//! Hybrid logical clock (HLC) state for sync conflict resolution.
//!
//! AF stamps each local pending change with a minted HLC and records the
//! committed HLC per `(table, row_id)` in `hlc_ledger`, so synckit's conflict
//! resolution can drop a stale remote change instead of clobbering a newer local
//! value (replacing the prior blind last-writer-wins in `apply_remote_changes`).
//!
//! All HLC logic lives here in the sync layer; the content store and DB schema
//! stay sync-agnostic. The device's `node` is its registered `device_id`; the
//! running clock is persisted as `sync_state.last_hlc`.
//!
//! Convergence rests on `device_id` (the HLC `node`) being unique per install: it
//! is the final, deterministic tiebreak when two HLCs share `wall_ms`+`counter`.
//! A restored backup that copies the `device_id` violates this; synckit mitigates
//! the exact-HLC collision with a canonical-payload tiebreak, but unique
//! `device_id`s are the design assumption.

use rusqlite::{Connection, OptionalExtension};
use synckit_client::{DeviceId, Hlc};

use super::{get_sync_state, set_sync_state};
use crate::error::Result;

/// Wall-clock milliseconds since the Unix epoch (the HLC physical component).
fn now_ms() -> i64 {
    chrono::Utc::now().timestamp_millis()
}

/// Load the persisted device clock. Returns the zero clock on first sync (no
/// persisted value); on a *corrupt* persisted value it seeds from physical time
/// rather than zero, see below.
pub(crate) fn load_clock(conn: &Connection, node: DeviceId) -> Hlc {
    match get_sync_state(conn, "last_hlc") {
        Ok(s) if !s.is_empty() => serde_json::from_str(&s).unwrap_or_else(|e| {
            // Corrupt persisted clock: do NOT reset to zero. Zero sends the clock
            // backward, so a subsequent local edit could mint an HLC that loses to
            // an already-committed row, the one place HLC monotonicity isn't
            // structurally guaranteed. Seed from physical time instead (forward of
            // any sanely-timestamped past HLC); `observe()` advances it past
            // anything seen on the next pull.
            tracing::warn!("corrupt last_hlc ({e}); seeding clock from physical time");
            Hlc::tick(Hlc::zero(node), now_ms(), node)
        }),
        _ => Hlc::zero(node), // first sync: genuinely no prior clock
    }
}

fn store_clock(conn: &Connection, clock: &Hlc) -> Result<()> {
    // Hlc is a plain derive-Serialize struct, so serialization cannot fail; make
    // that an invariant (panic on the impossible) rather than silently persisting
    // an empty string that load_clock would then treat as a corrupt clock.
    let json = serde_json::to_string(clock).expect("Hlc always serializes");
    set_sync_state(conn, "last_hlc", &json)
}

/// The committed HLC for a row, if any, the oracle `CleanChanges::gated` uses to
/// drop remote changes that are not newer than what this device already holds.
///
/// A genuine query error propagates. It was previously swallowed as "no committed
/// HLC", which is the one answer that gates a remote change *in* (`is_none_or`), so
/// a failed ledger read could let a stale remote value clobber a newer local one.
/// A *corrupt* stored HLC still degrades to `None` with a warning, matching how
/// `load_clock` treats a corrupt `last_hlc`, one bad ledger row must not wedge the
/// entire sync.
pub(crate) fn committed_hlc(conn: &Connection, table: &str, row_id: &str) -> Result<Option<Hlc>> {
    let stored: Option<String> = conn
        .query_row(
            "SELECT hlc FROM hlc_ledger WHERE table_name = ?1 AND row_id = ?2",
            rusqlite::params![table, row_id],
            |r| r.get::<_, String>(0),
        )
        .optional()?;
    Ok(stored.and_then(|s| match serde_json::from_str(&s) {
        Ok(hlc) => Some(hlc),
        Err(e) => {
            tracing::warn!("corrupt committed HLC for {table}/{row_id} ({e}); treating as absent");
            None
        }
    }))
}

