//! The `SyncStore` integration (Groups p4 / GO -> SyncStore, M2).
//!
//! GoingsOn hand-rolled its cloud sync in `sync_service/` against the raw
//! `SyncKitClient`. To gain group sync (and shed the glue), it migrates to the
//! engine's `SyncStore`, which owns the changelog, triggers, HLC, apply, blobs,
//! and scheduler. This module assembles that store from GO's pieces:
//! - the [`goingson_schema`] manifest (declared in M1),
//! - [`AttachmentBlobs`], GO's blob policy,
//! - [`GoSyncObserver`], which forwards engine events to the frontend,
//! - and [`run_cutover`], the one-time local migration from GO's bookkeeping to
//!   the engine's.
//!
//! The engine opens its own rusqlite connection (WAL) to the same `goingson.db`
//! file the sqlx pool uses; the two coexist by design.

mod blobs;
pub(crate) mod manifest;
mod observer;
pub(crate) mod sync_state;

use std::path::{Path, PathBuf};
use std::sync::Arc;

use rusqlite::{Connection, OptionalExtension};
use synckit_client::store::install_policy;
use synckit_client::{DbSource, SyncKitClient, SyncStore};
use tauri::{AppHandle, Manager};

use crate::config_key::CONFIG;

use crate::state::AppState;
use blobs::AttachmentBlobs;
use manifest::goingson_schema;
use observer::GoSyncObserver;

/// Build GoingsOn's `SyncStore` over the app's SQLite file. The device name and
/// platform default to the host's (the engine's default), matching GO's prior
/// registration behaviour.
pub(crate) fn build_store(
    db_path: PathBuf,
    data_dir: PathBuf,
    client: SyncKitClient,
) -> Arc<SyncStore<SyncKitClient>> {
    let store = SyncStore::builder(DbSource::path(db_path), client, goingson_schema())
        .blob_policy(Arc::new(AttachmentBlobs::new(data_dir)))
        .build();
    Arc::new(store)
}

/// Spawn the background cloud-sync scheduler for the managed `AppState`'s
/// `SyncStore`, if one is configured. The engine's loop races a timer against the
/// server's SSE change stream, applies the auth / key-loaded / auto-sync-enabled /
/// backoff gate chain, and reports status through [`GoSyncObserver`].
///
/// Must be called from within the async runtime (the engine spawns the loop with
/// `tokio::spawn`). The returned handle is intentionally dropped: the loop runs
/// for the app's lifetime (dropping the handle does not stop it), which matches
/// GoingsOn's prior always-on scheduler. A no-op when sync is unconfigured.
pub fn spawn_scheduler(app: &AppHandle) {
    let Some(state) = app.try_state::<Arc<AppState>>() else {
        return;
    };
    let Some(store) = state.sync_store.clone() else {
        return;
    };
    let _handle = store.spawn_scheduler(Arc::new(GoSyncObserver::new(app.clone())));
}

/// One-time migration of a device's local sync bookkeeping from GoingsOn's
/// hand-rolled triggers/changelog to the engine's. Idempotent and guarded by the
/// `syncstore_migrated` flag, so it is a cheap no-op on every launch after the
/// first. Runs synchronously (rusqlite) at app init, after the sqlx migrations
/// and before the scheduler spawns.
///
/// Steps:
/// 1. Open a configured connection; this runs the engine's `ensure_scope_schema`,
///    which adds `sync_changelog.scope` to a legacy changelog and seeds
///    `sync_scope_cursor('')` from the old single `pull_cursor`.
/// 2. Drop GO's legacy capture triggers (two naming conventions; see
///    [`legacy_sync_triggers`]).
/// 3. Run the engine's generated DDL: bookkeeping tables (`CREATE IF NOT EXISTS`)
///    and one trigger set per table (each with its own `DROP IF EXISTS`).
/// 4. Re-baseline. The server changelog is authoritative, so reset the personal
///    pull cursor to 0 (re-pull and re-apply idempotently) and clear
///    `initial_snapshot_done` (re-snapshot every local row for the first push, so
///    any local-only data survives the cutover). This is the one-time cost the
///    migration plan accepts.
///
/// The whole rewrite runs in one transaction; a failure leaves the legacy
/// triggers in place and the `syncstore_migrated` flag unset, so the next launch
/// retries from a clean state.
pub(crate) fn run_cutover(db_path: &Path) -> Result<(), String> {
    let conn = DbSource::path(db_path).open().map_err(str_err)?;

    if flag_set(&conn, "syncstore_migrated")? {
        return Ok(());
    }

    let legacy = legacy_sync_triggers(&conn)?;

    let tx = conn.unchecked_transaction().map_err(str_err)?;
    for name in &legacy {
        tx.execute_batch(&format!("DROP TRIGGER IF EXISTS \"{name}\";"))
            .map_err(str_err)?;
    }
    // The manifest is group-scoped (M3): its generated triggers read
    // NEW.group_id. The column comes from migration 059 (which runs before this
    // at app init), so it is guaranteed present here.
    tx.execute_batch(&goingson_schema().migration_sql())
        .map_err(str_err)?;
    tx.execute_batch(
        "UPDATE sync_scope_cursor SET cursor = 0 WHERE scope = '';\n\
         UPDATE sync_state SET value = '0' WHERE key = 'pull_cursor';\n\
         UPDATE sync_state SET value = '0' WHERE key = 'initial_snapshot_done';\n\
         INSERT INTO sync_state (key, value) VALUES ('syncstore_migrated', '1')\n\
             ON CONFLICT(key) DO UPDATE SET value = '1';\n\
         INSERT INTO sync_state (key, value) VALUES ('syncstore_groups_migrated', '1')\n\
             ON CONFLICT(key) DO UPDATE SET value = '1';",
    )
    .map_err(str_err)?;
    tx.commit().map_err(str_err)?;
    Ok(())
}

