fraiseql-cli 2.15.0

//! Doctor command — systematic diagnostic checks for common FraiseQL setup problems.
//!
//! Usage:
//!   fraiseql doctor
//!   fraiseql doctor --config fraiseql.toml --schema schema.compiled.json
//!   fraiseql doctor --json

use std::{net::TcpStream, path::Path, sync::Arc, time::Duration};

use fraiseql_core::{
    db::postgres::PostgresAdapter,
    runtime::{Executor, RuntimeConfig},
    schema::{ArgumentDefinition, CompiledSchema, MutationDefinition, QueryDefinition},
};
use fraiseql_observers::migrations::ENTITY_CHANGE_LOG_CONTRACT;
use serde::{Deserialize, Serialize};

use crate::{
    config::toml_schema::TomlSchema,
    schema::{
        database_validator::{create_introspector, validate_schema_against_database},
        mutation_contract::{ContractReport, Severity, validate_mutation_contract},
        pg_catalog::{
            CaptureFnSecurity, ChangeLogActorStats, ChangeLogRlsStatus, LiveColumn, PgCatalog,
            PlpgsqlCheckOutcome, PublicGrant, SecurityInvokerAudit,
        },
    },
};

// ─── Types ────────────────────────────────────────────────────────────────────

/// Outcome of a single diagnostic check.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
#[non_exhaustive]
pub enum CheckStatus {
    /// Check passed.
    Pass,
    /// Check produced a non-fatal warning.
    Warn,
    /// Check failed (fatal).
    Fail,
}

/// A single diagnostic check result.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DoctorCheck {
    /// Short display name shown in the report.
    pub name:   &'static str,
    /// Outcome of the check.
    pub status: CheckStatus,
    /// One-line detail appended after the name.
    pub detail: String,
    /// Optional actionable hint shown on the next line when status is not Pass.
    pub hint:   Option<String>,
}

impl DoctorCheck {
    pub(crate) fn pass(name: &'static str, detail: impl Into<String>) -> Self {
        Self {
            name,
            status: CheckStatus::Pass,
            detail: detail.into(),
            hint: None,
        }
    }

    pub(crate) fn warn(
        name: &'static str,
        detail: impl Into<String>,
        hint: impl Into<String>,
    ) -> Self {
        Self {
            name,
            status: CheckStatus::Warn,
            detail: detail.into(),
            hint: Some(hint.into()),
        }
    }

    pub(crate) fn fail(
        name: &'static str,
        detail: impl Into<String>,
        hint: impl Into<String>,
    ) -> Self {
        Self {
            name,
            status: CheckStatus::Fail,
            detail: detail.into(),
            hint: Some(hint.into()),
        }
    }
}

// ─── Individual checks ────────────────────────────────────────────────────────

/// Check that the compiled schema file exists and is readable.
pub fn check_schema_exists(path: &Path) -> DoctorCheck {
    if path.exists() {
        DoctorCheck::pass("Schema file exists", path.display().to_string())
    } else {
        DoctorCheck::fail(
            "Schema file exists",
            format!("not found: {}", path.display()),
            "Run `fraiseql compile fraiseql.toml` to generate schema.compiled.json",
        )
    }
}

/// Check that the compiled schema file is valid JSON.
pub fn check_schema_parses(path: &Path) -> DoctorCheck {
    match std::fs::read_to_string(path) {
        Err(e) => DoctorCheck::fail(
            "Schema parses",
            format!("cannot read: {e}"),
            "Check file permissions or run `fraiseql compile`",
        ),
        Ok(content) => match serde_json::from_str::<serde_json::Value>(&content) {
            Err(e) => DoctorCheck::fail(
                "Schema parses",
                format!("JSON parse error: {e}"),
                "Run `fraiseql compile fraiseql.toml` to regenerate the schema",
            ),
            Ok(schema) => {
                let types = schema.get("types").and_then(|v| v.as_array()).map_or(0, Vec::len);
                let queries = schema.get("queries").and_then(|v| v.as_array()).map_or(0, Vec::len);
                let mutations =
                    schema.get("mutations").and_then(|v| v.as_array()).map_or(0, Vec::len);
                DoctorCheck::pass(
                    "Schema parses",
                    format!("types={types}, queries={queries}, mutations={mutations}"),
                )
            },
        },
    }
}

/// Check that the compiled schema was produced by *this* build (#1304).
///
/// Since #1304 the server refuses any compiled schema its own build did not
/// produce, so this answers the question an operator actually has — would this
/// file boot here? — and a schema that cannot boot is reported as an error, not
/// as advice: `doctor` exits 1 on a failed check and 0 on a warning.
///
/// It read a top-level `version` key until #1320, and nothing has ever written
/// one — the compiled field was `schema_format_version` and is now
/// `fraiseql_version`. So every schema reported "no version field (older
/// schema)" and the remedy named the command that had just produced the file.
/// The three arms were pinned against a hand-written `{"version":1}` fixture no
/// compiler emits, which is how the suite stayed green over a check that never
/// once read a real artifact; the tests now serialize a `CompiledSchema`.
pub fn check_schema_version(path: &Path) -> DoctorCheck {
    const LABEL: &str = "Compiled schema build";

    let Ok(content) = std::fs::read_to_string(path) else {
        return DoctorCheck::warn(
            LABEL,
            "could not read schema file",
            "Ensure schema.compiled.json is readable",
        );
    };
    let Ok(schema) = serde_json::from_str::<serde_json::Value>(&content) else {
        return DoctorCheck::warn(
            LABEL,
            "schema is not valid JSON — build check skipped",
            "Run `fraiseql compile` to regenerate",
        );
    };

    let current = fraiseql_core::schema::CURRENT_FRAISEQL_VERSION;
    let recompile = format!("Recompile it with fraiseql-cli {current}");

    match schema.get("fraiseql_version").and_then(serde_json::Value::as_str) {
        Some(v) if v == current => {
            DoctorCheck::pass(LABEL, format!("fraiseql_version={v} (this build)"))
        },
        Some(v) => DoctorCheck::fail(
            LABEL,
            format!(
                "produced by fraiseql {v}, but this is fraiseql {current} — the server refuses it"
            ),
            recompile,
        ),
        None => DoctorCheck::fail(
            LABEL,
            format!(
                "no fraiseql_version stamp — the server refuses it (this is fraiseql {current})"
            ),
            recompile,
        ),
    }
}

/// Check whether `fraiseql.toml` exists.
pub fn check_toml_exists(path: &Path) -> DoctorCheck {
    if path.exists() {
        DoctorCheck::pass("fraiseql.toml found", path.display().to_string())
    } else {
        DoctorCheck::warn(
            "fraiseql.toml found",
            format!("not found: {} (using defaults)", path.display()),
            "Create fraiseql.toml with `fraiseql init` or provide --config",
        )
    }
}

/// The effective database URL for connectivity checks: `--db-url`, else
/// `--against-db`, else (inside the checks) the `DATABASE_URL` env var.
///
/// `--against-db` is the URL the live passes already connect with, so a run that
/// passes only `--against-db` must not report "DATABASE_URL not set" (#488).
#[must_use]
pub fn effective_db_url<'a>(
    db_url: Option<&'a str>,
    against_db: Option<&'a str>,
) -> Option<&'a str> {
    db_url.or(against_db)
}

/// Validate the `--config` file as TOML, choosing the right parse for its role.
///
/// When `--schema` is a compiled JSON, `--config` is a **runtime** config (server
/// settings), so only its *syntax* is validated — parsing it as a schema-definition
/// TOML would spuriously fail on the common Python-SDK + `--against-db` flow (#488).
/// When `--schema` is itself a TOML (or no compiled JSON is given), `--config` may
/// be the schema source, so the existing schema parse runs.
#[must_use]
pub fn check_config_toml(config_path: &Path, schema_path: &Path) -> DoctorCheck {
    if schema_is_compiled_json(schema_path) {
        check_toml_syntax(config_path)
    } else {
        check_toml_parses(config_path)
    }
}

/// Whether `--schema` points at a compiled-JSON schema (vs a `.toml` schema
/// source). Path-based by necessity: `run_checks` sees only paths, never schema
/// *content*, so there is no richer "schema came from TOML" signal to prefer. The
/// default `--schema schema.compiled.json` is covered by the `.json` extension.
fn schema_is_compiled_json(schema_path: &Path) -> bool {
    schema_path.extension().is_some_and(|e| e.eq_ignore_ascii_case("json"))
}

