//! DB-layer contract tests for the SyncKit change log, blobs and keys
//! (`db::synckit::{log, blobs, keys}`), which had none of their own.
//!
//! Its siblings (`db_synckit_rotation`, `db_synckit_groups`,
//! `db_synckit_invitations`) already pin the rotation state machine, the group
//! changelog and the invitation lifecycle at this layer;
//! `db_synckit_accounts_layer` covers devices, apps, subscriptions and the
//! audit log. What was reachable only through the HTTP workflows is the owner
//! scoping every one of these queries carries, the `app_id = $1 AND
//! user_id = $2` pair: a route test authenticates as one user, so it cannot
//! see a query that returns another user's rows.
//!
//! Deliberately not re-asserted here, because it is covered elsewhere: the
//! table and `since` pull filters (`synckit_selective`), the storage quota and
//! blob-delete paths through a real subscription (`synckit_paid_sync`,
//! `synckit_per_key_storage`), and compaction against a freshly registered
//! device (`synckit_paid_sync`).

use super::db_synckit_accounts_layer::seed_active_subscription;
use crate::harness::db::TestDb;
use crate::harness::seed_user;

use makenotwork::db::synckit;
use makenotwork::db::synckit::BlobConfirm;
use makenotwork::db::{SyncAppId, SyncDeviceId, SyncPlatform, UserId};
use uuid::Uuid;

/// Seed a sync app owned by `user`, with its usage row.
async fn seed_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
    synckit::create_sync_app(pool, user, name, &format!("key_{name}_padding"), None, None)
        .await
        .expect("seed sync app")
        .id
}

/// Mark an app first-party, so the end-user subscription model applies.
async fn make_internal(pool: &sqlx::PgPool, app: SyncAppId) {
    sqlx::query("UPDATE sync_apps SET is_internal = true WHERE id = $1")
        .bind(app)
        .execute(pool)
        .await
        .expect("mark internal");
}

async fn seed_device(
    pool: &sqlx::PgPool,
    app: SyncAppId,
    user: UserId,
    name: &str,
) -> SyncDeviceId {
    synckit::upsert_sync_device(pool, app, user, name, SyncPlatform::Macos, None)
        .await
        .expect("seed device")
        .id
}

/// One INSERT change tuple in the shape `push_sync_changes` expects.
fn change(
    table: &str,
    row: &str,
) -> (
    String,
    String,
    String,
    chrono::DateTime<chrono::Utc>,
    Option<serde_json::Value>,
) {
    (
        table.to_string(),
        "INSERT".to_string(),
        row.to_string(),
        chrono::Utc::now(),
        Some(serde_json::json!({ "row": row })),
    )
}

// ── log ─────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn an_append_keeps_every_entry_and_the_order_it_arrived_in() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklog_append").await;
    let app = seed_app(&db.pool, user, "logappend").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    let changes: Vec<_> = (0..5).map(|i| change("tasks", &format!("r{i}"))).collect();
    let cursor = synckit::push_sync_changes(&db.pool, app, user, device, Uuid::new_v4(), &changes)
        .await
        .unwrap();

    let entries = synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
        .await
        .unwrap();
    assert_eq!(entries.len(), 5, "an append drops nothing");
    let rows: Vec<&str> = entries.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(rows, ["r0", "r1", "r2", "r3", "r4"], "and reorders nothing");
    assert!(
        entries.windows(2).all(|w| w[0].seq < w[1].seq),
        "seq is strictly increasing: {:?}",
        entries.iter().map(|e| e.seq).collect::<Vec<_>>()
    );
    assert_eq!(
        cursor,
        entries.last().unwrap().seq,
        "the returned cursor is the highest seq assigned"
    );
}

#[tokio::test]
async fn cursor_paging_returns_each_entry_exactly_once() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklog_page").await;
    let app = seed_app(&db.pool, user, "logpage").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    for i in 0..5 {
        synckit::push_sync_changes(
            &db.pool,
            app,
            user,
            device,
            Uuid::new_v4(),
            &[change("tasks", &format!("r{i}"))],
        )
        .await
        .unwrap();
    }

    // Walk the log two at a time the way a client does, carrying the last seq
    // forward as the next cursor.
    let mut seen: Vec<String> = Vec::new();
    let mut cursor = 0i64;
    loop {
        let page = synckit::pull_sync_changes(&db.pool, app, user, cursor, 2)
            .await
            .unwrap();
        if page.is_empty() {
            break;
        }
        cursor = page.last().unwrap().seq;
        seen.extend(page.into_iter().map(|e| e.row_id));
        assert!(seen.len() <= 5, "paging must terminate: {seen:?}");
    }
    assert_eq!(
        seen,
        ["r0", "r1", "r2", "r3", "r4"],
        "every entry once, in order, with no gap at a page boundary"
    );
}

