Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
634 linesCodeBlameRaw
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 // Say which call's answer did not read, and why: a bare serde error
61 // ("JSON serialization error") is all a log would otherwise show. The
62 // body is left out, as it may carry a signed link.
63 let text = response.text().await?;
64 read_answer(method, &text)
65}
66
67/// Calls `method` like [`call`], waiting at most `within` for the answer:
68/// `None` when none came in time. Nothing on Workers ends a subrequest that
69/// stops answering, so a caller that must answer someone (a click on Merge)
70/// bounds the calls it makes this way and says what happened instead of
71/// hanging until its own caller gives up. The call is not cancelled: what
72/// it was doing may still finish, so only a call that is safe to make
73/// again (or whose late result is harmless) should be bounded.
74pub async fn call_within<A: Serialize, R: DeserializeOwned>(
75 service: &Fetcher,
76 method: &str,
77 arguments: &A,
78 within: std::time::Duration,
79) -> Result<Option<R>> {
80 match first_of(call(service, method, arguments), worker::Delay::from(within)).await {
81 Some(answer) => answer.map(Some),
82 None => {
83 worker::console_error!("{method} did not answer within {} s", within.as_secs());
84 Ok(None)
85 }
86 }
87}
88
89/// `work`'s output if it finishes before `deadline` does, `None` otherwise.
90pub async fn first_of<W: std::future::Future, D: std::future::Future>(work: W, deadline: D) -> Option<W::Output> {
91 use std::task::Poll;
92 let mut work = Box::pin(work);
93 let mut deadline = Box::pin(deadline);
94 std::future::poll_fn(|cx| {
95 if let Poll::Ready(output) = work.as_mut().poll(cx) {
96 return Poll::Ready(Some(output));
97 }
98 if deadline.as_mut().poll(cx).is_ready() {
99 return Poll::Ready(None);
100 }
101 Poll::Pending
102 })
103 .await
104}
105
106#[cfg(test)]
107mod deadline_tests {
108 use super::first_of;
109 use std::future::{pending, ready};
110 use std::task::{Context, Poll, Waker};
111
112 fn now<F: std::future::Future>(future: F) -> Poll<F::Output> {
113 let mut future = Box::pin(future);
114 future.as_mut().poll(&mut Context::from_waker(Waker::noop()))
115 }
116
117 #[test]
118 fn an_answer_in_time_is_passed_through_and_a_late_one_is_not_waited_for() {
119 assert_eq!(now(first_of(ready(7), pending::<()>())), Poll::Ready(Some(7)));
120 assert_eq!(now(first_of(pending::<u8>(), ready(()))), Poll::Ready(None));
121 // Both ready: the answer wins.
122 assert_eq!(now(first_of(ready(1), ready(()))), Poll::Ready(Some(1)));
123 assert_eq!(now(first_of(pending::<u8>(), pending::<()>())), Poll::Pending);
124 }
125}
126
127/// A method's answer, read from its body.
128pub fn read_answer<R: DeserializeOwned>(method: &str, text: &str) -> Result<R> {
129 serde_json::from_str(text).map_err(|error| {
130 worker::Error::RustError(format!("{method} answered with what could not be read: {error}"))
131 })
132}
133
134pub mod d1;
135pub mod limits;
136pub mod pause;
137pub mod wire;
138
139/// Helpers for bindings that workers-rs has no typed wrapper for, such as
140/// Artifacts and Email Sending. Values cross the boundary as JSON.
141pub mod js {
142 use std::fmt;
143
144 use serde::Serialize;
145 use serde::de::DeserializeOwned;
146 use worker::js_sys::{Array, Function, JSON, Promise, Reflect};
147 use worker::wasm_bindgen::{JsCast, JsValue};
148 use worker::wasm_bindgen_futures::JsFuture;
149 use worker::{Env, Error, Result};
150
151 /// An exception thrown by JavaScript, with its `code` if it had one.
152 #[derive(Debug)]
153 pub struct Thrown {
154 pub code: Option<String>,
155 pub message: String,
156 }
157
158 impl Thrown {
159 fn from_value(value: JsValue) -> Self {
160 let property = |name: &str| {
161 Reflect::get(&value, &name.into())
162 .ok()
163 .and_then(|property| property.as_string())
164 };
165 Thrown {
166 code: property("code"),
167 message: property("message").unwrap_or_else(|| format!("{value:?}")),
168 }
169 }
170
171 pub fn is(&self, code: &str) -> bool {
172 self.code.as_deref() == Some(code)
173 }
174 }
175
176 impl fmt::Display for Thrown {
177 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
178 match &self.code {
179 Some(code) => write!(f, "{code}: {}", self.message),
180 None => f.write_str(&self.message),
181 }
182 }
183 }
184
185 impl From<Thrown> for Error {
186 fn from(thrown: Thrown) -> Self {
187 Error::RustError(thrown.to_string())
188 }
189 }
190
191 /// The binding called `name`, as a raw JavaScript value.
192 pub fn binding(env: &Env, name: &str) -> Result<JsValue> {
193 let value = Reflect::get(env.as_ref(), &name.into()).map_err(Thrown::from_value)?;
194 if value.is_undefined() {
195 return Err(Error::RustError(format!(
196 "binding {name} is not configured"
197 )));
198 }
199 Ok(value)
200 }
201
202 /// Reads a property of a JavaScript object.
203 pub fn get(target: &JsValue, name: &str) -> JsValue {
204 Reflect::get(target, &name.into()).unwrap_or(JsValue::UNDEFINED)
205 }
206
207 /// Sets a property on an object.
208 pub fn set(target: &JsValue, name: &str, value: &JsValue) {
209 let _ = Reflect::set(target, &name.into(), value);
210 }
211
212 pub fn to_js<T: Serialize>(value: &T) -> Result<JsValue> {
213 Ok(JSON::parse(&serde_json::to_string(value)?).map_err(Thrown::from_value)?)
214 }
215
216 pub fn from_js<T: DeserializeOwned>(value: &JsValue) -> Result<T> {
217 let text = if value.is_undefined() {
218 None
219 } else {
220 JSON::stringify(value)
221 .map_err(Thrown::from_value)?
222 .as_string()
223 };
224 let text = text.as_deref().unwrap_or("null");
225 serde_json::from_str(text).map_err(|error| {
226 // Say what arrived; a bare serde error is useless in a log.
227 let seen: String = text.chars().take(300).collect();
228 Error::RustError(format!(
229 "unexpected value from JavaScript ({error}): {seen}"
230 ))
231 })
232 }
233
234 /// Calls `target[method](...args)` and awaits the result if it is a
235 /// thenable. `method` may be a name or a symbol.
236 pub async fn call_key(
237 target: &JsValue,
238 method: &JsValue,
239 args: &[JsValue],
240 ) -> std::result::Result<JsValue, Thrown> {
241 let function: Function = Reflect::get(target, method)
242 .map_err(Thrown::from_value)?
243 .dyn_into()
244 .map_err(|_| Thrown {
245 code: None,
246 message: format!("{method:?} is not a function"),
247 })?;
248 let arguments: Array = args.iter().collect();
249 // An RPC stub treats every property access as a remote method, so
250 // `function.apply(...)` would be sent over the wire as a call to
251 // "apply". Reflect.apply invokes the function without touching it.
252 let returned = Reflect::apply(&function, target, &arguments).map_err(Thrown::from_value)?;
253 // Worker RPC returns its own thenable rather than a Promise, so
254 // resolve whatever came back instead of testing its type.
255 JsFuture::from(Promise::resolve(&returned))
256 .await
257 .map_err(Thrown::from_value)
258 }
259
260 /// Calls `target.method(...args)`; see [`call_key`].
261 pub async fn call(
262 target: &JsValue,
263 method: &str,
264 args: &[JsValue],
265 ) -> std::result::Result<JsValue, Thrown> {
266 call_key(target, &method.into(), args).await
267 }
268}
269
270/// Moving a service's rows when a workspace is renamed.
271pub mod rename {
272 use std::collections::HashMap;
273
274 use g1t_contracts::events::{Event, WorkspaceRenamed};
275 use g1t_contracts::identity::UsernamesArgs;
276 use worker::wasm_bindgen::JsValue;
277 use worker::{D1Database, Env, Result};
278
279 /// Handles `workspace.renamed` with `statements`, and says whether
280 /// `event` was one. Each statement uses `?1` for the workspace's current
281 /// slug (asked of identity by id, so renames delivered twice or out of
282 /// order converge) and `?2` for a slug its rows may still be under; the
283 /// statements run in one batch per such slug. A statement that matches
284 /// nothing changes nothing, so running them again is harmless.
285 pub async fn on_event(env: &Env, db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
286 if event.kind != "workspace.renamed" {
287 return Ok(false);
288 }
289 let Ok(renamed) = serde_json::from_value::<WorkspaceRenamed>(event.data.clone()) else {
290 worker::console_error!("workspace.renamed {} could not be read", event.id);
291 return Ok(true);
292 };
293 let names: HashMap<String, String> = crate::call(
294 &env.service("IDENTITY")?,
295 "usernames",
296 &UsernamesArgs {
297 ids: vec![renamed.workspace_id.clone()],
298 },
299 )
300 .await?;
301 let current = names
302 .get(&renamed.workspace_id)
303 .cloned()
304 .unwrap_or_else(|| renamed.to.clone());
305 for stale in renamed.stale_slugs(&current) {
306 let values: [JsValue; 2] = [current.as_str().into(), stale.as_str().into()];
307 let mut batch = Vec::with_capacity(statements.len());
308 for sql in statements {
309 batch.push(db.prepare(*sql).bind(&values[..parameters(sql)])?);
310 }
311 db.batch(batch).await?;
312 }
313 Ok(true)
314 }
315
316 /// How many values a statement takes: its highest `?N`.
317 pub fn parameters(sql: &str) -> usize {
318 sql.split('?')
319 .skip(1)
320 .filter_map(|rest| {
321 let digits: String = rest.chars().take_while(char::is_ascii_digit).collect();
322 digits.parse().ok()
323 })
324 .max()
325 .unwrap_or(0)
326 }
327
328 #[cfg(test)]
329 mod tests {
330 use super::parameters;
331
332 #[test]
333 fn counts_numbered_parameters() {
334 assert_eq!(parameters("UPDATE t SET a = ?1 WHERE a = ?2"), 2);
335 assert_eq!(parameters("DELETE FROM t WHERE a = ?2"), 2);
336 assert_eq!(parameters("UPDATE t SET a = ?1"), 1);
337 assert_eq!(parameters("DELETE FROM t"), 0);
338 }
339 }
340}
341
342/// Moving a service's rows when a repository's path changes: transferred
343/// to another workspace (`repo.transferred`) or renamed within its own
344/// (`repo.renamed`). Both are handled the same way, so a service that
345/// follows transfers follows renames too.
346pub mod transfer {
347 use g1t_contracts::events::{Event, RepoRenamed, RepoTransferred};
348 use g1t_contracts::repos::{PathByIdArgs, RepoPath};
349 use worker::wasm_bindgen::JsValue;
350 use worker::{D1Database, Env, Result};
351
352 pub use crate::rename::parameters;
353
354 /// A repository's path change, from either event.
355 #[derive(Clone, Debug, PartialEq, Eq)]
356 pub struct Moved {
357 pub repo_id: String,
358 /// The paths (`namespace/name`) the event names, old then new.
359 pub paths: [String; 2],
360 }
361
362 impl Moved {
363 /// The paths whose rows move to `current`: the two the event
364 /// names, minus `current`.
365 pub fn stale_paths(&self, current: &str) -> Vec<String> {
366 let mut paths: Vec<String> = Vec::new();
367 for path in &self.paths {
368 if path != current && !paths.contains(path) {
369 paths.push(path.clone());
370 }
371 }
372 paths
373 }
374
375 /// Where the event says it went, for when repos does not know.
376 pub fn destination(&self) -> &str {
377 &self.paths[1]
378 }
379 }
380
381 impl From<RepoTransferred> for Moved {
382 fn from(t: RepoTransferred) -> Self {
383 Moved {
384 paths: [format!("{}/{}", t.from, t.name), format!("{}/{}", t.to, t.name)],
385 repo_id: t.repo_id,
386 }
387 }
388 }
389
390 impl From<RepoRenamed> for Moved {
391 fn from(r: RepoRenamed) -> Self {
392 Moved {
393 paths: [format!("{}/{}", r.namespace, r.from), format!("{}/{}", r.namespace, r.to)],
394 repo_id: r.repo_id,
395 }
396 }
397 }
398
399 /// The values the statements are bound with for one stale path, in
400 /// order: `?1` the current path (`namespace/name`), `?2` the stale
401 /// path, `?3` the current workspace, `?4` the stale workspace, `?5` the
402 /// repository's id, `?6` the current name, `?7` the stale name.
403 pub fn values(current: &str, stale: &str, repo_id: &str) -> [String; 7] {
404 let namespace = |path: &str| path.split_once('/').map_or(path, |(ns, _)| ns).to_owned();
405 let name = |path: &str| path.split_once('/').map_or("", |(_, name)| name).to_owned();
406 [
407 current.to_owned(),
408 stale.to_owned(),
409 namespace(current),
410 namespace(stale),
411 repo_id.to_owned(),
412 name(current),
413 name(stale),
414 ]
415 }
416
417 /// Handles `repo.transferred` and `repo.renamed` with `statements`, and
418 /// says whether `event` was one. The repository's current path is asked
419 /// of repos by id, so path changes delivered twice or out of order
420 /// converge; the statements run in one batch per path its rows may
421 /// still be under (see [`values`] for the parameters). A statement that
422 /// matches nothing changes nothing, so running them again is harmless.
423 pub async fn on_event(env: &Env, db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
424 let Some(moved) = read(event) else {
425 return Ok(false);
426 };
427 let current = current_path(env, &moved).await?;
428 for stale in moved.stale_paths(&current) {
429 let values = values(&current, &stale, &moved.repo_id);
430 let values: Vec<JsValue> = values.iter().map(|v| JsValue::from(v.as_str())).collect();
431 let mut batch = Vec::with_capacity(statements.len());
432 for sql in statements {
433 batch.push(db.prepare(*sql).bind(&values[..parameters(sql)])?);
434 }
435 db.batch(batch).await?;
436 }
437 Ok(true)
438 }
439
440 /// The path change `event` announces, if it is one.
441 pub fn read(event: &Event) -> Option<Moved> {
442 let read = match event.kind.as_str() {
443 "repo.transferred" => serde_json::from_value::<RepoTransferred>(event.data.clone()).map(Moved::from),
444 "repo.renamed" => serde_json::from_value::<RepoRenamed>(event.data.clone()).map(Moved::from),
445 _ => return None,
446 };
447 if read.is_err() {
448 worker::console_error!("{} {} could not be read", event.kind, event.id);
449 }
450 read.ok()
451 }
452
453 /// The repository's path now, as `namespace/name`: asked of repos, or
454 /// where the event says it went when repos does not know it.
455 pub async fn current_path(env: &Env, moved: &Moved) -> Result<String> {
456 let path: Option<RepoPath> = crate::call(
457 &env.service("REPOS")?,
458 "path_by_id",
459 &PathByIdArgs {
460 id: moved.repo_id.clone(),
461 },
462 )
463 .await?;
464 Ok(path.map_or_else(
465 || moved.destination().to_owned(),
466 |path| format!("{}/{}", path.namespace, path.name),
467 ))
468 }
469
470 #[cfg(test)]
471 mod tests {
472 use super::*;
473
474 #[test]
475 fn binds_paths_workspaces_names_and_the_id() {
476 assert_eq!(
477 values("flagon-io/g1t", "syntaqx/g1t", "rep_1"),
478 ["flagon-io/g1t", "syntaqx/g1t", "flagon-io", "syntaqx", "rep_1", "g1t", "g1t"].map(String::from)
479 );
480 assert_eq!(values("acme/new", "acme/old", "rep_1")[5..], ["new".to_owned(), "old".to_owned()]);
481 }
482
483 #[test]
484 fn a_rename_and_a_transfer_are_both_moves() {
485 let renamed: Moved = RepoRenamed {
486 repo_id: "rep_1".into(),
487 namespace: "acme".into(),
488 from: "old".into(),
489 to: "new".into(),
490 }
491 .into();
492 assert_eq!(renamed.stale_paths("acme/new"), vec!["acme/old"]);
493 assert_eq!(renamed.destination(), "acme/new");
494 let transferred: Moved = RepoTransferred {
495 repo_id: "rep_1".into(),
496 name: "g1t".into(),
497 from: "a".into(),
498 to: "b".into(),
499 }
500 .into();
501 assert_eq!(transferred.stale_paths("c/g1t"), vec!["a/g1t", "b/g1t"]);
502 }
503 }
504}
505
506/// Reading the repository lifecycle events every service reacts to:
507/// `repo.deleted` (stop and hide; it may come back), `repo.restored`
508/// (start again) and `repo.purged` (drop every row kept by its id).
509pub mod lifecycle {
510 use g1t_contracts::events::{Event, RepoDeleted, RepoPurged, RepoRestored};
511 use worker::{D1Database, Result};
512
513 /// One of the three, read.
514 #[derive(Debug)]
515 pub enum Lifecycle {
516 Deleted(RepoDeleted),
517 Restored(RepoRestored),
518 Purged(RepoPurged),
519 }
520
521 impl Lifecycle {
522 pub fn repo_id(&self) -> &str {
523 match self {
524 Lifecycle::Deleted(e) => &e.repo_id,
525 Lifecycle::Restored(e) => &e.repo_id,
526 Lifecycle::Purged(e) => &e.repo_id,
527 }
528 }
529 }
530
531 /// The lifecycle event `event` is, if it is one.
532 pub fn read(event: &Event) -> Option<Lifecycle> {
533 let data = event.data.clone();
534 let read = match event.kind.as_str() {
535 "repo.deleted" => serde_json::from_value(data).map(Lifecycle::Deleted),
536 "repo.restored" => serde_json::from_value(data).map(Lifecycle::Restored),
537 "repo.purged" => serde_json::from_value(data).map(Lifecycle::Purged),
538 _ => return None,
539 };
540 if read.is_err() {
541 worker::console_error!("{} {} could not be read", event.kind, event.id);
542 }
543 read.ok()
544 }
545
546 /// Handles `repo.purged` with `statements`, each taking the
547 /// repository's id as `?1`, in one batch; says whether `event` was
548 /// one. Running them again changes nothing.
549 pub async fn on_purged(db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
550 let Some(Lifecycle::Purged(purged)) = read(event) else {
551 return Ok(false);
552 };
553 if statements.is_empty() {
554 return Ok(true);
555 }
556 let mut batch = Vec::with_capacity(statements.len());
557 for sql in statements {
558 batch.push(db.prepare(*sql).bind(&[purged.repo_id.as_str().into()])?);
559 }
560 db.batch(batch).await?;
561 Ok(true)
562 }
563}
564
565/// What a service does when an account is purged (`user.deleted`): drop
566/// what it keeps for the account alone, and show what it wrote as `ghost`.
567pub mod user_deleted {
568 use g1t_contracts::events::{Event, UserDeleted};
569 use worker::wasm_bindgen::JsValue;
570 use worker::{D1Database, Result};
571
572 /// What each statement is given: the account's id as `?1`, and its
573 /// username (lowercase) as `?2` when the statement names `?2`.
574 pub fn binds<'a>(sql: &str, user_id: &'a str, username: &'a str) -> Vec<&'a str> {
575 let mut binds = vec![user_id];
576 if sql.contains("?2") {
577 binds.push(username);
578 }
579 binds
580 }
581
582 /// Handles `user.deleted` with `statements` in one batch; says whether
583 /// `event` was one. Running them again changes nothing.
584 pub async fn on_event(db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
585 if event.kind != "user.deleted" {
586 return Ok(false);
587 }
588 let Ok(deleted) = serde_json::from_value::<UserDeleted>(event.data.clone()) else {
589 worker::console_error!("user.deleted {} could not be read", event.id);
590 return Ok(true);
591 };
592 let username = deleted.username.to_lowercase();
593 if deleted.user_id.is_empty() || username.is_empty() || statements.is_empty() {
594 return Ok(true);
595 }
596 let mut batch = Vec::with_capacity(statements.len());
597 for sql in statements {
598 let values: Vec<JsValue> = binds(sql, &deleted.user_id, &username).into_iter().map(JsValue::from).collect();
599 batch.push(db.prepare(*sql).bind(&values)?);
600 }
601 db.batch(batch).await?;
602 Ok(true)
603 }
604}
605
606/// Dropping what a service keeps for a workspace alone when the workspace
607/// is deleted.
608pub mod deleted {
609 use g1t_contracts::events::{Event, WorkspaceDeleted};
610 use worker::{D1Database, Result};
611
612 /// Handles `workspace.deleted` with `statements`, each taking the
613 /// workspace's slug as `?1`, in one batch; says whether `event` was
614 /// one. Running them again changes nothing.
615 pub async fn on_event(db: &D1Database, event: &Event, statements: &[&str]) -> Result<bool> {
616 if event.kind != "workspace.deleted" {
617 return Ok(false);
618 }
619 let Ok(deleted) = serde_json::from_value::<WorkspaceDeleted>(event.data.clone()) else {
620 worker::console_error!("workspace.deleted {} could not be read", event.id);
621 return Ok(true);
622 };
623 let slug = deleted.slug.to_lowercase();
624 if slug.is_empty() || statements.is_empty() {
625 return Ok(true);
626 }
627 let mut batch = Vec::with_capacity(statements.len());
628 for sql in statements {
629 batch.push(db.prepare(*sql).bind(&[slug.as_str().into()])?);
630 }
631 db.batch(batch).await?;
632 Ok(true)
633 }
634}