//! Schema bootstrap tests against the harness server
//! (`docs/plans/implementation.md`'s test posture: dockerized
//! `postgres:16-alpine` on :15432 — connection settings ride the
//! environment, never hardcoded; tests without a reachable server
//! skip cleanly so the workspace gates stay green server-less).
//!
//! A fresh schema per test (unique name per test run) is the
//! isolation guarantee — the POC's shared-server
//! parallel-interference caveat is answered by schema-per-test
//! isolation, not sequential-only harnesses.

use std::sync::atomic::{AtomicU64, Ordering};

use alkstore_postgres::{bootstrap, quote_identifier, tables};

const ENV_HOST: &str = "ALKSTORE_PG_HOST";
const ENV_PORT: &str = "ALKSTORE_PG_PORT";
const ENV_USER: &str = "ALKSTORE_PG_USER";
const ENV_PASSWORD: &str = "ALKSTORE_PG_PASSWORD";
const ENV_DB: &str = "ALKSTORE_PG_DB";

fn harness_dsn() -> Option<String> {
    let host = std::env::var(ENV_HOST).ok()?;
    let port: u16 = std::env::var(ENV_PORT).ok()?.parse().ok()?;
    let user = std::env::var(ENV_USER).ok()?;
    let password = std::env::var(ENV_PASSWORD).ok()?;
    let db = std::env::var(ENV_DB).unwrap_or_else(|_| "postgres".to_string());
    Some(format!(
        "host={host} port={port} user={user} password={password} dbname={db}"
    ))
}

async fn harness_client() -> Option<tokio_postgres::Client> {
    let (client, connection) = tokio_postgres::connect(&harness_dsn()?, tokio_postgres::NoTls)
        .await
        .ok()?;
    tokio::spawn(async move {
        let _ = connection.await;
    });
    Some(client)
}

fn instance_namer(tag: &str) -> impl Fn() -> String + use<'_> {
    let counter = AtomicU64::new(0);
    move || {
        format!(
            "{tag}_{}_{}_{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_nanos(),
            counter.fetch_add(1, Ordering::SeqCst),
        )
    }
}

async fn drop_schema(client: &tokio_postgres::Client, schema: &str) {
    let sql = format!("DROP SCHEMA IF EXISTS {} CASCADE", quote_identifier(schema));
    let _ = client.batch_execute(&sql).await;
}

async fn table_names(client: &tokio_postgres::Client, schema: &str) -> Vec<String> {
    client
        .query(
            "SELECT tablename FROM pg_tables WHERE schemaname = $1 ORDER BY tablename",
            &[&schema],
        )
        .await
        .unwrap_or_default()
        .iter()
        .map(|r| r.get::<_, String>(0))
        .collect()
}

async fn column_names(client: &tokio_postgres::Client, schema: &str, table: &str) -> Vec<String> {
    let sql = "SELECT column_name FROM information_schema.columns
         WHERE table_schema = $1 AND table_name = $2
         ORDER BY ordinal_position"
        .to_string();
    client
        .query(&sql, &[&schema, &table])
        .await
        .unwrap_or_default()
        .iter()
        .map(|r| r.get::<_, String>(0))
        .collect()
}

async fn index_names(client: &tokio_postgres::Client, schema: &str) -> Vec<String> {
    client
        .query(
            "SELECT indexname FROM pg_indexes WHERE schemaname = $1 ORDER BY indexname",
            &[&schema],
        )
        .await
        .unwrap_or_default()
        .iter()
        .map(|r| r.get::<_, String>(0))
        .collect()
}

#[tokio::test]
async fn bootstrap_is_idempotent() {
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    let schema = instance_namer("idempotent")();
    assert!(bootstrap(&client, &schema).await.is_ok());
    assert!(bootstrap(&client, &schema).await.is_ok());
    third_run_converges(&client, &schema).await;
    drop_schema(&client, &schema).await;
}

async fn third_run_converges(client: &tokio_postgres::Client, schema: &str) {
    assert!(bootstrap(client, schema).await.is_ok());
    let names = table_names(client, schema).await;
    assert_eq!(names.len(), 6, "expected the 6-table family, saw {names:?}");
}

