use super::*;

async fn ledgerless_v1(pool: &sqlx::PgPool) {
    MigrationAuthority::apply(pool).await.unwrap();
    remove_v4_schema(pool).await;
    remove_v3_schema(pool).await;
    sqlx::raw_sql(
        "drop table __crank_migrations, __crank_migration_legacy_audit;
         drop table __crank_mcp_migrations;
         drop index mcp_transport_sessions_expires_at_idx;
         drop table __crank_ext_migrations;
         create table __crank_ext_migrations (
             extension_name text not null,
             version integer not null,
             applied_at timestamptz not null default now(),
             primary key (extension_name, version)
         );
         drop table __crank_core_migrations;",
    )
    .execute(pool)
    .await
    .unwrap();
}

#[tokio::test]
async fn ledgerless_baseline_with_optional_index_drift_is_rejected_without_writes() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_index").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::raw_sql(
        "drop index mcp_transport_sessions_workspace_agent_idx;
         create index mcp_transport_sessions_workspace_agent_idx
             on mcp_transport_sessions(id);",
    )
    .execute(&pool)
    .await
    .unwrap();

    let error = MigrationAuthority::apply(&pool).await.unwrap_err();
    assert_eq!(error.code(), "partial_sequence");
    assert_eq!(error.stage(), "preflight.legacy_fingerprint");
    let core_exists: bool = sqlx::query_scalar(
        "select to_regclass(format('%I.%I', current_schema(), '__crank_core_migrations')) is not null",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(!core_exists, "rejected adoption must remain read-only");
}

#[tokio::test]
async fn ledgerless_baseline_with_disabled_integrity_triggers_is_rejected_without_writes() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_triggers").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::query("alter table memberships disable trigger all")
        .execute(&pool)
        .await
        .unwrap();

    let error = MigrationAuthority::apply(&pool).await.unwrap_err();
    assert_eq!(error.code(), "partial_sequence");
    assert_eq!(error.stage(), "preflight.legacy_fingerprint");
    let core_exists: bool = sqlx::query_scalar(
        "select to_regclass(format('%I.%I', current_schema(), '__crank_core_migrations')) is not null",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(!core_exists, "rejected adoption must remain read-only");
}

#[tokio::test]
async fn published_ledgerless_baseline_upgrades_without_data_loss() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_baseline").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::query(
        "insert into operations
            (id, workspace_id, name, display_name, protocol, status, created_at, updated_at)
         values ('op_ledgerless', 'ws_default', 'ledgerless', 'Ledgerless', 'rest', 'draft', now(), now())",
    )
    .execute(&pool)
    .await
    .unwrap();

    assert_eq!(
        MigrationAuthority::preflight(&pool).await.unwrap(),
        MigrationPreflight::MigrationRequired {
            current: 0,
            target: 13,
        }
    );
    MigrationAuthority::apply(&pool).await.unwrap();

    let operation_count: i64 =
        sqlx::query_scalar("select count(*) from operations where id = 'op_ledgerless'")
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_eq!(operation_count, 1);
    assert_eq!(
        MigrationAuthority::preflight(&pool).await.unwrap(),
        MigrationPreflight::Current { version: 13 }
    );
}

