The ledger: hash-chained events in the frozen envelope, signed checkpoints, admin verify, check and rebuild (#3) 36537112fcac

-- The ledger (spec §9.1, §9.5): every event, in order, chained by hash.

-- Columns are the frozen envelope; `pruned` marks a stub whose payload and

-- proof were removed by `spectra admin prune` (its hash and prev stay).

CREATE TABLE events (

    seq             INTEGER PRIMARY KEY,

    id              TEXT NOT NULL UNIQUE,

    at_ms           INTEGER NOT NULL,

    hlc_wall        INTEGER NOT NULL,

    hlc_logical     INTEGER NOT NULL,

    prev            BLOB NOT NULL,

    hash            BLOB NOT NULL,

    auth_key        BLOB,

    auth_method     INTEGER NOT NULL,

    auth_assurance  INTEGER NOT NULL,

    auth_proof      BLOB,

    kind            TEXT NOT NULL,

    actor           TEXT NOT NULL,

    session         TEXT,

    project         TEXT,

    subject         TEXT,

    payload         BLOB,

    trace           TEXT,

    pruned          INTEGER NOT NULL DEFAULT 0

);

CREATE INDEX events_kind ON events(kind);

CREATE INDEX events_subject ON events(subject);

-- Facts about this database: the ledger format it was written with.

CREATE TABLE meta (

    key   TEXT PRIMARY KEY,

    value TEXT NOT NULL

);

-- Projection: how many events of each kind (Radar's Log header, `admin check`).

CREATE TABLE event_counts (

    kind  TEXT PRIMARY KEY,

    n     INTEGER NOT NULL

);

//! The event envelope (spec §9.1) and how it is encoded and hashed (§9.5).

//!

//! An event's body is deterministic CBOR with integer keys, through

//! `oxy-core`'s encoder, exactly as Oxygen encodes its objects (oxygen spec

//! §5.1); its hash is `BLAKE3("spectra1 event " || len || 0x00 || body)`, the

//! domain-separated form of oxygen spec §5.2. The hash and the proof are not

//! in the body: signatures sit outside what they sign (oxygen spec §5.4).

//!

//! The envelope is frozen from the first release: fields may be added at new

//! keys, never changed or removed, so a hub merges logs from any version.

use crate::hlc::Hlc;

use oxy_core::cbor::{self, Value};

/// The envelope's version, key 0 of the body.

pub const ENVELOPE_VERSION: u64 = 1;

/// The field names of the envelope, in order: the frozen set (spec §9.1).

pub const ENVELOPE_FIELDS: &[&str] =

    &["seq", "id", "at_ms", "hlc", "prev", "hash", "auth", "kind", "actor", "session", "project", "subject", "payload", "trace"];

/// How the event's cause was authenticated (spec §8.5).

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub enum Method {

    /// Made by Spectra itself (presence timers, checkpoints of its own);

    /// covered by the next checkpoint's signature.

    System = 0,

    /// A bearer token: possession only.

    Token = 1,

    /// A request signed with a key on disk.

    Signature = 2,

    /// A passkey assertion with user presence.

    Passkey = 3,

}

impl Method {

    pub fn from_u8(n: u8) -> Option<Method> {

        Some(match n {

            0 => Method::System,

            1 => Method::Token,

            2 => Method::Signature,

            3 => Method::Passkey,

            _ => return None,

        })

    }

    pub fn name(self) -> &'static str {

        match self {

            Method::System => "system",

            Method::Token => "token",

            Method::Signature => "signature",

            Method::Passkey => "passkey",

        }

    }

}

/// Who caused an event and the proof of it. `key` and the assurance level

/// are hashed; `proof` is carried beside the hash.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Auth {

    pub key: Option<[u8; 32]>,

    pub method: Method,

    /// 0 for system events, else 1 to 3 (spec §8.5).

    pub assurance: u8,

    pub proof: Option<Vec<u8>>,

}

impl Auth {

    /// An event Spectra makes itself.

    pub fn system() -> Auth {

        Auth { key: None, method: Method::System, assurance: 0, proof: None }

    }

}

/// An event before it is appended: what the caller decides.

#[derive(Clone, Debug)]

pub struct NewEvent {

    pub kind: String,

    pub actor: String,

    pub session: Option<String>,

    pub project: Option<String>,

    pub subject: Option<String>,

    pub payload: serde_json::Value,

    pub trace: Option<String>,

    pub auth: Auth,

}

impl NewEvent {

    /// A system event of `kind` with `payload`.

    pub fn system(kind: &str, payload: serde_json::Value) -> NewEvent {

        NewEvent {

            kind: kind.into(),

            actor: "system".into(),

            session: None,

            project: None,

            subject: None,

            payload,

            trace: None,

            auth: Auth::system(),

        }

    }

    pub fn subject(mut self, s: impl Into<String>) -> NewEvent {

        self.subject = Some(s.into());

        self

    }

    pub fn project(mut self, p: impl Into<String>) -> NewEvent {

        self.project = Some(p.into());

        self

    }

    pub fn session(mut self, s: impl Into<String>) -> NewEvent {

        self.session = Some(s.into());

        self

    }

    pub fn by(mut self, actor: impl Into<String>, auth: Auth) -> NewEvent {

        self.actor = actor.into();

        self.auth = auth;

        self

    }

}

/// An event in the ledger.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Event {

    pub seq: u64,

    pub id: String,

    pub at_ms: u64,

    pub hlc: Hlc,

    pub prev: [u8; 32],

    pub hash: [u8; 32],

    pub auth: Auth,

    pub kind: String,

    pub actor: String,

    pub session: Option<String>,

    pub project: Option<String>,

    pub subject: Option<String>,

    /// Canonical JSON (sorted keys, no whitespace), hashed as stored;

    /// `None` once pruned to a stub.

    pub payload: Option<Vec<u8>>,

    pub trace: Option<String>,

}

impl Event {

    /// The payload as JSON (`null` for a stub).

    pub fn json(&self) -> serde_json::Value {

        self.payload.as_deref().and_then(|b| serde_json::from_slice(b).ok()).unwrap_or(serde_json::Value::Null)

    }

    /// The canonical body: every envelope field but `hash` and the proof.

