pgroles-inspect 0.8.0

//! Database introspection for pgroles.
//!
//! Queries `pg_catalog` tables to build a [`pgroles_core::model::RoleGraph`]
//! representing the current state of roles, grants, default privileges, and
//! memberships in a PostgreSQL database.

pub mod cloud;
mod defaults;
mod memberships;
mod privileges;
mod public_grants;
mod roles;
mod safety;
mod version;

use std::collections::{BTreeMap, BTreeSet};
use std::time::{Duration, Instant};

use sqlx::PgPool;
use thiserror::Error;
use tracing::debug;

use pgroles_core::manifest::{ObjectType, Privilege};
use pgroles_core::model::RoleGraph;
use pgroles_core::ownership::ManagedScope;

// Re-export the sub-modules' public items for testing / advanced use.
pub use cloud::{CloudProvider, PrivilegeLevel, detect_privilege_level};
pub use defaults::fetch_default_privileges;
pub use memberships::fetch_memberships;
pub use privileges::{
    fetch_column_level_grants, fetch_database_privileges, fetch_object_inventory, fetch_privileges,
    fetch_relation_inventory,
};
pub use public_grants::{PublicGrants, fetch_public_grants, format_public_grants};
pub use roles::fetch_roles;
pub use safety::{
    DropRoleSafetyAssessment, DropRoleSafetyIssue, DropRoleSafetyReport, inspect_drop_role_safety,
};
pub use version::{PgVersion, detect_pg_version};

// ---------------------------------------------------------------------------
// Errors
// ---------------------------------------------------------------------------

