//! The suite-driving test target: the Postgres engine's backlog column
//! (ADR-022's option (a) — the rows live in `alkstore-contract-suite`,
//! this target is the executor that runs them against the pg factory;
//! wave 5 runs the cross-engine equivalence rows on top).
//!
//! The factory: a **fresh schema per `open`** (the SQLite factory's
//! fresh-temp-file precedent; the POC's shared-server
//! parallel-interference caveat is answered by exactly this isolation —
//! each property's rows live in their own schema, so parallel property
//! runs never observe one another), teardown `DROP SCHEMA IF EXISTS …
//! CASCADE` over exactly the schemas this factory instance minted,
//! tracked per-instance (an idempotent, owned teardown — never the
//! shared server's other schemas). The drop is safe against an
//! abandoned store: `CASCADE` removes the dependents server-side while
//! the store's pooled connections fail closed on their next use (the
//! documented post-close posture). Close-arm rows drop the store handle
//! (the dispose carrier) — teardown under a live store is not a close
//! mechanism.
//!
//! Server-less environments skip cleanly: the rows are gated behind a
//! reachability probe so the workspace gates stay green; wave
//! acceptance runs them against the harness server (dockerized
//! `postgres:16-alpine` on :15432, connection settings riding the
//! environment — the schema task's convention).

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

use alkstore::{BoxedFuture, Error, Result, Store};
use alkstore_contract_suite::StoreFactory;
use alkstore_postgres::{PgOpts, open, quote_identifier};

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 admin_connect(dsn: &str) -> Result<tokio_postgres::Client> {
    let (client, connection) = tokio_postgres::connect(dsn, tokio_postgres::NoTls)
        .await
        .map_err(Error::database)?;
    tokio::spawn(async move {
        let _ = connection.await;
    });
    Ok(client)
}

/// The Postgres suite factory: a fresh engine-owned schema per `open`
/// (unique per call — the isolation guarantee), tracked on the
/// instance, tearing down with `DROP SCHEMA IF EXISTS … CASCADE` over
/// exactly the schemas it minted. `Send + Sync` — instance state is
/// the DSN, a counter, and the schema registry. Default `PgOpts` (pool
/// size, `synchronous_commit` ship config) ride; only the schema name
/// is factory policy.
struct PgFactory {
    dsn: String,
    schema_seq: AtomicU64,
    schemas: std::sync::Mutex<Vec<String>>,
}

impl PgFactory {
    fn new(tag: &str) -> Self {
        let dsn = harness_dsn().expect("the harness server is configured for the suite");
        PgFactory {
            dsn,
            // Seeded per (tag, pid, time): two factory instances never
            // share a schema name, racing opens included.
            schema_seq: AtomicU64::new(
                (std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.subsec_nanos() as u64)
                    .unwrap_or(0))
                    ^ (tag.len() as u64)
                    ^ ((std::process::id() as u64) << 8),
            ),
            schemas: std::sync::Mutex::new(Vec::new()),
        }
    }

    async fn fresh_schema(&self) -> String {
        let name = format!(
            "alkstore_suite_{}_{}_{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0),
            self.schema_seq.fetch_add(1, Ordering::SeqCst),
        );
        self.schemas
            .lock()
            .expect("factory schema registry poisoned")
            .push(name.clone());
        name
    }
}

impl StoreFactory for PgFactory {
    fn open(&self) -> BoxedFuture<'_, Result<Box<dyn Store>>> {
        Box::pin(async move {
            let schema = self.fresh_schema().await;
            let opts = PgOpts {
                schema,
                ..PgOpts::default()
            };
            // Unreachable-server: the typed `Database` error surfaces
            // (tests gate on reachability first so server-less runs
            // skip).
            open(&self.dsn, opts).await
        })
    }

    fn teardown(&self) -> BoxedFuture<'_, Result<()>> {
        let dsn = self.dsn.clone();
        let schemas = self
            .schemas
            .lock()
            .expect("factory schema registry poisoned")
            .clone();
        Box::pin(async move {
            let client = admin_connect(&dsn).await?;
            for schema in &schemas {
                let sql = format!("DROP SCHEMA IF EXISTS {} CASCADE", quote_identifier(schema));
                client.batch_execute(&sql).await.map_err(Error::database)?;
            }
            Ok(())
        })
    }
}

