//! The conflict stash: every version last-write-wins threw away, kept.
//!
//! LWW always discards one side. Which side is a detail of clock order, so from
//! the user's seat the outcome is the same either way: an edit somebody made is
//! gone, with no record that it existed. Multi-user editing is not a first-class
//! feature in SyncKit and is not becoming one; this is the safety net under it.
//!
//! Local-only by construction. `sync_conflict_stash` is absent from every sync
//! manifest, so it is never group-scoped, never pushed, and never rides a shared
//! changelog. A stash row is per-device evidence about a decision this device
//! made, not shared state, and pushing it would leak one member's discarded
//! plaintext into a group log.
//!
//! Nothing in the engine reads these rows back. A consuming app decides whether
//! and how to surface them (a conflicts view, a badge on the row, an annotation
//! in context) and when to mark one reviewed.
//!
//! Design: wiki synckit-groups-design.

use rusqlite::Connection;

use crate::conflict::canonical_payload;
use crate::error::Result;
use crate::types::{ChangeEntry, Hlc, PulledChange};

/// Which side of the contest lost, as stored in `losing_side`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum LosingSide {
    /// This device's own edit was discarded: a remote change won.
    Local,
    /// The other writer's edit was discarded: our value stood.
    Remote,
}

impl LosingSide {
    fn as_str(self) -> &'static str {
        match self {
            LosingSide::Local => "local",
            LosingSide::Remote => "remote",
        }
    }
}

/// How many stash rows to keep per device before the oldest are trimmed.
///
/// The stash is evidence a human might read, not an audit log. Unbounded, a
/// pathological sync loop between two devices would grow it without limit; at
/// this size it stays small next to the changelog and still holds far more than
/// anyone will review.
pub(crate) const MAX_STASH_ROWS: i64 = 1_000;

/// Format an HLC for storage. Sortable and human-legible, so a stash row can be
/// ordered and read without decoding.
fn hlc_text(hlc: &Hlc) -> String {
    format!("{}:{}:{}", hlc.wall_ms, hlc.counter, hlc.node)
}

/// Record one discarded version.
///
/// Returns `Ok(false)` without writing when the two payloads are byte-identical
/// under [`canonical_payload`], the comparison [`crate::conflict::resolve_lww`]
/// already uses for its exact-HLC tiebreak. Every echo and every unchanged
/// re-save would otherwise stash, and a table full of no-ops is one nobody reads.
pub(crate) fn stash_loser(
    conn: &Connection,
    scope: &str,
    side: LosingSide,
    losing: &ChangeEntry,
    losing_device: crate::ids::DeviceId,
    winning: &ChangeEntry,
) -> Result<bool> {
    if canonical_payload(losing.data.as_ref()) == canonical_payload(winning.data.as_ref()) {
        return Ok(false);
    }

    let payload = losing
        .data
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| crate::error::SyncKitError::Internal(format!("stash payload: {e}")))?;

    conn.execute(
        "INSERT INTO sync_conflict_stash
           (table_name, row_id, scope, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
        rusqlite::params![
            losing.table,
            losing.row_id,
            scope,
            side.as_str(),
            payload,
            hlc_text(&losing.hlc),
            losing_device.to_string(),
            hlc_text(&winning.hlc),
        ],
    )?;

    tracing::debug!(
        table = %losing.table,
        row_id = %losing.row_id,
        side = side.as_str(),
        "stashed the losing side of a conflict"
    );
    Ok(true)
}

