Coverage / crates/spectra-core/src/ledger.rs 95.0%

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

        let lock_path = path.with_extension("lock");

        let lock = std::fs::OpenOptions::new()

            .create(true)

            .truncate(false)

            .write(true)

            .open(&lock_path)

            .with_context(|| format!("opening {}", lock_path.display()))?;

        match lock.try_lock() {

            Ok(()) => {}

                bail!("another process is writing to {} (is `spectra serve` running? stop it, or do this through it)", path.display())

        let mut l = Ledger::open_unlocked(path, instance).await?;

        l._lock = Some(lock);

        Ok(l)

    }

    pub async fn open_read_only(path: &Path) -> Result<Ledger> {

        Ledger::open_unlocked(path, None).await

    }

    async fn open_unlocked(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>) {

                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 => {}

                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,

            _lock: None,

            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,

        })

    }

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

        &self.pool

    }

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

        self.bus.subscribe()

    }

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

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

        (h.seq, h.hash)

    }

    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);

            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() {

        }

        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)

    }

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

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

    }

    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

    }

    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(

            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

    }

    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()

    }

    pub async fn events_of_kinds(&self, from: u64, kinds: &[&str], limit: u32) -> Result<Vec<Event>> {

        let marks = vec!["?"; kinds.len()].join(",");

        let sql = format!("{SELECT_EVENT} WHERE seq >= ? AND kind IN ({marks}) ORDER BY seq LIMIT ?");

        let mut q = sqlx::query(&sql).bind(from as i64);

        for k in kinds {

            q = q.bind(*k);

        }

        let rows = q.bind(limit as i64).fetch_all(&self.pool).await?;

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

    }

    pub async fn verify(&self, keys: &mut 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;

            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 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)

    }

    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(())

    }

    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 {

            }

        tx.rollback().await?;

        Ok(differ)

    }

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) {

    }

    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}"))

}

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

        match e.auth.method {

            _ => Ok(()),

    }

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

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

}

    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 });

    }