//! Hybrid logical clock ledger and conflict-strategy orchestration.
//!
//! This is the wiring between the changelog and the crate's audited conflict
//! primitives ([`detect_conflicts`], [`CleanChanges::gated`](crate::conflict),
//! [`resolve_lww_at`]). It owns three pieces of persistent HLC state:
//!
//! - the **local clock** (this device's monotonic HLC), kept in `sync_state`;
//! - the **per-row stamp** on each `sync_changelog` row (`hlc_wall`/`hlc_counter`);
//! - the **committed ledger** (`sync_committed_hlc`), the HLC last applied for a
//!   row, which the gate uses to drop a re-pulled older change.
//!
//! [`resolve_pull`] is the entry point: it turns a pulled batch into the resolved
//! `Vec<ChangeEntry>` the apply engine writes, dispatching on the schema's
//! [`ConflictStrategy`].

use std::collections::HashMap;

use chrono::{DateTime, Utc};
use rusqlite::{Connection, OptionalExtension};
use uuid::Uuid;

use super::db::{get_sync_state, set_sync_state};
use super::schema::{ConflictStrategy, SyncSchema};
use crate::conflict::{Resolution, detect_conflicts, resolve_lww_at};
use crate::error::Result;
use crate::ids::DeviceId;
use crate::types::{ChangeEntry, ChangeOp, Hlc, PulledChange};

// local clock (persisted in sync_state)

/// Load this device's persisted clock, or `Hlc::zero(node)` if never set.
pub fn load_clock(conn: &Connection, node: DeviceId) -> Result<Hlc> {
    let wall_ms = get_sync_state(conn, "hlc_wall")?
        .and_then(|s| s.parse::<i64>().ok())
        .unwrap_or(0);
    let counter = get_sync_state(conn, "hlc_counter")?
        .and_then(|s| s.parse::<u32>().ok())
        .unwrap_or(0);
    Ok(Hlc {
        wall_ms,
        counter,
        node,
    })
}

fn save_clock(conn: &Connection, clock: &Hlc) -> Result<()> {
    set_sync_state(conn, "hlc_wall", &clock.wall_ms.to_string())?;
    set_sync_state(conn, "hlc_counter", &clock.counter.to_string())?;
    Ok(())
}

/// Mint an HLC for every unpushed changelog row that lacks one, advancing the
/// local clock once per row. Returns how many rows were stamped.
pub fn stamp_pending(conn: &Connection, node: DeviceId, now_ms: i64) -> Result<usize> {
    let ids: Vec<i64> = {
        let mut stmt = conn.prepare(
            "SELECT id FROM sync_changelog WHERE pushed = 0 AND hlc_wall IS NULL ORDER BY id",
        )?;
        stmt.query_map([], |r| r.get(0))?
            .collect::<rusqlite::Result<_>>()?
    };
    if ids.is_empty() {
        return Ok(0);
    }
    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_wall = ?1, hlc_counter = ?2 WHERE id = ?3",
            (clock.wall_ms, clock.counter, id),
        )?;
    }
    save_clock(conn, &clock)?;
    Ok(ids.len())
}

/// Advance the local clock past every observed remote HLC (monotonic merge).
pub fn observe(
    conn: &Connection,
    node: DeviceId,
    remotes: impl IntoIterator<Item = Hlc>,
    now_ms: i64,
) -> Result<()> {
    let mut clock = load_clock(conn, node)?;
    let mut any = false;
    for remote in remotes {
        clock = Hlc::observe(clock, remote, now_ms, node);
        any = true;
    }
    if any {
        save_clock(conn, &clock)?;
    }
    Ok(())
}

// committed ledger

/// The HLC currently committed for a row, or `None` if never applied locally.
pub fn committed_hlc(conn: &Connection, table: &str, row_id: &str) -> Result<Option<Hlc>> {
    conn.query_row(
        "SELECT hlc_wall, hlc_counter, hlc_node FROM sync_committed_hlc \
         WHERE table_name = ?1 AND row_id = ?2",
        (table, row_id),
        |r| {
            let wall_ms: i64 = r.get(0)?;
            let counter: i64 = r.get(1)?;
            let node: String = r.get(2)?;
            Ok((wall_ms, counter as u32, node))
        },
    )
    .optional()?
    .map(|(wall_ms, counter, node)| {
        let node = Uuid::parse_str(&node)
            .map(DeviceId::new)
            .map_err(|e| crate::error::SyncKitError::Database(format!("bad hlc_node: {e}")))?;
        Ok(Hlc {
            wall_ms,
            counter,
            node,
        })
    })
    .transpose()
}

