Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
11 kB · 313 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314//! owner-side record emission: `knot-static repo record <repoDid>`.
use std::sync::Arc;
use anyhow::Context as _;use http::StatusCode;use knot_runtime::HttpTransport;use knot_types::RepoDid;
use crate::config::ValidatedConfig;use crate::registry::format_rfc3339;
pub async fn record( config: Arc<ValidatedConfig>, repo_did: &str, publish: bool,) -> anyhow::Result<()> { let repo_did = RepoDid::new(repo_did.to_string()).context("invalid repo DID")?; let plan = build_record_plan(&config, &repo_did)?; let RecordPlan { owner: owner_str, payload, .. } = &plan;
println!("{}", serde_json::to_string_pretty(&payload)?);
if publish { let owner_account = knot_types::AccountDid::new(owner_str.clone()) .context("invalid owner DID in plan")?; let mut session = crate::oauth::ensure_auth_for_owner(&config, &owner_account, true) .await? .context("OAuth session required for publishing")?; publish_with_resync_ping(config, &repo_did, &plan, &mut session).await?; println!( "{}", serde_json::to_string_pretty(&serde_json::json!({ "published": true, }))? ); } Ok(())}
/// everything needed to emit (and optionally publish) one repo's record:/// rkey/name/createdAt/owner from the registry COB, falling back to the index.pub(crate) struct RecordPlan { pub rkey: String, pub owner: String, pub payload: serde_json::Value,}
fn build_record_plan( config: &Arc<ValidatedConfig>, repo_did: &RepoDid,) -> anyhow::Result<RecordPlan> { let layout = config.layout()?; let meta_path = layout.meta_path(&config.knot_did)?; let meta_repo = layout .bootstrap_meta(&config.knot_did) .context("open meta.git")?;
let cob_store = knot_cob::CobStore::new(&meta_repo); let cob_objects = cob_store.list::<knot_cobs::RepoRegistryCob>()?; let registry_state = cob_objects .first() .map(|&object| cob_store.get::<knot_cobs::RepoRegistryCob>(object)) .transpose()? .map(|cob| cob.into_state());
let (rkey_str, name_str, created_at_iso, owner_str) = if let Some(record) = registry_state .as_ref() .and_then(|reg| reg.record_of(repo_did)) { ( record.rkey.to_string(), record.name.to_string(), chrono::DateTime::from_timestamp(record.created_at.get(), 0) .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)) .unwrap_or_else(|| format_rfc3339(record.created_at.get())), record.owner.to_string(), ) } else { let index = knot_index::Index::new(&meta_path, layout); let _ = index.rebuild(); match index.ownership_of(repo_did) { knot_index::Resolved::Ready(Some(ownership)) => { let rkey = ownership.rkey.to_string(); let owner = ownership.owner.to_string(); let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_secs() as i64; // the index has no cosmetic name; fall back to the rkey (rkey.clone(), rkey, format_rfc3339(now), owner) } _ => anyhow::bail!("repo {repo_did} is not registered"), } };
let payload = serde_json::json!({ "collection": "sh.tangled.repo", "rkey": rkey_str, "record": { "$type": "sh.tangled.repo", "name": name_str, "knot": config.hostname.as_str(), // `repo` is the lexicon field (the repo DID); repoDid is kept for // compatibility with our own earlier readers. "repo": repo_did.as_str(), "repoDid": repo_did.as_str(), "createdAt": created_at_iso, } });
Ok(RecordPlan { rkey: rkey_str, owner: owner_str, payload, })}
/// re-publish changed repos' records for owners with an active OAuth session./// a static knot emits no firehose, so a record update is the only signal that/// re-enqueues a knotmirror resync.////// a re-publish with identical bytes yields an identical cid, which mirrors treat as/// no change. so when the built payload matches what's already published, createdAt is/// nudged +-1s across two writes: the record ends up semantically identical but both/// writes carry fresh cids, forcing the change through cid-deduping pipelines./// per-repo failures are logged; they never fail an activation.pub(crate) async fn publish_all(config: Arc<ValidatedConfig>, changed_repos: &[String]) { let layout = match config.layout() { Ok(layout) => layout, Err(error) => { eprintln!("warn: record publish: {error:#}"); return; } }; let meta_path = match layout.meta_path(&config.knot_did) { Ok(meta_path) => meta_path, Err(error) => { eprintln!("warn: record publish: {error:#}"); return; } }; let index = knot_index::Index::new(&meta_path, layout.clone()); let _ = index.rebuild();
let mut targets = Vec::new(); for did in index.hosted_repos() { if !changed_repos .iter() .any(|repo_did| repo_did == did.as_str()) { continue; } let knot_index::Resolved::Ready(Some(ownership)) = index.ownership_of(&did) else { continue; }; let owner_did_str = ownership.owner.to_string(); let Ok(owner_account) = knot_types::AccountDid::new(owner_did_str.clone()) else { continue; }; let plan = match build_record_plan(&config, &did) { Ok(plan) => plan, Err(error) => { eprintln!("warn: record publish {}: {error:#}", did.as_str()); continue; } }; targets.push((did, owner_account, plan)); }
if targets.is_empty() { return; }
let mut sessions: std::collections::HashMap<String, Option<crate::oauth::StoredSession>> = std::collections::HashMap::new(); for (_, owner_account, _) in &targets { let owner_key = owner_account.to_string(); if sessions.contains_key(&owner_key) { continue; } let session = match crate::oauth::get_or_refresh_session(&config, owner_account).await { Ok(Some(s)) => Some(s), Ok(None) => { eprintln!("record publish: skipped (no OAuth session for owner {owner_key})"); None } Err(error) => { eprintln!("warn: oauth session lookup for {owner_key}: {error:#}"); None } }; sessions.insert(owner_key, session); }
let publish_futures = targets.into_iter().filter_map(|(did, owner_account, plan)| { let session = sessions.get(&owner_account.to_string())?.clone()?; let config = config.clone(); Some(async move { let mut session = session; match publish_with_resync_ping(config, &did, &plan, &mut session).await { Ok(()) => eprintln!( "published sh.tangled.repo record for {} via OAuth (mirror resync requested)", did.as_str() ), Err(error) => eprintln!("warn: record publish {}: {error:#}", did.as_str()), } }) });
futures::future::join_all(publish_futures).await;}
/// publish the record; when the built payload equals what's already published,/// the +-1s createdAt nudge makes the write observable downstream.async fn publish_with_resync_ping( config: Arc<ValidatedConfig>, _did: &RepoDid, plan: &RecordPlan, session: &mut crate::oauth::StoredSession,) -> anyhow::Result<()> { let payload = plan.payload.clone(); crate::oauth::put_record_oauth( session, "sh.tangled.repo", &plan.rkey, &payload["record"], ) .await?;
let published = fetch_published_record(&config, plan).await?; if published.as_ref() != Some(&payload["record"]) { // first write was a real change: nothing was published before, or the content drifted. return Ok(()); }
// Identical bytes published: nudge createdAt +1s then restore to force fresh CIDs downstream. let original = payload["record"]["createdAt"] .as_str() .context("record payload missing createdAt")? .to_owned(); let shifted = shift_rfc3339_seconds(&original, 1).context("shift createdAt")?;
let mut nudged = payload["record"].clone(); nudged["createdAt"] = serde_json::Value::String(shifted); crate::oauth::put_record_oauth(session, "sh.tangled.repo", &plan.rkey, &nudged).await?;
let mut restored = payload["record"].clone(); restored["createdAt"] = serde_json::Value::String(original); crate::oauth::put_record_oauth(session, "sh.tangled.repo", &plan.rkey, &restored).await?; Ok(())}
/// read the currently published record from the repo owner's PDS for the no-change check.async fn fetch_published_record( config: &Arc<ValidatedConfig>, plan: &RecordPlan,) -> anyhow::Result<Option<serde_json::Value>> { let transport = knot_runtime::ReqwestHttp::new(knot_runtime::HttpLimits::default()) .context("build http transport")?; let atproto = knot_atproto::Atproto::new( knot_runtime::ReqwestHttp::new(knot_runtime::HttpLimits::default()) .context("build identity http transport")?, knot_runtime::SystemClock, config.knot_did.clone(), knot_atproto::PlcDirectory::new(config.plc_directory.clone())?, ); let owner_account = knot_types::AccountDid::new(plan.owner.clone()) .context("repo owner DID is not a valid account DID")?; let identity = atproto .resolve_identity(&owner_account) .await .context("resolve repo owner DID to find its PDS")?; let mut get_url = identity .pds .url() .join("xrpc/com.atproto.repo.getRecord") .context("build getRecord url")?; get_url .query_pairs_mut() .append_pair("repo", plan.owner.as_str()) .append_pair("collection", "sh.tangled.repo") .append_pair("rkey", &plan.rkey); let mut request = knot_runtime::HttpRequest::get(get_url); request.headers.insert( http::header::ACCEPT, http::HeaderValue::from_static("application/json"), ); let response = transport .execute(request) .await .context("getRecord request")?; match response.status { StatusCode::NOT_FOUND => Ok(None), StatusCode::OK => { let body: serde_json::Value = serde_json::from_slice(&response.body).context("parse getRecord response")?; Ok(body.get("value").cloned()) } status => anyhow::bail!("getRecord failed with HTTP {status}"), }}
/// shift an RFC3339 timestamp by whole seconds, keeping second precision.fn shift_rfc3339_seconds(stamp: &str, delta: i64) -> Option<String> { chrono::DateTime::parse_from_rfc3339(stamp).ok().map(|dt| { (dt + chrono::Duration::seconds(delta)).to_rfc3339_opts(chrono::SecondsFormat::Secs, true) })}