//! Push/pull loops and changelog lifecycle.
//!
//! This ties B2–B4 together against a transport: the drain-loop push (stamp ->
//! batch -> send -> mark-pushed + record committed), the paginated pull (fetch ->
//! resolve -> apply -> record committed -> persist cursor), plus the initial
//! snapshot, changelog cleanup, and retention cap.
//!
//! The engine's SQLite connection is `!Send`, so every database step runs on a
//! blocking thread (`spawn_blocking`) with a fresh connection from the
//! [`DbSource`](super::db::DbSource); the transport calls are the only `.await`
//! points, and nothing `!Send` is held across them. The transport is abstracted
//! behind [`SyncTransport`] so the loops test end-to-end against an in-memory
//! fake, and `SyncKitClient` satisfies it in production.

use std::collections::HashSet;
use std::future::Future;

use chrono::Utc;
use rusqlite::Connection;

use super::apply::{ApplyOutcome, apply_remote_changes};
use super::db::{
    DbSource, device_id, get_scope_cursor, set_device_id, set_scope_cursor, set_sync_state,
};
use super::hlc::{record_committed, resolve_pull, set_committed, stamp_pending};
use super::migrate::{json_object, row_id_expr};
use super::schema::{SyncMode, SyncSchema};
use crate::client::SyncKitClient;
use crate::error::{Result, SyncKitError};
use crate::ids::{DeviceId, GroupId};
use crate::types::{ChangeEntry, ChangeOp, Device, Hlc, PulledChange};

/// A sync scope: the personal changelog, or one group's shared changelog. Each
/// scope has its own changelog partition (`sync_changelog.scope`), pull cursor
/// (`sync_scope_cursor`), key, and endpoint. See the wiki design note
/// `synckit-multiscope-design`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncScope {
    /// The user's personal data, encrypted under the master key.
    Personal,
    /// A group's shared data, encrypted under the group's GCK at `gck_version`.
    Group {
        /// Group whose shared data this scope covers.
        id: GroupId,
        /// GCK version the group's data is encrypted under.
        gck_version: i32,
    },
}

impl SyncScope {
    /// The `sync_changelog.scope` / `sync_scope_cursor.scope` string for this
    /// scope: `""` for personal, the group id otherwise.
    pub fn key(&self) -> String {
        match self {
            SyncScope::Personal => String::new(),
            SyncScope::Group { id, .. } => id.to_string(),
        }
    }
}

/// Maximum changes sent in one push batch.
pub const PUSH_BATCH_LIMIT: usize = 500;

/// The two operations the sync loops need from a server.
///
/// A seam for testing (an in-memory fake) and the reference point a future
/// non-Rust SDK mirrors. [`SyncKitClient`] implements it by forwarding to its
/// inherent methods.
pub trait SyncTransport: Sync {
    /// Register (or look up) this device with the server.
    fn register_device(
        &self,
        device_name: &str,
        platform: &str,
    ) -> impl Future<Output = Result<Device>> + Send;

    /// Push encrypted changes; returns the server's new cursor.
    fn push(
        &self,
        device_id: DeviceId,
        changes: Vec<ChangeEntry>,
    ) -> impl Future<Output = Result<i64>> + Send;

