use super::{
    AedbConfig, AedbError, AedbInstance, CallerContext, ColumnDef, ColumnType, ConsistencyMode,
    DdlOperation, Mutation, Permission, REACTIVE_ACK_CACHE_MAX_ENTRIES,
    ReactiveCheckpointAckCacheKey, ReactiveCheckpointAckState,
};
use crate::catalog::SYSTEM_PROJECT_ID;
use tempfile::tempdir;

#[tokio::test]
async fn event_outbox_and_reactive_processor_checkpoint_lag_work() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");
    db.create_scope("arcana", "app").await.expect("scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: Some("system".into()),
        if_not_exists: true,
        project_id: SYSTEM_PROJECT_ID.into(),
        scope_id: "app".into(),
        table_name: "reactive_processor_checkpoints".into(),
        columns: vec![
            ColumnDef {
                name: "processor_name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "checkpoint_seq".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "updated_at_micros".into(),
                col_type: ColumnType::Timestamp,
                nullable: false,
            },
        ],
        primary_key: vec!["processor_name".into()],
    }))
    .await
    .expect("create processor checkpoint table");

    let first = db
        .emit_event(
            "arcana",
            "app",
            "hand_settled",
            "hand_1".into(),
            r#"{"user_id":"u1","wager":100,"pnl":-100}"#.into(),
        )
        .await
        .expect("emit first event");
    let second = db
        .emit_event(
            "arcana",
            "app",
            "hand_settled",
            "hand_2".into(),
            r#"{"user_id":"u2","wager":75,"pnl":50}"#.into(),
        )
        .await
        .expect("emit second event");

    let page = db
        .read_event_stream(Some("hand_settled"), 0, 10, ConsistencyMode::AtLatest)
        .await
        .expect("read stream");
    assert_eq!(page.events.len(), 2);
    assert_eq!(page.events[0].event_key, "hand_1");
    assert_eq!(page.events[1].event_key, "hand_2");
    assert_eq!(page.next_commit_seq, Some(second.commit_seq));

    db.ack_reactive_processor_checkpoint("points_processor", first.commit_seq)
        .await
        .expect("ack checkpoint");
    let lag = db
        .reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
        .await
        .expect("lag");
    assert_eq!(lag.processor_name, "points_processor");
    assert_eq!(lag.checkpoint_seq, first.commit_seq);
    assert!(lag.head_seq >= second.commit_seq);
    assert_eq!(
        lag.lag_commits,
        lag.head_seq.saturating_sub(first.commit_seq)
    );
}

#[tokio::test]
async fn reactive_processor_checkpoint_ack_batches_by_watermark() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");
    db.create_scope("arcana", "app").await.expect("scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: Some("system".into()),
        if_not_exists: true,
        project_id: SYSTEM_PROJECT_ID.into(),
        scope_id: "app".into(),
        table_name: "reactive_processor_checkpoints".into(),
        columns: vec![
            ColumnDef {
                name: "processor_name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "checkpoint_seq".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "updated_at_micros".into(),
                col_type: ColumnType::Timestamp,
                nullable: false,
            },
        ],
        primary_key: vec!["processor_name".into()],
    }))
    .await
    .expect("create processor checkpoint table");

    let first = db
        .emit_event(
            "arcana",
            "app",
            "hand_settled",
            "hand_1".into(),
            r#"{"user_id":"u1","wager":100,"pnl":-100}"#.into(),
        )
        .await
        .expect("emit first event");
    let second = db
        .emit_event(
            "arcana",
            "app",
            "hand_settled",
            "hand_2".into(),
            r#"{"user_id":"u2","wager":75,"pnl":50}"#.into(),
        )
        .await
        .expect("emit second event");

    let persisted = db
        .ack_reactive_processor_checkpoint_batched("points_processor", first.commit_seq, 2)
        .await
        .expect("first batched ack");
    assert!(persisted.is_some(), "first ack should persist baseline");

    let deferred = db
        .ack_reactive_processor_checkpoint_batched("points_processor", first.commit_seq + 1, 2)
        .await
        .expect("deferred batched ack");
    assert!(deferred.is_none(), "ack below watermark should be deferred");

    let lag_after_defer = db
        .reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
        .await
        .expect("lag after deferred ack");
    assert_eq!(lag_after_defer.checkpoint_seq, first.commit_seq);

    let persisted_after_watermark = db
        .ack_reactive_processor_checkpoint_batched("points_processor", first.commit_seq + 2, 2)
        .await
        .expect("persist on watermark");
    assert!(
        persisted_after_watermark.is_some(),
        "watermark crossing should persist checkpoint"
    );

    let lag = db
        .reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
        .await
        .expect("lag");
    assert_eq!(lag.processor_name, "points_processor");
    assert_eq!(lag.checkpoint_seq, first.commit_seq + 2);
    assert!(lag.head_seq >= second.commit_seq);
}

