//! The `SyncStore` facade: the public handle that assembles the schema, the
//! client, and the database into a one-call sync (`sync_now`) and an optional
//! background scheduler.

use std::collections::HashSet;
use std::sync::Arc;
use std::time::{Duration, Instant};

use chrono::{DateTime, Utc};

use super::blob::{BlobOutcome, BlobPolicy, BlobTransport, sync_blobs};
use super::db::{
    DbSource, clear_applying_remote, count_pending_changes, get_sync_state, get_sync_state_or,
    set_sync_state,
};
use super::scheduler::{
    SyncObserver, SyncState, backoff_delay, interval_elapsed, is_auth_lost,
    is_subscription_required,
};
use super::schema::SyncSchema;
use super::sync::{
    SyncScope, SyncTransport, cleanup_changelog, create_initial_snapshot,
    enforce_changelog_retention, ensure_device_registered, pull_changes, pull_scope, push_changes,
    push_scope,
};
use crate::client::SyncKitClient;
use crate::error::{Result, SyncKitError};

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

/// Tunables for a [`SyncStore`].
#[derive(Debug, Clone)]
pub struct SyncConfig {
    /// How often the scheduler wakes to consider syncing (the timer granularity;
    /// the actual sync cadence is the `sync_interval_minutes` sync-state value).
    pub check_interval: Duration,
    /// Changelog size cap enforced after each cycle (`None` = no cap).
    pub retention_cap: Option<i64>,
}

impl Default for SyncConfig {
    fn default() -> Self {
        Self {
            check_interval: Duration::from_mins(1),
            retention_cap: None,
        }
    }
}

/// Outcome of one [`SyncStore::sync_now`] cycle.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct SyncOutcome {
    /// Number of local changes pushed to the server.
    pub pushed: u64,
    /// Number of remote changes pulled and applied.
    pub pulled: u64,
    /// Tables whose rows changed as a result of the pull.
    pub changed_tables: HashSet<String>,
    /// Blob upload/download tallies for this cycle.
    pub blobs: BlobOutcome,
}

/// The engine handle. Generic over the transport so it can be driven by the real
/// [`SyncKitClient`] or a test double; the scheduler loop is available only for
/// the concrete client (it needs SSE and auth state).
pub struct SyncStore<C> {
    db: DbSource,
    client: Arc<C>,
    schema: SyncSchema,
    device_name: String,
    platform: String,
    config: SyncConfig,
    blob_policy: Option<Arc<dyn BlobPolicy>>,
}

/// Builder for [`SyncStore`].
pub struct SyncStoreBuilder<C> {
    db: DbSource,
    client: C,
    schema: SyncSchema,
    device_name: String,
    platform: String,
    config: SyncConfig,
    blob_policy: Option<Arc<dyn BlobPolicy>>,
}

fn default_device_name() -> String {
    std::env::var("HOSTNAME")
        .or_else(|_| std::env::var("COMPUTERNAME"))
        .unwrap_or_else(|_| "SyncKit Device".to_string())
}

impl<C> SyncStore<C> {
    /// Start building a store from its database, client, and schema.
    pub fn builder(db: DbSource, client: C, schema: SyncSchema) -> SyncStoreBuilder<C> {
        SyncStoreBuilder {
            db,
            client,
            schema,
            device_name: default_device_name(),
            platform: std::env::consts::OS.to_string(),
            config: SyncConfig::default(),
            blob_policy: None,
        }
    }

    /// The engine's device name (for display / registration).
    pub fn device_name(&self) -> &str {
        &self.device_name
    }

    /// The underlying transport client. A consumer that also drives auth,
    /// subscription, or tier flows against the same client (GoingsOn's sync
    /// commands) reaches it here instead of holding a second handle.
    pub fn client(&self) -> &Arc<C> {
        &self.client
    }
}

impl<C> SyncStoreBuilder<C> {
    /// Override the device name and platform used at registration.
    #[must_use]
    pub fn device(mut self, name: impl Into<String>, platform: impl Into<String>) -> Self {
        self.device_name = name.into();
        self.platform = platform.into();
        self
    }

