//! Pull remote changes and apply them to the local database.

use chrono::Utc;
use goingson_core::CoreError;
use sqlx::{SqliteConnection, SqlitePool};
use synckit_client::{
    ChangeEntry, ChangeOp, DeviceId, Hlc, Resolution, SyncKitClient,
    detect_conflicts, resolve_lww,
};
use tracing::{debug, warn};
use uuid::Uuid;

use super::state::{get_sync_state, set_sync_state};
use super::hlc::{assign_pending_hlcs, load_committed_hlcs, observe_remote, record_committed_hlc};
use super::{UPSERT_ORDER, DELETE_ORDER};
use super::apply::{apply_upsert, apply_delete};

#[tracing::instrument(skip_all)]
pub async fn pull_changes(
    pool: &SqlitePool,
    client: &SyncKitClient,
    device_id: Uuid,
) -> Result<PullOutcome, CoreError> {
    let cursor_str = get_sync_state(pool, "pull_cursor").await?;
    let mut cursor: i64 = cursor_str.parse().unwrap_or(0);
    let mut total_applied: i64 = 0;
    let mut changed_tables: std::collections::HashSet<String> = std::collections::HashSet::new();

    // Read local pending changes once (unpushed changelog entries). This also
    // stamps them with their HLC so conflict resolution can compare clocks.
    let local_pending = read_local_pending(pool, device_id).await?;

    loop {
        let (changes, new_cursor, has_more) = client
            .pull_rich(DeviceId::new(device_id), cursor)
            .await
            .map_err(|e| CoreError::sync(format!("pull failed: {}", e)))?;

        if changes.is_empty() {
            set_sync_state(pool, "pull_cursor", &new_cursor.to_string()).await?;
            break;
        }

        // Advance our clock past everything in this batch, so future local changes
        // causally follow the remote ones. Observing the batch maximum suffices.
        if let Some(max_remote) = changes.iter().map(|p| p.entry.hlc).max() {
            observe_remote(pool, max_remote, device_id).await?;
        }

        // Detect conflicts between remote changes and local pending
        let (clean, conflicts) = detect_conflicts(changes, &local_pending, DeviceId::new(device_id));

        // Apply non-conflicting changes — but HLC-gate them first. "Clean" only
        // means no local *pending* edit contests the row; an older remote edit for
        // an already-committed row also lands here, and applying it blindly would
        // clobber the newer local value. Pre-fetch the rows' committed clocks, then
        // let SyncKit's CleanChanges gate drop anything older-or-equal.
        let clean_count = clean.len() as i64;
        if !clean.is_empty() {
            let keys: Vec<(String, String)> = clean
                .row_keys()
                .map(|(t, r)| (t.to_string(), r.to_string()))
                .collect();
            let committed = load_committed_hlcs(pool, &keys).await?;
            let clean_entries: Vec<ChangeEntry> =
                clean.gated(|t, r| committed.get(&(t.to_string(), r.to_string())).copied());
            if !clean_entries.is_empty() {
                changed_tables.extend(clean_entries.iter().map(|e| e.table.clone()));
                apply_remote_changes(pool, clean_entries).await?;
            }
        }

        // Resolve conflicts with LWW
        let mut resolved_count: i64 = 0;
        if !conflicts.is_empty() {
            let conflict_count = conflicts.len();
            let mut resolved_entries: Vec<ChangeEntry> = Vec::new();

            for pair in &conflicts {
                match resolve_lww(&pair.local, &pair.remote) {
                    Resolution::KeepRemote => {
                        resolved_entries.push(pair.remote.entry.clone());
                    }
                    Resolution::KeepLocal => {
                        // Skip — local version will push on next cycle
                    }
                    Resolution::Merged(data) => {
                        // Apply merged data using the remote entry as template
                        let mut merged = pair.remote.entry.clone();
                        merged.data = Some(data);
                        resolved_entries.push(merged);
                    }
                    Resolution::Skip => {
                        // Skip both
                    }
                }
            }

            resolved_count = resolved_entries.len() as i64;
            if !resolved_entries.is_empty() {
                changed_tables.extend(resolved_entries.iter().map(|e| e.table.clone()));
                apply_remote_changes(pool, resolved_entries).await?;
            }

            debug!(
                "Resolved {} conflicts ({} applied as remote wins)",
                conflict_count,
                conflict_count - conflicts.iter().filter(|p| {
                    matches!(resolve_lww(&p.local, &p.remote), Resolution::KeepLocal | Resolution::Skip)
                }).count(),
            );
        }

        total_applied += clean_count + resolved_count;

        // Save cursor after each batch (crash-safe)
        set_sync_state(pool, "pull_cursor", &new_cursor.to_string()).await?;
        cursor = new_cursor;

        if !has_more {
            break;
        }
    }

    if total_applied > 0 {
        debug!("Pulled and applied {} remote changes", total_applied);
    }
    Ok(PullOutcome { applied: total_applied, changed_tables })
}

