//! DB-layer contract tests for `db::synckit::rotation`, the SyncKit encryption
//! key-rotation state machine.
//!
//! Audit Run 16 flagged `rotation` as a Concurrency/Testing cold spot (B+), and
//! Phase 3 of the Run 16 remediation made `begin_key_rotation` transactional
//! (`SELECT ... FOR UPDATE` on the `sync_keys` row) precisely to close a
//! check-then-insert race where two devices both observe "no rotation" and both
//! INSERT. These tests pin that fix and the rest of the begin/complete/cancel
//! lifecycle at the DB layer, calling the `db::synckit::rotation` functions
//! directly against real Postgres.

use crate::harness::db::TestDb;
use makenotwork::db::synckit;
use makenotwork::db::{SyncAppId, SyncDeviceId, UserId};

/// Seed a verified user.
async fn seed_user(pool: &sqlx::PgPool, username: &str) -> UserId {
    let hash = makenotwork::auth::hash_password("password123").expect("hash");
    sqlx::query_scalar::<_, UserId>(
        "INSERT INTO users (username, email, password_hash, email_verified)
         VALUES ($1, $2, $3, true) RETURNING id",
    )
    .bind(username)
    .bind(format!("{username}@test.com"))
    .bind(&hash)
    .fetch_one(pool)
    .await
    .expect("seed user")
}

/// Seed a sync app owned by `user`.
async fn seed_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
    sqlx::query_scalar::<_, SyncAppId>(
        "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix)
         VALUES ($1, $2, $3, $4) RETURNING id",
    )
    .bind(user)
    .bind(name)
    .bind(format!("hash_{name}"))
    .bind(&name[..name.len().min(8)])
    .fetch_one(pool)
    .await
    .expect("seed sync app")
}

/// Seed the app's sync_keys row (key_version=1, key_id=1 by default).
async fn seed_key(pool: &sqlx::PgPool, app: SyncAppId, user: UserId) {
    sqlx::query("INSERT INTO sync_keys (app_id, user_id, encrypted_key) VALUES ($1, $2, 'enc_v1')")
        .bind(app)
        .bind(user)
        .execute(pool)
        .await
        .expect("seed sync key");
}

/// Seed a device row (`sync_key_rotations.device_id` FKs `sync_devices`).
async fn seed_device(
    pool: &sqlx::PgPool,
    app: SyncAppId,
    user: UserId,
    name: &str,
) -> SyncDeviceId {
    sqlx::query_scalar::<_, SyncDeviceId>(
        "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
         VALUES ($1, $2, $3, 'macos') RETURNING id",
    )
    .bind(app)
    .bind(user)
    .bind(name)
    .fetch_one(pool)
    .await
    .expect("seed device")
}

async fn rotation_row_count(pool: &sqlx::PgPool, app: SyncAppId) -> i64 {
    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM sync_key_rotations WHERE app_id = $1")
        .bind(app)
        .fetch_one(pool)
        .await
        .expect("count rotations")
}

// ── begin_key_rotation preconditions ─────────────────────────────────────────

#[tokio::test]
async fn begin_rotation_creates_a_row_at_matching_version() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_begin").await;
    let app = seed_app(&db.pool, user, "rotbegin").await;
    seed_key(&db.pool, app, user).await;
    let device = seed_device(&db.pool, app, user, "dev").await;

    let out = synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
        .await
        .unwrap();
    let created = out.expect("matching version begins a rotation");
    // Fresh key was key_id=1, so the rotation targets key_id=2.
    assert_eq!(created.new_key_id, 2);
    assert_eq!(created.device_id, device);
    assert_eq!(rotation_row_count(&db.pool, app).await, 1);
}

#[tokio::test]
async fn begin_rotation_rejects_version_mismatch() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_ver").await;
    let app = seed_app(&db.pool, user, "rotver").await;
    seed_key(&db.pool, app, user).await;

    // Client believes the key is at version 99; server is at 1 -> caller 409s.
    let out = synckit::begin_key_rotation(&db.pool, app, user, SyncDeviceId::new(), "enc_v2", 99)
        .await
        .unwrap();
    assert_eq!(out.err(), Some("key version mismatch"));
    assert_eq!(
        rotation_row_count(&db.pool, app).await,
        0,
        "a mismatch inserts nothing"
    );
}

#[tokio::test]
async fn begin_rotation_errors_when_no_key_exists() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_nokey").await;
    let app = seed_app(&db.pool, user, "rotnokey").await;
    // Deliberately no seed_key.

    let out = synckit::begin_key_rotation(&db.pool, app, user, SyncDeviceId::new(), "enc_v2", 1)
        .await
        .unwrap();
    assert_eq!(out.err(), Some("no encryption key exists"));
}

#[tokio::test]
async fn begin_rotation_is_resumable_by_the_same_device() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_resume").await;
    let app = seed_app(&db.pool, user, "rotresume").await;
    seed_key(&db.pool, app, user).await;
    let device = seed_device(&db.pool, app, user, "dev").await;

    let first = synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
        .await
        .unwrap()
        .expect("first begin");
    // The same device re-issuing begin resumes the identical rotation, not a new one.
    let resumed = synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
        .await
        .unwrap()
        .expect("resume");
    assert_eq!(
        resumed.id, first.id,
        "the same device resumes its own rotation"
    );
    assert_eq!(
        rotation_row_count(&db.pool, app).await,
        1,
        "resume must not create a second row"
    );
}