/// Record a remote change the committed-HLC gate discarded.
///
/// This is the quiet loss: no [`crate::conflict::ConflictPair`] is ever built for
/// it, because no local *pending* edit contests the row. It happens when this
/// device already applied and pushed a newer edit and then pulls an older remote
/// one, which means the other writer's edit is dropped without anything looking
/// like a conflict. There is no losing `ChangeEntry` to compare against, only the
/// committed clock, so the payload-identity check cannot apply here.
pub(crate) fn stash_superseded(
    conn: &Connection,
    scope: &str,
    dropped: &PulledChange,
    committed: &Hlc,
) -> Result<()> {
    let payload = dropped
        .entry
        .data
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| crate::error::SyncKitError::Internal(format!("stash payload: {e}")))?;

    conn.execute(
        "INSERT INTO sync_conflict_stash
           (table_name, row_id, scope, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc)
         VALUES (?1, ?2, ?3, 'remote', ?4, ?5, ?6, ?7)",
        rusqlite::params![
            dropped.entry.table,
            dropped.entry.row_id,
            scope,
            payload,
            hlc_text(&dropped.entry.hlc),
            dropped.device_id.to_string(),
            hlc_text(committed),
        ],
    )?;

    tracing::debug!(
        table = %dropped.entry.table,
        row_id = %dropped.entry.row_id,
        "stashed a remote change superseded by the committed HLC"
    );
    Ok(())
}

/// Trim the stash to [`MAX_STASH_ROWS`], oldest first. Reviewed rows are trimmed
/// like any other: marking one reviewed says a human saw it, not that it must be
/// kept forever.
pub(crate) fn trim_stash(conn: &Connection) -> Result<usize> {
    let removed = conn.execute(
        "DELETE FROM sync_conflict_stash WHERE id NOT IN
           (SELECT id FROM sync_conflict_stash ORDER BY id DESC LIMIT ?1)",
        rusqlite::params![MAX_STASH_ROWS],
    )?;
    if removed > 0 {
        tracing::debug!(removed, "trimmed the conflict stash");
    }
    Ok(removed)
}

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

    fn hlc(wall_ms: i64, counter: u32, node: &str) -> Hlc {
        Hlc {
            wall_ms,
            counter,
            node: DeviceId::new(Uuid::parse_str(node).unwrap()),
        }
    }

    #[test]
    fn hlc_text_is_wall_counter_node() {
        let h = hlc(1_700_000_000_123, 7, "3f2504e0-4f89-41d3-9a0c-0305e82c3301");
        assert_eq!(
            hlc_text(&h),
            "1700000000123:7:3f2504e0-4f89-41d3-9a0c-0305e82c3301"
        );
    }

    #[test]
    fn hlc_text_ordering_tracks_hlc_ordering() {
        // The lexical form is only order-preserving while the numeric components
        // share a width: "9:0:.." sorts above "10:0:..". Every fixture here uses
        // the same 13-digit wall_ms and single-digit counter, which is the shape
        // a live clock produces.
        let node_a = "00000000-0000-0000-0000-00000000000a";
        let node_b = "00000000-0000-0000-0000-00000000000b";
        let fixtures = [
            hlc(1_700_000_000_000, 0, node_a),
            hlc(1_700_000_000_000, 0, node_b),
            hlc(1_700_000_000_000, 1, node_a),
            hlc(1_700_000_000_000, 9, node_b),
            hlc(1_700_000_000_001, 0, node_a),
            hlc(1_899_999_999_999, 4, node_b),
        ];

        for (i, left) in fixtures.iter().enumerate() {
            for (j, right) in fixtures.iter().enumerate() {
                assert_eq!(
                    hlc_text(left).cmp(&hlc_text(right)),
                    left.cmp(right),
                    "lexical order disagreed with Hlc order for fixtures {i} and {j}"
                );
            }
        }
    }

    #[test]
    fn hlc_text_distinguishes_every_component() {
        let node_a = "00000000-0000-0000-0000-00000000000a";
        let node_b = "00000000-0000-0000-0000-00000000000b";
        let base = hlc(1_700_000_000_000, 0, node_a);
        assert_ne!(
            hlc_text(&base),
            hlc_text(&hlc(1_700_000_000_001, 0, node_a))
        );
        assert_ne!(
            hlc_text(&base),
            hlc_text(&hlc(1_700_000_000_000, 1, node_a))
        );
        assert_ne!(
            hlc_text(&base),
            hlc_text(&hlc(1_700_000_000_000, 0, node_b))
        );
    }
}