//! The storage-version gate: refuse to sync across a breaking manifest change.
//!
//! Standing policy. When a storage change is breaking, a client refuses to sync
//! rather than degrading, and the local migration runs first. The alternative
//! failure is silent: a client that syncs everything it recognises and ignores
//! the rest looks like it worked, and the user learns about the gap much later
//! from missing data rather than immediately from a refusal.
//!
//! There are two comparisons, and they answer different questions.
//!
//! **This device against its own store** ([`enforce_local`]). The store carries
//! the version of the manifest that last shaped it, in `sync_state`. A build
//! whose manifest is *older* than the store is a downgrade meeting data it does
//! not fully understand; a build whose manifest is *newer* has not run its local
//! migration yet, and migration is not a thing the sync does, it is a thing that
//! has to have already happened for the sync to be allowed. Either way the cycle
//! stops before it registers a device or reads a row.
//!
//! **This device against a peer** ([`check_peer`], driven from the pull loop).
//! Clients share one changelog, so a peer on a different manifest is the case the
//! policy actually exists for. Every pushed change seals its storage version into
//! the (end-to-end encrypted) HLC envelope, so a pull can read the other side's
//! number and refuse before anything is applied.
//!
//! An app that has not declared a version leaves both gates off, which is what
//! every consumer predating this got. Adopting the gate is
//! [`SyncSchema::storage_version`](super::schema::SyncSchema::storage_version).
//!
//! <!-- wiki: synckit-overview -->

use rusqlite::Connection;

use super::db::{get_sync_state, set_sync_state};
use super::schema::SyncSchema;
use crate::error::{Result, StorageVersionRefusal, SyncKitError, VersionSource};

/// The `sync_state` key holding the storage version this store was last shaped
/// by.
pub const STORAGE_VERSION_KEY: &str = "storage_version";

/// Read the store's stamped storage version, or `None` if it has never been
/// stamped (every store predating the gate).
pub fn stored_version(conn: &Connection) -> Result<Option<u32>> {
    match get_sync_state(conn, STORAGE_VERSION_KEY)? {
        None => Ok(None),
        Some(raw) if raw.is_empty() => Ok(None),
        Some(raw) => raw.parse::<u32>().map(Some).map_err(|_| {
            SyncKitError::Database(format!(
                "sync_state.{STORAGE_VERSION_KEY} is not an integer: {raw:?}"
            ))
        }),
    }
}

/// Stamp the store with `version`.
///
/// An app calls this at the end of its own local migration, in the same
/// transaction where it can: the stamp is the claim that the store now matches
/// the manifest, so writing it before the migration finishes is what would make a
/// half-migrated store observable to a peer.
pub fn stamp_version(conn: &Connection, version: u32) -> Result<()> {
    set_sync_state(conn, STORAGE_VERSION_KEY, &version.to_string())
}

/// Gate this device against its own store, and adopt the stamp on a store that
/// has never carried one.
///
/// Returns the declared version when the gate is on, `None` when the manifest has
/// not adopted it.
///
/// Adoption is deliberate rather than a refusal: a store with no stamp is one
/// written before the gate existed, not one written by a version we disagree
/// with, and there is nothing to migrate. Refusing there would break every
/// existing install on the upgrade that turns the gate on.
pub fn enforce_local(conn: &Connection, schema: &SyncSchema) -> Result<Option<u32>> {
    let Some(mine) = schema.declared_storage_version() else {
        return Ok(None);
    };
    match stored_version(conn)? {
        None => {
            stamp_version(conn, mine)?;
            Ok(Some(mine))
        }
        Some(theirs) if theirs == mine => Ok(Some(mine)),
        Some(theirs) => Err(SyncKitError::StorageVersion(StorageVersionRefusal {
            mine,
            theirs,
            source: VersionSource::LocalStore,
        })),
    }
}