#[tokio::test]
async fn begin_rotation_blocks_a_second_device() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_block").await;
    let app = seed_app(&db.pool, user, "rotblock").await;
    seed_key(&db.pool, app, user).await;
    let dev_a = seed_device(&db.pool, app, user, "dev-a").await;
    let dev_b = seed_device(&db.pool, app, user, "dev-b").await;

    synckit::begin_key_rotation(&db.pool, app, user, dev_a, "enc_v2", 1)
        .await
        .unwrap()
        .expect("first device begins");
    let second = synckit::begin_key_rotation(&db.pool, app, user, dev_b, "enc_v2", 1)
        .await
        .unwrap();
    assert_eq!(
        second.err(),
        Some("rotation already in progress by another device")
    );
    assert_eq!(rotation_row_count(&db.pool, app).await, 1);
}

/// The Phase 3 remediation's raison d'être: two devices racing `begin` for the
/// same (app, user). The `FOR UPDATE` on the sync_keys row serializes the
/// check-then-insert, so exactly one wins and exactly one rotation row exists,
/// never two rotations, never a unique-constraint 500.
#[tokio::test]
async fn concurrent_begins_yield_exactly_one_rotation() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_race").await;
    let app = seed_app(&db.pool, user, "rotrace").await;
    seed_key(&db.pool, app, user).await;

    let mut handles = Vec::new();
    for i in 0..8 {
        let pool = db.pool.clone();
        let device = seed_device(&db.pool, app, user, &format!("dev-{i}")).await;
        handles.push(tokio::spawn(async move {
            synckit::begin_key_rotation(&pool, app, user, device, "enc_v2", 1).await
        }));
    }

    let mut winners = 0;
    for h in handles {
        // No task may error out (a lost race resolves to Ok(Err(..)), not Err).
        let out = h
            .await
            .expect("task panicked")
            .expect("begin must not hard-error under contention");
        if out.is_ok() {
            winners += 1;
        }
    }

    assert_eq!(
        winners, 1,
        "exactly one racing device may open the rotation"
    );
    assert_eq!(
        rotation_row_count(&db.pool, app).await,
        1,
        "the FOR UPDATE serialization must leave exactly one rotation row"
    );
}

// ── complete + cancel ────────────────────────────────────────────────────────

#[tokio::test]
async fn complete_rotation_with_no_entries_swaps_the_key() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_complete").await;
    let app = seed_app(&db.pool, user, "rotcomp").await;
    seed_key(&db.pool, app, user).await;
    let device = seed_device(&db.pool, app, user, "dev").await;

    let started = synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
        .await
        .unwrap()
        .expect("begin");

    // No sync_log entries exist, so nothing needs re-encryption: complete succeeds.
    let done = synckit::complete_key_rotation(&db.pool, app, user, started.id)
        .await
        .unwrap();
    assert_eq!(
        done,
        Ok(started.new_key_id),
        "complete returns the new key id"
    );

    // The rotation is consumed and the live key advanced.
    assert_eq!(
        rotation_row_count(&db.pool, app).await,
        0,
        "the rotation row is deleted"
    );
    let (version, key_id, enc): (i32, i32, String) =
        sqlx::query_as("SELECT key_version, key_id, encrypted_key FROM sync_keys WHERE app_id = $1 AND user_id = $2")
            .bind(app)
            .bind(user)
            .fetch_one(&db.pool)
            .await
            .unwrap();
    assert_eq!(version, 2, "key_version is bumped on completion");
    assert_eq!(
        key_id, started.new_key_id,
        "the new key id is now the active one"
    );
    assert_eq!(enc, "enc_v2", "the new encrypted key is now live");
}

#[tokio::test]
async fn complete_rotation_is_noop_for_unknown_rotation() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_unknown").await;
    let app = seed_app(&db.pool, user, "rotunkn").await;
    seed_key(&db.pool, app, user).await;

    // A random rotation id that was never started -> Err(0), no key change.
    let out = synckit::complete_key_rotation(&db.pool, app, user, uuid::Uuid::new_v4())
        .await
        .unwrap();
    assert_eq!(out, Err(0));
}

#[tokio::test]
async fn cancel_stale_rotation_respects_the_age_threshold() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "rot_stale").await;
    let app = seed_app(&db.pool, user, "rotstale").await;
    seed_key(&db.pool, app, user).await;
    let device = seed_device(&db.pool, app, user, "dev").await;

    synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
        .await
        .unwrap()
        .expect("begin");

    // A fresh rotation is not stale, so a 24h cancel is a no-op.
    assert!(
        !synckit::cancel_stale_rotation(&db.pool, app, user, 24)
            .await
            .unwrap(),
        "a fresh rotation must not be cancelled as stale"
    );
    assert_eq!(rotation_row_count(&db.pool, app).await, 1);

    // Age it past the threshold -> the stale sweep removes it, unblocking new begins.
    sqlx::query(
        "UPDATE sync_key_rotations SET updated_at = NOW() - INTERVAL '48 hours' WHERE app_id = $1",
    )
    .bind(app)
    .execute(&db.pool)
    .await
    .unwrap();
    assert!(
        synckit::cancel_stale_rotation(&db.pool, app, user, 24)
            .await
            .unwrap()
    );
    assert_eq!(rotation_row_count(&db.pool, app).await, 0);
}