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