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