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/crates/kit/src/lib.rs

519 lines19,178 bytesCodeBlame
1//! 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}
62
63pub mod wire;
64
65/// Helpers for bindings that workers-rs has no typed wrapper for, such as
66/// Artifacts and Email Sending. Values cross the boundary as JSON.
67pub mod js {
68 use std::fmt;
69
70 use serde::Serialize;
71 use serde::de::DeserializeOwned;
72 use worker::js_sys::{Array, Function, JSON, Promise, Reflect};
73 use worker::wasm_bindgen::{JsCast, JsValue};
74 use worker::wasm_bindgen_futures::JsFuture;
75 use worker::{Env, Error, Result};
76
77 /// An exception thrown by JavaScript, with its `code` if it had one.
78 #[derive(Debug)]
79 pub struct Thrown {
80 pub code: Option<String>,
81 pub message: String,
82 }
83
84 impl Thrown {
85 fn from_value(value: JsValue) -> Self {
86 let property = |name: &str| {
87 Reflect::get(&value, &name.into())
88 .ok()
89 .and_then(|property| property.as_string())
90 };
91 Thrown {
92 code: property("code"),
93 message: property("message").unwrap_or_else(|| format!("{value:?}")),
94 }
95 }
96
97 pub fn is(&self, code: &str) -> bool {
98 self.code.as_deref() == Some(code)
99 }
100 }
101
102 impl fmt::Display for Thrown {
103 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
104 match &self.code {
105 Some(code) => write!(f, "{code}: {}", self.message),
106 None => f.write_str(&self.message),
107 }
108 }
109 }
110
111 impl From<Thrown> for Error {
112 fn from(thrown: Thrown) -> Self {
113 Error::RustError(thrown.to_string())
114 }
115 }
116
117 /// The binding called `name`, as a raw JavaScript value.
118 pub fn binding(env: &Env, name: &str) -> Result<JsValue> {
119 let value = Reflect::get(env.as_ref(), &name.into()).map_err(Thrown::from_value)?;
120 if value.is_undefined() {
121 return Err(Error::RustError(format!(
122 "binding {name} is not configured"
123 )));
124 }
125 Ok(value)
126 }
127
128 /// Reads a property of a JavaScript object.
129 pub fn get(target: &JsValue, name: &str) -> JsValue {
130 Reflect::get(target, &name.into()).unwrap_or(JsValue::UNDEFINED)
131 }
132
133 /// Sets a property on an object.
134 pub fn set(target: &JsValue, name: &str, value: &JsValue) {
135 let _ = Reflect::set(target, &name.into(), value);
136 }
137
138 pub fn to_js<T: Serialize>(value: &T) -> Result<JsValue> {
139 Ok(JSON::parse(&serde_json::to_string(value)?).map_err(Thrown::from_value)?)
140 }
141
142 pub fn from_js<T: DeserializeOwned>(value: &JsValue) -> Result<T> {
143 let text = if value.is_undefined() {
144 None
145 } else {
146 JSON::stringify(value)
147 .map_err(Thrown::from_value)?
148 .as_string()
149 };
150 let text = text.as_deref().unwrap_or("null");
151 serde_json::from_str(text).map_err(|error| {
152 // Say what arrived; a bare serde error is useless in a log.
153 let seen: String = text.chars().take(300).collect();
154 Error::RustError(format!(
155 "unexpected value from JavaScript ({error}): {seen}"
156 ))
157 })
158 }
159
160 /// Calls `target[method](...args)` and awaits the result if it is a
161 /// thenable. `method` may be a name or a symbol.
162 pub async fn call_key(
163 target: &JsValue,
164 method: &JsValue,
165 args: &[JsValue],
166 ) -> std::result::Result<JsValue, Thrown> {
167 let function: Function = Reflect::get(target, method)
168 .map_err(Thrown::from_value)?
169 .dyn_into()
170 .map_err(|_| Thrown {
171 code: None,
172 message: format!("{method:?} is not a function"),
173 })?;
174 let arguments: Array = args.iter().collect();
175 // An RPC stub treats every property access as a remote method, so
176 // `function.apply(...)` would be sent over the wire as a call to
177 // "apply". Reflect.apply invokes the function without touching it.
178 let returned = Reflect::apply(&function, target, &arguments).map_err(Thrown::from_value)?;
179 // Worker RPC returns its own thenable rather than a Promise, so
180 // resolve whatever came back instead of testing its type.
181 JsFuture::from(Promise::resolve(&returned))
182 .await
183 .map_err(Thrown::from_value)
184 }
185
186 /// Calls `target.method(...args)`; see [`call_key`].
187 pub async fn call(
188 target: &JsValue,
189 method: &str,
190 args: &[JsValue],
191 ) -> std::result::Result<JsValue, Thrown> {
192 call_key(target, &method.into(), args).await
193 }
194}
195
196/// Moving a service's rows when a workspace is renamed.
197pub mod rename {
198 use std::collections::HashMap;
199
200 use g1t_contracts::events::{Event, WorkspaceRenamed};
201 use g1t_contracts::identity::UsernamesArgs;
202 use worker::wasm_bindgen::JsValue;
203 use worker::{D1Database, Env, Result};
204
205 /// Handles `workspace.renamed` with `statements`, and says whether
206 /// `event` was one. Each statement uses `?1` for the workspace's current
207 /// slug (asked of identity by id, so renames delivered twice or out of
208 /// order converge) and `?2` for a slug its rows may still be under; the
209 /// statements run in one batch per such slug. A statement that matches
210 /// nothing changes nothing, so running them again is harmless.
211 pub async fn on_event(env: &Env, db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
212 if event.kind != "workspace.renamed" {
213 return Ok(false);
214 }
215 let Ok(renamed) = serde_json::from_value::<WorkspaceRenamed>(event.data.clone()) else {
216 worker::console_error!("workspace.renamed {} could not be read", event.id);
217 return Ok(true);
218 };
219 let names: HashMap<String, String> = crate::call(
220 &env.service("IDENTITY")?,
221 "usernames",
222 &UsernamesArgs {
223 ids: vec![renamed.workspace_id.clone()],
224 },
225 )
226 .await?;
227 let current = names
228 .get(&renamed.workspace_id)
229 .cloned()
230 .unwrap_or_else(|| renamed.to.clone());
231 for stale in renamed.stale_slugs(&current) {
232 let values: [JsValue; 2] = [current.as_str().into(), stale.as_str().into()];
233 let mut batch = Vec::with_capacity(statements.len());
234 for sql in statements {
235 batch.push(db.prepare(*sql).bind(&values[..parameters(sql)])?);
236 }
237 db.batch(batch).await?;
238 }
239 Ok(true)
240 }
241
242 /// How many values a statement takes: its highest `?N`.
243 pub fn parameters(sql: &str) -> usize {
244 sql.split('?')
245 .skip(1)
246 .filter_map(|rest| {
247 let digits: String = rest.chars().take_while(char::is_ascii_digit).collect();
248 digits.parse().ok()
249 })
250 .max()
251 .unwrap_or(0)
252 }
253
254 #[cfg(test)]
255 mod tests {
256 use super::parameters;
257
258 #[test]
259 fn counts_numbered_parameters() {
260 assert_eq!(parameters("UPDATE t SET a = ?1 WHERE a = ?2"), 2);
261 assert_eq!(parameters("DELETE FROM t WHERE a = ?2"), 2);
262 assert_eq!(parameters("UPDATE t SET a = ?1"), 1);
263 assert_eq!(parameters("DELETE FROM t"), 0);
264 }
265 }
266}
267
268/// Moving a service's rows when a repository's path changes: transferred
269/// to another workspace (`repo.transferred`) or renamed within its own
270/// (`repo.renamed`). Both are handled the same way, so a service that
271/// follows transfers follows renames too.
272pub mod transfer {
273 use g1t_contracts::events::{Event, RepoRenamed, RepoTransferred};
274 use g1t_contracts::repos::{PathByIdArgs, RepoPath};
275 use worker::wasm_bindgen::JsValue;
276 use worker::{D1Database, Env, Result};
277
278 pub use crate::rename::parameters;
279
280 /// A repository's path change, from either event.
281 #[derive(Clone, Debug, PartialEq, Eq)]
282 pub struct Moved {
283 pub repo_id: String,
284 /// The paths (`namespace/name`) the event names, old then new.
285 pub paths: [String; 2],
286 }
287
288 impl Moved {
289 /// The paths whose rows move to `current`: the two the event
290 /// names, minus `current`.
291 pub fn stale_paths(&self, current: &str) -> Vec<String> {
292 let mut paths: Vec<String> = Vec::new();
293 for path in &self.paths {
294 if path != current && !paths.contains(path) {
295 paths.push(path.clone());
296 }
297 }
298 paths
299 }
300
301 /// Where the event says it went, for when repos does not know.
302 pub fn destination(&self) -> &str {
303 &self.paths[1]
304 }
305 }
306
307 impl From<RepoTransferred> for Moved {
308 fn from(t: RepoTransferred) -> Self {
309 Moved {
310 paths: [format!("{}/{}", t.from, t.name), format!("{}/{}", t.to, t.name)],
311 repo_id: t.repo_id,
312 }
313 }
314 }
315
316 impl From<RepoRenamed> for Moved {
317 fn from(r: RepoRenamed) -> Self {
318 Moved {
319 paths: [format!("{}/{}", r.namespace, r.from), format!("{}/{}", r.namespace, r.to)],
320 repo_id: r.repo_id,
321 }
322 }
323 }
324
325 /// The values the statements are bound with for one stale path, in
326 /// order: `?1` the current path (`namespace/name`), `?2` the stale
327 /// path, `?3` the current workspace, `?4` the stale workspace, `?5` the
328 /// repository's id, `?6` the current name, `?7` the stale name.
329 pub fn values(current: &str, stale: &str, repo_id: &str) -> [String; 7] {
330 let namespace = |path: &str| path.split_once('/').map_or(path, |(ns, _)| ns).to_owned();
331 let name = |path: &str| path.split_once('/').map_or("", |(_, name)| name).to_owned();
332 [
333 current.to_owned(),
334 stale.to_owned(),
335 namespace(current),
336 namespace(stale),
337 repo_id.to_owned(),
338 name(current),
339 name(stale),
340 ]
341 }
342
343 /// Handles `repo.transferred` and `repo.renamed` with `statements`, and
344 /// says whether `event` was one. The repository's current path is asked
345 /// of repos by id, so path changes delivered twice or out of order
346 /// converge; the statements run in one batch per path its rows may
347 /// still be under (see [`values`] for the parameters). A statement that
348 /// matches nothing changes nothing, so running them again is harmless.
349 pub async fn on_event(env: &Env, db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
350 let Some(moved) = read(event) else {
351 return Ok(false);
352 };
353 let current = current_path(env, &moved).await?;
354 for stale in moved.stale_paths(&current) {
355 let values = values(&current, &stale, &moved.repo_id);
356 let values: Vec<JsValue> = values.iter().map(|v| JsValue::from(v.as_str())).collect();
357 let mut batch = Vec::with_capacity(statements.len());
358 for sql in statements {
359 batch.push(db.prepare(*sql).bind(&values[..parameters(sql)])?);
360 }
361 db.batch(batch).await?;
362 }
363 Ok(true)
364 }
365
366 /// The path change `event` announces, if it is one.
367 pub fn read(event: &Event) -> Option<Moved> {
368 let read = match event.kind.as_str() {
369 "repo.transferred" => serde_json::from_value::<RepoTransferred>(event.data.clone()).map(Moved::from),
370 "repo.renamed" => serde_json::from_value::<RepoRenamed>(event.data.clone()).map(Moved::from),
371 _ => return None,
372 };
373 if read.is_err() {
374 worker::console_error!("{} {} could not be read", event.kind, event.id);
375 }
376 read.ok()
377 }
378
379 /// The repository's path now, as `namespace/name`: asked of repos, or
380 /// where the event says it went when repos does not know it.
381 pub async fn current_path(env: &Env, moved: &Moved) -> Result<String> {
382 let path: Option<RepoPath> = crate::call(
383 &env.service("REPOS")?,
384 "path_by_id",
385 &PathByIdArgs {
386 id: moved.repo_id.clone(),
387 },
388 )
389 .await?;
390 Ok(path.map_or_else(
391 || moved.destination().to_owned(),
392 |path| format!("{}/{}", path.namespace, path.name),
393 ))
394 }
395
396 #[cfg(test)]
397 mod tests {
398 use super::*;
399
400 #[test]
401 fn binds_paths_workspaces_names_and_the_id() {
402 assert_eq!(
403 values("flagon-io/g1t", "syntaqx/g1t", "rep_1"),
404 ["flagon-io/g1t", "syntaqx/g1t", "flagon-io", "syntaqx", "rep_1", "g1t", "g1t"].map(String::from)
405 );
406 assert_eq!(values("acme/new", "acme/old", "rep_1")[5..], ["new".to_owned(), "old".to_owned()]);
407 }
408
409 #[test]
410 fn a_rename_and_a_transfer_are_both_moves() {
411 let renamed: Moved = RepoRenamed {
412 repo_id: "rep_1".into(),
413 namespace: "acme".into(),
414 from: "old".into(),
415 to: "new".into(),
416 }
417 .into();
418 assert_eq!(renamed.stale_paths("acme/new"), vec!["acme/old"]);
419 assert_eq!(renamed.destination(), "acme/new");
420 let transferred: Moved = RepoTransferred {
421 repo_id: "rep_1".into(),
422 name: "g1t".into(),
423 from: "a".into(),
424 to: "b".into(),
425 }
426 .into();
427 assert_eq!(transferred.stale_paths("c/g1t"), vec!["a/g1t", "b/g1t"]);
428 }
429 }
430}
431
432/// Reading the repository lifecycle events every service reacts to:
433/// `repo.deleted` (stop and hide; it may come back), `repo.restored`
434/// (start again) and `repo.purged` (drop every row kept by its id).
435pub mod lifecycle {
436 use g1t_contracts::events::{Event, RepoDeleted, RepoPurged, RepoRestored};
437 use worker::{D1Database, Result};
438
439 /// One of the three, read.
440 #[derive(Debug)]
441 pub enum Lifecycle {
442 Deleted(RepoDeleted),
443 Restored(RepoRestored),
444 Purged(RepoPurged),
445 }
446
447 impl Lifecycle {
448 pub fn repo_id(&self) -> &str {
449 match self {
450 Lifecycle::Deleted(e) => &e.repo_id,
451 Lifecycle::Restored(e) => &e.repo_id,
452 Lifecycle::Purged(e) => &e.repo_id,
453 }
454 }
455 }
456
457 /// The lifecycle event `event` is, if it is one.
458 pub fn read(event: &Event) -> Option<Lifecycle> {
459 let data = event.data.clone();
460 let read = match event.kind.as_str() {
461 "repo.deleted" => serde_json::from_value(data).map(Lifecycle::Deleted),
462 "repo.restored" => serde_json::from_value(data).map(Lifecycle::Restored),
463 "repo.purged" => serde_json::from_value(data).map(Lifecycle::Purged),
464 _ => return None,
465 };
466 if read.is_err() {
467 worker::console_error!("{} {} could not be read", event.kind, event.id);
468 }
469 read.ok()
470 }
471
472 /// Handles `repo.purged` with `statements`, each taking the
473 /// repository's id as `?1`, in one batch; says whether `event` was
474 /// one. Running them again changes nothing.
475 pub async fn on_purged(db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
476 let Some(Lifecycle::Purged(purged)) = read(event) else {
477 return Ok(false);
478 };
479 if statements.is_empty() {
480 return Ok(true);
481 }
482 let mut batch = Vec::with_capacity(statements.len());
483 for sql in statements {
484 batch.push(db.prepare(*sql).bind(&[purged.repo_id.as_str().into()])?);
485 }
486 db.batch(batch).await?;
487 Ok(true)
488 }
489}
490
491/// Dropping what a service keeps for a workspace alone when the workspace
492/// is deleted.
493pub mod deleted {
494 use g1t_contracts::events::{Event, WorkspaceDeleted};
495 use worker::{D1Database, Result};
496
497 /// Handles `workspace.deleted` with `statements`, each taking the
498 /// workspace's slug as `?1`, in one batch; says whether `event` was
499 /// one. Running them again changes nothing.
500 pub async fn on_event(db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
501 if event.kind != "workspace.deleted" {
502 return Ok(false);
503 }
504 let Ok(deleted) = serde_json::from_value::<WorkspaceDeleted>(event.data.clone()) else {
505 worker::console_error!("workspace.deleted {} could not be read", event.id);
506 return Ok(true);
507 };
508 let slug = deleted.slug.to_lowercase();
509 if slug.is_empty() || statements.is_empty() {
510 return Ok(true);
511 }
512 let mut batch = Vec::with_capacity(statements.len());
513 for sql in statements {
514 batch.push(db.prepare(*sql).bind(&[slug.as_str().into()])?);
515 }
516 db.batch(batch).await?;
517 Ok(true)
518 }
519}