#[tokio::test]
async fn a_replayed_batch_appends_nothing_and_returns_the_same_cursor() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklog_replay").await;
    let app = seed_app(&db.pool, user, "logreplay").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    let batch = Uuid::new_v4();
    let changes = [change("tasks", "r0"), change("tasks", "r1")];
    let first = synckit::push_sync_changes(&db.pool, app, user, device, batch, &changes)
        .await
        .unwrap();
    // The client never saw the response and retried the same batch id.
    let second = synckit::push_sync_changes(&db.pool, app, user, device, batch, &changes)
        .await
        .unwrap();

    assert_eq!(first, second, "at most once: the same cursor comes back");
    let entries = synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
        .await
        .unwrap();
    assert_eq!(entries.len(), 2, "the retry inserted nothing: {entries:?}");
}

#[tokio::test]
async fn a_pull_never_reaches_another_user_or_another_app() {
    let db = TestDb::new().await;
    let alice = seed_user(&db.pool, "sklog_alice").await;
    let bob = seed_user(&db.pool, "sklog_bob").await;
    let app = seed_app(&db.pool, alice, "logscope").await;
    let other_app = seed_app(&db.pool, alice, "logscope2").await;
    let alice_dev = seed_device(&db.pool, app, alice, "alice-laptop").await;
    let bob_dev = seed_device(&db.pool, app, bob, "bob-laptop").await;
    let alice_other_dev = seed_device(&db.pool, other_app, alice, "alice-phone").await;

    synckit::push_sync_changes(
        &db.pool,
        app,
        alice,
        alice_dev,
        Uuid::new_v4(),
        &[change("tasks", "alice")],
    )
    .await
    .unwrap();
    synckit::push_sync_changes(
        &db.pool,
        app,
        bob,
        bob_dev,
        Uuid::new_v4(),
        &[change("tasks", "bob")],
    )
    .await
    .unwrap();
    synckit::push_sync_changes(
        &db.pool,
        other_app,
        alice,
        alice_other_dev,
        Uuid::new_v4(),
        &[change("tasks", "alice-other-app")],
    )
    .await
    .unwrap();

    let alice_entries = synckit::pull_sync_changes(&db.pool, app, alice, 0, 100)
        .await
        .unwrap();
    let rows: Vec<&str> = alice_entries.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        rows,
        ["alice"],
        "one user's log is one user's, per app: {rows:?}"
    );
}

// ── blobs ───────────────────────────────────────────────────────────────────

#[tokio::test]
async fn a_blob_is_scoped_to_the_user_who_stored_it() {
    let db = TestDb::new().await;
    let alice = seed_user(&db.pool, "skblob_alice").await;
    let bob = seed_user(&db.pool, "skblob_bob").await;
    let app = seed_app(&db.pool, alice, "blobscope").await;
    make_internal(&db.pool, app).await;
    seed_active_subscription(&db.pool, alice, app, "sub_blob_alice", 1_000_000).await;
    seed_active_subscription(&db.pool, bob, app, "sub_blob_bob", 1_000_000).await;

    let stored =
        synckit::confirm_internal_blob(&db.pool, app, alice, "hash-a", 400, "s3/alice", "default")
            .await
            .unwrap();
    assert_eq!(stored, BlobConfirm::Stored);

    assert!(
        synckit::get_sync_blob_by_hash(&db.pool, app, alice, "hash-a")
            .await
            .unwrap()
            .is_some()
    );
    assert!(
        synckit::get_sync_blob_by_hash(&db.pool, app, bob, "hash-a")
            .await
            .unwrap()
            .is_none(),
        "the hash is the same bytes, but it is not bob's row"
    );
    assert_eq!(
        synckit::storage_used_bytes(&db.pool, app, bob)
            .await
            .unwrap(),
        0,
        "and it is not charged to bob's quota"
    );
    assert_eq!(
        synckit::storage_used_bytes(&db.pool, app, alice)
            .await
            .unwrap(),
        400
    );
}