    /// Pull decrypted changes since `cursor`; returns `(changes, new_cursor,
    /// has_more)` with each change's originating device and sequence.
    fn pull_rich(
        &self,
        device_id: DeviceId,
        cursor: i64,
    ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send;

    /// The groups this user belongs to, as `(group_id, current_gck_version)`,
    /// the scopes to sync beyond personal. Defaults to none, so a transport that
    /// does not support groups (a test double, a future minimal SDK) never syncs
    /// any group scope.
    // `-> impl Future + Send` (not `async fn`) keeps the explicit Send bound the
    // store's spawned tasks need; that is the whole trait's convention.
    #[allow(clippy::manual_async_fn)]
    fn list_group_scopes(&self) -> impl Future<Output = Result<Vec<(GroupId, i32)>>> + Send {
        async move { Ok(Vec::new()) }
    }

    /// Push encrypted changes to a group's shared changelog. The implementation
    /// resolves the group's GCK internally; the caller never handles keys. The
    /// default errors, it is only reached if [`list_group_scopes`] returns a
    /// group, which the default never does.
    #[allow(clippy::manual_async_fn)]
    fn group_scope_push(
        &self,
        _group_id: GroupId,
        _gck_version: i32,
        _device_id: DeviceId,
        _changes: Vec<ChangeEntry>,
    ) -> impl Future<Output = Result<i64>> + Send {
        async move {
            Err(SyncKitError::Internal(
                "group sync not supported by this transport".into(),
            ))
        }
    }

    /// Pull a group's changes since `cursor`, decrypted under its GCK. Default
    /// errors; see [`group_scope_push`](Self::group_scope_push).
    #[allow(clippy::manual_async_fn)]
    fn group_scope_pull(
        &self,
        _group_id: GroupId,
        _gck_version: i32,
        _device_id: DeviceId,
        _cursor: i64,
    ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
        async move {
            Err(SyncKitError::Internal(
                "group sync not supported by this transport".into(),
            ))
        }
    }
}

// The group methods below return `-> impl Future + Send` with an `async move`
// body (not `async fn`) to keep the explicit Send bound the trait requires.
#[allow(clippy::manual_async_fn)]
impl SyncTransport for SyncKitClient {
    fn register_device(
        &self,
        device_name: &str,
        platform: &str,
    ) -> impl Future<Output = Result<Device>> + Send {
        SyncKitClient::register_device(self, device_name, platform)
    }

    fn push(
        &self,
        device_id: DeviceId,
        changes: Vec<ChangeEntry>,
    ) -> impl Future<Output = Result<i64>> + Send {
        // Inherent method (preferred over the trait method of the same name).
        SyncKitClient::push(self, device_id, changes)
    }

    fn pull_rich(
        &self,
        device_id: DeviceId,
        cursor: i64,
    ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
        SyncKitClient::pull_rich(self, device_id, cursor)
    }

    fn list_group_scopes(&self) -> impl Future<Output = Result<Vec<(GroupId, i32)>>> + Send {
        async move {
            Ok(self
                .list_groups()
                .await?
                .into_iter()
                .map(|g| (g.id, g.gck_version))
                .collect())
        }
    }

    fn group_scope_push(
        &self,
        group_id: GroupId,
        gck_version: i32,
        device_id: DeviceId,
        changes: Vec<ChangeEntry>,
    ) -> impl Future<Output = Result<i64>> + Send {
        async move {
            let gck = self.group_content_key(group_id, gck_version).await?;
            self.group_push(group_id, &gck, device_id, changes).await
        }
    }

    fn group_scope_pull(
        &self,
        group_id: GroupId,
        gck_version: i32,
        device_id: DeviceId,
        cursor: i64,
    ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
        async move {
            let gck = self.group_content_key(group_id, gck_version).await?;
            match self
                .group_pull_rich(group_id, &gck, device_id, cursor)
                .await
            {
                // A decrypt failure means this client is holding a stale GCK (it
                // missed a rotation): drop it, re-fetch the current grant, retry
                // once. A second failure surfaces.
                Err(SyncKitError::DecryptionFailed) => {
                    self.invalidate_gck(group_id);
                    let gck = self.group_content_key(group_id, gck_version).await?;
                    self.group_pull_rich(group_id, &gck, device_id, cursor)
                        .await
                }
                other => other,
            }
        }
    }
}

/// Ensure this install has a registered device, returning its id (which doubles
/// as the HLC node). Fast path: a stored `device_id` returns without a network
/// call. Otherwise register with the server and persist the assigned id.
pub async fn ensure_device_registered<T: SyncTransport>(
    db: &DbSource,
    client: &T,
    device_name: &str,
    platform: &str,
) -> Result<DeviceId> {
    let db_read = db.clone();
    let existing = tokio::task::spawn_blocking(move || device_id(&db_read.open()?))
        .await
        .map_err(|e| join_err(&e))??;
    if let Some(id) = existing {
        return Ok(id);
    }

    let device = client.register_device(device_name, platform).await?;
    let id = device.id;

    let db_write = db.clone();
    tokio::task::spawn_blocking(move || set_device_id(&db_write.open()?, id))
        .await
        .map_err(|e| join_err(&e))??;
    Ok(id)
}

/// Outcome of a pull pass.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct PullOutcome {
    /// Number of remote changes applied.
    pub applied: u64,
    /// Tables whose rows changed as a result of the pull.
    pub changed_tables: HashSet<String>,
}