    pub fn body(&self) -> Vec<u8> {

        let opt = |s: &Option<String>| s.as_ref().map(|t| Value::Text(t.clone())).unwrap_or(Value::Null);

        cbor::encode(&Value::map([

            (0, Value::Uint(ENVELOPE_VERSION)),

            (1, Value::Uint(self.seq)),

            (2, Value::Text(self.id.clone())),

            (3, Value::Uint(self.at_ms)),

            (4, Value::Array(vec![Value::Uint(self.hlc.wall_ms), Value::Uint(self.hlc.logical as u64)])),

            (5, Value::Bytes(self.prev.to_vec())),

            (

                6,

                Value::map([

                    (0, self.auth.key.map(|k| Value::Bytes(k.to_vec())).unwrap_or(Value::Null)),

                    (1, Value::Uint(self.auth.method as u64)),

                    (2, Value::Uint(self.auth.assurance as u64)),

                ]),

            ),

            (7, Value::Text(self.kind.clone())),

            (8, Value::Text(self.actor.clone())),

            (9, opt(&self.session)),

            (10, opt(&self.project)),

            (11, opt(&self.subject)),

            (12, self.payload.as_ref().map(|p| Value::Bytes(p.clone())).unwrap_or(Value::Null)),

            (13, opt(&self.trace)),

        ]))

    }

    /// The hash this event's fields give; equal to `hash` unless edited.

    pub fn computed_hash(&self) -> [u8; 32] {

        hash_body(&self.body())

    }

}

/// `BLAKE3("spectra1 event " || decimal(len) || 0x00 || body)`.

pub fn hash_body(body: &[u8]) -> [u8; 32] {

    let mut h = blake3::Hasher::new();

    h.update(b"spectra1 event ");

    h.update(body.len().to_string().as_bytes());

    h.update(&[0]);

    h.update(body);

    *h.finalize().as_bytes()

}

/// JSON with object keys sorted at every level and no whitespace, so equal

/// payloads are equal bytes whatever built them.

pub fn canonical_json(v: &serde_json::Value) -> Vec<u8> {

    fn sort(v: &serde_json::Value) -> serde_json::Value {

        match v {

            serde_json::Value::Object(m) => {

                let mut keys: Vec<&String> = m.keys().collect();

                keys.sort();

                let mut out = serde_json::Map::new();

                for k in keys {

                    out.insert(k.clone(), sort(&m[k]));

                }

                serde_json::Value::Object(out)

            }

            serde_json::Value::Array(a) => serde_json::Value::Array(a.iter().map(sort).collect()),

            other => other.clone(),

        }

    }

    serde_json::to_vec(&sort(v)).expect("JSON values always serialise")

}

#[cfg(test)]

mod tests {

    use super::*;

    fn sample() -> Event {

        Event {

            seq: 7,

            id: "01J0000000000000000000000Z".into(),

            at_ms: 1_700_000_000_000,

            hlc: Hlc { wall_ms: 1_700_000_000_000, logical: 2 },

            prev: [9; 32],

            hash: [0; 32],

            auth: Auth { key: Some([1; 32]), method: Method::Signature, assurance: 2, proof: Some(vec![5, 5]) },

            kind: "message.posted".into(),

            actor: "agent:claude-oxygen".into(),

            session: None,

            project: Some("oxygen".into()),

            subject: Some("m1".into()),

            payload: Some(canonical_json(&serde_json::json!({"b": 1, "a": [{"z": 1, "y": 2}]}))),

            trace: None,

        }

    }

    #[test]

    fn every_field_but_the_proof_changes_the_hash() {

        let e = sample();

        let h = e.computed_hash();

        let mut p = e.clone();

        p.auth.proof = None;

        assert_eq!(p.computed_hash(), h, "the proof is outside the hash");

        type Edit = Box<dyn Fn(&mut Event)>;

        let edits: Vec<Edit> = vec![

            Box::new(|e| e.seq += 1),

            Box::new(|e| e.id.push('X')),

            Box::new(|e| e.at_ms += 1),

            Box::new(|e| e.hlc.logical += 1),

            Box::new(|e| e.prev[0] ^= 1),

            Box::new(|e| e.auth.key = None),

            Box::new(|e| e.auth.assurance = 3),

            Box::new(|e| e.auth.method = Method::Token),

            Box::new(|e| e.kind.push('x')),

            Box::new(|e| e.actor.push('x')),

            Box::new(|e| e.session = Some("s".into())),

            Box::new(|e| e.project = None),

            Box::new(|e| e.subject = None),

            Box::new(|e| e.payload = Some(b"{}".to_vec())),

            Box::new(|e| e.trace = Some("t".into())),

        ];

        for edit in edits {

            let mut x = e.clone();

            edit(&mut x);

            assert_ne!(x.computed_hash(), h);

        }

    }

    #[test]

    fn canonical_json_sorts_keys_at_every_level() {

        let a = canonical_json(&serde_json::json!({"b": 1, "a": {"d": 1, "c": 2}}));

        assert_eq!(a, br#"{"a":{"c":2,"d":1},"b":1}"#);

    }

    #[test]

    fn the_envelope_is_frozen() {

        assert_eq!(

            ENVELOPE_FIELDS.join(","),

            "seq,id,at_ms,hlc,prev,hash,auth,kind,actor,session,project,subject,payload,trace",

            "the envelope is frozen at 0.1: add fields at new keys, never change these (spec §9.1)"

        );

        assert_eq!(ENVELOPE_VERSION, 1);

    }

}

//! The hybrid logical clock on every event (spec §9.1), with Oxygen's drift

//! bound (oxygen spec §6.4): observing a remote time never moves this clock

//! more than five minutes ahead of the local wall clock.

use std::sync::Mutex;

/// How far ahead of the local clock an observed time may move ours.

pub const MAX_DRIFT_MS: u64 = 5 * 60 * 1000;

/// A hybrid logical clock reading: wall milliseconds and a counter.

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]

pub struct Hlc {

    pub wall_ms: u64,

    pub logical: u32,

}

/// The clock: monotonic across calls, whatever the wall clock does.

#[derive(Debug, Default)]

pub struct Clock {

    last: Mutex<Hlc>,

}

impl Clock {

    /// A clock that continues from `last` (the ledger's newest reading).

    pub fn starting_at(last: Hlc) -> Clock {

        Clock { last: Mutex::new(last) }

    }

    /// The next reading for a local event, at wall time `now_ms`.

    pub fn tick(&self, now_ms: u64) -> Hlc {

        let mut last = self.last.lock().unwrap();

        *last = if now_ms > last.wall_ms {

            Hlc { wall_ms: now_ms, logical: 0 }

        } else {

            Hlc { wall_ms: last.wall_ms, logical: last.logical.saturating_add(1) }

        };

        *last

    }

    /// Take a remote reading into account, as on receiving another

    /// Spectra's event; the remote reading is kept as it is, only this

    /// clock's own advance is bounded.