/// Record the committed HLC for a row (one of our pushes, or an applied remote).
pub(crate) fn set_committed(conn: &Connection, table: &str, row_id: &str, hlc: &Hlc) -> Result<()> {
    conn.execute(
        "INSERT INTO hlc_ledger (table_name, row_id, hlc) VALUES (?1, ?2, ?3)
         ON CONFLICT(table_name, row_id) DO UPDATE SET hlc = excluded.hlc",
        rusqlite::params![
            table,
            row_id,
            serde_json::to_string(hlc).expect("Hlc always serializes")
        ],
    )?;
    Ok(())
}

/// Assign an HLC to every un-stamped pending changelog row (`pushed = 0` and
/// `hlc IS NULL`), in `id` order, advancing and persisting the device clock.
///
/// Idempotent: a row that already carries an HLC keeps it, so the value is stable
/// whether it is read first by conflict detection (pull) or by push.
pub(crate) fn stamp_pending(conn: &Connection, node: DeviceId) -> Result<()> {
    let ids: Vec<i64> = {
        let mut stmt = conn.prepare(
            "SELECT id FROM sync_changelog WHERE pushed = 0 AND hlc IS NULL ORDER BY id",
        )?;
        let rows = stmt.query_map([], |r| r.get(0))?;
        rows.collect::<std::result::Result<_, _>>()?
    };
    if ids.is_empty() {
        return Ok(());
    }
    let mut clock = load_clock(conn, node);
    for id in ids {
        clock = Hlc::tick(clock, now_ms(), node);
        conn.execute(
            "UPDATE sync_changelog SET hlc = ?1 WHERE id = ?2",
            rusqlite::params![
                serde_json::to_string(&clock).expect("Hlc always serializes"),
                id
            ],
        )?;
    }
    store_clock(conn, &clock)
}

/// Advance the device clock by observing a remote HLC, so our next local edit is
/// ordered after every change we've seen (standard HLC receive rule).
pub(crate) fn observe(conn: &Connection, node: DeviceId, remote: Hlc) -> Result<()> {
    let next = Hlc::observe(load_clock(conn, node), remote, now_ms(), node);
    store_clock(conn, &next)
}

#[cfg(test)]
mod tests {
    use super::*;
    use uuid::Uuid;

    fn mem() -> Connection {
        // A standalone in-memory DB with just the tables the HLC layer touches.
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(
            "CREATE TABLE sync_state (key TEXT PRIMARY KEY, value TEXT NOT NULL);
             CREATE TABLE sync_changelog (id INTEGER PRIMARY KEY AUTOINCREMENT,
                table_name TEXT, op TEXT, row_id TEXT, timestamp TEXT, data TEXT,
                pushed INTEGER NOT NULL DEFAULT 0, hlc TEXT);
             CREATE TABLE hlc_ledger (table_name TEXT, row_id TEXT, hlc TEXT,
                PRIMARY KEY (table_name, row_id));
             INSERT INTO sync_state (key, value) VALUES ('last_hlc', '');",
        )
        .unwrap();
        conn
    }

    #[test]
    fn stamp_pending_assigns_monotonic_hlcs() {
        let conn = mem();
        let node = DeviceId::new(Uuid::from_u128(1));
        for i in 0..3 {
            conn.execute(
                "INSERT INTO sync_changelog (table_name, op, row_id) VALUES ('tags', 'INSERT', ?1)",
                rusqlite::params![format!("r{i}")],
            )
            .unwrap();
        }
        stamp_pending(&conn, node).unwrap();
        let hlcs: Vec<Hlc> = {
            let mut stmt = conn
                .prepare("SELECT hlc FROM sync_changelog ORDER BY id")
                .unwrap();
            stmt.query_map([], |r| r.get::<_, String>(0))
                .unwrap()
                .map(|s| serde_json::from_str(&s.unwrap()).unwrap())
                .collect()
        };
        assert_eq!(hlcs.len(), 3);
        assert!(
            hlcs[0] < hlcs[1] && hlcs[1] < hlcs[2],
            "HLCs must be strictly increasing"
        );
        // Re-stamping is a no-op: existing HLCs are stable.
        stamp_pending(&conn, node).unwrap();
        let again: Hlc = serde_json::from_str(
            &conn
                .query_row("SELECT hlc FROM sync_changelog WHERE id = 1", [], |r| {
                    r.get::<_, String>(0)
                })
                .unwrap(),
        )
        .unwrap();
        assert_eq!(again, hlcs[0]);
    }