/// M3 group migration for a device that already ran the M2 cutover: regenerate
/// the engine triggers so the shareable tables become group-scoped (they stamp
/// the changelog scope from `group_id`). Idempotent, guarded by
/// `syncstore_groups_migrated`, transactional. The `group_id` columns come from
/// migration 059; a fresh install already generated group-scoped triggers inside
/// [`run_cutover`] and set the flag, so this is a no-op there; only an
/// M2-migrated device (personal-only triggers) reaches the body.
pub(crate) fn run_group_migration(db_path: &Path) -> Result<(), String> {
    let conn = DbSource::path(db_path).open().map_err(str_err)?;
    if flag_set(&conn, "syncstore_groups_migrated")? {
        return Ok(());
    }
    let tx = conn.unchecked_transaction().map_err(str_err)?;
    // Regenerate every trigger; the shareable tables' triggers now stamp the
    // changelog scope from group_id. Each generated trigger has its own DROP.
    tx.execute_batch(&goingson_schema().migration_sql())
        .map_err(str_err)?;
    tx.execute_batch(
        "INSERT INTO sync_state (key, value) VALUES ('syncstore_groups_migrated', '1')\n\
             ON CONFLICT(key) DO UPDATE SET value = '1';",
    )
    .map_err(str_err)?;
    tx.commit().map_err(str_err)?;
    Ok(())
}

/// Seed `config_key_policy` from the config spec ([`CONFIG`]).
///
/// Runs every launch and is idempotent ([`install_policy`] clears and reinserts
/// the closed set), so the spec stays the single source of truth for which
/// `user_config` keys sync. **Must run before any
/// `goingson_schema().migration_sql()`** — the cutover, the group migration, the
/// config migration, and the store build all regenerate the engine's triggers,
/// and the generated config triggers join this table, so it has to exist and be
/// seeded first.
pub(crate) fn seed_config_policy(db_path: &Path) -> Result<(), String> {
    let mut conn = DbSource::path(db_path).open().map_err(str_err)?;
    install_policy(&mut conn, &CONFIG).map_err(str_err)?;
    Ok(())
}

/// One-time: regenerate the engine triggers so the `user_config` config table
/// gains its export/import triggers on a device that already ran the M2/M3
/// cutover (both are guarded and no-op now, so neither would create them).
/// Idempotent, guarded by `syncstore_config_migrated`, transactional. A fresh
/// install already generated the config triggers inside [`run_cutover`] (the
/// manifest now carries the config table) and never reaches the body.
///
/// [`seed_config_policy`] must have run first this launch, since the regenerated
/// config triggers join `config_key_policy`.
pub(crate) fn run_config_migration(db_path: &Path) -> Result<(), String> {
    let conn = DbSource::path(db_path).open().map_err(str_err)?;
    if flag_set(&conn, "syncstore_config_migrated")? {
        return Ok(());
    }
    let tx = conn.unchecked_transaction().map_err(str_err)?;
    // Each generated trigger has its own DROP IF EXISTS, so regenerating the full
    // set is safe; it adds the config table's triggers to an already-migrated
    // device.
    tx.execute_batch(&goingson_schema().migration_sql())
        .map_err(str_err)?;
    tx.execute_batch(
        "INSERT INTO sync_state (key, value) VALUES ('syncstore_config_migrated', '1')\n\
             ON CONFLICT(key) DO UPDATE SET value = '1';",
    )
    .map_err(str_err)?;
    tx.commit().map_err(str_err)?;
    Ok(())
}

fn flag_set(conn: &Connection, key: &str) -> Result<bool, String> {
    let v: Option<String> = conn
        .query_row("SELECT value FROM sync_state WHERE key = ?1", [key], |r| {
            r.get(0)
        })
        .optional()
        .map_err(str_err)?;
    Ok(v.as_deref() == Some("1"))
}

fn str_err<E: std::fmt::Display>(e: E) -> String {
    e.to_string()
}

/// GO's hand-written capture triggers use two naming conventions: `sync_trg_*`
/// (migration 030 and later) and `*_changelog` (migration 037's `sync_accounts`).
/// Both write to `sync_changelog` and must be dropped before the engine's own
/// triggers replace them. The engine's generated triggers are named
/// `sync_<table>_<op>` (no `sync_trg_` prefix, no `_changelog` suffix), so this
/// query never matches them. The FTS triggers on `emails` (`emails_ai/ad/au`)
/// also never match, so email full-text search is untouched.
fn legacy_sync_triggers(conn: &Connection) -> Result<Vec<String>, String> {
    let mut stmt = conn
        .prepare(
            "SELECT name FROM sqlite_master WHERE type = 'trigger' \
             AND (name LIKE 'sync\\_trg\\_%' ESCAPE '\\' \
                  OR name LIKE '%\\_changelog' ESCAPE '\\')",
        )
        .map_err(str_err)?;
    let names = stmt
        .query_map([], |r| r.get::<_, String>(0))
        .map_err(str_err)?;
    names.collect::<rusqlite::Result<Vec<_>>>().map_err(str_err)
}