/// Syntax-only TOML parse — catches malformed TOML without requiring
/// schema-definition structure. Used for runtime-config `--config` files (#488).
fn check_toml_syntax(path: &Path) -> DoctorCheck {
    const NAME: &str = "config TOML syntax valid";
    let content = match std::fs::read_to_string(path) {
        Ok(c) => c,
        Err(e) => {
            return DoctorCheck::fail(NAME, format!("cannot read: {e}"), "Check file permissions");
        },
    };
    match toml::from_str::<toml::Value>(&content) {
        Ok(_) => DoctorCheck::pass(NAME, ""),
        Err(e) => {
            let short = e.to_string().lines().next().unwrap_or("parse error").to_string();
            DoctorCheck::fail(NAME, format!("parse error: {short}"), "Fix TOML syntax and retry")
        },
    }
}

/// Parse `fraiseql.toml`. Only called when the file actually exists.
pub fn check_toml_parses(path: &Path) -> DoctorCheck {
    let content = match std::fs::read_to_string(path) {
        Ok(c) => c,
        Err(e) => {
            return DoctorCheck::fail(
                "TOML syntax valid",
                format!("cannot read: {e}"),
                "Check file permissions",
            );
        },
    };
    match TomlSchema::parse_toml(&content) {
        Ok(_) => DoctorCheck::pass("TOML syntax valid", ""),
        Err(e) => {
            // Keep only the first line of the error to avoid overwhelming output.
            let first_line = e.to_string();
            let short = first_line.lines().next().unwrap_or("parse error");
            DoctorCheck::fail(
                "TOML syntax valid",
                format!("parse error: {short}"),
                "Fix TOML syntax in fraiseql.toml and retry",
            )
        },
    }
}

/// Check whether `DATABASE_URL` is set in the environment.
pub fn check_database_url_set(db_url_override: Option<&str>) -> DoctorCheck {
    let val = db_url_override
        .map(std::borrow::Cow::Borrowed)
        .or_else(|| std::env::var("DATABASE_URL").ok().map(std::borrow::Cow::Owned));
    if val.is_some() {
        DoctorCheck::pass("DATABASE_URL set", "")
    } else {
        DoctorCheck::fail(
            "DATABASE_URL set",
            "not set",
            "Set DATABASE_URL=postgres://user:pass@host:port/dbname in your environment",
        )
    }
}

/// Attempt a TCP connection to the database host:port extracted from the URL.
///
/// This does **not** run any SQL — it only validates that a TCP socket can be
/// opened within a 5-second timeout.
pub fn check_db_reachable(db_url_override: Option<&str>) -> DoctorCheck {
    let url_str = match db_url_override
        .map(std::borrow::Cow::Borrowed)
        .or_else(|| std::env::var("DATABASE_URL").ok().map(std::borrow::Cow::Owned))
    {
        Some(u) => u.into_owned(),
        None => {
            return DoctorCheck::fail(
                "DATABASE_URL reachable",
                "DATABASE_URL not set — cannot check connectivity",
                "Set DATABASE_URL first",
            );
        },
    };

    match parse_host_port(&url_str) {
        None => DoctorCheck::warn(
            "DATABASE_URL reachable",
            format!("could not parse host:port from URL: {url_str}"),
            "Ensure DATABASE_URL is a valid postgres:// or mysql:// URL",
        ),
        Some((host, port)) => {
            let addr = format!("{host}:{port}");
            match probe_tcp(&host, port) {
                Ok(()) => DoctorCheck::pass("DATABASE_URL reachable", addr),
                Err(ProbeError::Resolution(e)) => DoctorCheck::fail(
                    "DATABASE_URL reachable",
                    format!("host does not resolve ({addr}): {e}"),
                    format!("Check the hostname in DATABASE_URL — DNS lookup of '{host}' failed"),
                ),
                Err(ProbeError::Connect(e)) => DoctorCheck::fail(
                    "DATABASE_URL reachable",
                    format!("connection refused ({addr}): {e}"),
                    format!(
                        "Check that the database is running: pg_isready -h {host} -p {port}\n\
                         Or set DATABASE_URL=postgres://user:pass@host:port/dbname"
                    ),
                ),
            }
        },
    }
}

/// Why a TCP probe failed: the name never resolved, or every resolved address
/// refused the connection. The two need different operator guidance.
enum ProbeError {
    /// DNS/name resolution failed (or yielded no addresses).
    Resolution(String),
    /// The name resolved but no address accepted a connection.
    Connect(std::io::Error),
}

/// Resolve `host:port` through the system resolver (hostnames, IPv4 and IPv6
/// literals alike) and try a 5-second TCP connect to each resolved address.
///
/// `str::parse::<SocketAddr>` performs **no** name resolution, so the previous
/// implementation dialed a sentinel `0.0.0.0:0` for every hostname-based URL
/// and reported a healthy database as "connection refused" (#819). A sentinel
/// address is never substituted: resolution failure is reported as its own
/// failure mode.
fn probe_tcp(host: &str, port: u16) -> std::result::Result<(), ProbeError> {
    use std::net::ToSocketAddrs;

    let addrs: Vec<std::net::SocketAddr> = (host, port)
        .to_socket_addrs()
        .map_err(|e| ProbeError::Resolution(e.to_string()))?
        .collect();
    if addrs.is_empty() {
        return Err(ProbeError::Resolution("no addresses returned".to_string()));
    }

    let mut last_err: Option<std::io::Error> = None;
    for addr in addrs {
        match TcpStream::connect_timeout(&addr, Duration::from_secs(5)) {
            Ok(_) => return Ok(()),
            Err(e) => last_err = Some(e),
        }
    }
    // `addrs` is non-empty, so at least one connect ran and set `last_err`.
    Err(ProbeError::Connect(last_err.unwrap_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "no address accepted")
    })))
}

/// Check whether `FRAISEQL_JWT_SECRET` is set.
pub fn check_jwt_secret() -> DoctorCheck {
    if std::env::var("FRAISEQL_JWT_SECRET").is_ok() {
        DoctorCheck::pass("FRAISEQL_JWT_SECRET", "set")
    } else {
        DoctorCheck::warn(
            "FRAISEQL_JWT_SECRET",
            "not set (auth will reject all tokens)",
            "Set FRAISEQL_JWT_SECRET in your environment or .env file",
        )
    }
}

/// Check Redis if `REDIS_URL` is set.
pub fn check_redis_reachable() -> DoctorCheck {
    let Ok(url_str) = std::env::var("REDIS_URL") else {
        return DoctorCheck::pass("FRAISEQL_REDIS_URL", "not set (OK: cache disabled)");
    };

    match parse_host_port(&url_str) {
        None => DoctorCheck::warn(
            "FRAISEQL_REDIS_URL",
            format!("could not parse host:port from REDIS_URL: {url_str}"),
            "Ensure REDIS_URL is a valid redis:// URL",
        ),
        Some((host, port)) => {
            let addr = format!("{host}:{port}");
            match probe_tcp(&host, port) {
                Ok(()) => DoctorCheck::pass("FRAISEQL_REDIS_URL", format!("reachable ({addr})")),
                Err(ProbeError::Resolution(e)) => DoctorCheck::fail(
                    "FRAISEQL_REDIS_URL",
                    format!("host does not resolve ({addr}): {e}"),
                    "Check the hostname in REDIS_URL, or unset REDIS_URL to disable caching",
                ),
                Err(ProbeError::Connect(e)) => DoctorCheck::fail(
                    "FRAISEQL_REDIS_URL",
                    format!("set but not reachable ({addr}): {e}"),
                    "Check that Redis is running or unset REDIS_URL to disable caching",
                ),
            }
        },
    }
}