#[tokio::test]
async fn reactive_processor_checkpoint_ack_rejects_regression() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");
    db.create_scope("arcana", "app").await.expect("scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: Some("system".into()),
        if_not_exists: true,
        project_id: SYSTEM_PROJECT_ID.into(),
        scope_id: "app".into(),
        table_name: "reactive_processor_checkpoints".into(),
        columns: vec![
            ColumnDef {
                name: "processor_name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "checkpoint_seq".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "updated_at_micros".into(),
                col_type: ColumnType::Timestamp,
                nullable: false,
            },
        ],
        primary_key: vec!["processor_name".into()],
    }))
    .await
    .expect("create processor checkpoint table");

    let first = db
        .emit_event(
            "arcana",
            "app",
            "hand_settled",
            "hand_1".into(),
            r#"{"user_id":"u1","wager":100,"pnl":-100}"#.into(),
        )
        .await
        .expect("emit first event");
    let second = db
        .emit_event(
            "arcana",
            "app",
            "hand_settled",
            "hand_2".into(),
            r#"{"user_id":"u2","wager":75,"pnl":50}"#.into(),
        )
        .await
        .expect("emit second event");

    db.ack_reactive_processor_checkpoint("points_processor", second.commit_seq)
        .await
        .expect("persist newer checkpoint");

    let err = db
        .ack_reactive_processor_checkpoint("points_processor", first.commit_seq)
        .await
        .expect_err("checkpoint regression must fail");
    assert!(matches!(err, AedbError::Validation(_)));

    let lag = db
        .reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
        .await
        .expect("lag");
    assert_eq!(lag.checkpoint_seq, second.commit_seq);
}

#[tokio::test]
async fn reactive_processor_checkpoint_batched_rejects_stale_persist_after_concurrent_advance() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");
    db.create_scope("arcana", "app").await.expect("scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: Some("system".into()),
        if_not_exists: true,
        project_id: SYSTEM_PROJECT_ID.into(),
        scope_id: "app".into(),
        table_name: "reactive_processor_checkpoints".into(),
        columns: vec![
            ColumnDef {
                name: "processor_name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "checkpoint_seq".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "updated_at_micros".into(),
                col_type: ColumnType::Timestamp,
                nullable: false,
            },
        ],
        primary_key: vec!["processor_name".into()],
    }))
    .await
    .expect("create processor checkpoint table");

    let first = db
        .emit_event(
            "arcana",
            "app",
            "hand_settled",
            "hand_1".into(),
            r#"{"user_id":"u1","wager":100,"pnl":-100}"#.into(),
        )
        .await
        .expect("emit first event");
    let second = db
        .emit_event(
            "arcana",
            "app",
            "hand_settled",
            "hand_2".into(),
            r#"{"user_id":"u2","wager":75,"pnl":50}"#.into(),
        )
        .await
        .expect("emit second event");

    let stale = db
        .build_reactive_processor_checkpoint_envelope("points_processor", first.commit_seq)
        .await
        .expect("stale envelope");

    db.ack_reactive_processor_checkpoint("points_processor", second.commit_seq)
        .await
        .expect("persist newer checkpoint");

    let err = db
        .commit_envelope_prevalidated_internal("stale_reactive_ack_test", stale)
        .await
        .expect_err("stale asserted ack must fail");
    assert!(matches!(err, AedbError::Conflict(_)));

    let lag = db
        .reactive_processor_lag("points_processor", ConsistencyMode::AtLatest)
        .await
        .expect("lag");
    assert_eq!(lag.checkpoint_seq, second.commit_seq);
}

