//! Docker-gated read tests for object grants + ownership.
//!
//! Each test spins up an ephemeral Postgres container, sets up objects with
//! explicit ownership and grants, reads the catalog back, and asserts that
//! `owner` and `grants` are populated correctly.
//!
//! Skipped when Docker is unavailable or `PGEVOLVE_DISABLE_DOCKER_TESTS` is
//! set. Pattern mirrors `catalog_round_trip.rs`.

use anyhow::{Context, Result, anyhow};
use pgevolve_core::catalog::{CatalogFilter, PgVersion, read_catalog};
use pgevolve_core::identifier::Identifier;
use pgevolve_core::ir::default_privileges::DefaultPrivObjectType;
use pgevolve_core::ir::grant::{GrantTarget, Privilege};
use pgevolve_testkit::ephemeral_pg::{EphemeralPostgres, docker_available};
use pgevolve_testkit::pg_querier::PgCatalogQuerier;

// ---------------------------------------------------------------------------
// Helper
// ---------------------------------------------------------------------------

async fn read_catalog_from_sql(sql: &str) -> Result<pgevolve_core::ir::catalog::Catalog> {
    let pg = EphemeralPostgres::start(PgVersion::Pg17)
        .await
        .context("start ephemeral postgres")?;
    pg.exec_sql(sql).await.context("exec setup SQL")?;
    let client = pg.connect().await.context("connect")?;
    let querier = PgCatalogQuerier::new(client).map_err(|e| anyhow!(e))?;
    let managed = vec![Identifier::from_unquoted("app").map_err(|e| anyhow!(e))?];
    let filter = CatalogFilter::new(managed, vec![]).map_err(|e| anyhow!(e.to_string()))?;
    let (catalog, _drift) = tokio::task::spawn_blocking(move || read_catalog(&querier, &filter))
        .await?
        .map_err(|e| anyhow!(e.to_string()))?;
    catalog.canonicalize().map_err(|e| anyhow!(e.to_string()))
}

// ---------------------------------------------------------------------------
// Test 1: table owner + object-level grants + column-level grants
// ---------------------------------------------------------------------------

/// Verify that `owner` and `grants` (object-level + column-level) are
/// populated correctly for tables.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reads_table_grants_and_owner() {
    if !docker_available() {
        eprintln!("skipping reads_table_grants_and_owner: Docker not available");
        return;
    }

    let cat = read_catalog_from_sql(
        r"
        CREATE SCHEMA app;
        CREATE ROLE app_owner;
        CREATE ROLE readers;
        ALTER SCHEMA app OWNER TO app_owner;
        CREATE TABLE app.t (id bigint, name text);
        ALTER TABLE app.t OWNER TO app_owner;
        GRANT SELECT ON app.t TO readers;
        GRANT INSERT (name) ON app.t TO readers;
        ",
    )
    .await
    .expect("read catalog");

    let t = cat
        .tables
        .iter()
        .find(|t| t.qname.name.as_str() == "t")
        .expect("table app.t must appear in catalog");

    assert_eq!(
        t.owner.as_ref().map(Identifier::as_str),
        Some("app_owner"),
        "table owner must be app_owner"
    );

    // After stripping owner self-grants, exactly 2 grants remain:
    // an object-level SELECT and a column-level INSERT(name), both to `readers`.
    assert_eq!(
        t.grants.len(),
        2,
        "expected SELECT object-level + INSERT(name) column-level only; owner self-grants filtered; grants: {:#?}",
        t.grants
    );
    assert!(
        t.grants.iter().any(|g| g.privilege == Privilege::Select
            && g.columns.is_none()
            && matches!(&g.grantee, GrantTarget::Role(r) if r.as_str() == "readers")),
        "must have readers object-level SELECT grant; grants: {:#?}",
        t.grants
    );
    assert!(
        t.grants.iter().any(|g| g.privilege == Privilege::Insert
            && g.columns.is_some()
            && matches!(&g.grantee, GrantTarget::Role(r) if r.as_str() == "readers")),
        "must have readers column-level INSERT(name) grant; grants: {:#?}",
        t.grants
    );
}

// ---------------------------------------------------------------------------
// Test 2: schema owner + grants (named role + PUBLIC)
// ---------------------------------------------------------------------------

/// Verify schema `owner` and `grants` are correctly populated.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reads_schema_grants_and_owner() {
    if !docker_available() {
        eprintln!("skipping reads_schema_grants_and_owner: Docker not available");
        return;
    }

    let cat = read_catalog_from_sql(
        r"
        CREATE ROLE app_owner;
        CREATE ROLE readers;
        CREATE SCHEMA app;
        ALTER SCHEMA app OWNER TO app_owner;
        GRANT USAGE ON SCHEMA app TO readers;
        GRANT USAGE, CREATE ON SCHEMA app TO PUBLIC;
        ",
    )
    .await
    .expect("read catalog");

    let s = cat
        .schemas
        .iter()
        .find(|s| s.name.as_str() == "app")
        .expect("schema app must appear in catalog");

    assert_eq!(
        s.owner.as_ref().map(Identifier::as_str),
        Some("app_owner"),
        "schema owner must be app_owner"
    );

    // 1 grant to readers (USAGE) + 2 to PUBLIC (USAGE, CREATE) = 3 after canon.
    assert!(
        s.grants
            .iter()
            .any(|g| g.privilege == Privilege::Usage && matches!(g.grantee, GrantTarget::Public)),
        "must have PUBLIC USAGE grant on schema"
    );
    assert!(
        s.grants
            .iter()
            .any(|g| g.privilege == Privilege::Create && matches!(g.grantee, GrantTarget::Public)),
        "must have PUBLIC CREATE grant on schema"
    );
    assert!(
        s.grants.iter().any(|g| g.privilege == Privilege::Usage
            && matches!(&g.grantee, GrantTarget::Role(r) if r.as_str() == "readers")),
        "must have readers USAGE grant on schema"
    );
}