/// Check TLS: if the TOML config enables TLS, the cert file must exist.
///
/// Reads the config as a generic `toml::Value` rather than as a `TomlSchema`. That matters
/// because `TomlSchema` is `deny_unknown_fields` and carries no `[metrics]`/`[tracing]`-style
/// runtime tables, so any *runtime* server config failed the parse — the normal case for the
/// documented compiled-schema + runtime-config flow. This function then returned
/// `DoctorCheck::pass(…, "TOML unreadable — TLS check skipped")`, which `print_text_report`
/// renders as `[✓]` and `run_with_db_checks` counts as a pass. A missing certificate that
/// would abort server boot was reported as a green tick (#868 item 1).
///
/// A check that did not run is now a **warning**, never a pass. And because the generic parse
/// works for both the schema-TOML and runtime-config shapes, it usually does run.
pub fn check_tls(config_path: &Path) -> DoctorCheck {
    let Ok(content) = std::fs::read_to_string(config_path) else {
        return DoctorCheck::warn(
            "TLS certificate",
            format!("could not read {} — TLS not verified", config_path.display()),
            "Point --config at a readable config file, or remove the flag if TLS is not used",
        );
    };
    let Ok(value) = content.parse::<toml::Value>() else {
        return DoctorCheck::warn(
            "TLS certificate",
            "config is not valid TOML — TLS not verified",
            "Fix the TOML syntax so the TLS certificate check can run",
        );
    };

    let tls = value.get("server").and_then(|s| s.get("tls"));
    let enabled = tls
        .and_then(|t| t.get("enabled"))
        .and_then(toml::Value::as_bool)
        .unwrap_or(false);
    if !enabled {
        return DoctorCheck::pass("TLS certificate", "not configured (OK: TLS disabled)");
    }

    let cert = tls
        .and_then(|t| t.get("cert_file"))
        .and_then(toml::Value::as_str)
        .unwrap_or_default()
        .to_string();
    let cert = &cert;
    if cert.is_empty() {
        return DoctorCheck::fail(
            "TLS certificate",
            "TLS enabled but cert_file is empty",
            "Set [server.tls] cert_file and key_file in fraiseql.toml",
        );
    }
    if Path::new(cert).exists() {
        DoctorCheck::pass("TLS certificate", format!("found: {cert}"))
    } else {
        DoctorCheck::fail(
            "TLS certificate",
            format!("TLS enabled but cert_file not found: {cert}"),
            "Provide a valid PEM certificate at the configured path",
        )
    }
}

/// Cross-check: warn if caching is enabled without any authorization policy.
///
/// When caching is active but no authorization policies are configured, cached
/// results may be served to unauthenticated users — a potential data-leak.
/// Read as a generic `toml::Value` for the same reason as [`check_tls`], and warn rather than
/// pass when the check cannot run — this is one of the two security-relevant checks that used
/// to render as `[✓]` on the documented flow (#868 item 1).
pub fn check_rls_cache_coherence(config_path: &Path) -> DoctorCheck {
    let Ok(content) = std::fs::read_to_string(config_path) else {
        return DoctorCheck::warn(
            "Cache + auth coherence",
            format!("could not read {} — coherence not verified", config_path.display()),
            "Point --config at a readable config file",
        );
    };
    let Ok(value) = content.parse::<toml::Value>() else {
        return DoctorCheck::warn(
            "Cache + auth coherence",
            "config is not valid TOML — coherence not verified",
            "Fix the TOML syntax so the cache/auth coherence check can run",
        );
    };

    let caching_enabled = value
        .get("caching")
        .and_then(|c| c.get("enabled"))
        .and_then(toml::Value::as_bool)
        .unwrap_or(false);
    let security = value.get("security");
    // #983: keyed on the mechanisms that actually gate a cached read — database RLS,
    // and role definitions backing `requires_scope`. It used to accept the presence of
    // `[security] default_policy`, a key no enforcer read, so a schema with no access
    // control at all could pass this check by declaring one word.
    let has_rls = security
        .and_then(|s| s.get("rls"))
        .and_then(|r| r.get("enabled"))
        .and_then(toml::Value::as_bool)
        .unwrap_or(false);
    let has_roles = security
        .and_then(|s| s.get("role_definitions"))
        .and_then(toml::Value::as_array)
        .is_some_and(|a| !a.is_empty());
    let has_auth_policy = has_rls || has_roles;

    match (caching_enabled, has_auth_policy) {
        (false, _) => {
            DoctorCheck::pass("Cache + auth coherence", "cache disabled — no cross-user risk")
        },
        (true, true) => {
            DoctorCheck::pass("Cache + auth coherence", "caching + auth policy both configured")
        },
        (true, false) => DoctorCheck::warn(
            "Cache + auth coherence",
            "caching enabled without authorization policy — cached results may leak across users",
            "Declare [security.rls] enabled = true (with policies in the database), or \
             define [[security.role_definitions]] and mark fields `requires_scope`",
        ),
    }
}

// ─── Helpers ─────────────────────────────────────────────────────────────────

/// Extract (host, port) from a URL like `postgres://user:pass@host:5432/db`.
///
/// Returns `None` if the URL cannot be parsed.
pub(crate) fn parse_host_port(url: &str) -> Option<(String, u16)> {
    // Strip the scheme prefix and credentials; we only need the host:port.
    let after_scheme = url.split("://").nth(1)?;
    // Drop path/query after the first `/` following host:port.
    let host_part = after_scheme.split('/').next()?;
    // Drop user:pass@.
    let host_port = host_part.split('@').next_back()?;

    // Handle IPv6 addresses: [::1]:5432
    if host_port.starts_with('[') {
        let bracket_end = host_port.find(']')?;
        let host = host_port[1..bracket_end].to_string();
        let after_bracket = &host_port[bracket_end + 1..];
        let port = after_bracket.trim_start_matches(':').parse::<u16>().ok()?;
        return Some((host, port));
    }

    let mut parts = host_port.rsplitn(2, ':');
    let port = parts.next()?.parse::<u16>().ok()?;
    let host = parts.next().unwrap_or("localhost").to_string();
    Some((host, port))
}

// ─── Output ───────────────────────────────────────────────────────────────────

/// Print the doctor report in text format to stdout.
pub fn print_text_report(checks: &[DoctorCheck]) {
    println!("\nChecking FraiseQL setup...\n");

    for check in checks {
        let symbol = match check.status {
            CheckStatus::Pass => "✓",
            CheckStatus::Warn => "!",
            CheckStatus::Fail => "✗",
        };
        let detail = if check.detail.is_empty() {
            String::new()
        } else {
            format!("    {}", check.detail)
        };
        println!("  [{symbol}] {:<30}{detail}", check.name);
        if let Some(hint) = &check.hint {
            for line in hint.lines() {
                println!("       → {line}");
            }
        }
    }

    let errors = checks.iter().filter(|c| c.status == CheckStatus::Fail).count();
    let warnings = checks.iter().filter(|c| c.status == CheckStatus::Warn).count();

    println!();
    match (errors, warnings) {
        (0, 0) => println!("All checks passed."),
        (0, w) => println!("Summary: 0 errors, {w} warning(s)"),
        (e, 0) => println!("Summary: {e} error(s), 0 warnings"),
        (e, w) => println!("Summary: {e} error(s), {w} warning(s)"),
    }
}

/// Print the doctor report as JSON to stdout.
pub fn print_json_report(checks: &[DoctorCheck]) {
    let json = serde_json::to_string_pretty(checks).unwrap_or_else(|_| "[]".to_string());
    println!("{json}");
}

// ─── Entry point ──────────────────────────────────────────────────────────────

/// Run all doctor checks and return the list of results.
pub fn run_checks(
    config_path: &Path,
    schema_path: &Path,
    db_url_override: Option<&str>,
) -> Vec<DoctorCheck> {
    let mut checks = Vec::new();

    // Schema checks
    checks.push(check_schema_exists(schema_path));
    if schema_path.exists() {
        checks.push(check_schema_parses(schema_path));
        checks.push(check_schema_version(schema_path));
    }

    // TOML config checks. When `--schema` is a compiled JSON, `--config` is a
    // *runtime* config (not a schema definition), so only its syntax is checked —
    // schema-parsing it would spuriously fail (#488).
    checks.push(check_toml_exists(config_path));
    if config_path.exists() {
        checks.push(check_config_toml(config_path, schema_path));
    }

    // Environment / connectivity checks
    checks.push(check_database_url_set(db_url_override));
    checks.push(check_db_reachable(db_url_override));
    checks.push(check_jwt_secret());
    checks.push(check_redis_reachable());

    // TLS and coherence checks (only meaningful when config is present)
    checks.push(check_tls(config_path));
    checks.push(check_rls_cache_coherence(config_path));

    checks
}