    /// Override the config.
    #[must_use]
    pub fn config(mut self, config: SyncConfig) -> Self {
        self.config = config;
        self
    }

    /// Attach a blob policy (enables the blob pass in each cycle).
    #[must_use]
    pub fn blob_policy(mut self, policy: Arc<dyn BlobPolicy>) -> Self {
        self.blob_policy = Some(policy);
        self
    }

    /// Finish building.
    pub fn build(self) -> SyncStore<C> {
        SyncStore {
            db: self.db,
            client: Arc::new(self.client),
            schema: self.schema,
            device_name: self.device_name,
            platform: self.platform,
            config: self.config,
            blob_policy: self.blob_policy,
        }
    }
}

impl<C: SyncTransport + BlobTransport> SyncStore<C> {
    /// Run one full sync cycle: ensure the device is registered, take the initial
    /// snapshot if needed, push local changes, pull and apply remote changes, sync
    /// blobs (if a policy is set), then clean up and stamp the last-sync time.
    pub async fn sync_now(&self) -> Result<SyncOutcome> {
        // Defensive: a database written by an older client might carry a stale flag.
        self.blocking(clear_applying_remote).await?;

        let device_id =
            ensure_device_registered(&self.db, &*self.client, &self.device_name, &self.platform)
                .await?;
        self.maybe_snapshot().await?;

        let mut pushed = push_changes(&self.db, &*self.client, device_id).await?;
        let pull = pull_changes(&self.db, &*self.client, &self.schema, device_id).await?;
        let mut pulled = pull.applied;
        let mut changed_tables = pull.changed_tables;

        // Group scopes: sync each group the user belongs to, under its own key and
        // cursor. A transport without groups reports none (the default), so this
        // loop is a no-op for personal-only apps.
        for (id, gck_version) in self.client.list_group_scopes().await? {
            let scope = SyncScope::Group { id, gck_version };
            pushed += push_scope(&self.db, &*self.client, device_id, scope).await?;
            let gp = pull_scope(&self.db, &*self.client, &self.schema, device_id, scope).await?;
            pulled += gp.applied;
            changed_tables.extend(gp.changed_tables);
        }

        let blobs = match &self.blob_policy {
            Some(policy) => sync_blobs(&self.db, &*self.client, policy).await?,
            None => BlobOutcome::default(),
        };

        self.finalize().await?;
        Ok(SyncOutcome {
            pushed,
            pulled,
            changed_tables,
            blobs,
        })
    }

    /// Count unpushed local changes.
    pub async fn pending_changes(&self) -> Result<i64> {
        self.blocking(count_pending_changes).await
    }

    async fn maybe_snapshot(&self) -> Result<()> {
        let db = self.db.clone();
        let schema = self.schema.clone();
        tokio::task::spawn_blocking(move || -> Result<()> {
            let conn = db.open()?;
            if get_sync_state_or(&conn, "initial_snapshot_done", "0")? != "1" {
                create_initial_snapshot(&conn, &schema)?;
            }
            Ok(())
        })
        .await
        .map_err(|e| join_err(&e))?
    }

    async fn finalize(&self) -> Result<()> {
        let db = self.db.clone();
        let cap = self.config.retention_cap;
        tokio::task::spawn_blocking(move || -> Result<()> {
            let conn = db.open()?;
            cleanup_changelog(&conn)?;
            if let Some(c) = cap {
                enforce_changelog_retention(&conn, c)?;
            }
            set_sync_state(&conn, "last_sync_at", &Utc::now().to_rfc3339())?;
            Ok(())
        })
        .await
        .map_err(|e| join_err(&e))?
    }

    /// Run a synchronous DB closure on a blocking thread with a fresh connection.
    async fn blocking<F, R>(&self, f: F) -> Result<R>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R> + Send + 'static,
        R: Send + 'static,
    {
        let db = self.db.clone();
        tokio::task::spawn_blocking(move || f(&db.open()?))
            .await
            .map_err(|e| join_err(&e))?
    }
}