#[derive(Debug, Error)]
pub enum InspectError {
    #[error("database query error: {0}")]
    Database(#[from] sqlx::Error),
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct InspectionDiagnostics {
    pub unsatisfiable_wildcard_grants: Vec<UnsatisfiableWildcardGrant>,
    /// Column-level ACL entries (`GRANT ... (column) ON table TO role`) found
    /// on relations inside managed schemas. pgroles does not manage
    /// column-level privileges — these are surfaced as an advisory warning
    /// during `diff`/`apply`, not a blocking error: unlike
    /// [`unsatisfiable_wildcard_grants`](Self::unsatisfiable_wildcard_grants),
    /// their presence never stops inspection or reconciliation from
    /// proceeding.
    pub column_level_grants: Vec<ColumnLevelGrantDiagnostic>,
}

impl InspectionDiagnostics {
    pub fn is_empty(&self) -> bool {
        self.unsatisfiable_wildcard_grants.is_empty() && self.column_level_grants.is_empty()
    }

    /// Render the blocking diagnostics (unsatisfiable wildcard grants) as the
    /// error message `diff`/`apply` and the operator fail with, one per line.
    /// Returns `None` when nothing blocks reconciliation. Advisory
    /// diagnostics ([`column_level_grants`](Self::column_level_grants)) are
    /// deliberately excluded — callers surface those as warnings separately.
    pub fn blocking_message(&self) -> Option<String> {
        if self.unsatisfiable_wildcard_grants.is_empty() {
            return None;
        }
        Some(
            self.unsatisfiable_wildcard_grants
                .iter()
                .map(ToString::to_string)
                .collect::<Vec<_>>()
                .join("\n"),
        )
    }
}

/// Combined rendering of all diagnostics, blocking and advisory alike.
///
/// NOTE: production paths do NOT use this impl — the CLI and operator render
/// [`InspectionDiagnostics::blocking_message`] for the failure path and
/// iterate `column_level_grants` for the warning path separately, so severity
/// stays visible. This impl exists for logging/debugging convenience.
impl std::fmt::Display for InspectionDiagnostics {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut wrote_any = false;
        for diagnostic in &self.unsatisfiable_wildcard_grants {
            if wrote_any {
                writeln!(f)?;
            }
            write!(f, "{diagnostic}")?;
            wrote_any = true;
        }
        for diagnostic in &self.column_level_grants {
            if wrote_any {
                writeln!(f)?;
            }
            write!(f, "{diagnostic}")?;
            wrote_any = true;
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnsatisfiableWildcardGrant {
    pub role: String,
    pub object_type: ObjectType,
    pub schema: String,
    pub privileges: std::collections::BTreeSet<Privilege>,
    pub executor: String,
    pub skipped_count: usize,
    pub examples: Vec<UnsatisfiableWildcardObject>,
}

impl std::fmt::Display for UnsatisfiableWildcardGrant {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let privileges = self
            .privileges
            .iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join(", ");
        let examples = self
            .examples
            .iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join("; ");
        write!(
            f,
            "UnsatisfiableWildcardGrant: cannot fully satisfy wildcard grant \
             {privileges} ON {} * IN SCHEMA \"{}\" TO \"{}\" as executor \"{}\"; \
             {} matching object(s) are missing the desired privilege and are not grantable",
            self.object_type, self.schema, self.role, self.executor, self.skipped_count
        )?;
        if !examples.is_empty() {
            write!(f, " (examples: {examples})")?;
        }
        Ok(())
    }
}

/// A column-level grant detected on a relation inside a managed schema,
/// aggregated by `(schema, relation, grantee)`.
///
/// pgroles only manages table/view/etc.-level ACLs (`pg_class.relacl`); it
/// never reads or writes `pg_attribute.attacl`. When column-level grants
/// exist, the manifest is not the whole truth for that relation — this
/// diagnostic surfaces that gap without attempting to manage it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ColumnLevelGrantDiagnostic {
    pub schema: String,
    pub relation: String,
    /// The grantee role name, or the literal string `"PUBLIC"` for grants to
    /// the PUBLIC pseudo-role (ACL grantee OID 0).
    pub grantee: String,
    /// Up to [`COLUMN_LEVEL_GRANT_EXAMPLE_LIMIT`] affected column names,
    /// sorted; the overflow count lives in `skipped_columns`. Capped at
    /// construction (like `UnsatisfiableWildcardGrant::examples`) so a wide
    /// table doesn't keep thousands of names resident per diagnostic.
    pub columns: Vec<String>,
    /// Number of additional affected columns beyond `columns`.
    pub skipped_columns: usize,
    pub privileges: std::collections::BTreeSet<Privilege>,
}

/// Maximum number of column names carried by a [`ColumnLevelGrantDiagnostic`];
/// the remainder is summarized as `skipped_columns` at aggregation time.
pub(crate) const COLUMN_LEVEL_GRANT_EXAMPLE_LIMIT: usize = 8;

impl std::fmt::Display for ColumnLevelGrantDiagnostic {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let privileges = self
            .privileges
            .iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join(", ");
        let mut columns = self.columns.join(", ");
        if self.skipped_columns > 0 {
            columns.push_str(&format!(", … (+{} more)", self.skipped_columns));
        }
        write!(
            f,
            "ColumnLevelGrant: \"{}\".\"{}\" has column-level grant(s) [{privileges}] to \"{}\" \
             on column(s) [{columns}]; pgroles does not manage column-level privileges — they are \
             not diffed, revoked, or included in `generate` output. See \
             https://hardbyte.github.io/pgroles/docs/limitations/ for details.",
            self.schema, self.relation, self.grantee
        )
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnsatisfiableWildcardObject {
    pub name: String,
    pub owner: String,
    pub privileges: std::collections::BTreeSet<Privilege>,
}

impl std::fmt::Display for UnsatisfiableWildcardObject {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let privileges = self
            .privileges
            .iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join(", ");
        write!(
            f,
            "\"{}\" owned by \"{}\" missing [{}]",
            self.name, self.owner, privileges
        )
    }
}

#[derive(Debug, Clone)]
pub struct InspectionResult {
    pub graph: RoleGraph,
    pub diagnostics: InspectionDiagnostics,
    pub stats: InspectionStats,
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct InspectionStats {
    pub roles: usize,
    pub memberships: usize,
    pub schemas: usize,
    pub grants: usize,
    pub default_privileges: usize,
    pub phase_durations: BTreeMap<&'static str, Duration>,
    pub wildcard: WildcardInspectionStats,
}

impl InspectionStats {
    fn record_phase(&mut self, phase: &'static str, duration: Duration) {
        self.phase_durations.insert(phase, duration);
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct WildcardInspectionStats {
    pub configured_grants: usize,
    pub configured_scopes: usize,
    pub inventory_objects: usize,
    pub unsatisfied_grants: usize,
    pub unsatisfied_scopes: usize,
    pub grantability_queries: usize,
    pub grantability_objects: usize,
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) struct WildcardGrantPattern {
    pub role: String,
    pub object_type: pgroles_core::manifest::ObjectType,
    pub schema: String,
    /// The desired privileges for this wildcard grant. Used to construct a
    /// vacuously-satisfied wildcard when no objects of this type exist in the
    /// schema, so the diff engine sees exact parity and produces no change.
    pub privileges: std::collections::BTreeSet<pgroles_core::manifest::Privilege>,
}

// ---------------------------------------------------------------------------
// Configuration
// ---------------------------------------------------------------------------

/// Configuration for what to inspect from the database.
///
/// Scoped to only the roles and schemas that the manifest manages, so we
/// don't pull in the entire pg_catalog.
#[derive(Debug, Clone)]
pub struct InspectConfig {
    /// The role names that the manifest manages (created by pgroles).
    /// Privileges and memberships are filtered to only include these roles.
    pub managed_roles: Vec<String>,

    /// The schema names that the manifest manages for schema-owner inspection.
    pub managed_schemas: Vec<String>,

    /// The schema names whose grants/default privileges are managed.
    pub privilege_schemas: Vec<String>,

    /// Whether to also inspect database-level privileges (CONNECT, CREATE, TEMPORARY).
    /// Usually only needed if the manifest includes database-level grants.
    pub include_database_privileges: bool,

    /// Wildcard grant selectors from the desired manifest.
    pub(crate) wildcard_grants: Vec<WildcardGrantPattern>,
}

impl InspectConfig {
    /// Create an `InspectConfig` from an expanded manifest by extracting
    /// the unique set of managed role names and schema names.
    pub fn from_expanded(
        expanded: &pgroles_core::manifest::ExpandedManifest,
        include_database_privileges: bool,
    ) -> Self {
        let mut managed_roles: BTreeSet<String> = BTreeSet::new();
        let mut managed_schemas: BTreeSet<String> = BTreeSet::new();
        // Key for deduplicating wildcard grants: (role, object_type, schema).
        type WildcardKey = (String, pgroles_core::manifest::ObjectType, String);
        let mut wildcard_map: BTreeMap<WildcardKey, BTreeSet<pgroles_core::manifest::Privilege>> =
            BTreeMap::new();

        // Collect role names
        for role_def in &expanded.roles {
            managed_roles.insert(role_def.name.clone());
        }

        // Collect schema names from grants
        for grant in &expanded.grants {
            if let Some(ref schema) = grant.object.schema {
                managed_schemas.insert(schema.clone());
            }
            // Schema-level grants use the name field as the schema name
            if grant.object.object_type == pgroles_core::manifest::ObjectType::Schema
                && let Some(ref name) = grant.object.name
            {
                managed_schemas.insert(name.clone());
            }
            if grant.object.name.as_deref() == Some("*")
                && !matches!(
                    grant.object.object_type,
                    pgroles_core::manifest::ObjectType::Schema
                        | pgroles_core::manifest::ObjectType::Database
                )
                && let Some(schema) = &grant.object.schema
            {
                let key = (grant.role.clone(), grant.object.object_type, schema.clone());
                wildcard_map
                    .entry(key)
                    .or_default()
                    .extend(grant.privileges.iter().copied());
            }
        }

        // Collect schema names from default privileges
        for dp in &expanded.default_privileges {
            managed_schemas.insert(dp.schema.clone());
        }

        for schema in &expanded.schemas {
            managed_schemas.insert(schema.name.clone());
        }

        Self {
            managed_roles: managed_roles.into_iter().collect(),
            managed_schemas: managed_schemas.clone().into_iter().collect(),
            privilege_schemas: managed_schemas.into_iter().collect(),
            include_database_privileges,
            wildcard_grants: wildcard_map
                .into_iter()
                .map(
                    |((role, object_type, schema), privileges)| WildcardGrantPattern {
                        role,
                        object_type,
                        schema,
                        privileges,
                    },
                )
                .collect(),
        }
    }

    /// Create an `InspectConfig` from a managed scope plus an expanded desired
    /// manifest so current-state inspection can be restricted to composed policy
    /// boundaries.
    pub fn from_managed_scope(
        scope: &ManagedScope,
        expanded: &pgroles_core::manifest::ExpandedManifest,
        include_database_privileges: bool,
    ) -> Self {
        let base = Self::from_expanded(expanded, include_database_privileges);

        Self {
            managed_roles: scope.roles.iter().cloned().collect(),
            managed_schemas: scope.schemas.keys().cloned().collect(),
            privilege_schemas: scope
                .schemas
                .iter()
                .filter_map(|(schema, managed)| managed.bindings.then_some(schema.clone()))
                .collect(),
            include_database_privileges,
            wildcard_grants: base
                .wildcard_grants
                .into_iter()
                .filter(|pattern| {
                    scope
                        .schemas
                        .get(&pattern.schema)
                        .is_some_and(|managed| managed.bindings)
                })
                .collect(),
        }
    }

    /// Extend the managed role scope with additional explicit role names.
    pub fn with_additional_roles<I>(mut self, roles: I) -> Self
    where
        I: IntoIterator<Item = String>,
    {
        let mut managed_roles: BTreeSet<String> = self.managed_roles.into_iter().collect();
        managed_roles.extend(roles);
        self.managed_roles = managed_roles.into_iter().collect();
        self
    }
}

// ---------------------------------------------------------------------------
// Top-level inspect function
// ---------------------------------------------------------------------------

/// Configuration for unscoped inspection (used by `generate` command).
#[derive(Debug, Clone)]
pub struct InspectAllConfig {
    /// Whether to exclude PostgreSQL system roles (pg_*, postgres).
    pub exclude_system_roles: bool,
}

/// Inspect all non-system roles and their privileges for manifest generation.
///
/// Unlike [`inspect`], this does not require a manifest to scope the query.
/// It discovers all user-defined roles, schemas they have access to, and
/// reconstructs the full RoleGraph.
pub async fn inspect_all(
    pool: &PgPool,
    config: &InspectAllConfig,
) -> Result<RoleGraph, InspectError> {
    let mut graph = RoleGraph::default();

    // Fetch all non-system roles.
    // fetch_roles(None) already excludes pg_* and postgres system roles.
    // The exclude_system_roles flag is reserved for future use with broader filtering.
    let _ = config.exclude_system_roles;
    let role_rows = fetch_roles(pool, None).await?;
    for row in &role_rows {
        graph.roles.insert(row.rolname.clone(), row.to_role_state());
    }
    debug!(found = graph.roles.len(), "roles discovered for generation");

    let role_names: Vec<String> = graph.roles.keys().cloned().collect();
    let role_refs: Vec<&str> = role_names.iter().map(|s| s.as_str()).collect();

    // Discover schemas these roles have access to
    let schema_rows: Vec<(String,)> = sqlx::query_as(
        r#"
        SELECT nspname::text FROM pg_namespace
        WHERE nspname NOT LIKE 'pg_%'
          AND nspname <> 'information_schema'
        ORDER BY nspname
        "#,
    )
    .fetch_all(pool)
    .await?;
    let schema_names: Vec<String> = schema_rows.into_iter().map(|r| r.0).collect();
    let schema_refs: Vec<&str> = schema_names.iter().map(|s| s.as_str()).collect();

    // Memberships
    let membership_rows = fetch_memberships(pool, Some(&role_refs)).await?;
    for row in &membership_rows {
        graph.memberships.insert(row.to_membership_edge());
    }

    // Schemas
    let schema_rows = fetch_schemas(pool, &schema_refs).await?;
    for row in &schema_rows {
        graph.schemas.insert(
            row.schema_name.clone(),
            pgroles_core::model::SchemaState {
                owner: Some(row.owner_name.clone()),
                owner_privileges: row.owner_privileges(),
            },
        );
    }

    if graph.roles.is_empty() && graph.schemas.is_empty() {
        return Ok(graph);
    }

    // Object privileges (no wildcard patterns for unscoped inspection)
    if !schema_refs.is_empty() {
        let privilege_grants = privileges::fetch_privileges_with_wildcards(
            pool,
            &schema_refs,
            &role_refs,
            &[], // no wildcard patterns
        )
        .await?
        .grants;
        for (key, state) in privilege_grants {
            graph.grants.insert(key, state);
        }
        remove_redundant_schema_owner_grants(&mut graph);
    }

    // Database privileges
    let db_grants = fetch_database_privileges(pool, &role_refs).await?;
    for (key, state) in db_grants {
        graph.grants.insert(key, state);
    }

    // Default privileges
    if !schema_refs.is_empty() {
        let default_privs = fetch_default_privileges(pool, &schema_refs, &role_refs).await?;
        for (key, state) in default_privs {
            graph.default_privileges.insert(key, state);
        }
    }

    Ok(graph)
}

/// Inspect the current state of the database and build a `RoleGraph`.
///
/// Queries roles, memberships, object privileges, and default privileges,
/// scoped to the managed set defined by `config`.
pub async fn inspect(pool: &PgPool, config: &InspectConfig) -> Result<RoleGraph, InspectError> {
    Ok(inspect_with_diagnostics(pool, config).await?.graph)
}

/// Inspect the current database state and return diagnostics for desired-state
/// intent that cannot be satisfied by the current executor.
pub async fn inspect_with_diagnostics(
    pool: &PgPool,
    config: &InspectConfig,
) -> Result<InspectionResult, InspectError> {
    let mut graph = RoleGraph::default();
    let mut diagnostics = InspectionDiagnostics::default();
    let mut stats = InspectionStats::default();

    // Build &str slices for the query functions
    let role_refs: Vec<&str> = config.managed_roles.iter().map(|s| s.as_str()).collect();
    let schema_refs: Vec<&str> = config.managed_schemas.iter().map(|s| s.as_str()).collect();
    let privilege_schema_refs: Vec<&str> = config
        .privilege_schemas
        .iter()
        .map(|s| s.as_str())
        .collect();

    // --- Roles ---
    debug!(
        count = role_refs.len(),
        "inspecting managed roles from pg_roles"
    );
    let phase_started_at = Instant::now();
    let role_rows = fetch_roles(pool, Some(&role_refs)).await?;
    stats.record_phase("roles", phase_started_at.elapsed());
    for row in &role_rows {
        graph.roles.insert(row.rolname.clone(), row.to_role_state());
    }
    stats.roles = graph.roles.len();
    debug!(found = graph.roles.len(), "roles inspected");

    // --- Memberships ---
    debug!("inspecting memberships from pg_auth_members");
    let phase_started_at = Instant::now();
    let membership_rows = fetch_memberships(pool, Some(&role_refs)).await?;
    stats.record_phase("memberships", phase_started_at.elapsed());
    for row in &membership_rows {
        graph.memberships.insert(row.to_membership_edge());
    }
    // Also add memberships where the member (not the group) is a managed role.
    // This captures cases like "user@example.com is a member of inventory-editor"
    // where inventory-editor is the group (managed) and user@example.com is the member.
    // The fetch above already handles this (filters on group role = managed).
    stats.memberships = graph.memberships.len();
    debug!(found = graph.memberships.len(), "memberships inspected");

    // --- Schemas ---
    if !schema_refs.is_empty() {
        debug!(schemas = ?schema_refs, "inspecting schemas from pg_namespace");
        let phase_started_at = Instant::now();
        let schema_rows = fetch_schemas(pool, &schema_refs).await?;
        stats.record_phase("schemas", phase_started_at.elapsed());
        for row in &schema_rows {
            graph.schemas.insert(
                row.schema_name.clone(),
                pgroles_core::model::SchemaState {
                    owner: Some(row.owner_name.clone()),
                    owner_privileges: row.owner_privileges(),
                },
            );
        }
        stats.schemas = graph.schemas.len();
        debug!(found = graph.schemas.len(), "schemas inspected");
    }

    // --- Object privileges ---
    if !privilege_schema_refs.is_empty() {
        debug!(
            schemas = ?privilege_schema_refs,
            "inspecting object privileges via aclexplode"
        );
        let phase_started_at = Instant::now();
        let privilege_result = privileges::fetch_privileges_with_wildcards(
            pool,
            &privilege_schema_refs,
            &role_refs,
            &config.wildcard_grants,
        )
        .await?;
        stats.record_phase("object_privileges", phase_started_at.elapsed());
        stats.wildcard = privilege_result.wildcard_stats;
        diagnostics
            .unsatisfiable_wildcard_grants
            .extend(privilege_result.diagnostics);
        let privilege_grants = privilege_result.grants;
        for (key, state) in privilege_grants {
            graph.grants.insert(key, state);
        }
        remove_redundant_schema_owner_grants(&mut graph);
        stats.grants = graph.grants.len();
        debug!(found = graph.grants.len(), "privilege grants inspected");

        debug!(
            schemas = ?privilege_schema_refs,
            "inspecting column-level grants via pg_attribute.attacl"
        );
        let phase_started_at = Instant::now();
        diagnostics.column_level_grants =
            privileges::fetch_column_level_grants(pool, &privilege_schema_refs).await?;
        stats.record_phase("column_level_grants", phase_started_at.elapsed());
        if !diagnostics.column_level_grants.is_empty() {
            debug!(
                found = diagnostics.column_level_grants.len(),
                "column-level grants detected (unmanaged)"
            );
        }
    }

    // --- Database-level privileges ---
    if config.include_database_privileges {
        debug!("inspecting database-level privileges");
        let phase_started_at = Instant::now();
        let db_grants = fetch_database_privileges(pool, &role_refs).await?;
        stats.record_phase("database_privileges", phase_started_at.elapsed());
        for (key, state) in db_grants {
            graph.grants.insert(key, state);
        }
        stats.grants = graph.grants.len();
        debug!(
            total = graph.grants.len(),
            "grants after database privileges"
        );
    }

    // --- Default privileges ---
    if !privilege_schema_refs.is_empty() {
        debug!("inspecting default privileges from pg_default_acl");
        let phase_started_at = Instant::now();
        let default_privs =
            fetch_default_privileges(pool, &privilege_schema_refs, &role_refs).await?;
        stats.record_phase("default_privileges", phase_started_at.elapsed());
        for (key, state) in default_privs {
            graph.default_privileges.insert(key, state);
        }
        stats.default_privileges = graph.default_privileges.len();
        debug!(
            found = graph.default_privileges.len(),
            "default privileges inspected"
        );
    }

    Ok(InspectionResult {
        graph,
        diagnostics,
        stats,
    })
}

/// Fetch the names of all non-system schemas in the target database.
///
/// Used for pre-flight validation — the operator checks that every schema
/// referenced by a policy exists before rendering GRANT statements that would
/// otherwise fail mid-transaction with `schema "X" does not exist`.
///
/// Returns a [`BTreeSet`] for efficient membership lookup. Excludes
/// `pg_catalog`, `pg_toast`, other `pg_*` schemas, and `information_schema`.
pub async fn fetch_existing_schemas(
    pool: &PgPool,
) -> Result<std::collections::BTreeSet<String>, InspectError> {
    let rows: Vec<(String,)> = sqlx::query_as(
        r#"
        SELECT nspname::text FROM pg_namespace
        WHERE nspname NOT LIKE 'pg_%'
          AND nspname <> 'information_schema'
        "#,
    )
    .fetch_all(pool)
    .await?;
    Ok(rows.into_iter().map(|r| r.0).collect())
}

#[derive(Debug, sqlx::FromRow)]
pub struct SchemaRow {
    pub schema_name: String,
    pub owner_name: String,
    pub owner_has_create: bool,
    pub owner_has_usage: bool,
}

impl SchemaRow {
    fn owner_privileges(&self) -> BTreeSet<Privilege> {
        let mut privileges = BTreeSet::new();
        if self.owner_has_create {
            privileges.insert(Privilege::Create);
        }
        if self.owner_has_usage {
            privileges.insert(Privilege::Usage);
        }
        privileges
    }
}

pub async fn fetch_schemas(
    pool: &PgPool,
    managed_schemas: &[&str],
) -> Result<Vec<SchemaRow>, InspectError> {
    let rows = sqlx::query_as::<_, SchemaRow>(
        r#"
        SELECT
            n.nspname AS schema_name,
            owner_role.rolname AS owner_name,
            has_schema_privilege(owner_role.rolname, n.nspname, 'CREATE') AS owner_has_create,
            has_schema_privilege(owner_role.rolname, n.nspname, 'USAGE') AS owner_has_usage
        FROM pg_namespace n
        JOIN pg_roles owner_role ON owner_role.oid = n.nspowner
        WHERE n.nspname = ANY($1)
        ORDER BY n.nspname
        "#,
    )
    .bind(managed_schemas)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

fn remove_redundant_schema_owner_grants(graph: &mut RoleGraph) {
    // Keep ordinary owner CREATE/USAGE management in SchemaState instead of the
    // grants map. This avoids noisy self-grants while still preserving drift
    // when the owner's ordinary privileges have been revoked.
    graph.grants.retain(|key, _| {
        if key.object_type != pgroles_core::manifest::ObjectType::Schema {
            return true;
        }

        let Some(schema_name) = key.name.as_deref() else {
            return true;
        };

        let Some(schema_state) = graph.schemas.get(schema_name) else {
            return true;
        };

        schema_state.owner.as_deref() != Some(key.role.as_str())
    });
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use pgroles_core::manifest::{expand_manifest, parse_manifest};
    use pgroles_core::ownership::ManagedSchemaScope;

    #[test]
    fn inspect_config_from_expanded_manifest() {
        let yaml = r#"
default_owner: app_owner

profiles:
  editor:
    grants:
      - privileges: [USAGE]
        object: { type: schema }
      - privileges: [SELECT, INSERT]
        object: { type: table, name: "*" }
    default_privileges:
      - privileges: [SELECT, INSERT]
        on_type: table

schemas:
  - name: inventory
    profiles: [editor]
  - name: catalog
    profiles: [editor]

roles:
  - name: analytics
    login: true

grants:
  - role: analytics
    privileges: [CONNECT]
    object: { type: database, name: mydb }
"#;
        let manifest = parse_manifest(yaml).unwrap();
        let expanded = expand_manifest(&manifest).unwrap();
        let config = InspectConfig::from_expanded(&expanded, true);

        // Managed roles: inventory-editor, catalog-editor, analytics
        assert_eq!(config.managed_roles.len(), 3);
        assert!(
            config
                .managed_roles
                .contains(&"inventory-editor".to_string())
        );
        assert!(config.managed_roles.contains(&"catalog-editor".to_string()));
        assert!(config.managed_roles.contains(&"analytics".to_string()));

        // Managed schemas: inventory, catalog
        assert_eq!(config.managed_schemas.len(), 2);
        assert!(config.managed_schemas.contains(&"inventory".to_string()));
        assert!(config.managed_schemas.contains(&"catalog".to_string()));

        assert!(config.include_database_privileges);
        assert_eq!(config.privilege_schemas.len(), 2);
        assert_eq!(config.wildcard_grants.len(), 2);
    }

    #[test]
    fn inspect_config_can_include_retired_roles() {
        let yaml = r#"
roles:
  - name: analytics
"#;
        let manifest = parse_manifest(yaml).unwrap();
        let expanded = expand_manifest(&manifest).unwrap();
        let config = InspectConfig::from_expanded(&expanded, false)
            .with_additional_roles(vec!["legacy-app".to_string(), "analytics".to_string()]);

        assert_eq!(config.managed_roles.len(), 2);
        assert!(config.managed_roles.contains(&"analytics".to_string()));
        assert!(config.managed_roles.contains(&"legacy-app".to_string()));
    }

    #[test]
    fn inspect_config_from_managed_scope_limits_privileges_to_binding_schemas() {
        let yaml = r#"
default_owner: app_owner

profiles:
  editor:
    grants:
      - privileges: [USAGE]
        object: { type: schema }

schemas:
  - name: inventory
    owner: app_owner
    profiles: [editor]

roles:
  - name: app_owner
    login: false
"#;
        let manifest = parse_manifest(yaml).unwrap();
        let expanded = expand_manifest(&manifest).unwrap();
        let scope = ManagedScope {
            roles: BTreeSet::from(["app_owner".to_string(), "inventory-editor".to_string()]),
            schemas: BTreeMap::from([(
                "inventory".to_string(),
                ManagedSchemaScope {
                    owner: true,
                    bindings: false,
                },
            )]),
        };

        let config = InspectConfig::from_managed_scope(&scope, &expanded, false);

        assert_eq!(config.managed_schemas, vec!["inventory".to_string()]);
        assert!(config.privilege_schemas.is_empty());
        assert!(config.wildcard_grants.is_empty());
    }

    #[test]
    fn remove_redundant_schema_owner_grants_keeps_only_non_owner_schema_grants() {
        let mut graph = RoleGraph::default();
        graph.schemas.insert(
            "inventory".to_string(),
            pgroles_core::model::SchemaState {
                owner: Some("inventory_owner".to_string()),
                owner_privileges: [pgroles_core::manifest::Privilege::Create]
                    .into_iter()
                    .collect(),
            },
        );
        graph.grants.insert(
            pgroles_core::model::GrantKey {
                role: "inventory_owner".to_string(),
                object_type: pgroles_core::manifest::ObjectType::Schema,
                schema: None,
                name: Some("inventory".to_string()),
            },
            pgroles_core::model::GrantState {
                privileges: [pgroles_core::manifest::Privilege::Usage]
                    .into_iter()
                    .collect(),
            },
        );
        graph.grants.insert(
            pgroles_core::model::GrantKey {
                role: "inventory_reader".to_string(),
                object_type: pgroles_core::manifest::ObjectType::Schema,
                schema: None,
                name: Some("inventory".to_string()),
            },
            pgroles_core::model::GrantState {
                privileges: [pgroles_core::manifest::Privilege::Usage]
                    .into_iter()
                    .collect(),
            },
        );

        remove_redundant_schema_owner_grants(&mut graph);

        assert_eq!(graph.grants.len(), 1);
        assert!(
            graph
                .grants
                .keys()
                .all(|key| key.role == "inventory_reader")
        );
    }

    fn sample_column_level_grant(grantee: &str, columns: &[&str]) -> ColumnLevelGrantDiagnostic {
        ColumnLevelGrantDiagnostic {
            schema: "inventory".to_string(),
            relation: "widgets".to_string(),
            grantee: grantee.to_string(),
            columns: columns.iter().map(|c| c.to_string()).collect(),
            skipped_columns: 0,
            privileges: BTreeSet::from([Privilege::Select]),
        }
    }

    #[test]
    fn column_level_grant_display_names_schema_relation_and_grantee() {
        let diagnostic = sample_column_level_grant("analytics", &["secret"]);
        let rendered = diagnostic.to_string();

        assert!(rendered.contains("ColumnLevelGrant"));
        assert!(rendered.contains("\"inventory\".\"widgets\""));
        assert!(rendered.contains("\"analytics\""));
        assert!(rendered.contains("SELECT"));
        assert!(rendered.contains("secret"));
        assert!(rendered.contains("does not manage column-level privileges"));
        assert!(rendered.contains("https://hardbyte.github.io/pgroles/docs/limitations/"));
    }

    #[test]
    fn column_level_grant_display_renders_public_grantee_explicitly() {
        let diagnostic = sample_column_level_grant("PUBLIC", &["secret"]);
        let rendered = diagnostic.to_string();

        assert!(rendered.contains("\"PUBLIC\""));
    }

    #[test]
    fn column_level_grant_display_summarizes_skipped_columns() {
        // Capping happens at aggregation time (see privileges.rs tests);
        // Display just renders the carried examples plus the overflow count.
        let mut diagnostic = sample_column_level_grant("analytics", &["col_a", "col_b"]);
        diagnostic.skipped_columns = 4;

        let rendered = diagnostic.to_string();
        assert!(rendered.contains("col_a, col_b, … (+4 more)"));
    }

    #[test]
    fn inspection_diagnostics_is_empty_requires_both_fields_empty() {
        let mut diagnostics = InspectionDiagnostics::default();
        assert!(diagnostics.is_empty());

        diagnostics
            .column_level_grants
            .push(sample_column_level_grant("analytics", &["secret"]));
        assert!(!diagnostics.is_empty());
    }

    #[test]
    fn inspection_diagnostics_display_includes_column_level_grants() {
        let mut diagnostics = InspectionDiagnostics::default();
        diagnostics
            .column_level_grants
            .push(sample_column_level_grant("analytics", &["secret"]));

        let rendered = diagnostics.to_string();
        assert!(rendered.contains("ColumnLevelGrant"));
    }

    #[test]
    fn inspection_diagnostics_display_joins_wildcard_and_column_level_diagnostics() {
        let mut diagnostics = InspectionDiagnostics::default();
        diagnostics
            .unsatisfiable_wildcard_grants
            .push(UnsatisfiableWildcardGrant {
                role: "reader".to_string(),
                object_type: ObjectType::Table,
                schema: "inventory".to_string(),
                privileges: BTreeSet::from([Privilege::Select]),
                executor: "app_owner".to_string(),
                skipped_count: 1,
                examples: vec![],
            });
        diagnostics
            .column_level_grants
            .push(sample_column_level_grant("analytics", &["secret"]));

        let rendered = diagnostics.to_string();
        let wildcard_pos = rendered.find("UnsatisfiableWildcardGrant").unwrap();
        let column_pos = rendered.find("ColumnLevelGrant").unwrap();
        assert!(
            wildcard_pos < column_pos,
            "wildcard diagnostics should render before column-level diagnostics"
        );
        // Both diagnostics must be on their own line.
        assert_eq!(rendered.lines().count(), 2);
    }
}