#[tokio::test]
async fn reactive_processor_checkpoint_batched_as_isolated_by_caller() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");
    db.create_scope("arcana", "app").await.expect("scope");
    db.emit_event("arcana", "app", "bootstrap", "evt-1".into(), "{}".into())
        .await
        .expect("bootstrap system scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: Some("system".into()),
        if_not_exists: true,
        project_id: SYSTEM_PROJECT_ID.into(),
        scope_id: "app".into(),
        table_name: "reactive_processor_checkpoints".into(),
        columns: vec![
            ColumnDef {
                name: "processor_name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "checkpoint_seq".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "updated_at_micros".into(),
                col_type: ColumnType::Timestamp,
                nullable: false,
            },
        ],
        primary_key: vec!["processor_name".into()],
    }))
    .await
    .expect("create processor checkpoint table");
    db.ack_reactive_processor_checkpoint("bootstrap", 1)
        .await
        .expect("bootstrap processor checkpoint table");

    for caller in ["alice", "bob"] {
        db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
            actor_id: None,
            delegable: false,
            caller_id: caller.into(),
            permission: Permission::TableWrite {
                project_id: SYSTEM_PROJECT_ID.into(),
                scope_id: "app".into(),
                table_name: "reactive_processor_checkpoints".into(),
            },
        }))
        .await
        .expect("grant system table write");
    }

    let first = db
        .ack_reactive_processor_checkpoint_batched_as(
            CallerContext::new("alice"),
            "shared_processor",
            42,
            100,
        )
        .await
        .expect("alice ack");
    assert!(first.is_some(), "alice baseline ack should persist");

    let second = db
        .ack_reactive_processor_checkpoint_batched_as(
            CallerContext::new("bob"),
            "shared_processor",
            42,
            100,
        )
        .await
        .expect("bob ack");
    assert!(
        second.is_some(),
        "bob baseline ack should persist independently"
    );
}

#[tokio::test]
async fn reactive_processor_checkpoint_batched_as_does_not_poison_cache_on_permission_failure() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");
    db.create_scope("arcana", "app").await.expect("scope");
    db.emit_event("arcana", "app", "bootstrap", "evt-1".into(), "{}".into())
        .await
        .expect("bootstrap system scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: Some("system".into()),
        if_not_exists: true,
        project_id: SYSTEM_PROJECT_ID.into(),
        scope_id: "app".into(),
        table_name: "reactive_processor_checkpoints".into(),
        columns: vec![
            ColumnDef {
                name: "processor_name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "checkpoint_seq".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "updated_at_micros".into(),
                col_type: ColumnType::Timestamp,
                nullable: false,
            },
        ],
        primary_key: vec!["processor_name".into()],
    }))
    .await
    .expect("create processor checkpoint table");
    db.ack_reactive_processor_checkpoint("bootstrap", 1)
        .await
        .expect("bootstrap processor checkpoint table");

    let denied = db
        .ack_reactive_processor_checkpoint_batched_as(
            CallerContext::new("mallory"),
            "perm_test_processor",
            100,
            100,
        )
        .await
        .expect_err("mallory should be denied before grant");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "mallory".into(),
        permission: Permission::TableWrite {
            project_id: SYSTEM_PROJECT_ID.into(),
            scope_id: "app".into(),
            table_name: "reactive_processor_checkpoints".into(),
        },
    }))
    .await
    .expect("grant system table write");

    let persisted = db
        .ack_reactive_processor_checkpoint_batched_as(
            CallerContext::new("mallory"),
            "perm_test_processor",
            100,
            100,
        )
        .await
        .expect("mallory should succeed after grant");
    assert!(
        persisted.is_some(),
        "failed pre-grant attempt must not suppress later successful persist"
    );
}

#[tokio::test]
async fn reactive_processor_checkpoint_batched_cache_is_bounded() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");
    db.create_scope("arcana", "app").await.expect("scope");
    db.emit_event("arcana", "app", "bootstrap", "evt-1".into(), "{}".into())
        .await
        .expect("bootstrap system scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: Some("system".into()),
        if_not_exists: true,
        project_id: SYSTEM_PROJECT_ID.into(),
        scope_id: "app".into(),
        table_name: "reactive_processor_checkpoints".into(),
        columns: vec![
            ColumnDef {
                name: "processor_name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "checkpoint_seq".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "updated_at_micros".into(),
                col_type: ColumnType::Timestamp,
                nullable: false,
            },
        ],
        primary_key: vec!["processor_name".into()],
    }))
    .await
    .expect("create processor checkpoint table");
    {
        let mut cache = db.reactive_processor_ack_watermarks.lock();
        for i in 0..(REACTIVE_ACK_CACHE_MAX_ENTRIES + 200) {
            cache.insert(
                ReactiveCheckpointAckCacheKey {
                    processor_name: format!("cache-seed-{i}"),
                    caller_id: None,
                },
                ReactiveCheckpointAckState {
                    last_persisted_seq: i as u64,
                    last_touch_micros: i as u64,
                },
            );
        }
    }

    let persisted = db
        .ack_reactive_processor_checkpoint_batched("cache-boundary", 1, 1)
        .await
        .expect("ack should succeed");
    assert!(persisted.is_some());
    let cache_len = db.reactive_processor_ack_watermarks.lock().len();
    assert!(
        cache_len <= REACTIVE_ACK_CACHE_MAX_ENTRIES,
        "cache should be pruned to cap; got {}",
        cache_len
    );
}