fn join_err(e: &tokio::task::JoinError) -> SyncKitError {
    SyncKitError::Internal(format!("blocking task failed: {e}"))
}

// push

struct PushBatch {
    wire: Vec<ChangeEntry>,
    wire_ids: Vec<i64>,
    /// Rows marked pushed without sending (corrupt/unknown-op) so they can't wedge.
    skip_ids: Vec<i64>,
    total_read: usize,
}

/// Drain all unpushed local changes for the **personal** scope. Thin wrapper over
/// [`push_scope`] preserved for the personal call sites.
pub async fn push_changes<T: SyncTransport>(
    db: &DbSource,
    client: &T,
    device_id: DeviceId,
) -> Result<u64> {
    push_scope(db, client, device_id, SyncScope::Personal).await
}

/// Drain all unpushed local changes for one `scope` to the server, batch by batch.
///
/// Each batch is stamped and read on a blocking thread, sent to the scope's
/// endpoint (personal push, or a group push under its GCK), then marked pushed
/// with its HLC recorded as the committed clock (so a peer's later stale re-pull
/// of this row is gated out). `device_id` doubles as the HLC node. The drain
/// reads only rows whose `sync_changelog.scope` matches, so a group's changes
/// never leak into the personal push and vice versa.
pub async fn push_scope<T: SyncTransport>(
    db: &DbSource,
    client: &T,
    device_id: DeviceId,
    scope: SyncScope,
) -> Result<u64> {
    let scope_key = scope.key();
    let mut pushed_total = 0u64;
    loop {
        let db_read = db.clone();
        let sk = scope_key.clone();
        let batch =
            tokio::task::spawn_blocking(move || read_push_batch(&db_read.open()?, device_id, &sk))
                .await
                .map_err(|e| join_err(&e))??;

        if batch.wire.is_empty() && batch.skip_ids.is_empty() {
            break;
        }

        if !batch.wire.is_empty() {
            match scope {
                SyncScope::Personal => {
                    client.push(device_id, batch.wire.clone()).await?;
                }
                SyncScope::Group { id, gck_version } => {
                    client
                        .group_scope_push(id, gck_version, device_id, batch.wire.clone())
                        .await?;
                }
            }
        }

        let db_write = db.clone();
        let PushBatch {
            wire,
            wire_ids,
            skip_ids,
            total_read,
        } = batch;
        let sent = wire.len() as u64;
        tokio::task::spawn_blocking(move || {
            mark_pushed_committed(&mut db_write.open()?, &wire_ids, &skip_ids, &wire)
        })
        .await
        .map_err(|e| join_err(&e))??;

        pushed_total += sent;
        if total_read < PUSH_BATCH_LIMIT {
            break;
        }
    }
    Ok(pushed_total)
}

fn read_push_batch(conn: &Connection, device_id: DeviceId, scope: &str) -> Result<PushBatch> {
    stamp_pending(conn, device_id, Utc::now().timestamp_millis())?;
    let mut stmt = conn.prepare(
        "SELECT id, table_name, op, row_id, data, hlc_wall, hlc_counter \
         FROM sync_changelog WHERE pushed = 0 AND scope = ?2 ORDER BY id LIMIT ?1",
    )?;
    let rows = stmt.query_map(rusqlite::params![PUSH_BATCH_LIMIT as i64, scope], |r| {
        Ok((
            r.get::<_, i64>(0)?,
            r.get::<_, String>(1)?,
            r.get::<_, String>(2)?,
            r.get::<_, String>(3)?,
            r.get::<_, Option<String>>(4)?,
            r.get::<_, i64>(5)?,
            r.get::<_, i64>(6)?,
        ))
    })?;

    let mut batch = PushBatch {
        wire: Vec::new(),
        wire_ids: Vec::new(),
        skip_ids: Vec::new(),
        total_read: 0,
    };
    for row in rows {
        batch.total_read += 1;
        let (id, table, op, row_id, data, wall_ms, counter) = row?;
        // Corrupt/unknown rows are marked pushed (not sent) so they can't wedge
        // the drain by being re-read forever.
        let Some(op) = ChangeOp::from_str_opt(&op) else {
            batch.skip_ids.push(id);
            continue;
        };
        let Ok(data) = data.as_deref().map(serde_json::from_str).transpose() else {
            batch.skip_ids.push(id);
            continue;
        };
        batch.wire.push(ChangeEntry {
            table,
            op,
            row_id,
            timestamp: Utc::now(),
            hlc: Hlc {
                wall_ms,
                counter: counter as u32,
                node: device_id,
            },
            data,
            extra: serde_json::Map::default(),
        });
        batch.wire_ids.push(id);
    }
    Ok(batch)
}