#[tokio::test]
async fn published_ledgerless_historical_column_layout_upgrades_without_data_loss() {
    let database_url =
        crank_test_support::postgres_schema_url("test_ledgerless_historical_layout").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::raw_sql(
        "alter table platform_api_keys
             drop column key_kind,
             drop column expires_at,
             drop column allowed_origins_json;
         alter table platform_api_keys
             add column key_kind text not null default 'mcp_client',
             add column expires_at timestamptz null,
             add column allowed_origins_json jsonb not null default '[]'::jsonb;

         drop index approval_requests_pending_fingerprint_idx;
         alter table approval_requests
             drop column execution_started_at,
             drop column execution_attempts,
             drop column request_fingerprint,
             add column confirmation_title text not null default '',
             add column confirmation_body text not null default '';
         alter table approval_requests
             alter column confirmation_title drop default,
             alter column confirmation_body drop default,
             drop column confirmation_title,
             drop column confirmation_body,
             add column execution_started_at timestamptz null,
             add column execution_attempts integer not null default 0,
             add column request_fingerprint text null;
         create unique index approval_requests_pending_fingerprint_idx
             on approval_requests(agent_id, operation_id, operation_version, request_fingerprint)
             where status = 'pending' and request_fingerprint is not null;

         insert into operations
            (id, workspace_id, name, display_name, protocol, status, created_at, updated_at)
         values ('op_historical_layout', 'ws_default', 'historical-layout', 'Historical layout', 'rest', 'draft', now(), now());
         insert into agents
            (id, workspace_id, slug, display_name, description, status, created_at, updated_at)
         values ('agent_historical_layout', 'ws_default', 'historical-layout', 'Historical layout', '', 'draft', now(), now());
         insert into platform_api_keys
            (id, workspace_id, agent_id, name, prefix, secret_hash, key_kind, scopes_json, status,
             created_at, expires_at, allowed_origins_json)
         values ('key_historical_layout', 'ws_default', 'agent_historical_layout', 'Historical layout',
                 'cp_', 'hash', 'admin', '[]'::jsonb, 'active', now(),
                 '2030-01-02 03:04:05+00'::timestamptz, '[\"https://example.test\"]'::jsonb);
         insert into approval_requests
            (id, workspace_id, agent_id, operation_id, operation_version, status, risk_level,
             request_payload_json, created_at, expires_at, execution_started_at,
             execution_attempts, request_fingerprint)
         values ('approval_historical_layout', 'ws_default', 'agent_historical_layout', 'op_historical_layout', 1,
                 'pending', 'high', '{\"layout\":\"historical\"}'::jsonb, now(), now() + interval '1 hour',
                 '2029-02-03 04:05:06+00'::timestamptz, 3, 'historical-fingerprint');",
    )
    .execute(&pool)
    .await
    .unwrap();

    let historical_ordinals: Vec<(String, i32)> = sqlx::query_as(
        "select table_name || '.' || column_name, ordinal_position
         from information_schema.columns
         where table_schema = current_schema()
           and ((table_name = 'platform_api_keys'
                 and column_name in ('key_kind', 'expires_at', 'allowed_origins_json'))
             or (table_name = 'approval_requests'
                 and column_name in ('execution_started_at', 'execution_attempts', 'request_fingerprint')))
         order by table_name, ordinal_position",
    )
    .fetch_all(&pool)
    .await
    .unwrap();
    assert_eq!(
        historical_ordinals,
        vec![
            ("approval_requests.execution_started_at".to_owned(), 22),
            ("approval_requests.execution_attempts".to_owned(), 23),
            ("approval_requests.request_fingerprint".to_owned(), 24),
            ("platform_api_keys.key_kind".to_owned(), 15),
            ("platform_api_keys.expires_at".to_owned(), 16),
            ("platform_api_keys.allowed_origins_json".to_owned(), 17),
        ],
        "fixture must reproduce the published in-place column layout",
    );

    assert_eq!(
        MigrationAuthority::preflight(&pool).await.unwrap(),
        MigrationPreflight::MigrationRequired {
            current: 0,
            target: 13,
        }
    );
    MigrationAuthority::apply(&pool).await.unwrap();

    let key_count: i64 = sqlx::query_scalar(
        "select count(*) from platform_api_keys
         where id = 'key_historical_layout'
           and key_kind = 'admin'
           and expires_at = '2030-01-02 03:04:05+00'::timestamptz
           and allowed_origins_json = '[\"https://example.test\"]'::jsonb",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(
        key_count, 1,
        "platform key row must survive the layout upgrade"
    );
    let approval_count: i64 = sqlx::query_scalar(
        "select count(*) from approval_requests
         where id = 'approval_historical_layout'
           and request_payload_json = '{\"layout\":\"historical\"}'::jsonb
           and execution_started_at = '2029-02-03 04:05:06+00'::timestamptz
           and execution_attempts = 3
           and request_fingerprint = 'historical-fingerprint'",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(
        approval_count, 1,
        "approval row must survive the layout upgrade"
    );
    assert_eq!(
        MigrationAuthority::preflight(&pool).await.unwrap(),
        MigrationPreflight::Current { version: 13 }
    );
}

#[tokio::test]
async fn published_initial_mcp_layout_is_accepted() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_initial_mcp").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::raw_sql(
        "drop table mcp_transport_sessions;
         create table mcp_transport_sessions (
             id text primary key,
             protocol_version text not null,
             initialized boolean not null default false,
             workspace_slug text not null,
             agent_slug text not null,
             created_at timestamptz not null,
             updated_at timestamptz not null,
             expires_at timestamptz null
         );
         create index mcp_transport_sessions_workspace_agent_idx
             on mcp_transport_sessions(workspace_slug, agent_slug, updated_at desc);",
    )
    .execute(&pool)
    .await
    .unwrap();

    assert_eq!(
        MigrationAuthority::preflight(&pool).await.unwrap(),
        MigrationPreflight::MigrationRequired {
            current: 0,
            target: 13,
        }
    );
    MigrationAuthority::apply(&pool).await.unwrap();
    let supports_elicitation: bool = sqlx::query_scalar(
        "select exists (
             select 1 from information_schema.columns
             where table_schema = current_schema()
               and table_name = 'mcp_transport_sessions'
               and column_name = 'supports_elicitation'
         )",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(supports_elicitation);
    assert_eq!(
        MigrationAuthority::preflight(&pool).await.unwrap(),
        MigrationPreflight::Current { version: 13 }
    );
}