/// Handle to a running background scheduler. Dropping it does not stop the loop;
/// call [`stop`](Self::stop).
pub struct SchedulerHandle {
    handle: tokio::task::JoinHandle<()>,
}

impl SchedulerHandle {
    /// Stop the scheduler.
    pub fn stop(self) {
        self.handle.abort();
    }
}

// The scheduler and session teardown need the concrete client's SSE stream and
// auth state, so they are not part of the generic impl.
impl SyncStore<SyncKitClient> {
    /// Spawn the background sync loop, reporting through `observer`. The loop
    /// races a timer against the server's change notifications, applies the gate
    /// chain (authenticated, key loaded, auto-sync enabled, not backed off), and
    /// backs off exponentially on failure (an hour on a subscription block).
    pub fn spawn_scheduler(self: Arc<Self>, observer: Arc<dyn SyncObserver>) -> SchedulerHandle {
        let handle = tokio::spawn(async move { self.run_scheduler(observer).await });
        SchedulerHandle { handle }
    }

    async fn run_scheduler(self: Arc<Self>, observer: Arc<dyn SyncObserver>) {
        let mut failures: u32 = 0;
        let mut backoff_until: Option<Instant> = None;
        let mut ticker = tokio::time::interval(self.config.check_interval);
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        // The SDK's stream auto-reconnects internally; None means SSE is
        // unavailable and we fall back to the timer alone.
        let mut sse = self.client.subscribe().await.ok();

        loop {
            let mut sse_triggered = false;
            match sse.as_mut() {
                Some(stream) => {
                    tokio::select! {
                        _ = ticker.tick() => {}
                        change = stream.next_change() => match change {
                            Some(()) => sse_triggered = true,
                            None => sse = None, // stream ended fatally → timer-only
                        }
                    }
                }
                None => {
                    ticker.tick().await;
                }
            }

            // Gate chain.
            if self.client.session_info().is_none() || self.client.is_token_expired() {
                observer.on_status(SyncState::LoggedOut);
                continue;
            }
            if !self.client.has_master_key() {
                continue;
            }
            if !self.auto_sync_enabled().await.unwrap_or(false) {
                continue;
            }
            if backoff_until.is_some_and(|u| Instant::now() < u) {
                continue;
            }
            if !sse_triggered && !self.timer_due().await.unwrap_or(true) {
                continue;
            }

            observer.on_status(SyncState::Syncing);
            match self.sync_now().await {
                Ok(outcome) => {
                    failures = 0;
                    backoff_until = None;
                    observer.on_status(SyncState::Idle);
                    if outcome.pulled > 0 {
                        observer.on_changes_applied(&outcome.changed_tables);
                    }
                }
                Err(e) if is_subscription_required(&e) => {
                    observer.on_subscription_required();
                    backoff_until = Some(Instant::now() + Duration::from_hours(1));
                }
                Err(e) if is_auth_lost(&e) => {
                    observer.on_status(SyncState::LoggedOut);
                    backoff_until = Some(Instant::now() + backoff_delay(failures));
                }
                Err(e) => {
                    failures = failures.saturating_add(1);
                    observer.on_status(SyncState::Error(e.to_string()));
                    backoff_until = Some(Instant::now() + backoff_delay(failures));
                }
            }
        }
    }

    /// Clear the in-memory session and master key. Does not touch the OS keychain
    /// or the local database (a later re-auth reuses the stored `device_id`).
    pub fn disconnect(&self) {
        self.client.clear_session();
    }

    async fn auto_sync_enabled(&self) -> Result<bool> {
        self.blocking(|c| Ok(get_sync_state_or(c, "auto_sync_enabled", "1")? == "1"))
            .await
    }