fn mark_pushed_committed(
    conn: &mut Connection,
    wire_ids: &[i64],
    skip_ids: &[i64],
    wire: &[ChangeEntry],
) -> Result<()> {
    let tx = conn.transaction()?;
    for id in wire_ids.iter().chain(skip_ids) {
        tx.execute("UPDATE sync_changelog SET pushed = 1 WHERE id = ?1", [id])?;
    }
    // Push-side committed-ledger update: our own edit's HLC becomes the row's
    // committed clock, so a peer's later stale re-pull of it is gated out.
    for e in wire {
        set_committed(&tx, &e.table, &e.row_id, &e.hlc)?;
    }
    tx.commit()?;
    Ok(())
}

// pull

/// Pull, resolve, and apply all remote changes for the **personal** scope. Thin
/// wrapper over [`pull_scope`] preserved for the personal call sites.
pub async fn pull_changes<T: SyncTransport>(
    db: &DbSource,
    client: &T,
    schema: &SyncSchema,
    device_id: DeviceId,
) -> Result<PullOutcome> {
    pull_scope(db, client, schema, device_id, SyncScope::Personal).await
}

/// Pull, resolve, and apply all remote changes for one `scope`, page by page.
///
/// Reads from the scope's own cursor (`sync_scope_cursor`), pulls from the
/// scope's endpoint (personal pull, or a group pull under its GCK), and applies
/// with the scope stamped so group rows are tagged with their group. Follows the
/// SDK contract: the cursor advances only *after* the batch is applied and
/// committed, in the same blocking step, so a crash re-pulls a batch the
/// idempotent apply/gate absorbs.
pub async fn pull_scope<T: SyncTransport>(
    db: &DbSource,
    client: &T,
    schema: &SyncSchema,
    device_id: DeviceId,
    scope: SyncScope,
) -> Result<PullOutcome> {
    let scope_key = scope.key();
    let mut out = PullOutcome::default();
    loop {
        let db_cursor = db.clone();
        let sk = scope_key.clone();
        let cursor = tokio::task::spawn_blocking(move || get_scope_cursor(&db_cursor.open()?, &sk))
            .await
            .map_err(|e| join_err(&e))??;

        let (pulled, new_cursor, has_more) = match scope {
            SyncScope::Personal => client.pull_rich(device_id, cursor).await?,
            SyncScope::Group { id, gck_version } => {
                client
                    .group_scope_pull(id, gck_version, device_id, cursor)
                    .await?
            }
        };

        if pulled.is_empty() {
            let db_c = db.clone();
            let sk = scope_key.clone();
            tokio::task::spawn_blocking(move || set_scope_cursor(&db_c.open()?, &sk, new_cursor))
                .await
                .map_err(|e| join_err(&e))??;
            break;
        }

        let db_apply = db.clone();
        let schema = schema.clone();
        let sk = scope_key.clone();
        let outcome = tokio::task::spawn_blocking(move || {
            apply_pull(
                &mut db_apply.open()?,
                &schema,
                device_id,
                pulled,
                new_cursor,
                &sk,
            )
        })
        .await
        .map_err(|e| join_err(&e))??;

        out.applied += outcome.applied as u64;
        out.changed_tables.extend(outcome.changed_tables);
        if !has_more {
            break;
        }
    }
    Ok(out)
}

fn apply_pull(
    conn: &mut Connection,
    schema: &SyncSchema,
    device_id: DeviceId,
    pulled: Vec<PulledChange>,
    new_cursor: i64,
    scope_key: &str,
) -> Result<ApplyOutcome> {
    let resolved = resolve_pull(conn, schema, device_id, pulled, Utc::now())?;
    let outcome = apply_remote_changes(conn, schema, &resolved, scope_key)?;
    record_committed(conn, &resolved)?;
    set_scope_cursor(conn, scope_key, new_cursor)?;
    Ok(outcome)
}