/// Record a row's committed HLC, advancing only (never regress the ledger).
pub fn set_committed(conn: &Connection, table: &str, row_id: &str, hlc: &Hlc) -> Result<()> {
    if let Some(existing) = committed_hlc(conn, table, row_id)?
        && cmp_hlc(&existing, hlc).is_ge()
    {
        return Ok(());
    }
    conn.execute(
        "INSERT INTO sync_committed_hlc (table_name, row_id, hlc_wall, hlc_counter, hlc_node) \
         VALUES (?1, ?2, ?3, ?4, ?5) \
         ON CONFLICT(table_name, row_id) DO UPDATE SET \
         hlc_wall = excluded.hlc_wall, hlc_counter = excluded.hlc_counter, hlc_node = excluded.hlc_node",
        (
            table,
            row_id,
            hlc.wall_ms,
            hlc.counter,
            hlc.node.as_uuid().to_string(),
        ),
    )?;
    Ok(())
}

/// Advance the committed ledger for a batch of applied entries.
pub fn record_committed(conn: &Connection, entries: &[ChangeEntry]) -> Result<()> {
    for e in entries {
        set_committed(conn, &e.table, &e.row_id, &e.hlc)?;
    }
    Ok(())
}

// pull resolution

/// Turn a pulled batch into the resolved changes to apply, per the schema's
/// [`ConflictStrategy`].
///
/// `ServerOrder` trusts the server's sequence order and returns the entries
/// verbatim (last delivered wins). `HybridLogicalClock` runs the full pipeline:
/// advance the clock past the remote HLCs, stamp local pending edits, split into
/// clean vs conflicting against local pending, HLC-gate the clean set against the
/// committed ledger, resolve conflicts by `resolve_lww`, and collapse to one
/// entry per row (highest HLC wins, operation-agnostic).
pub fn resolve_pull(
    conn: &Connection,
    schema: &SyncSchema,
    node: DeviceId,
    pulled: Vec<PulledChange>,
    now: DateTime<Utc>,
) -> Result<Vec<ChangeEntry>> {
    match schema.conflict {
        ConflictStrategy::ServerOrder => Ok(pulled.into_iter().map(|p| p.entry).collect()),
        ConflictStrategy::HybridLogicalClock => {
            let now_ms = now.timestamp_millis();
            observe(conn, node, pulled.iter().map(|p| p.entry.hlc), now_ms)?;
            stamp_pending(conn, node, now_ms)?;

            let local_pending = load_local_pending(conn, node)?;
            let (clean, conflicts) = detect_conflicts(pulled, &local_pending, node);

            let mut resolved: Vec<ChangeEntry> =
                clean.gated_at(now, |t, r| lookup_committed(conn, t, r));

            for pair in conflicts {
                match resolve_lww_at(&pair.local, &pair.remote, now) {
                    Resolution::KeepRemote => resolved.push(pair.remote.entry),
                    Resolution::KeepLocal | Resolution::Skip => {}
                    Resolution::Merged(data) => {
                        let hlc = max_hlc(pair.local.hlc, pair.remote.entry.hlc);
                        resolved.push(ChangeEntry {
                            table: pair.remote.entry.table,
                            op: ChangeOp::Update,
                            row_id: pair.remote.entry.row_id,
                            timestamp: pair.remote.entry.timestamp,
                            hlc,
                            data: Some(data),
                            extra: serde_json::Map::default(),
                        });
                    }
                }
            }

            Ok(collapse_max_hlc(resolved))
        }
    }
}

/// Committed-HLC lookup for the gate. A read error is logged and treated as
/// "never applied", the gate then keeps the change, which the idempotent apply
/// path can safely re-run.
fn lookup_committed(conn: &Connection, table: &str, row_id: &str) -> Option<Hlc> {
    match committed_hlc(conn, table, row_id) {
        Ok(v) => v,
        Err(e) => {
            tracing::warn!(
                table,
                row_id,
                "committed_hlc lookup failed, treating as unset: {e}"
            );
            None
        }
    }
}

/// Load the unpushed, HLC-stamped local changelog rows as `ChangeEntry` values
/// for conflict detection. `node` is this device's id, the node of every local
/// stamp.
fn load_local_pending(conn: &Connection, node: DeviceId) -> Result<Vec<ChangeEntry>> {
    let mut stmt = conn.prepare(
        "SELECT table_name, op, row_id, data, hlc_wall, hlc_counter \
         FROM sync_changelog WHERE pushed = 0 AND hlc_wall IS NOT NULL ORDER BY id",
    )?;
    let rows = stmt.query_map([], |r| {
        let table: String = r.get(0)?;
        let op: String = r.get(1)?;
        let row_id: String = r.get(2)?;
        let data: Option<String> = r.get(3)?;
        let wall_ms: i64 = r.get(4)?;
        let counter: i64 = r.get(5)?;
        Ok((table, op, row_id, data, wall_ms, counter as u32))
    })?;

    let mut out = Vec::new();
    for row in rows {
        let (table, op, row_id, data, wall_ms, counter) = row?;
        let Some(op) = ChangeOp::from_str_opt(&op) else {
            continue; // unknown op, not a resolvable local edit
        };
        let data = match data {
            Some(s) => Some(serde_json::from_str(&s)?),
            None => None,
        };
        out.push(ChangeEntry {
            table,
            op,
            row_id,
            timestamp: Utc::now(),
            hlc: Hlc {
                wall_ms,
                counter,
                node,
            },
            data,
            extra: serde_json::Map::default(),
        });
    }
    Ok(out)
}

