//! Hybrid logical clock persistence for cross-device sync conflict resolution.
//!
//! SyncKit orders conflicts by an [`Hlc`] rather than a bare wall-clock timestamp
//! (ultra-fuzz Run #28): a skewed-fast device no longer wins every conflict, and a
//! strictly-newer edit beats an older delete. This module owns the device's
//! persistent clock — `(wall_ms, counter)` in the `hlc_state` table; the HLC's node
//! component is always this device's id, so it is not stored.
//!
//! Local changelog rows are stamped **lazily**: when a row is first read for push or
//! conflict detection it gets the next HLC, and the stamp is written back onto the
//! row. This keeps the changelog triggers unchanged and guarantees push and
//! conflict-detection read the same HLC for a given change.

use chrono::Utc;
use goingson_core::CoreError;
use sqlx::SqlitePool;
use std::collections::{HashMap, HashSet};
use synckit_client::{DeviceId, Hlc};
use uuid::Uuid;

/// Load the persistent clock, reattaching `node` (which is not stored).
async fn load_clock(pool: &SqlitePool, node: Uuid) -> Result<Hlc, CoreError> {
    let (wall_ms, counter): (i64, i64) =
        sqlx::query_as("SELECT wall_ms, counter FROM hlc_state WHERE id = 1")
            .fetch_one(pool)
            .await
            .map_err(CoreError::database)?;
    Ok(Hlc { wall_ms, counter: counter as u32, node: DeviceId::new(node) })
}

/// Stamp every unpushed changelog row that lacks an HLC, in id order, advancing the
/// persistent clock once per row. Idempotent — already-stamped rows are skipped — so
/// push and conflict-detection observe identical stamps. Runs in one transaction so a
/// crash mid-stamp leaves the clock and the rows consistent (all-or-nothing).
#[tracing::instrument(skip_all)]
pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(), CoreError> {
    let ids: Vec<(i64, String, String)> = sqlx::query_as(
        "SELECT id, table_name, row_id FROM sync_changelog \
         WHERE pushed = 0 AND hlc_wall IS NULL ORDER BY id ASC",
    )
    .fetch_all(pool)
    .await
    .map_err(CoreError::database)?;

    if ids.is_empty() {
        return Ok(());
    }

    let mut tx = pool.begin().await.map_err(CoreError::database)?;

    let (wall_ms, counter): (i64, i64) =
        sqlx::query_as("SELECT wall_ms, counter FROM hlc_state WHERE id = 1")
            .fetch_one(&mut *tx)
            .await
            .map_err(CoreError::database)?;
    let node = DeviceId::new(device_id);
    let mut clock = Hlc { wall_ms, counter: counter as u32, node };

    for (id, table_name, row_id) in ids {
        clock = Hlc::tick(clock, Utc::now().timestamp_millis(), node);
        sqlx::query("UPDATE sync_changelog SET hlc_wall = ?, hlc_counter = ? WHERE id = ?")
            .bind(clock.wall_ms)
            .bind(clock.counter as i64)
            .bind(id)
            .execute(&mut *tx)
            .await
            .map_err(CoreError::database)?;
        // The row now holds this local value at `clock`; record it as the row's
        // committed HLC so a later older remote edit gates out instead of clobbering it.
        record_committed_hlc(&mut *tx, &table_name, &row_id, clock).await?;
    }

    sqlx::query("UPDATE hlc_state SET wall_ms = ?, counter = ? WHERE id = 1")
        .bind(clock.wall_ms)
        .bind(clock.counter as i64)
        .execute(&mut *tx)
        .await
        .map_err(CoreError::database)?;

    tx.commit().await.map_err(CoreError::database)?;
    Ok(())
}