// lifecycle (synchronous)

/// One-time backfill: snapshot every existing row into the changelog so a fresh
/// device's pre-existing data pushes on the first sync. Idempotent, rows already
/// represented in the changelog are skipped. Reuses the trigger's row-id and
/// payload expressions, so snapshot entries are byte-identical to trigger entries.
pub fn create_initial_snapshot(conn: &Connection, schema: &SyncSchema) -> Result<u64> {
    let mut total = 0u64;
    for table in &schema.tables {
        let name = table.name;
        let cols = table.emitted_columns();
        let row_id = row_id_expr(table, name);
        let data = json_object(name, &cols);
        // PartialUpdate tables reflect state changes; snapshot them as UPDATEs so
        // a peer applies them through the same partial path.
        let op = if matches!(table.mode, SyncMode::PartialUpdate { .. }) {
            "UPDATE"
        } else {
            "INSERT"
        };
        let exclude = table
            .exclude_where
            .map(|p| format!(" AND ({})", p.replace("{row}", name)))
            .unwrap_or_default();

        let sql = format!(
            "INSERT INTO sync_changelog (table_name, op, row_id, data) \
             SELECT '{name}', '{op}', {row_id}, {data} FROM {name} \
             WHERE NOT EXISTS ( \
                 SELECT 1 FROM sync_changelog sc \
                 WHERE sc.table_name = '{name}' AND sc.row_id = {row_id} \
             ){exclude}"
        );
        total += conn.execute(&sql, [])? as u64;
    }
    set_sync_state(conn, "initial_snapshot_done", "1")?;
    Ok(total)
}

/// Prune pushed changelog entries older than seven days.
pub fn cleanup_changelog(conn: &Connection) -> Result<u64> {
    let n = conn.execute(
        "DELETE FROM sync_changelog WHERE pushed = 1 AND timestamp < datetime('now', '-7 days')",
        [],
    )?;
    Ok(n as u64)
}

/// Cap changelog size: drop the oldest *pushed* entries beyond the most recent
/// `cap`. Unpushed entries are never dropped (that would lose un-synced data), so
/// a backlog larger than `cap` is preserved until it syncs.
pub fn enforce_changelog_retention(conn: &Connection, cap: i64) -> Result<u64> {
    let n = conn.execute(
        "DELETE FROM sync_changelog WHERE pushed = 1 AND id NOT IN ( \
             SELECT id FROM sync_changelog ORDER BY id DESC LIMIT ?1 \
         )",
        [cap],
    )?;
    Ok(n as u64)
}

#[cfg(test)]
mod tests {
    use super::super::db::get_sync_state_or;
    use super::super::schema::SyncTable;
    use super::*;
    use std::sync::{Arc, Mutex};

    /// A shared in-memory "server": an append-only personal log of (origin
    /// device, entry), plus a group log of (group, origin device, entry).
    #[derive(Clone, Default)]
    struct FakeServer {
        log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>,
        group_log: Arc<Mutex<Vec<(GroupId, DeviceId, ChangeEntry)>>>,
    }

    impl SyncTransport for FakeServer {
        fn group_scope_push(
            &self,
            group_id: GroupId,
            _gck_version: i32,
            device_id: DeviceId,
            changes: Vec<ChangeEntry>,
        ) -> impl Future<Output = Result<i64>> + Send {
            let group_log = self.group_log.clone();
            async move {
                let mut l = group_log.lock().unwrap();
                for c in changes {
                    l.push((group_id, device_id, c));
                }
                Ok(l.len() as i64)
            }
        }

        async fn register_device(&self, _name: &str, _platform: &str) -> Result<Device> {
            Ok(Device {
                id: DeviceId::new(uuid::Uuid::from_u128(0xDE)),
                app_id: crate::ids::AppId::nil(),
                user_id: crate::ids::UserId::nil(),
                device_name: "fake".into(),
                platform: "test".into(),
                last_seen_at: Utc::now(),
                created_at: Utc::now(),
            })
        }

