//! 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::identity::KeyId;

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

use crate::keys::KeyPair;

use crate::{ids, projections};

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

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,

    /// Held for as long as this ledger may append: one writer per data

    /// directory, so the chain never forks (`None` when opened read-only).

    _lock: Option<std::fs::File>,

    w: Arc<Writer>,

}

/// What writing needs, shared with the task that does it. Every write runs

/// in a task of its own, so a caller that goes away (a client that hangs up

/// mid-request) can never leave a transaction half done: a cancelled

/// transaction on a pooled connection would swallow every later commit

/// made on that connection.

struct Writer {

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

            Err(std::fs::TryLockError::WouldBlock) => {

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

            }

            Err(std::fs::TryLockError::Error(e)) => return Err(e).with_context(|| format!("locking {}", lock_path.display())),

        }

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

        l._lock = Some(lock);

        Ok(l)

    }

    /// Open for reading and verifying only, while another process may be

    /// writing: no lock, and `append` must not be called.

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

            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: pool.clone(),

            _lock: None,

            w: Arc::new(Writer {

                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.w.bus.subscribe()

    }

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

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

        self.w.instance.as_ref()

    }

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

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

        let h = self.w.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. The work runs to

    /// completion even if the caller is dropped.

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

        let w = self.w.clone();

        tokio::spawn(async move { w.append(events).await }).await.context("the ledger's writer stopped")?

    }

    /// 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 w = self.w.clone();

        tokio::spawn(async move { w.checkpoint().await }).await.context("the ledger's writer stopped")?

    }

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

    }

    /// Events of the given kinds from `from` (inclusive), oldest first.

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

    }

    /// 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: &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;

        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 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 w = self.w.clone();

        tokio::spawn(async move { w.rebuild().await }).await.context("the ledger's writer stopped")?

    }

    /// 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 w = self.w.clone();

        tokio::spawn(async move { w.check().await }).await.context("the ledger's writer stopped")?

    }

}

impl Writer {

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

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

    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)

    }

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

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

    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(crate::cbor::encode(&sig.to_value())) },

        );

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

        h.since_checkpoint = 0;

        h.oldest_unchecked_ms = None;

        e

    }

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

    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.

    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 = crate::cbor::decode(proof).ok().and_then(|v| Signature::from_value(&v).ok()).ok_or("malformed signature")?;

    crate::identity::verify(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`).

/// It is given every event but checkpoints, in order, so it can follow

/// identities as they are recorded.

pub trait KeyResolver: Send {

    fn check(&mut 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(&mut self, e: &Event) -> std::result::Result<(), String> {

        match e.auth.method {

            Method::Signature | Method::Passkey => Err("no identity is known for its key".into()),

            _ => Ok(()),

        }

    }

}

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