/// Advance the persistent clock past a remote HLC observed during pull, so that
/// subsequent local changes causally follow it. Monotonic: advancing past the batch
/// maximum is enough to cover every change in the batch.
#[tracing::instrument(skip_all)]
pub async fn observe_remote(pool: &SqlitePool, remote: Hlc, device_id: Uuid) -> Result<(), CoreError> {
    let clock = load_clock(pool, device_id).await?;
    let advanced = Hlc::observe(clock, remote, Utc::now().timestamp_millis(), DeviceId::new(device_id));
    sqlx::query("UPDATE hlc_state SET wall_ms = ?, counter = ? WHERE id = 1")
        .bind(advanced.wall_ms)
        .bind(advanced.counter as i64)
        .execute(pool)
        .await
        .map_err(CoreError::database)?;
    Ok(())
}

/// Record the committed HLC for a row, keeping the maximum.
///
/// This is the committed-clock store SyncKit's `CleanChanges` gate reads: a clean
/// remote change older-or-equal to a row's committed HLC is dropped rather than
/// clobbering a newer local value. Called on every applied change — a local stamp
/// ([`assign_pending_hlcs`]) or a remote apply (`pull::apply_changes_inner`) — so
/// the stored clock always reflects the value currently in the row. The
/// `ON CONFLICT ... WHERE` keeps the larger HLC, comparing `(wall, counter, node)`
/// as a row value (node is the canonical lowercase UUID string, whose lexicographic
/// order matches the byte order [`Hlc`] compares on).
pub(crate) async fn record_committed_hlc<'e, E>(
    executor: E,
    table: &str,
    row_id: &str,
    hlc: Hlc,
) -> Result<(), CoreError>
where
    E: sqlx::Executor<'e, Database = sqlx::Sqlite>,
{
    sqlx::query(
        "INSERT INTO sync_committed_hlc (table_name, row_id, hlc_wall, hlc_counter, hlc_node) \
         VALUES (?, ?, ?, ?, ?) \
         ON CONFLICT(table_name, row_id) DO UPDATE SET \
             hlc_wall = excluded.hlc_wall, \
             hlc_counter = excluded.hlc_counter, \
             hlc_node = excluded.hlc_node \
         WHERE (excluded.hlc_wall, excluded.hlc_counter, excluded.hlc_node) > \
               (sync_committed_hlc.hlc_wall, sync_committed_hlc.hlc_counter, sync_committed_hlc.hlc_node)",
    )
    .bind(table)
    .bind(row_id)
    .bind(hlc.wall_ms)
    .bind(hlc.counter as i64)
    .bind(hlc.node.to_string())
    .execute(executor)
    .await
    .map_err(CoreError::database)?;
    Ok(())
}

/// Load the committed HLCs for a set of `(table, row_id)` keys, for pre-fetching
/// before the `CleanChanges` gate. Queries by `row_id` (one statement) and filters
/// the exact `(table, row_id)` pairs in memory, so a row never seen locally is
/// simply absent from the map (the gate then keeps that change).
pub(crate) async fn load_committed_hlcs(
    pool: &SqlitePool,
    keys: &[(String, String)],
) -> Result<HashMap<(String, String), Hlc>, CoreError> {
    let mut map = HashMap::new();
    if keys.is_empty() {
        return Ok(map);
    }

    let placeholders = vec!["?"; keys.len()].join(",");
    let sql = format!(
        "SELECT table_name, row_id, hlc_wall, hlc_counter, hlc_node \
         FROM sync_committed_hlc WHERE row_id IN ({placeholders})"
    );
    let mut query = sqlx::query_as::<_, (String, String, i64, i64, String)>(&sql);
    for (_, row_id) in keys {
        query = query.bind(row_id);
    }
    let rows = query.fetch_all(pool).await.map_err(CoreError::database)?;

    let want: HashSet<(&str, &str)> =
        keys.iter().map(|(t, r)| (t.as_str(), r.as_str())).collect();
    for (table, row_id, wall, counter, node) in rows {
        if want.contains(&(table.as_str(), row_id.as_str()))
            && let Ok(node) = Uuid::parse_str(&node)
        {
            map.insert(
                (table, row_id),
                Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(node) },
            );
        }
    }
    Ok(map)
}