    pub fn observe(&self, remote: Hlc, now_ms: u64) {

        let mut last = self.last.lock().unwrap();

        let bound = now_ms.saturating_add(MAX_DRIFT_MS);

        let remote = if remote.wall_ms > bound { Hlc { wall_ms: bound, logical: 0 } } else { remote };

        if remote > *last {

            *last = remote;

        }

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn ticks_are_monotonic_when_the_wall_clock_stalls_or_steps_back() {

        let c = Clock::default();

        let a = c.tick(1_000);

        let b = c.tick(1_000);

        let d = c.tick(500);

        assert!(a < b && b < d);

        assert_eq!(c.tick(2_000), Hlc { wall_ms: 2_000, logical: 0 });

    }

    #[test]

    fn a_far_future_remote_time_moves_the_clock_by_the_drift_bound_at_most() {

        let c = Clock::default();

        c.observe(Hlc { wall_ms: 10_000_000_000, logical: 7 }, 1_000);

        assert_eq!(c.tick(1_000).wall_ms, 1_000 + MAX_DRIFT_MS);

    }

}

//! Identifiers (spec §9.4): ULIDs, 26 characters of Crockford base32, time

//! ordered and globally unique, so an id made on one machine merges into a

//! hub without renaming.

use rand::RngCore;

const CROCKFORD: &[u8; 32] = b"0123456789ABCDEFGHJKMNPQRSTVWXYZ";

/// A new ULID for the current time.

pub fn ulid() -> String {

    let mut r = [0u8; 16];

    rand::rngs::OsRng.fill_bytes(&mut r);

    ulid_from(now_ms(), u128::from_be_bytes(r))

}

/// A ULID from a millisecond time and 80 bits taken from `random`.

pub fn ulid_from(ms: u64, random: u128) -> String {

    let v: u128 = ((ms as u128 & ((1 << 48) - 1)) << 80) | (random & ((1u128 << 80) - 1));

    (0..26).map(|i| CROCKFORD[((v >> (125 - 5 * i)) & 31) as usize] as char).collect()

}

/// Milliseconds since the Unix epoch.

pub fn now_ms() -> u64 {

    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0)

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn ulids_are_26_chars_and_sort_by_time() {

        let a = ulid_from(1_000, 5);

        let b = ulid_from(2_000, 0);

        assert_eq!(a.len(), 26);

        assert!(a < b);

        assert_eq!(ulid_from(0, 0), "00000000000000000000000000");

        assert_ne!(ulid(), ulid());

    }

}

//! Keys and identities on disk, laid out exactly as `oxy` lays them out

//! (spec §8.1, §21.3): an `OXY_HOME` holds `keys/<KEYID>.secret` (private),

//! `keys/<KEYID>.identity` and `config.toml` naming the key as `author.key`.

//! Spectra writes them with `oxy-core`'s own functions, so `oxy` reads what

//! Spectra made and the other way round.

//!

//! The instance key (spec §9.5) is kept in the data directory instead, at

//! `keys/instance.secret` and `keys/instance.identity`.

use anyhow::{bail, Context, Result};

use oxy_core::durable::{write_atomic, write_atomic_private};

use oxy_core::hash::KeyId;

use oxy_core::identity::{Identity, SecretKey, Signature};

use std::fs;

use std::path::{Path, PathBuf};

/// A key pair with its identity: the instance, an agent, or the owner.

#[derive(Clone, Debug)]

pub struct KeyPair {

    pub key: KeyId,

    pub identity: Identity,

    secret: SecretKey,

}

impl KeyPair {

    /// A fresh key pair named `name`, made now.

    pub fn generate(name: &str, email: &str) -> KeyPair {

        let secret = SecretKey::generate();

        let identity = secret.identity(name, email, crate::ids::now_ms());

        KeyPair { key: identity.hash(), identity, secret }

    }

    /// Sign `message` (a Spectra domain-prefixed byte string, at least 16

    /// bytes) through Oxygen's challenge signing: the target is

    /// `BLAKE3("oxy1 challenge " || message)`, so a Spectra signature can

    /// never be taken for a change, an attestation or a policy.

    pub fn sign(&self, message: &[u8]) -> Signature {

        self.secret.sign_challenge(&self.key, message).expect("Spectra messages are longer than 16 bytes")

    }

    /// Make a key pair in an `OXY_HOME` laid out as `oxy identity create`

    /// lays one out, and name it the home's author.

    pub fn create_home(home: &Path, name: &str, email: &str) -> Result<KeyPair> {

        let pair = KeyPair::generate(name, email);

        let keys = home.join("keys");

        fs::create_dir_all(&keys).with_context(|| format!("creating {}", keys.display()))?;

        #[cfg(unix)]

        {

            use std::os::unix::fs::PermissionsExt;

            fs::set_permissions(home, fs::Permissions::from_mode(0o700))?;

        }

        write_atomic_private(&keys.join(format!("{}.secret", pair.key.to_hex())), &pair.secret.encode(&pair.key))?;

        write_atomic(&keys.join(format!("{}.identity", pair.key.to_hex())), &pair.identity.encode())?;

        let config = home.join("config.toml");

        oxy_core::repo::Config::set_user(&config, "author.name", name)?;

        oxy_core::repo::Config::set_user(&config, "author.email", email)?;

        oxy_core::repo::Config::set_user(&config, "author.key", &pair.key.to_hex())?;

        Ok(pair)

    }

    /// The key pair an `OXY_HOME` signs with: `author.key` from its

    /// `config.toml`, else its only secret key.

    pub fn load_home(home: &Path) -> Result<KeyPair> {

        let keys = home.join("keys");

        let key = match author_key(home)? {

            Some(k) => k,

            None => {

                let secrets: Vec<PathBuf> = fs::read_dir(&keys)

                    .with_context(|| format!("reading {}", keys.display()))?

                    .filter_map(|e| e.ok().map(|e| e.path()))

                    .filter(|p| p.extension().is_some_and(|x| x == "secret"))

                    .collect();

                match secrets.as_slice() {

                    [one] => KeyId::from_hex(one.file_stem().and_then(|s| s.to_str()).unwrap_or(""))

                        .context("a secret key file is not named by its key")?,

                    [] => bail!("{} holds no secret key: an agent home is made by `spectra agent add`", home.display()),

                    _ => bail!("{} holds several keys and its config.toml names none (author.key)", home.display()),

                }

            }

        };

        KeyPair::load(&keys.join(format!("{}.secret", key.to_hex())), &keys.join(format!("{}.identity", key.to_hex())))

    }

    /// Read a secret and its identity, checking that they belong together.