/// Result of a pull: how many changes landed and which tables they touched.
///
/// The table set drives selective cache invalidation in the UI so a pull that
/// only touched, say, `tasks` doesn't force the compose screen to re-hydrate
/// every contact.
pub struct PullOutcome {
    pub applied: i64,
    pub changed_tables: std::collections::HashSet<String>,
}

/// Read unpushed changelog entries as ChangeEntry values for conflict detection.
///
/// Stamps any not-yet-stamped rows with their HLC first, so the clocks used here
/// match the ones the push path will send.
/// A pending sync_changelog row: (table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter).
type PendingRow = (String, String, String, String, Option<String>, Option<i64>, Option<i64>);

async fn read_local_pending(pool: &SqlitePool, device_id: Uuid) -> Result<Vec<ChangeEntry>, CoreError> {
    assign_pending_hlcs(pool, device_id).await?;

    let rows: Vec<PendingRow> = sqlx::query_as(
        "SELECT table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter \
         FROM sync_changelog WHERE pushed = 0 ORDER BY id ASC"
    )
    .fetch_all(pool)
    .await
    .map_err(CoreError::database)?;

    let entries: Vec<ChangeEntry> = rows
        .into_iter()
        .filter_map(|(table, op, row_id, timestamp, data, hlc_wall, hlc_counter)| {
            let ts = chrono::DateTime::parse_from_rfc3339(&timestamp)
                .map(|dt| dt.with_timezone(&Utc))
                .unwrap_or(chrono::DateTime::UNIX_EPOCH);

            let change_op = ChangeOp::from_str_opt(&op)?;
            let json_data = data.and_then(|d| serde_json::from_str(&d).ok());

            let hlc = match (hlc_wall, hlc_counter) {
                (Some(wall), Some(counter)) => Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(device_id) },
                _ => Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id)),
            };

            Some(ChangeEntry {
                table,
                op: change_op,
                row_id,
                timestamp: ts,
                hlc,
                data: json_data,
                extra: Default::default(),
            })
        })
        .collect();

    Ok(entries)
}