#[tokio::test]
async fn table_family_exists_with_pinned_shapes() {
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    let schema = instance_namer("shapes")();
    bootstrap(&client, &schema).await.unwrap();

    // Job: the field list of record (Job::from_row, ADR-019 §3 /
    // ADR-021 §2) + the stamped QueueOpts (ADR-010 §3a).
    assert_eq!(
        column_names(&client, &schema, tables::JOB).await,
        vec![
            "id",
            "queue",
            "state",
            "payload",
            "priority",
            "run_at",
            "attempts",
            "max_attempts",
            "worker_id",
            "claimed_at",
            "claim_expires_at",
            "created_at",
            "expires_at",
            "visibility_timeout_s",
            "backoff_base_s",
            "dead_letter_retention_s"
        ]
    );

    // Dead: the job row's diagnosis fields + died_at.
    assert_eq!(
        column_names(&client, &schema, tables::DEAD).await,
        vec![
            "id",
            "queue",
            "payload",
            "priority",
            "run_at",
            "attempts",
            "max_attempts",
            "worker_id",
            "claimed_at",
            "claim_expires_at",
            "created_at",
            "expires_at",
            "visibility_timeout_s",
            "backoff_base_s",
            "dead_letter_retention_s",
            "last_error",
            "died_at"
        ]
    );

    // Events: bigserial offset, nullable key, payload, created_at.
    assert_eq!(
        column_names(&client, &schema, tables::EVENTS).await,
        vec!["id", "stream", "key", "payload", "created_at"]
    );

    // Offsets: (stream, consumer) → offset, the checkpoint.
    assert_eq!(
        column_names(&client, &schema, tables::OFFSETS).await,
        vec!["stream", "consumer", "offset"]
    );

    // Schedule: the substrate's __alkstore_scheduler_tasks shape
    // re-owned (no `enabled`).
    assert_eq!(
        column_names(&client, &schema, tables::SCHEDULE).await,
        vec![
            "name",
            "spec",
            "queue",
            "payload",
            "priority",
            "expires_s",
            "next_fire_at",
            "max_attempts",
            "visibility_timeout_s",
            "backoff_base_s",
            "dead_letter_retention_s"
        ]
    );

    // Locks: name (unique — PK), owner, expiry TTL.
    assert_eq!(
        column_names(&client, &schema, tables::LOCKS).await,
        vec!["name", "owner", "expires_at"]
    );

    drop_schema(&client, &schema).await;
}

#[tokio::test]
async fn payload_columns_are_bytea() {
    let expected_payload_type = "bytea";
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    let schema = instance_namer("bytea")();
    bootstrap(&client, &schema).await.unwrap();

    for table in [tables::JOB, tables::DEAD, tables::EVENTS, tables::SCHEDULE] {
        let table = table.to_string();
        let sql = "SELECT data_type FROM information_schema.columns
             WHERE table_schema = $1 AND table_name = $2 AND column_name = 'payload'"
            .to_string();
        let rows = client.query(&sql, &[&schema, &table]).await.unwrap();
        assert_eq!(rows.len(), 1, "{table}: payload column missing");
        let ty: String = rows[0].get(0);
        assert_eq!(ty, expected_payload_type, "{table}: payload not bytea");
    }

    drop_schema(&client, &schema).await;
}

#[tokio::test]
async fn bigserial_offset_is_monotone_with_gaps_on_delete() {
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    let schema = instance_namer("offsets")();
    bootstrap(&client, &schema).await.unwrap();

    let mut events: Vec<i64> = Vec::new();
    for _ in 0..3 {
        let row = client
            .query_one(
                &format!(
                    "INSERT INTO {} (stream, key, payload, created_at)
                     VALUES ('s', NULL, '\\x7b7d', 1) RETURNING id",
                    alkstore_postgres::QualifiedTable {
                        schema: &schema,
                        name: tables::EVENTS,
                    }
                ),
                &[],
            )
            .await
            .unwrap();
        events.push(row.get(0));
    }
    assert_eq!(events, vec![1, 2, 3], "bigserial starts at 1, step 1");

    client
        .execute(
            &format!(
                "DELETE FROM {} WHERE id = $1",
                alkstore_postgres::QualifiedTable {
                    schema: &schema,
                    name: tables::EVENTS,
                }
            ),
            &[&events[1]],
        )
        .await
        .unwrap();

    let row = client
        .query_one(
            &format!(
                "INSERT INTO {} (stream, key, payload, created_at)
                 VALUES ('s', NULL, '\\x7b7d', 1) RETURNING id",
                alkstore_postgres::QualifiedTable {
                    schema: &schema,
                    name: tables::EVENTS,
                }
            ),
            &[],
        )
        .await
        .unwrap();
    let after_delete: i64 = row.get(0);
    assert_eq!(
        after_delete, 4,
        "the offset never renumbers — deletion leaves a gap, the sequence advances"
    );

    drop_schema(&client, &schema).await;
}