/// Collapse to one entry per `(table, row_id)`, keeping the highest HLC,
/// operation-agnostic, so a newer delete beats an older edit and vice versa, and
/// every device converges on the same winner. First-seen order is preserved
/// (apply re-orders by schema anyway).
fn collapse_max_hlc(entries: Vec<ChangeEntry>) -> Vec<ChangeEntry> {
    let mut best: HashMap<(String, String), usize> = HashMap::new();
    let mut kept: Vec<ChangeEntry> = Vec::new();
    for e in entries {
        let key = (e.table.clone(), e.row_id.clone());
        match best.get(&key) {
            Some(&i) if cmp_hlc(&kept[i].hlc, &e.hlc).is_ge() => {}
            Some(&i) => kept[i] = e,
            None => {
                best.insert(key, kept.len());
                kept.push(e);
            }
        }
    }
    kept
}

fn cmp_hlc(a: &Hlc, b: &Hlc) -> std::cmp::Ordering {
    (a.wall_ms, a.counter, a.node.as_uuid()).cmp(&(b.wall_ms, b.counter, b.node.as_uuid()))
}

fn max_hlc(a: Hlc, b: Hlc) -> Hlc {
    if cmp_hlc(&a, &b).is_ge() { a } else { b }
}

#[cfg(test)]
mod tests {
    use super::super::apply::apply_remote_changes;
    use super::super::db::configure_connection;
    use super::super::schema::{SyncSchema, SyncTable};
    use super::*;
    use rusqlite::Connection;

    fn node(n: u128) -> DeviceId {
        DeviceId::new(Uuid::from_u128(n))
    }

