//! SyncKit change log: the push/pull sync protocol core, append a device's
//! changes and pull the changes since a cursor (optionally filtered), the basis
//! of the multi-device convergence protocol.

use serde_json::Value as JsonValue;
use sqlx::PgPool;

use crate::db::models::{DbSyncGroupLogEntry, DbSyncLogEntry};
use crate::db::{SyncAppId, SyncDeviceId, SyncGroupId, UserId};
use crate::error::Result;

// ── Sync Log ──

/// Push a batch of changes to the sync log. Returns the highest seq assigned.
///
/// `batch_id` is a client-generated UUID for idempotent push. If a batch with
/// the same ID has already been committed for this app+user, the existing max
/// seq is returned without re-inserting (at-most-once semantics).
///
/// All changes are inserted within a single transaction for atomicity and
/// performance (one fsync instead of N).
#[allow(clippy::type_complexity)]
#[tracing::instrument(skip_all)]
pub async fn push_sync_changes(
    pool: &PgPool,
    app_id: SyncAppId,
    user_id: UserId,
    device_id: SyncDeviceId,
    batch_id: uuid::Uuid,
    changes: &[(
        String,
        String,
        String,
        chrono::DateTime<chrono::Utc>,
        Option<JsonValue>,
    )],
) -> Result<i64> {
    if changes.is_empty() {
        return Ok(0);
    }

    // Decompose into parallel Vecs for UNNEST-based batch INSERT
    let mut table_names: Vec<String> = Vec::with_capacity(changes.len());
    let mut operations: Vec<String> = Vec::with_capacity(changes.len());
    let mut row_ids: Vec<String> = Vec::with_capacity(changes.len());
    let mut client_timestamps: Vec<chrono::DateTime<chrono::Utc>> =
        Vec::with_capacity(changes.len());
    let mut data_values: Vec<JsonValue> = Vec::with_capacity(changes.len());

    for (table_name, operation, row_id, client_timestamp, data) in changes {
        table_names.push(table_name.clone());
        operations.push(operation.clone());
        row_ids.push(row_id.clone());
        client_timestamps.push(*client_timestamp);
        data_values.push(data.clone().unwrap_or(JsonValue::Null));
    }

    let mut tx = pool.begin().await?;

    // Serialize concurrent pushes of the SAME batch. A batch legitimately
    // inserts many sync_log rows sharing one batch_id, so migration 086 had to
    // drop the unique index that used to backstop idempotency, which left the
    // MAX(seq) check-then-insert below open to a TOCTOU: two concurrent
    // redeliveries of the same batch both read MAX = NULL and both insert,
    // duplicating the client's changes. This per-(app,user,batch) transaction
    // advisory lock makes the check-and-insert atomic; the loser blocks, then
    // sees the winner's rows and returns the existing cursor. Same
    // `hashtextextended('ns:' || ...)` pattern as the Stripe webhook lock.
    sqlx::query(
        "SELECT pg_advisory_xact_lock(hashtextextended('synckit_push:' || $1::text || ':' || $2::text || ':' || $3::text, 0))",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(batch_id)
    .execute(&mut *tx)
    .await?;

    // Idempotency check inside the transaction, race-free under the lock above.
    let existing: (Option<i64>,) = sqlx::query_as(
        "SELECT MAX(seq) FROM sync_log WHERE app_id = $1 AND user_id = $2 AND batch_id = $3",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(batch_id)
    .fetch_one(&mut *tx)
    .await?;

    if let Some(max_seq) = existing.0 {
        tracing::debug!(batch_id = %batch_id, cursor = max_seq, "Push batch already committed, returning existing cursor");
        tx.rollback().await.ok();
        return Ok(max_seq);
    }

    // Read the current key_id from sync_keys so pushed entries are stamped
    // with the active encryption key. Falls back to NULL if no key exists yet
    // (pre-encryption setup, entries have no encrypted data anyway).
    let current_key_id: Option<i32> =
        sqlx::query_scalar("SELECT key_id FROM sync_keys WHERE app_id = $1 AND user_id = $2")
            .bind(app_id)
            .bind(user_id)
            .fetch_optional(&mut *tx)
            .await?;

    let seqs: Vec<i64> = sqlx::query_scalar(
        r"
        INSERT INTO sync_log (app_id, user_id, device_id, batch_id, table_name, operation, row_id, client_timestamp, data, key_id)
        SELECT $1, $2, $3, $4, t.*, $10
        FROM UNNEST($5::text[], $6::text[], $7::text[], $8::timestamptz[], $9::jsonb[]) AS t
        RETURNING seq
        ",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(device_id)
    .bind(batch_id)
    .bind(&table_names)
    .bind(&operations)
    .bind(&row_ids)
    .bind(&client_timestamps)
    .bind(&data_values)
    .bind(current_key_id)
    .fetch_all(&mut *tx)
    .await?;

    tx.commit().await?;

    let max_seq = seqs.iter().copied().max().unwrap_or(0);

    Ok(max_seq)
}

/// Push a batch of changes to a **group's** shared changelog (`sync_group_log`).
/// Returns the highest seq assigned. The group scope (not the pushing user) owns
/// the entries, so every member's pull sees them.
///
/// Mirrors [`push_sync_changes`] but writes the separate `sync_group_log` table
/// and scopes idempotency by `(app_id, group_id, batch_id)`. Group entries carry
/// no `key_id`: their key generation is the group's `gck_version`, not the
/// per-user `sync_keys.key_id`. The dedicated table is what keeps group rows out
/// of every personal-scope query with no `group_id IS NULL` guard to remember.
/// Membership is enforced by the caller.
#[allow(clippy::type_complexity)]
#[tracing::instrument(skip_all)]
pub async fn push_group_changes(
    pool: &PgPool,
    app_id: SyncAppId,
    group_id: SyncGroupId,
    user_id: UserId,
    device_id: SyncDeviceId,
    batch_id: uuid::Uuid,
    changes: &[(
        String,
        String,
        String,
        chrono::DateTime<chrono::Utc>,
        Option<JsonValue>,
    )],
) -> Result<i64> {
    if changes.is_empty() {
        return Ok(0);
    }

    let mut table_names: Vec<String> = Vec::with_capacity(changes.len());
    let mut operations: Vec<String> = Vec::with_capacity(changes.len());
    let mut row_ids: Vec<String> = Vec::with_capacity(changes.len());
    let mut client_timestamps: Vec<chrono::DateTime<chrono::Utc>> =
        Vec::with_capacity(changes.len());
    let mut data_values: Vec<JsonValue> = Vec::with_capacity(changes.len());

    for (table_name, operation, row_id, client_timestamp, data) in changes {
        table_names.push(table_name.clone());
        operations.push(operation.clone());
        row_ids.push(row_id.clone());
        client_timestamps.push(*client_timestamp);
        data_values.push(data.clone().unwrap_or(JsonValue::Null));
    }

    let mut tx = pool.begin().await?;

    // Serialize concurrent redeliveries of the same batch within this group, so
    // the MAX(seq) idempotency check below is race-free (same pattern as the
    // per-user push lock, keyed by group instead of user).
    sqlx::query(
        "SELECT pg_advisory_xact_lock(hashtextextended('synckit_group_push:' || $1::text || ':' || $2::text, 0))",
    )
    .bind(group_id)
    .bind(batch_id)
    .execute(&mut *tx)
    .await?;

    let existing: (Option<i64>,) = sqlx::query_as(
        "SELECT MAX(seq) FROM sync_group_log WHERE app_id = $1 AND group_id = $2 AND batch_id = $3",
    )
    .bind(app_id)
    .bind(group_id)
    .bind(batch_id)
    .fetch_one(&mut *tx)
    .await?;

    if let Some(max_seq) = existing.0 {
        tx.rollback().await.ok();
        return Ok(max_seq);
    }

    let seqs: Vec<i64> = sqlx::query_scalar(
        r"
        INSERT INTO sync_group_log (app_id, user_id, device_id, group_id, batch_id, table_name, operation, row_id, client_timestamp, data)
        SELECT $1, $2, $3, $4, $5, t.*
        FROM UNNEST($6::text[], $7::text[], $8::text[], $9::timestamptz[], $10::jsonb[]) AS t
        RETURNING seq
        ",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(device_id)
    .bind(group_id)
    .bind(batch_id)
    .bind(&table_names)
    .bind(&operations)
    .bind(&row_ids)
    .bind(&client_timestamps)
    .bind(&data_values)
    .fetch_all(&mut *tx)
    .await?;

    tx.commit().await?;

    Ok(seqs.iter().copied().max().unwrap_or(0))
}

/// Pull a group's changes since a cursor from `sync_group_log`, with the same
/// optional table/timestamp filters as [`pull_sync_changes_filtered`], scoped by
/// `(app_id, group_id)`. Membership is enforced by the caller.
#[tracing::instrument(skip_all)]
pub async fn pull_group_changes_filtered(
    pool: &PgPool,
    app_id: SyncAppId,
    group_id: SyncGroupId,
    cursor: i64,
    limit: i64,
    tables: Option<&[String]>,
    since: Option<chrono::DateTime<chrono::Utc>>,
) -> Result<Vec<DbSyncGroupLogEntry>> {
    let entries = sqlx::query_as::<_, DbSyncGroupLogEntry>(
        r"
        SELECT * FROM sync_group_log
        WHERE app_id = $1 AND group_id = $2 AND seq > $3
          AND ($5::text[] IS NULL OR table_name = ANY($5))
          AND ($6::timestamptz IS NULL OR client_timestamp >= $6)
        ORDER BY seq ASC
        LIMIT $4
        ",
    )
    .bind(app_id)
    .bind(group_id)
    .bind(cursor)
    .bind(limit)
    .bind(tables)
    .bind(since)
    .fetch_all(pool)
    .await?;

    Ok(entries)
}

/// Pull changes since a cursor for a user within an app.
///
/// Prefer `pull_sync_changes_filtered` for new code; it supports optional
/// table and timestamp filters. This function is kept for backward compatibility.
#[allow(dead_code)]
#[tracing::instrument(skip_all)]
pub async fn pull_sync_changes(
    pool: &PgPool,
    app_id: SyncAppId,
    user_id: UserId,
    cursor: i64,
    limit: i64,
) -> Result<Vec<DbSyncLogEntry>> {
    let entries = sqlx::query_as::<_, DbSyncLogEntry>(
        r"
        SELECT * FROM sync_log
        WHERE app_id = $1 AND user_id = $2 AND seq > $3
        ORDER BY seq ASC
        LIMIT $4
        ",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(cursor)
    .bind(limit)
    .fetch_all(pool)
    .await?;

    Ok(entries)
}

/// Pull changes since a cursor with optional table and timestamp filters.
///
/// When `tables` is `Some`, only entries whose `table_name` is in the list are returned.
/// When `since` is `Some`, only entries with `client_timestamp >= since` are returned.
/// Both filters compose (AND). Passing `None` for both is identical to `pull_sync_changes`.
#[tracing::instrument(skip_all)]
pub async fn pull_sync_changes_filtered(
    pool: &PgPool,
    app_id: SyncAppId,
    user_id: UserId,
    cursor: i64,
    limit: i64,
    tables: Option<&[String]>,
    since: Option<chrono::DateTime<chrono::Utc>>,
) -> Result<Vec<DbSyncLogEntry>> {
    let entries = sqlx::query_as::<_, DbSyncLogEntry>(
        r"
        SELECT * FROM sync_log
        WHERE app_id = $1 AND user_id = $2 AND seq > $3
          AND ($5::text[] IS NULL OR table_name = ANY($5))
          AND ($6::timestamptz IS NULL OR client_timestamp >= $6)
        ORDER BY seq ASC
        LIMIT $4
        ",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(cursor)
    .bind(limit)
    .bind(tables)
    .bind(since)
    .fetch_all(pool)
    .await?;

    Ok(entries)
}