#[tokio::test]
async fn claim_ordering_index_exists() {
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    let schema = instance_namer("idx")();
    bootstrap(&client, &schema).await.unwrap();

    let indexes = index_names(&client, &schema).await;
    let prefixed = |suffix: &str| format!("{schema}_{suffix}");
    let contains = |needle: String| indexes.iter().any(|i| i == &needle);
    assert!(
        contains(prefixed("job_claim_idx")),
        "claim-ordering index missing; saw {indexes:?}"
    );
    assert!(
        contains(prefixed("job_pending_idx")) && contains(prefixed("job_deadline_idx")),
        "hot-path partial indexes missing; saw {indexes:?}"
    );
    assert!(
        contains(prefixed("dead_retention_idx")),
        "dead retention index missing; saw {indexes:?}"
    );
    assert!(
        contains(prefixed("events_read_idx")),
        "events (stream, id) index missing; saw {indexes:?}"
    );
    assert!(
        contains(prefixed("schedule_fire_idx")) && contains(prefixed("locks_expiry_idx")),
        "schedule/locks indexes missing; saw {indexes:?}"
    );

    // The ordering row's column set + directionality (priority DESC,
    // run_at ASC, enqueue order = id) — pg_indexes' indexdef carries
    // it.
    let def = client
        .query_one(
            "SELECT indexdef FROM pg_indexes
             WHERE schemaname = $1 AND indexname = $2",
            &[&schema, &prefixed("job_claim_idx")],
        )
        .await
        .unwrap()
        .get::<_, String>(0);
    for fragment in ["queue", "priority DESC", "run_at", "id"] {
        assert!(
            def.contains(fragment),
            "claim index def missing {fragment}: {def}"
        );
    }
    assert!(
        def.contains("pending"),
        "claim index not partial over the live states: {def}"
    );

    drop_schema(&client, &schema).await;
}

#[tokio::test]
async fn schema_name_is_a_parameter() {
    let expected_default = "alkstore";
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    assert_eq!(alkstore_postgres::DEFAULT_SCHEMA, expected_default);

    // A non-default schema name lands its objects under that name
    // (quoted, reserved-word-safe) — the PgOpts layer's knob works.
    let schema = "select";
    assert!(bootstrap(&client, schema).await.is_ok());
    let names = table_names(&client, schema).await;
    assert_eq!(
        names.len(),
        6,
        "reserved-word schema name quoted and populated"
    );
    drop_schema(&client, schema).await;
}

#[tokio::test]
async fn bootstrap_is_atomic_in_the_schema_scope() {
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    let schema = instance_namer("atomic")();
    assert!(bootstrap(&client, &schema).await.is_ok());

    // A consumer table co-temporaries the instance untouched by a
    // re-bootstrap (the co-tenancy posture — engine DDL never
    // touches non-engine objects in other scopes).
    client
        .execute(
            &format!(
                "CREATE TABLE {}.consumer_thing (x INT)",
                quote_identifier(&schema)
            ),
            &[],
        )
        .await
        .unwrap();
    assert!(bootstrap(&client, &schema).await.is_ok());
    let names = table_names(&client, &schema).await;
    assert_eq!(
        names.len(),
        7,
        "engine re-bootstrap left the co-tenant intact"
    );
    drop_schema(&client, &schema).await;
}

#[tokio::test]
async fn co_tenant_index_name_never_blocks_the_bootstrap() {
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    // A co-tenant carries an index with the bare name the engine's
    // bootstrap would use — `CREATE INDEX IF NOT EXISTS` checks
    // existence across all schemas, so an un-prefixed engine index
    // name would be silently skipped here (the defect class the
    // schema-prefixed index naming exists to exclude).
    client
        .execute("CREATE TABLE public.co_tenant (x INT)", &[])
        .await
        .unwrap();
    client
        .execute("CREATE INDEX job_claim_idx ON public.co_tenant (x)", &[])
        .await
        .unwrap();

    let schema = instance_namer("cotenant")();
    bootstrap(&client, &schema).await.unwrap();
    let indexes = index_names(&client, &schema).await;
    assert!(
        indexes
            .iter()
            .any(|i| i == &format!("{schema}_job_claim_idx")),
        "co-tenant's bare-name index must not skip the engine's; saw {indexes:?}"
    );

    client
        .execute("DROP INDEX public.job_claim_idx", &[])
        .await
        .unwrap();
    client
        .execute("DROP TABLE public.co_tenant", &[])
        .await
        .unwrap();
    drop_schema(&client, &schema).await;
}

#[tokio::test]
async fn custom_schema_quoting_rejects_injection() {
    let Some(client) = harness_client().await else {
        eprintln!("skip: no harness server");
        return;
    };
    // An identifier containing a double-quote must be inert SQL text
    // (doubled) — never a breakout into a second object.
    let hostile = "alk\"; DROP SCHEMA public; CREATE SCHEMA sneaky; --";
    assert!(bootstrap(&client, hostile).await.is_ok());
    let names = table_names(&client, hostile).await;
    assert_eq!(
        names.len(),
        6,
        "hostile-named schema populated exactly once"
    );
    let sneaky = table_names(&client, "sneaky").await;
    assert!(sneaky.is_empty(), "injection created a second schema");
    drop_schema(&client, hostile).await;
    drop_schema(&client, "sneaky").await;
}