    #[test]
    fn corrupt_clock_seeds_forward_not_zero() {
        let conn = mem();
        let node = DeviceId::new(Uuid::from_u128(1));
        // First sync (empty) is genuinely the zero clock.
        assert_eq!(load_clock(&conn, node), Hlc::zero(node));
        // A corrupt persisted clock must NOT reset to zero (that moves the clock
        // backward); it seeds from physical time, so it's well ahead of zero.
        set_sync_state(&conn, "last_hlc", "{not valid json").unwrap();
        let recovered = load_clock(&conn, node);
        assert!(
            recovered.wall_ms > 0 && recovered > Hlc::zero(node),
            "corrupt clock seeded backward: {recovered:?}"
        );
    }

    #[test]
    fn ledger_roundtrip_and_gate() {
        let conn = mem();
        let node = DeviceId::new(Uuid::from_u128(7));
        assert!(committed_hlc(&conn, "tags", "r1").unwrap().is_none());
        let older = Hlc {
            wall_ms: 100,
            counter: 0,
            node,
        };
        let newer = Hlc {
            wall_ms: 200,
            counter: 0,
            node,
        };
        set_committed(&conn, "tags", "r1", &newer).unwrap();
        assert_eq!(committed_hlc(&conn, "tags", "r1").unwrap(), Some(newer));
        // A stale remote (older HLC) must be recognisable as not-newer.
        assert!(older < committed_hlc(&conn, "tags", "r1").unwrap().unwrap());
        // Upsert keeps the latest.
        set_committed(&conn, "tags", "r1", &older).unwrap();
        assert_eq!(committed_hlc(&conn, "tags", "r1").unwrap(), Some(older));
    }

    #[test]
    fn committed_hlc_propagates_query_error() {
        // A genuine read failure must surface as an error, not collapse to "no
        // committed HLC" (which gates a stale remote change *in*).
        let conn = mem();
        conn.execute_batch("DROP TABLE hlc_ledger").unwrap();
        assert!(committed_hlc(&conn, "tags", "r1").is_err());
    }

    #[test]
    fn committed_hlc_degrades_corrupt_row_to_none() {
        // A corrupt stored HLC is not a query error: it degrades to None (with a
        // warning) so one bad ledger row can't wedge the whole sync.
        let conn = mem();
        conn.execute(
            "INSERT INTO hlc_ledger (table_name, row_id, hlc) VALUES ('tags', 'r1', ?1)",
            rusqlite::params!["{not valid json"],
        )
        .unwrap();
        assert_eq!(committed_hlc(&conn, "tags", "r1").unwrap(), None);
    }

    #[test]
    fn observe_keeps_clock_ahead() {
        let conn = mem();
        let node = DeviceId::new(Uuid::from_u128(2));
        let remote = Hlc {
            wall_ms: i64::MAX / 2,
            counter: 5,
            node: DeviceId::new(Uuid::from_u128(99)),
        };
        observe(&conn, node, remote).unwrap();
        let clock = load_clock(&conn, node);
        assert!(
            clock >= remote,
            "local clock must dominate the observed remote"
        );
        assert_eq!(clock.node, node, "our clock carries our node");
    }
}