    fn schema() -> SyncSchema {
        SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])])
    }

    fn server_order_schema() -> SyncSchema {
        schema().conflict_strategy(ConflictStrategy::ServerOrder)
    }

    /// A device: in-memory DB with the note table + migration, and a node id.
    fn device(n: u128) -> (Connection, DeviceId) {
        let conn = Connection::open_in_memory().unwrap();
        configure_connection(&conn).unwrap();
        conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
            .unwrap();
        conn.execute_batch(&schema().migration_sql()).unwrap();
        (conn, node(n))
    }

    /// Make a local edit (domain write → trigger captures it), stamp it at
    /// `now_ms`, and return it as a pulled change from `node` (as a peer would
    /// receive it via the server).
    fn local_edit_as_pulled(
        conn: &Connection,
        node: DeviceId,
        id: &str,
        name: &str,
        now_ms: i64,
        seq: i64,
    ) -> PulledChange {
        conn.execute(
            "INSERT INTO note (id, name) VALUES (?1, ?2) \
             ON CONFLICT(id) DO UPDATE SET name = excluded.name",
            (id, name),
        )
        .unwrap();
        stamp_pending(conn, node, now_ms).unwrap();
        let entry = load_local_pending(conn, node)
            .unwrap()
            .into_iter()
            .find(|e| e.row_id == id)
            .unwrap();
        PulledChange {
            entry,
            device_id: node,
            seq,
        }
    }

    fn note_name(conn: &Connection, id: &str) -> Option<String> {
        conn.query_row("SELECT name FROM note WHERE id = ?1", [id], |r| r.get(0))
            .optional()
            .unwrap()
    }

    fn pull_apply(
        conn: &mut Connection,
        s: &SyncSchema,
        node: DeviceId,
        pulled: Vec<PulledChange>,
    ) {
        let now = Utc::now();
        let resolved = resolve_pull(conn, s, node, pulled, now).unwrap();
        apply_remote_changes(conn, s, &resolved, "").unwrap();
        record_committed(conn, &resolved).unwrap();
    }

    #[test]
    fn stamp_pending_assigns_monotonic_hlcs() {
        let (conn, n) = device(1);
        conn.execute("INSERT INTO note (id, name) VALUES ('a', '1')", [])
            .unwrap();
        conn.execute("INSERT INTO note (id, name) VALUES ('b', '2')", [])
            .unwrap();
        assert_eq!(stamp_pending(&conn, n, 1000).unwrap(), 2);
        // Re-stamping is a no-op (both already stamped).
        assert_eq!(stamp_pending(&conn, n, 2000).unwrap(), 0);
        let stamps: Vec<(i64, i64)> = {
            let mut s = conn
                .prepare("SELECT hlc_wall, hlc_counter FROM sync_changelog ORDER BY id")
                .unwrap();
            s.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
                .unwrap()
                .map(|r| r.unwrap())
                .collect()
        };
        assert!(stamps.iter().all(|(w, _)| *w == 1000));
        assert_eq!(
            stamps[0].1 + 1,
            stamps[1].1,
            "counter increments within a wall_ms"
        );
    }

    #[test]
    fn committed_ledger_advances_only() {
        let (conn, _) = device(1);
        let older = Hlc {
            wall_ms: 100,
            counter: 0,
            node: node(9),
        };
        let newer = Hlc {
            wall_ms: 200,
            counter: 0,
            node: node(9),
        };
        set_committed(&conn, "note", "r", &newer).unwrap();
        set_committed(&conn, "note", "r", &older).unwrap(); // must not regress
        assert_eq!(
            committed_hlc(&conn, "note", "r").unwrap().unwrap().wall_ms,
            200
        );
    }

    #[test]
    fn conflicting_edits_converge_to_higher_hlc_both_directions() {
        // A edits at t=100, B edits at t=200 → B wins everywhere.
        let (mut a, an) = device(1);
        let (mut b, bn) = device(2);
        let a_change = local_edit_as_pulled(&a, an, "r", "from-A", 100, 1);
        let b_change = local_edit_as_pulled(&b, bn, "r", "from-B", 200, 1);

        pull_apply(&mut a, &schema(), an, vec![b_change]); // A pulls B (newer) → adopts B
        pull_apply(&mut b, &schema(), bn, vec![a_change]); // B pulls A (older) → keeps B

        assert_eq!(note_name(&a, "r").as_deref(), Some("from-B"));
        assert_eq!(note_name(&b, "r").as_deref(), Some("from-B"));
    }

    #[test]
    fn gate_drops_repulled_older_change() {
        let (mut a, an) = device(1);
        let (b, bn) = device(2);
        // B's change is applied on A.
        let b_change = local_edit_as_pulled(&b, bn, "r", "v-old", 100, 1);
        pull_apply(&mut a, &schema(), an, vec![b_change.clone()]);
        // A then makes a NEWER local edit and commits it.
        let _a_new = local_edit_as_pulled(&a, an, "r", "v-new", 300, 2);
        // Mark A's edit committed (as a push would) so the gate has a committed clock.
        let a_pending = load_local_pending(&a, an);
        record_committed(&a, &a_pending.unwrap()).unwrap();
        // Re-pulling B's OLD change must be gated out (older than committed).
        let resolved = resolve_pull(&a, &schema(), an, vec![b_change], Utc::now()).unwrap();
        assert!(
            resolved.iter().all(|e| e.row_id != "r"),
            "stale re-pull must be gated"
        );
    }

    #[test]
    fn newer_delete_beats_older_edit() {
        let (mut a, an) = device(1);
        let (b, bn) = device(2);
        // A has an older edit locally.
        local_edit_as_pulled(&a, an, "r", "edit", 100, 1);
        // B deletes the same row, newer.
        b.execute("INSERT INTO note (id, name) VALUES ('r', 'x')", [])
            .unwrap();
        stamp_pending(&b, bn, 50).unwrap();
        b.execute("DELETE FROM note WHERE id = 'r'", []).unwrap();
        stamp_pending(&b, bn, 200).unwrap();
        let del = load_local_pending(&b, bn)
            .unwrap()
            .into_iter()
            .find(|e| e.op == ChangeOp::Delete)
            .unwrap();
        let pulled = PulledChange {
            entry: del,
            device_id: bn,
            seq: 2,
        };
        pull_apply(&mut a, &schema(), an, vec![pulled]);
        assert_eq!(
            note_name(&a, "r"),
            None,
            "newer delete wins over older edit"
        );
    }

    #[test]
    fn server_order_applies_last_delivered_no_hlc() {
        let (mut a, an) = device(1);
        let s = server_order_schema();
        // Two pulled changes for the same row; server order = last wins, HLC ignored.
        let (src, sn) = device(2);
        let first = local_edit_as_pulled(&src, sn, "r", "first", 999, 1); // higher wall
        let (src2, sn2) = device(3);
        let second = local_edit_as_pulled(&src2, sn2, "r", "second", 1, 2); // lower wall, later seq
        let resolved = resolve_pull(&a, &s, an, vec![first, second], Utc::now()).unwrap();
        apply_remote_changes(&mut a, &s, &resolved, "").unwrap();
        assert_eq!(
            note_name(&a, "r").as_deref(),
            Some("second"),
            "server order: last delivered wins"
        );
    }
}