#[tokio::test]
async fn re_confirming_the_same_hash_stores_one_row_and_charges_once() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "skblob_dedup").await;
    let app = seed_app(&db.pool, user, "blobdedup").await;
    make_internal(&db.pool, app).await;
    seed_active_subscription(&db.pool, user, app, "sub_blob_dedup", 1_000).await;

    let first =
        synckit::confirm_internal_blob(&db.pool, app, user, "hash-x", 400, "s3/x", "default")
            .await
            .unwrap();
    let second =
        synckit::confirm_internal_blob(&db.pool, app, user, "hash-x", 400, "s3/x", "default")
            .await
            .unwrap();
    assert_eq!(first, BlobConfirm::Stored);
    assert_eq!(
        second,
        BlobConfirm::AlreadyStored,
        "content-addressed: the same hash is the same object"
    );

    assert_eq!(
        synckit::storage_used_bytes(&db.pool, app, user)
            .await
            .unwrap(),
        400,
        "a re-upload must not count twice, or a retry would eat the cap"
    );
    let rows: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = 'hash-x'",
    )
    .bind(app)
    .bind(user)
    .fetch_one(&db.pool)
    .await
    .unwrap();
    assert_eq!(rows, 1);
}

// ── keys ────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn a_key_upsert_takes_only_the_version_the_caller_expected() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "skkey_occ").await;
    let app = seed_app(&db.pool, user, "keyocc").await;

    assert!(
        synckit::upsert_sync_key(&db.pool, app, user, "env_v1", 0)
            .await
            .unwrap(),
        "the first key inserts"
    );
    let info = synckit::get_sync_key(&db.pool, app, user)
        .await
        .unwrap()
        .expect("key exists");
    assert_eq!(info.encrypted_key, "env_v1");
    assert_eq!(info.key_version, 1);
    assert!(info.pending_key.is_none());

    // A second device that still believes it is at version 0 loses.
    assert!(
        !synckit::upsert_sync_key(&db.pool, app, user, "env_stale", 0)
            .await
            .unwrap(),
        "a stale expected_version is a conflict, not a write"
    );
    assert_eq!(
        synckit::get_sync_key(&db.pool, app, user)
            .await
            .unwrap()
            .unwrap()
            .encrypted_key,
        "env_v1",
        "and the losing envelope must not have landed"
    );

    assert!(
        synckit::upsert_sync_key(&db.pool, app, user, "env_v2", 1)
            .await
            .unwrap()
    );
    let info = synckit::get_sync_key(&db.pool, app, user)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(info.encrypted_key, "env_v2");
    assert_eq!(info.key_version, 2);
}

#[tokio::test]
async fn a_key_belongs_to_one_user_within_one_app() {
    let db = TestDb::new().await;
    let alice = seed_user(&db.pool, "skkey_alice").await;
    let bob = seed_user(&db.pool, "skkey_bob").await;
    let app = seed_app(&db.pool, alice, "keyscope").await;
    synckit::upsert_sync_key(&db.pool, app, alice, "alice_env", 0)
        .await
        .unwrap();

    assert!(
        synckit::get_sync_key(&db.pool, app, bob)
            .await
            .unwrap()
            .is_none(),
        "bob has no key here, and must not be handed alice's envelope"
    );
}

#[tokio::test]
async fn pruning_refuses_a_non_positive_horizon() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "skkey_prune").await;
    let app = seed_app(&db.pool, user, "keyprune").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;
    synckit::push_sync_changes(
        &db.pool,
        app,
        user,
        device,
        Uuid::new_v4(),
        &[change("tasks", "r0")],
    )
    .await
    .unwrap();

    for horizon in [0, -1] {
        assert_eq!(
            synckit::prune_sync_log(&db.pool, horizon).await.unwrap(),
            0,
            "a zero or negative retention is a mistake, not an instruction to \
             delete the whole log"
        );
    }
    assert_eq!(
        synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
            .await
            .unwrap()
            .len(),
        1
    );
    // A real horizon spares entries inside it.
    assert_eq!(synckit::prune_sync_log(&db.pool, 30).await.unwrap(), 0);
}

#[tokio::test]
async fn compaction_holds_back_a_log_no_device_has_pulled() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "skkey_compact").await;
    let app = seed_app(&db.pool, user, "keycompact").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;
    synckit::push_sync_changes(
        &db.pool,
        app,
        user,
        device,
        Uuid::new_v4(),
        &[change("tasks", "r0")],
    )
    .await
    .unwrap();

    assert_eq!(
        synckit::compact_sync_log(&db.pool, app, user, 0)
            .await
            .unwrap(),
        0,
        "a zero-day safety margin compacts nothing"
    );
    assert_eq!(
        synckit::compact_sync_log(&db.pool, app, user, 7)
            .await
            .unwrap(),
        0,
        "the device sits at cursor 0, so nothing is known-pulled"
    );
    assert_eq!(
        synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
            .await
            .unwrap()
            .len(),
        1
    );
}