Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/repos/src/registry.rs

904 lines34,141 bytesCodeBlame
1//! Repository metadata in D1.
2
3use std::cell::RefCell;
4use std::collections::HashMap;
5
6use g1t_contracts::Viewer;
7use g1t_contracts::access::{self, Capability, RepoRole};
8use g1t_contracts::repos::{Repo, RepoPath};
9use serde::Deserialize;
10use worker::wasm_bindgen::JsValue;
11use worker::{D1Database, Result};
12
13#[derive(Deserialize)]
14pub(crate) struct RepoRow {
15 id: String,
16 namespace: String,
17 name: String,
18 description: Option<String>,
19 is_private: u8,
20 owner_id: String,
21 default_branch: String,
22 fork_of: Option<String>,
23 protected: u8,
24 created_at: String,
25 /// Null only on rows written before the column existed and not yet
26 /// migrated; their key is the one worked out from the path.
27 #[serde(default)]
28 store: Option<String>,
29 /// JSON; absent on rows read before the column existed.
30 #[serde(default)]
31 topics: Option<String>,
32 #[serde(default)]
33 website: Option<String>,
34 #[serde(default)]
35 archived_at: Option<String>,
36 #[serde(default)]
37 deleted_at: Option<String>,
38 /// Bumped by everything that changes the repository's refs; see
39 /// [`RefsState`]. Absent on rows read before the column existed.
40 #[serde(default)]
41 refs_version: Option<f64>,
42 #[serde(default)]
43 refs_open_until: Option<f64>,
44 /// A pull request working copy whose git data was removed, and the
45 /// head it had (forks.rs). Absent before the columns existed.
46 #[serde(default)]
47 retired_at: Option<String>,
48 #[serde(default)]
49 retired_head: Option<String>,
50}
51
52/// Where a repository's refs stand, as its row last said: `version` goes up
53/// with every change g1t makes to them, so an answer that lists them (see
54/// refs_cache.rs) is kept under the version it was made at, and a change
55/// leaves it behind. Until `open_until` (milliseconds) a credential that
56/// can change them is out of g1t's hands, and nothing is kept.
57#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
58pub struct RefsState {
59 pub version: u64,
60 pub open_until: u64,
61}
62
63/// The newest [`RefsState`] this isolate has read or written, by
64/// repository id. A version only goes up, so an older read finishing late
65/// never takes a newer one back.
66#[derive(Default)]
67pub struct RefsStates {
68 states: HashMap<String, RefsState>,
69}
70
71impl RefsStates {
72 pub fn note(&mut self, id: &str, state: RefsState) {
73 let kept = self.states.entry(id.to_owned()).or_default();
74 kept.version = kept.version.max(state.version);
75 kept.open_until = kept.open_until.max(state.open_until);
76 }
77
78 pub fn get(&self, id: &str) -> Option<RefsState> {
79 self.states.get(id).copied()
80 }
81}
82
83thread_local! {
84 static REFS: RefCell<RefsStates> = RefCell::new(RefsStates::default());
85}
86
87/// Where the refs of the repository with this id stand, as this isolate
88/// last read them; `None` before the column existed or before its row was
89/// read here.
90pub fn refs_state(id: &str) -> Option<RefsState> {
91 REFS.with(|refs| refs.borrow().get(id))
92}
93
94fn note_refs(id: &str, version: Option<f64>, open_until: Option<f64>) {
95 if let Some(version) = version {
96 let state = RefsState {
97 version: version as u64,
98 open_until: open_until.unwrap_or(0.0) as u64,
99 };
100 REFS.with(|refs| refs.borrow_mut().note(id, state));
101 }
102}
103
104thread_local! {
105 /// Working copies whose git data was removed, by id, with the head
106 /// each had (forks.rs). Filled whenever a row is read.
107 static RETIRED: RefCell<HashMap<String, String>> = RefCell::new(HashMap::new());
108}
109
110/// The head a removed working copy had, if the repository with this id is one.
111pub fn retired(id: &str) -> Option<String> {
112 RETIRED.with(|retired| retired.borrow().get(id).cloned())
113}
114
115/// Records whether the repository with this id is a removed working copy.
116pub fn note_retired(id: &str, head: Option<&str>) {
117 RETIRED.with(|retired| {
118 let mut retired = retired.borrow_mut();
119 match head {
120 Some(head) => {
121 retired.insert(id.to_owned(), head.to_owned());
122 }
123 None => {
124 retired.remove(id);
125 }
126 }
127 });
128}
129
130thread_local! {
131 /// Store keys that differ from the one a repository's path gives: those
132 /// of repositories whose workspace was renamed after they were made.
133 /// Filled whenever a row is read or written, so every `Repo` this
134 /// service holds has its key here. A key never changes once given, so
135 /// requests sharing the isolate can share the map.
136 static MOVED: RefCell<HashMap<String, String>> = RefCell::new(HashMap::new());
137}
138
139/// How long a fetch may go by a repository's row as it was read a moment
140/// ago: a clone is two or three requests in quick succession, and each
141/// would otherwise read the same row. Short enough that making a repository
142/// private, archiving or deleting it applies within seconds.
143pub const RECENT_MS: u64 = 5_000;
144
145/// Repositories read in the last [`RECENT_MS`], by path. Only rows that
146/// were found are kept, so a repository just made is never missed.
147#[derive(Default)]
148pub struct Recent {
149 rows: HashMap<(String, String), (Repo, u64)>,
150}
151
152impl Recent {
153 fn key(path: &RepoPath) -> (String, String) {
154 (path.namespace.to_lowercase(), path.name.to_lowercase())
155 }
156
157 pub fn get(&self, path: &RepoPath, now: u64) -> Option<Repo> {
158 self.rows
159 .get(&Self::key(path))
160 .filter(|(_, read)| now.saturating_sub(*read) < RECENT_MS)
161 .map(|(repo, _)| repo.clone())
162 }
163
164 pub fn keep(&mut self, path: &RepoPath, repo: &Repo, now: u64) {
165 self.rows.retain(|_, (_, read)| now.saturating_sub(*read) < RECENT_MS);
166 self.rows.insert(Self::key(path), (repo.clone(), now));
167 }
168}
169
170thread_local! {
171 static RECENT: RefCell<Recent> = RefCell::new(Recent::default());
172}
173
174/// The key a repository's path gives: what every repository was stored
175/// under before workspaces could be renamed.
176pub fn path_key(repo: &Repo) -> String {
177 format!("{}--{}", repo.namespace, repo.name)
178}
179
180/// Records where a repository is stored, when its path does not say.
181pub fn remember_store(repo: &Repo, store: &str) {
182 if store != path_key(repo) {
183 MOVED.with(|moved| moved.borrow_mut().insert(repo.id.clone(), store.to_owned()));
184 }
185}
186
187impl From<RepoRow> for Repo {
188 fn from(row: RepoRow) -> Self {
189 let repo = Repo {
190 id: row.id,
191 namespace: row.namespace,
192 name: row.name,
193 description: row.description,
194 is_private: row.is_private != 0,
195 owner_id: row.owner_id,
196 default_branch: row.default_branch,
197 fork_of: row.fork_of,
198 protected: row.protected != 0,
199 created_at: row.created_at,
200 topics: row
201 .topics
202 .as_deref()
203 .and_then(|topics| serde_json::from_str(topics).ok())
204 .unwrap_or_default(),
205 website: row.website,
206 archived_at: row.archived_at,
207 };
208 if let Some(store) = &row.store {
209 remember_store(&repo, store);
210 }
211 note_refs(&repo.id, row.refs_version, row.refs_open_until);
212 note_retired(&repo.id, row.retired_at.as_ref().and(row.retired_head.as_deref()));
213 repo
214 }
215}
216
217/// The key a repo is stored under in the git store.
218pub fn store_key(repo: &Repo) -> String {
219 let key = MOVED
220 .with(|moved| moved.borrow().get(&repo.id).cloned())
221 .unwrap_or_else(|| path_key(repo));
222 // Its interactions with the store are metered for its workspace.
223 crate::meters::note_owner(&key, &repo.namespace);
224 key
225}
226
227/// The viewer's role on `repo` (see `g1t_contracts::access`): ownership of
228/// its workspace, the workspace's base permission, a direct grant, or
229/// Read on a public repository. A pull request's fork is its author's to
230/// write; whoever can read the repository it came from can read it too,
231/// which `Repos::may_read` checks.
232pub fn role(repo: &Repo, viewer: &Viewer) -> Option<RepoRole> {
233 if repo.fork_of.is_some() {
234 let author = viewer.as_ref().is_some_and(|user| user.id == repo.owner_id);
235 return if author {
236 Some(RepoRole::Write)
237 } else if repo.is_private {
238 None
239 } else {
240 Some(RepoRole::Read)
241 };
242 }
243 access::permission(viewer.as_ref(), repo)
244}
245
246/// Whether the viewer may read `repo`, going by the repository alone.
247pub fn can_read(repo: &Repo, viewer: &Viewer) -> bool {
248 role(repo, viewer).is_some()
249}
250
251/// Whether the viewer may push to `repo`: Write or higher, or the author
252/// of a pull request's fork.
253pub fn can_write(repo: &Repo, viewer: &Viewer) -> bool {
254 can(repo, viewer, Capability::Push)
255}
256
257/// Whether the viewer may do `capability` in `repo`. A fork has only its
258/// author's Write.
259pub fn can(repo: &Repo, viewer: &Viewer, capability: Capability) -> bool {
260 if repo.fork_of.is_some() {
261 return role(repo, viewer).is_some_and(|role| access::allows(role, capability))
262 && !access::OWNER_ONLY.contains(&capability);
263 }
264 access::can(viewer.as_ref(), repo, capability)
265}
266
267fn optional(value: &Option<String>) -> JsValue {
268 value.as_deref().map_or(JsValue::NULL, JsValue::from)
269}
270
271pub struct Registry {
272 pub db: D1Database,
273}
274
275impl Registry {
276 pub async fn by_path(&self, path: &RepoPath) -> Result<Option<Repo>> {
277 Ok(self
278 .db
279 .prepare("SELECT * FROM repos WHERE namespace = ? AND name = ? AND deleted_at IS NULL")
280 .bind(&[
281 path.namespace.to_lowercase().into(),
282 path.name.to_lowercase().into(),
283 ])?
284 .first::<RepoRow>(None)
285 .await?
286 .map(Repo::from))
287 }
288
289 /// The repository at `path`, as read in the last few seconds if it was
290 /// (see [`RECENT_MS`]). For fetches only: a push always reads the row.
291 pub async fn by_path_recent(&self, path: &RepoPath) -> Result<Option<Repo>> {
292 let now = g1t_kit::now_ms();
293 if let Some(repo) = RECENT.with(|recent| recent.borrow().get(path, now)) {
294 return Ok(Some(repo));
295 }
296 let found = self.by_path(path).await?;
297 if let Some(repo) = &found {
298 RECENT.with(|recent| recent.borrow_mut().keep(path, repo, now));
299 }
300 Ok(found)
301 }
302
303 /// Its details; who can see it changes with `set_private`.
304 pub async fn update(
305 &self,
306 id: &str,
307 description: Option<&str>,
308 protected: bool,
309 topics: &[String],
310 website: Option<&str>,
311 ) -> Result<()> {
312 self.db
313 .prepare("UPDATE repos SET description = ?, protected = ?, topics = ?, website = ? WHERE id = ?")
314 .bind(&[
315 description.map_or(JsValue::NULL, JsValue::from),
316 u32::from(protected).into(),
317 serde_json::to_string(topics)?.into(),
318 website.map_or(JsValue::NULL, JsValue::from),
319 id.into(),
320 ])?
321 .run()
322 .await?;
323 Ok(())
324 }
325
326 pub async fn by_id(&self, id: &str) -> Result<Option<Repo>> {
327 Ok(self
328 .db
329 .prepare("SELECT * FROM repos WHERE id = ? AND deleted_at IS NULL")
330 .bind(&[id.into()])?
331 .first::<RepoRow>(None)
332 .await?
333 .map(Repo::from))
334 }
335
336 /// The repository at `path`, deleted or not: what holds the name.
337 pub async fn by_path_any(&self, path: &RepoPath) -> Result<Option<(Repo, Option<String>)>> {
338 Ok(self
339 .db
340 .prepare("SELECT * FROM repos WHERE namespace = ? AND name = ?")
341 .bind(&[
342 path.namespace.to_lowercase().into(),
343 path.name.to_lowercase().into(),
344 ])?
345 .first::<RepoRow>(None)
346 .await?
347 .map(|mut row| {
348 let deleted_at = row.deleted_at.take();
349 (Repo::from(row), deleted_at)
350 }))
351 }
352
353 /// Repos the viewer may see, newest first. Excludes pull request forks.
354 /// With `member_only`, only repos in the viewer's own workspaces.
355 pub async fn list(
356 &self,
357 viewer: &Viewer,
358 query: Option<&str>,
359 namespace: Option<&str>,
360 member_only: bool,
361 ) -> Result<Vec<Repo>> {
362 let workspaces: Vec<&str> = viewer
363 .iter()
364 .flat_map(|user| &user.workspaces)
365 .map(|membership| membership.slug.as_str())
366 .collect();
367 // The workspaces whose private repositories the viewer reads all
368 // of (an owner, or a base permission other than none), and the
369 // repositories they were given a role on: see access.rs. A probe
370 // repository in each workspace stands for all of them.
371 let reading: Vec<&str> = viewer
372 .iter()
373 .flat_map(|user| {
374 user.workspaces.iter().filter(move |membership| {
375 let probe = access::RepoRef { id: "", namespace: &membership.slug, private: true };
376 access::granted(user, probe).is_some()
377 })
378 })
379 .map(|membership| membership.slug.as_str())
380 .collect();
381 let granted: Vec<&str> = viewer
382 .iter()
383 .flat_map(|user| &user.grants)
384 .map(|grant| grant.repo_id.as_str())
385 .collect();
386 let mut params: Vec<JsValue> = vec![
387 serde_json::to_string(&reading)?.into(),
388 serde_json::to_string(&granted)?.into(),
389 ];
390 let private_ok = "(namespace IN (SELECT value FROM json_each(?)) OR id IN (SELECT value FROM json_each(?)))";
391 let mut conditions = vec![
392 "fork_of IS NULL AND deleted_at IS NULL".to_owned(),
393 format!("(is_private = 0 OR {private_ok})"),
394 ];
395 if member_only {
396 conditions.push("namespace IN (SELECT value FROM json_each(?))".to_owned());
397 params.push(serde_json::to_string(&workspaces)?.into());
398 }
399 if let Some(namespace) = namespace {
400 conditions.push("namespace = ?".to_owned());
401 params.push(namespace.to_lowercase().into());
402 }
403 if let Some(query) = query.map(str::trim).filter(|query| !query.is_empty()) {
404 conditions
405 .push("(name LIKE ? ESCAPE '\\' OR description LIKE ? ESCAPE '\\')".to_owned());
406 // LIKE wildcards in the query are matched literally.
407 let escaped: String = query
408 .chars()
409 .flat_map(|c| match c {
410 '\\' | '%' | '_' => vec!['\\', c],
411 _ => vec![c],
412 })
413 .collect();
414 let pattern = format!("%{escaped}%");
415 params.push(pattern.as_str().into());
416 params.push(pattern.into());
417 }
418 let sql = format!(
419 "SELECT * FROM repos WHERE {} ORDER BY created_at DESC, id DESC LIMIT 50",
420 conditions.join(" AND ")
421 );
422 let rows = self
423 .db
424 .prepare(sql)
425 .bind(&params)?
426 .all()
427 .await?
428 .results::<RepoRow>()?;
429 Ok(rows.into_iter().map(Repo::from).collect())
430 }
431
432 /// Of these ids, the repositories (not forks) the viewer may read.
433 pub async fn readable(&self, ids: &[String], viewer: &Viewer) -> Result<Vec<Repo>> {
434 let ids: Vec<&String> = ids.iter().take(g1t_contracts::repos::MAX_READABLE).collect();
435 if ids.is_empty() {
436 return Ok(Vec::new());
437 }
438 // One parameter however many ids: D1 binds at most 100.
439 let rows = self
440 .db
441 .prepare(
442 "SELECT * FROM repos
443 WHERE id IN (SELECT value FROM json_each(?)) AND fork_of IS NULL AND deleted_at IS NULL",
444 )
445 .bind(&[serde_json::to_string(&ids)?.into()])?
446 .all()
447 .await?
448 .results::<RepoRow>()?;
449 Ok(rows
450 .into_iter()
451 .map(Repo::from)
452 .filter(|repo| can_read(repo, viewer))
453 .collect())
454 }
455
456 /// The workspaces in which this account made a public repository.
457 pub async fn public_namespaces(&self, owner_id: &str) -> Result<Vec<String>> {
458 #[derive(Deserialize)]
459 struct Row {
460 namespace: String,
461 }
462 Ok(self
463 .db
464 .prepare(
465 "SELECT DISTINCT namespace FROM repos
466 WHERE owner_id = ? AND is_private = 0 AND fork_of IS NULL AND deleted_at IS NULL
467 ORDER BY namespace",
468 )
469 .bind(&[owner_id.into()])?
470 .all()
471 .await?
472 .results::<Row>()?
473 .into_iter()
474 .map(|row| row.namespace)
475 .collect())
476 }
477
478 /// Repositories that are not forks, by id, a page at a time.
479 pub async fn ids_after(&self, after: Option<&str>, limit: u32) -> Result<Vec<String>> {
480 #[derive(Deserialize)]
481 struct Row {
482 id: String,
483 }
484 Ok(self
485 .db
486 .prepare("SELECT id FROM repos WHERE fork_of IS NULL AND deleted_at IS NULL AND id > ? ORDER BY id LIMIT ?")
487 .bind(&[after.unwrap_or("").into(), limit.into()])?
488 .all()
489 .await?
490 .results::<Row>()?
491 .into_iter()
492 .map(|row| row.id)
493 .collect())
494 }
495
496 /// Adds a pushed pack's bytes to what the repository is counted as
497 /// holding: its own, or, for a pull request's working copy, the
498 /// repository it is a copy of, whose storage it is.
499 pub async fn add_stored_bytes(&self, repo: &Repo, bytes: u64) -> Result<()> {
500 if bytes == 0 {
501 return Ok(());
502 }
503 let root = repo.fork_of.as_deref().unwrap_or(&repo.id);
504 self.db
505 .prepare("UPDATE repos SET stored_bytes = stored_bytes + ? WHERE id = ?")
506 .bind(&[(bytes as f64).into(), root.into()])?
507 .run()
508 .await?;
509 Ok(())
510 }
511
512 /// Which of these `namespace/name` paths are private. A working copy
513 /// answers as its repository. Unknown paths are left out.
514 pub async fn visibility(&self, paths: &[String]) -> Result<Vec<g1t_contracts::repos::RepoVisibility>> {
515 let mut out = Vec::new();
516 for path in paths.iter().take(50) {
517 let Some((namespace, name)) = path.split_once('/') else { continue };
518 let Some(repo) = self
519 .by_path(&RepoPath { namespace: namespace.to_owned(), name: name.to_owned() })
520 .await?
521 else {
522 continue;
523 };
524 let is_private = match &repo.fork_of {
525 Some(parent) => self.by_id(parent).await?.map_or(repo.is_private, |parent| parent.is_private),
526 None => repo.is_private,
527 };
528 out.push(g1t_contracts::repos::RepoVisibility { path: path.clone(), is_private });
529 }
530 Ok(out)
531 }
532
533 /// What each workspace's repositories are counted as holding, private
534 /// and public apart. Working copies count toward their repository.
535 pub async fn storage(&self) -> Result<Vec<g1t_contracts::repos::WorkspaceStorage>> {
536 #[derive(Deserialize)]
537 struct Row {
538 namespace: String,
539 private_bytes: Option<f64>,
540 public_bytes: Option<f64>,
541 }
542 Ok(self
543 .db
544 .prepare(
545 "SELECT namespace,
546 SUM(CASE WHEN is_private = 1 THEN stored_bytes ELSE 0 END) AS private_bytes,
547 SUM(CASE WHEN is_private = 0 THEN stored_bytes ELSE 0 END) AS public_bytes
548 FROM repos WHERE fork_of IS NULL AND deleted_at IS NULL AND stored_bytes > 0 GROUP BY namespace",
549 )
550 .all()
551 .await?
552 .results::<Row>()?
553 .into_iter()
554 .map(|row| g1t_contracts::repos::WorkspaceStorage {
555 namespace: row.namespace,
556 private_bytes: row.private_bytes.unwrap_or(0.0) as i64,
557 public_bytes: row.public_bytes.unwrap_or(0.0) as i64,
558 })
559 .collect())
560 }
561
562 /// What one workspace's private repositories are counted as holding.
563 pub async fn private_bytes(&self, namespace: &str) -> Result<i64> {
564 #[derive(Deserialize)]
565 struct Row {
566 bytes: Option<f64>,
567 }
568 Ok(self
569 .db
570 .prepare("SELECT SUM(stored_bytes) AS bytes FROM repos WHERE namespace = ? AND is_private = 1 AND fork_of IS NULL AND deleted_at IS NULL")
571 .bind(&[namespace.into()])?
572 .first::<Row>(None)
573 .await?
574 .and_then(|row| row.bytes)
575 .unwrap_or(0.0) as i64)
576 }
577
578 /// Forgets a repository that could not be filled.
579 pub async fn remove(&self, id: &str) -> Result<()> {
580 self.db
581 .prepare("DELETE FROM repos WHERE id = ?")
582 .bind(&[id.into()])?
583 .run()
584 .await?;
585 Ok(())
586 }
587
588 /// Picks the store key for a repository about to be made, and
589 /// remembers it: the one its path gives, unless a repository already
590 /// holds that (one made in a workspace that has since been renamed,
591 /// whose old name this workspace now has), when its id.
592 ///
593 /// `namespace` is the git store namespace it goes in (shards.rs), or
594 /// `None` for the default, `default`. A name is taken in any of them.
595 pub async fn claim_store_key(&self, repo: &Repo, namespace: Option<&str>, default: &str) -> Result<String> {
596 let wanted = path_key(repo);
597 let held = self
598 .db
599 .prepare(
600 "SELECT 1 AS held FROM repos
601 WHERE store = ?1 OR (instr(store, '/') > 0 AND substr(store, instr(store, '/') + 1) = ?1)",
602 )
603 .bind(&[wanted.as_str().into()])?
604 .first::<serde_json::Value>(None)
605 .await?
606 .is_some();
607 let name = if held { repo.id.clone() } else { wanted };
608 let key = crate::shards::compose(namespace, &name, default);
609 remember_store(repo, &key);
610 Ok(key)
611 }
612
613
614 /// Moves a renamed workspace's repositories to its current slug, from
615 /// any of `stale`. A repository whose name the current slug already has
616 /// (one pushed there in the moment before this ran) stays where it is;
617 /// returns how many did.
618 pub async fn rename_namespace(&self, stale: &[String], current: &str) -> Result<usize> {
619 if stale.is_empty() {
620 return Ok(0);
621 }
622 let marks = vec!["?"; stale.len()].join(", ");
623 let mut moved: Vec<JsValue> = vec![current.into()];
624 moved.extend(stale.iter().map(|slug| JsValue::from(slug.as_str())));
625 let left: Vec<JsValue> = stale.iter().map(|slug| JsValue::from(slug.as_str())).collect();
626 let results = self
627 .db
628 .batch(vec![
629 self.db
630 .prepare(format!(
631 "UPDATE OR IGNORE repos SET namespace = ? WHERE namespace IN ({marks})"
632 ))
633 .bind(&moved)?,
634 self.db
635 .prepare(format!(
636 "SELECT count(*) AS left FROM repos WHERE namespace IN ({marks})"
637 ))
638 .bind(&left)?,
639 // Git operations follow the workspace, added together.
640 self.db
641 .prepare(format!(
642 "INSERT INTO git_operations (namespace, hour, operations)
643 SELECT ?, hour, SUM(operations) FROM git_operations WHERE namespace IN ({marks}) GROUP BY hour
644 ON CONFLICT (namespace, hour) DO UPDATE SET operations = git_operations.operations + excluded.operations"
645 ))
646 .bind(&moved)?,
647 self.db
648 .prepare(format!("DELETE FROM git_operations WHERE namespace IN ({marks})"))
649 .bind(&left)?,
650 // Paths repositories were transferred away from follow the
651 // workspace too, so the old slug's redirect then finds them.
652 self.db
653 .prepare(format!(
654 "UPDATE OR IGNORE repo_redirects SET namespace = ? WHERE namespace IN ({marks})"
655 ))
656 .bind(&moved)?,
657 ])
658 .await?;
659 #[derive(Deserialize)]
660 struct Left {
661 left: usize,
662 }
663 Ok(results
664 .get(1)
665 .map(|result| result.results::<Left>())
666 .transpose()?
667 .and_then(|rows| rows.into_iter().next())
668 .map_or(0, |row| row.left))
669 }
670
671 /// Records that the refs of the repository with this id changed, after
672 /// they did: what anything that lists them keeps goes stale.
673 pub async fn refs_moved(&self, id: &str) -> Result<()> {
674 self.bump_refs(
675 "UPDATE repos SET refs_version = refs_version + 1 WHERE id = ?
676 RETURNING refs_version, refs_open_until",
677 &[id.into()],
678 id,
679 )
680 .await
681 }
682
683 /// Records that a credential able to change the refs of the repository
684 /// with this id was handed out of g1t's hands, until `until`
685 /// (milliseconds): until then, nothing that lists them is kept.
686 pub async fn refs_open(&self, id: &str, until: u64) -> Result<()> {
687 self.bump_refs(
688 "UPDATE repos SET refs_version = refs_version + 1,
689 refs_open_until = max(coalesce(refs_open_until, 0), ?)
690 WHERE id = ? RETURNING refs_version, refs_open_until",
691 &[(until as f64).into(), id.into()],
692 id,
693 )
694 .await
695 }
696
697 async fn bump_refs(&self, sql: &str, params: &[JsValue], id: &str) -> Result<()> {
698 #[derive(Deserialize)]
699 struct Bumped {
700 refs_version: Option<f64>,
701 refs_open_until: Option<f64>,
702 }
703 let bumped = self
704 .db
705 .prepare(sql)
706 .bind(params)?
707 .first::<Bumped>(None)
708 .await?;
709 if let Some(bumped) = bumped {
710 note_refs(id, bumped.refs_version, bumped.refs_open_until);
711 }
712 Ok(())
713 }
714
715 pub async fn insert(&self, repo: &Repo) -> Result<()> {
716 self.db
717 .prepare(
718 "INSERT INTO repos
719 (id, namespace, name, description, is_private, owner_id,
720 default_branch, fork_of, created_at, store)
721 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
722 )
723 .bind(&[
724 repo.id.as_str().into(),
725 repo.namespace.as_str().into(),
726 repo.name.as_str().into(),
727 optional(&repo.description),
728 (repo.is_private as u8).into(),
729 repo.owner_id.as_str().into(),
730 repo.default_branch.as_str().into(),
731 optional(&repo.fork_of),
732 repo.created_at.as_str().into(),
733 store_key(repo).into(),
734 ])?
735 .run()
736 .await?;
737 Ok(())
738 }
739}
740
741#[cfg(test)]
742mod tests {
743 use super::*;
744 use g1t_contracts::access::{BasePermission, RepoGrant};
745 use g1t_contracts::{Membership, Role, User};
746
747 #[test]
748 fn the_refs_state_kept_only_moves_forward() {
749 let mut states = RefsStates::default();
750 assert_eq!(states.get("rep_1"), None);
751 states.note("rep_1", RefsState { version: 3, open_until: 0 });
752 // A read that started before a bump and finished after it.
753 states.note("rep_1", RefsState { version: 2, open_until: 0 });
754 assert_eq!(states.get("rep_1").unwrap().version, 3);
755 states.note("rep_1", RefsState { version: 4, open_until: 9_000 });
756 states.note("rep_1", RefsState { version: 5, open_until: 0 });
757 assert_eq!(states.get("rep_1"), Some(RefsState { version: 5, open_until: 9_000 }));
758 assert_eq!(states.get("rep_2"), None);
759 }
760
761 #[test]
762 fn a_row_from_before_the_column_has_no_refs_state() {
763 let row = |version: Option<f64>| RepoRow {
764 id: format!("rep_row_{}", version.is_some()),
765 namespace: "acme".into(),
766 name: "rocket".into(),
767 description: None,
768 is_private: 0,
769 owner_id: "usr_owner".into(),
770 default_branch: "main".into(),
771 fork_of: None,
772 protected: 0,
773 created_at: String::new(),
774 store: None,
775 topics: None,
776 website: None,
777 archived_at: None,
778 deleted_at: None,
779 refs_version: version,
780 refs_open_until: None,
781 retired_at: None,
782 retired_head: None,
783 };
784 let old = Repo::from(row(None));
785 assert_eq!(refs_state(&old.id), None);
786 let new = Repo::from(row(Some(7.0)));
787 assert_eq!(refs_state(&new.id), Some(RefsState { version: 7, open_until: 0 }));
788 }
789
790 #[test]
791 fn a_removed_working_copy_is_known_by_its_row() {
792 note_retired("rep_fork", Some("abc"));
793 assert_eq!(retired("rep_fork").as_deref(), Some("abc"));
794 note_retired("rep_fork", None);
795 assert_eq!(retired("rep_fork"), None);
796 }
797
798 #[test]
799 fn a_repository_read_a_moment_ago_is_reused_for_a_few_seconds() {
800 let mut recent = Recent::default();
801 let path = RepoPath {
802 namespace: "Acme".into(),
803 name: "Rocket".into(),
804 };
805 recent.keep(&path, &repo(false), 1_000);
806 // Paths are matched as the table matches them, ignoring case.
807 let lower = RepoPath {
808 namespace: "acme".into(),
809 name: "rocket".into(),
810 };
811 assert_eq!(recent.get(&lower, 1_000 + RECENT_MS - 1).unwrap().id, "rep_1");
812 assert!(recent.get(&lower, 1_000 + RECENT_MS).is_none());
813 let other = RepoPath {
814 namespace: "acme".into(),
815 name: "booster".into(),
816 };
817 assert!(recent.get(&other, 1_000).is_none());
818 // Keeping another later drops the stale row.
819 recent.keep(&other, &repo(true), 1_000 + RECENT_MS);
820 assert_eq!(recent.rows.len(), 1);
821 }
822
823 fn repo(private: bool) -> Repo {
824 Repo {
825 id: "rep_1".into(),
826 namespace: "acme".into(),
827 name: "rocket".into(),
828 description: None,
829 is_private: private,
830 owner_id: "usr_owner".into(),
831 default_branch: "main".into(),
832 fork_of: None,
833 protected: false,
834 created_at: String::new(),
835 topics: Vec::new(),
836 website: None,
837 archived_at: None,
838 }
839 }
840
841 fn person(id: &str, memberships: Vec<Membership>, grants: Vec<(&str, RepoRole)>) -> Viewer {
842 Some(User {
843 id: id.into(),
844 username: id.into(),
845 verified: true,
846 workspaces: memberships,
847 grants: grants
848 .into_iter()
849 .map(|(repo_id, role)| RepoGrant { repo_id: repo_id.into(), workspace: "acme".into(), role })
850 .collect(),
851 ..User::default()
852 })
853 }
854
855 /// What git asks: clone and fetch need Read on a private repository,
856 /// push needs Write.
857 #[test]
858 fn git_reads_with_read_and_pushes_with_write() {
859 let private = repo(true);
860 let reader = person("usr_r", vec![], vec![("rep_1", RepoRole::Read)]);
861 assert!(can_read(&private, &reader));
862 assert!(!can_write(&private, &reader));
863 let writer = person("usr_w", vec![], vec![("rep_1", RepoRole::Write)]);
864 assert!(can_read(&private, &writer) && can_write(&private, &writer));
865 let stranger = person("usr_s", vec![], vec![("rep_2", RepoRole::Admin)]);
866 assert!(!can_read(&private, &stranger) && !can_write(&private, &stranger));
867 assert!(!can_read(&private, &None));
868 // A public repository: anyone clones, nobody without Write pushes.
869 let public = repo(false);
870 assert!(can_read(&public, &None) && !can_write(&public, &None));
871 assert!(can_read(&public, &stranger) && !can_write(&public, &stranger));
872 }
873
874 #[test]
875 fn members_follow_the_base_permission_and_owners_have_admin() {
876 let private = repo(true);
877 let default_member = person("usr_m", vec![Membership::member("acme")], vec![]);
878 assert!(can_write(&private, &default_member));
879 assert!(!can(&private, &default_member, Capability::ManageIntegrations));
880 let none = Membership { base_permission: Some(BasePermission::None), ..Membership::member("acme") };
881 let locked_out = person("usr_n", vec![none.clone()], vec![]);
882 assert!(!can_read(&private, &locked_out));
883 let given = person("usr_g", vec![none], vec![("rep_1", RepoRole::Triage)]);
884 assert!(can_read(&private, &given) && !can_write(&private, &given));
885 let owner = person("usr_o", vec![Membership { role: Role::Owner, ..Membership::member("acme") }], vec![]);
886 assert_eq!(role(&private, &owner), Some(RepoRole::Admin));
887 assert!(can(&private, &owner, Capability::Delete));
888 }
889
890 #[test]
891 fn a_pull_requests_fork_is_its_authors() {
892 let fork = Repo {
893 namespace: "pulls".into(),
894 fork_of: Some("rep_1".into()),
895 owner_id: "usr_a".into(),
896 ..repo(true)
897 };
898 let author = person("usr_a", vec![], vec![]);
899 assert!(can_write(&fork, &author));
900 assert!(!can(&fork, &author, Capability::ManageSettings));
901 let other = person("usr_b", vec![Membership::member("acme")], vec![]);
902 assert!(!can_write(&fork, &other));
903 }
904}