From b7beac96d099ae665a119a8934102f95b84f12bf Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Sun, 18 Oct 2026 02:30:20 +0900 Subject: [PATCH] knot2: notify spindle on git push event Signed-off-by: Seongmin Lee --- knot2/crates/knot-atproto/src/lib.rs | 90 ++++++++++++++- knot2/crates/knot-atproto/src/resolve.rs | 50 ++++++++- knot2/crates/knot-xrpc/src/refrecords.rs | 133 +++++++++++++++++++++-- 3 files changed, 256 insertions(+), 17 deletions(-) diff --git a/knot2/crates/knot-atproto/src/lib.rs b/knot2/crates/knot-atproto/src/lib.rs index eee68614e..c26f50094 100644 --- a/knot2/crates/knot-atproto/src/lib.rs +++ b/knot2/crates/knot-atproto/src/lib.rs @@ -69,8 +69,8 @@ use knot_runtime::{ UnixMicros, }; use knot_types::{ - AccountDid, Collection, Handle, HttpStatus, KnotId, Nsid, OfferedKey, RepoDid, RepoRkey, Rkey, - UnixSeconds, + AccountDid, AtUri, Collection, Handle, HttpStatus, KnotId, Nsid, OfferedKey, RepoDid, RepoRkey, + Rkey, ServiceDid, UnixSeconds, }; use pubkeys::Cursor; use serde::{Deserialize, Serialize}; @@ -136,6 +136,13 @@ pub enum AtprotoError { PointerEncode(String), #[error("putRecord response isn't a valid receipt: {0}")] MalformedReceipt(String), + #[error("notifyEvent to {spindle} returned HTTP {status}")] + NotifyEvent { + spindle: ServiceDid, + status: HttpStatus, + }, + #[error("notifyEvent response isn't valid: {0}")] + MalformedNotify(String), } impl AtprotoError { @@ -149,7 +156,8 @@ impl AtprotoError { | AtprotoError::PlcSubmit { status, .. } | AtprotoError::PlcFetch { status, .. } | AtprotoError::PutRecord { status, .. } - | AtprotoError::GetRecord { status, .. } => status.is_transient(), + | AtprotoError::GetRecord { status, .. } + | AtprotoError::NotifyEvent { status, .. } => status.is_transient(), _ => false, } } @@ -171,7 +179,9 @@ impl AtprotoError { | AtprotoError::PutRecord { .. } | AtprotoError::GetRecord { .. } | AtprotoError::PointerEncode(_) - | AtprotoError::MalformedReceipt(_) => false, + | AtprotoError::MalformedReceipt(_) + | AtprotoError::NotifyEvent { .. } + | AtprotoError::MalformedNotify(_) => false, } } } @@ -727,6 +737,78 @@ impl Atproto { pointer::receipt_from_response(&response.body) } + /// Calls `org.tangled.event.notifyEvent` on `spindle` under this knot's + /// service auth, answering the pipeline the spindle scheduled, if any. + pub async fn notify_spindle( + &self, + authorizer: &dyn PointerAuthorizer, + spindle: &ServiceDid, + body: bytes::Bytes, + ) -> Result, AtprotoError> { + let method = + Nsid::new_static("org.tangled.event.notifyEvent").expect("Literal NSID parses"); + let url = self.spindle_xrpc_url(spindle, &method).await?; + let now = UnixSeconds::new((self.clock.now_unix_micros().get() / 1_000_000) as i64); + let mut request = json_post(url, body); + authorizer.authorize( + &mut request, + &PointerAuth { + issuer: &self.knot_did, + audience: spindle, + lxm: &method, + now_unix: now, + }, + )?; + let response = self.http.execute(request).await?; + if !response.status.is_success() { + return Err(AtprotoError::NotifyEvent { + spindle: spindle.clone(), + status: HttpStatus::from(response.status), + }); + } + #[derive(Deserialize)] + struct NotifyOutput { + pipeline: Option, + } + let output: NotifyOutput = serde_json::from_slice(&response.body) + .map_err(|error| AtprotoError::MalformedNotify(error.to_string()))?; + output + .pipeline + .map(|uri| AtUri::new_owned(&uri)) + .transpose() + .map_err(|error| AtprotoError::MalformedNotify(error.to_string())) + } + + async fn spindle_xrpc_url( + &self, + spindle: &ServiceDid, + method: &Nsid, + ) -> Result { + let did = AccountDid::new(spindle.as_str()).map_err(|_| ResolveError::Unresolvable { + value: spindle.as_str().to_string(), + })?; + let url = self.document_url(&did)?; + resolve::guard_fetch_url(&url)?; + let response = self + .http + .execute(HttpRequest::get(url)) + .await + .map_err(ResolveError::from)?; + if !response.status.is_success() { + return Err(ResolveError::Status { + status: HttpStatus::from(response.status), + } + .into()); + } + let endpoint = resolve::spindle_endpoint(spindle, &response.body)?; + let base = PdsEndpoint::new(endpoint.clone()).map_err(|_| ResolveError::BadSpindle { + value: endpoint.to_string(), + })?; + let url = xrpc_url(&base, method)?; + resolve::guard_fetch_url(&url)?; + Ok(url) + } + pub async fn verify_account_document( &self, did: &RepoDid, diff --git a/knot2/crates/knot-atproto/src/resolve.rs b/knot2/crates/knot-atproto/src/resolve.rs index ea4f310be..e9c2c398b 100644 --- a/knot2/crates/knot-atproto/src/resolve.rs +++ b/knot2/crates/knot-atproto/src/resolve.rs @@ -3,11 +3,13 @@ use std::borrow::Cow; use knot_runtime::PublicKeyBytes; use knot_types::crypto::{KeyCodec, PublicKey as CryptoKey}; use knot_types::did_doc::DidDocument; -use knot_types::{AccountDid, Handle, HttpStatus, RepoDid}; +use knot_types::{AccountDid, Handle, HttpStatus, RepoDid, ServiceDid}; use url::{Host, Url}; const LEGACY_K256_KIND: &str = "EcdsaSecp256k1VerificationKey2019"; const LEGACY_P256_KIND: &str = "EcdsaSecp256r1VerificationKey2019"; +const SPINDLE_SERVICE_ID: &str = "#tangled_spindle"; +const SPINDLE_SERVICE_TYPE: &str = "TangledSpindle"; #[derive(Debug, Clone, thiserror::Error)] pub enum ResolveError { @@ -35,6 +37,10 @@ pub enum ResolveError { MissingPds, #[error("atproto_pds endpoint {value:?} isn't valid URL")] BadPds { value: String }, + #[error("DID document declares no tangled_spindle service endpoint")] + MissingSpindle, + #[error("tangled_spindle endpoint {value:?} isn't valid URL")] + BadSpindle { value: String }, #[error("PLC directory {value:?} isn't valid http(s) base URL")] BadPlcDirectory { value: String }, #[error("identity {did} recently failed to resolve and is negatively cached")] @@ -82,6 +88,8 @@ impl ResolveError { | ResolveError::BadSigningKey(_) | ResolveError::MissingPds | ResolveError::BadPds { .. } + | ResolveError::MissingSpindle + | ResolveError::BadSpindle { .. } | ResolveError::BadPlcDirectory { .. } | ResolveError::RecentlyFailed { .. } | ResolveError::ExpectedKeyAbsent { .. } @@ -238,6 +246,31 @@ pub(crate) fn identity_from_document( }) } +/// The `#tangled_spindle` service endpoint of a spindle's DID document. +pub(crate) fn spindle_endpoint(did: &ServiceDid, body: &[u8]) -> Result { + let document: DidDocument = + serde_json::from_slice(body).map_err(|error| ResolveError::Malformed(error.to_string()))?; + if document.id.as_str() != did.as_str() { + return Err(ResolveError::IdMismatch { + requested: did.as_str().to_string(), + document: document.id.as_str().to_string(), + }); + } + let endpoint = document + .service + .iter() + .flatten() + .find(|service| { + service.id.as_str().ends_with(SPINDLE_SERVICE_ID) + && service.r#type.as_str() == SPINDLE_SERVICE_TYPE + }) + .and_then(|service| service.service_endpoint.as_ref()?.as_str()) + .ok_or(ResolveError::MissingSpindle)?; + Url::parse(endpoint).map_err(|_| ResolveError::BadSpindle { + value: endpoint.to_string(), + }) +} + pub(crate) fn document_publishes_account_shape( did: &RepoDid, body: &[u8], @@ -634,4 +667,19 @@ mod tests { as an account that stopped existing" ); } + + #[test] + fn a_spindle_answers_at_its_tangled_spindle_service() { + let did = ServiceDid::new("did:web:crow.tngl.boltless.dev").unwrap(); + let crow = br##"{"@context":["https://www.w3.org/ns/did/v1"],"id":"did:web:crow.tngl.boltless.dev","service":[{"id":"#tangled_spindle","serviceEndpoint":"https://crow.tngl.boltless.dev","type":"TangledSpindle"}]}"##; + assert_eq!( + spindle_endpoint(&did, crow).unwrap().as_str(), + "https://crow.tngl.boltless.dev/" + ); + let knot = br##"{"id":"did:web:crow.tngl.boltless.dev","service":[{"id":"#tangled_knot","serviceEndpoint":"https://crow.tngl.boltless.dev","type":"TangledKnot"}]}"##; + assert!(matches!( + spindle_endpoint(&did, knot), + Err(ResolveError::MissingSpindle) + )); + } } diff --git a/knot2/crates/knot-xrpc/src/refrecords.rs b/knot2/crates/knot-xrpc/src/refrecords.rs index f76aa7cc4..052307b0c 100644 --- a/knot2/crates/knot-xrpc/src/refrecords.rs +++ b/knot2/crates/knot-xrpc/src/refrecords.rs @@ -1,10 +1,15 @@ use std::sync::Arc; -use knot_git::{RefUpdate, Repo}; +use jacquard_repo::storage::BlockStore; +use knot_atproto::ServiceAuth; +use knot_git::{RefClass, RefUpdate, Repo}; +use knot_postreceive::Ci; +use knot_record::blocks::GitBlocks; use knot_record::chain; use knot_record::git_refs; -use knot_runtime::{Clock, HttpTransport}; -use knot_types::{AccountDid, Oid, PushOptions, RefName, RepoDid}; +use knot_record::manifest::{MANIFEST_COLLECTION, MANIFEST_RKEY}; +use knot_runtime::{Clock, HttpTransport, Signer}; +use knot_types::{AccountDid, ObjectFormat, Oid, PushOptions, RefName, RepoDid, ServiceDid}; use crate::error::XrpcError; use crate::run_blocking; @@ -168,15 +173,18 @@ impl knot_receive::RefSurface for RefProjection firehose::publish_push( - &self.state.firehose, - repo_did, - pusher, - applied, - options, - &signer, - self.state.now(), - ), + Ok(signer) => { + firehose::publish_push( + &self.state.firehose, + repo_did, + pusher, + applied, + options, + &signer, + self.state.now(), + ); + notify_spindle(&self.state, repo, repo_did, applied, options, &signer); + } Err(error) => tracing::error!( repo = repo_did.as_str(), %error, @@ -185,3 +193,104 @@ impl knot_receive::RefSurface for RefProjection( + state: &Arc>, + repo: &Repo, + did: &RepoDid, + applied: &[RefUpdate], + options: &PushOptions, + signer: &dyn Signer, +) { + if matches!(knot_receive::ci_from_push_options(options, None), Ci::Skip) { + return; + } + let spindle = match get_repo_spindle(repo) { + Ok(Some(spindle)) => spindle, + Ok(None) => return, + Err(error) => { + tracing::warn!(repo = did.as_str(), %error, "the manifest wouldn't read, so the spindle doesn't hear of this push"); + return; + } + }; + let auth = ServiceAuth::new(signer, state.entropy.as_ref()); + for update in applied + .iter() + .filter(|update| RefClass::of(update.name()) == RefClass::Public) + { + let body = push_event_body(repo.object_format(), did, update, options); + let sent = futures::executor::block_on(state.atproto.notify_spindle( + &auth, + &spindle, + body.to_string().into(), + )); + match sent { + // TODO: show spindle-provided friendly messsage about scheduled pipeline + Ok(pipeline) => tracing::info!( + repo = did.as_str(), + ref_name = update.name().as_str(), + spindle = spindle.as_str(), + pipeline = pipeline.as_ref().map(|uri| uri.as_str()), + "the spindle heard of the push" + ), + Err(error) => tracing::warn!( + repo = did.as_str(), + ref_name = update.name().as_str(), + spindle = spindle.as_str(), + %error, + "the spindle didn't take the push event" + ), + } + } +} + +fn push_event_body( + format: ObjectFormat, + did: &RepoDid, + update: &RefUpdate, + _options: &PushOptions, +) -> serde_json::Value { + let transition = update.transition(); + let null = format.null_oid(); + serde_json::json!({ + "repo": did, + "event": { + "$type": "org.tangled.event.push", + "repo": did, + "ref": update.name(), + "old": transition.old_oid().unwrap_or(null).to_hex(), + "new": transition.new_oid().unwrap_or(null).to_hex(), + // TODO: include push options + }, + }) +} + +fn get_repo_spindle(repo: &Repo) -> Result, String> { + #[derive(serde::Deserialize)] + struct Manifest { + spindle: Option, + } + let Some(tip) = chain::tip(repo).map_err(|error| error.to_string())? else { + return Ok(None); + }; + let key = format!("{MANIFEST_COLLECTION}/{MANIFEST_RKEY}"); + futures::executor::block_on(async { + let Some(cid) = chain::load_tree(repo, &tip) + .get(&key) + .await + .map_err(|error| error.to_string())? + else { + return Ok(None); + }; + let Some(bytes) = GitBlocks::at(repo, tip.blocks) + .get(&cid) + .await + .map_err(|error| error.to_string())? + else { + return Ok(None); + }; + serde_ipld_dagcbor::from_slice::(&bytes) + .map(|manifest| manifest.spindle) + .map_err(|error| error.to_string()) + }) +} -- 2.54.0