Skip to content
Sign up

flagon-io/g1t

public
00

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