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

632 lines23,959 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}
39
40thread_local! {
41 /// Store keys that differ from the one a repository's path gives: those
42 /// of repositories whose workspace was renamed after they were made.
43 /// Filled whenever a row is read or written, so every `Repo` this
44 /// service holds has its key here. A key never changes once given, so
45 /// requests sharing the isolate can share the map.
46 static MOVED: RefCell<HashMap<String, String>> = RefCell::new(HashMap::new());
47}
48
49/// The key a repository's path gives: what every repository was stored
50/// under before workspaces could be renamed.
51pub fn path_key(repo: &Repo) -> String {
52 format!("{}--{}", repo.namespace, repo.name)
53}
54
55/// Records where a repository is stored, when its path does not say.
56pub fn remember_store(repo: &Repo, store: &str) {
57 if store != path_key(repo) {
58 MOVED.with(|moved| moved.borrow_mut().insert(repo.id.clone(), store.to_owned()));
59 }
60}
61
62impl From<RepoRow> for Repo {
63 fn from(row: RepoRow) -> Self {
64 let repo = Repo {
65 id: row.id,
66 namespace: row.namespace,
67 name: row.name,
68 description: row.description,
69 is_private: row.is_private != 0,
70 owner_id: row.owner_id,
71 default_branch: row.default_branch,
72 fork_of: row.fork_of,
73 protected: row.protected != 0,
74 created_at: row.created_at,
75 topics: row
76 .topics
77 .as_deref()
78 .and_then(|topics| serde_json::from_str(topics).ok())
79 .unwrap_or_default(),
80 website: row.website,
81 archived_at: row.archived_at,
82 };
83 if let Some(store) = &row.store {
84 remember_store(&repo, store);
85 }
86 repo
87 }
88}
89
90/// The key a repo is stored under in the git store.
91pub fn store_key(repo: &Repo) -> String {
92 MOVED
93 .with(|moved| moved.borrow().get(&repo.id).cloned())
94 .unwrap_or_else(|| path_key(repo))
95}
96
97/// The viewer's role on `repo` (see `g1t_contracts::access`): ownership of
98/// its workspace, the workspace's base permission, a direct grant, or
99/// Read on a public repository. A pull request's fork is its author's to
100/// write; whoever can read the repository it came from can read it too,
101/// which `Repos::may_read` checks.
102pub fn role(repo: &Repo, viewer: &Viewer) -> Option<RepoRole> {
103 if repo.fork_of.is_some() {
104 let author = viewer.as_ref().is_some_and(|user| user.id == repo.owner_id);
105 return if author {
106 Some(RepoRole::Write)
107 } else if repo.is_private {
108 None
109 } else {
110 Some(RepoRole::Read)
111 };
112 }
113 access::permission(viewer.as_ref(), repo)
114}
115
116/// Whether the viewer may read `repo`, going by the repository alone.
117pub fn can_read(repo: &Repo, viewer: &Viewer) -> bool {
118 role(repo, viewer).is_some()
119}
120
121/// Whether the viewer may push to `repo`: Write or higher, or the author
122/// of a pull request's fork.
123pub fn can_write(repo: &Repo, viewer: &Viewer) -> bool {
124 can(repo, viewer, Capability::Push)
125}
126
127/// Whether the viewer may do `capability` in `repo`. A fork has only its
128/// author's Write.
129pub fn can(repo: &Repo, viewer: &Viewer, capability: Capability) -> bool {
130 if repo.fork_of.is_some() {
131 return role(repo, viewer).is_some_and(|role| access::allows(role, capability))
132 && !access::OWNER_ONLY.contains(&capability);
133 }
134 access::can(viewer.as_ref(), repo, capability)
135}
136
137fn optional(value: &Option<String>) -> JsValue {
138 value.as_deref().map_or(JsValue::NULL, JsValue::from)
139}
140
141pub struct Registry {
142 pub db: D1Database,
143}
144
145impl Registry {
146 pub async fn by_path(&self, path: &RepoPath) -> Result<Option<Repo>> {
147 Ok(self
148 .db
149 .prepare("SELECT * FROM repos WHERE namespace = ? AND name = ? AND deleted_at IS NULL")
150 .bind(&[
151 path.namespace.to_lowercase().into(),
152 path.name.to_lowercase().into(),
153 ])?
154 .first::<RepoRow>(None)
155 .await?
156 .map(Repo::from))
157 }
158
159 /// Its details; who can see it changes with `set_private`.
160 pub async fn update(
161 &self,
162 id: &str,
163 description: Option<&str>,
164 protected: bool,
165 topics: &[String],
166 website: Option<&str>,
167 ) -> Result<()> {
168 self.db
169 .prepare("UPDATE repos SET description = ?, protected = ?, topics = ?, website = ? WHERE id = ?")
170 .bind(&[
171 description.map_or(JsValue::NULL, JsValue::from),
172 u32::from(protected).into(),
173 serde_json::to_string(topics)?.into(),
174 website.map_or(JsValue::NULL, JsValue::from),
175 id.into(),
176 ])?
177 .run()
178 .await?;
179 Ok(())
180 }
181
182 pub async fn by_id(&self, id: &str) -> Result<Option<Repo>> {
183 Ok(self
184 .db
185 .prepare("SELECT * FROM repos WHERE id = ? AND deleted_at IS NULL")
186 .bind(&[id.into()])?
187 .first::<RepoRow>(None)
188 .await?
189 .map(Repo::from))
190 }
191
192 /// The repository at `path`, deleted or not: what holds the name.
193 pub async fn by_path_any(&self, path: &RepoPath) -> Result<Option<(Repo, Option<String>)>> {
194 Ok(self
195 .db
196 .prepare("SELECT * FROM repos WHERE namespace = ? AND name = ?")
197 .bind(&[
198 path.namespace.to_lowercase().into(),
199 path.name.to_lowercase().into(),
200 ])?
201 .first::<RepoRow>(None)
202 .await?
203 .map(|mut row| {
204 let deleted_at = row.deleted_at.take();
205 (Repo::from(row), deleted_at)
206 }))
207 }
208
209 /// Repos the viewer may see, newest first. Excludes pull request forks.
210 /// With `member_only`, only repos in the viewer's own workspaces.
211 pub async fn list(
212 &self,
213 viewer: &Viewer,
214 query: Option<&str>,
215 namespace: Option<&str>,
216 member_only: bool,
217 ) -> Result<Vec<Repo>> {
218 let workspaces: Vec<&str> = viewer
219 .iter()
220 .flat_map(|user| &user.workspaces)
221 .map(|membership| membership.slug.as_str())
222 .collect();
223 // The workspaces whose private repositories the viewer reads all
224 // of (an owner, or a base permission other than none), and the
225 // repositories they were given a role on: see access.rs. A probe
226 // repository in each workspace stands for all of them.
227 let reading: Vec<&str> = viewer
228 .iter()
229 .flat_map(|user| {
230 user.workspaces.iter().filter(move |membership| {
231 let probe = access::RepoRef { id: "", namespace: &membership.slug, private: true };
232 access::granted(user, probe).is_some()
233 })
234 })
235 .map(|membership| membership.slug.as_str())
236 .collect();
237 let granted: Vec<&str> = viewer
238 .iter()
239 .flat_map(|user| &user.grants)
240 .map(|grant| grant.repo_id.as_str())
241 .collect();
242 let mut params: Vec<JsValue> = vec![
243 serde_json::to_string(&reading)?.into(),
244 serde_json::to_string(&granted)?.into(),
245 ];
246 let private_ok = "(namespace IN (SELECT value FROM json_each(?)) OR id IN (SELECT value FROM json_each(?)))";
247 let mut conditions = vec![
248 "fork_of IS NULL AND deleted_at IS NULL".to_owned(),
249 format!("(is_private = 0 OR {private_ok})"),
250 ];
251 if member_only {
252 conditions.push("namespace IN (SELECT value FROM json_each(?))".to_owned());
253 params.push(serde_json::to_string(&workspaces)?.into());
254 }
255 if let Some(namespace) = namespace {
256 conditions.push("namespace = ?".to_owned());
257 params.push(namespace.to_lowercase().into());
258 }
259 if let Some(query) = query.map(str::trim).filter(|query| !query.is_empty()) {
260 conditions
261 .push("(name LIKE ? ESCAPE '\\' OR description LIKE ? ESCAPE '\\')".to_owned());
262 // LIKE wildcards in the query are matched literally.
263 let escaped: String = query
264 .chars()
265 .flat_map(|c| match c {
266 '\\' | '%' | '_' => vec!['\\', c],
267 _ => vec![c],
268 })
269 .collect();
270 let pattern = format!("%{escaped}%");
271 params.push(pattern.as_str().into());
272 params.push(pattern.into());
273 }
274 let sql = format!(
275 "SELECT * FROM repos WHERE {} ORDER BY created_at DESC, id DESC LIMIT 50",
276 conditions.join(" AND ")
277 );
278 let rows = self
279 .db
280 .prepare(sql)
281 .bind(&params)?
282 .all()
283 .await?
284 .results::<RepoRow>()?;
285 Ok(rows.into_iter().map(Repo::from).collect())
286 }
287
288 /// Of these ids, the repositories (not forks) the viewer may read.
289 pub async fn readable(&self, ids: &[String], viewer: &Viewer) -> Result<Vec<Repo>> {
290 let ids: Vec<&String> = ids.iter().take(g1t_contracts::repos::MAX_READABLE).collect();
291 if ids.is_empty() {
292 return Ok(Vec::new());
293 }
294 // One parameter however many ids: D1 binds at most 100.
295 let rows = self
296 .db
297 .prepare(
298 "SELECT * FROM repos
299 WHERE id IN (SELECT value FROM json_each(?)) AND fork_of IS NULL AND deleted_at IS NULL",
300 )
301 .bind(&[serde_json::to_string(&ids)?.into()])?
302 .all()
303 .await?
304 .results::<RepoRow>()?;
305 Ok(rows
306 .into_iter()
307 .map(Repo::from)
308 .filter(|repo| can_read(repo, viewer))
309 .collect())
310 }
311
312 /// The workspaces in which this account made a public repository.
313 pub async fn public_namespaces(&self, owner_id: &str) -> Result<Vec<String>> {
314 #[derive(Deserialize)]
315 struct Row {
316 namespace: String,
317 }
318 Ok(self
319 .db
320 .prepare(
321 "SELECT DISTINCT namespace FROM repos
322 WHERE owner_id = ? AND is_private = 0 AND fork_of IS NULL AND deleted_at IS NULL
323 ORDER BY namespace",
324 )
325 .bind(&[owner_id.into()])?
326 .all()
327 .await?
328 .results::<Row>()?
329 .into_iter()
330 .map(|row| row.namespace)
331 .collect())
332 }
333
334 /// Repositories that are not forks, by id, a page at a time.
335 pub async fn ids_after(&self, after: Option<&str>, limit: u32) -> Result<Vec<String>> {
336 #[derive(Deserialize)]
337 struct Row {
338 id: String,
339 }
340 Ok(self
341 .db
342 .prepare("SELECT id FROM repos WHERE fork_of IS NULL AND deleted_at IS NULL AND id > ? ORDER BY id LIMIT ?")
343 .bind(&[after.unwrap_or("").into(), limit.into()])?
344 .all()
345 .await?
346 .results::<Row>()?
347 .into_iter()
348 .map(|row| row.id)
349 .collect())
350 }
351
352 /// Adds a pushed pack's bytes to what the repository is counted as
353 /// holding: its own, or, for a pull request's working copy, the
354 /// repository it is a copy of, whose storage it is.
355 pub async fn add_stored_bytes(&self, repo: &Repo, bytes: u64) -> Result<()> {
356 if bytes == 0 {
357 return Ok(());
358 }
359 let root = repo.fork_of.as_deref().unwrap_or(&repo.id);
360 self.db
361 .prepare("UPDATE repos SET stored_bytes = stored_bytes + ? WHERE id = ?")
362 .bind(&[(bytes as f64).into(), root.into()])?
363 .run()
364 .await?;
365 Ok(())
366 }
367
368 /// Which of these `namespace/name` paths are private. A working copy
369 /// answers as its repository. Unknown paths are left out.
370 pub async fn visibility(&self, paths: &[String]) -> Result<Vec<g1t_contracts::repos::RepoVisibility>> {
371 let mut out = Vec::new();
372 for path in paths.iter().take(50) {
373 let Some((namespace, name)) = path.split_once('/') else { continue };
374 let Some(repo) = self
375 .by_path(&RepoPath { namespace: namespace.to_owned(), name: name.to_owned() })
376 .await?
377 else {
378 continue;
379 };
380 let is_private = match &repo.fork_of {
381 Some(parent) => self.by_id(parent).await?.map_or(repo.is_private, |parent| parent.is_private),
382 None => repo.is_private,
383 };
384 out.push(g1t_contracts::repos::RepoVisibility { path: path.clone(), is_private });
385 }
386 Ok(out)
387 }
388
389 /// What each workspace's repositories are counted as holding, private
390 /// and public apart. Working copies count toward their repository.
391 pub async fn storage(&self) -> Result<Vec<g1t_contracts::repos::WorkspaceStorage>> {
392 #[derive(Deserialize)]
393 struct Row {
394 namespace: String,
395 private_bytes: Option<f64>,
396 public_bytes: Option<f64>,
397 }
398 Ok(self
399 .db
400 .prepare(
401 "SELECT namespace,
402 SUM(CASE WHEN is_private = 1 THEN stored_bytes ELSE 0 END) AS private_bytes,
403 SUM(CASE WHEN is_private = 0 THEN stored_bytes ELSE 0 END) AS public_bytes
404 FROM repos WHERE fork_of IS NULL AND deleted_at IS NULL AND stored_bytes > 0 GROUP BY namespace",
405 )
406 .all()
407 .await?
408 .results::<Row>()?
409 .into_iter()
410 .map(|row| g1t_contracts::repos::WorkspaceStorage {
411 namespace: row.namespace,
412 private_bytes: row.private_bytes.unwrap_or(0.0) as i64,
413 public_bytes: row.public_bytes.unwrap_or(0.0) as i64,
414 })
415 .collect())
416 }
417
418 /// What one workspace's private repositories are counted as holding.
419 pub async fn private_bytes(&self, namespace: &str) -> Result<i64> {
420 #[derive(Deserialize)]
421 struct Row {
422 bytes: Option<f64>,
423 }
424 Ok(self
425 .db
426 .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")
427 .bind(&[namespace.into()])?
428 .first::<Row>(None)
429 .await?
430 .and_then(|row| row.bytes)
431 .unwrap_or(0.0) as i64)
432 }
433
434 /// Forgets a repository that could not be filled.
435 pub async fn remove(&self, id: &str) -> Result<()> {
436 self.db
437 .prepare("DELETE FROM repos WHERE id = ?")
438 .bind(&[id.into()])?
439 .run()
440 .await?;
441 Ok(())
442 }
443
444 /// Picks the store key for a repository about to be made, and
445 /// remembers it: the one its path gives, unless a repository already
446 /// holds that (one made in a workspace that has since been renamed,
447 /// whose old name this workspace now has), when its id.
448 pub async fn claim_store_key(&self, repo: &Repo) -> Result<String> {
449 let wanted = path_key(repo);
450 let held = self
451 .db
452 .prepare("SELECT 1 AS held FROM repos WHERE store = ?")
453 .bind(&[wanted.as_str().into()])?
454 .first::<serde_json::Value>(None)
455 .await?
456 .is_some();
457 let key = if held { repo.id.clone() } else { wanted };
458 remember_store(repo, &key);
459 Ok(key)
460 }
461
462 /// Moves a renamed workspace's repositories to its current slug, from
463 /// any of `stale`. A repository whose name the current slug already has
464 /// (one pushed there in the moment before this ran) stays where it is;
465 /// returns how many did.
466 pub async fn rename_namespace(&self, stale: &[String], current: &str) -> Result<usize> {
467 if stale.is_empty() {
468 return Ok(0);
469 }
470 let marks = vec!["?"; stale.len()].join(", ");
471 let mut moved: Vec<JsValue> = vec![current.into()];
472 moved.extend(stale.iter().map(|slug| JsValue::from(slug.as_str())));
473 let left: Vec<JsValue> = stale.iter().map(|slug| JsValue::from(slug.as_str())).collect();
474 let results = self
475 .db
476 .batch(vec![
477 self.db
478 .prepare(format!(
479 "UPDATE OR IGNORE repos SET namespace = ? WHERE namespace IN ({marks})"
480 ))
481 .bind(&moved)?,
482 self.db
483 .prepare(format!(
484 "SELECT count(*) AS left FROM repos WHERE namespace IN ({marks})"
485 ))
486 .bind(&left)?,
487 // Git operations follow the workspace, added together.
488 self.db
489 .prepare(format!(
490 "INSERT INTO git_operations (namespace, hour, operations)
491 SELECT ?, hour, SUM(operations) FROM git_operations WHERE namespace IN ({marks}) GROUP BY hour
492 ON CONFLICT (namespace, hour) DO UPDATE SET operations = git_operations.operations + excluded.operations"
493 ))
494 .bind(&moved)?,
495 self.db
496 .prepare(format!("DELETE FROM git_operations WHERE namespace IN ({marks})"))
497 .bind(&left)?,
498 // Paths repositories were transferred away from follow the
499 // workspace too, so the old slug's redirect then finds them.
500 self.db
501 .prepare(format!(
502 "UPDATE OR IGNORE repo_redirects SET namespace = ? WHERE namespace IN ({marks})"
503 ))
504 .bind(&moved)?,
505 ])
506 .await?;
507 #[derive(Deserialize)]
508 struct Left {
509 left: usize,
510 }
511 Ok(results
512 .get(1)
513 .map(|result| result.results::<Left>())
514 .transpose()?
515 .and_then(|rows| rows.into_iter().next())
516 .map_or(0, |row| row.left))
517 }
518
519 pub async fn insert(&self, repo: &Repo) -> Result<()> {
520 self.db
521 .prepare(
522 "INSERT INTO repos
523 (id, namespace, name, description, is_private, owner_id,
524 default_branch, fork_of, created_at, store)
525 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
526 )
527 .bind(&[
528 repo.id.as_str().into(),
529 repo.namespace.as_str().into(),
530 repo.name.as_str().into(),
531 optional(&repo.description),
532 (repo.is_private as u8).into(),
533 repo.owner_id.as_str().into(),
534 repo.default_branch.as_str().into(),
535 optional(&repo.fork_of),
536 repo.created_at.as_str().into(),
537 store_key(repo).into(),
538 ])?
539 .run()
540 .await?;
541 Ok(())
542 }
543}
544
545#[cfg(test)]
546mod tests {
547 use super::*;
548 use g1t_contracts::access::{BasePermission, RepoGrant};
549 use g1t_contracts::{Membership, Role, User};
550
551 fn repo(private: bool) -> Repo {
552 Repo {
553 id: "rep_1".into(),
554 namespace: "acme".into(),
555 name: "rocket".into(),
556 description: None,
557 is_private: private,
558 owner_id: "usr_owner".into(),
559 default_branch: "main".into(),
560 fork_of: None,
561 protected: false,
562 created_at: String::new(),
563 topics: Vec::new(),
564 website: None,
565 archived_at: None,
566 }
567 }
568
569 fn person(id: &str, memberships: Vec<Membership>, grants: Vec<(&str, RepoRole)>) -> Viewer {
570 Some(User {
571 id: id.into(),
572 username: id.into(),
573 verified: true,
574 workspaces: memberships,
575 grants: grants
576 .into_iter()
577 .map(|(repo_id, role)| RepoGrant { repo_id: repo_id.into(), workspace: "acme".into(), role })
578 .collect(),
579 ..User::default()
580 })
581 }
582
583 /// What git asks: clone and fetch need Read on a private repository,
584 /// push needs Write.
585 #[test]
586 fn git_reads_with_read_and_pushes_with_write() {
587 let private = repo(true);
588 let reader = person("usr_r", vec![], vec![("rep_1", RepoRole::Read)]);
589 assert!(can_read(&private, &reader));
590 assert!(!can_write(&private, &reader));
591 let writer = person("usr_w", vec![], vec![("rep_1", RepoRole::Write)]);
592 assert!(can_read(&private, &writer) && can_write(&private, &writer));
593 let stranger = person("usr_s", vec![], vec![("rep_2", RepoRole::Admin)]);
594 assert!(!can_read(&private, &stranger) && !can_write(&private, &stranger));
595 assert!(!can_read(&private, &None));
596 // A public repository: anyone clones, nobody without Write pushes.
597 let public = repo(false);
598 assert!(can_read(&public, &None) && !can_write(&public, &None));
599 assert!(can_read(&public, &stranger) && !can_write(&public, &stranger));
600 }
601
602 #[test]
603 fn members_follow_the_base_permission_and_owners_have_admin() {
604 let private = repo(true);
605 let default_member = person("usr_m", vec![Membership::member("acme")], vec![]);
606 assert!(can_write(&private, &default_member));
607 assert!(!can(&private, &default_member, Capability::ManageIntegrations));
608 let none = Membership { base_permission: Some(BasePermission::None), ..Membership::member("acme") };
609 let locked_out = person("usr_n", vec![none.clone()], vec![]);
610 assert!(!can_read(&private, &locked_out));
611 let given = person("usr_g", vec![none], vec![("rep_1", RepoRole::Triage)]);
612 assert!(can_read(&private, &given) && !can_write(&private, &given));
613 let owner = person("usr_o", vec![Membership { role: Role::Owner, ..Membership::member("acme") }], vec![]);
614 assert_eq!(role(&private, &owner), Some(RepoRole::Admin));
615 assert!(can(&private, &owner, Capability::Delete));
616 }
617
618 #[test]
619 fn a_pull_requests_fork_is_its_authors() {
620 let fork = Repo {
621 namespace: "pulls".into(),
622 fork_of: Some("rep_1".into()),
623 owner_id: "usr_a".into(),
624 ..repo(true)
625 };
626 let author = person("usr_a", vec![], vec![]);
627 assert!(can_write(&fork, &author));
628 assert!(!can(&fork, &author, Capability::ManageSettings));
629 let other = person("usr_b", vec![Membership::member("acme")], vec![]);
630 assert!(!can_write(&fork, &other));
631 }
632}