/// Gate this device against one peer's stamp.
///
/// `theirs` is `None` for a change pushed by a build that predates the stamp, or
/// by one whose manifest declares no version. There is nothing to compare, so it
/// passes: the gate protects from the version it lands in forward, and treating
/// an absent stamp as a mismatch would refuse every pre-adoption row already on
/// the changelog.
pub fn check_peer(mine: Option<u32>, theirs: Option<u32>) -> Result<()> {
    let (Some(mine), Some(theirs)) = (mine, theirs) else {
        return Ok(());
    };
    if mine == theirs {
        return Ok(());
    }
    Err(SyncKitError::StorageVersion(StorageVersionRefusal {
        mine,
        theirs,
        source: VersionSource::Peer,
    }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::error::SyncKitError;
    use crate::store::schema::SyncTable;

    fn schema_at(version: u32) -> SyncSchema {
        SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]).storage_version(version)
    }

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

    fn db(schema: &SyncSchema) -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
            .unwrap();
        conn.execute_batch(&schema.migration_sql()).unwrap();
        conn
    }

    fn refusal(err: SyncKitError) -> StorageVersionRefusal {
        match err {
            SyncKitError::StorageVersion(r) => r,
            other => panic!("expected a storage-version refusal, got {other:?}"),
        }
    }

    #[test]
    fn a_store_that_has_never_been_stamped_adopts_the_declared_version() {
        let s = schema_at(4);
        let conn = db(&s);
        assert_eq!(stored_version(&conn).unwrap(), None);
        assert_eq!(enforce_local(&conn, &s).unwrap(), Some(4));
        assert_eq!(
            stored_version(&conn).unwrap(),
            Some(4),
            "adoption stamps, so the next open compares against a real number"
        );
    }

    #[test]
    fn an_undeclared_manifest_leaves_the_gate_off_and_stamps_nothing() {
        let s = undeclared();
        let conn = db(&s);
        assert_eq!(enforce_local(&conn, &s).unwrap(), None);
        assert_eq!(stored_version(&conn).unwrap(), None);
    }

    #[test]
    fn an_older_build_meeting_a_newer_store_refuses() {
        let conn = db(&schema_at(5));
        stamp_version(&conn, 5).unwrap();

        let r = refusal(enforce_local(&conn, &schema_at(4)).unwrap_err());
        assert_eq!((r.mine, r.theirs), (4, 5));
        assert_eq!(r.source, VersionSource::LocalStore);
        assert!(r.local_is_older());
        assert_eq!(
            r.message(),
            "This store uses a newer format. Update to sync."
        );
        assert!(r.to_string().contains("nothing was written"));
    }

    #[test]
    fn a_newer_build_meeting_an_unmigrated_store_refuses_with_the_other_message() {
        let conn = db(&schema_at(4));
        stamp_version(&conn, 4).unwrap();

        let r = refusal(enforce_local(&conn, &schema_at(5)).unwrap_err());
        assert_eq!((r.mine, r.theirs), (5, 4));
        assert!(!r.local_is_older());
        assert_eq!(
            r.message(),
            "This store has not been migrated to the current format yet."
        );
    }

    #[test]
    fn stamping_after_the_migration_reopens_the_gate() {
        let s = schema_at(5);
        let conn = db(&s);
        stamp_version(&conn, 4).unwrap();
        assert!(enforce_local(&conn, &s).is_err());

        // What an app calls at the end of its own local migration.
        stamp_version(&conn, 5).unwrap();
        assert_eq!(enforce_local(&conn, &s).unwrap(), Some(5));
    }

    #[test]
    fn a_non_integer_stamp_is_an_error_rather_than_a_silent_zero() {
        let s = schema_at(1);
        let conn = db(&s);
        set_sync_state(&conn, STORAGE_VERSION_KEY, "four").unwrap();
        assert!(matches!(
            stored_version(&conn),
            Err(SyncKitError::Database(_))
        ));
    }

    #[test]
    fn the_peer_gate_passes_only_on_equality() {
        assert!(check_peer(Some(4), Some(4)).is_ok());

        let r = refusal(check_peer(Some(4), Some(5)).unwrap_err());
        assert_eq!(r.source, VersionSource::Peer);
        assert_eq!(r.message(), "Update this device.");

        let r = refusal(check_peer(Some(5), Some(4)).unwrap_err());
        assert_eq!(r.message(), "Another device is out of date.");
    }

    /// Equality, not a floor: an additive change is still breaking for the older
    /// peer, because clients share one changelog.
    #[test]
    fn a_higher_peer_version_is_not_forward_compatible() {
        assert!(check_peer(Some(4), Some(5)).is_err());
        assert!(check_peer(Some(5), Some(4)).is_err());
    }

    #[test]
    fn an_absent_stamp_on_either_side_passes() {
        assert!(check_peer(None, Some(5)).is_ok());
        assert!(check_peer(Some(5), None).is_ok());
        assert!(check_peer(None, None).is_ok());
    }
}