    pub fn load(secret: &Path, identity: &Path) -> Result<KeyPair> {

        let (key, secret_key) = SecretKey::decode(&fs::read(secret).with_context(|| format!("reading {}", secret.display()))?)

            .map_err(|e| anyhow::anyhow!("{}: {e}", secret.display()))?;

        let identity = Identity::decode(&fs::read(identity).with_context(|| format!("reading {}", identity.display()))?)

            .map_err(|e| anyhow::anyhow!("{}: {e}", identity.display()))?;

        if identity.hash() != key {

            bail!("{} is not the identity of {}", identity.name, secret.display());

        }

        if secret_key.identity(&identity.name, &identity.email, identity.created).ed25519_pk != identity.ed25519_pk {

            bail!("the secret key in {} does not match its identity", secret.display());

        }

        Ok(KeyPair { key, identity, secret: secret_key })

    }

    /// The data directory's instance key, made on first use (spec §9.5).

    pub fn instance(data: &Path, name: &str) -> Result<KeyPair> {

        let keys = data.join("keys");

        let (secret, identity) = (keys.join("instance.secret"), keys.join("instance.identity"));

        if secret.exists() {

            return KeyPair::load(&secret, &identity);

        }

        fs::create_dir_all(&keys)?;

        let pair = KeyPair::generate(name, "");

        write_atomic_private(&secret, &pair.secret.encode(&pair.key))?;

        write_atomic(&identity, &pair.identity.encode())?;

        Ok(pair)

    }

}

/// The identity an `OXY_HOME` names as its author (`oxy identity export`

/// without the CLI): for reading the owner's identity at `spectra init`.

pub fn home_identity(home: &Path) -> Result<Identity> {

    let key = author_key(home)?.with_context(|| format!("{} names no identity (author.key in config.toml)", home.display()))?;

    let path = home.join("keys").join(format!("{}.identity", key.to_hex()));

    Identity::decode(&fs::read(&path).with_context(|| format!("reading {}", path.display()))?)

        .map_err(|e| anyhow::anyhow!("{}: {e}", path.display()))

}

fn author_key(home: &Path) -> Result<Option<KeyId>> {

    let config = home.join("config.toml");

    let Ok(text) = fs::read_to_string(&config) else { return Ok(None) };

    let table: toml::Table = text.parse().with_context(|| format!("reading {}", config.display()))?;

    let found = table.get("author").and_then(|a| a.get("key")).or_else(|| table.get("author.key")).and_then(|v| v.as_str());

    Ok(match found {

        Some(hex) => Some(KeyId::from_hex(hex).with_context(|| format!("{}: author.key is not a key", config.display()))?),

        None => None,

    })

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn a_home_made_here_reads_back_and_signs_verifiably() {

        let dir = tempfile::tempdir().unwrap();

        let made = KeyPair::create_home(dir.path(), "claude-oxygen", "").unwrap();

        let read = KeyPair::load_home(dir.path()).unwrap();

        assert_eq!(made.key, read.key);

        assert_eq!(home_identity(dir.path()).unwrap().hash(), made.key);

        let msg = b"spectra1 test message of some length";

        let sig = read.sign(msg);

        oxy_core::identity::verify_challenge(&made.identity, &sig, msg).unwrap();

        assert!(oxy_core::identity::verify_challenge(&made.identity, &sig, b"spectra1 another message entirely").is_err());

    }

    #[test]

    fn the_instance_key_is_made_once() {

        let dir = tempfile::tempdir().unwrap();

        let a = KeyPair::instance(dir.path(), "laptop").unwrap();

        let b = KeyPair::instance(dir.path(), "laptop").unwrap();

        assert_eq!(a.key, b.key);

    }

}

//! The ledger (spec §9, §9.5): the append-only, hash-chained event log in

//! SQLite, its checkpoints signed by the instance key, its verification,

//! and the bus that tells everything else what was appended.

//!

//! One writer at a time (a lock around the chain head and SQLite's own),

//! so `seq` is gap-free and every `prev` is the hash of the event before.

//! Readers never wait for it (WAL).

use crate::event::{canonical_json, Auth, Event, Method, NewEvent};

use crate::hlc::{Clock, Hlc};

use crate::keys::KeyPair;

use crate::{ids, projections};

use anyhow::{bail, Context, Result};

use oxy_core::hash::KeyId;

use oxy_core::identity::{Identity, Signature};

use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteRow, SqliteSynchronous};

use sqlx::{Row, SqlitePool};

use std::path::Path;

use std::sync::Arc;

use tokio::sync::{broadcast, Mutex};

/// Events between checkpoints (spec §9.5).

pub const CHECKPOINT_EVERY: u64 = 256;

/// The longest a written event waits for a checkpoint (spec §9.5).

pub const CHECKPOINT_INTERVAL_MS: u64 = 10 * 60 * 1000;

const CHECKPOINT_DOMAIN: &[u8] = b"spectra1 checkpoint ";

/// The ledger: the database, the chain head, the clock and the bus.

pub struct Ledger {

    pool: SqlitePool,

    head: Mutex<Head>,

    clock: Clock,

    bus: broadcast::Sender<Arc<Event>>,

    instance: Option<KeyPair>,

}

#[derive(Clone, Copy, Debug)]

struct Head {

    seq: u64,

    hash: [u8; 32],

    since_checkpoint: u64,

    oldest_unchecked_ms: Option<u64>,

}

impl Ledger {

    /// Open (or create) `path`, apply migrations, and refuse a ledger

    /// written by a newer format. Checkpoints are signed with `instance`;

    /// without one (tests, read-only tools) none are made.

    pub async fn open(path: &Path, instance: Option<KeyPair>) -> Result<Ledger> {

        let opts = SqliteConnectOptions::new()

            .filename(path)

            .create_if_missing(true)

            .journal_mode(SqliteJournalMode::Wal)

            .synchronous(SqliteSynchronous::Normal)

            .foreign_keys(true)

            .busy_timeout(std::time::Duration::from_secs(10));

        let pool =

            SqlitePoolOptions::new().max_connections(8).connect_with(opts).await.with_context(|| format!("opening {}", path.display()))?;

        sqlx::migrate!("./migrations").run(&pool).await.context("applying migrations")?;

        let format: Option<String> = sqlx::query_scalar("SELECT value FROM meta WHERE key = 'ledger_format'").fetch_optional(&pool).await?;

        match format.as_deref().map(str::parse::<u32>) {

            None => {

                sqlx::query("INSERT INTO meta (key, value) VALUES ('ledger_format', ?)")

                    .bind(crate::LEDGER_FORMAT.to_string())

                    .execute(&pool)

                    .await?;

            }

            Some(Ok(f)) if f <= crate::LEDGER_FORMAT => {}

            Some(_) => {

                bail!("{} was written by a newer Spectra (ledger format {}): upgrade spectra", path.display(), format.unwrap_or_default())

            }

        }

        let last =

            sqlx::query("SELECT seq, hash, hlc_wall, hlc_logical FROM events ORDER BY seq DESC LIMIT 1").fetch_optional(&pool).await?;

        let (seq, hash, hlc) = match last {

            Some(r) => (

                r.get::<i64, _>(0) as u64,

                to32(r.get::<Vec<u8>, _>(1))?,

                Hlc { wall_ms: r.get::<i64, _>(2) as u64, logical: r.get::<i64, _>(3) as u32 },

            ),

            None => (0, [0; 32], Hlc::default()),

        };

        let checkpoint_seq: i64 =

            sqlx::query_scalar("SELECT COALESCE(MAX(seq), 0) FROM events WHERE kind = 'ledger.checkpoint'").fetch_one(&pool).await?;

        let since = seq.saturating_sub(checkpoint_seq as u64);

        let (bus, _) = broadcast::channel(4096);

        Ok(Ledger {

            pool,

            head: Mutex::new(Head { seq, hash, since_checkpoint: since, oldest_unchecked_ms: (since > 0).then(ids::now_ms) }),

            clock: Clock::starting_at(hlc),

            bus,

            instance,

        })

    }