/// Execute the doctor command.
///
/// Returns `true` if all checks passed (exit 0), `false` if any check failed
/// (exit 1). Warnings do not trigger an exit-1.
/// Execute the doctor command including the optional `--against-db` live-DB
/// passes: the change-log contract drift check (#380), the PL/pgSQL
/// body-resolution pass (#409), and the mutation→function contract drift check
/// (#384).
///
/// Runs the standard checks, then — when `against_db` is set — appends the
/// change-log contract/RLS/grant/capture reports, the internal-call resolution
/// results for `schemas`, and the mutation→function contract drift report.
/// Returns `true` if all checks passed (exit 0), `false` if any failed (exit 1).
pub async fn run_with_db_checks(
    config: &Path,
    schema: &Path,
    db_url: Option<&str>,
    against_db: Option<&str>,
    runtime: bool,
    schemas: &[String],
    json: bool,
) -> bool {
    // Connectivity prefers the explicit URL the run already connected with, so a
    // `--against-db`-only run no longer reports "DATABASE_URL not set" (#488).
    // Precedence: --db-url > --against-db > env (the env fallback lives in the
    // checks themselves, for the no-flag case).
    let mut checks = run_checks(config, schema, effective_db_url(db_url, against_db));
    if let Some(url) = against_db {
        checks.extend(changelog_contract_checks(url).await);
        checks.extend(changelog_rls_checks(url).await);
        checks.extend(changelog_public_grants_checks(url).await);
        checks.extend(changelog_actor_checks(url).await);
        checks.extend(capture_fn_security_checks(url).await);
        checks.extend(body_resolution_checks(url, schemas).await);
        checks.extend(mutation_contract_checks(url, schema).await);
        checks.extend(view_drift_checks(url, schema).await);
        checks.extend(rls_security_invoker_checks(url, schema).await);
        checks.extend(pagination_index_checks(url, schema).await);
    }

    // Runtime smoke (#501): actually execute each probeable root operation. Prefers
    // the live `--against-db` URL, falling back to `--db-url`.
    if runtime {
        checks.extend(runtime_probe_checks(against_db.or(db_url), schema).await);
    }

    if json {
        print_json_report(&checks);
    } else {
        print_text_report(&checks);
    }

    checks.iter().all(|c| c.status != CheckStatus::Fail)
}

// ─── Pagination index advice (#1307) ────────────────────────────────────────────

const PAGINATION_INDEX_NAME: &str = "Pagination index";

