Sign in

flagon-io/g1t

public

Git for AI scale: a forge for thousands of agents working on the same code at once.

gitagentscloudflarerustforgemcp

g1t/services/repos/src/registry.rs

351 lines11,996 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Rust repos service with shipping; pull requests kept in the model1//! Repository metadata in D1.
2
Agents and memory, checks and conflicts, profiles, slug renames, custom domains3use std::cell::RefCell;
4use std::collections::HashMap;
5
Rust repos service with shipping; pull requests kept in the model6use g1t_contracts::Viewer;
7use g1t_contracts::repos::{Repo, RepoPath};
8use serde::Deserialize;
9use worker::wasm_bindgen::JsValue;
10use worker::{D1Database, Result};
11
12#[derive(Deserialize)]
13struct RepoRow {
14 id: String,
15 namespace: String,
16 name: String,
17 description: Option<String>,
18 is_private: u8,
19 owner_id: String,
20 default_branch: String,
21 fork_of: Option<String>,
Agents as a team: lifecycle, merge queue, billing and a new shell22 protected: u8,
RFC 3339 timestamps in identity and repos23 created_at: String,
Agents and memory, checks and conflicts, profiles, slug renames, custom domains24 /// Null only on rows written before the column existed and not yet
25 /// migrated; their key is the one worked out from the path.
26 #[serde(default)]
27 store: Option<String>,
Rust repos service with shipping; pull requests kept in the model28}
29
Agents and memory, checks and conflicts, profiles, slug renames, custom domains30thread_local! {
31 /// Store keys that differ from the one a repository's path gives: those
32 /// of repositories whose workspace was renamed after they were made.
33 /// Filled whenever a row is read or written, so every `Repo` this
34 /// service holds has its key here. A key never changes once given, so
35 /// requests sharing the isolate can share the map.
36 static MOVED: RefCell<HashMap<String, String>> = RefCell::new(HashMap::new());
37}
38
39/// The key a repository's path gives: what every repository was stored
40/// under before workspaces could be renamed.
41pub fn path_key(repo: &Repo) -> String {
42 format!("{}--{}", repo.namespace, repo.name)
43}
44
45/// Records where a repository is stored, when its path does not say.
46pub fn remember_store(repo: &Repo, store: &str) {
47 if store != path_key(repo) {
48 MOVED.with(|moved| moved.borrow_mut().insert(repo.id.clone(), store.to_owned()));
49 }
50}
51
Rust repos service with shipping; pull requests kept in the model52impl From<RepoRow> for Repo {
53 fn from(row: RepoRow) -> Self {
Agents and memory, checks and conflicts, profiles, slug renames, custom domains54 let repo = Repo {
Rust repos service with shipping; pull requests kept in the model55 id: row.id,
56 namespace: row.namespace,
57 name: row.name,
58 description: row.description,
59 is_private: row.is_private != 0,
60 owner_id: row.owner_id,
61 default_branch: row.default_branch,
62 fork_of: row.fork_of,
Agents as a team: lifecycle, merge queue, billing and a new shell63 protected: row.protected != 0,
RFC 3339 timestamps in identity and repos64 created_at: row.created_at,
Agents and memory, checks and conflicts, profiles, slug renames, custom domains65 };
66 if let Some(store) = &row.store {
67 remember_store(&repo, store);
Rust repos service with shipping; pull requests kept in the model68 }
Agents and memory, checks and conflicts, profiles, slug renames, custom domains69 repo
Rust repos service with shipping; pull requests kept in the model70 }
71}
72
73/// The key a repo is stored under in the git store.
74pub fn store_key(repo: &Repo) -> String {
Agents and memory, checks and conflicts, profiles, slug renames, custom domains75 MOVED
76 .with(|moved| moved.borrow().get(&repo.id).cloned())
77 .unwrap_or_else(|| path_key(repo))
Rust repos service with shipping; pull requests kept in the model78}
79
Members can read a private repository's pull request forks80/// Whether the viewer may read `repo`, going by the repository alone. A
81/// private pull request fork is also readable by whoever can read the
82/// repository it came from, which `Repos::may_read` checks.
Rust repos service with shipping; pull requests kept in the model83pub fn can_read(repo: &Repo, viewer: &Viewer) -> bool {
84 !repo.is_private || can_write(repo, viewer)
85}
86
Members can read a private repository's pull request forks87/// A repository belongs to its workspace, so any member may write to it. A
Issues and pull requests replace intents and attempts88/// pull request's fork belongs to whoever opened the pull request.
Rust repos service with shipping; pull requests kept in the model89pub fn can_write(repo: &Repo, viewer: &Viewer) -> bool {
Workspaces own repositories90 viewer.as_ref().is_some_and(|user| {
91 if repo.fork_of.is_some() {
92 user.id == repo.owner_id
93 } else {
94 user.is_member(&repo.namespace)
95 }
96 })
Rust repos service with shipping; pull requests kept in the model97}
98
99fn optional(value: &Option<String>) -> JsValue {
100 value.as_deref().map_or(JsValue::NULL, JsValue::from)
101}
102
103pub struct Registry {
104 pub db: D1Database,
105}
106
107impl Registry {
108 pub async fn by_path(&self, path: &RepoPath) -> Result<Option<Repo>> {
109 Ok(self
110 .db
111 .prepare("SELECT * FROM repos WHERE namespace = ? AND name = ?")
112 .bind(&[
113 path.namespace.to_lowercase().into(),
114 path.name.to_lowercase().into(),
115 ])?
116 .first::<RepoRow>(None)
117 .await?
118 .map(Repo::from))
119 }
120
Agents as a team: lifecycle, merge queue, billing and a new shell121 pub async fn update(
122 &self,
123 id: &str,
124 description: Option<&str>,
125 is_private: bool,
126 protected: bool,
127 ) -> Result<()> {
128 self.db
129 .prepare("UPDATE repos SET description = ?, is_private = ?, protected = ? WHERE id = ?")
130 .bind(&[
131 description.map_or(JsValue::NULL, JsValue::from),
132 u32::from(is_private).into(),
133 u32::from(protected).into(),
134 id.into(),
135 ])?
136 .run()
137 .await?;
138 Ok(())
139 }
140
Rust repos service with shipping; pull requests kept in the model141 pub async fn by_id(&self, id: &str) -> Result<Option<Repo>> {
142 Ok(self
143 .db
144 .prepare("SELECT * FROM repos WHERE id = ?")
145 .bind(&[id.into()])?
146 .first::<RepoRow>(None)
147 .await?
148 .map(Repo::from))
149 }
150
Issues and pull requests replace intents and attempts151 /// Repos the viewer may see, newest first. Excludes pull request forks.
Workspaces own repositories152 /// With `member_only`, only repos in the viewer's own workspaces.
Rust repos service with shipping; pull requests kept in the model153 pub async fn list(
154 &self,
155 viewer: &Viewer,
156 query: Option<&str>,
157 namespace: Option<&str>,
Workspaces own repositories158 member_only: bool,
Rust repos service with shipping; pull requests kept in the model159 ) -> Result<Vec<Repo>> {
Workspaces own repositories160 let workspaces: Vec<&str> = viewer
161 .iter()
162 .flat_map(|user| &user.workspaces)
163 .map(|membership| membership.slug.as_str())
164 .collect();
165 // An empty IN list is not valid SQL, so a viewer in no workspace
166 // gets a name no workspace can have.
167 let mut params: Vec<JsValue> = if workspaces.is_empty() {
168 vec!["".into()]
169 } else {
170 workspaces.iter().map(|slug| JsValue::from(*slug)).collect()
171 };
172 let mine = format!("namespace IN ({})", vec!["?"; params.len()].join(", "));
173 let mut conditions = vec![
174 "fork_of IS NULL".to_owned(),
175 if member_only {
176 mine
177 } else {
178 format!("(is_private = 0 OR {mine})")
179 },
180 ];
Rust repos service with shipping; pull requests kept in the model181 if let Some(namespace) = namespace {
Workspaces own repositories182 conditions.push("namespace = ?".to_owned());
Rust repos service with shipping; pull requests kept in the model183 params.push(namespace.to_lowercase().into());
184 }
185 if let Some(query) = query.map(str::trim).filter(|query| !query.is_empty()) {
Workspaces own repositories186 conditions
187 .push("(name LIKE ? ESCAPE '\\' OR description LIKE ? ESCAPE '\\')".to_owned());
Rust repos service with shipping; pull requests kept in the model188 // LIKE wildcards in the query are matched literally.
189 let escaped: String = query
190 .chars()
191 .flat_map(|c| match c {
192 '\\' | '%' | '_' => vec!['\\', c],
193 _ => vec![c],
194 })
195 .collect();
196 let pattern = format!("%{escaped}%");
197 params.push(pattern.as_str().into());
198 params.push(pattern.into());
199 }
200 let sql = format!(
201 "SELECT * FROM repos WHERE {} ORDER BY created_at DESC, id DESC LIMIT 50",
202 conditions.join(" AND ")
203 );
204 let rows = self
205 .db
206 .prepare(sql)
207 .bind(&params)?
208 .all()
209 .await?
210 .results::<RepoRow>()?;
211 Ok(rows.into_iter().map(Repo::from).collect())
212 }
213
Agents and memory, checks and conflicts, profiles, slug renames, custom domains214 /// Of these ids, the repositories (not forks) the viewer may read.
215 pub async fn readable(&self, ids: &[String], viewer: &Viewer) -> Result<Vec<Repo>> {
216 let ids: Vec<&String> = ids.iter().take(g1t_contracts::repos::MAX_READABLE).collect();
217 if ids.is_empty() {
218 return Ok(Vec::new());
219 }
220 // One parameter however many ids: D1 binds at most 100.
221 let rows = self
222 .db
223 .prepare(
224 "SELECT * FROM repos
225 WHERE id IN (SELECT value FROM json_each(?)) AND fork_of IS NULL",
226 )
227 .bind(&[serde_json::to_string(&ids)?.into()])?
228 .all()
229 .await?
230 .results::<RepoRow>()?;
231 Ok(rows
232 .into_iter()
233 .map(Repo::from)
234 .filter(|repo| can_read(repo, viewer))
235 .collect())
236 }
237
238 /// The workspaces in which this account made a public repository.
239 pub async fn public_namespaces(&self, owner_id: &str) -> Result<Vec<String>> {
240 #[derive(Deserialize)]
241 struct Row {
242 namespace: String,
243 }
244 Ok(self
245 .db
246 .prepare(
247 "SELECT DISTINCT namespace FROM repos
248 WHERE owner_id = ? AND is_private = 0 AND fork_of IS NULL
249 ORDER BY namespace",
250 )
251 .bind(&[owner_id.into()])?
252 .all()
253 .await?
254 .results::<Row>()?
255 .into_iter()
256 .map(|row| row.namespace)
257 .collect())
258 }
259
Agents as a team: lifecycle, merge queue, billing and a new shell260 /// Forgets a repository that could not be filled.
261 pub async fn remove(&self, id: &str) -> Result<()> {
262 self.db
263 .prepare("DELETE FROM repos WHERE id = ?")
264 .bind(&[id.into()])?
265 .run()
266 .await?;
267 Ok(())
268 }
269
Agents and memory, checks and conflicts, profiles, slug renames, custom domains270 /// Picks the store key for a repository about to be made, and
271 /// remembers it: the one its path gives, unless a repository already
272 /// holds that (one made in a workspace that has since been renamed,
273 /// whose old name this workspace now has), when its id.
274 pub async fn claim_store_key(&self, repo: &Repo) -> Result<String> {
275 let wanted = path_key(repo);
276 let held = self
277 .db
278 .prepare("SELECT 1 AS held FROM repos WHERE store = ?")
279 .bind(&[wanted.as_str().into()])?
280 .first::<serde_json::Value>(None)
281 .await?
282 .is_some();
283 let key = if held { repo.id.clone() } else { wanted };
284 remember_store(repo, &key);
285 Ok(key)
286 }
287
288 /// Moves a renamed workspace's repositories to its current slug, from
289 /// any of `stale`. A repository whose name the current slug already has
290 /// (one pushed there in the moment before this ran) stays where it is;
291 /// returns how many did.
292 pub async fn rename_namespace(&self, stale: &[String], current: &str) -> Result<usize> {
293 if stale.is_empty() {
294 return Ok(0);
295 }
296 let marks = vec!["?"; stale.len()].join(", ");
297 let mut moved: Vec<JsValue> = vec![current.into()];
298 moved.extend(stale.iter().map(|slug| JsValue::from(slug.as_str())));
299 let left: Vec<JsValue> = stale.iter().map(|slug| JsValue::from(slug.as_str())).collect();
300 let results = self
301 .db
302 .batch(vec![
303 self.db
304 .prepare(format!(
305 "UPDATE OR IGNORE repos SET namespace = ? WHERE namespace IN ({marks})"
306 ))
307 .bind(&moved)?,
308 self.db
309 .prepare(format!(
310 "SELECT count(*) AS left FROM repos WHERE namespace IN ({marks})"
311 ))
312 .bind(&left)?,
313 ])
314 .await?;
315 #[derive(Deserialize)]
316 struct Left {
317 left: usize,
318 }
319 Ok(results
320 .get(1)
321 .map(|result| result.results::<Left>())
322 .transpose()?
323 .and_then(|rows| rows.into_iter().next())
324 .map_or(0, |row| row.left))
325 }
326
Rust repos service with shipping; pull requests kept in the model327 pub async fn insert(&self, repo: &Repo) -> Result<()> {
328 self.db
329 .prepare(
330 "INSERT INTO repos
331 (id, namespace, name, description, is_private, owner_id,
Agents and memory, checks and conflicts, profiles, slug renames, custom domains332 default_branch, fork_of, created_at, store)
333 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
Rust repos service with shipping; pull requests kept in the model334 )
335 .bind(&[
336 repo.id.as_str().into(),
337 repo.namespace.as_str().into(),
338 repo.name.as_str().into(),
339 optional(&repo.description),
340 (repo.is_private as u8).into(),
341 repo.owner_id.as_str().into(),
342 repo.default_branch.as_str().into(),
343 optional(&repo.fork_of),
RFC 3339 timestamps in identity and repos344 repo.created_at.as_str().into(),
Agents and memory, checks and conflicts, profiles, slug renames, custom domains345 store_key(repo).into(),
Rust repos service with shipping; pull requests kept in the model346 ])?
347 .run()
348 .await?;
349 Ok(())
350 }
351}