    /// The database, for projections' readers.

    pub fn pool(&self) -> &SqlitePool {

        &self.pool

    }

    /// Every appended event, as it commits.

    pub fn subscribe(&self) -> broadcast::Receiver<Arc<Event>> {

        self.bus.subscribe()

    }

    /// The instance key, when this ledger signs checkpoints.

    pub fn instance(&self) -> Option<&KeyPair> {

        self.instance.as_ref()

    }

    /// The newest event's sequence number and hash.

    pub async fn head(&self) -> (u64, [u8; 32]) {

        let h = self.head.lock().await;

        (h.seq, h.hash)

    }

    /// Append `events` in one transaction, chained and projected, then

    /// publish them; a checkpoint follows when one is due.

    pub async fn append(&self, events: Vec<NewEvent>) -> Result<Vec<Arc<Event>>> {

        let mut head = self.head.lock().await;

        let mut tx = self.pool.begin().await?;

        let mut out = Vec::with_capacity(events.len() + 1);

        let mut h = *head;

        for n in events {

            let e = self.chain(&mut h, n);

            insert(&mut tx, &e).await?;

            projections::apply(&mut tx, &e).await?;

            h.since_checkpoint += 1;

            h.oldest_unchecked_ms.get_or_insert(e.at_ms);

            out.push(e);

            // Every 256 events exactly, even inside a large batch.

            if h.since_checkpoint >= CHECKPOINT_EVERY && self.instance.is_some() {

                let cp = self.checkpoint_event(&mut h);

                insert(&mut tx, &cp).await?;

                projections::apply(&mut tx, &cp).await?;

                out.push(cp);

            }

        }

        let due = h.oldest_unchecked_ms.is_some_and(|t| ids::now_ms().saturating_sub(t) >= CHECKPOINT_INTERVAL_MS);

        if due && self.instance.is_some() {

            let cp = self.checkpoint_event(&mut h);

            insert(&mut tx, &cp).await?;

            projections::apply(&mut tx, &cp).await?;

            out.push(cp);

        }

        tx.commit().await?;

        *head = h;

        drop(head);

        let out: Vec<Arc<Event>> = out.into_iter().map(Arc::new).collect();

        for e in &out {

            let _ = self.bus.send(e.clone());

        }

        Ok(out)

    }

    /// Append one event.

    pub async fn append_one(&self, e: NewEvent) -> Result<Arc<Event>> {

        Ok(self.append(vec![e]).await?.remove(0))

    }

    /// Write a checkpoint now if any event is not covered by one (at

    /// shutdown, and from the ten-minute timer).

    pub async fn checkpoint(&self) -> Result<Option<Arc<Event>>> {

        let Some(_) = &self.instance else { return Ok(None) };

        let mut head = self.head.lock().await;

        if head.since_checkpoint == 0 {

            return Ok(None);

        }

        let mut h = *head;

        let cp = self.checkpoint_event(&mut h);

        let mut tx = self.pool.begin().await?;

        insert(&mut tx, &cp).await?;

        projections::apply(&mut tx, &cp).await?;

        tx.commit().await?;

        *head = h;

        drop(head);

        let cp = Arc::new(cp);

        let _ = self.bus.send(cp.clone());

        Ok(Some(cp))

    }

    fn chain(&self, h: &mut Head, n: NewEvent) -> Event {

        let at_ms = ids::now_ms();

        let mut e = Event {

            seq: h.seq + 1,

            id: ids::ulid(),

            at_ms,

            hlc: self.clock.tick(at_ms),

            prev: h.hash,

            hash: [0; 32],

            auth: n.auth,

            kind: n.kind,

            actor: n.actor,

            session: n.session,

            project: n.project,

            subject: n.subject,

            payload: Some(canonical_json(&n.payload)),

            trace: n.trace,

        };

        e.hash = e.computed_hash();

        h.seq = e.seq;

        h.hash = e.hash;

        e

    }

    /// A checkpoint naming the head `h` holds, signed by the instance key

    /// over `"spectra1 checkpoint " || hash || seq` (spec §9.5).

    fn checkpoint_event(&self, h: &mut Head) -> Event {

        let instance = self.instance.as_ref().expect("checked by callers");

        let (seq, hash, count) = (h.seq, h.hash, h.since_checkpoint);

        let sig = instance.sign(&checkpoint_message(seq, &hash));

        let n = NewEvent::system("ledger.checkpoint", serde_json::json!({ "seq": seq, "hash": hex::encode(hash), "count": count })).by(

            "system",

            Auth {

                key: Some(instance.key.0),

                method: Method::Signature,

                assurance: 2,

                proof: Some(oxy_core::cbor::encode(&sig.to_value())),

            },

        );

        let e = self.chain(h, n);

        h.since_checkpoint = 0;

        h.oldest_unchecked_ms = None;

        e

    }

    /// Events from `from` (inclusive), oldest first, at most `limit`.

    pub async fn events(&self, from: u64, limit: u32) -> Result<Vec<Event>> {

        let rows = sqlx::query(&format!("{SELECT_EVENT} WHERE seq >= ? ORDER BY seq LIMIT ?"))

            .bind(from as i64)

            .bind(limit as i64)

            .fetch_all(&self.pool)

            .await?;

        rows.iter().map(row_to_event).collect()

    }

    /// Recompute the chain and check every checkpoint and every proof it

    /// can, from the ledger alone (spec §9.5, `spectra admin verify`).