/// Report the indexes that would remove a sort from every paginated read.
///
/// Since #1287 a client sort over a non-unique key is tie-broken by the entity
/// identity, and where the sort key is indexed without the tie-break the plan
/// degrades to an incremental sort. The author is the only one who can create the
/// index, and nothing told them. Measured on 200k rows, `ORDER BY status, pk`:
/// separate indexes give `Incremental Sort`, the composite gives `Index Scan`.
///
/// Warnings, never failures: the query is correct either way, and `doctor` exits
/// non-zero only on a failed check — performance advice must not break a pipeline.
///
/// Indexes are read from the view's **base** relation, resolved through
/// `pg_rewrite`: a view carries no indexes of its own, so introspecting the
/// `sql_source` directly could only ever report "no index".
async fn pagination_index_checks(db_url: &str, schema_path: &Path) -> Vec<DoctorCheck> {
    use crate::commands::pagination_index_advice::{Advice, advise};

    let schema = match std::fs::read_to_string(schema_path)
        .map_err(|e| format!("cannot read schema: {e}"))
        .and_then(|c| {
            serde_json::from_str::<CompiledSchema>(&c)
                .map_err(|e| format!("schema parse error: {e}"))
        }) {
        Ok(s) => s,
        Err(detail) => {
            return vec![DoctorCheck::warn(
                PAGINATION_INDEX_NAME,
                format!("skipped — {detail}"),
                "Run `fraiseql compile` to produce a valid schema.compiled.json",
            )];
        },
    };

    let introspector = match crate::commands::compile::build_postgres_introspector(db_url) {
        Ok(i) => i,
        Err(e) => {
            return vec![DoctorCheck::fail(
                PAGINATION_INDEX_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    let mut checks = Vec::new();
    let mut examined = 0_usize;

    for query in &schema.queries {
        let (Some(view), true) = (query.sql_source.as_deref(), query.pagination_order.is_some())
        else {
            continue;
        };

        let bases = match introspector.resolve_base_relations(view).await {
            Ok(b) => b,
            Err(e) => {
                checks.push(DoctorCheck::warn(
                    PAGINATION_INDEX_NAME,
                    format!("{}: could not resolve '{view}' — {e}", query.name),
                    "Check that the view exists and the role can read the catalog",
                ));
                continue;
            },
        };

        // A join has no single relation to index, and guessing one would produce a
        // `CREATE INDEX` against a table that may not carry the sort key at all.
        let [base] = bases.as_slice() else {
            if bases.len() > 1 {
                checks.push(DoctorCheck::warn(
                    PAGINATION_INDEX_NAME,
                    format!(
                        "{}: '{view}' reads {} relations — cannot attribute the page order to one",
                        query.name,
                        bases.len()
                    ),
                    "Index the relation the ORDER BY resolves against, by hand",
                ));
            }
            continue;
        };

        let indexes = match introspector.get_index_definitions(base).await {
            Ok(i) => i,
            Err(e) => {
                checks.push(DoctorCheck::warn(
                    PAGINATION_INDEX_NAME,
                    format!("{}: could not read indexes on '{base}' — {e}", query.name),
                    "Check that the role can read the catalog",
                ));
                continue;
            },
        };

        examined += 1;
        for found in advise(query, base, &indexes) {
            let detail = match &found.advice {
                Advice::TieBreakUnindexed { tie_break } => format!(
                    "{}: nothing on '{base}' leads with {tie_break}, so every page sorts the whole relation",
                    query.name
                ),
                Advice::MissingComposite {
                    sort_keys,
                    tie_break,
                } => format!(
                    "{}: {} indexed on '{base}' without {tie_break} following — an incremental sort per page",
                    query.name,
                    sort_keys.join(", ")
                ),
            };
            checks.push(DoctorCheck::warn(PAGINATION_INDEX_NAME, detail, found.ddl.join(" ")));
        }
    }

    if checks.is_empty() {
        let subject = if examined == 1 {
            "query is"
        } else {
            "queries are"
        };
        checks.push(DoctorCheck::pass(
            PAGINATION_INDEX_NAME,
            format!("{examined} paginated {subject} ordered by an index"),
        ));
    }
    checks
}

// ─── Runtime smoke (#501) ───────────────────────────────────────────────────────

const RUNTIME_NAME: &str = "Runtime smoke";
const RUNTIME_QUERY_LABEL: &str = "Runtime query probe";
const RUNTIME_MUTATION_LABEL: &str = "Runtime mutation probe (dry-run)";

/// Boot the compiled schema in-process and probe each root operation end-to-end.
///
/// For every root query (and every mutation, dry-run / rolled back) that takes no
/// required arguments, a minimal operation is executed against the database; a
/// success becomes a `Pass`, a resolution error a `Fail`. Operations that require
/// arguments cannot be probed automatically (input would have to be fabricated)
/// and are reported as `Warn` (skipped) — `Warn` never fails the overall run.
///
/// Never panics: a missing URL, an unreadable schema, or a connection failure all
/// become a single `Fail` check.
pub async fn runtime_probe_checks(db_url: Option<&str>, schema_path: &Path) -> Vec<DoctorCheck> {
    let Some(url) = db_url else {
        return vec![DoctorCheck::fail(
            RUNTIME_NAME,
            "no database URL provided for --runtime",
            "Pass --against-db <postgres-url> (or --db-url) together with --runtime.",
        )];
    };
    if !(url.starts_with("postgres://") || url.starts_with("postgresql://")) {
        return vec![DoctorCheck::fail(
            RUNTIME_NAME,
            "runtime smoke supports PostgreSQL only",
            "Pass a postgres:// URL to --against-db.",
        )];
    }

    let schema = match load_compiled_schema(schema_path) {
        Ok(schema) => schema,
        Err(detail) => {
            return vec![DoctorCheck::fail(
                RUNTIME_NAME,
                detail,
                "Run `fraiseql compile` to (re)generate the compiled schema.",
            )];
        },
    };

    let adapter = match PostgresAdapter::new(url).await {
        Ok(adapter) => Arc::new(adapter),
        Err(e) => {
            return vec![DoctorCheck::fail(
                RUNTIME_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db.",
            )];
        },
    };

    // dry_run_mutations: mutation probes run inside a rolled-back transaction so a
    // smoke check never commits side effects.
    let config = RuntimeConfig {
        dry_run_mutations: true,
        ..RuntimeConfig::default()
    };
    // Clone for iteration so the borrow is independent of the executor that owns
    // the schema (one-shot CLI; the clone cost is negligible).
    let probe_schema = schema.clone();
    let executor = Executor::with_config(schema, adapter, config);

    let mut checks = Vec::new();
    for query in &probe_schema.queries {
        if let Some(op) = minimal_query_probe(query, &probe_schema) {
            let result = executor.execute(&op, None).await;
            checks.push(probe_outcome(RUNTIME_QUERY_LABEL, &query.name, &result));
        } else {
            checks.push(DoctorCheck::warn(
                RUNTIME_QUERY_LABEL,
                format!("{} skipped — requires arguments", query.name),
                format!("Probe manually: fraiseql query '{{ {}(…) {{ … }} }}'", query.name),
            ));
        }
    }
    for mutation in &probe_schema.mutations {
        if let Some(op) = minimal_mutation_probe(mutation, &probe_schema) {
            let result = executor.execute(&op, None).await;
            checks.push(probe_outcome(RUNTIME_MUTATION_LABEL, &mutation.name, &result));
        } else {
            checks.push(DoctorCheck::warn(
                RUNTIME_MUTATION_LABEL,
                format!("{} skipped — requires arguments", mutation.name),
                format!(
                    "Dry-run manually: fraiseql query --dry-run 'mutation {{ {}(…) {{ … }} }}'",
                    mutation.name
                ),
            ));
        }
    }

    if checks.is_empty() {
        checks.push(DoctorCheck::warn(
            RUNTIME_NAME,
            "no root queries or mutations to probe",
            "Compile a schema that declares queries or mutations.",
        ));
    }
    checks
}

/// Read and parse a compiled schema, mapping any failure to a display string.
fn load_compiled_schema(path: &Path) -> Result<CompiledSchema, String> {
    let text = std::fs::read_to_string(path)
        .map_err(|e| format!("cannot read {}: {e}", path.display()))?;
    CompiledSchema::from_json(&text, false)
        .map_err(|e| format!("cannot parse {}: {e}", path.display()))
}

/// Map a single probe execution result to a doctor check.
fn probe_outcome(
    label: &'static str,
    op_name: &str,
    result: &fraiseql_core::error::Result<serde_json::Value>,
) -> DoctorCheck {
    match result {
        Ok(value) if !crate::commands::query::has_errors(value) => {
            DoctorCheck::pass(label, format!("{op_name} resolves"))
        },
        Ok(value) => DoctorCheck::fail(
            label,
            format!(
                "{op_name}: {}",
                first_error_message(value).unwrap_or_else(|| "resolution error".to_string())
            ),
            "Inspect with `fraiseql query`.",
        ),
        Err(e) => {
            DoctorCheck::fail(label, format!("{op_name}: {e}"), "Inspect with `fraiseql query`.")
        },
    }
}

/// Extract the first GraphQL error message from a response envelope, if any.
fn first_error_message(value: &serde_json::Value) -> Option<String> {
    value
        .get("errors")?
        .as_array()?
        .first()?
        .get("message")?
        .as_str()
        .map(ToString::to_string)
}

/// Whether any declared argument is required (non-null and has no default), in
/// which case the operation cannot be probed without fabricating input.
pub(crate) fn probe_requires_arguments(args: &[ArgumentDefinition]) -> bool {
    args.iter().any(|a| !a.nullable && a.default_value.is_none())
}

/// Build a minimal sub-selection for a composite return type: the first scalar
/// field, falling back to `__typename`. Returns `None` when `type_name` is a leaf
/// (scalar / enum) that needs no sub-selection.
///
/// The three branches mirror `requires_selection_set` in
/// `fraiseql-core`'s `selection_validation` — object, union, unknown — because a
/// probe document this builds is adjudicated by that rule (§ 5.3.3, #1357). They
/// disagreed for exactly one shape: a mutation returning a **union** is absent from
/// `schema.types` (unions live in `schema.unions`), so this emitted a bare
/// `mutation { createUser }` for it, which the engine now refuses. Union payloads
/// are the ordinary shape for a mutation that reports success-or-error (#212,
/// #450/#451, #698), so doctor probed them into a validation failure rather than
/// the write it was checking for.
pub(crate) fn minimal_selection_for_type(
    type_name: &str,
    schema: &CompiledSchema,
) -> Option<String> {
    if let Some(type_def) = schema.types.iter().find(|t| t.name.as_str() == type_name) {
        let field = type_def.fields.iter().find(|f| f.field_type.is_scalar());
        return Some(
            field.map_or_else(|| "__typename".to_string(), |f| f.name.as_str().to_string()),
        );
    }
    // A union has no fields of its own to probe, but `__typename` is valid on every
    // composite selection set and is enough to make the document well-formed.
    if schema.unions.iter().any(|u| u.name.as_str() == type_name) {
        return Some("__typename".to_string());
    }
    // A leaf, or a type the compiled schema does not carry: no sub-selection is the
    // correct document, and the engine passes both for the same reason.
    None
}

/// Build a minimal probe document for a root query, or `None` if it requires args.
pub(crate) fn minimal_query_probe(
    query: &QueryDefinition,
    schema: &CompiledSchema,
) -> Option<String> {
    if probe_requires_arguments(&query.arguments) {
        return None;
    }
    Some(match minimal_selection_for_type(&query.return_type, schema) {
        Some(sel) => format!("{{ {} {{ {sel} }} }}", query.name),
        None => format!("{{ {} }}", query.name),
    })
}

/// Build a minimal probe document for a root mutation, or `None` if it requires args.
pub(crate) fn minimal_mutation_probe(
    mutation: &MutationDefinition,
    schema: &CompiledSchema,
) -> Option<String> {
    if probe_requires_arguments(&mutation.arguments) {
        return None;
    }
    Some(match minimal_selection_for_type(&mutation.return_type, schema) {
        Some(sel) => format!("mutation {{ {} {{ {sel} }} }}", mutation.name),
        None => format!("mutation {{ {} }}", mutation.name),
    })
}

/// Run the PL/pgSQL body-resolution pass and map its outcome to doctor checks.
///
/// Never panics: connection or analysis failures become `Fail` checks, an
/// absent `plpgsql_check` extension becomes a `Warn` (skipped), and each
/// unresolved internal call becomes its own `Fail`.
async fn body_resolution_checks(db_url: &str, schemas: &[String]) -> Vec<DoctorCheck> {
    const NAME: &str = "PL/pgSQL body resolution";

    let catalog = match PgCatalog::connect(db_url) {
        Ok(c) => c,
        Err(e) => {
            return vec![DoctorCheck::fail(
                NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    match catalog.plpgsql_check_unresolved_calls(schemas).await {
        Err(e) => vec![DoctorCheck::fail(
            NAME,
            format!("pass failed: {e}"),
            "Ensure the connecting role may CREATE EXTENSION plpgsql_check",
        )],
        Ok(PlpgsqlCheckOutcome::Unavailable) => vec![DoctorCheck::warn(
            NAME,
            "plpgsql_check not installed — skipped",
            "Install the plpgsql_check extension to enable internal-call checking",
        )],
        Ok(PlpgsqlCheckOutcome::Ran { errors }) if errors.is_empty() => vec![DoctorCheck::pass(
            NAME,
            format!("all internal calls resolve in {}", schemas.join(", ")),
        )],
        Ok(PlpgsqlCheckOutcome::Ran { errors }) => errors
            .iter()
            .map(|e| {
                let location = e.lineno.map_or_else(String::new, |l| format!(" (line {l})"));
                DoctorCheck::fail(
                    NAME,
                    format!("{}{location}: {}", e.caller, e.message),
                    "A migration changed a function signature — update its internal callers",
                )
            })
            .collect(),
    }
}

const MUTATION_CONTRACT_NAME: &str = "Mutation→function contract";

/// Convert a mutation-contract [`ContractReport`] into doctor checks (#384).
///
/// A clean report (no violations) yields a single `Pass` summarising how many
/// DB-backed mutations resolved. Otherwise each violation becomes its own
/// `Fail`/`Warn` naming the mutation and its `sql_source`, so a declared
/// mutation with a missing or mismatched backing function — the most common
/// "failed to prepare" root cause — is surfaced at check time, not first call.
///
/// Pure (no I/O) so it is unit-tested without a database.
pub(crate) fn mutation_contract_drift(report: &ContractReport) -> Vec<DoctorCheck> {
    if report.mutations.is_empty() {
        return vec![DoctorCheck::pass(
            MUTATION_CONTRACT_NAME,
            format!(
                "all {} DB-backed mutation(s) resolve to a matching function ({} skipped)",
                report.checked, report.skipped
            ),
        )];
    }
    report
        .mutations
        .iter()
        .flat_map(|m| {
            m.violations.iter().map(move |v| {
                let detail = format!("{} (sql_source: {}): {v}", m.mutation, m.sql_source);
                match v.severity() {
                    Severity::Error => DoctorCheck::fail(
                        MUTATION_CONTRACT_NAME,
                        detail,
                        "Declared mutation does not match its backing function — create or fix \
                         the function, or correct the schema",
                    ),
                    Severity::Warn => DoctorCheck::warn(
                        MUTATION_CONTRACT_NAME,
                        detail,
                        "The backing function differs from the declaration (non-fatal)",
                    ),
                }
            })
        })
        .collect()
}

/// Run the mutation→function contract drift check against a live database (#384).
///
/// Loads the compiled schema, then for every DB-backed mutation verifies its
/// `sql_source` resolves to exactly one function whose argument arity/shape and
/// `mutation_response` return columns match what the runtime will send and
/// decode — the same static check as `fraiseql validate --against-db`, surfaced
/// in `doctor` so it is the single drift report.
///
/// Never panics: a schema-load, connection, or catalog failure becomes a single
/// check rather than aborting the doctor run.
async fn mutation_contract_checks(db_url: &str, schema_path: &Path) -> Vec<DoctorCheck> {
    let schema = match std::fs::read_to_string(schema_path)
        .map_err(|e| format!("cannot read schema: {e}"))
        .and_then(|c| {
            serde_json::from_str::<CompiledSchema>(&c)
                .map_err(|e| format!("schema parse error: {e}"))
        }) {
        Ok(s) => s,
        Err(detail) => {
            return vec![DoctorCheck::warn(
                MUTATION_CONTRACT_NAME,
                format!("skipped — {detail}"),
                "Run `fraiseql compile` to produce a valid schema.compiled.json",
            )];
        },
    };

    let catalog = match PgCatalog::connect(db_url) {
        Ok(c) => c,
        Err(e) => {
            return vec![DoctorCheck::fail(
                MUTATION_CONTRACT_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    match validate_mutation_contract(&schema, &catalog).await {
        Ok(report) => mutation_contract_drift(&report),
        Err(e) => vec![DoctorCheck::fail(
            MUTATION_CONTRACT_NAME,
            format!("contract check failed: {e}"),
            "Ensure the connecting role can read pg_catalog for the function schema",
        )],
    }
}

const VIEW_DRIFT_NAME: &str = "View drift";

/// Run the view-composition linter against a live database (#384: L1
/// `sql_source` relation existence, L2 column shape, L3 JSONB key sampling).
///
/// The same pass `compile --database` runs, surfaced as structured checks so
/// CI can consume the report via `--json`: an error-severity finding becomes a
/// `Fail`, a warn-severity finding a `Warn`, a clean report a single `Pass`.
/// A connection or introspection failure becomes a single `Fail` check (never
/// panics).
async fn view_drift_checks(db_url: &str, schema_path: &Path) -> Vec<DoctorCheck> {
    let schema = match std::fs::read_to_string(schema_path)
        .map_err(|e| format!("cannot read schema: {e}"))
        .and_then(|c| {
            serde_json::from_str::<CompiledSchema>(&c)
                .map_err(|e| format!("schema parse error: {e}"))
        }) {
        Ok(s) => s,
        Err(detail) => {
            return vec![DoctorCheck::warn(
                VIEW_DRIFT_NAME,
                format!("skipped — {detail}"),
                "Run `fraiseql compile` to produce a valid schema.compiled.json",
            )];
        },
    };

    let introspector = match create_introspector(db_url).await {
        Ok(i) => i,
        Err(e) => {
            return vec![DoctorCheck::fail(
                VIEW_DRIFT_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    match validate_schema_against_database(&schema, &introspector).await {
        Ok(report) if report.warnings.is_empty() => {
            vec![DoctorCheck::pass(
                VIEW_DRIFT_NAME,
                "no schema↔database drift found",
            )]
        },
        Ok(report) => report
            .warnings
            .iter()
            .map(|w| match w.severity() {
                Severity::Error => DoctorCheck::fail(
                    VIEW_DRIFT_NAME,
                    w.to_string(),
                    "Fix the declaration or the database object it names",
                ),
                Severity::Warn => DoctorCheck::warn(
                    VIEW_DRIFT_NAME,
                    w.to_string(),
                    "Advisory — see `compile --database` for the same finding at build time",
                ),
            })
            .collect(),
        Err(e) => vec![DoctorCheck::fail(
            VIEW_DRIFT_NAME,
            format!("view-drift check failed: {e}"),
            "Ensure the connecting role can read the relation catalogs",
        )],
    }
}

const CHANGELOG_CONTRACT_NAME: &str = "Change-log contract";

/// Run the change-log contract drift check against a live database (#380).
///
/// Connects, reads `core.tb_entity_change_log` from `information_schema.columns`,
/// and classifies the drift via [`changelog_contract_drift`]. A connection or
/// introspection failure becomes a single `Fail` check (never panics).
async fn changelog_contract_checks(db_url: &str) -> Vec<DoctorCheck> {
    let catalog = match PgCatalog::connect(db_url) {
        Ok(c) => c,
        Err(e) => {
            return vec![DoctorCheck::fail(
                CHANGELOG_CONTRACT_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    match catalog.table_columns("core", "tb_entity_change_log").await {
        Ok(live) => changelog_contract_drift(&live),
        Err(e) => vec![DoctorCheck::fail(
            CHANGELOG_CONTRACT_NAME,
            format!("introspection failed: {e}"),
            "Ensure the connecting role can read information_schema for the core schema",
        )],
    }
}

const CHANGELOG_RLS_NAME: &str = "Change-log RLS";

/// Run the change-log RLS posture check against a live database (#437 F6 / #443).
///
/// Connects, reads whether RLS is enabled on `core.tb_entity_change_log` and
/// whether the connecting role can read it, and classifies via
/// [`changelog_rls_check`]. A connection or introspection failure becomes a single
/// `Fail` check (never panics).
async fn changelog_rls_checks(db_url: &str) -> Vec<DoctorCheck> {
    let catalog = match PgCatalog::connect(db_url) {
        Ok(c) => c,
        Err(e) => {
            return vec![DoctorCheck::fail(
                CHANGELOG_RLS_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    match catalog.change_log_rls_status().await {
        Ok(status) => vec![changelog_rls_check(status.as_ref())],
        Err(e) => vec![DoctorCheck::fail(
            CHANGELOG_RLS_NAME,
            format!("introspection failed: {e}"),
            "Ensure the connecting role can read pg_class and pg_roles",
        )],
    }
}

/// Classify the change-log RLS posture into a doctor check.
///
/// Pure — no database access — so the matrix is unit-tested without a connection.
/// The high-value case is the **silent-empty footgun**: RLS is enabled but the
/// connecting role neither owns the table nor has `BYPASSRLS`, so the change-log
/// poller, the NATS bridges, and the admin change-log query read **zero** rows
/// without any error. That is a `Warn`. An absent table or RLS-disabled posture is
/// an informational `Pass` (single-tenant or pre-migration-12 deployments).
pub(crate) fn changelog_rls_check(status: Option<&ChangeLogRlsStatus>) -> DoctorCheck {
    let Some(s) = status else {
        return DoctorCheck::pass(
            CHANGELOG_RLS_NAME,
            "core.tb_entity_change_log not present — skipped",
        );
    };
    if !s.rls_enabled {
        return DoctorCheck::pass(
            CHANGELOG_RLS_NAME,
            "Row-Level Security is not enabled on core.tb_entity_change_log (single-tenant or \
             pre-migration-12). Apply 12_enable_change_log_rls.sql to fail-close cross-tenant reads.",
        );
    }
    if s.can_bypass || s.is_owner {
        let how = if s.is_owner {
            "owns the table"
        } else {
            "has BYPASSRLS"
        };
        return DoctorCheck::pass(
            CHANGELOG_RLS_NAME,
            format!(
                "RLS enabled; the connecting role '{}' {how} — change-log consumers read all tenants.",
                s.role_name
            ),
        );
    }
    DoctorCheck::warn(
        CHANGELOG_RLS_NAME,
        format!(
            "RLS is enabled on core.tb_entity_change_log but the connecting role '{}' is neither \
             the table owner nor BYPASSRLS — the change-log poller, the NATS bridges, and the admin \
             change-log query will silently read ZERO rows.",
            s.role_name
        ),
        "Grant BYPASSRLS to this role (ALTER ROLE … BYPASSRLS) or connect the change-log consumers \
         as the table owner.",
    )
}

const CHANGELOG_ACTOR_NAME: &str = "Change-log actor attribution";

/// Run the change-log actor-attribution check against a live database (#390).
///
/// Connects, reads how the live `core.tb_entity_change_log` rows are attributed
/// (NULL / out-of-contract `actor_type` values, CHECK constraint presence), and
/// classifies via [`changelog_actor_check`]. A connection or introspection
/// failure becomes a single `Fail` check (never panics).
async fn changelog_actor_checks(db_url: &str) -> Vec<DoctorCheck> {
    let catalog = match PgCatalog::connect(db_url) {
        Ok(c) => c,
        Err(e) => {
            return vec![DoctorCheck::fail(
                CHANGELOG_ACTOR_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    let tokens: Vec<String> = fraiseql_core::security::ActorType::ALL
        .iter()
        .map(|a| a.as_str().to_string())
        .collect();
    match catalog.change_log_actor_stats(&tokens).await {
        Ok(stats) => changelog_actor_check(stats.as_ref()),
        Err(e) => vec![DoctorCheck::fail(
            CHANGELOG_ACTOR_NAME,
            format!("introspection failed: {e}"),
            "Ensure the connecting role can read core.tb_entity_change_log and pg_constraint",
        )],
    }
}

/// Classify the change-log actor-attribution posture into doctor checks (#390).
///
/// Pure — no database access — so the matrix is unit-tested without a connection.
/// Severity model:
/// - **out-of-contract `actor_type` values → `Fail`**: only a rogue writer (or a pre-constraint
///   one) can produce them, and every downstream per-actor forensic query silently mis-buckets
///   those rows;
/// - **missing CHECK constraint → `Warn`**: the domain is enforced only by application-layer
///   discipline until migration 08 is (re-)applied;
/// - **NULL `actor_type` rows → `Warn`**: pre-actor-model rows or writes made without an
///   authenticated security context — the actor model cannot attribute them retroactively;
/// - otherwise → `Pass`.
pub(crate) fn changelog_actor_check(stats: Option<&ChangeLogActorStats>) -> Vec<DoctorCheck> {
    let Some(s) = stats else {
        return vec![DoctorCheck::pass(
            CHANGELOG_ACTOR_NAME,
            "core.tb_entity_change_log not present — skipped",
        )];
    };

    let mut checks = Vec::new();

    if s.unknown_values.is_empty() {
        if !s.constraint_installed {
            checks.push(DoctorCheck::warn(
                CHANGELOG_ACTOR_NAME,
                "chk_entity_change_log_actor_type is not installed — an arbitrary string can \
                 reach actor_type",
                "Re-apply migration 08 (08_create_entity_change_log_contract.sql) or re-run \
                 `fraiseql setup`",
            ));
        }
    } else {
        checks.push(DoctorCheck::fail(
            CHANGELOG_ACTOR_NAME,
            format!(
                "out-of-contract actor_type values in core.tb_entity_change_log: {} — a rogue \
                 or pre-constraint writer{}",
                s.unknown_values.join(", "),
                if s.constraint_installed {
                    " (the CHECK constraint now blocks new ones)"
                } else {
                    "; chk_entity_change_log_actor_type is also missing, so new ones are NOT \
                     blocked"
                }
            ),
            "Identify the writer; valid tokens are human_user, service_account, ai_agent, \
             system_job. Re-apply migration 08 if the constraint is missing",
        ));
    }

    if s.null_rows > 0 {
        checks.push(DoctorCheck::warn(
            CHANGELOG_ACTOR_NAME,
            format!(
                "{} change-log row(s) carry no actor attribution (actor_type IS NULL) — \
                 pre-actor-model rows, or writes made without an authenticated security context",
                s.null_rows
            ),
            "New writes through the mutation runner are stamped from the SecurityContext; \
             unauthenticated deployments record NULL. Enable authentication to attribute every \
             action",
        ));
    }

    if checks.is_empty() {
        checks.push(DoctorCheck::pass(
            CHANGELOG_ACTOR_NAME,
            "every change-log row carries a canonical actor_type and the domain CHECK \
             constraint is installed",
        ));
    }
    checks
}

const CHANGELOG_PUBLIC_GRANTS_NAME: &str = "Change-log PUBLIC grants";

/// Run the change-log PUBLIC-grants check against a live database (#443).
///
/// Connects, reads which privileges `PUBLIC` holds on the change-log table and its
/// two views, and classifies via [`changelog_public_grants_check`]. A connection
/// or introspection failure becomes a single `Fail` check (never panics).
async fn changelog_public_grants_checks(db_url: &str) -> Vec<DoctorCheck> {
    let catalog = match PgCatalog::connect(db_url) {
        Ok(c) => c,
        Err(e) => {
            return vec![DoctorCheck::fail(
                CHANGELOG_PUBLIC_GRANTS_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    match catalog.change_log_public_grants().await {
        Ok(grants) => vec![changelog_public_grants_check(&grants)],
        Err(e) => vec![DoctorCheck::fail(
            CHANGELOG_PUBLIC_GRANTS_NAME,
            format!("introspection failed: {e}"),
            "Ensure the connecting role can read pg_class.relacl for the core schema",
        )],
    }
}

/// Classify the PUBLIC-grants posture of the change-log relations into a doctor
/// check (#443).
///
/// Pure — no database access — so the matrix is unit-tested without a connection.
/// `REVOKE ALL … FROM PUBLIC` (migration 12) keeps the change-log — every tenant's
/// before/after payload — off the world-readable `PUBLIC` pseudo-role, making
/// explicit grants the primary control and RLS defence-in-depth. A relation that
/// still grants any privilege to PUBLIC is the **world-readable footgun** → `Warn`.
/// An empty input (none of the three relations present) is an informational `Pass`
/// (single-tenant or pre-migration deployments).
pub(crate) fn changelog_public_grants_check(grants: &[PublicGrant]) -> DoctorCheck {
    if grants.is_empty() {
        return DoctorCheck::pass(
            CHANGELOG_PUBLIC_GRANTS_NAME,
            "change-log table/views not present — skipped",
        );
    }

    let exposed: Vec<String> = grants
        .iter()
        .filter(|g| !g.privileges.is_empty())
        .map(|g| {
            // array_agg order is unspecified; sort so the message is deterministic.
            let mut privs = g.privileges.clone();
            privs.sort();
            format!("core.{} ({})", g.relname, privs.join(", "))
        })
        .collect();

    if exposed.is_empty() {
        return DoctorCheck::pass(
            CHANGELOG_PUBLIC_GRANTS_NAME,
            "no PUBLIC privileges on the change-log table or views (least-privilege baseline \
             intact)",
        );
    }

    DoctorCheck::warn(
        CHANGELOG_PUBLIC_GRANTS_NAME,
        format!(
            "PUBLIC can still access the change-log — every tenant's before/after payload is \
             world-readable: {}",
            exposed.join("; ")
        ),
        "REVOKE ALL ON <relation> FROM PUBLIC (or re-run migration \
         12_enable_change_log_rls.sql) so only explicitly granted roles read the change-log.",
    )
}

const SECURITY_INVOKER_NAME: &str = "RLS view security_invoker";

/// Warn when any `sql_source` view lacks `security_invoker` while the database uses
/// RLS. A default view runs as the view owner and BYPASSES the caller's RLS, so
/// ordinary reads — and cascades assembled from that view — silently leak
/// cross-tenant rows. Turns that silent leak class into a diagnosable one.
///
/// A connection or introspection failure becomes a single `Fail` (never panics).
async fn rls_security_invoker_checks(db_url: &str, schema_path: &Path) -> Vec<DoctorCheck> {
    let schema = match load_compiled_schema(schema_path) {
        Ok(s) => s,
        Err(e) => {
            return vec![DoctorCheck::fail(
                SECURITY_INVOKER_NAME,
                format!("cannot load compiled schema: {e}"),
                "Compile the schema before running --against-db.",
            )];
        },
    };
    // Distinct bare view names backing entity types (strip any schema qualifier).
    let mut views: Vec<String> = schema
        .types
        .iter()
        .map(|t| t.sql_source.as_str())
        .filter(|s| !s.is_empty())
        .map(|s| s.rsplit('.').next().unwrap_or(s).to_string())
        .collect();
    views.sort();
    views.dedup();
    if views.is_empty() {
        return vec![DoctorCheck::pass(
            SECURITY_INVOKER_NAME,
            "no sql_source views to audit",
        )];
    }

    let catalog = match PgCatalog::connect(db_url) {
        Ok(c) => c,
        Err(e) => {
            return vec![DoctorCheck::fail(
                SECURITY_INVOKER_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db.",
            )];
        },
    };
    match catalog.security_invoker_audit(&views).await {
        Ok(audit) => vec![security_invoker_check(&audit)],
        Err(e) => vec![DoctorCheck::fail(
            SECURITY_INVOKER_NAME,
            format!("introspection failed: {e}"),
            "Ensure the connecting role can read pg_class and pg_policies.",
        )],
    }
}

/// Classify a [`SecurityInvokerAudit`] into a doctor check.
pub(crate) fn security_invoker_check(audit: &SecurityInvokerAudit) -> DoctorCheck {
    if !audit.rls_in_use {
        return DoctorCheck::pass(
            SECURITY_INVOKER_NAME,
            "No RLS policies in the database — security_invoker views are not required for \
             row isolation.",
        );
    }
    if audit.views_without_invoker.is_empty() {
        return DoctorCheck::pass(
            SECURITY_INVOKER_NAME,
            "RLS is in use and every sql_source view is security_invoker — base-table RLS is \
             honoured on reads and in cascades.",
        );
    }
    DoctorCheck::warn(
        SECURITY_INVOKER_NAME,
        format!(
            "RLS is in use, but these sql_source views are NOT security_invoker, so they run as \
             the view owner and BYPASS the caller's RLS — a cross-tenant leak on ordinary reads \
             AND in cascades: {}",
            audit.views_without_invoker.join(", ")
        ),
        "Recreate each as `CREATE VIEW <v> WITH (security_invoker = true) AS …` (or \
         `ALTER VIEW <v> SET (security_invoker = true)`), PostgreSQL 15+.",
    )
}

const CAPTURE_FN_SECURITY_NAME: &str = "Change-log capture function";

/// Run the capture-function security check against a live database (#443 /
/// #437 F6).
///
/// Connects, reads whether `core.fn_entity_change_log_capture()` is
/// `SECURITY DEFINER` with a pinned `search_path`, and classifies via
/// [`capture_fn_security_check`]. A connection or introspection failure becomes a
/// single `Fail` check (never panics).
async fn capture_fn_security_checks(db_url: &str) -> Vec<DoctorCheck> {
    let catalog = match PgCatalog::connect(db_url) {
        Ok(c) => c,
        Err(e) => {
            return vec![DoctorCheck::fail(
                CAPTURE_FN_SECURITY_NAME,
                format!("cannot connect: {e}"),
                "Pass a reachable postgres:// URL to --against-db",
            )];
        },
    };

    match catalog.capture_fn_security().await {
        Ok(status) => vec![capture_fn_security_check(status.as_ref())],
        Err(e) => vec![DoctorCheck::fail(
            CAPTURE_FN_SECURITY_NAME,
            format!("introspection failed: {e}"),
            "Ensure the connecting role can read pg_proc for the core schema",
        )],
    }
}

/// Classify the capture-function security posture into a doctor check (#443 /
/// #437 F6).
///
/// Pure — no database access — so the matrix is unit-tested without a connection.
/// `core.fn_entity_change_log_capture()` runs from triggers on arbitrary tables
/// and must be `SECURITY DEFINER` (so external-write capture runs as the table
/// owner, exempt under change-log RLS) **with a pinned `search_path`** (a DEFINER
/// function with a mutable `search_path` is a classic privilege-escalation
/// vector). An absent function is an informational `Pass` (single-tenant or
/// pre-migration-11). `SECURITY INVOKER`, or DEFINER without a pinned
/// `search_path`, is a `Warn`.
pub(crate) fn capture_fn_security_check(status: Option<&CaptureFnSecurity>) -> DoctorCheck {
    let Some(s) = status else {
        return DoctorCheck::pass(
            CAPTURE_FN_SECURITY_NAME,
            "core.fn_entity_change_log_capture not present — skipped",
        );
    };

    if !s.security_definer {
        return DoctorCheck::warn(
            CAPTURE_FN_SECURITY_NAME,
            "core.fn_entity_change_log_capture is SECURITY INVOKER — under change-log RLS, an \
             external write by a NOBYPASSRLS role runs the capture as that role instead of the \
             table owner, so the capture row is not written.",
            "Re-apply migration 11_create_change_log_capture_trigger.sql so the capture function \
             is SECURITY DEFINER with a pinned search_path.",
        );
    }

    match &s.search_path {
        None => DoctorCheck::warn(
            CAPTURE_FN_SECURITY_NAME,
            "core.fn_entity_change_log_capture is SECURITY DEFINER but has no pinned search_path — \
             a DEFINER function reachable from a trigger on any schema with a mutable search_path \
             is a privilege-escalation vector.",
            "ALTER FUNCTION core.fn_entity_change_log_capture() SET search_path = pg_catalog, core \
             (or re-apply migration 11).",
        ),
        Some(sp) => DoctorCheck::pass(
            CAPTURE_FN_SECURITY_NAME,
            format!("SECURITY DEFINER with pinned search_path = {sp}"),
        ),
    }
}

/// Compare a live `core.tb_entity_change_log` against the shipped contract and
/// map the drift to doctor checks.
///
/// Pure — no database access — so the full classification matrix is unit-tested
/// without a connection. The expected column set is the single source of truth
/// in [`fraiseql_observers::migrations::ENTITY_CHANGE_LOG_CONTRACT`].
///
/// - **Type mismatch** on a pre-existing column → `Fail`. The contract migration is purely additive
///   (`ADD COLUMN IF NOT EXISTS`), so it no-ops on a column that already exists and **cannot retype
///   it** — e.g. a legacy `object_id text` the contract wants as `uuid` (this bit the #149 e2e). A
///   manual `ALTER COLUMN … TYPE` is required.
/// - **Missing** contract column → `Warn`: `fraiseql migrate up` adds it.
/// - **Extra** non-contract column → `Warn`: the migration leaves it untouched.
/// - Otherwise → `Pass`.
///
/// An empty `live` means the table is absent → a single `Warn` pointing at the
/// migration.
pub(crate) fn changelog_contract_drift(live: &[LiveColumn]) -> Vec<DoctorCheck> {
    if live.is_empty() {
        return vec![DoctorCheck::warn(
            CHANGELOG_CONTRACT_NAME,
            "core.tb_entity_change_log not found",
            "Run `fraiseql setup` (installs the change-log contract) or `fraiseql migrate up`. \
             Every default mutation writes this table via its transactional-outbox CTE, so \
             without it the first mutation fails at prepare time (#569).",
        )];
    }

    let mut checks = Vec::new();

    // Type drift on pre-existing columns — the additive migration cannot fix it.
    for contract in ENTITY_CHANGE_LOG_CONTRACT {
        let Some(col) = live.iter().find(|c| c.name == contract.name) else {
            continue;
        };
        if col.udt_name == contract.udt {
            continue;
        }
        checks.push(DoctorCheck::fail(
            CHANGELOG_CONTRACT_NAME,
            format!(
                "column `{}` is `{}`, contract expects `{}`",
                contract.name, col.udt_name, contract.udt
            ),
            format!(
                "The additive migration cannot retype an existing column; ALTER it manually, \
                 e.g. ALTER TABLE core.tb_entity_change_log ALTER COLUMN {name} TYPE {udt} \
                 USING {name}::{udt};",
                name = contract.name,
                udt = contract.udt,
            ),
        ));
    }

    // Missing contract columns — the migration adds them (additive).
    let missing: Vec<&str> = ENTITY_CHANGE_LOG_CONTRACT
        .iter()
        .map(|c| c.name)
        .filter(|name| !live.iter().any(|c| c.name == *name))
        .collect();
    if !missing.is_empty() {
        checks.push(DoctorCheck::warn(
            CHANGELOG_CONTRACT_NAME,
            format!("{} contract column(s) missing: {}", missing.len(), missing.join(", ")),
            "Run `fraiseql migrate up` — the additive migration adds these columns",
        ));
    }

    // Extra columns — present live, not in the contract; the migration keeps them.
    let extra: Vec<&str> = live
        .iter()
        .map(|c| c.name.as_str())
        .filter(|name| !ENTITY_CHANGE_LOG_CONTRACT.iter().any(|c| c.name == *name))
        .collect();
    if !extra.is_empty() {
        checks.push(DoctorCheck::warn(
            CHANGELOG_CONTRACT_NAME,
            format!("{} non-contract column(s) present: {}", extra.len(), extra.join(", ")),
            "Left untouched by the migration — app-specific columns are safe to keep",
        ));
    }

    if checks.is_empty() {
        checks.push(DoctorCheck::pass(
            CHANGELOG_CONTRACT_NAME,
            format!(
                "all {} contract columns present and correctly typed",
                ENTITY_CHANGE_LOG_CONTRACT.len()
            ),
        ));
    }

    checks
}