//! Last-synced row snapshots: the base version a three-way field merge needs.
//!
//! [`resolve_field_merge`](crate::conflict::resolve_field_merge) has always been
//! able to merge two edits that touched different columns, and nothing could call
//! it, because a three-way merge needs the version both devices started from and
//! the engine kept no such version. This is that version.
//!
//! A snapshot is the wire payload of the last change for a row that **both
//! devices have seen**, which is exactly the two moments a row becomes common
//! ground: a remote change this device applied, and a local change the server
//! acknowledged. Written at both, so the base re-bases itself on every sync and
//! never needs a separate reconciliation pass.
//!
//! The payload is stored rather than the row read back, deliberately. A merge
//! compares two payloads against the base, and a row read carries columns the
//! wire never sends (`preserve_local` secrets, group provenance, anything outside
//! the manifest). Basing a merge on those would report a field as "changed" on
//! every device that has a different local secret. The payload is the only shape
//! all three sides share.
//!
//! Opt-in per table ([`SyncTable::field_merge`](super::schema::SyncTable::field_merge)):
//! a table nobody opted in stores nothing here and resolves conflicts exactly as
//! it did before. Storage roughly doubles for the tables that do opt in, which is
//! why it is not simply on for everything.

use rusqlite::{Connection, OptionalExtension};
use serde_json::Value;

use crate::error::Result;

/// DDL for the snapshot store, shared by
/// [`SyncSchema::migration_sql`](super::schema::SyncSchema::migration_sql) and the
/// per-connection upgrade in [`db::ensure_scope_schema`](super::db::ensure_scope_schema),
/// so an install that predates field merge gets the table on its next connection
/// open rather than waiting for the app to re-run its migration.
///
/// One row per `(table_name, row_id)`. Not scoped: a row lives in one scope at a
/// time and moving it between scopes rewrites the row, so a per-scope base would
/// be a second copy of the same answer.
pub(crate) const SNAPSHOT_DDL: &str = "\
CREATE TABLE IF NOT EXISTS sync_row_snapshot (
    table_name TEXT NOT NULL,
    row_id     TEXT NOT NULL,
    payload    TEXT NOT NULL,
    updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    PRIMARY KEY (table_name, row_id)
) WITHOUT ROWID;
";

/// Record `payload` as the last-synced base for a row.
///
/// Called on both sides of common ground: after a remote change is applied, and
/// after a local change is acknowledged by a push. A non-object payload is not
/// stored, since it could never serve as a merge base.
pub(crate) fn record(conn: &Connection, table: &str, row_id: &str, payload: &Value) -> Result<()> {
    if !payload.is_object() {
        return Ok(());
    }
    conn.execute(
        "INSERT INTO sync_row_snapshot (table_name, row_id, payload) VALUES (?1, ?2, ?3) \
         ON CONFLICT(table_name, row_id) DO UPDATE SET \
           payload = excluded.payload, \
           updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
        rusqlite::params![table, row_id, serde_json::to_string(payload)?],
    )?;
    Ok(())
}

/// Drop a row's base, because the row is gone.
///
/// A stale base outliving its row would be handed to the merge if the same key
/// were later recreated, and it would describe a version of a different row.
pub(crate) fn forget(conn: &Connection, table: &str, row_id: &str) -> Result<()> {
    conn.execute(
        "DELETE FROM sync_row_snapshot WHERE table_name = ?1 AND row_id = ?2",
        rusqlite::params![table, row_id],
    )?;
    Ok(())
}

/// The base for a row, or [`Value::Null`] when there is none.
///
/// Null rather than an error or an `Option` because that is what the merge takes:
/// [`resolve_field_merge`](crate::conflict::resolve_field_merge) treats a
/// non-object base as "no usable base" and falls back on its own. A read error is
/// logged and reported as absent, since a merge this device cannot base is a
/// merge that should not happen, not a sync that should fail.
pub(crate) fn load(conn: &Connection, table: &str, row_id: &str) -> Value {
    let stored: Option<String> = match conn
        .query_row(
            "SELECT payload FROM sync_row_snapshot WHERE table_name = ?1 AND row_id = ?2",
            rusqlite::params![table, row_id],
            |r| r.get(0),
        )
        .optional()
    {
        Ok(v) => v,
        Err(e) => {
            tracing::warn!(
                table,
                row_id,
                "snapshot lookup failed, treating as absent: {e}"
            );
            None
        }
    };
    stored
        .and_then(|s| serde_json::from_str(&s).ok())
        .unwrap_or(Value::Null)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn conn() -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(SNAPSHOT_DDL).unwrap();
        conn
    }

    #[test]
    fn round_trips_and_overwrites() {
        let c = conn();
        record(&c, "note", "r1", &json!({"name": "one"})).unwrap();
        assert_eq!(load(&c, "note", "r1"), json!({"name": "one"}));

        record(&c, "note", "r1", &json!({"name": "two"})).unwrap();
        assert_eq!(
            load(&c, "note", "r1"),
            json!({"name": "two"}),
            "a later sync must re-base the row, not accumulate versions"
        );
    }

    #[test]
    fn absent_and_forgotten_rows_read_as_null() {
        let c = conn();
        assert_eq!(load(&c, "note", "missing"), Value::Null);

        record(&c, "note", "r1", &json!({"name": "one"})).unwrap();
        forget(&c, "note", "r1").unwrap();
        assert_eq!(
            load(&c, "note", "r1"),
            Value::Null,
            "a deleted row's base must not survive to be merged against"
        );
    }

    /// A delete carries no object payload, and an entry with no payload has no
    /// base to offer. Storing one would put a scalar where the merge expects an
    /// object.
    #[test]
    fn a_non_object_payload_is_not_stored() {
        let c = conn();
        record(&c, "note", "r1", &json!("not an object")).unwrap();
        assert_eq!(load(&c, "note", "r1"), Value::Null);
    }
}