    async fn timer_due(&self) -> Result<bool> {
        self.blocking(|c| {
            let interval: i64 = get_sync_state_or(c, "sync_interval_minutes", "15")?
                .parse()
                .unwrap_or(15);
            let last = get_sync_state(c, "last_sync_at")?
                .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
                .map(|d| d.with_timezone(&Utc));
            Ok(interval_elapsed(last, interval, Utc::now()))
        })
        .await
    }
}

#[cfg(test)]
mod tests {
    use super::super::blob::BlobTransport;
    use super::super::schema::SyncTable;
    use super::*;
    use crate::client::BlobUploadOutcome;
    use crate::ids::{AppId, DeviceId, UserId};
    use crate::types::{ChangeEntry, Device, PulledChange};
    use std::collections::HashMap;
    use std::future::Future;
    use std::path::PathBuf;
    use std::sync::Mutex;

    /// A combined fake implementing both transports over a shared changelog log
    /// and blob store, tagged with this device's registered id.
    type GroupLog = Arc<Mutex<Vec<(crate::ids::GroupId, DeviceId, ChangeEntry)>>>;

    #[derive(Clone)]
    struct Fake {
        device_id: DeviceId,
        log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>,
        blobs: Arc<Mutex<HashMap<String, Vec<u8>>>>,
        /// Groups this member belongs to, reported by `list_group_scopes`.
        groups: Vec<(crate::ids::GroupId, i32)>,
        /// Shared group changelog (plaintext, the fake bypasses encryption, which
        /// is covered by the `client::groups` tests).
        group_log: GroupLog,
    }
    impl Fake {
        fn new(device_id: u128, log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>) -> Self {
            Self {
                device_id: DeviceId::new(uuid::Uuid::from_u128(device_id)),
                log,
                blobs: Arc::new(Mutex::new(HashMap::new())),
                groups: Vec::new(),
                group_log: Arc::new(Mutex::new(Vec::new())),
            }
        }

        /// Enroll this member in a group sharing `group_log`.
        fn in_group(mut self, id: crate::ids::GroupId, version: i32, group_log: GroupLog) -> Self {
            self.groups.push((id, version));
            self.group_log = group_log;
            self
        }
    }

    impl SyncTransport for Fake {
        fn register_device(
            &self,
            _name: &str,
            _platform: &str,
        ) -> impl Future<Output = Result<Device>> + Send {
            let id = self.device_id;
            async move {
                Ok(Device {
                    id,
                    app_id: AppId::nil(),
                    user_id: 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, (d, e))| PulledChange {
                        entry: e.clone(),
                        device_id: *d,
                        seq: i as i64 + 1,
                    })
                    .collect();
                Ok((out, l.len() as i64, false))
            }
        }

        fn list_group_scopes(
            &self,
        ) -> impl Future<Output = Result<Vec<(crate::ids::GroupId, i32)>>> + Send {
            let groups = self.groups.clone();
            async move { Ok(groups) }
        }