        fn push(
            &self,
            device_id: DeviceId,
            changes: Vec<ChangeEntry>,
        ) -> impl Future<Output = Result<i64>> + Send {
            let log = self.log.clone();
            async move {
                let mut l = log.lock().unwrap();
                for c in changes {
                    l.push((device_id, c));
                }
                Ok(l.len() as i64)
            }
        }

        fn pull_rich(
            &self,
            _device_id: DeviceId,
            cursor: i64,
        ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
            let log = self.log.clone();
            async move {
                let l = log.lock().unwrap();
                let out: Vec<PulledChange> = l
                    .iter()
                    .enumerate()
                    .filter(|(i, _)| (*i as i64 + 1) > cursor)
                    .map(|(i, (dev, entry))| PulledChange {
                        entry: entry.clone(),
                        device_id: *dev,
                        seq: i as i64 + 1,
                    })
                    .collect();
                Ok((out, l.len() as i64, false))
            }
        }
    }

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

    /// A schema with a group-scoped table: `task` carries a local `group_id`
    /// provenance column (not a synced column), declared via `group_scoped`.
    fn group_schema() -> SyncSchema {
        SyncSchema::new(vec![
            SyncTable::full("task", &["id", "name"]).group_scoped("group_id"),
        ])
    }

    fn group_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
        let db = DbSource::path(path);
        let conn = db.open().unwrap();
        conn.execute_batch("CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);")
            .unwrap();
        conn.execute_batch(&group_schema().migration_sql()).unwrap();
        (db, DeviceId::new(uuid::Uuid::from_u128(n)))
    }

    #[tokio::test]
    async fn push_scope_drains_only_its_own_scope() {
        let dir = tempdir();
        let (db, node) = group_device(&dir.join("g.db"), 7);
        let server = FakeServer::default();
        let gid = GroupId::new(uuid::Uuid::from_u128(0x6971));

        // A personal task (group_id NULL) and a group task (group_id = gid).
        {
            let c = db.open().unwrap();
            c.execute(
                "INSERT INTO task (id, name, group_id) VALUES ('p', 'personal', NULL)",
                [],
            )
            .unwrap();
            c.execute(
                "INSERT INTO task (id, name, group_id) VALUES ('g', 'grouped', ?1)",
                [gid.to_string()],
            )
            .unwrap();
        }

        // Personal push drains only the personal row.
        let pushed = push_scope(&db, &server, node, SyncScope::Personal)
            .await
            .unwrap();
        assert_eq!(pushed, 1);
        assert_eq!(server.log.lock().unwrap().len(), 1);
        assert_eq!(server.log.lock().unwrap()[0].1.row_id, "p");
        assert!(server.group_log.lock().unwrap().is_empty());

        // Group push drains only the group row, to that group.
        let pushed = push_scope(
            &db,
            &server,
            node,
            SyncScope::Group {
                id: gid,
                gck_version: 1,
            },
        )
        .await
        .unwrap();
        assert_eq!(pushed, 1);
        let gl = server.group_log.lock().unwrap();
        assert_eq!(gl.len(), 1);
        assert_eq!(gl[0].0, gid);
        assert_eq!(gl[0].2.row_id, "g");
        // The personal log did not grow.
        assert_eq!(server.log.lock().unwrap().len(), 1);
    }

    fn device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
        let db = DbSource::path(path);
        let conn = db.open().unwrap();
        conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
            .unwrap();
        conn.execute_batch(&schema().migration_sql()).unwrap();
        (db, DeviceId::new(uuid::Uuid::from_u128(n)))
    }

    fn edit(db: &DbSource, id: &str, name: &str) {
        let conn = db.open().unwrap();
        conn.execute(
            "INSERT INTO note (id, name) VALUES (?1, ?2) ON CONFLICT(id) DO UPDATE SET name = excluded.name",
            (id, name),
        )
        .unwrap();
    }

    fn stamp_at(db: &DbSource, node: DeviceId, now_ms: i64) {
        stamp_pending(&db.open().unwrap(), node, now_ms).unwrap();
    }

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

    #[tokio::test]
    async fn two_device_push_pull_converges_to_higher_hlc() {
        let dir = tempdir();
        let (da, na) = device(&dir.join("a.db"), 1);
        let (db_, nb) = device(&dir.join("b.db"), 2);
        let server = FakeServer::default();

        // A edits first (t=100), B edits the same row later (t=200) → B wins.
        edit(&da, "r", "from-A");
        stamp_at(&da, na, 100);
        edit(&db_, "r", "from-B");
        stamp_at(&db_, nb, 200);

        push_changes(&da, &server, na).await.unwrap();
        push_changes(&db_, &server, nb).await.unwrap();

        pull_changes(&da, &server, &schema(), na).await.unwrap();
        pull_changes(&db_, &server, &schema(), nb).await.unwrap();

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

    #[tokio::test]
    async fn push_marks_rows_and_advances_cursor() {
        let dir = tempdir();
        let (da, na) = device(&dir.join("a.db"), 1);
        let server = FakeServer::default();
        edit(&da, "r1", "x");
        edit(&da, "r2", "y");

        let pushed = push_changes(&da, &server, na).await.unwrap();
        assert_eq!(pushed, 2);
        // All local rows are marked pushed.
        let unpushed: i64 = da
            .open()
            .unwrap()
            .query_row(
                "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(unpushed, 0);

        // A fresh device pulls both and advances its cursor to 2.
        let (db_, nb) = device(&dir.join("b.db"), 2);
        let out = pull_changes(&db_, &server, &schema(), nb).await.unwrap();
        assert_eq!(out.applied, 2);
        assert_eq!(note_name(&db_, "r1").as_deref(), Some("x"));
        // Personal pull advances the personal ('') scope cursor.
        let cursor = get_scope_cursor(&db_.open().unwrap(), "").unwrap();
        assert_eq!(cursor, 2);
    }

    #[test]
    fn initial_snapshot_captures_existing_rows_once() {
        let dir = tempdir();
        let (da, _) = device(&dir.join("a.db"), 1);
        let conn = da.open().unwrap();
        // Pre-existing rows inserted with triggers suppressed (as if before sync).
        conn.execute("INSERT INTO note (id, name) VALUES ('r1', 'a')", [])
            .unwrap();
        conn.execute("INSERT INTO note (id, name) VALUES ('r2', 'b')", [])
            .unwrap();
        // (the inserts above DID fire triggers; clear the changelog to simulate a
        // pre-sync backfill scenario cleanly)
        conn.execute("DELETE FROM sync_changelog", []).unwrap();

        let n = create_initial_snapshot(&conn, &schema()).unwrap();
        assert_eq!(n, 2);
        // Idempotent: a second snapshot adds nothing.
        assert_eq!(create_initial_snapshot(&conn, &schema()).unwrap(), 0);
        let done = get_sync_state_or(&conn, "initial_snapshot_done", "0").unwrap();
        assert_eq!(done, "1");
    }

    #[test]
    fn retention_caps_pushed_but_keeps_unpushed() {
        let dir = tempdir();
        let (da, _) = device(&dir.join("a.db"), 1);
        let conn = da.open().unwrap();
        conn.execute("DELETE FROM sync_changelog", []).unwrap();
        // 5 pushed + 2 unpushed entries.
        for i in 0..5 {
            conn.execute(
                "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',1)",
                [format!("p{i}")],
            )
            .unwrap();
        }
        for i in 0..2 {
            conn.execute(
                "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',0)",
                [format!("u{i}")],
            )
            .unwrap();
        }
        // Cap to 3: keep the 3 most recent rows, drop older PUSHED ones only.
        let dropped = enforce_changelog_retention(&conn, 3).unwrap();
        let unpushed: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(unpushed, 2, "unpushed entries are never dropped");
        assert!(dropped >= 1);
    }

    // minimal temp-dir helper (no external dep)
    fn tempdir() -> std::path::PathBuf {
        let mut p = std::env::temp_dir();
        // Unique-ish per test via a monotonic counter; tests run in the same
        // process so a static AtomicU64 keeps paths distinct.
        use std::sync::atomic::{AtomicU64, Ordering};
        static N: AtomicU64 = AtomicU64::new(0);
        p.push(format!(
            "synckit_b5_{}_{}",
            std::process::id(),
            N.fetch_add(1, Ordering::Relaxed)
        ));
        std::fs::create_dir_all(&p).unwrap();
        p
    }
}