diff --git a/crates/vibescrobble-index/tests/deploy/cold.rs b/crates/vibescrobble-index/tests/deploy/cold.rs new file mode 100644 index 00000000..c41fe248 --- /dev/null +++ b/crates/vibescrobble-index/tests/deploy/cold.rs @@ -0,0 +1,156 @@ +//! A start with no saved state reads the relay; one with a saved state takes +//! it back. Either way the board a reader is served, with each account's +//! kind, burn and weeks, is the one the network holds. A restored start that +//! lost every file beside `state.json`, or kept one written before kinds +//! were, ends up there too once its start-up passes have run. + +use std::collections::BTreeMap; + +use serde_json::Value; + +use crate::node::{self, start, start_keeping, DataDir, Node}; +use crate::world::{World, ACCOUNTS, KEEP_DORMANT}; + +/// The process that was running: a cold start, the stream, and its passes, +/// stopped on `dir`. +async fn ran(dir: &DataDir, world: &std::sync::Arc, keep_dormant: usize) { + let first = start_keeping(dir, world, keep_dormant).await; + first.read_relay().await; + first.follow(world.stream(1_000)).await; + world.advance(); + first.sweep().await; + first.fill().await; + first.stop().await; +} + +/// A start with no files at all, on the network as it is now. +async fn cold(dir: &DataDir, world: &std::sync::Arc, keep_dormant: usize) -> Node { + let cold = start_keeping(dir, world, keep_dormant).await; + assert!(!cold.restored); + cold.read_relay().await; + cold.kinds().await; + cold.sweep().await; + cold.fill().await; + cold +} + +#[tokio::test] +async fn a_cold_start_and_a_restored_start_serve_the_same_board() { + let world = World::new(); + let kept = DataDir::new("cold-kept"); + ran(&kept, &world, KEEP_DORMANT).await; + + let restored = start(&kept, &world).await; + assert!(restored.restored); + restored.kinds().await; + restored.fill().await; + let empty = DataDir::new("cold-empty"); + let cold = cold(&empty, &world, KEEP_DORMANT).await; + + let served = cold.surfaces().await; + let kinds: Vec<&Value> = served["/view?all"]["body"]["agents"] + .as_array() + .expect("agents") + .iter() + .map(|agent| &agent["registeredKind"]) + .collect(); + assert_eq!( + kinds.iter().filter(|kind| kind.is_string()).count(), + 4, + "every held didbot account has its kind: {kinds:?}" + ); + node::same(&served, &restored.surfaces().await, "cold, then restored"); +} + +/// A start with `state.json` alone, written before kinds were kept: the +/// burn store and the ledgers are lost, or were never there. +#[tokio::test] +async fn a_start_from_the_state_alone_comes_to_what_a_cold_start_serves() { + let world = World::new(); + let kept = DataDir::new("cold-state-alone"); + // With room for every dormant account: one let go for room is in + // neither the state nor the repositories the fill reads. + ran(&kept, &world, ACCOUNTS.len()).await; + lose(&kept, &["burn.json", "burn.json.journal", "ledgers.sqlite"]); + let path = kept.file("state.json"); + let mut saved: Value = + serde_json::from_slice(&std::fs::read(&path).expect("the state")).expect("json"); + for agent in saved["view"]["agents"].as_array_mut().expect("agents") { + agent + .as_object_mut() + .expect("an agent") + .remove("registeredKind"); + } + std::fs::write(&path, saved.to_string()).expect("written back"); + + let empty = DataDir::new("cold-state-alone-empty"); + let cold = cold(&empty, &world, ACCOUNTS.len()).await; + let served = cold.surfaces().await; + + // Every line the board holds inside the burn window is counted from the + // first request; what is older is the fill pass's. + let restored = start_keeping(&kept, &world, ACCOUNTS.len()).await; + assert!(restored.restored); + node::same( + &recent_burn(&served), + &recent_burn(&restored.surfaces().await), + "restored, before the fill", + ); + + restored.kinds().await; + restored.fill().await; + node::same( + &served, + &restored.surfaces().await, + "cold, then restored from the state alone", + ); +} + +/// The burn store let go, as one in an older format is at start-up, beside +/// a board that has let an account go for room. +#[tokio::test] +#[ignore = "issue 3myrrithsun2l: the burn of an account let go for room is not read back"] +async fn a_burn_store_let_go_is_rebuilt_with_the_accounts_let_go_for_room() { + let world = World::new(); + let kept = DataDir::new("cold-burn-let-go"); + ran(&kept, &world, KEEP_DORMANT).await; + lose(&kept, &["burn.json.journal"]); + std::fs::write(kept.file("burn.json"), r#"{"version":1}"#).expect("an older store"); + + let restored = start(&kept, &world).await; + restored.fill().await; + let empty = DataDir::new("cold-burn-let-go-empty"); + let cold = cold(&empty, &world, KEEP_DORMANT).await; + + node::same( + &cold.surfaces().await, + &restored.surfaces().await, + "cold, then restored with its burn store let go", + ); +} + +/// Every burn total in `served` over the last week or less. +fn recent_burn(served: &BTreeMap) -> BTreeMap { + served + .iter() + .filter(|(path, _)| { + (path.starts_with("/leaderboard") && !path.ends_with("window=all")) + || path.starts_with("/usage") + }) + .map(|(path, answer)| { + let mut answer = answer.clone(); + for whose in ["own", "subtree"] { + if let Some(spans) = answer["body"][whose].as_object_mut() { + spans.remove("all"); + } + } + (path.clone(), answer) + }) + .collect() +} + +fn lose(dir: &DataDir, files: &[&str]) { + for lost in files { + std::fs::remove_file(dir.file(lost)).expect("a file the run wrote"); + } +} diff --git a/crates/vibescrobble-index/tests/deploy/main.rs b/crates/vibescrobble-index/tests/deploy/main.rs new file mode 100644 index 00000000..7c36b58e --- /dev/null +++ b/crates/vibescrobble-index/tests/deploy/main.rs @@ -0,0 +1,13 @@ +//! What a deploy does to the board: an index stops, and the next one starts +//! on the same data directory, or with none, or with one an older or newer +//! binary wrote. +//! +//! [`node`] starts and stops an index on a directory as the binary does, and +//! [`world`] is the network it reads: a relay, one didbot server and the +//! accounts on it. Each test drives a deploy through both and compares what +//! a reader is served. + +mod cold; +mod node; +mod restart; +mod world; diff --git a/crates/vibescrobble-index/tests/deploy/node.rs b/crates/vibescrobble-index/tests/deploy/node.rs new file mode 100644 index 00000000..b1518dcc --- /dev/null +++ b/crates/vibescrobble-index/tests/deploy/node.rs @@ -0,0 +1,309 @@ +//! One index process on a data directory: started as `run_jetstream` starts +//! one, with every file a deploy keeps, and stopped as it stops one. + +use std::collections::BTreeMap; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +use serde_json::Value; +use tokio::sync::mpsc; +use vibescrobble_index::activity::DEFAULT_PRUNE; +use vibescrobble_index::backfill::{from_relay, Relisting}; +use vibescrobble_index::burn::{Burn, Store, RETENTION_DAYS}; +use vibescrobble_index::identity::{Identities, Registry}; +use vibescrobble_index::jetstream::Update; +use vibescrobble_index::state::{self, Stream}; +use vibescrobble_index::survey::Retry; +use vibescrobble_index::view::HISTORY; +use vibescrobble_index::worker::Limits; +use vibescrobble_index::{app, moderation, relay, DidResolver, Indexer, Shared}; + +use crate::world::{self, World, ACCOUNTS, HUMAN, JETSTREAM, KEEP_DORMANT}; + +/// A data directory, removed when dropped. +#[derive(Debug)] +pub struct DataDir(PathBuf); + +impl DataDir { + pub fn new(name: &str) -> Self { + static MADE: AtomicUsize = AtomicUsize::new(0); + let path = std::env::temp_dir().join(format!( + "vibescrobble-deploy-{}-{}-{name}", + std::process::id(), + MADE.fetch_add(1, Ordering::Relaxed) + )); + let _ = std::fs::remove_dir_all(&path); + std::fs::create_dir_all(&path).expect("a data directory"); + Self(path) + } + + pub fn path(&self) -> &Path { + &self.0 + } + + pub fn file(&self, name: &str) -> PathBuf { + self.0.join(name) + } +} + +impl Drop for DataDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +/// The files the binary is pointed at, as vibescrobble.com names them. +pub const FILES: [&str; 5] = [ + "state.json", + "ledgers.sqlite", + "identities.jsonl", + "burn.json", + "burn.json.journal", +]; + +pub struct Node { + pub shared: Shared, + pub registry: Arc>>, + world: Arc, + dir: PathBuf, + /// Whether a saved state was taken back. + pub restored: bool, +} + +/// `run` and `run_jetstream`'s start-up, in their order: the burn store, the +/// ledgers, the denylist, the prune window, the identities, and the saved +/// state. +pub async fn start(dir: &DataDir, world: &Arc) -> Node { + start_keeping(dir, world, KEEP_DORMANT).await +} + +/// [`start`], keeping `keep_dormant` dormant accounts. +pub async fn start_keeping(dir: &DataDir, world: &Arc, keep_dormant: usize) -> Node { + let burn = Store::open(&dir.file("burn.json"), RETENTION_DAYS).expect("the burn store opens"); + let shared = Shared::new().with_burn(Burn::Local(Arc::new(burn))); + let ledgers = vibescrobble_index::ledgers::Store::disk(&dir.file("ledgers.sqlite")) + .expect("the ledgers open"); + shared.view.lock().await.use_ledgers(ledgers); + moderation::reload(&shared, &dir.file("denylist.txt"), &mut None).await; + let announced = { + let mut view = shared.view.lock().await; + view.set_prune(Some(DEFAULT_PRUNE)); + view.set_keep_dormant(Some(keep_dormant)); + view.sweep(time::OffsetDateTime::now_utc()) + }; + shared.announce(announced).await; + let identities = Identities::open(dir.file("identities.jsonl")).expect("the identities open"); + let registry = Arc::new(Registry::new( + DidResolver::new(Arc::clone(world)), + identities, + )); + let restored = match state::read(&dir.file("state.json")) { + Ok(saved) => { + state::restore(&shared, ®istry, saved, &[(Stream::Relay, JETSTREAM)]).await; + true + } + Err(_) => false, + }; + Node { + shared, + registry, + world: Arc::clone(world), + dir: dir.path().to_owned(), + restored, + } +} + +impl Node { + /// What the binary does when it is stopped: the board saved, and the + /// open week written. + pub async fn stop(self) { + let saved = + state::gather(&self.shared, &self.registry, &[(Stream::Relay, JETSTREAM)]).await; + state::write(&self.dir.join("state.json"), &saved).expect("the state is written"); + self.shared.view.lock().await.flush_ledgers(); + } + + /// Follows the stream for as long as it carries `updates`, from where + /// the saved state was when one was taken back. + pub async fn follow(&self, updates: Vec) { + let (events, incoming) = mpsc::channel(updates.len().max(1)); + for update in updates { + events.send(update).await.expect("the loop is listening"); + } + drop(events); + let start = if self.restored { + relay::Start::Restored + } else { + relay::Start::Cold + }; + relay::follow_from( + self.shared.clone(), + incoming, + Arc::clone(&self.registry), + JETSTREAM.to_owned(), + JETSTREAM.to_owned(), + None, + None, + Limits::default(), + start, + ) + .await; + } + + /// Reads the relay's population in, as a cold start does before the + /// stream is applied. + pub async fn read_relay(&self) { + from_relay( + &self.shared, + &self.registry, + &self.relisting(), + JETSTREAM, + None, + ) + .await; + } + + fn relisting(&self) -> Relisting> { + Relisting { + indexer: Indexer::new(Arc::clone(&self.world)), + relay: world::relay(), + repositories: 100, + records: HISTORY, + retry: Retry::NEVER, + again: Retry::NEVER, + } + } + + /// One pass of the ledger and burn fill, as `fill::run` makes a minute + /// after start-up. + pub async fn fill(&self) -> vibescrobble_index::fill::Filled { + vibescrobble_index::fill::fill( + &self.shared, + &self.registry, + &Indexer::new(Arc::clone(&self.world)), + std::time::Duration::ZERO, + vibescrobble_index::fill::PAGES, + time::OffsetDateTime::now_utc(), + ) + .await + } + + /// The pass every start runs beside the stream, reading the + /// registration of each held account without its kind. + pub async fn kinds(&self) { + vibescrobble_index::backfill::kinds( + &self.shared, + &self.registry, + std::time::Duration::ZERO, + ) + .await; + } + + /// One beat of the prune loop. + pub async fn sweep(&self) { + let announced = self + .shared + .view + .lock() + .await + .sweep(time::OffsetDateTime::now_utc()); + self.shared.announce(announced).await; + } + + /// Everything a reader can ask the index about the accounts in + /// [`ACCOUNTS`] and their operator, by path, as status and body. + pub async fn surfaces(&self) -> BTreeMap { + let year = time::OffsetDateTime::now_utc().year(); + let mut paths = vec!["/view".to_owned(), "/view?all".to_owned()]; + for by in ["operator", "agent", "model", "vendor", "repo"] { + for window in ["hour", "day", "week", "all"] { + paths.push(format!("/leaderboard?by={by}&window={window}")); + } + } + for did in ACCOUNTS.iter().chain([&HUMAN]) { + paths.push(format!("/usage/{did}")); + paths.push(format!("/ledgers/{did}")); + paths.push(format!("/ledgers/{did}/y/{year}")); + } + paths.push(format!("/usage/operator/{HUMAN}")); + serve(&self.shared, &paths).await + } +} + +/// Asks the index's router each of `paths`, over a socket, as a reader would. +pub async fn serve(shared: &Shared, paths: &[String]) -> BTreeMap { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("a port"); + let addr: SocketAddr = listener.local_addr().expect("an address"); + let router = app(shared.clone()); + let server = tokio::spawn(async move { + vibescrobble_index::service::serve(listener, router) + .await + .ok(); + }); + let client = reqwest::Client::new(); + let mut answers = BTreeMap::new(); + for path in paths { + let response = client + .get(format!("http://{addr}{path}")) + .send() + .await + .expect("the index answers"); + let status = response.status().as_u16(); + let mut body: Value = response.json().await.unwrap_or(Value::Null); + // The stream's cursor names the process's run, which every start + // changes, so a reader's stream seeds again. + if let Some(view) = body.as_object_mut() { + view.remove("cursor"); + } + answers.insert( + path.clone(), + serde_json::json!({ "status": status, "body": body }), + ); + } + server.abort(); + answers +} + +/// Asserts `after` serves what `before` did, naming where each path differs. +pub fn same(before: &BTreeMap, after: &BTreeMap, what: &str) { + let mut differ = Vec::new(); + for (path, body) in before { + differences( + path, + body, + after.get(path).unwrap_or(&Value::Null), + &mut differ, + ); + } + assert!(differ.is_empty(), "{what}:\n{}", differ.join("\n")); +} + +fn differences(at: &str, before: &Value, after: &Value, differ: &mut Vec) { + match (before, after) { + (Value::Object(before), Value::Object(after)) => { + for key in before + .keys() + .chain(after.keys().filter(|key| !before.contains_key(*key))) + { + let null = Value::Null; + differences( + &format!("{at} {key}"), + before.get(key).unwrap_or(&null), + after.get(key).unwrap_or(&null), + differ, + ); + } + } + (Value::Array(one), Value::Array(other)) if one.len() == other.len() => { + for (n, (before, after)) in one.iter().zip(other).enumerate() { + differences(&format!("{at} {n}"), before, after, differ); + } + } + _ if before != after => differ.push(format!("{at}: {before} became {after}")), + _ => {} + } +} diff --git a/crates/vibescrobble-index/tests/deploy/restart.rs b/crates/vibescrobble-index/tests/deploy/restart.rs new file mode 100644 index 00000000..2d4a2d9d --- /dev/null +++ b/crates/vibescrobble-index/tests/deploy/restart.rs @@ -0,0 +1,104 @@ +//! A deploy stops the index and starts the next one on the same data +//! directory. Every total a reader is served is the same on either side of +//! it: the board, each account's burn over every span, the leaderboards by +//! every key, and the weekly ledgers, for accounts shown, dormant and let go +//! for room alike. + +use vibescrobble_index::state::Stream; + +use crate::node::{self, start, DataDir, FILES}; +use crate::world::{World, DORMANT, HOST, HUMAN, LONER, RELEASED, SESSION, SUBAGENT}; + +#[tokio::test] +async fn a_restart_on_the_same_data_dir_serves_every_total_it_served_before() { + let world = World::new(); + let dir = DataDir::new("restart"); + + // A cold start: the relay's population, then the stream, then the + // prune loop and the fill pass. + let first = start(&dir, &world).await; + assert!(!first.restored); + first.read_relay().await; + first.follow(world.stream(1_000)).await; + world.advance(); + first.sweep().await; + first.fill().await; + let before = first.surfaces().await; + + // What the board holds before the restart, so a total that is lost + // both times is not taken for one that is kept. + let shown = before["/view"]["body"]["agents"] + .as_array() + .expect("agents"); + assert_eq!(shown.len(), 4, "the four accounts that spoke today"); + let held = before["/view?all"]["body"]["agents"] + .as_array() + .expect("agents"); + assert!(held.iter().any(|agent| agent["did"] == DORMANT)); + assert!( + !held.iter().any(|agent| agent["did"] == RELEASED), + "let go for room" + ); + for did in [HOST, SESSION, LONER, DORMANT] { + let usage = &before[&format!("/usage/{did}")]; + assert_eq!(usage["status"], 200, "{did}: {usage}"); + assert_eq!( + usage["body"]["own"]["all"]["output"], + world.output_of(did), + "{did}: {usage}" + ); + } + let ranked = before["/leaderboard?by=agent&window=all"]["body"]["rows"] + .as_array() + .expect("rows"); + let released = ranked.iter().find(|row| row["key"] == RELEASED); + assert_eq!( + released.map(|row| row["output"].clone()), + Some(world.output_of(RELEASED).into()), + "an account let go for room is still ranked" + ); + let operated: u64 = [HOST, SESSION, SUBAGENT, DORMANT, RELEASED] + .map(|did| world.output_of(did)) + .iter() + .sum(); + assert_eq!( + before[&format!("/usage/operator/{HUMAN}")]["body"]["subtree"]["all"]["output"], + operated, + "the operator's total holds every account in the tree" + ); + let weeks = before[&format!("/ledgers/{SESSION}")]["body"] + .as_array() + .map_or(0, Vec::len); + assert_eq!( + weeks, + world.weeks_of(SESSION), + "the session's weeks, two of them before the burn window" + ); + first.stop().await; + for file in FILES { + assert!(dir.file(file).exists(), "{file} is written"); + } + + // The next process, on the same files, with the stream not yet resumed. + world.asked_for(""); + let second = start(&dir, &world).await; + assert!(second.restored); + node::same(&before, &second.surfaces().await, "after a restart"); + let cursor = second.shared.followed.cursor(Stream::Relay); + assert!(cursor.is_some(), "the stream resumes where it was"); + + // Its fill pass has nothing to read: every account was read whole. + assert_eq!(second.fill().await.reads, 0, "pages read"); + let read = world.asked_for("listRecords"); + assert!(read.is_empty(), "no repository is read again: {read:?}"); + + // And a second restart, from what the restored process wrote. + second.stop().await; + let third = start(&dir, &world).await; + node::same(&before, &third.surfaces().await, "after a second restart"); + assert_eq!( + third.shared.followed.cursor(Stream::Relay), + cursor, + "a process stopped before the stream said anything keeps its cursor" + ); +} diff --git a/crates/vibescrobble-index/tests/deploy/world.rs b/crates/vibescrobble-index/tests/deploy/world.rs new file mode 100644 index 00000000..df71b52f --- /dev/null +++ b/crates/vibescrobble-index/tests/deploy/world.rs @@ -0,0 +1,480 @@ +//! The network an index reads: a relay listing every repository, the didbot +//! server holding them, each account's DID document, and the person who +//! operates them. It answers from the scrobbles in [`LINES`], dated back +//! from the moment it was made. +//! +//! A few lines arrive on the stream rather than being there at start-up, and +//! one is deleted on it. [`World::stream`] is those, as Jetstream carries +//! them, and after [`World::advance`] the repositories hold what the stream +//! said. + +use std::future::Future; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use serde_json::{json, Value}; +use time::{Duration, OffsetDateTime}; +use vibescrobble_index::http::{HttpGet, HttpResponse, TransportError}; +use vibescrobble_index::jetstream::{Commit, Update}; +use vibescrobble_index::{AgentDid, ServerRef}; + +pub const DEV: &str = "http://localhost:3000"; +pub const RELAY: &str = "http://relay.localhost:2470"; +pub const JETSTREAM: &str = "ws://jetstream.localhost:8080"; +const SCROBBLE: &str = "com.vibescrobble.scrobble"; + +/// The person every registration names. +pub const HUMAN: &str = "did:plc:kq3vv6mbfqhzpldiwtoeyrxs"; +const HUMAN_PDS: &str = "https://pds.elsewhere.example"; + +/// A host the human created. +pub const HOST: &str = "did:web:quarry.agents.localhost%3A3000"; +/// A session under it, and the busiest account. +pub const SESSION: &str = "did:web:sluice.agents.localhost%3A3000"; +/// A subagent under the session. +pub const SUBAGENT: &str = "did:web:tamarack.agents.localhost%3A3000"; +/// A session that went quiet two days ago: dormant, and held. +pub const DORMANT: &str = "did:web:fescue.agents.localhost%3A3000"; +/// A session that went quiet six days ago, let go for room. +pub const RELEASED: &str = "did:web:bracken.agents.localhost%3A3000"; +/// A bot that is no didbot agent: a `did:plc`, with no registration. +pub const LONER: &str = "did:plc:wicketwicketwicketwicket"; + +/// Every account, in the order the relay lists them: children first. +pub const ACCOUNTS: [&str; 6] = [SUBAGENT, RELEASED, SESSION, LONER, DORMANT, HOST]; + +/// How many dormant accounts the board keeps: one of the two goes. +pub const KEEP_DORMANT: usize = 1; + +/// How an account came to be, as its registration says. +fn lineage(did: &str) -> Option<&'static [&'static str]> { + match did { + HOST => Some(&[]), + SESSION | DORMANT | RELEASED => Some(&[HOST]), + SUBAGENT => Some(&[HOST, SESSION]), + _ => None, + } +} + +/// One scrobble. +#[derive(Debug, Clone, Copy)] +struct Line { + did: &'static str, + /// Minutes before the world was made. + ago: i64, + model: Option<&'static str>, + repo: Option<&'static str>, + /// Output tokens it says it spent, when it says. + output: Option, + /// Written on the stream after start-up, rather than there already. + streamed: bool, + /// Deleted on the stream. + deleted: bool, +} + +const fn line(did: &'static str, ago: i64, model: &'static str, output: u64) -> Line { + Line { + did, + ago, + model: Some(model), + repo: None, + output: Some(output), + streamed: false, + deleted: false, + } +} + +const MINUTE: i64 = 1; +const HOUR: i64 = 60; +const DAY: i64 = 24 * HOUR; +const OPUS: &str = "claude-opus-5-5"; +const SONNET: &str = "claude-sonnet-5"; +const GPT: &str = "gpt-5.2"; +const KILN: &str = "https://tangled.org/permadeath.com/kiln"; +const QUERN: &str = "https://github.com/quernstone/quern"; + +/// Lines in the last hour, the last day, the last week, this burn store's +/// window and before it, so every span a total is asked over holds some and +/// some are folded into dailies. None is near a span's edge. +const LINES: &[Line] = &[ + line(HOST, 10 * MINUTE, OPUS, 100), + line(HOST, 3 * DAY, OPUS, 200), + Line { + repo: Some(KILN), + ..line(SESSION, 20 * MINUTE, OPUS, 1_000) + }, + Line { + repo: Some(KILN), + ..line(SESSION, 2 * HOUR, OPUS, 2_000) + }, + Line { + repo: Some(QUERN), + ..line(SESSION, 3 * DAY, OPUS, 3_000) + }, + line(SESSION, 10 * DAY, SONNET, 4_000), + line(SESSION, 40 * DAY, OPUS, 5_000), + Line { + model: None, + output: None, + ..line(SESSION, 50 * DAY, OPUS, 0) + }, + Line { + deleted: true, + ..line(SESSION, 5 * HOUR, OPUS, 9_999) + }, + Line { + streamed: true, + repo: Some(KILN), + ..line(SESSION, 5 * MINUTE, OPUS, 500) + }, + Line { + repo: Some(QUERN), + ..line(SUBAGENT, 30 * MINUTE, SONNET, 300) + }, + line(SUBAGENT, 5 * DAY, SONNET, 600), + line(LONER, 15 * MINUTE, GPT, 7_000), + line(LONER, 12 * DAY, GPT, 8_000), + Line { + streamed: true, + ..line(LONER, 4 * MINUTE, GPT, 50) + }, + line(DORMANT, 2 * DAY, SONNET, 900), + line(DORMANT, 9 * DAY, SONNET, 1_100), + line(RELEASED, 6 * DAY, OPUS, 1_300), + line(RELEASED, 20 * DAY, OPUS, 1_700), +]; + +/// The network, as one table of answers and a note of what was asked. +#[derive(Debug)] +pub struct World { + now: OffsetDateTime, + lines: Vec, + /// Whether the repositories hold what the stream said. + advanced: AtomicBool, + /// Accounts whose server says they are taken down. + taken_down: Mutex>, + asked: Mutex>, +} + +impl World { + pub fn new() -> Arc { + Arc::new(Self { + now: OffsetDateTime::now_utc(), + lines: LINES.to_vec(), + advanced: AtomicBool::new(false), + taken_down: Mutex::default(), + asked: Mutex::default(), + }) + } + + /// The repositories hold what [`Self::stream`] said from now on. + pub fn advance(&self) { + self.advanced.store(true, Ordering::SeqCst); + } + + /// What was asked whose URL holds `part`, since the last time. + pub fn asked_for(&self, part: &str) -> Vec { + let mut asked = self.asked.lock().expect("the list"); + let matching = asked + .iter() + .filter(|url| url.contains(part)) + .cloned() + .collect(); + asked.clear(); + matching + } + + fn at(&self, line: &Line) -> OffsetDateTime { + self.now - Duration::minutes(line.ago) + } + + /// The lines a repository holds now. + fn held(&self) -> impl Iterator { + let advanced = self.advanced.load(Ordering::SeqCst); + self.lines.iter().filter(move |line| { + if advanced { + !line.deleted + } else { + !line.streamed + } + }) + } + + /// Output tokens `did`'s lines say they spent, all of them. + pub fn output_of(&self, did: &str) -> u64 { + self.lines + .iter() + .filter(|line| line.did == did && !line.deleted) + .filter_map(|line| line.output) + .sum() + } + + /// How many ISO weeks `did`'s lines fall in. + pub fn weeks_of(&self, did: &str) -> usize { + self.lines + .iter() + .filter(|line| line.did == did && !line.deleted) + .map(|line| { + let (year, week, _) = self.at(line).to_iso_week_date(); + (year, week) + }) + .collect::>() + .len() + } + + /// What the stream carries after start-up: the lines written then, and + /// the delete. + pub fn stream(&self, first: u64) -> Vec { + let mut updates: Vec<(OffsetDateTime, Update)> = Vec::new(); + for line in self.lines.iter().filter(|line| line.streamed) { + let at = self.at(line); + updates.push(( + at, + commit( + line.did, + &tid(at), + at, + Some(record(self, line)["value"].clone()), + ), + )); + } + for line in self.lines.iter().filter(|line| line.deleted) { + let at = self.deleted_at(); + updates.push((at, commit(line.did, &tid(self.at(line)), at, None))); + } + updates.sort_by_key(|(at, _)| *at); + updates + .into_iter() + .enumerate() + .map(|(n, (_, mut update))| { + if let Update::Commit(commit) = &mut update { + commit.seq = first + n as u64; + } + update + }) + .collect() + } + + fn answer(&self, url: &str) -> Option { + let parsed = url::Url::parse(url).ok()?; + let query = |name: &str| { + parsed + .query_pairs() + .find(|(key, _)| key == name) + .map(|(_, value)| value.into_owned()) + }; + let host = parsed.host_str()?; + let method = parsed.path().strip_prefix("/xrpc/"); + if host == "relay.localhost" && method == Some("com.atproto.sync.listRepos") { + let taken_down = self.taken_down.lock().expect("the list").clone(); + let repos: Vec = ACCOUNTS + .iter() + .map(|did| { + let mut repo = json!({ + "did": did, + "head": format!("bafy-{did}"), + "rev": tid(self.last(did)), + }); + if taken_down.contains(did) { + repo["active"] = json!(false); + repo["status"] = json!("takendown"); + } + repo + }) + .collect(); + return Some(json!({ "repos": repos })); + } + if host == "plc.directory" { + let did = parsed.path().trim_start_matches('/'); + return match did { + HUMAN => Some(document(HUMAN, "operator.example", HUMAN_PDS)), + LONER => Some(document(LONER, "wicket.example", DEV)), + _ => None, + }; + } + if parsed.path() == "/.well-known/did.json" { + let did = ACCOUNTS + .into_iter() + .find(|did| did.starts_with("did:web:") && document_url(did) == url)?; + let label = did.trim_start_matches("did:web:").split('.').next()?; + return Some(document(did, &format!("{label}.agents.localhost"), DEV)); + } + let repo = query("repo"); + let collection = query("collection"); + match (method?, repo.as_deref(), collection.as_deref()) { + ("com.atproto.repo.getRecord", Some(HUMAN), Some("bot.did.operator")) => Some(json!({ + "uri": format!("at://{HUMAN}/bot.did.operator/localhost"), + "value": { + "$type": "bot.did.operator", + "subject": "did:web:localhost", + "createdAt": "2026-08-24T15:00:00Z", + }, + })), + ("com.atproto.repo.getRecord", Some(did), Some("bot.did.registration")) => { + let lineage = lineage(did)?; + let mut value = json!({ + "$type": "bot.did.registration", + "did": did, + "operator": HUMAN, + "kind": if lineage.is_empty() { "host" } else { "agent" }, + "createdAt": "2026-08-24T16:00:00Z", + }); + if !lineage.is_empty() { + value["lineage"] = json!(lineage); + } + Some( + json!({ "uri": format!("at://{did}/bot.did.registration/self"), "value": value }), + ) + } + ("com.atproto.repo.getRecord", Some(did), Some("app.bsky.actor.profile")) => { + Some(json!({ + "uri": format!("at://{did}/app.bsky.actor.profile/self"), + "value": { + "$type": "app.bsky.actor.profile", + "displayName": "flittermouse", + "labels": { + "$type": "com.atproto.label.defs#selfLabels", + "values": [{ "val": "bot" }], + }, + }, + })) + } + ("com.atproto.repo.listRecords", Some(did), Some(SCROBBLE)) => { + let limit: usize = query("limit").and_then(|limit| limit.parse().ok())?; + let cursor = query("cursor"); + let mut lines: Vec<(String, &Line)> = self + .held() + .filter(|line| line.did == did) + .map(|line| (tid(self.at(line)), line)) + .filter(|(key, _)| cursor.as_ref().is_none_or(|cursor| key < cursor)) + .collect(); + lines.sort_by(|a, b| b.0.cmp(&a.0)); + let more = lines.len() > limit; + lines.truncate(limit); + let mut page = json!({ + "records": lines.iter().map(|(_, line)| record(self, line)).collect::>(), + }); + if more { + page["cursor"] = json!(lines.last().map(|(key, _)| key.clone())); + } + Some(page) + } + ("com.atproto.repo.listRecords", Some(_), Some(_)) => Some(json!({ "records": [] })), + _ => None, + } + } + + /// When `did` last committed: a line, or a delete on the stream. + fn last(&self, did: &str) -> OffsetDateTime { + let advanced = self.advanced.load(Ordering::SeqCst); + let deleted = self + .lines + .iter() + .filter(|line| advanced && line.deleted && line.did == did) + .map(|_| self.deleted_at()); + self.held() + .filter(|line| line.did == did) + .map(|line| self.at(line)) + .chain(deleted) + .max() + .unwrap_or(self.now - Duration::days(400)) + } + + /// When the stream deletes a line. + fn deleted_at(&self) -> OffsetDateTime { + self.now - Duration::minutes(3) + } +} + +impl HttpGet for World { + fn get(&self, url: &str) -> impl Future> + Send { + self.asked.lock().expect("the list").push(url.to_owned()); + let answer = match self.answer(url) { + Some(body) => HttpResponse::new(200, body.to_string()), + None => HttpResponse::new(404, r#"{"error":"NotFound"}"#), + }; + async move { Ok(answer) } + } +} + +/// `line` as `listRecords` serves it. +fn record(world: &World, line: &Line) -> Value { + let at = world.at(line); + let mut value = json!({ + "$type": SCROBBLE, + "text": format!("quernstone, {} minutes ago", line.ago), + "emoji": "\u{1f9ed}", + "createdAt": format(at), + }); + if let Some(model) = line.model { + value["model"] = json!(model); + } + if let Some(output) = line.output { + value["usage"] = json!({ "input": 10, "output": output, "cacheRead": 20, "cacheWrite": 0 }); + } + if let Some(repo) = line.repo { + value["repo"] = json!({ "url": repo }); + } + json!({ + "uri": format!("at://{}/{SCROBBLE}/{}", line.did, tid(at)), + "cid": format!("bafy-{}", tid(at)), + "value": value, + }) +} + +/// A write as Jetstream carries it, numbered later. +fn commit(did: &str, rkey: &str, at: OffsetDateTime, record: Option) -> Update { + Update::Commit(Commit { + seq: 0, + did: did.to_owned(), + time: format(at), + rev: tid(at), + operation: if record.is_some() { "create" } else { "delete" }.to_owned(), + collection: SCROBBLE.to_owned(), + rkey: rkey.to_owned(), + cid: record.as_ref().map(|_| format!("bafy-{rkey}")), + record, + }) +} + +fn document(did: &str, handle: &str, pds: &str) -> Value { + json!({ + "@context": ["https://www.w3.org/ns/did/v1"], + "id": did, + "alsoKnownAs": [format!("at://{handle}")], + "service": [{ + "id": format!("{did}#atproto_pds"), + "type": "AtprotoPersonalDataServer", + "serviceEndpoint": pds, + }], + }) +} + +fn document_url(did: &str) -> String { + AgentDid::parse(did) + .expect("a valid did:web") + .did_json_url() +} + +fn format(at: OffsetDateTime) -> String { + at.format(&time::format_description::well_known::Rfc3339) + .expect("a time") +} + +/// A record key or revision made at `at`, as a TID. +pub fn tid(at: OffsetDateTime) -> String { + const ALPHABET: &[u8] = b"234567abcdefghijklmnopqrstuvwxyz"; + let micros = (at.unix_timestamp_nanos() / 1000) as u64; + let mut value = micros << 10; + let mut out = [b'2'; 13]; + for slot in out.iter_mut().rev() { + *slot = ALPHABET[(value % 32) as usize]; + value /= 32; + } + String::from_utf8(out.to_vec()).expect("ascii") +} + +/// The relay, as a deployment names it. +pub fn relay() -> ServerRef { + ServerRef::parse(RELAY).expect("a valid base url") +}