    pub async fn verify(&self, keys: &dyn KeyResolver) -> Result<Verification> {

        let mut v = Verification::default();

        let mut prev = [0u8; 32];

        let mut expect = 1u64;

        let mut instance: Option<Identity> = None;

        let mut from = 1;

        loop {

            let batch = self.events(from, 2_000).await?;

            if batch.is_empty() {

                break;

            }

            for e in &batch {

                v.events += 1;

                if e.seq != expect {

                    v.finding(e.seq, Problem::Gap, format!("expected event {expect}"));

                    expect = e.seq;

                }

                if e.prev != prev {

                    v.finding(e.seq, Problem::ChainBreak, "prev is not the hash of the event before".into());

                }

                if e.payload.is_some() && e.computed_hash() != e.hash {

                    v.finding(e.seq, Problem::HashMismatch, "the event's fields do not give its hash: it was edited".into());

                }

                if e.kind == "identity.instance" {

                    instance = e

                        .json()

                        .get("identity")

                        .and_then(|h| h.as_str())

                        .and_then(|h| hex::decode(h).ok())

                        .and_then(|b| Identity::decode(&b).ok());

                }

                if e.kind == "ledger.checkpoint" {

                    v.checkpoints += 1;

                    match check_checkpoint(e, prev, instance.as_ref()) {

                        Ok(()) => v.last_checkpoint = Some(e.seq),

                        Err(why) => v.finding(e.seq, Problem::BadCheckpoint, why),

                    }

                } else if e.auth.method == Method::Signature || e.auth.method == Method::Passkey {

                    if let Err(why) = keys.check(e) {

                        v.finding(e.seq, Problem::BadProof, why);

                    }

                }

                prev = e.hash;

                expect += 1;

            }

            from = batch.last().map(|e| e.seq + 1).unwrap_or(from);

        }

        v.head = prev;

        v.uncovered = v.events.saturating_sub(v.last_checkpoint.unwrap_or(0));

        Ok(v)

    }

    /// Clear every projection and replay the ledger into them.

    pub async fn rebuild(&self) -> Result<()> {

        let _head = self.head.lock().await;

        let mut tx = self.pool.begin().await?;

        replay(&mut tx).await?;

        tx.commit().await?;

        Ok(())

    }

    /// Whether the projections equal a rebuild from the ledger: rebuild in

    /// a transaction, compare, and roll back. Returns the tables that differ.

    pub async fn check(&self) -> Result<Vec<String>> {

        let _head = self.head.lock().await;

        let mut tx = self.pool.begin().await?;

        let mut before = Vec::new();

        for t in projections::TABLES {

            before.push(dump(&mut tx, t).await?);

        }

        replay(&mut tx).await?;

        let mut differ = Vec::new();

        for (t, b) in projections::TABLES.iter().zip(before) {

            if dump(&mut tx, t).await? != b {

                differ.push(t.to_string());

            }

        }

        tx.rollback().await?;

        Ok(differ)

    }

}

/// `"spectra1 checkpoint " || hash || seq (8 bytes, big-endian)`.

pub fn checkpoint_message(seq: u64, hash: &[u8; 32]) -> Vec<u8> {

    let mut m = CHECKPOINT_DOMAIN.to_vec();

    m.extend_from_slice(hash);

    m.extend_from_slice(&seq.to_be_bytes());

    m

}

fn check_checkpoint(e: &Event, prev: [u8; 32], instance: Option<&Identity>) -> std::result::Result<(), String> {

    let p = e.json();

    let seq = p.get("seq").and_then(|s| s.as_u64()).ok_or("no seq")?;

    let hash: [u8; 32] =

        p.get("hash").and_then(|h| h.as_str()).and_then(|h| hex::decode(h).ok()).and_then(|b| b.try_into().ok()).ok_or("no hash")?;

    if seq + 1 != e.seq || hash != prev {

        return Err("names a head other than the event before it".into());

    }

    let instance = instance.ok_or("no instance identity precedes it")?;

    if e.auth.key != Some(instance.hash().0) {

        return Err("signed by a key other than the instance's".into());

    }

    let proof = e.auth.proof.as_deref().ok_or("unsigned")?;

    let sig = oxy_core::cbor::decode(proof).ok().and_then(|v| Signature::from_value(&v).ok()).ok_or("malformed signature")?;

    oxy_core::identity::verify_challenge(instance, &sig, &checkpoint_message(seq, &hash))

        .map_err(|e| format!("signature does not verify: {e}"))

}

/// Checks the proof of a signed event: the identities and delegations that

/// say whose key it was (spec §8.5). Principals supply it (`principals`).

pub trait KeyResolver: Send + Sync {

    fn check(&self, e: &Event) -> std::result::Result<(), String>;

}

/// A resolver that knows no keys: every signed event is reported.

pub struct NoKeys;

impl KeyResolver for NoKeys {

    fn check(&self, _: &Event) -> std::result::Result<(), String> {

        Err("no identity is known for its key".into())

    }

}

/// What `verify` found.

#[derive(Clone, Debug, Default, serde::Serialize)]

pub struct Verification {

    pub events: u64,

    pub checkpoints: u64,

    pub last_checkpoint: Option<u64>,

    /// Events after the last good checkpoint: chained, not yet signed.

    pub uncovered: u64,

    #[serde(serialize_with = "ser_hex")]

    pub head: [u8; 32],

    pub findings: Vec<Finding>,

}

fn ser_hex<S: serde::Serializer>(h: &[u8; 32], s: S) -> std::result::Result<S::Ok, S::Error> {

    s.serialize_str(&hex::encode(h))

}

impl Verification {

    pub fn clean(&self) -> bool {

        self.findings.is_empty()

    }

    fn finding(&mut self, seq: u64, problem: Problem, detail: String) {

        self.findings.push(Finding { seq, problem, detail });

    }

}

/// One thing wrong with the ledger.

#[derive(Clone, Debug, serde::Serialize)]

pub struct Finding {

    pub seq: u64,

    pub problem: Problem,

    pub detail: String,

}

#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)]

#[serde(rename_all = "snake_case")]

pub enum Problem {

    Gap,

    ChainBreak,

    HashMismatch,

    BadCheckpoint,

    BadProof,

}

const SELECT_EVENT: &str = "SELECT seq, id, at_ms, hlc_wall, hlc_logical, prev, hash, auth_key, auth_method, auth_assurance, auth_proof, kind, actor, session, project, subject, payload, trace FROM events";