/// The server reachability probe: server-less environments skip the
/// rows (the gates stay green); the harness server's presence runs
/// them. One cheap admin connection per test; `teardown` reconnects
/// independently (no shared admin client lifetime).
async fn harness_ready() -> bool {
    let Some(dsn) = harness_dsn() else {
        eprintln!("skip: no harness server configured");
        return false;
    };
    match tokio::time::timeout(std::time::Duration::from_secs(3), admin_connect(&dsn)).await {
        Ok(Ok(_)) => true,
        Ok(Err(e)) => {
            eprintln!("skip: harness server unreachable ({e})");
            false
        }
        Err(_) => {
            eprintln!("skip: harness server unreachable (timeout)");
            false
        }
    }
}

mod rows {
    use alkstore_contract_suite::properties::{
        backoff_curve_equivalence, concurrent_try_lock_loser_is_a_value,
        drop_rollback_leaves_no_ghosts, duration_refusal_on_non_positive_ttl,
        enqueue_opts_resolution, extent_clamp_semantics, in_tx_reads_see_own_writes,
        job_handle_validity_predicate, lock_ttl_expiry_and_reacquisition,
        name_validation_rejects_empty_and_reserved, outbox_enqueue_tx_commit_atomicity,
        payload_round_trip_stores_exact_encoding, payload_too_large_produced_on_pg,
        publish_with_key_tx_commit_atomicity, queue_depth_reclaim_and_dead_letter,
        receiver_close_and_save_arms, receiver_offset_pre_yield_is_zero, scheduler_boundary_fires,
        scheduler_bounded_catchup, scheduler_leadership_discipline, stream_ordering_equivalence,
        sweep_no_stranded_rows, trim_to_semantics, wake_receiver_shapes,
        with_tx_panicking_closure_rolls_back,
    };

    use super::PgFactory;

    fn factory(tag: &str) -> PgFactory {
        PgFactory::new(tag)
    }

    macro_rules! harness_row {
        ($fn_name:ident, $row_fn:path, $tag:literal) => {
            #[tokio::test(flavor = "multi_thread")]
            async fn $fn_name() {
                if !super::harness_ready().await {
                    return;
                }
                $row_fn(&factory($tag)).await;
            }
        };
    }