        fn group_scope_push(
            &self,
            group_id: crate::ids::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)
            }
        }

        fn group_scope_pull(
            &self,
            group_id: crate::ids::GroupId,
            _gck_version: i32,
            _device_id: DeviceId,
            cursor: i64,
        ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
            let group_log = self.group_log.clone();
            async move {
                let l = group_log.lock().unwrap();
                // Filter to this group, then page by global position (mirrors the
                // personal fake's seq scheme).
                let out: Vec<PulledChange> = l
                    .iter()
                    .enumerate()
                    .filter(|(i, (gid, _, _))| *gid == group_id && (*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))
            }
        }
    }

    impl BlobTransport for Fake {
        async fn upload_file(
            &self,
            hash: &str,
            path: &std::path::Path,
        ) -> Result<BlobUploadOutcome> {
            if self.blobs.lock().unwrap().contains_key(hash) {
                return Ok(BlobUploadOutcome::AlreadyPresent);
            }
            let data = std::fs::read(path)
                .map_err(|e| crate::error::SyncKitError::Internal(e.to_string()))?;
            self.blobs.lock().unwrap().insert(hash.to_string(), data);
            Ok(BlobUploadOutcome::Uploaded)
        }
        async fn confirm(&self, _hash: &str, _size: u64) -> Result<()> {
            Ok(())
        }
        async fn download_url(&self, hash: &str) -> Result<String> {
            Ok(format!("fake://{hash}"))
        }
        async fn download(&self, hash: &str, _url: &str) -> Result<Vec<u8>> {
            self.blobs
                .lock()
                .unwrap()
                .get(hash)
                .cloned()
                .ok_or_else(|| SyncKitError::Internal("no blob".into()))
        }
    }

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

    fn tempdir() -> PathBuf {
        use std::sync::atomic::{AtomicU64, Ordering};
        static N: AtomicU64 = AtomicU64::new(0);
        let mut p = std::env::temp_dir();
        p.push(format!(
            "synckit_b7_{}_{}",
            std::process::id(),
            N.fetch_add(1, Ordering::Relaxed)
        ));
        std::fs::create_dir_all(&p).unwrap();
        p
    }

    fn make_db(path: &std::path::Path) -> DbSource {
        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
    }

    fn store(db: DbSource, fake: Fake) -> SyncStore<Fake> {
        SyncStore::builder(db, fake, schema())
            .device("dev", "test")
            .build()
    }

    // ── Group-scope sync ──

    fn group_schema() -> SyncSchema {
        SyncSchema::new(vec![
            SyncTable::full("task", &["id", "name"]).group_scoped("group_id"),
        ])
    }

    fn group_make_db(path: &std::path::Path) -> DbSource {
        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
    }

    fn group_store(db: DbSource, fake: Fake) -> SyncStore<Fake> {
        SyncStore::builder(db, fake, group_schema())
            .device("dev", "test")
            .build()
    }

    #[tokio::test]
    async fn two_members_converge_through_a_group_scope() {
        let dir = tempdir();
        let gid = crate::ids::GroupId::new(uuid::Uuid::from_u128(0x9));
        let group_log = Arc::new(Mutex::new(Vec::new()));
        let plog = Arc::new(Mutex::new(Vec::new()));

        let da = group_make_db(&dir.join("a.db"));
        let dbb = group_make_db(&dir.join("b.db"));
        let sa = group_store(
            da.clone(),
            Fake::new(0xA, plog.clone()).in_group(gid, 1, group_log.clone()),
        );
        let sb = group_store(
            dbb.clone(),
            Fake::new(0xB, plog.clone()).in_group(gid, 1, group_log.clone()),
        );

        // Member A creates a task in the group.
        da.open()
            .unwrap()
            .execute(
                "INSERT INTO task (id, name, group_id) VALUES ('t', 'from-A', ?1)",
                [gid.to_string()],
            )
            .unwrap();

        sa.sync_now().await.unwrap(); // pushes the group task to the shared log
        sb.sync_now().await.unwrap(); // pulls it and applies, stamping group_id

        // Member B now has the task, tagged with the group it came from (stamped
        // from the channel, not the payload, the payload carried only id + name).
        let (name, group_id): (String, String) = dbb
            .open()
            .unwrap()
            .query_row("SELECT name, group_id FROM task WHERE id = 't'", [], |r| {
                Ok((r.get(0)?, r.get(1)?))
            })
            .unwrap();
        assert_eq!(name, "from-A");
        assert_eq!(group_id, gid.to_string());

        // The group write never touched B's personal changelog scope.
        let personal_rows: i64 = dbb
            .open()
            .unwrap()
            .query_row(
                "SELECT COUNT(*) FROM sync_changelog WHERE scope = ''",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(personal_rows, 0);
    }

    /// Pre-stamp a device's pending edit at a controlled time so cross-device HLC
    /// ordering is deterministic (sync_now's internal stamp is then a no-op).
    fn stamp_pending(db: &DbSource, node: DeviceId, now_ms: i64) {
        super::super::hlc::stamp_pending(&db.open().unwrap(), node, now_ms).unwrap();
    }

    #[tokio::test]
    async fn sync_now_registers_snapshots_pushes_and_finalizes() {
        let dir = tempdir();
        let db = make_db(&dir.join("a.db"));
        // Pre-existing row (trigger fired), plus clear changelog to test snapshot.
        {
            let c = db.open().unwrap();
            c.execute("INSERT INTO note (id, name) VALUES ('r', 'v')", [])
                .unwrap();
            c.execute("DELETE FROM sync_changelog", []).unwrap();
        }
        let log = Arc::new(Mutex::new(Vec::new()));
        let s = store(db.clone(), Fake::new(0xA, log.clone()));

        let out = s.sync_now().await.unwrap();
        assert_eq!(out.pushed, 1, "the snapshot row is pushed");
        // device_id registered, last_sync stamped, snapshot marked done.
        let c = db.open().unwrap();
        assert!(get_sync_state(&c, "device_id").unwrap().is_some());
        assert_eq!(
            get_sync_state_or(&c, "initial_snapshot_done", "0").unwrap(),
            "1"
        );
        assert!(
            !get_sync_state_or(&c, "last_sync_at", "")
                .unwrap()
                .is_empty()
        );
        // The server received the snapshot entry.
        assert_eq!(log.lock().unwrap().len(), 1);
    }

    #[tokio::test]
    async fn two_devices_converge_through_sync_now() {
        let dir = tempdir();
        let log = Arc::new(Mutex::new(Vec::new()));
        let da = make_db(&dir.join("a.db"));
        let db = make_db(&dir.join("b.db"));

        // A edits at t=100, B edits the same row at t=200 (B wins).
        da.open()
            .unwrap()
            .execute("INSERT INTO note (id,name) VALUES ('r','from-A')", [])
            .unwrap();
        stamp_pending(&da, DeviceId::new(uuid::Uuid::from_u128(0xA)), 100);
        db.open()
            .unwrap()
            .execute("INSERT INTO note (id,name) VALUES ('r','from-B')", [])
            .unwrap();
        stamp_pending(&db, DeviceId::new(uuid::Uuid::from_u128(0xB)), 200);

        let sa = store(da.clone(), Fake::new(0xA, log.clone()));
        let sb = store(db.clone(), Fake::new(0xB, log.clone()));

        // Push order then pull order.
        sa.sync_now().await.unwrap();
        sb.sync_now().await.unwrap();
        sa.sync_now().await.unwrap();
        sb.sync_now().await.unwrap();

        let name_a: String = da
            .open()
            .unwrap()
            .query_row("SELECT name FROM note WHERE id='r'", [], |r| r.get(0))
            .unwrap();
        let name_b: String = db
            .open()
            .unwrap()
            .query_row("SELECT name FROM note WHERE id='r'", [], |r| r.get(0))
            .unwrap();
        assert_eq!(name_a, "from-B");
        assert_eq!(name_b, "from-B");
    }

    #[tokio::test]
    async fn builder_defaults_and_overrides() {
        let dir = tempdir();
        let db = make_db(&dir.join("a.db"));
        let log = Arc::new(Mutex::new(Vec::new()));
        let s = SyncStore::builder(db, Fake::new(0xA, log), schema())
            .device("My Mac", "macos")
            .config(SyncConfig {
                check_interval: Duration::from_secs(30),
                retention_cap: Some(1000),
            })
            .build();
        assert_eq!(s.device_name(), "My Mac");
        assert_eq!(s.platform, "macos");
        assert_eq!(s.config.retention_cap, Some(1000));
    }

    #[tokio::test]
    async fn pending_changes_counts_unpushed() {
        let dir = tempdir();
        let db = make_db(&dir.join("a.db"));
        db.open()
            .unwrap()
            .execute("INSERT INTO note (id,name) VALUES ('r','v')", [])
            .unwrap();
        let log = Arc::new(Mutex::new(Vec::new()));
        let s = store(db, Fake::new(0xA, log));
        assert_eq!(s.pending_changes().await.unwrap(), 1);
    }
}