async fn insert(conn: &mut sqlx::SqliteConnection, e: &Event) -> sqlx::Result<()> {

    sqlx::query(

        "INSERT INTO events (seq, id, at_ms, hlc_wall, hlc_logical, prev, hash, auth_key, auth_method, auth_assurance, auth_proof, kind, actor, session, project, subject, payload, trace) \

         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",

    )

    .bind(e.seq as i64)

    .bind(&e.id)

    .bind(e.at_ms as i64)

    .bind(e.hlc.wall_ms as i64)

    .bind(e.hlc.logical as i64)

    .bind(e.prev.to_vec())

    .bind(e.hash.to_vec())

    .bind(e.auth.key.map(|k| k.to_vec()))

    .bind(e.auth.method as i64)

    .bind(e.auth.assurance as i64)

    .bind(e.auth.proof.clone())

    .bind(&e.kind)

    .bind(&e.actor)

    .bind(&e.session)

    .bind(&e.project)

    .bind(&e.subject)

    .bind(e.payload.clone())

    .bind(&e.trace)

    .execute(conn)

    .await?;

    Ok(())

}

fn row_to_event(r: &SqliteRow) -> Result<Event> {

    Ok(Event {

        seq: r.get::<i64, _>(0) as u64,

        id: r.get(1),

        at_ms: r.get::<i64, _>(2) as u64,

        hlc: Hlc { wall_ms: r.get::<i64, _>(3) as u64, logical: r.get::<i64, _>(4) as u32 },

        prev: to32(r.get(5))?,

        hash: to32(r.get(6))?,

        auth: Auth {

            key: r.get::<Option<Vec<u8>>, _>(7).map(to32).transpose()?,

            method: Method::from_u8(r.get::<i64, _>(8) as u8).context("unknown auth method")?,

            assurance: r.get::<i64, _>(9) as u8,

            proof: r.get(10),

        },

        kind: r.get(11),

        actor: r.get(12),

        session: r.get(13),

        project: r.get(14),

        subject: r.get(15),

        payload: r.get(16),

        trace: r.get(17),

    })

}

fn to32(v: Vec<u8>) -> Result<[u8; 32]> {

    v.try_into().map_err(|_| anyhow::anyhow!("a stored hash is not 32 bytes"))

}

async fn replay(conn: &mut sqlx::SqliteConnection) -> Result<()> {

    for t in projections::TABLES {

        sqlx::query(&format!("DELETE FROM {t}")).execute(&mut *conn).await?;

    }

    let mut from = 1i64;

    loop {

        let rows = sqlx::query(&format!("{SELECT_EVENT} WHERE seq >= ? ORDER BY seq LIMIT 2000")).bind(from).fetch_all(&mut *conn).await?;

        if rows.is_empty() {

            break;

        }

        for r in &rows {

            let e = row_to_event(r)?;

            projections::apply(conn, &e).await?;

            from = e.seq as i64 + 1;

        }

    }

    Ok(())

}

async fn dump(conn: &mut sqlx::SqliteConnection, table: &str) -> Result<Vec<String>> {

    // Every column as text, rows in a fixed order: enough to compare.

    let rows = sqlx::query(&format!("SELECT * FROM {table} ORDER BY 1, 2")).fetch_all(&mut *conn).await?;

    Ok(rows

        .iter()

        .map(|r| {

            (0..r.columns().len())

                .map(|i| match r.try_get::<Option<String>, _>(i) {

                    Ok(v) => format!("{v:?}"),

                    Err(_) => match r.try_get::<Option<i64>, _>(i) {

                        Ok(n) => format!("{n:?}"),

                        Err(_) => format!("{:?}", r.try_get::<Option<Vec<u8>>, _>(i).ok().flatten()),

                    },

                })

                .collect::<Vec<_>>()

                .join("|")

        })

        .collect())

}

/// The identity of `key` among `identities`, by KeyId.

pub fn find_identity<'a>(identities: &'a [Identity], key: &KeyId) -> Option<&'a Identity> {

    identities.iter().find(|i| &i.hash() == key)

}

//! Projections (spec §9.3): tables kept from the ledger in the same

//! transaction as each append, and rebuilt from it by `spectra admin

//! rebuild`. Every table here must be a function of the events alone; the

//! test harness and `spectra admin check` hold them to it.

use crate::event::Event;

use sqlx::SqliteConnection;

/// The projection tables, in the order they are cleared and compared.

pub const TABLES: &[&str] = &["event_counts"];

/// Apply one event to every projection.

pub async fn apply(conn: &mut SqliteConnection, e: &Event) -> sqlx::Result<()> {

    sqlx::query("INSERT INTO event_counts (kind, n) VALUES (?, 1) ON CONFLICT(kind) DO UPDATE SET n = n + 1")

        .bind(&e.kind)

        .execute(&mut *conn)

        .await?;

    Ok(())

}

//! The ledger's promises (spec §9.5, goal G6): an untouched ledger verifies,

//! with checkpoints signed by the instance key; an edited, removed or

//! re-signed event is caught; projections always equal a rebuild.

use rand::{Rng, SeedableRng};

use spectra_core::event::NewEvent;

use spectra_core::keys::KeyPair;

use spectra_core::ledger::{Ledger, NoKeys, Problem, CHECKPOINT_EVERY};

use std::path::Path;

async fn open(dir: &Path) -> Ledger {

    let instance = KeyPair::instance(dir, "test-instance").unwrap();

    let ledger = Ledger::open(&dir.join("spectra.db"), Some(instance.clone())).await.unwrap();

    if ledger.head().await.0 == 0 {

        ledger

            .append_one(NewEvent::system(

                "identity.instance",

                serde_json::json!({ "key": instance.key.to_hex(), "identity": hex::encode(instance.identity.encode()) }),

            ))

            .await

            .unwrap();

    }

    ledger

}

fn msg(i: usize) -> NewEvent {

    NewEvent::system("test.message", serde_json::json!({ "n": i, "body": format!("message {i}") })).subject(format!("m{i}"))

}

async fn raw(dir: &Path, sql: &str) {

    let pool = sqlx::SqlitePool::connect(&format!("sqlite://{}", dir.join("spectra.db").display())).await.unwrap();

    sqlx::query(sql).execute(&pool).await.unwrap();

    pool.close().await;

}

#[tokio::test]

async fn an_untouched_ledger_verifies_with_signed_checkpoints() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    for i in 0..(CHECKPOINT_EVERY as usize + 10) {

        ledger.append_one(msg(i)).await.unwrap();

    }

    let cp = ledger.checkpoint().await.unwrap().expect("events since the last checkpoint");

    assert_eq!(cp.kind, "ledger.checkpoint");

    let v = ledger.verify(&NoKeys).await.unwrap();

    assert!(v.clean(), "{:?}", v.findings);

    assert_eq!(v.checkpoints, 2, "one at 256 events, one on demand");

    assert_eq!(v.uncovered, 0);

    assert_eq!(v.head, ledger.head().await.1);

    assert!(ledger.checkpoint().await.unwrap().is_none(), "nothing new to cover");

}

