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

379 lines12,980 bytesCodeBlame
1//! Repository metadata in D1.
2
3use std::cell::RefCell;
4use std::collections::HashMap;
5
6use 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>,
22 protected: u8,
23 created_at: String,
24 /// 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>,
28 /// JSON; absent on rows read before the column existed.
29 #[serde(default)]
30 topics: Option<String>,
31}
32
33thread_local! {
34 /// Store keys that differ from the one a repository's path gives: those
35 /// of repositories whose workspace was renamed after they were made.
36 /// Filled whenever a row is read or written, so every `Repo` this
37 /// service holds has its key here. A key never changes once given, so
38 /// requests sharing the isolate can share the map.
39 static MOVED: RefCell<HashMap<String, String>> = RefCell::new(HashMap::new());
40}
41
42/// The key a repository's path gives: what every repository was stored
43/// under before workspaces could be renamed.
44pub fn path_key(repo: &Repo) -> String {
45 format!("{}--{}", repo.namespace, repo.name)
46}
47
48/// Records where a repository is stored, when its path does not say.
49pub fn remember_store(repo: &Repo, store: &str) {
50 if store != path_key(repo) {
51 MOVED.with(|moved| moved.borrow_mut().insert(repo.id.clone(), store.to_owned()));
52 }
53}
54
55impl From<RepoRow> for Repo {
56 fn from(row: RepoRow) -> Self {
57 let repo = Repo {
58 id: row.id,
59 namespace: row.namespace,
60 name: row.name,
61 description: row.description,
62 is_private: row.is_private != 0,
63 owner_id: row.owner_id,
64 default_branch: row.default_branch,
65 fork_of: row.fork_of,
66 protected: row.protected != 0,
67 created_at: row.created_at,
68 topics: row
69 .topics
70 .as_deref()
71 .and_then(|topics| serde_json::from_str(topics).ok())
72 .unwrap_or_default(),
73 };
74 if let Some(store) = &row.store {
75 remember_store(&repo, store);
76 }
77 repo
78 }
79}
80
81/// The key a repo is stored under in the git store.
82pub fn store_key(repo: &Repo) -> String {
83 MOVED
84 .with(|moved| moved.borrow().get(&repo.id).cloned())
85 .unwrap_or_else(|| path_key(repo))
86}
87
88/// Whether the viewer may read `repo`, going by the repository alone. A
89/// private pull request fork is also readable by whoever can read the
90/// repository it came from, which `Repos::may_read` checks.
91pub fn can_read(repo: &Repo, viewer: &Viewer) -> bool {
92 !repo.is_private || can_write(repo, viewer)
93}
94
95/// A repository belongs to its workspace, so any member may write to it. A
96/// pull request's fork belongs to whoever opened the pull request.
97pub fn can_write(repo: &Repo, viewer: &Viewer) -> bool {
98 viewer.as_ref().is_some_and(|user| {
99 if repo.fork_of.is_some() {
100 user.id == repo.owner_id
101 } else {
102 user.is_member(&repo.namespace)
103 }
104 })
105}
106
107fn optional(value: &Option<String>) -> JsValue {
108 value.as_deref().map_or(JsValue::NULL, JsValue::from)
109}
110
111pub struct Registry {
112 pub db: D1Database,
113}
114
115impl Registry {
116 pub async fn by_path(&self, path: &RepoPath) -> Result<Option<Repo>> {
117 Ok(self
118 .db
119 .prepare("SELECT * FROM repos WHERE namespace = ? AND name = ?")
120 .bind(&[
121 path.namespace.to_lowercase().into(),
122 path.name.to_lowercase().into(),
123 ])?
124 .first::<RepoRow>(None)
125 .await?
126 .map(Repo::from))
127 }
128
129 pub async fn update(
130 &self,
131 id: &str,
132 description: Option<&str>,
133 is_private: bool,
134 protected: bool,
135 topics: &[String],
136 ) -> Result<()> {
137 self.db
138 .prepare("UPDATE repos SET description = ?, is_private = ?, protected = ?, topics = ? WHERE id = ?")
139 .bind(&[
140 description.map_or(JsValue::NULL, JsValue::from),
141 u32::from(is_private).into(),
142 u32::from(protected).into(),
143 serde_json::to_string(topics)?.into(),
144 id.into(),
145 ])?
146 .run()
147 .await?;
148 Ok(())
149 }
150
151 pub async fn by_id(&self, id: &str) -> Result<Option<Repo>> {
152 Ok(self
153 .db
154 .prepare("SELECT * FROM repos WHERE id = ?")
155 .bind(&[id.into()])?
156 .first::<RepoRow>(None)
157 .await?
158 .map(Repo::from))
159 }
160
161 /// Repos the viewer may see, newest first. Excludes pull request forks.
162 /// With `member_only`, only repos in the viewer's own workspaces.
163 pub async fn list(
164 &self,
165 viewer: &Viewer,
166 query: Option<&str>,
167 namespace: Option<&str>,
168 member_only: bool,
169 ) -> Result<Vec<Repo>> {
170 let workspaces: Vec<&str> = viewer
171 .iter()
172 .flat_map(|user| &user.workspaces)
173 .map(|membership| membership.slug.as_str())
174 .collect();
175 // An empty IN list is not valid SQL, so a viewer in no workspace
176 // gets a name no workspace can have.
177 let mut params: Vec<JsValue> = if workspaces.is_empty() {
178 vec!["".into()]
179 } else {
180 workspaces.iter().map(|slug| JsValue::from(*slug)).collect()
181 };
182 let mine = format!("namespace IN ({})", vec!["?"; params.len()].join(", "));
183 let mut conditions = vec![
184 "fork_of IS NULL".to_owned(),
185 if member_only {
186 mine
187 } else {
188 format!("(is_private = 0 OR {mine})")
189 },
190 ];
191 if let Some(namespace) = namespace {
192 conditions.push("namespace = ?".to_owned());
193 params.push(namespace.to_lowercase().into());
194 }
195 if let Some(query) = query.map(str::trim).filter(|query| !query.is_empty()) {
196 conditions
197 .push("(name LIKE ? ESCAPE '\\' OR description LIKE ? ESCAPE '\\')".to_owned());
198 // LIKE wildcards in the query are matched literally.
199 let escaped: String = query
200 .chars()
201 .flat_map(|c| match c {
202 '\\' | '%' | '_' => vec!['\\', c],
203 _ => vec![c],
204 })
205 .collect();
206 let pattern = format!("%{escaped}%");
207 params.push(pattern.as_str().into());
208 params.push(pattern.into());
209 }
210 let sql = format!(
211 "SELECT * FROM repos WHERE {} ORDER BY created_at DESC, id DESC LIMIT 50",
212 conditions.join(" AND ")
213 );
214 let rows = self
215 .db
216 .prepare(sql)
217 .bind(&params)?
218 .all()
219 .await?
220 .results::<RepoRow>()?;
221 Ok(rows.into_iter().map(Repo::from).collect())
222 }
223
224 /// Of these ids, the repositories (not forks) the viewer may read.
225 pub async fn readable(&self, ids: &[String], viewer: &Viewer) -> Result<Vec<Repo>> {
226 let ids: Vec<&String> = ids.iter().take(g1t_contracts::repos::MAX_READABLE).collect();
227 if ids.is_empty() {
228 return Ok(Vec::new());
229 }
230 // One parameter however many ids: D1 binds at most 100.
231 let rows = self
232 .db
233 .prepare(
234 "SELECT * FROM repos
235 WHERE id IN (SELECT value FROM json_each(?)) AND fork_of IS NULL",
236 )
237 .bind(&[serde_json::to_string(&ids)?.into()])?
238 .all()
239 .await?
240 .results::<RepoRow>()?;
241 Ok(rows
242 .into_iter()
243 .map(Repo::from)
244 .filter(|repo| can_read(repo, viewer))
245 .collect())
246 }
247
248 /// The workspaces in which this account made a public repository.
249 pub async fn public_namespaces(&self, owner_id: &str) -> Result<Vec<String>> {
250 #[derive(Deserialize)]
251 struct Row {
252 namespace: String,
253 }
254 Ok(self
255 .db
256 .prepare(
257 "SELECT DISTINCT namespace FROM repos
258 WHERE owner_id = ? AND is_private = 0 AND fork_of IS NULL
259 ORDER BY namespace",
260 )
261 .bind(&[owner_id.into()])?
262 .all()
263 .await?
264 .results::<Row>()?
265 .into_iter()
266 .map(|row| row.namespace)
267 .collect())
268 }
269
270 /// Repositories that are not forks, by id, a page at a time.
271 pub async fn ids_after(&self, after: Option<&str>, limit: u32) -> Result<Vec<String>> {
272 #[derive(Deserialize)]
273 struct Row {
274 id: String,
275 }
276 Ok(self
277 .db
278 .prepare("SELECT id FROM repos WHERE fork_of IS NULL AND id > ? ORDER BY id LIMIT ?")
279 .bind(&[after.unwrap_or("").into(), limit.into()])?
280 .all()
281 .await?
282 .results::<Row>()?
283 .into_iter()
284 .map(|row| row.id)
285 .collect())
286 }
287
288 /// Forgets a repository that could not be filled.
289 pub async fn remove(&self, id: &str) -> Result<()> {
290 self.db
291 .prepare("DELETE FROM repos WHERE id = ?")
292 .bind(&[id.into()])?
293 .run()
294 .await?;
295 Ok(())
296 }
297
298 /// Picks the store key for a repository about to be made, and
299 /// remembers it: the one its path gives, unless a repository already
300 /// holds that (one made in a workspace that has since been renamed,
301 /// whose old name this workspace now has), when its id.
302 pub async fn claim_store_key(&self, repo: &Repo) -> Result<String> {
303 let wanted = path_key(repo);
304 let held = self
305 .db
306 .prepare("SELECT 1 AS held FROM repos WHERE store = ?")
307 .bind(&[wanted.as_str().into()])?
308 .first::<serde_json::Value>(None)
309 .await?
310 .is_some();
311 let key = if held { repo.id.clone() } else { wanted };
312 remember_store(repo, &key);
313 Ok(key)
314 }
315
316 /// Moves a renamed workspace's repositories to its current slug, from
317 /// any of `stale`. A repository whose name the current slug already has
318 /// (one pushed there in the moment before this ran) stays where it is;
319 /// returns how many did.
320 pub async fn rename_namespace(&self, stale: &[String], current: &str) -> Result<usize> {
321 if stale.is_empty() {
322 return Ok(0);
323 }
324 let marks = vec!["?"; stale.len()].join(", ");
325 let mut moved: Vec<JsValue> = vec![current.into()];
326 moved.extend(stale.iter().map(|slug| JsValue::from(slug.as_str())));
327 let left: Vec<JsValue> = stale.iter().map(|slug| JsValue::from(slug.as_str())).collect();
328 let results = self
329 .db
330 .batch(vec![
331 self.db
332 .prepare(format!(
333 "UPDATE OR IGNORE repos SET namespace = ? WHERE namespace IN ({marks})"
334 ))
335 .bind(&moved)?,
336 self.db
337 .prepare(format!(
338 "SELECT count(*) AS left FROM repos WHERE namespace IN ({marks})"
339 ))
340 .bind(&left)?,
341 ])
342 .await?;
343 #[derive(Deserialize)]
344 struct Left {
345 left: usize,
346 }
347 Ok(results
348 .get(1)
349 .map(|result| result.results::<Left>())
350 .transpose()?
351 .and_then(|rows| rows.into_iter().next())
352 .map_or(0, |row| row.left))
353 }
354
355 pub async fn insert(&self, repo: &Repo) -> Result<()> {
356 self.db
357 .prepare(
358 "INSERT INTO repos
359 (id, namespace, name, description, is_private, owner_id,
360 default_branch, fork_of, created_at, store)
361 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
362 )
363 .bind(&[
364 repo.id.as_str().into(),
365 repo.namespace.as_str().into(),
366 repo.name.as_str().into(),
367 optional(&repo.description),
368 (repo.is_private as u8).into(),
369 repo.owner_id.as_str().into(),
370 repo.default_branch.as_str().into(),
371 optional(&repo.fork_of),
372 repo.created_at.as_str().into(),
373 store_key(repo).into(),
374 ])?
375 .run()
376 .await?;
377 Ok(())
378 }
379}