/// Apply remote changes to local DB with triggers suppressed and FK enforcement off.
///
/// FK enforcement is disabled so that tasks with `source_email_id` pointing to
/// emails not yet fetched locally can be inserted without error. Uses a dedicated
/// connection (same pattern as `crates/db-sqlite/src/migrations.rs`).
///
/// The `applying_remote` flag is set inside a transaction on the dedicated connection.
/// In WAL mode, uncommitted changes are only visible to the writing connection, so
/// other connections (handling user edits) never see the flag and their triggers fire
/// normally. The flag is reset to '0' before commit so it's never globally visible as '1'.
///
/// Uses `detach()` to prevent returning the connection to the pool with FK OFF
/// if pragma restoration fails. The pool will create a fresh connection (with
/// FK ON via connect options) to replace it.
pub(crate) async fn apply_remote_changes(pool: &SqlitePool, changes: Vec<ChangeEntry>) -> Result<(), CoreError> {
    // Acquire a dedicated connection for FK pragma and transaction
    let mut conn = pool.acquire().await.map_err(CoreError::database)?;

    // FK pragma must be set outside a transaction (SQLite requirement)
    sqlx::query("PRAGMA foreign_keys = OFF")
        .execute(&mut *conn)
        .await
        .map_err(CoreError::database)?;

    // Use a transaction so applying_remote is only visible to this connection (WAL isolation).
    // Other connections see the committed value ('0') and their triggers fire normally.
    sqlx::query("BEGIN IMMEDIATE")
        .execute(&mut *conn)
        .await
        .map_err(CoreError::database)?;

    sqlx::query("UPDATE sync_state SET value = '1' WHERE key = 'applying_remote'")
        .execute(&mut *conn)
        .await
        .map_err(CoreError::database)?;

    let result = apply_changes_inner(&mut conn, changes).await;

    // Reset flag before commit so it's never globally visible as '1'
    let _ = sqlx::query("UPDATE sync_state SET value = '0' WHERE key = 'applying_remote'")
        .execute(&mut *conn)
        .await;

    match result {
        Ok(skipped) => {
            if let Err(e) = sqlx::query("COMMIT").execute(&mut *conn).await {
                let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
                let _ = sqlx::query("PRAGMA foreign_keys = ON").execute(&mut *conn).await;
                return Err(CoreError::database(e));
            }
            if skipped > 0 {
                warn!(
                    skipped,
                    "Skipped un-appliable remote change entries; the rest of the batch \
                     was applied and the pull cursor advanced (sync not wedged)."
                );
            }
        }
        Err(ref _e) => {
            let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
        }
    }

    // Always restore FK enforcement. If this fails, detach the connection
    // so it doesn't return to the pool with FK OFF.
    if let Err(e) = sqlx::query("PRAGMA foreign_keys = ON")
        .execute(&mut *conn)
        .await
    {
        conn.detach();
        return Err(CoreError::database(e));
    }

    result.map(|_| ())
}

/// Apply a batch of remote changes within an open transaction.
///
/// Per-entry failures are skipped and logged rather than aborting the whole batch:
/// before, one un-appliable remote row (e.g. a payload missing a NOT NULL column)
/// rolled back the transaction, `apply_remote_changes` returned Err, and the pull
/// cursor never advanced -- so every subsequent pull re-fetched the same poisoned
/// batch and failed identically, wedging sync permanently (ultra-fuzz Run #27
/// Data S-1). A failed statement does not poison the SQLite transaction, so the
/// good rows still commit and the cursor advances. Returns the count of skipped
/// entries (the caller logs a summary); only a catastrophic error returns `Err`.
pub(crate) async fn apply_changes_inner(
    conn: &mut SqliteConnection,
    changes: Vec<ChangeEntry>,
) -> Result<usize, CoreError> {
    // Separate upserts from deletes
    let mut upserts: Vec<&ChangeEntry> = Vec::new();
    let mut deletes: Vec<&ChangeEntry> = Vec::new();

    for change in &changes {
        match change.op {
            ChangeOp::Insert | ChangeOp::Update => upserts.push(change),
            ChangeOp::Delete => deletes.push(change),
        }
    }

    let mut skipped = 0usize;

    // Apply upserts in parent-first FK order
    for table in UPSERT_ORDER {
        for change in &upserts {
            if change.table != *table {
                continue;
            }
            let Some(ref data) = change.data else {
                continue;
            };
            match apply_upsert(&mut *conn, table, &change.row_id, data).await {
                Ok(()) => {
                    record_committed_hlc(&mut *conn, table, &change.row_id, change.hlc).await?;
                }
                Err(e) => {
                    warn!(table = *table, row_id = %change.row_id, error = %e,
                        "Skipping un-appliable remote upsert");
                    skipped += 1;
                }
            }
        }
    }

    // Apply deletes in child-first FK order
    for table in DELETE_ORDER {
        for change in &deletes {
            if change.table != *table {
                continue;
            }
            match apply_delete(&mut *conn, table, &change.row_id).await {
                Ok(()) => {
                    record_committed_hlc(&mut *conn, table, &change.row_id, change.hlc).await?;
                }
                Err(e) => {
                    warn!(table = *table, row_id = %change.row_id, error = %e,
                        "Skipping un-appliable remote delete");
                    skipped += 1;
                }
            }
        }
    }

    Ok(skipped)
}