    harness_row!(
        row_name_validation_rejects_empty_and_reserved,
        name_validation_rejects_empty_and_reserved,
        "row-name-validation"
    );
    harness_row!(
        row_extent_clamp_semantics,
        extent_clamp_semantics,
        "row-extent-clamp"
    );
    harness_row!(
        row_duration_refusal_on_non_positive_ttl,
        duration_refusal_on_non_positive_ttl,
        "row-duration-refusal"
    );
    harness_row!(
        row_lock_ttl_expiry_and_reacquisition,
        lock_ttl_expiry_and_reacquisition,
        "row-lock-ttl-expiry"
    );
    harness_row!(
        row_concurrent_try_lock_loser_is_a_value,
        concurrent_try_lock_loser_is_a_value,
        "row-lock-loser-value"
    );
    harness_row!(
        row_payload_round_trip_stores_exact_encoding,
        payload_round_trip_stores_exact_encoding,
        "row-payload-round-trip"
    );
    harness_row!(
        row_payload_too_large_produced_on_pg,
        payload_too_large_produced_on_pg,
        "row-too-large"
    );
    harness_row!(
        row_drop_rollback_leaves_no_ghosts,
        drop_rollback_leaves_no_ghosts,
        "row-drop-rollback"
    );
    harness_row!(
        row_in_tx_reads_see_own_writes,
        in_tx_reads_see_own_writes,
        "row-in-tx-ryow"
    );
    harness_row!(
        row_enqueue_opts_resolution,
        enqueue_opts_resolution,
        "row-opts-resolution"
    );
    harness_row!(
        row_backoff_curve_equivalence,
        backoff_curve_equivalence,
        "row-backoff-curve"
    );
    harness_row!(
        row_receiver_close_and_save_arms,
        receiver_close_and_save_arms,
        "row-receiver-arms"
    );
    harness_row!(
        row_receiver_offset_pre_yield_is_zero,
        receiver_offset_pre_yield_is_zero,
        "row-receiver-offset-preyield"
    );
    harness_row!(
        row_job_handle_validity_predicate,
        job_handle_validity_predicate,
        "row-handle-predicate"
    );
    harness_row!(
        row_queue_depth_reclaim_and_dead_letter,
        queue_depth_reclaim_and_dead_letter,
        "row-queue-depth"
    );
    harness_row!(
        row_sweep_no_stranded_rows,
        sweep_no_stranded_rows,
        "row-sweep-stranded"
    );
    harness_row!(
        row_scheduler_boundary_fires,
        scheduler_boundary_fires,
        "row-scheduler-boundary"
    );
    harness_row!(
        row_scheduler_bounded_catchup,
        scheduler_bounded_catchup,
        "row-scheduler-catchup"
    );
    harness_row!(
        row_scheduler_leadership_discipline,
        scheduler_leadership_discipline,
        "row-scheduler-leadership"
    );
    harness_row!(
        row_outbox_enqueue_tx_commit_atomicity,
        outbox_enqueue_tx_commit_atomicity,
        "row-outbox-tx-atomicity"
    );
    harness_row!(
        row_publish_with_key_tx_commit_atomicity,
        publish_with_key_tx_commit_atomicity,
        "row-keyed-tx-atomicity"
    );
    harness_row!(
        row_stream_ordering_equivalence,
        stream_ordering_equivalence,
        "row-stream-ordering"
    );
    harness_row!(row_trim_to_semantics, trim_to_semantics, "row-trim-to");
    harness_row!(
        row_wake_receiver_shapes,
        wake_receiver_shapes,
        "row-wake-shapes"
    );
    harness_row!(
        row_with_tx_panicking_closure_rolls_back,
        with_tx_panicking_closure_rolls_back,
        "row-panic-rollback"
    );
}

mod factory_shape {
    use alkstore_contract_suite::StoreFactory;

    use super::PgFactory;

    /// ADR-022's factory contract: every open yield is isolated (no two
    /// opens share rows — proved by a row one store lands being
    /// invisible to the other) and teardown is idempotent per instance;
    /// the failing-assertion early-exit (the panic path) leaves teardown
    /// safe against the abandoned store (`CASCADE`).
    #[tokio::test(flavor = "multi_thread")]
    async fn opens_are_isolated_and_teardown_is_idempotent() {
        if !super::harness_ready().await {
            return;
        }
        let factory = PgFactory::new("factory-shape");

        let a = factory.open().await.unwrap();
        let b = factory.open().await.unwrap();

        // Isolation: a row `a` publishes lands under `a`'s schema alone
        // — `b`'s read of the same stream name sees nothing (each
        // store owns its schema's tables; parallel property runs never
        // observe one another).
        let sa = a.stream("iso").await.unwrap();
        let event_offset = sa.publish(serde_json::json!({"iso": true})).await.unwrap();
        assert!(event_offset > 0);
        let sb = b.stream("iso").await.unwrap();
        let seen = sb.read_since(0, 100).await.unwrap();
        assert!(
            seen.is_empty(),
            "two opens must never share rows — b saw {seen:?}"
        );
        drop(sa);
        drop(sb);
        drop(a);
        drop(b);

        let schemas: Vec<String> = {
            let guard = factory.schemas.lock().expect("schema registry poisoned");
            guard.clone()
        };
        factory.teardown().await.unwrap();
        factory.teardown().await.unwrap();

        // Idempotent teardown: both opens' schemas are gone server-side.
        let client = super::admin_connect(&factory.dsn).await.unwrap();
        for schema in schemas {
            let count: i64 = client
                .query_one(
                    "SELECT count(*) FROM pg_namespace WHERE nspname = $1",
                    &[&schema],
                )
                .await
                .unwrap()
                .get(0);
            assert_eq!(count, 0, "teardown dropped the schema {schema}");
        }
    }
}