// ---------------------------------------------------------------------------
// Test 3: sequence owner + grants
// ---------------------------------------------------------------------------

/// Verify sequence `owner` and `grants` are correctly populated.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reads_sequence_grants_and_owner() {
    if !docker_available() {
        eprintln!("skipping reads_sequence_grants_and_owner: Docker not available");
        return;
    }

    let cat = read_catalog_from_sql(
        r"
        CREATE SCHEMA app;
        CREATE ROLE app_owner;
        CREATE ROLE readers;
        CREATE SEQUENCE app.seq1;
        ALTER SEQUENCE app.seq1 OWNER TO app_owner;
        GRANT USAGE ON SEQUENCE app.seq1 TO readers;
        ",
    )
    .await
    .expect("read catalog");

    let seq = cat
        .sequences
        .iter()
        .find(|s| s.qname.name.as_str() == "seq1")
        .expect("sequence app.seq1 must appear in catalog");

    assert_eq!(
        seq.owner.as_ref().map(Identifier::as_str),
        Some("app_owner"),
        "sequence owner must be app_owner"
    );
    assert!(
        seq.grants.iter().any(|g| g.privilege == Privilege::Usage
            && matches!(&g.grantee, GrantTarget::Role(r) if r.as_str() == "readers")),
        "must have readers USAGE grant on sequence"
    );
}

// ---------------------------------------------------------------------------
// Test 4: function owner + grants
// ---------------------------------------------------------------------------

/// Verify function `owner` and `grants` are correctly populated.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reads_function_grants_and_owner() {
    if !docker_available() {
        eprintln!("skipping reads_function_grants_and_owner: Docker not available");
        return;
    }

    let cat = read_catalog_from_sql(
        r"
        CREATE SCHEMA app;
        CREATE ROLE app_owner;
        CREATE ROLE callers;
        CREATE FUNCTION app.add(a integer, b integer) RETURNS integer
            LANGUAGE sql AS $$ SELECT a + b $$;
        ALTER FUNCTION app.add(integer, integer) OWNER TO app_owner;
        GRANT EXECUTE ON FUNCTION app.add(integer, integer) TO callers;
        ",
    )
    .await
    .expect("read catalog");

    let f = cat
        .functions
        .iter()
        .find(|f| f.qname.name.as_str() == "add")
        .expect("function app.add must appear in catalog");

    assert_eq!(
        f.owner.as_ref().map(Identifier::as_str),
        Some("app_owner"),
        "function owner must be app_owner"
    );
    assert!(
        f.grants.iter().any(|g| g.privilege == Privilege::Execute
            && matches!(&g.grantee, GrantTarget::Role(r) if r.as_str() == "callers")),
        "must have callers EXECUTE grant on function"
    );
}

// ---------------------------------------------------------------------------
// Test 5: reads_default_privileges
// ---------------------------------------------------------------------------

/// Verify that `catalog.default_privileges` is populated from `pg_default_acl`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reads_default_privileges() {
    if !docker_available() {
        eprintln!("skipping reads_default_privileges: Docker not available");
        return;
    }

    let cat = read_catalog_from_sql(
        r"
        CREATE SCHEMA app;
        CREATE ROLE app_owner;
        CREATE ROLE readers;
        ALTER DEFAULT PRIVILEGES FOR ROLE app_owner IN SCHEMA app
            GRANT SELECT ON TABLES TO readers;
        ",
    )
    .await
    .expect("read catalog");

    let dp = &cat.default_privileges;
    assert_eq!(
        dp.len(),
        1,
        "expected exactly one default-privilege rule; got: {dp:#?}"
    );
    assert_eq!(dp[0].target_role.as_str(), "app_owner");
    assert_eq!(dp[0].schema.as_ref().map(Identifier::as_str), Some("app"));
    assert_eq!(dp[0].object_type, DefaultPrivObjectType::Tables);
    assert_eq!(
        dp[0].grants.len(),
        1,
        "expected exactly one grant in rule; got: {:#?}",
        dp[0].grants
    );
    assert_eq!(dp[0].grants[0].privilege, Privilege::Select);
    assert!(
        matches!(&dp[0].grants[0].grantee, GrantTarget::Role(r) if r.as_str() == "readers"),
        "grantee must be 'readers'; got: {:?}",
        dp[0].grants[0].grantee
    );
}

// ---------------------------------------------------------------------------
// Test 6: default_privileges_filters_pg_predefined_roles
// ---------------------------------------------------------------------------

/// Verify that predefined `pg_*` roles are excluded from `default_privileges`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn default_privileges_filters_pg_predefined_roles() {
    if !docker_available() {
        eprintln!("skipping default_privileges_filters_pg_predefined_roles: Docker not available");
        return;
    }

    let cat = read_catalog_from_sql("CREATE SCHEMA app;")
        .await
        .expect("read catalog");

    for rule in &cat.default_privileges {
        assert!(
            !rule.target_role.as_str().starts_with("pg_"),
            "predefined pg_* role must not appear in default_privileges: {:?}",
            rule.target_role
        );
    }
}