#[tokio::test]
async fn published_in_place_mcp_upgrade_layout_is_accepted() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_mcp_upgrade").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::raw_sql(
        "drop table mcp_transport_sessions;
         create table mcp_transport_sessions (
             id text primary key,
             protocol_version text not null,
             initialized boolean not null default false,
             workspace_slug text not null,
             agent_slug text not null,
             created_at timestamptz not null,
             updated_at timestamptz not null,
             expires_at timestamptz null
         );
         create index mcp_transport_sessions_workspace_agent_idx
             on mcp_transport_sessions(workspace_slug, agent_slug, updated_at desc);
         alter table mcp_transport_sessions
             add column supports_elicitation boolean not null default false;",
    )
    .execute(&pool)
    .await
    .unwrap();

    assert_eq!(
        MigrationAuthority::preflight(&pool).await.unwrap(),
        MigrationPreflight::MigrationRequired {
            current: 0,
            target: 13,
        }
    );
    MigrationAuthority::apply(&pool).await.unwrap();
    assert_eq!(
        MigrationAuthority::preflight(&pool).await.unwrap(),
        MigrationPreflight::Current { version: 13 }
    );
}

#[tokio::test]
async fn ledgerless_baseline_with_future_drift_is_rejected_without_writes() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_drift").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::query("alter table invocation_logs add column trace_id text")
        .execute(&pool)
        .await
        .unwrap();

    let error = MigrationAuthority::apply(&pool).await.unwrap_err();
    assert_eq!(error.code(), "partial_sequence");
    assert_eq!(error.stage(), "preflight.legacy_fingerprint");
    let core_exists: bool = sqlx::query_scalar(
        "select to_regclass(format('%I.%I', current_schema(), '__crank_core_migrations')) is not null",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(!core_exists, "rejected adoption must remain read-only");
}

#[tokio::test]
async fn ledgerless_baseline_with_default_drift_is_rejected_without_writes() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_default").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::query("alter table workspaces alter column status set default 'active'")
        .execute(&pool)
        .await
        .unwrap();

    let error = MigrationAuthority::apply(&pool).await.unwrap_err();
    assert_eq!(error.code(), "partial_sequence");
    assert_eq!(error.stage(), "preflight.legacy_fingerprint");
    let core_exists: bool = sqlx::query_scalar(
        "select to_regclass(format('%I.%I', current_schema(), '__crank_core_migrations')) is not null",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(!core_exists, "rejected adoption must remain read-only");
}

#[tokio::test]
async fn ledgerless_baseline_with_constraint_drift_is_rejected_without_writes() {
    let database_url =
        crank_test_support::postgres_schema_url("test_ledgerless_constraint_definition").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::query(
        "alter table workspaces
         add constraint workspaces_status_nonempty check (status <> '')",
    )
    .execute(&pool)
    .await
    .unwrap();

    let error = MigrationAuthority::apply(&pool).await.unwrap_err();
    assert_eq!(error.code(), "partial_sequence");
    assert_eq!(error.stage(), "preflight.legacy_fingerprint");
    let core_exists: bool = sqlx::query_scalar(
        "select to_regclass(format('%I.%I', current_schema(), '__crank_core_migrations')) is not null",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(!core_exists, "rejected adoption must remain read-only");
}

#[tokio::test]
async fn ledgerless_baseline_with_rls_drift_is_rejected_without_writes() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_rls").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::query("alter table workspaces enable row level security")
        .execute(&pool)
        .await
        .unwrap();

    let error = MigrationAuthority::apply(&pool).await.unwrap_err();
    assert_eq!(error.code(), "partial_sequence");
    assert_eq!(error.stage(), "preflight.legacy_fingerprint");
    let core_exists: bool = sqlx::query_scalar(
        "select to_regclass(format('%I.%I', current_schema(), '__crank_core_migrations')) is not null",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(!core_exists, "rejected adoption must remain read-only");
}

#[tokio::test]
async fn ledgerless_baseline_with_nullability_drift_is_rejected_without_writes() {
    let database_url = crank_test_support::postgres_schema_url("test_ledgerless_constraint").await;
    let pool = sqlx::PgPool::connect(&database_url).await.unwrap();
    ledgerless_v1(&pool).await;
    sqlx::query("alter table workspaces alter column status drop not null")
        .execute(&pool)
        .await
        .unwrap();

    let error = MigrationAuthority::apply(&pool).await.unwrap_err();
    assert_eq!(error.code(), "partial_sequence");
    assert_eq!(error.stage(), "preflight.legacy_fingerprint");
    let core_exists: bool = sqlx::query_scalar(
        "select to_regclass(format('%I.%I', current_schema(), '__crank_core_migrations')) is not null",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(!core_exists, "rejected adoption must remain read-only");
}