Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022use axum::body::Bytes;use base64::Engine;use base64::engine::general_purpose::URL_SAFE_NO_PAD;use futures::future::join_all;use futures::stream::StreamExt;use http::Method;use http::header::AUTHORIZATION;use serde_json::{Value, json};use std::collections::{BTreeMap, BTreeSet};use std::net::SocketAddr;use std::sync::Arc;use tower::ServiceExt;
use knot_runtime::{Entropy, K256Signer, SeededEntropy, Signer};use knot_types::{ AccountDid, Datetime, HttpStatus, KnotHostname, Nsid, RecordCollection, RepoDid, UnixSeconds,};
use crate::harness::{Harness, subject_dids};use crate::trace::{OperationIndex, Outcome, Projection, RoundNumber, Step, Trace, fnv1a};
const SKEW_BACKDATE_SECS: i64 = 600;const SKEW_LIFETIME_SECS: i64 = 60;const CREATE_RECORD_NSID: &str = "com.atproto.repo.createRecord";const DELETE_RECORD_NSID: &str = "com.atproto.repo.deleteRecord";pub(crate) fn member_collection() -> RecordCollection { RecordCollection::new("org.tangled.knot.member").expect("literal NSID parses")}
pub(crate) fn block_collection() -> RecordCollection { RecordCollection::new("org.tangled.knot.block").expect("literal NSID parses")}
pub(crate) fn collaborator_collection() -> RecordCollection { RecordCollection::new("org.tangled.graph.collaborator").expect("literal NSID parses")}
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]struct RepoIndex(usize);
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]struct SubjectIndex(usize);
#[derive(Clone, Copy)]struct StrangerIndex(usize);
#[derive(Clone, Copy)]enum ReadOp { Version, DescribeKnot, ListMembers, DidJson, InfoRefs(RepoIndex), Branches(RepoIndex), Log(RepoIndex), DescribeRepo(RepoIndex), Tree(RepoIndex), Blob(RepoIndex), Languages(RepoIndex), ListCollaborators(RepoIndex),}
impl ReadOp { fn name(self) -> &'static str { match self { ReadOp::Version => "version", ReadOp::DescribeKnot => "describeKnot", ReadOp::ListMembers => "listMembers", ReadOp::DidJson => "didJson", ReadOp::InfoRefs(_) => "infoRefs", ReadOp::Branches(_) => "branches", ReadOp::Log(_) => "log", ReadOp::DescribeRepo(_) => "describeRepo", ReadOp::Tree(_) => "tree", ReadOp::Blob(_) => "blob", ReadOp::Languages(_) => "languages", ReadOp::ListCollaborators(_) => "listCollaborators", } }}
#[derive(Clone, Copy)]enum AdminOp { AddMember(SubjectIndex), RemoveMember(SubjectIndex), Ban(SubjectIndex), Unban(SubjectIndex), CreateRepo(u32), AddCollaborator(RepoIndex, SubjectIndex),}
impl AdminOp { fn name(self) -> &'static str { match self { AdminOp::AddMember(_) => "addMember", AdminOp::RemoveMember(_) => "removeMember", AdminOp::Ban(_) => "ban", AdminOp::Unban(_) => "unban", AdminOp::CreateRepo(_) => "createRepo", AdminOp::AddCollaborator(_, _) => "addCollaborator", } }}
#[derive(Clone, Copy)]enum Mutation { Admin { op: AdminOp, skew: bool }, Probe { stranger: StrangerIndex, drop: bool }, Maintain { repo: RepoIndex },}
#[derive(Clone, Copy)]struct Observation { op: ReadOp, killed: bool,}
#[derive(Clone, Copy)]enum Planned { Mutate(Mutation), Observe(Observation),}
impl Planned { fn creates_a_repo(self) -> bool { matches!( self, Planned::Mutate(Mutation::Admin { op: AdminOp::CreateRepo(_), skew: false, }) ) }}
enum Phase { Mutate(Vec<Mutation>), Observe(Vec<Observation>),}
impl Phase { fn ops(&self) -> Vec<Planned> { match self { Phase::Mutate(ops) => ops.iter().copied().map(Planned::Mutate).collect(), Phase::Observe(ops) => ops.iter().copied().map(Planned::Observe).collect(), } }}
pub(crate) struct Round { phase: Phase, advance: std::time::Duration,}
pub(crate) struct Rng(SeededEntropy);
impl Rng { pub(crate) fn new(seed: u64) -> Self { Self(SeededEntropy::new(seed ^ 0x57ee_d000)) }
pub(crate) fn below(&self, n: u64) -> u64 { self.0.next_u64() % n.max(1) }
pub(crate) fn chance(&self, num: u64, den: u64) -> bool { assert!(num <= den, "chance numerator exceeds denominator"); self.below(den) < num }}
pub(crate) fn plan(seed: u64, rounds: u32, subjects: usize) -> Vec<Round> { let rng = Rng::new(seed); let mut available: u32 = 1; let mut rkey: u32 = 0; let mut stranger: usize = 0; (0..rounds) .map(|round| { let (phase, fresh_repos) = match round % 2 { 0 => { let (ops, fresh_repos) = mutate_round(&rng, subjects, available, &mut rkey, &mut stranger); (Phase::Mutate(ops), fresh_repos) } _ => (Phase::Observe(observe_round(&rng, available)), 0), }; let advance = std::time::Duration::from_micros(rng.below(3_000_000)); available += fresh_repos; Round { phase, advance } }) .collect()}
#[derive(Default)]struct Model { members: BTreeSet<AccountDid>, blocked: BTreeSet<AccountDid>, collaborators: BTreeMap<usize, BTreeSet<AccountDid>>,}
pub(crate) fn predict(seed: u64, rounds: u32) -> Projection { let subjects = subject_dids(); let subject_did = |subject: SubjectIndex| subjects[subject.0].clone(); let plan = plan(seed, rounds, subjects.len()); let ops = || plan.iter().flat_map(|round| round.phase.ops()); let repos = 1 + ops().filter(|planned| planned.creates_a_repo()).count(); let model = ops().fold(Model::default(), |mut model, planned| { match planned { Planned::Mutate(Mutation::Admin { op, skew: false }) => match op { AdminOp::AddMember(subject) => { model.members.insert(subject_did(subject)); } AdminOp::RemoveMember(subject) => { model.members.remove(&subject_did(subject)); } AdminOp::Ban(subject) => { model.blocked.insert(subject_did(subject)); } AdminOp::Unban(subject) => { model.blocked.remove(&subject_did(subject)); } AdminOp::AddCollaborator(repo, subject) => { model .collaborators .entry(repo.0) .or_default() .insert(subject_did(subject)); } AdminOp::CreateRepo(_) => {} }, Planned::Mutate(Mutation::Admin { skew: true, .. }) | Planned::Mutate(Mutation::Probe { .. }) | Planned::Mutate(Mutation::Maintain { .. }) | Planned::Observe(_) => {} } model }); let mut rosters = model.collaborators; Projection { members: model.members.into_iter().collect(), blocked: model.blocked.into_iter().collect(), collaborators: (0..repos) .map(|repo| { rosters .remove(&repo) .unwrap_or_default() .into_iter() .collect() }) .collect(), }}
fn mutate_round( rng: &Rng, subjects: usize, available: u32, rkey: &mut u32, stranger: &mut usize,) -> (Vec<Mutation>, u32) { let mut ops: Vec<Mutation> = (0..subjects) .filter(|_| rng.chance(2, 3)) .map(|subject| { let subject = SubjectIndex(subject); let op = match rng.below(4) { 0 => AdminOp::AddMember(subject), 1 => AdminOp::RemoveMember(subject), 2 => AdminOp::Ban(subject), _ => AdminOp::Unban(subject), }; Mutation::Admin { op, skew: rng.chance(1, 5), } }) .collect();
(0..available) .filter(|_| rng.chance(1, 2)) .for_each(|repo| { let repo = RepoIndex(repo as usize); let planned = if rng.chance(1, 2) { let subject = SubjectIndex(rng.below(subjects as u64) as usize); Mutation::Admin { op: AdminOp::AddCollaborator(repo, subject), skew: rng.chance(1, 6), } } else { Mutation::Maintain { repo } }; ops.push(planned); });
let mut fresh_repos = 0; (0..rng.below(3)).for_each(|_| { let key = *rkey; *rkey += 1; let skew = rng.chance(1, 8); if !skew { fresh_repos += 1; } ops.push(Mutation::Admin { op: AdminOp::CreateRepo(key), skew, }); });
(0..rng.below(3)).for_each(|_| { let stranger_index = StrangerIndex(*stranger); *stranger += 1; ops.push(Mutation::Probe { stranger: stranger_index, drop: rng.chance(1, 2), }); });
(ops, fresh_repos)}
fn observe_round(rng: &Rng, available: u32) -> Vec<Observation> { let mut ops: Vec<Observation> = [ ReadOp::Version, ReadOp::DescribeKnot, ReadOp::ListMembers, ReadOp::DidJson, ] .into_iter() .map(|op| Observation { op, killed: rng.chance(1, 5), }) .collect(); (0..available) .filter(|_| rng.chance(2, 3)) .for_each(|repo| { let repo = RepoIndex(repo as usize); let op = match rng.below(8) { 0 => ReadOp::Branches(repo), 1 => ReadOp::Log(repo), 2 => ReadOp::DescribeRepo(repo), 3 => ReadOp::InfoRefs(repo), 4 => ReadOp::Tree(repo), 5 => ReadOp::Blob(repo), 6 => ReadOp::ListCollaborators(repo), _ => ReadOp::Languages(repo), }; ops.push(Observation { op, killed: rng.chance(1, 5), }); }); ops}
struct OpResult { step: Step, created: Option<RepoDid>,}
pub(crate) async fn execute(harness: Arc<Harness>, seed: u64, plan: Vec<Round>) -> Trace { let initial = ( vec![harness.seed_repo.clone()], Vec::<Step>::new(), Vec::new(), ); let harness = &harness; let (repos, steps, snapshots) = futures::stream::iter(plan.into_iter().enumerate()) .fold( initial, |(repos, mut steps, mut snapshots), (round_index, round)| { let harness = Arc::clone(harness); async move { let round_no = RoundNumber::new(round_index as u32); let ops = round.phase.ops(); let drops = arm_drops(&harness, &ops); let repos = Arc::new(repos); let tasks = ops .iter() .enumerate() .map(|(index, planned)| { let harness = Arc::clone(&harness); let repos = Arc::clone(&repos); let planned = *planned; tokio::spawn(async move { run_op( &harness, &repos, round_no, OperationIndex::new(index as u32), planned, ) .await }) }) .collect::<Vec<_>>(); let results: Vec<OpResult> = join_all(tasks) .await .into_iter() .map(|joined| joined.expect("sim op task mustn't panic")) .collect(); drops .iter() .for_each(|host| harness.faults.clear_host(host));
let created: Vec<RepoDid> = results .iter() .filter_map(|result| result.created.clone()) .collect(); let planned_creates = ops .iter() .filter(|planned| planned.creates_a_repo()) .count(); assert_eq!( planned_creates, created.len(), "round {}: {planned_creates} non-skew creates planned but \ {} materialized, so plan/execute repo indices have drifted apart", round_no.get(), created.len() ); created.iter().for_each(|did| harness.populate(did)); steps.extend(results.into_iter().map(|result| result.step)); let mut repos = Arc::into_inner(repos) .expect("all op tasks released the round repo snapshot"); repos.extend(created);
harness.advance(round.advance); snapshots.push(harness.snapshot(round_no, &repos)); (repos, steps, snapshots) } }, ) .await; let no_fault_creates = steps .iter() .filter(|step| step.op == "createRepo" && step.fault == "none") .count(); let materialized = repos.len() - 1; assert_eq!( no_fault_creates, materialized, "no-fault createRepo count {no_fault_creates} doesn't match the {materialized} repos \ materialized: a planned create silently failed and repo_at would have masked the drift" ); Trace { seed, steps, snapshots, }}
fn arm_drops(harness: &Harness, ops: &[Planned]) -> Vec<KnotHostname> { let hosts: Vec<KnotHostname> = ops .iter() .filter_map(|planned| match planned { Planned::Mutate(Mutation::Probe { stranger, drop: true, }) => Some(harness.strangers[stranger.0].host.clone()), _ => None, }) .collect(); hosts.iter().for_each(|host| harness.faults.drop_host(host)); hosts}
async fn run_op( harness: &Harness, repos: &[RepoDid], round: RoundNumber, index: OperationIndex, planned: Planned,) -> OpResult { let make = |op: &'static str, actor: String, fault: &'static str, outcome: Outcome| Step { round, index, op, actor, fault, outcome, };
match planned { Planned::Mutate(Mutation::Maintain { repo }) => { let repo = repo_at(repos, repo); OpResult { step: make( "maintain", "knot".to_string(), "none", maintain_outcome(harness, repo), ), created: None, } } Planned::Observe(Observation { op, killed }) => { let request = read_request(harness, repos, op); if killed { drive_kill(harness.router(), request.method, &request.uri, request.body).await; return OpResult { step: make(op.name(), request.actor, "killed", Outcome::Killed), created: None, }; } let (status, body) = http_call( harness.router(), request.method, &request.uri, None, request.body, ) .await; OpResult { step: make( op.name(), request.actor, "none", Outcome::Served { status, body: body_digest(&body), }, ), created: None, } } Planned::Mutate(Mutation::Admin { op, skew }) => { let request = admin_request(harness, repos, op, skew, round, index); let (status, body) = http_call( harness.router(), request.method, &request.uri, request.token.as_deref(), request.body, ) .await; let created = match op { AdminOp::CreateRepo(_) if status == HttpStatus::new(200) => repo_did_of(&body), _ => None, }; OpResult { step: make( op.name(), request.actor, if skew { "clock_skew" } else { "none" }, Outcome::Served { status, body: body_digest(&body), }, ), created, } } Planned::Mutate(Mutation::Probe { stranger, drop }) => { let request = probe_request(harness, stranger, round, index); let (status, body) = http_call( harness.router(), request.method, &request.uri, request.token.as_deref(), request.body, ) .await; OpResult { step: make( "probe", request.actor, if drop { "drop_identity" } else { "none" }, Outcome::Served { status, body: body_digest(&body), }, ), created: None, } } }}
pub(crate) struct Request { pub(crate) method: Method, pub(crate) uri: String, pub(crate) token: Option<String>, pub(crate) body: Bytes, pub(crate) actor: String,}
impl Request { pub(crate) fn get(uri: impl Into<String>) -> Self { Self { method: Method::GET, uri: uri.into(), token: None, body: Bytes::new(), actor: "anon".to_string(), } }}
fn read_request(harness: &Harness, repos: &[RepoDid], op: ReadOp) -> Request { match op { ReadOp::Version => get("/xrpc/org.tangled.knot.describeKnot"), ReadOp::DescribeKnot => get("/xrpc/org.tangled.knot.describeKnot"), ReadOp::ListMembers => get(format!( "/xrpc/com.atproto.repo.listRecords?repo={}&collection={}", enc(&harness.knot_aud), member_collection() ) .as_str()), ReadOp::DidJson => get("/.well-known/did.json"), ReadOp::InfoRefs(repo) => Request { method: Method::GET, uri: format!( "/{}/info/refs?service=git-upload-pack", repo_at(repos, repo) ), token: None, body: Bytes::new(), actor: "anon".to_string(), }, ReadOp::Branches(repo) => repo_get("org.tangled.git.listBranches", repo_at(repos, repo)), ReadOp::Log(repo) => get(format!( "/xrpc/org.tangled.git.listCommits?repo={}&ref=HEAD", enc(repo_at(repos, repo)) ) .as_str()), ReadOp::DescribeRepo(repo) => { repo_get("com.atproto.repo.describeRepo", repo_at(repos, repo)) } ReadOp::Tree(repo) => repo_get("org.tangled.git.getTree", repo_at(repos, repo)), ReadOp::Languages(repo) => repo_get("org.tangled.git.listLanguages", repo_at(repos, repo)), ReadOp::ListCollaborators(repo) => get(format!( "/xrpc/com.atproto.repo.listRecords?repo={}&collection={}", enc(repo_at(repos, repo)), collaborator_collection() ) .as_str()), ReadOp::Blob(repo) => Request { method: Method::GET, uri: format!( "/xrpc/org.tangled.git.getBlob?repo={}&path=README.md&ref=HEAD", enc(repo_at(repos, repo)) ), token: None, body: Bytes::new(), actor: "anon".to_string(), }, }}
fn admin_request( harness: &Harness, repos: &[RepoDid], op: AdminOp, skew: bool, round: RoundNumber, index: OperationIndex,) -> Request { let subjects = &harness.subjects; let now = iso_utc(harness.now_seconds()); match op { AdminOp::AddMember(subject) => admin_post( harness, "addMember", &nsid(CREATE_RECORD_NSID), &harness.knot_aud, grant_body( harness.knot_repo(), &member_collection(), &subjects[subject.0], Some(&now), ), OpSlot { skew, round, index }, ), AdminOp::RemoveMember(subject) => admin_post( harness, "removeMember", &nsid(DELETE_RECORD_NSID), &harness.knot_aud, grant_body( harness.knot_repo(), &member_collection(), &subjects[subject.0], None, ), OpSlot { skew, round, index }, ), AdminOp::Ban(subject) => admin_post( harness, "ban", &nsid(CREATE_RECORD_NSID), &harness.knot_aud, grant_body( harness.knot_repo(), &block_collection(), &subjects[subject.0], Some(&now), ), OpSlot { skew, round, index }, ), AdminOp::Unban(subject) => admin_post( harness, "unban", &nsid(DELETE_RECORD_NSID), &harness.knot_aud, grant_body( harness.knot_repo(), &block_collection(), &subjects[subject.0], None, ), OpSlot { skew, round, index }, ), AdminOp::CreateRepo(key) => { let name = format!("repo{key}"); admin_post( harness, "create", &nsid("org.tangled.repo.createRepo"), &harness.knot_aud, json!({ "name": name }), OpSlot { skew, round, index }, ) } AdminOp::AddCollaborator(repo, subject) => admin_post( harness, "addCollaborator", &nsid(CREATE_RECORD_NSID), &repo_at(repos, repo), grant_body( repo_at(repos, repo), &collaborator_collection(), &subjects[subject.0], Some(&now), ), OpSlot { skew, round, index }, ), }}
fn probe_request( harness: &Harness, stranger: StrangerIndex, round: RoundNumber, index: OperationIndex,) -> Request { let actor = &harness.strangers[stranger.0]; let mut request = signed_post( harness, &actor.signer, &actor.did, &nsid(CREATE_RECORD_NSID), &harness.knot_aud, grant_body( harness.knot_repo(), &member_collection(), &harness.subjects[0], Some(&iso_utc(harness.now_seconds())), ), OpSlot { skew: false, round, index, }, ); request.actor = actor.host.to_string(); request}
fn get(path: &str) -> Request { Request::get(path)}
fn repo_get(method: &str, repo: &RepoDid) -> Request { Request::get(format!("/xrpc/{method}?repo={}", enc(repo)))}
fn nsid(name: &str) -> Nsid { Nsid::new_owned(name).expect("the xrpc method is an NSID")}
pub(crate) fn signed_post( harness: &Harness, signer: &K256Signer, did: &AccountDid, method: &Nsid, aud: &impl AsRef<str>, body: Value, OpSlot { skew, round, index }: OpSlot,) -> Request { let token = mint( signer, did, aud, method, jwt_window(harness, skew), round, index, ); Request { method: Method::POST, uri: format!("/xrpc/{method}"), token: Some(token), body: encode_body(body), actor: String::new(), }}
pub(crate) fn maintain_outcome(harness: &Harness, repo: &RepoDid) -> Outcome { match harness.maintain(repo) { Ok(()) => Outcome::Served { status: HttpStatus::new(200), body: 0, }, Err(error) => Outcome::Served { status: HttpStatus::new(500), body: fnv1a(error.to_string().as_bytes()), }, }}
pub(crate) struct OpSlot { pub(crate) skew: bool, pub(crate) round: RoundNumber, pub(crate) index: OperationIndex,}
fn admin_post( harness: &Harness, method_short: &str, method: &Nsid, aud: &impl AsRef<str>, body: Value, OpSlot { skew, round, index }: OpSlot,) -> Request { let mut request = signed_post( harness, &harness.admin.signer, &harness.admin.did, method, aud, body, OpSlot { skew, round, index }, ); request.actor = format!("admin:{method_short}"); request}
pub(crate) fn jwt_window(harness: &Harness, skew: bool) -> (UnixSeconds, UnixSeconds) { let now = harness.now_seconds(); if skew { ( now.saturating_sub_secs(SKEW_BACKDATE_SECS + SKEW_LIFETIME_SECS), now.saturating_sub_secs(SKEW_BACKDATE_SECS), ) } else { (now, now.saturating_add_secs(60)) }}
pub(crate) fn mint( signer: &K256Signer, issuer: &AccountDid, aud: &impl AsRef<str>, nsid: &Nsid, window: (UnixSeconds, UnixSeconds), round: RoundNumber, index: OperationIndex,) -> String { let header = URL_SAFE_NO_PAD.encode(br#"{"alg":"ES256K","typ":"JWT"}"#); let payload = URL_SAFE_NO_PAD.encode( serde_json::to_vec(&json!({ "iss": issuer, "aud": aud.as_ref(), "iat": window.0.get(), "exp": window.1.get(), "jti": format!("sim-{}-{}", round.get(), index.get()), "lxm": nsid, })) .expect("claims serialize"), ); let signing_input = format!("{header}.{payload}"); let signature = signer.sign(signing_input.as_bytes()); format!( "{signing_input}.{}", URL_SAFE_NO_PAD.encode(signature.as_bytes()) )}
pub(crate) async fn http_call( router: axum::Router, method: Method, uri: &str, token: Option<&str>, body: Bytes,) -> (HttpStatus, Bytes) { let mut request = http::Request::builder() .method(method) .uri(uri) .body(axum::body::Body::from(body)) .expect("request builds"); if let Some(token) = token { request.headers_mut().insert( AUTHORIZATION, http::HeaderValue::from_str(&format!("Bearer {token}")).expect("bearer header"), ); } request .extensions_mut() .insert(axum::extract::ConnectInfo(SocketAddr::from(( [127, 0, 0, 1], 4242, )))); let response = router .oneshot(request) .await .expect("router must return a response"); let status = HttpStatus::from(response.status()); let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) .await .expect("response body"); (status, bytes)}
pub(crate) async fn drive_kill(router: axum::Router, method: Method, uri: &str, body: Bytes) { let call = http_call(router, method, uri, None, body); futures::pin_mut!(call); tokio::select! { biased; _ = &mut call => {} _ = tokio::task::yield_now() => {} }}
pub(crate) fn repo_did_of(body: &Bytes) -> Option<RepoDid> { serde_json::from_slice::<Value>(body) .ok() .and_then(|value| value.get("did").and_then(Value::as_str).map(str::to_string)) .and_then(|did| RepoDid::new(did).ok())}
pub(crate) fn body_digest(body: &Bytes) -> u64 { match serde_json::from_slice::<Value>(body) { Ok(mut value) => { canonicalize(&mut value); fnv1a(&serde_json::to_vec(&value).expect("canonical body serializes")) } Err(_) => fnv1a(body), }}
fn canonicalize(value: &mut Value) { match value { Value::Array(items) => { items.iter_mut().for_each(canonicalize); items.sort_by_cached_key(|item| serde_json::to_string(item).expect("array item")); } Value::Object(map) => map.values_mut().for_each(canonicalize), _ => {} }}
pub(crate) fn encode_body(value: Value) -> Bytes { Bytes::from(serde_json::to_vec(&value).expect("request body serializes"))}
pub(crate) fn enc(value: impl std::fmt::Display) -> String { value.to_string().replace(':', "%3A")}
pub(crate) fn grant_body( repo: &RepoDid, collection: &RecordCollection, rkey: &AccountDid, created_at: Option<&Datetime>,) -> Value { match created_at { Some(now) => json!({ "repo": repo, "collection": collection, "rkey": rkey, "record": { "createdAt": now } }), None => json!({ "repo": repo, "collection": collection, "rkey": rkey, }), }}
pub(crate) fn iso_utc(seconds: UnixSeconds) -> Datetime { let seconds = seconds.get(); let (days, secs_of_day) = (seconds.div_euclid(86_400), seconds.rem_euclid(86_400)); let z = days + 719_468; let era = z.div_euclid(146_097); let doe = z - era * 146_097; let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); let mp = (5 * doy + 2) / 153; let day = doy - (153 * mp + 2) / 5 + 1; let (month, year) = if mp < 10 { (mp + 3, yoe + era * 400) } else { (mp - 9, yoe + era * 400 + 1) }; Datetime::raw_str(format!( "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}Z", secs_of_day / 3600, secs_of_day % 3600 / 60, secs_of_day % 60 ))}
fn repo_at(repos: &[RepoDid], index: RepoIndex) -> &RepoDid { repos.get(index.0).unwrap_or_else(|| { panic!( "plan/execute repo drift: index {} exceeds {} repos created so far", index.0, repos.len() ) })}