#[tokio::test]

async fn an_edited_event_is_caught() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    for i in 0..20 {

        ledger.append_one(msg(i)).await.unwrap();

    }

    ledger.checkpoint().await.unwrap();

    drop(ledger);

    raw(dir.path(), r#"UPDATE events SET payload = CAST('{"body":"message 5, edited","n":5}' AS BLOB) WHERE subject = 'm5'"#).await;

    let v = open(dir.path()).await.verify(&NoKeys).await.unwrap();

    assert_eq!(v.findings.len(), 1, "{:?}", v.findings);

    assert_eq!(v.findings[0].problem, Problem::HashMismatch);

}

#[tokio::test]

async fn an_edit_with_its_hash_recomputed_breaks_the_chain_or_the_checkpoint() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    for i in 0..20 {

        ledger.append_one(msg(i)).await.unwrap();

    }

    ledger.checkpoint().await.unwrap();

    // A careful forger recomputes the edited event's hash: the next event's

    // prev no longer matches, and the checkpoint names the old head.

    let mut e = ledger.events(1, 100).await.unwrap().into_iter().find(|e| e.subject.as_deref() == Some("m19")).unwrap();

    e.payload = Some(br#"{"body":"forged","n":19}"#.to_vec());

    let forged = hex::encode(e.computed_hash());

    drop(ledger);

    raw(

        dir.path(),

        &format!(r#"UPDATE events SET payload = CAST('{{"body":"forged","n":19}}' AS BLOB), hash = X'{forged}' WHERE seq = {}"#, e.seq),

    )

    .await;

    let v = open(dir.path()).await.verify(&NoKeys).await.unwrap();

    assert!(v.findings.iter().any(|f| f.problem == Problem::ChainBreak || f.problem == Problem::BadCheckpoint), "{:?}", v.findings);

}

#[tokio::test]

async fn a_removed_event_is_caught() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    for i in 0..10 {

        ledger.append_one(msg(i)).await.unwrap();

    }

    drop(ledger);

    raw(dir.path(), "DELETE FROM events WHERE subject = 'm4'").await;

    let v = open(dir.path()).await.verify(&NoKeys).await.unwrap();

    assert!(v.findings.iter().any(|f| f.problem == Problem::Gap), "{:?}", v.findings);

    assert!(v.findings.iter().any(|f| f.problem == Problem::ChainBreak), "{:?}", v.findings);

}

#[tokio::test]

async fn a_checkpoint_signed_by_another_key_is_caught() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    ledger.append_one(msg(1)).await.unwrap();

    let cp = ledger.checkpoint().await.unwrap().unwrap();

    drop(ledger);

    let mallory = KeyPair::generate("mallory", "");

    let sig = mallory.sign(&spectra_core::ledger::checkpoint_message(cp.seq - 1, &cp.prev));

    let proof = hex::encode(oxy_core::cbor::encode(&sig.to_value()));

    raw(dir.path(), &format!("UPDATE events SET auth_proof = X'{proof}' WHERE seq = {}", cp.seq)).await;

    let v = open(dir.path()).await.verify(&NoKeys).await.unwrap();

    assert_eq!(v.findings.len(), 1, "{:?}", v.findings);

    assert_eq!(v.findings[0].problem, Problem::BadCheckpoint);

}

#[tokio::test]

async fn projections_equal_a_rebuild_over_ten_thousand_random_events() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    let mut rng = rand::rngs::StdRng::seed_from_u64(7);

    let kinds = ["message.posted", "agent.presence", "task.created", "claim.taken", "tool.called"];

    let started = std::time::Instant::now();

    let mut appended = 0;

    while appended < 10_000 {

        let n = rng.gen_range(1..50);

        let batch = (0..n)

            .map(|i| {

                NewEvent::system(kinds[rng.gen_range(0..kinds.len())], serde_json::json!({ "i": appended + i, "r": rng.gen::<u32>() }))

            })

            .collect();

        ledger.append(batch).await.unwrap();

        appended += n;

    }

    let per_batch_ms = started.elapsed().as_millis() as f64 / (10_000.0 / 25.0);

    assert!(ledger.check().await.unwrap().is_empty(), "projections drifted from the ledger");

    ledger.rebuild().await.unwrap();

    assert!(ledger.check().await.unwrap().is_empty());

    let v = ledger.verify(&NoKeys).await.unwrap();

    assert!(v.clean(), "{:?}", &v.findings[..v.findings.len().min(3)]);

    assert!(v.checkpoints >= 10_000 / CHECKPOINT_EVERY);

    // Generous in a debug build; the release number is in docs/plan.md.

    assert!(per_batch_ms < 200.0, "{per_batch_ms} ms a batch");

}

#[tokio::test]

async fn a_ledger_from_a_newer_spectra_is_refused() {

    let dir = tempfile::tempdir().unwrap();

    drop(open(dir.path()).await);

    raw(dir.path(), "UPDATE meta SET value = '99' WHERE key = 'ledger_format'").await;

    let err = Ledger::open(&dir.path().join("spectra.db"), None).await.err().expect("refused");

    assert!(err.to_string().contains("newer Spectra"), "{err}");

}

#[tokio::test]

async fn the_bus_delivers_what_was_appended_in_order() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    let mut rx = ledger.subscribe();

    ledger.append(vec![msg(1), msg(2)]).await.unwrap();

    assert_eq!(rx.recv().await.unwrap().subject.as_deref(), Some("m1"));

    assert_eq!(rx.recv().await.unwrap().subject.as_deref(), Some("m2"));

}

#[tokio::test]

async fn reopening_continues_the_chain() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    ledger.append_one(msg(1)).await.unwrap();

    let (seq, hash) = ledger.head().await;

    drop(ledger);

    let ledger = open(dir.path()).await;

    let e = ledger.append_one(msg(2)).await.unwrap();

    assert_eq!((e.seq, e.prev), (seq + 1, hash));

    assert!(ledger.verify(&NoKeys).await.unwrap().clean());

}

#[tokio::test]

async fn a_single_append_is_fast() {

    let dir = tempfile::tempdir().unwrap();

    let ledger = open(dir.path()).await;

    let started = std::time::Instant::now();

    for i in 0..1_000 {

        ledger.append_one(msg(i)).await.unwrap();

    }

    let each_us = started.elapsed().as_micros() / 1_000;

    eprintln!("append: {each_us} µs an event over 1,000 single appends");

    // The target is 10 ms with fsync (spec §24); debug builds get headroom.

    assert!(each_us < 50_000, "{each_us} µs");