Skip to content
Sign up

flagon-io/g1t

public
0
562 linesCodeBlameRaw

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.

API and MCP server, Rust identity service, registration, site redesign1//! Plumbing shared by g1t services that run on Workers.
2//!
3//! Services talk to each other over service bindings with a small JSON
4//! protocol: `POST /rpc/<method>` with the method's arguments as the body,
5//! answered with the method's return value.
6
7use serde::Serialize;
8use serde::de::DeserializeOwned;
9use worker::{Date, Fetcher, Headers, Method, Request, RequestInit, Response, Result};
10
11/// The current time in milliseconds since the epoch.
12pub fn now_ms() -> u64 {
13 Date::now().as_millis()
14}
15
16/// The method name of an RPC request, or `None` if it is not one.
17pub fn rpc_method(request: &Request) -> Option<String> {
18 if request.method() != Method::Post {
19 return None;
20 }
21 request
22 .path()
23 .strip_prefix("/rpc/")
24 .map(|method| method.to_owned())
25}
26
27/// Deserializes a method's arguments.
28pub fn args<A: DeserializeOwned>(body: serde_json::Value) -> Result<A> {
29 serde_json::from_value(body)
30 .map_err(|error| worker::Error::RustError(format!("bad arguments: {error}")))
31}
32
33/// Serializes a method's return value as the response body.
34pub fn reply<R: Serialize>(value: &R) -> Result<Response> {
35 Response::from_json(value)
36}
37
38/// Calls `method` on another service through its binding.
39pub async fn call<A: Serialize, R: DeserializeOwned>(
40 service: &Fetcher,
41 method: &str,
42 arguments: &A,
43) -> Result<R> {
44 let headers = Headers::new();
45 headers.set("content-type", "application/json")?;
46 let mut init = RequestInit::new();
47 init.with_method(Method::Post)
48 .with_headers(headers)
49 .with_body(Some(serde_json::to_string(arguments)?.into()));
50 // The hostname is ignored; a service binding always reaches its service.
51 let request = Request::new_with_init(&format!("https://service/rpc/{method}"), &init)?;
52 let mut response = service.fetch_request(request).await?;
53 if response.status_code() != 200 {
54 return Err(worker::Error::RustError(format!(
55 "{method} failed with status {}: {}",
56 response.status_code(),
57 response.text().await.unwrap_or_default()
58 )));
59 }
60 response.json().await
61}
Email verification, password reset, and Git for AI scale positioning62
Fast pages, required checks on the branch, self-hosted runners, honest incidents63pub mod d1;
Merge branch 'worktree-agent-a8752162fea25f63f' into spend-guardrails64pub mod limits;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API65pub mod wire;
66
Email verification, password reset, and Git for AI scale positioning67/// Helpers for bindings that workers-rs has no typed wrapper for, such as
68/// Artifacts and Email Sending. Values cross the boundary as JSON.
69pub mod js {
Rust repos service with shipping; pull requests kept in the model70 use std::fmt;
71
Email verification, password reset, and Git for AI scale positioning72 use serde::Serialize;
73 use serde::de::DeserializeOwned;
Rust repos service with shipping; pull requests kept in the model74 use worker::js_sys::{Array, Function, JSON, Promise, Reflect};
Email verification, password reset, and Git for AI scale positioning75 use worker::wasm_bindgen::{JsCast, JsValue};
76 use worker::wasm_bindgen_futures::JsFuture;
77 use worker::{Env, Error, Result};
78
Rust repos service with shipping; pull requests kept in the model79 /// An exception thrown by JavaScript, with its `code` if it had one.
80 #[derive(Debug)]
81 pub struct Thrown {
82 pub code: Option<String>,
83 pub message: String,
84 }
85
86 impl Thrown {
87 fn from_value(value: JsValue) -> Self {
88 let property = |name: &str| {
89 Reflect::get(&value, &name.into())
Email verification, password reset, and Git for AI scale positioning90 .ok()
Rust repos service with shipping; pull requests kept in the model91 .and_then(|property| property.as_string())
92 };
93 Thrown {
94 code: property("code"),
95 message: property("message").unwrap_or_else(|| format!("{value:?}")),
96 }
97 }
98
99 pub fn is(&self, code: &str) -> bool {
100 self.code.as_deref() == Some(code)
101 }
Email verification, password reset, and Git for AI scale positioning102 }
103
Rust repos service with shipping; pull requests kept in the model104 impl fmt::Display for Thrown {
105 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
106 match &self.code {
107 Some(code) => write!(f, "{code}: {}", self.message),
108 None => f.write_str(&self.message),
109 }
110 }
111 }
112
113 impl From<Thrown> for Error {
114 fn from(thrown: Thrown) -> Self {
115 Error::RustError(thrown.to_string())
116 }
117 }
118
Email verification, password reset, and Git for AI scale positioning119 /// The binding called `name`, as a raw JavaScript value.
120 pub fn binding(env: &Env, name: &str) -> Result<JsValue> {
Rust repos service with shipping; pull requests kept in the model121 let value = Reflect::get(env.as_ref(), &name.into()).map_err(Thrown::from_value)?;
Email verification, password reset, and Git for AI scale positioning122 if value.is_undefined() {
123 return Err(Error::RustError(format!(
124 "binding {name} is not configured"
125 )));
126 }
127 Ok(value)
128 }
129
Rust repos service with shipping; pull requests kept in the model130 /// Reads a property of a JavaScript object.
131 pub fn get(target: &JsValue, name: &str) -> JsValue {
132 Reflect::get(target, &name.into()).unwrap_or(JsValue::UNDEFINED)
133 }
134
Events service in Rust, with RFC 3339 times and accurate push events135 /// Sets a property on an object.
136 pub fn set(target: &JsValue, name: &str, value: &JsValue) {
137 let _ = Reflect::set(target, &name.into(), value);
138 }
139
Email verification, password reset, and Git for AI scale positioning140 pub fn to_js<T: Serialize>(value: &T) -> Result<JsValue> {
Rust repos service with shipping; pull requests kept in the model141 Ok(JSON::parse(&serde_json::to_string(value)?).map_err(Thrown::from_value)?)
Email verification, password reset, and Git for AI scale positioning142 }
143
144 pub fn from_js<T: DeserializeOwned>(value: &JsValue) -> Result<T> {
Rust repos service with shipping; pull requests kept in the model145 let text = if value.is_undefined() {
146 None
147 } else {
148 JSON::stringify(value)
149 .map_err(Thrown::from_value)?
150 .as_string()
151 };
152 let text = text.as_deref().unwrap_or("null");
153 serde_json::from_str(text).map_err(|error| {
154 // Say what arrived; a bare serde error is useless in a log.
155 let seen: String = text.chars().take(300).collect();
156 Error::RustError(format!(
157 "unexpected value from JavaScript ({error}): {seen}"
158 ))
159 })
Email verification, password reset, and Git for AI scale positioning160 }
161
Rust repos service with shipping; pull requests kept in the model162 /// Calls `target[method](...args)` and awaits the result if it is a
163 /// thenable. `method` may be a name or a symbol.
164 pub async fn call_key(
165 target: &JsValue,
166 method: &JsValue,
167 args: &[JsValue],
168 ) -> std::result::Result<JsValue, Thrown> {
169 let function: Function = Reflect::get(target, method)
170 .map_err(Thrown::from_value)?
Email verification, password reset, and Git for AI scale positioning171 .dyn_into()
Rust repos service with shipping; pull requests kept in the model172 .map_err(|_| Thrown {
173 code: None,
174 message: format!("{method:?} is not a function"),
175 })?;
176 let arguments: Array = args.iter().collect();
177 // An RPC stub treats every property access as a remote method, so
178 // `function.apply(...)` would be sent over the wire as a call to
179 // "apply". Reflect.apply invokes the function without touching it.
180 let returned = Reflect::apply(&function, target, &arguments).map_err(Thrown::from_value)?;
181 // Worker RPC returns its own thenable rather than a Promise, so
182 // resolve whatever came back instead of testing its type.
183 JsFuture::from(Promise::resolve(&returned))
184 .await
185 .map_err(Thrown::from_value)
186 }
187
188 /// Calls `target.method(...args)`; see [`call_key`].
189 pub async fn call(
190 target: &JsValue,
191 method: &str,
192 args: &[JsValue],
193 ) -> std::result::Result<JsValue, Thrown> {
194 call_key(target, &method.into(), args).await
Email verification, password reset, and Git for AI scale positioning195 }
196}
Agents and memory, checks and conflicts, profiles, slug renames, custom domains197
198/// Moving a service's rows when a workspace is renamed.
199pub mod rename {
200 use std::collections::HashMap;
201
202 use g1t_contracts::events::{Event, WorkspaceRenamed};
203 use g1t_contracts::identity::UsernamesArgs;
204 use worker::wasm_bindgen::JsValue;
205 use worker::{D1Database, Env, Result};
206
207 /// Handles `workspace.renamed` with `statements`, and says whether
208 /// `event` was one. Each statement uses `?1` for the workspace's current
209 /// slug (asked of identity by id, so renames delivered twice or out of
210 /// order converge) and `?2` for a slug its rows may still be under; the
211 /// statements run in one batch per such slug. A statement that matches
212 /// nothing changes nothing, so running them again is harmless.
213 pub async fn on_event(env: &Env, db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
214 if event.kind != "workspace.renamed" {
215 return Ok(false);
216 }
217 let Ok(renamed) = serde_json::from_value::<WorkspaceRenamed>(event.data.clone()) else {
218 worker::console_error!("workspace.renamed {} could not be read", event.id);
219 return Ok(true);
220 };
221 let names: HashMap<String, String> = crate::call(
222 &env.service("IDENTITY")?,
223 "usernames",
224 &UsernamesArgs {
225 ids: vec![renamed.workspace_id.clone()],
226 },
227 )
228 .await?;
229 let current = names
230 .get(&renamed.workspace_id)
231 .cloned()
232 .unwrap_or_else(|| renamed.to.clone());
233 for stale in renamed.stale_slugs(&current) {
234 let values: [JsValue; 2] = [current.as_str().into(), stale.as_str().into()];
235 let mut batch = Vec::with_capacity(statements.len());
236 for sql in statements {
237 batch.push(db.prepare(*sql).bind(&values[..parameters(sql)])?);
238 }
239 db.batch(batch).await?;
240 }
241 Ok(true)
242 }
243
244 /// How many values a statement takes: its highest `?N`.
245 pub fn parameters(sql: &str) -> usize {
246 sql.split('?')
247 .skip(1)
248 .filter_map(|rest| {
249 let digits: String = rest.chars().take_while(char::is_ascii_digit).collect();
250 digits.parse().ok()
251 })
252 .max()
253 .unwrap_or(0)
254 }
255
256 #[cfg(test)]
257 mod tests {
258 use super::parameters;
259
260 #[test]
261 fn counts_numbered_parameters() {
262 assert_eq!(parameters("UPDATE t SET a = ?1 WHERE a = ?2"), 2);
263 assert_eq!(parameters("DELETE FROM t WHERE a = ?2"), 2);
264 assert_eq!(parameters("UPDATE t SET a = ?1"), 1);
265 assert_eq!(parameters("DELETE FROM t"), 0);
266 }
267 }
268}
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look269
270/// Moving a service's rows when a repository's path changes: transferred
271/// to another workspace (`repo.transferred`) or renamed within its own
272/// (`repo.renamed`). Both are handled the same way, so a service that
273/// follows transfers follows renames too.
274pub mod transfer {
275 use g1t_contracts::events::{Event, RepoRenamed, RepoTransferred};
276 use g1t_contracts::repos::{PathByIdArgs, RepoPath};
277 use worker::wasm_bindgen::JsValue;
278 use worker::{D1Database, Env, Result};
279
280 pub use crate::rename::parameters;
281
282 /// A repository's path change, from either event.
283 #[derive(Clone, Debug, PartialEq, Eq)]
284 pub struct Moved {
285 pub repo_id: String,
286 /// The paths (`namespace/name`) the event names, old then new.
287 pub paths: [String; 2],
288 }
289
290 impl Moved {
291 /// The paths whose rows move to `current`: the two the event
292 /// names, minus `current`.
293 pub fn stale_paths(&self, current: &str) -> Vec<String> {
294 let mut paths: Vec<String> = Vec::new();
295 for path in &self.paths {
296 if path != current && !paths.contains(path) {
297 paths.push(path.clone());
298 }
299 }
300 paths
301 }
302
303 /// Where the event says it went, for when repos does not know.
304 pub fn destination(&self) -> &str {
305 &self.paths[1]
306 }
307 }
308
309 impl From<RepoTransferred> for Moved {
310 fn from(t: RepoTransferred) -> Self {
311 Moved {
312 paths: [format!("{}/{}", t.from, t.name), format!("{}/{}", t.to, t.name)],
313 repo_id: t.repo_id,
314 }
315 }
316 }
317
318 impl From<RepoRenamed> for Moved {
319 fn from(r: RepoRenamed) -> Self {
320 Moved {
321 paths: [format!("{}/{}", r.namespace, r.from), format!("{}/{}", r.namespace, r.to)],
322 repo_id: r.repo_id,
323 }
324 }
325 }
326
327 /// The values the statements are bound with for one stale path, in
328 /// order: `?1` the current path (`namespace/name`), `?2` the stale
329 /// path, `?3` the current workspace, `?4` the stale workspace, `?5` the
330 /// repository's id, `?6` the current name, `?7` the stale name.
331 pub fn values(current: &str, stale: &str, repo_id: &str) -> [String; 7] {
332 let namespace = |path: &str| path.split_once('/').map_or(path, |(ns, _)| ns).to_owned();
333 let name = |path: &str| path.split_once('/').map_or("", |(_, name)| name).to_owned();
334 [
335 current.to_owned(),
336 stale.to_owned(),
337 namespace(current),
338 namespace(stale),
339 repo_id.to_owned(),
340 name(current),
341 name(stale),
342 ]
343 }
344
345 /// Handles `repo.transferred` and `repo.renamed` with `statements`, and
346 /// says whether `event` was one. The repository's current path is asked
347 /// of repos by id, so path changes delivered twice or out of order
348 /// converge; the statements run in one batch per path its rows may
349 /// still be under (see [`values`] for the parameters). A statement that
350 /// matches nothing changes nothing, so running them again is harmless.
351 pub async fn on_event(env: &Env, db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
352 let Some(moved) = read(event) else {
353 return Ok(false);
354 };
355 let current = current_path(env, &moved).await?;
356 for stale in moved.stale_paths(&current) {
357 let values = values(&current, &stale, &moved.repo_id);
358 let values: Vec<JsValue> = values.iter().map(|v| JsValue::from(v.as_str())).collect();
359 let mut batch = Vec::with_capacity(statements.len());
360 for sql in statements {
361 batch.push(db.prepare(*sql).bind(&values[..parameters(sql)])?);
362 }
363 db.batch(batch).await?;
364 }
365 Ok(true)
366 }
367
368 /// The path change `event` announces, if it is one.
369 pub fn read(event: &Event) -> Option<Moved> {
370 let read = match event.kind.as_str() {
371 "repo.transferred" => serde_json::from_value::<RepoTransferred>(event.data.clone()).map(Moved::from),
372 "repo.renamed" => serde_json::from_value::<RepoRenamed>(event.data.clone()).map(Moved::from),
373 _ => return None,
374 };
375 if read.is_err() {
376 worker::console_error!("{} {} could not be read", event.kind, event.id);
377 }
378 read.ok()
379 }
380
381 /// The repository's path now, as `namespace/name`: asked of repos, or
382 /// where the event says it went when repos does not know it.
383 pub async fn current_path(env: &Env, moved: &Moved) -> Result<String> {
384 let path: Option<RepoPath> = crate::call(
385 &env.service("REPOS")?,
386 "path_by_id",
387 &PathByIdArgs {
388 id: moved.repo_id.clone(),
389 },
390 )
391 .await?;
392 Ok(path.map_or_else(
393 || moved.destination().to_owned(),
394 |path| format!("{}/{}", path.namespace, path.name),
395 ))
396 }
397
398 #[cfg(test)]
399 mod tests {
400 use super::*;
401
402 #[test]
403 fn binds_paths_workspaces_names_and_the_id() {
404 assert_eq!(
405 values("flagon-io/g1t", "syntaqx/g1t", "rep_1"),
406 ["flagon-io/g1t", "syntaqx/g1t", "flagon-io", "syntaqx", "rep_1", "g1t", "g1t"].map(String::from)
407 );
408 assert_eq!(values("acme/new", "acme/old", "rep_1")[5..], ["new".to_owned(), "old".to_owned()]);
409 }
410
411 #[test]
412 fn a_rename_and_a_transfer_are_both_moves() {
413 let renamed: Moved = RepoRenamed {
414 repo_id: "rep_1".into(),
415 namespace: "acme".into(),
416 from: "old".into(),
417 to: "new".into(),
418 }
419 .into();
420 assert_eq!(renamed.stale_paths("acme/new"), vec!["acme/old"]);
421 assert_eq!(renamed.destination(), "acme/new");
422 let transferred: Moved = RepoTransferred {
423 repo_id: "rep_1".into(),
424 name: "g1t".into(),
425 from: "a".into(),
426 to: "b".into(),
427 }
428 .into();
429 assert_eq!(transferred.stale_paths("c/g1t"), vec!["a/g1t", "b/g1t"]);
430 }
431 }
432}
433
434/// Reading the repository lifecycle events every service reacts to:
435/// `repo.deleted` (stop and hide; it may come back), `repo.restored`
436/// (start again) and `repo.purged` (drop every row kept by its id).
437pub mod lifecycle {
438 use g1t_contracts::events::{Event, RepoDeleted, RepoPurged, RepoRestored};
439 use worker::{D1Database, Result};
440
441 /// One of the three, read.
442 #[derive(Debug)]
443 pub enum Lifecycle {
444 Deleted(RepoDeleted),
445 Restored(RepoRestored),
446 Purged(RepoPurged),
447 }
448
449 impl Lifecycle {
450 pub fn repo_id(&self) -> &str {
451 match self {
452 Lifecycle::Deleted(e) => &e.repo_id,
453 Lifecycle::Restored(e) => &e.repo_id,
454 Lifecycle::Purged(e) => &e.repo_id,
455 }
456 }
457 }
458
459 /// The lifecycle event `event` is, if it is one.
460 pub fn read(event: &Event) -> Option<Lifecycle> {
461 let data = event.data.clone();
462 let read = match event.kind.as_str() {
463 "repo.deleted" => serde_json::from_value(data).map(Lifecycle::Deleted),
464 "repo.restored" => serde_json::from_value(data).map(Lifecycle::Restored),
465 "repo.purged" => serde_json::from_value(data).map(Lifecycle::Purged),
466 _ => return None,
467 };
468 if read.is_err() {
469 worker::console_error!("{} {} could not be read", event.kind, event.id);
470 }
471 read.ok()
472 }
473
474 /// Handles `repo.purged` with `statements`, each taking the
475 /// repository's id as `?1`, in one batch; says whether `event` was
476 /// one. Running them again changes nothing.
477 pub async fn on_purged(db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
478 let Some(Lifecycle::Purged(purged)) = read(event) else {
479 return Ok(false);
480 };
481 if statements.is_empty() {
482 return Ok(true);
483 }
484 let mut batch = Vec::with_capacity(statements.len());
485 for sql in statements {
486 batch.push(db.prepare(*sql).bind(&[purged.repo_id.as_str().into()])?);
487 }
488 db.batch(batch).await?;
489 Ok(true)
490 }
491}
492
Merge account deletion: soft delete for 30 days, staff restore and purge, ghost for what remains (identity 0037)493/// What a service does when an account is purged (`user.deleted`): drop
494/// what it keeps for the account alone, and show what it wrote as `ghost`.
495pub mod user_deleted {
496 use g1t_contracts::events::{Event, UserDeleted};
497 use worker::wasm_bindgen::JsValue;
498 use worker::{D1Database, Result};
499
500 /// What each statement is given: the account's id as `?1`, and its
501 /// username (lowercase) as `?2` when the statement names `?2`.
502 pub fn binds<'a>(sql: &str, user_id: &'a str, username: &'a str) -> Vec<&'a str> {
503 let mut binds = vec![user_id];
504 if sql.contains("?2") {
505 binds.push(username);
506 }
507 binds
508 }
509
510 /// Handles `user.deleted` with `statements` in one batch; says whether
511 /// `event` was one. Running them again changes nothing.
512 pub async fn on_event(db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
513 if event.kind != "user.deleted" {
514 return Ok(false);
515 }
516 let Ok(deleted) = serde_json::from_value::<UserDeleted>(event.data.clone()) else {
517 worker::console_error!("user.deleted {} could not be read", event.id);
518 return Ok(true);
519 };
520 let username = deleted.username.to_lowercase();
521 if deleted.user_id.is_empty() || username.is_empty() || statements.is_empty() {
522 return Ok(true);
523 }
524 let mut batch = Vec::with_capacity(statements.len());
525 for sql in statements {
526 let values: Vec<JsValue> = binds(sql, &deleted.user_id, &username).into_iter().map(JsValue::from).collect();
527 batch.push(db.prepare(*sql).bind(&values)?);
528 }
529 db.batch(batch).await?;
530 Ok(true)
531 }
532}
533
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look534/// Dropping what a service keeps for a workspace alone when the workspace
535/// is deleted.
536pub mod deleted {
537 use g1t_contracts::events::{Event, WorkspaceDeleted};
538 use worker::{D1Database, Result};
539
540 /// Handles `workspace.deleted` with `statements`, each taking the
541 /// workspace's slug as `?1`, in one batch; says whether `event` was
542 /// one. Running them again changes nothing.
543 pub async fn on_event(db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
544 if event.kind != "workspace.deleted" {
545 return Ok(false);
546 }
547 let Ok(deleted) = serde_json::from_value::<WorkspaceDeleted>(event.data.clone()) else {
548 worker::console_error!("workspace.deleted {} could not be read", event.id);
549 return Ok(true);
550 };
551 let slug = deleted.slug.to_lowercase();
552 if slug.is_empty() || statements.is_empty() {
553 return Ok(true);
554 }
555 let mut batch = Vec::with_capacity(statements.len());
556 for sql in statements {
557 batch.push(db.prepare(*sql).bind(&[slug.as_str().into()])?);
558 }
559 db.batch(batch).await?;
560 Ok(true)
561 }
562}

This file's history is long; its oldest lines are credited to the oldest commit read.