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/store.rs

540 lines20,100 bytesCodeBlame
1//! The storage that actually holds git repositories.
2//!
3//! The service depends on the [`GitStore`] and [`GitRepo`] ports;
4//! [`ArtifactsStore`] is the adapter for Cloudflare Artifacts.
5
6use g1t_contracts::repos::{Branch, Commit, EntryKind, GitAccess, Signature, TreeEntry};
7use g1t_contracts::time::rfc3339;
8use g1t_kit::js;
9use serde::{Deserialize, Serialize};
10use std::cell::RefCell;
11use std::collections::HashMap;
12use std::rc::Rc;
13use worker::js_sys::{Reflect, Uint8Array};
14use worker::wasm_bindgen::{JsCast, JsValue};
15use worker::{Env, Result};
16
17/// How long a credential handed to git stays valid.
18const TOKEN_TTL_SECONDS: u32 = 300;
19/// The same, in milliseconds.
20pub const CREDENTIAL_LIFE_MS: u64 = TOKEN_TTL_SECONDS as u64 * 1000;
21/// How long a credential is reused for, so that every one used has at
22/// least two minutes left. Credentials never leave this service: g1t has
23/// already decided who may do what before one is used.
24const TOKEN_REUSE_MS: u64 = 180_000;
25
26#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
27pub enum Scope {
28 Read,
29 Write,
30}
31
32impl Scope {
33 fn as_str(self) -> &'static str {
34 match self {
35 Scope::Read => "read",
36 Scope::Write => "write",
37 }
38 }
39}
40
41/// Where a credential handed out came from, for `Server-Timing`.
42#[derive(Clone, Copy, PartialEq, Eq, Debug)]
43pub enum Kept {
44 /// This isolate made it, or had it from another, a moment ago.
45 Isolate,
46 /// Another isolate made it and shared it.
47 Shared,
48}
49
50impl Kept {
51 pub fn as_str(self) -> &'static str {
52 match self {
53 Kept::Isolate => "isolate",
54 Kept::Shared => "shared",
55 }
56 }
57}
58
59/// A place repositories live. `key` is the store's own name for a repo.
60#[allow(async_fn_in_trait)]
61pub trait GitStore {
62 type Repo: GitRepo;
63
64 /// Creates an empty repository. Succeeds if it already exists.
65 async fn create(
66 &self,
67 key: &str,
68 description: Option<&str>,
69 default_branch: &str,
70 ) -> Result<()>;
71 async fn open(&self, key: &str) -> Result<Self::Repo>;
72 /// A credential for `key` made a moment ago, if the store keeps one.
73 async fn kept_access(&self, _key: &str, _scope: Scope) -> Option<(GitAccess, Kept)> {
74 None
75 }
76 /// A new credential for `key`, which the store may keep for next time.
77 async fn mint_access(&self, key: &str, scope: Scope) -> Result<GitAccess> {
78 self.open(key).await?.access(scope).await
79 }
80 /// A remote URL and credential for git itself, for the repository at
81 /// `key`. A store may hand out one it made a moment ago.
82 async fn access(&self, key: &str, scope: Scope) -> Result<GitAccess> {
83 match self.kept_access(key, scope).await {
84 Some((access, _)) => Ok(access),
85 None => self.mint_access(key, scope).await,
86 }
87 }
88 /// Stops handing out the credentials it keeps for `key`: the store
89 /// turned one down, or the repository is gone.
90 async fn forget_access(&self, _key: &str) {}
91 /// Removes a repository and everything in it, for good. Succeeds if it
92 /// is already gone.
93 async fn delete(&self, key: &str) -> Result<()>;
94}
95
96/// One open repository.
97#[allow(async_fn_in_trait)]
98pub trait GitRepo {
99 /// A remote URL and short-lived credential for git itself.
100 async fn access(&self, scope: Scope) -> Result<GitAccess>;
101 /// Every branch and the commit it points to.
102 async fn branches(&self) -> Result<Vec<Branch>>;
103 /// Newest first along the first-parent chain; empty for an unknown ref.
104 async fn log(&self, git_ref: &str, limit: u32) -> Result<Vec<Commit>>;
105 /// The parents of a commit, or `None` if the commit does not exist.
106 async fn parents(&self, commit_hash: &str) -> Result<Option<Vec<String>>>;
107 async fn read_tree(&self, tree_hash: &str) -> Result<Option<Vec<TreeEntry>>>;
108 async fn read_blob(&self, blob_hash: &str) -> Result<Option<Vec<u8>>>;
109 /// `None` when the ref or path does not resolve to a file.
110 async fn read_file(&self, git_ref: &str, path: &str) -> Result<Option<Vec<u8>>>;
111 /// Makes a copy-on-write copy of this repository under `target_key`.
112 async fn fork(&self, target_key: &str) -> Result<()>;
113}
114
115pub struct ArtifactsStore {
116 binding: JsValue,
117 /// Where isolates share the credentials they make; see shared.rs.
118 shared: Option<Rc<crate::shared::Shared>>,
119}
120
121impl ArtifactsStore {
122 pub fn new(env: &Env, shared: Option<Rc<crate::shared::Shared>>) -> Result<Self> {
123 Ok(Self {
124 binding: js::binding(env, "ARTIFACTS")?,
125 shared,
126 })
127 }
128}
129
130/// A credential as isolates share it, sealed (see shared.rs): with when it
131/// was made, so that one shared is reused no longer than one kept here.
132#[derive(Serialize, Deserialize)]
133struct SharedCredential {
134 remote: String,
135 token: String,
136 made: u64,
137}
138
139/// The shared cache's key for a credential: the store's key for the
140/// repository, and the scope.
141fn shared_key(key: &str, scope: Scope) -> String {
142 format!("cred:{key}:{}", scope.as_str())
143}
144
145/// A shared credential, if it was made less than [`TOKEN_REUSE_MS`] before
146/// `now`; with when it was made.
147fn shared_credential(bytes: &[u8], now: u64) -> Option<(GitAccess, u64)> {
148 let kept: SharedCredential = serde_json::from_slice(bytes).ok()?;
149 (now.saturating_sub(kept.made) < TOKEN_REUSE_MS).then_some((
150 GitAccess {
151 remote: kept.remote,
152 token: kept.token,
153 },
154 kept.made,
155 ))
156}
157
158/// Credentials made in the last few minutes, by repository and scope.
159/// Making one is a round trip to the store on every git request; reusing
160/// it saves that, and the store's lookup of the repository with it.
161#[derive(Default)]
162pub struct Credentials {
163 kept: HashMap<(String, Scope), (GitAccess, u64)>,
164}
165
166impl Credentials {
167 /// One made for `key` and `scope` less than [`TOKEN_REUSE_MS`] before `now`.
168 pub fn get(&self, key: &str, scope: Scope, now: u64) -> Option<GitAccess> {
169 self.kept
170 .get(&(key.to_owned(), scope))
171 .filter(|(_, made)| now.saturating_sub(*made) < TOKEN_REUSE_MS)
172 .map(|(access, _)| access.clone())
173 }
174
175 pub fn keep(&mut self, key: &str, scope: Scope, access: GitAccess, now: u64) {
176 // Expired ones go first, so the map stays as small as the isolate's
177 // recent repositories.
178 self.kept
179 .retain(|_, (_, made)| now.saturating_sub(*made) < TOKEN_REUSE_MS);
180 self.kept.insert((key.to_owned(), scope), (access, now));
181 }
182
183 pub fn forget(&mut self, key: &str) {
184 self.kept.retain(|(kept, _), _| kept != key);
185 }
186}
187
188thread_local! {
189 static CREDENTIALS: RefCell<Credentials> = RefCell::new(Credentials::default());
190}
191
192impl GitStore for ArtifactsStore {
193 type Repo = ArtifactsRepo;
194
195 /// One kept in this isolate, else one another isolate shared. A shared
196 /// one is kept here only for the rest of its own reuse window.
197 async fn kept_access(&self, key: &str, scope: Scope) -> Option<(GitAccess, Kept)> {
198 let now = g1t_kit::now_ms();
199 if let Some(access) = CREDENTIALS.with(|kept| kept.borrow().get(key, scope, now)) {
200 return Some((access, Kept::Isolate));
201 }
202 let bytes = self.shared.as_ref()?.get(&shared_key(key, scope)).await?;
203 let (access, made) = shared_credential(&bytes, now)?;
204 CREDENTIALS.with(|kept| kept.borrow_mut().keep(key, scope, access.clone(), made));
205 Some((access, Kept::Shared))
206 }
207
208 /// Made by the store, then kept here and shared with other isolates.
209 async fn mint_access(&self, key: &str, scope: Scope) -> Result<GitAccess> {
210 let now = g1t_kit::now_ms();
211 let access = self.open(key).await?.access(scope).await?;
212 CREDENTIALS.with(|kept| kept.borrow_mut().keep(key, scope, access.clone(), now));
213 if let Some(shared) = &self.shared {
214 let value = SharedCredential {
215 remote: access.remote.clone(),
216 token: access.token.clone(),
217 made: now,
218 };
219 if let Ok(bytes) = serde_json::to_vec(&value) {
220 shared
221 .put(&shared_key(key, scope), &bytes, TOKEN_REUSE_MS / 1000)
222 .await;
223 }
224 }
225 Ok(access)
226 }
227
228 async fn forget_access(&self, key: &str) {
229 CREDENTIALS.with(|kept| kept.borrow_mut().forget(key));
230 if let Some(shared) = &self.shared {
231 futures_util::future::join(
232 shared.delete(&shared_key(key, Scope::Read)),
233 shared.delete(&shared_key(key, Scope::Write)),
234 )
235 .await;
236 }
237 }
238
239 async fn create(
240 &self,
241 key: &str,
242 description: Option<&str>,
243 default_branch: &str,
244 ) -> Result<()> {
245 let options = js::to_js(&serde_json::json!({
246 "description": description,
247 "setDefaultBranch": default_branch,
248 }))?;
249 match js::call(&self.binding, "create", &[key.into(), options]).await {
250 // Left behind by an earlier failed attempt; adopt it.
251 Err(thrown) if !thrown.is("ALREADY_EXISTS") => Err(thrown.into()),
252 _ => Ok(()),
253 }
254 }
255
256 async fn delete(&self, key: &str) -> Result<()> {
257 self.forget_access(key).await;
258 match js::call(&self.binding, "delete", &[key.into()]).await {
259 // Gone already: an earlier purge got this far.
260 Err(thrown) if !thrown.is("NOT_FOUND") => Err(thrown.into()),
261 _ => Ok(()),
262 }
263 }
264
265 async fn open(&self, key: &str) -> Result<ArtifactsRepo> {
266 Ok(ArtifactsRepo {
267 handle: js::call(&self.binding, "get", &[key.into()]).await?,
268 key: key.to_owned(),
269 })
270 }
271}
272
273/// A handle to one Artifacts repository. It is an RPC stub, so it is
274/// released when dropped.
275pub struct ArtifactsRepo {
276 handle: JsValue,
277 /// The repository's store key, which scopes its cached objects.
278 key: String,
279}
280
281/// Where cached git objects live. Trees and blobs are named by their
282/// content, so a cached one is never stale; each is kept under its own
283/// repository's key, so a repository only ever finds its own objects.
284const OBJECT_CACHE: &str = "https://objects.g1t.internal/";
285/// Blobs larger than this are not cached.
286const MAX_CACHED_BLOB: usize = 1024 * 1024;
287const OBJECT_MAX_AGE: &str = "public, max-age=31536000, immutable";
288
289impl ArtifactsRepo {
290 fn cache_url(&self, kind: &str, hash: &str) -> String {
291 format!("{OBJECT_CACHE}{}/{kind}/{hash}", self.key)
292 }
293
294 async fn cached(&self, kind: &str, hash: &str) -> Option<Vec<u8>> {
295 let mut response = worker::Cache::default()
296 .get(self.cache_url(kind, hash), false)
297 .await
298 .ok()??;
299 response.bytes().await.ok()
300 }
301
302 /// Keeps an object for next time. A failure only costs a later read.
303 async fn keep(&self, kind: &str, hash: &str, bytes: Vec<u8>) {
304 let Ok(mut response) = worker::Response::from_bytes(bytes) else {
305 return;
306 };
307 let _ = response.headers_mut().set("cache-control", OBJECT_MAX_AGE);
308 let _ = worker::Cache::default()
309 .put(self.cache_url(kind, hash), response)
310 .await;
311 }
312}
313
314impl Drop for ArtifactsRepo {
315 fn drop(&mut self) {
316 let symbol = js::get(&worker::js_sys::global(), "Symbol");
317 let dispose = js::get(&symbol, "dispose");
318 if let Ok(function) = Reflect::get(&self.handle, &dispose)
319 .and_then(|value| value.dyn_into::<worker::js_sys::Function>())
320 {
321 let _ = function.call0(&self.handle);
322 }
323 }
324}
325
326#[derive(Deserialize)]
327#[serde(rename_all = "camelCase")]
328struct RawCommit {
329 hash: String,
330 tree_hash: String,
331 message: String,
332 author: Signature,
333 parents: Vec<String>,
334 /// Seconds since the epoch.
335 authored_at: u64,
336}
337
338#[derive(Deserialize)]
339struct RawEntry {
340 name: String,
341 hash: String,
342 #[serde(rename = "type")]
343 kind: EntryKind,
344}
345
346#[derive(Deserialize)]
347struct RawInfo {
348 remote: String,
349}
350
351#[derive(Deserialize)]
352struct RawToken {
353 plaintext: String,
354}
355
356/// The bytes of a `Blob`, or `None` for null.
357async fn blob_bytes(blob: JsValue) -> Result<Option<Vec<u8>>> {
358 if blob.is_null() || blob.is_undefined() {
359 return Ok(None);
360 }
361 let buffer = js::call(&blob, "arrayBuffer", &[]).await?;
362 Ok(Some(Uint8Array::new(&buffer).to_vec()))
363}
364
365impl GitRepo for ArtifactsRepo {
366 async fn access(&self, scope: Scope) -> Result<GitAccess> {
367 let scope = scope.as_str();
368 // Two round trips to the store, at once.
369 let (info, token) = futures_util::future::join(
370 js::call(&self.handle, "info", &[]),
371 js::call(
372 &self.handle,
373 "createToken",
374 &[scope.into(), TOKEN_TTL_SECONDS.into()],
375 ),
376 )
377 .await;
378 let info: RawInfo = js::from_js(&info?)?;
379 let token: RawToken = js::from_js(&token?)?;
380 Ok(GitAccess {
381 remote: info.remote,
382 token: token.plaintext,
383 })
384 }
385
386 async fn branches(&self) -> Result<Vec<Branch>> {
387 crate::refs::branches(&self.access(Scope::Read).await?).await
388 }
389
390 async fn log(&self, git_ref: &str, limit: u32) -> Result<Vec<Commit>> {
391 let options = js::to_js(&serde_json::json!({ "ref": git_ref, "limit": limit }))?;
392 let commits: Vec<RawCommit> =
393 js::from_js(&js::call(&self.handle, "log", &[options]).await?)?;
394 Ok(commits
395 .into_iter()
396 .map(|commit| Commit {
397 hash: commit.hash,
398 tree_hash: commit.tree_hash,
399 message: commit.message,
400 author: commit.author,
401 parents: commit.parents,
402 authored_at: rfc3339(commit.authored_at * 1000),
403 })
404 .collect())
405 }
406
407 async fn parents(&self, commit_hash: &str) -> Result<Option<Vec<String>>> {
408 let commit: Option<RawCommit> =
409 js::from_js(&js::call(&self.handle, "readCommit", &[commit_hash.into()]).await?)?;
410 Ok(commit.map(|commit| commit.parents))
411 }
412
413 async fn read_tree(&self, tree_hash: &str) -> Result<Option<Vec<TreeEntry>>> {
414 if let Some(bytes) = self.cached("tree", tree_hash).await
415 && let Ok(entries) = serde_json::from_slice::<Vec<TreeEntry>>(&bytes) {
416 return Ok(Some(entries));
417 }
418 let entries: Option<Vec<RawEntry>> =
419 js::from_js(&js::call(&self.handle, "readTree", &[tree_hash.into()]).await?)?;
420 let entries: Option<Vec<TreeEntry>> = entries.map(|entries| {
421 entries
422 .into_iter()
423 .map(|entry| TreeEntry {
424 name: entry.name,
425 hash: entry.hash,
426 kind: entry.kind,
427 })
428 .collect()
429 });
430 if let Some(entries) = &entries
431 && let Ok(bytes) = serde_json::to_vec(entries) {
432 self.keep("tree", tree_hash, bytes).await;
433 }
434 Ok(entries)
435 }
436
437 async fn read_blob(&self, blob_hash: &str) -> Result<Option<Vec<u8>>> {
438 if let Some(bytes) = self.cached("blob", blob_hash).await {
439 return Ok(Some(bytes));
440 }
441 let bytes = blob_bytes(js::call(&self.handle, "readBlob", &[blob_hash.into()]).await?).await?;
442 if let Some(bytes) = bytes.as_ref().filter(|bytes| bytes.len() <= MAX_CACHED_BLOB) {
443 self.keep("blob", blob_hash, bytes.clone()).await;
444 }
445 Ok(bytes)
446 }
447
448 async fn read_file(&self, git_ref: &str, path: &str) -> Result<Option<Vec<u8>>> {
449 let args = js::to_js(&serde_json::json!({ "ref": git_ref, "path": path }))?;
450 blob_bytes(js::call(&self.handle, "readFile", &[args]).await?).await
451 }
452
453 async fn fork(&self, target_key: &str) -> Result<()> {
454 let options = js::to_js(&serde_json::json!({ "defaultBranchOnly": true }))?;
455 match js::call(&self.handle, "fork", &[target_key.into(), options]).await {
456 Err(thrown) if !thrown.is("ALREADY_EXISTS") => Err(thrown.into()),
457 _ => Ok(()),
458 }
459 }
460}
461
462#[cfg(test)]
463mod tests {
464 use super::{Credentials, GitAccess, Scope, SharedCredential, TOKEN_REUSE_MS, shared_credential, shared_key};
465
466 fn access(token: &str) -> GitAccess {
467 GitAccess {
468 remote: "https://store.example/acme--rocket.git".to_owned(),
469 token: token.to_owned(),
470 }
471 }
472
473 #[test]
474 fn a_credential_is_reused_only_while_it_has_time_left() {
475 let mut kept = Credentials::default();
476 kept.keep("acme--rocket", Scope::Read, access("r1"), 1_000);
477 assert_eq!(kept.get("acme--rocket", Scope::Read, 1_000).unwrap().token, "r1");
478 assert_eq!(
479 kept.get("acme--rocket", Scope::Read, 1_000 + TOKEN_REUSE_MS - 1).unwrap().token,
480 "r1"
481 );
482 assert!(kept.get("acme--rocket", Scope::Read, 1_000 + TOKEN_REUSE_MS).is_none());
483 }
484
485 #[test]
486 fn a_credential_is_kept_for_its_own_repository_and_scope() {
487 let mut kept = Credentials::default();
488 kept.keep("acme--rocket", Scope::Read, access("r1"), 1_000);
489 // A read credential never stands in for a write one.
490 assert!(kept.get("acme--rocket", Scope::Write, 1_000).is_none());
491 assert!(kept.get("acme--booster", Scope::Read, 1_000).is_none());
492 kept.keep("acme--rocket", Scope::Write, access("w1"), 1_000);
493 assert_eq!(kept.get("acme--rocket", Scope::Write, 1_000).unwrap().token, "w1");
494 assert_eq!(kept.get("acme--rocket", Scope::Read, 1_000).unwrap().token, "r1");
495 }
496
497 #[test]
498 fn a_shared_credential_is_reused_only_in_its_own_window() {
499 let value = serde_json::to_vec(&SharedCredential {
500 remote: "https://store.example/acme--rocket.git".to_owned(),
501 token: "r1".to_owned(),
502 made: 10_000,
503 })
504 .unwrap();
505 let (access, made) = shared_credential(&value, 10_000 + TOKEN_REUSE_MS - 1).unwrap();
506 assert_eq!(access.token, "r1");
507 // Kept here only for what is left of its window, not a new one.
508 assert_eq!(made, 10_000);
509 assert!(shared_credential(&value, 10_000 + TOKEN_REUSE_MS).is_none());
510 // Anything else is a miss.
511 assert!(shared_credential(b"not json", 10_000).is_none());
512 // Each repository and scope has its own key.
513 assert_eq!(shared_key("acme--rocket", Scope::Read), "cred:acme--rocket:read");
514 assert_ne!(shared_key("acme--rocket", Scope::Read), shared_key("acme--rocket", Scope::Write));
515 }
516
517 #[test]
518 fn a_shared_credential_kept_here_expires_with_the_original() {
519 let mut kept = Credentials::default();
520 // Made at 1_000 elsewhere, found here at 100_000.
521 kept.keep("acme--rocket", Scope::Read, access("r1"), 1_000);
522 assert!(kept.get("acme--rocket", Scope::Read, 100_000).is_some());
523 assert!(kept.get("acme--rocket", Scope::Read, 1_000 + TOKEN_REUSE_MS).is_none());
524 }
525
526 #[test]
527 fn a_turned_down_credential_is_forgotten_and_old_ones_are_dropped() {
528 let mut kept = Credentials::default();
529 kept.keep("acme--rocket", Scope::Read, access("r1"), 1_000);
530 kept.keep("acme--rocket", Scope::Write, access("w1"), 1_000);
531 kept.keep("acme--booster", Scope::Read, access("b1"), 1_000);
532 kept.forget("acme--rocket");
533 assert!(kept.get("acme--rocket", Scope::Read, 1_000).is_none());
534 assert!(kept.get("acme--rocket", Scope::Write, 1_000).is_none());
535 assert!(kept.get("acme--booster", Scope::Read, 1_000).is_some());
536 // Keeping another later drops the expired one from the map.
537 kept.keep("acme--other", Scope::Read, access("o1"), 1_000 + TOKEN_REUSE_MS);
538 assert_eq!(kept.kept.len(), 1);
539 }
540}