use std::sync::Arc;

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use lakecat_core::{LakeCatError, LakeCatResult, Principal, TableIdent, content_hash_json};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

#[async_trait]
pub trait GovernanceEngine: Send + Sync + 'static {
    async fn authorize(&self, request: AuthorizationRequest)
    -> LakeCatResult<AuthorizationReceipt>;
}

pub const ALLOW_ALL_LOCAL_ENGINE: &str = "lakecat-allow-all-local";

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct AuthorizationRequest {
    pub principal: Principal,
    pub action: CatalogAction,
    pub table: Option<TableIdent>,
    pub context: Value,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
pub enum CatalogAction {
    CatalogConfig,
    NamespaceCreate,
    NamespaceList,
    NamespaceLoad,
    NamespaceDrop,
    TableCreate,
    TableRegister,
    TableLoad,
    TablePlanScan,
    TableCommit,
    TableDrop,
    TableRestore,
    CredentialsVend,
    ViewLoad,
    ViewDrop,
    ServerManage,
    ProjectManage,
    WarehouseManage,
    StorageProfileManage,
    ViewManage,
    PolicyManage,
    GraphRead,
    LineageRead,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct AuthorizationReceipt {
    pub principal: Principal,
    pub action: CatalogAction,
    pub table: Option<TableIdent>,
    pub allowed: bool,
    pub engine: String,
    pub policy_hash: Option<String>,
    pub context: Value,
    pub checked_at: DateTime<Utc>,
}

impl AuthorizationReceipt {
    pub fn with_read_restriction_policy_hash(mut self) -> LakeCatResult<Self> {
        let Some(restriction) = self.context.get("read-restriction") else {
            return Ok(self);
        };
        let restriction: ReadRestriction =
            serde_json::from_value(restriction.clone()).map_err(|err| {
                LakeCatError::InvalidArgument(format!(
                    "authorization receipt carries invalid read restriction: {err}"
                ))
            })?;
        if restriction.policy_hashes.is_empty() {
            return Ok(self);
        }
        self.policy_hash = Some(content_hash_json(&json!({
            "engine": self.engine,
            "governance-policy-hash": self.policy_hash,
            "read-restriction-policy-hashes": restriction.policy_hashes,
            "action": self.action,
            "table": self.table,
        }))?);
        Ok(self)
    }
}

#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "kebab-case")]
pub struct ReadRestriction {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub allowed_columns: Option<Vec<String>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub row_predicate: Option<Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub purpose: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_credential_ttl_seconds: Option<u64>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub policy_hashes: Vec<String>,
}

impl ReadRestriction {
    pub fn unrestricted() -> Self {
        Self::default()
    }

    pub fn is_unrestricted(&self) -> bool {
        self.allowed_columns.is_none()
            && self.row_predicate.is_none()
            && self.purpose.is_none()
            && self.max_credential_ttl_seconds.is_none()
    }

    pub fn requires_governed_read(&self) -> bool {
        self.allowed_columns.is_some() || self.row_predicate.is_some()
    }

    pub fn from_odrl_policies<'a>(
        policies: impl IntoIterator<Item = &'a Value>,
    ) -> LakeCatResult<Self> {
        let mut restriction = Self::unrestricted();
        for odrl in policies {
            let policy_hash = content_hash_json(odrl)?;
            restriction.policy_hashes.push(policy_hash);
            if let Some(columns) = allowed_columns_from_odrl(odrl)? {
                restriction.allowed_columns = Some(match restriction.allowed_columns.take() {
                    Some(existing) => intersect_columns(&existing, &columns),
                    None => columns,
                });
            }
            if let Some(row_predicate) = row_predicate_from_odrl(odrl)? {
                restriction.row_predicate = Some(match restriction.row_predicate.take() {
                    Some(existing) => and_row_predicates(existing, row_predicate),
                    None => row_predicate,
                });
            }
            if let Some(purpose) = purpose_from_odrl(odrl)? {
                restriction.purpose = Some(match restriction.purpose.take() {
                    Some(existing) if existing == purpose => existing,
                    Some(existing) => {
                        return Err(LakeCatError::Conflict(format!(
                            "ODRL read restriction carries conflicting purposes {existing} and {purpose}"
                        )));
                    }
                    None => purpose,
                });
            }
            if let Some(ttl) = ttl_from_odrl(odrl)? {
                restriction.max_credential_ttl_seconds =
                    Some(match restriction.max_credential_ttl_seconds {
                        Some(existing) => existing.min(ttl),
                        None => ttl,
                    });
            }
        }
        Ok(restriction)
    }

    pub fn effective_projection(
        &self,
        requested_projection: &[String],
    ) -> LakeCatResult<Vec<String>> {
        let Some(allowed_columns) = self.allowed_columns.as_ref() else {
            return Ok(requested_projection.to_vec());
        };
        if allowed_columns.is_empty() {
            return Err(LakeCatError::Conflict(
                "read restriction leaves no columns available for scan planning".to_string(),
            ));
        }
        if requested_projection.is_empty() {
            return Ok(allowed_columns.clone());
        }
        let projection = requested_projection
            .iter()
            .filter(|column| allowed_columns.iter().any(|allowed| allowed == *column))
            .cloned()
            .collect::<Vec<_>>();
        if projection.is_empty() {
            return Err(LakeCatError::Conflict(
                "requested projection is outside the governed read restriction".to_string(),
            ));
        }
        Ok(projection)
    }

    pub fn effective_stats_fields(&self, requested: &[String]) -> Vec<String> {
        let Some(allowed_columns) = self.allowed_columns.as_ref() else {
            return requested.to_vec();
        };
        requested
            .iter()
            .filter(|column| allowed_columns.iter().any(|allowed| allowed == *column))
            .cloned()
            .collect()
    }

    pub fn mandatory_filters(&self) -> Vec<Value> {
        self.row_predicate.iter().cloned().collect()
    }
}

fn allowed_columns_from_odrl(odrl: &Value) -> LakeCatResult<Option<Vec<String>>> {
    for value in [
        odrl.get("allowed-columns"),
        odrl.get("allowedColumns"),
        odrl.get("columns"),
        odrl.get("lakecat:read-restriction")
            .and_then(|value| value.get("allowed-columns")),
        odrl.get("lakecat:read-restriction")
            .and_then(|value| value.get("allowedColumns")),
        odrl.get("read-restriction")
            .and_then(|value| value.get("allowed-columns")),
        odrl.get("readRestriction")
            .and_then(|value| value.get("allowedColumns")),
    ]
    .into_iter()
    .flatten()
    {
        return Ok(Some(string_list(value, "ODRL allowed columns")?));
    }

    let mut columns = Vec::new();
    for constraint in odrl_constraints(odrl) {
        let left = constraint_left_operand(constraint).unwrap_or_default();
        if matches!(
            left,
            "column" | "columns" | "allowed-columns" | "allowedColumns" | "lakecat:allowed-columns"
        ) {
            let value = constraint_right_operand(constraint, "allowed columns")?;
            require_constraint_operator(
                constraint,
                "allowed columns",
                &["eq", "isAnyOf", "isAllOf"],
            )?;
            columns.extend(string_list(value, "ODRL allowed columns")?);
        }
    }
    if columns.is_empty() {
        Ok(None)
    } else {
        Ok(Some(dedup_columns(columns)))
    }
}

fn row_predicate_from_odrl(odrl: &Value) -> LakeCatResult<Option<Value>> {
    for value in [
        odrl.get("row-predicate"),
        odrl.get("rowPredicate"),
        odrl.get("lakecat:row-predicate"),
        odrl.get("lakecat:read-restriction")
            .and_then(|value| value.get("row-predicate")),
        odrl.get("lakecat:read-restriction")
            .and_then(|value| value.get("rowPredicate")),
        odrl.get("read-restriction")
            .and_then(|value| value.get("row-predicate")),
        odrl.get("readRestriction")
            .and_then(|value| value.get("rowPredicate")),
    ]
    .into_iter()
    .flatten()
    {
        return Ok(Some(row_predicate_value(value)?));
    }

    let mut predicate = None;
    for constraint in odrl_constraints(odrl) {
        let left = constraint_left_operand(constraint).unwrap_or_default();
        if matches!(
            left,
            "row-predicate" | "rowPredicate" | "lakecat:row-predicate"
        ) {
            let value = constraint_right_operand(constraint, "row predicate")?;
            require_constraint_operator(constraint, "row predicate", &["eq"])?;
            let next = row_predicate_value(value)?;
            predicate = Some(match predicate {
                Some(existing) => and_row_predicates(existing, next),
                None => next,
            });
        }
    }
    Ok(predicate)
}

fn row_predicate_value(value: &Value) -> LakeCatResult<Value> {
    match value {
        Value::Object(_) => Ok(value.clone()),
        _ => Err(LakeCatError::InvalidArgument(
            "ODRL row predicate must be an Iceberg expression object".to_string(),
        )),
    }
}

fn and_row_predicates(left: Value, right: Value) -> Value {
    json!({
        "type": "and",
        "left": left,
        "right": right,
    })
}

fn purpose_from_odrl(odrl: &Value) -> LakeCatResult<Option<String>> {
    let mut purpose = odrl
        .get("purpose")
        .map(|value| nonblank_jsonld_string(value, "ODRL purpose"))
        .transpose()?
        .map(str::to_string);

    for constraint in odrl_constraints(odrl) {
        let left = constraint_left_operand(constraint).unwrap_or_default();
        if matches!(left, "purpose" | "lakecat:purpose") {
            require_constraint_operator(constraint, "purpose", &["eq"])?;
            let right_operand = constraint_right_operand(constraint, "purpose")?;
            let next = jsonld_string_value(right_operand)
                .ok_or_else(|| {
                    LakeCatError::InvalidArgument(
                        "ODRL purpose constraint must use a string right operand".to_string(),
                    )
                })
                .and_then(|value| require_nonblank_string(value, "ODRL purpose constraint"))?
                .to_string();
            purpose = Some(match purpose.take() {
                Some(existing) if existing == next => existing,
                Some(existing) => {
                    return Err(LakeCatError::Conflict(format!(
                        "ODRL read restriction carries conflicting purposes {existing} and {next}"
                    )));
                }
                None => next,
            });
        }
    }
    Ok(purpose)
}

fn ttl_from_odrl(odrl: &Value) -> LakeCatResult<Option<u64>> {
    let mut ttl = None;
    for value in [
        odrl.get("max-credential-ttl-seconds"),
        odrl.get("maxCredentialTtlSeconds"),
        odrl.get("lakecat:read-restriction")
            .and_then(|value| value.get("max-credential-ttl-seconds")),
        odrl.get("lakecat:read-restriction")
            .and_then(|value| value.get("maxCredentialTtlSeconds")),
        odrl.get("read-restriction")
            .and_then(|value| value.get("max-credential-ttl-seconds")),
        odrl.get("readRestriction")
            .and_then(|value| value.get("maxCredentialTtlSeconds")),
    ]
    .into_iter()
    .flatten()
    {
        ttl = tighten_ttl(ttl, ttl_value(value)?);
    }

    for constraint in odrl_constraints(odrl) {
        let left = constraint_left_operand(constraint).unwrap_or_default();
        if matches!(
            left,
            "max-credential-ttl-seconds"
                | "maxCredentialTtlSeconds"
                | "credential-ttl"
                | "credentialTtl"
                | "lakecat:max-credential-ttl-seconds"
                | "lakecat:credential-ttl"
        ) {
            let value = constraint_right_operand(constraint, "max credential TTL")?;
            require_constraint_operator(constraint, "max credential TTL", &["eq", "lteq", "lt"])?;
            ttl = tighten_ttl(ttl, ttl_value(value)?);
        }
    }
    Ok(ttl)
}

fn tighten_ttl(existing: Option<u64>, next: Option<u64>) -> Option<u64> {
    match (existing, next) {
        (Some(existing), Some(next)) => Some(existing.min(next)),
        (Some(existing), None) => Some(existing),
        (None, next) => next,
    }
}

fn require_constraint_operator(
    constraint: &Value,
    label: &str,
    allowed: &[&str],
) -> LakeCatResult<()> {
    let Some(operator) = constraint_operator(constraint) else {
        return Err(LakeCatError::InvalidArgument(format!(
            "ODRL {label} constraint must include an operator"
        )));
    };
    let normalized = operator
        .strip_prefix("odrl:")
        .or_else(|| operator.strip_prefix("http://www.w3.org/ns/odrl/2/"))
        .unwrap_or(operator);
    if allowed.iter().any(|allowed| allowed == &normalized) {
        Ok(())
    } else {
        Err(LakeCatError::InvalidArgument(format!(
            "ODRL {label} constraint uses unsupported operator {operator}"
        )))
    }
}

fn constraint_operator(constraint: &Value) -> Option<&str> {
    constraint
        .get("operator")
        .or_else(|| constraint.get("odrl:operator"))
        .and_then(jsonld_term)
}

fn constraint_left_operand(constraint: &Value) -> Option<&str> {
    constraint
        .get("leftOperand")
        .or_else(|| constraint.get("left-operand"))
        .or_else(|| constraint.get("odrl:leftOperand"))
        .and_then(jsonld_term)
}

fn jsonld_term(value: &Value) -> Option<&str> {
    value
        .as_str()
        .or_else(|| value.get("@id").and_then(Value::as_str))
}

fn constraint_right_operand<'a>(constraint: &'a Value, label: &str) -> LakeCatResult<&'a Value> {
    constraint
        .get("rightOperand")
        .or_else(|| constraint.get("right-operand"))
        .or_else(|| constraint.get("odrl:rightOperand"))
        .ok_or_else(|| {
            LakeCatError::InvalidArgument(format!(
                "ODRL {label} constraint must include a right operand"
            ))
        })
}

fn ttl_value(value: &Value) -> LakeCatResult<Option<u64>> {
    jsonld_u64_value(value).map(Some).ok_or_else(|| {
        LakeCatError::InvalidArgument(
            "ODRL max credential TTL must be an unsigned integer number of seconds".to_string(),
        )
    })
}

fn odrl_constraints(odrl: &Value) -> Vec<&Value> {
    let mut constraints = Vec::new();
    for permission in values_as_slice(odrl.get("permission")) {
        constraints.extend(values_as_slice(permission.get("constraint")));
    }
    constraints.extend(values_as_slice(odrl.get("constraint")));
    constraints
}

fn values_as_slice(value: Option<&Value>) -> Vec<&Value> {
    match value {
        Some(Value::Array(values)) => values.iter().collect(),
        Some(value) => vec![value],
        None => Vec::new(),
    }
}

fn string_list(value: &Value, label: &str) -> LakeCatResult<Vec<String>> {
    let raw = match value {
        Value::Array(values) => values
            .iter()
            .map(|item| nonblank_jsonld_string(item, label).map(ToString::to_string))
            .collect::<Result<Vec<_>, _>>()?,
        Value::Object(object) if object.contains_key("@list") => object
            .get("@list")
            .and_then(Value::as_array)
            .ok_or_else(|| {
                LakeCatError::InvalidArgument(format!("{label} @list must be an array"))
            })?
            .iter()
            .map(|item| nonblank_jsonld_string(item, label).map(ToString::to_string))
            .collect::<Result<Vec<_>, _>>()?,
        value if jsonld_string_value(value).is_some() => {
            vec![nonblank_jsonld_string(value, label)?.to_string()]
        }
        _ => {
            return Err(LakeCatError::InvalidArgument(format!(
                "{label} must be a string or string array"
            )));
        }
    };
    if raw.is_empty() {
        return Err(LakeCatError::InvalidArgument(format!(
            "{label} must not be empty"
        )));
    }
    Ok(dedup_columns(raw))
}

fn nonblank_jsonld_string<'a>(value: &'a Value, label: &str) -> LakeCatResult<&'a str> {
    let value = jsonld_string_value(value)
        .ok_or_else(|| LakeCatError::InvalidArgument(format!("{label} must be strings")))?;
    require_nonblank_string(value, label)
}

fn require_nonblank_string<'a>(value: &'a str, label: &str) -> LakeCatResult<&'a str> {
    if value.trim().is_empty() {
        return Err(LakeCatError::InvalidArgument(format!(
            "{label} must not be blank"
        )));
    }
    Ok(value)
}

fn jsonld_string_value(value: &Value) -> Option<&str> {
    value
        .as_str()
        .or_else(|| value.get("@value").and_then(Value::as_str))
}

fn jsonld_u64_value(value: &Value) -> Option<u64> {
    value.as_u64().or_else(|| {
        value.get("@value").and_then(|value| {
            value
                .as_u64()
                .or_else(|| value.as_str().and_then(|value| value.parse::<u64>().ok()))
        })
    })
}

fn dedup_columns(columns: Vec<String>) -> Vec<String> {
    let mut out = Vec::with_capacity(columns.len());
    for column in columns {
        if !out.iter().any(|existing| existing == &column) {
            out.push(column);
        }
    }
    out
}

fn intersect_columns(left: &[String], right: &[String]) -> Vec<String> {
    left.iter()
        .filter(|column| right.iter().any(|allowed| allowed == *column))
        .cloned()
        .collect()
}

#[derive(Debug, Clone, PartialEq)]
pub struct Capability<Action, Resource> {
    receipt: AuthorizationReceipt,
    resource: Resource,
    _action: std::marker::PhantomData<Action>,
}

impl<Action, Resource> Capability<Action, Resource> {
    pub fn receipt(&self) -> &AuthorizationReceipt {
        &self.receipt
    }

    pub fn resource(&self) -> &Resource {
        &self.resource
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanCreateTable;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanLoadTable;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanCommitTable;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanDropTable;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanRestoreTable;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanPlanScan;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanVendCredentials;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanManageServers;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanManageProjects;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanManageWarehouses;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanManageStorageProfiles;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanLoadViews;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanDropViews;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanManageViews;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanManagePolicies;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanReadGraph;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanReadLineage;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanReadCatalogConfig;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanCreateNamespace;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanListNamespaces;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanLoadNamespace;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CanDropNamespace;

pub type TableCreateCapability = Capability<CanCreateTable, TableIdent>;
pub type TableLoadCapability = Capability<CanLoadTable, TableIdent>;
pub type TableCommitCapability = Capability<CanCommitTable, TableIdent>;
pub type TableDropCapability = Capability<CanDropTable, TableIdent>;
pub type TableRestoreCapability = Capability<CanRestoreTable, TableIdent>;
pub type TableScanCapability = Capability<CanPlanScan, TableIdent>;
pub type CredentialsVendCapability = Capability<CanVendCredentials, TableIdent>;
pub type ServerManageCapability = Capability<CanManageServers, ()>;
pub type ProjectManageCapability = Capability<CanManageProjects, ()>;
pub type WarehouseManageCapability = Capability<CanManageWarehouses, ()>;
pub type StorageProfileManageCapability = Capability<CanManageStorageProfiles, ()>;
pub type ViewLoadCapability = Capability<CanLoadViews, ()>;
pub type ViewDropCapability = Capability<CanDropViews, ()>;
pub type ViewManageCapability = Capability<CanManageViews, ()>;
pub type PolicyManageCapability = Capability<CanManagePolicies, ()>;
pub type GraphReadCapability = Capability<CanReadGraph, ()>;
pub type LineageReadCapability = Capability<CanReadLineage, ()>;
pub type CatalogConfigCapability = Capability<CanReadCatalogConfig, ()>;
pub type NamespaceCreateCapability = Capability<CanCreateNamespace, ()>;
pub type NamespaceListCapability = Capability<CanListNamespaces, ()>;
pub type NamespaceLoadCapability = Capability<CanLoadNamespace, ()>;
pub type NamespaceDropCapability = Capability<CanDropNamespace, ()>;

impl TableCreateCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt, table: TableIdent) -> LakeCatResult<Self> {
        table_capability_from_receipt(receipt, table, CatalogAction::TableCreate, "create table")
    }

    pub fn table(&self) -> &TableIdent {
        self.resource()
    }
}

impl TableLoadCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt, table: TableIdent) -> LakeCatResult<Self> {
        table_capability_from_receipt(receipt, table, CatalogAction::TableLoad, "load table")
    }

    pub fn table(&self) -> &TableIdent {
        self.resource()
    }
}

impl TableCommitCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt, table: TableIdent) -> LakeCatResult<Self> {
        table_capability_from_receipt(receipt, table, CatalogAction::TableCommit, "commit table")
    }

    pub fn table(&self) -> &TableIdent {
        self.resource()
    }
}

impl TableDropCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt, table: TableIdent) -> LakeCatResult<Self> {
        table_capability_from_receipt(receipt, table, CatalogAction::TableDrop, "drop table")
    }

    pub fn table(&self) -> &TableIdent {
        self.resource()
    }
}

impl TableRestoreCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt, table: TableIdent) -> LakeCatResult<Self> {
        table_capability_from_receipt(receipt, table, CatalogAction::TableRestore, "restore table")
    }

    pub fn table(&self) -> &TableIdent {
        self.resource()
    }
}

impl TableScanCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt, table: TableIdent) -> LakeCatResult<Self> {
        table_capability_from_receipt(
            receipt,
            table,
            CatalogAction::TablePlanScan,
            "plan table scans",
        )
    }

    pub fn table(&self) -> &TableIdent {
        self.resource()
    }

    pub fn read_restriction(&self) -> LakeCatResult<ReadRestriction> {
        match self.receipt().context.get("read-restriction") {
            Some(value) => serde_json::from_value(value.clone()).map_err(|err| {
                LakeCatError::InvalidArgument(format!(
                    "authorization receipt carries invalid read restriction: {err}"
                ))
            }),
            None => Ok(ReadRestriction::unrestricted()),
        }
    }
}

impl CredentialsVendCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt, table: TableIdent) -> LakeCatResult<Self> {
        table_capability_from_receipt(
            receipt,
            table,
            CatalogAction::CredentialsVend,
            "vend table credentials",
        )
    }

    pub fn table(&self) -> &TableIdent {
        self.resource()
    }

    pub fn read_restriction(&self) -> LakeCatResult<ReadRestriction> {
        match self.receipt().context.get("read-restriction") {
            Some(value) => serde_json::from_value(value.clone()).map_err(|err| {
                LakeCatError::InvalidArgument(format!(
                    "authorization receipt carries invalid read restriction: {err}"
                ))
            }),
            None => Ok(ReadRestriction::unrestricted()),
        }
    }
}

impl ServerManageCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::ServerManage, "manage servers")
    }
}

impl StorageProfileManageCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(
            receipt,
            CatalogAction::StorageProfileManage,
            "manage storage profiles",
        )
    }
}

impl ViewManageCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::ViewManage, "manage views")
    }
}

impl ViewLoadCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::ViewLoad, "load views")
    }
}

impl ViewDropCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::ViewDrop, "drop views")
    }
}

impl WarehouseManageCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(
            receipt,
            CatalogAction::WarehouseManage,
            "manage warehouses",
        )
    }
}

impl ProjectManageCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::ProjectManage, "manage projects")
    }
}

impl PolicyManageCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::PolicyManage, "manage policies")
    }
}

impl GraphReadCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::GraphRead, "read catalog graph")
    }
}

impl LineageReadCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::LineageRead, "read catalog lineage")
    }
}

impl CatalogConfigCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(
            receipt,
            CatalogAction::CatalogConfig,
            "read catalog config",
        )
    }
}

impl NamespaceCreateCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(
            receipt,
            CatalogAction::NamespaceCreate,
            "create namespaces",
        )
    }
}

impl NamespaceListCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::NamespaceList, "list namespaces")
    }
}

impl NamespaceLoadCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::NamespaceLoad, "load namespaces")
    }
}

impl NamespaceDropCapability {
    pub fn from_receipt(receipt: AuthorizationReceipt) -> LakeCatResult<Self> {
        catalog_capability_from_receipt(receipt, CatalogAction::NamespaceDrop, "drop namespaces")
    }
}

fn catalog_capability_from_receipt<Action>(
    receipt: AuthorizationReceipt,
    expected_action: CatalogAction,
    action_description: &str,
) -> LakeCatResult<Capability<Action, ()>> {
    if !receipt.allowed {
        return Err(LakeCatError::Conflict(
            "authorization receipt is not allowed".to_string(),
        ));
    }
    if receipt.action != expected_action {
        return Err(LakeCatError::InvalidArgument(format!(
            "authorization receipt action {:?} cannot {action_description}",
            receipt.action,
        )));
    }
    if receipt.table.is_some() {
        return Err(LakeCatError::InvalidArgument(
            "catalog authorization receipt must not be table-scoped".to_string(),
        ));
    }
    Ok(Capability {
        receipt,
        resource: (),
        _action: std::marker::PhantomData,
    })
}

fn table_capability_from_receipt<Action>(
    receipt: AuthorizationReceipt,
    table: TableIdent,
    expected_action: CatalogAction,
    action_description: &str,
) -> LakeCatResult<Capability<Action, TableIdent>> {
    if !receipt.allowed {
        return Err(LakeCatError::Conflict(
            "authorization receipt is not allowed".to_string(),
        ));
    }
    if receipt.action != expected_action {
        return Err(LakeCatError::InvalidArgument(format!(
            "authorization receipt action {:?} cannot {action_description}",
            receipt.action,
        )));
    }
    if receipt.table.as_ref() != Some(&table) {
        return Err(LakeCatError::InvalidArgument(
            "authorization receipt table does not match scan table".to_string(),
        ));
    }
    Ok(Capability {
        receipt,
        resource: table,
        _action: std::marker::PhantomData,
    })
}

#[cfg(test)]
mod tests {
    use lakecat_core::{Namespace, PrincipalKind, TableName, WarehouseName};

    use super::*;

    #[test]
    fn read_restriction_composes_odrl_policy_documents() {
        let policy_a = serde_json::json!({
            "uid": "policy-a",
            "purpose": "resilience-demo",
            "lakecat:read-restriction": {
                "max-credential-ttl-seconds": 900,
                "allowed-columns": ["event_id", "payload"],
                "row-predicate": {
                    "type": "equal",
                    "term": "region",
                    "value": "west"
                }
            }
        });
        let policy_b = serde_json::json!({
            "uid": "policy-b",
            "max-credential-ttl-seconds": 300,
            "permission": [{
                "constraint": [
                    {
                        "leftOperand": "allowed-columns",
                        "operator": "eq",
                        "rightOperand": ["event_id", "severity"]
                    },
                    {
                        "leftOperand": "row-predicate",
                        "operator": "eq",
                        "rightOperand": {
                            "type": "greater-than-or-equal",
                            "term": "severity",
                            "value": 3
                        }
                    }
                ]
            }]
        });

        let restriction = ReadRestriction::from_odrl_policies([&policy_a, &policy_b]).unwrap();

        assert_eq!(
            restriction.allowed_columns,
            Some(vec!["event_id".to_string()])
        );
        assert_eq!(restriction.purpose.as_deref(), Some("resilience-demo"));
        assert_eq!(restriction.max_credential_ttl_seconds, Some(300));
        assert_eq!(restriction.policy_hashes.len(), 2);
        assert_eq!(
            restriction.row_predicate,
            Some(serde_json::json!({
                "type": "and",
                "left": {
                    "type": "equal",
                    "term": "region",
                    "value": "west"
                },
                "right": {
                    "type": "greater-than-or-equal",
                    "term": "severity",
                    "value": 3
                }
            }))
        );
    }

    #[test]
    fn read_restriction_parses_ttl_from_odrl_constraints_and_uses_tightest_ttl() {
        let policy_a = serde_json::json!({
            "uid": "policy-a",
            "lakecat:read-restriction": {
                "max-credential-ttl-seconds": 900
            }
        });
        let policy_b = serde_json::json!({
            "uid": "policy-b",
            "permission": [{
                "constraint": {
                    "leftOperand": "max-credential-ttl-seconds",
                    "operator": "lteq",
                    "rightOperand": 300
                }
            }]
        });

        let restriction = ReadRestriction::from_odrl_policies([&policy_a, &policy_b]).unwrap();

        assert_eq!(restriction.max_credential_ttl_seconds, Some(300));
    }

    #[test]
    fn read_restriction_uses_tightest_ttl_within_policy_document() {
        let policy = serde_json::json!({
            "uid": "policy-a",
            "max-credential-ttl-seconds": 900,
            "lakecat:read-restriction": {
                "max-credential-ttl-seconds": 600
            },
            "permission": [{
                "constraint": [
                    {
                        "leftOperand": "max-credential-ttl-seconds",
                        "operator": "lteq",
                        "rightOperand": 300
                    },
                    {
                        "leftOperand": "credential-ttl",
                        "operator": "eq",
                        "rightOperand": 1200
                    }
                ]
            }]
        });

        let restriction = ReadRestriction::from_odrl_policies([&policy]).unwrap();

        assert_eq!(restriction.max_credential_ttl_seconds, Some(300));
    }

    #[test]
    fn read_restriction_accepts_prefixed_odrl_constraint_operands() {
        let policy = serde_json::json!({
            "uid": "policy-a",
            "permission": [{
                "constraint": [
                    {
                        "odrl:leftOperand": "allowed-columns",
                        "odrl:operator": "odrl:eq",
                        "odrl:rightOperand": ["event_id", "severity"]
                    },
                    {
                        "odrl:leftOperand": "row-predicate",
                        "odrl:operator": "http://www.w3.org/ns/odrl/2/eq",
                        "odrl:rightOperand": {
                            "type": "equal",
                            "term": "region",
                            "value": "west"
                        }
                    },
                    {
                        "odrl:leftOperand": "purpose",
                        "odrl:operator": "odrl:eq",
                        "odrl:rightOperand": "resilience-demo"
                    },
                    {
                        "odrl:leftOperand": "credential-ttl",
                        "odrl:operator": "odrl:lteq",
                        "odrl:rightOperand": 300
                    }
                ]
            }]
        });

        let restriction = ReadRestriction::from_odrl_policies([&policy]).unwrap();

        assert_eq!(
            restriction.allowed_columns,
            Some(vec!["event_id".to_string(), "severity".to_string()])
        );
        assert_eq!(
            restriction.row_predicate,
            Some(serde_json::json!({
                "type": "equal",
                "term": "region",
                "value": "west"
            }))
        );
        assert_eq!(restriction.purpose.as_deref(), Some("resilience-demo"));
        assert_eq!(restriction.max_credential_ttl_seconds, Some(300));
    }

    #[test]
    fn read_restriction_accepts_jsonld_term_objects_for_constraint_terms() {
        let policy = serde_json::json!({
            "uid": "policy-a",
            "permission": [{
                "constraint": [
                    {
                        "leftOperand": { "@id": "lakecat:allowed-columns" },
                        "operator": { "@id": "odrl:eq" },
                        "rightOperand": ["event_id"]
                    },
                    {
                        "odrl:leftOperand": { "@id": "lakecat:row-predicate" },
                        "odrl:operator": { "@id": "http://www.w3.org/ns/odrl/2/eq" },
                        "odrl:rightOperand": {
                            "type": "equal",
                            "term": "region",
                            "value": "west"
                        }
                    }
                ]
            }]
        });

        let restriction = ReadRestriction::from_odrl_policies([&policy]).unwrap();

        assert_eq!(
            restriction.allowed_columns,
            Some(vec!["event_id".to_string()])
        );
        assert_eq!(
            restriction.row_predicate,
            Some(serde_json::json!({
                "type": "equal",
                "term": "region",
                "value": "west"
            }))
        );
    }

    #[test]
    fn read_restriction_accepts_jsonld_value_objects_for_right_operands() {
        let policy = serde_json::json!({
            "uid": "policy-a",
            "purpose": { "@value": "resilience-demo" },
            "permission": [{
                "constraint": [
                    {
                        "leftOperand": { "@id": "lakecat:allowed-columns" },
                        "operator": { "@id": "odrl:isAnyOf" },
                        "rightOperand": {
                            "@list": [
                                { "@value": "event_id" },
                                { "@value": "severity" },
                                { "@value": "event_id" }
                            ]
                        }
                    },
                    {
                        "leftOperand": { "@id": "lakecat:purpose" },
                        "operator": { "@id": "odrl:eq" },
                        "rightOperand": { "@value": "resilience-demo" }
                    },
                    {
                        "leftOperand": { "@id": "lakecat:credential-ttl" },
                        "operator": { "@id": "odrl:lteq" },
                        "rightOperand": {
                            "@value": "300",
                            "@type": "http://www.w3.org/2001/XMLSchema#unsignedLong"
                        }
                    }
                ]
            }]
        });

        let restriction = ReadRestriction::from_odrl_policies([&policy]).unwrap();

        assert_eq!(
            restriction.allowed_columns,
            Some(vec!["event_id".to_string(), "severity".to_string()])
        );
        assert_eq!(restriction.purpose.as_deref(), Some("resilience-demo"));
        assert_eq!(restriction.max_credential_ttl_seconds, Some(300));
    }

    #[test]
    fn read_restriction_rejects_malformed_jsonld_allowed_column_lists() {
        let policy = serde_json::json!({
            "uid": "policy-a",
            "permission": [{
                "constraint": {
                    "leftOperand": { "@id": "lakecat:allowed-columns" },
                    "operator": { "@id": "odrl:isAnyOf" },
                    "rightOperand": {
                        "@list": [
                            { "@value": "event_id" },
                            { "@id": "lakecat:not-a-column-value" }
                        ]
                    }
                }
            }]
        });

        let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

        assert!(
            err.to_string()
                .contains("ODRL allowed columns must be strings")
        );
    }

    #[test]
    fn read_restriction_rejects_empty_or_blank_allowed_column_lists() {
        for policy in [
            serde_json::json!({
                "uid": "policy-a",
                "lakecat:read-restriction": {
                    "allowed-columns": []
                }
            }),
            serde_json::json!({
                "uid": "policy-b",
                "lakecat:read-restriction": {
                    "allowed-columns": ["event_id", " "]
                }
            }),
            serde_json::json!({
                "uid": "policy-c",
                "permission": [{
                    "constraint": {
                        "leftOperand": { "@id": "lakecat:allowed-columns" },
                        "operator": { "@id": "odrl:isAnyOf" },
                        "rightOperand": {
                            "@list": [
                                { "@value": "event_id" },
                                { "@value": "" }
                            ]
                        }
                    }
                }]
            }),
        ] {
            let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

            assert!(
                err.to_string().contains("ODRL allowed columns must not be"),
                "unexpected error: {err}"
            );
        }
    }

    #[test]
    fn read_restriction_rejects_unsupported_odrl_constraint_operators() {
        let policy = serde_json::json!({
            "permission": [{
                "constraint": {
                    "leftOperand": "allowed-columns",
                    "operator": "neq",
                    "rightOperand": ["secret_payload"]
                }
            }]
        });

        let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

        assert!(
            err.to_string()
                .contains("ODRL allowed columns constraint uses unsupported operator")
        );
    }

    #[test]
    fn read_restriction_rejects_missing_odrl_constraint_operator() {
        let policy = serde_json::json!({
            "permission": [{
                "constraint": {
                    "leftOperand": "row-predicate",
                    "rightOperand": {
                        "type": "equal",
                        "term": "region",
                        "value": "west"
                    }
                }
            }]
        });

        let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

        assert!(
            err.to_string()
                .contains("ODRL row predicate constraint must include an operator")
        );
    }

    #[test]
    fn read_restriction_rejects_missing_odrl_constraint_right_operands() {
        for (left_operand_key, left_operand, label) in [
            ("leftOperand", "allowed-columns", "allowed columns"),
            ("leftOperand", "row-predicate", "row predicate"),
            ("leftOperand", "purpose", "purpose"),
            ("leftOperand", "credential-ttl", "max credential TTL"),
            ("odrl:leftOperand", "allowed-columns", "allowed columns"),
        ] {
            let mut constraint = serde_json::Map::new();
            constraint.insert(
                left_operand_key.to_string(),
                serde_json::Value::String(left_operand.to_string()),
            );
            constraint.insert(
                "operator".to_string(),
                serde_json::Value::String("eq".to_string()),
            );
            let policy = serde_json::json!({
                "permission": [{
                    "constraint": constraint
                }]
            });

            let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

            assert!(
                err.to_string().contains(&format!(
                    "ODRL {label} constraint must include a right operand"
                )),
                "unexpected error for {left_operand}: {err}"
            );
        }
    }

    #[test]
    fn read_restriction_rejects_non_equality_purpose_constraint() {
        let policy = serde_json::json!({
            "permission": [{
                "constraint": {
                    "leftOperand": "purpose",
                    "operator": "neq",
                    "rightOperand": "resilience-demo"
                }
            }]
        });

        let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

        assert!(
            err.to_string()
                .contains("ODRL purpose constraint uses unsupported operator")
        );
    }

    #[test]
    fn read_restriction_rejects_blank_odrl_purposes() {
        for policy in [
            serde_json::json!({
                "uid": "policy-a",
                "purpose": " "
            }),
            serde_json::json!({
                "uid": "policy-b",
                "permission": [{
                    "constraint": {
                        "leftOperand": "purpose",
                        "operator": "eq",
                        "rightOperand": { "@value": "" }
                    }
                }]
            }),
        ] {
            let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

            assert!(
                err.to_string().contains("ODRL purpose")
                    && err.to_string().contains("must not be blank"),
                "unexpected error: {err}"
            );
        }
    }

    #[test]
    fn read_restriction_rejects_conflicting_purpose_constraints() {
        let policy = serde_json::json!({
            "purpose": "resilience-demo",
            "permission": [{
                "constraint": {
                    "leftOperand": "purpose",
                    "operator": "eq",
                    "rightOperand": "training"
                }
            }]
        });

        let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

        assert!(
            err.to_string()
                .contains("ODRL read restriction carries conflicting purposes")
        );
    }

    #[test]
    fn read_restriction_rejects_conflicting_policy_purposes() {
        let policy_a = serde_json::json!({
            "uid": "policy-a",
            "purpose": "resilience-demo"
        });
        let policy_b = serde_json::json!({
            "uid": "policy-b",
            "permission": [{
                "constraint": {
                    "leftOperand": "purpose",
                    "operator": "eq",
                    "rightOperand": "training"
                }
            }]
        });

        let err = ReadRestriction::from_odrl_policies([&policy_a, &policy_b]).unwrap_err();

        assert!(
            err.to_string()
                .contains("ODRL read restriction carries conflicting purposes")
        );
    }

    #[test]
    fn read_restriction_rejects_non_numeric_ttl_constraints() {
        let policy = serde_json::json!({
            "permission": [{
                "constraint": {
                    "leftOperand": "credential-ttl",
                    "operator": "lteq",
                    "rightOperand": "five minutes"
                }
            }]
        });

        let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

        assert!(
            err.to_string()
                .contains("ODRL max credential TTL must be an unsigned integer")
        );
    }

    #[test]
    fn read_restriction_rejects_non_object_row_predicates() {
        let policy = serde_json::json!({
            "read-restriction": {
                "row-predicate": "severity >= 3"
            }
        });

        let err = ReadRestriction::from_odrl_policies([&policy]).unwrap_err();

        assert!(
            err.to_string()
                .contains("ODRL row predicate must be an Iceberg expression object")
        );
    }

    #[test]
    fn read_restriction_narrows_projection_stats_and_filters() {
        let row_predicate = serde_json::json!({
            "type": "equal",
            "term": "region",
            "value": "west"
        });
        let restriction = ReadRestriction {
            allowed_columns: Some(vec!["event_id".to_string(), "severity".to_string()]),
            row_predicate: Some(row_predicate.clone()),
            ..ReadRestriction::unrestricted()
        };

        assert_eq!(
            restriction.effective_projection(&[]).unwrap(),
            vec!["event_id".to_string(), "severity".to_string()]
        );
        assert_eq!(
            restriction
                .effective_projection(&["event_id".to_string(), "payload".to_string()])
                .unwrap(),
            vec!["event_id".to_string()]
        );
        assert!(
            restriction
                .effective_projection(&["payload".to_string()])
                .is_err()
        );
        assert_eq!(
            restriction.effective_stats_fields(&["payload".to_string(), "severity".to_string()]),
            vec!["severity".to_string()]
        );
        assert_eq!(restriction.mandatory_filters(), vec![row_predicate]);
    }

    #[test]
    fn table_capabilities_require_matching_allowed_receipts() {
        let table = TableIdent::new(
            WarehouseName::new("local").unwrap(),
            "default".parse::<Namespace>().unwrap(),
            TableName::new("events").unwrap(),
        );
        let receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:reader".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::TablePlanScan,
            table: Some(table.clone()),
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };

        let capability = TableScanCapability::from_receipt(receipt.clone(), table.clone())
            .expect("matching receipt should mint capability");
        assert_eq!(capability.table(), &table);
        assert_eq!(capability.receipt(), &receipt);

        let other_table = TableIdent::new(
            WarehouseName::new("local").unwrap(),
            "default".parse::<Namespace>().unwrap(),
            TableName::new("other").unwrap(),
        );
        assert!(TableScanCapability::from_receipt(receipt.clone(), other_table).is_err());

        let mut wrong_action_receipt = receipt;
        wrong_action_receipt.action = CatalogAction::TableLoad;
        assert!(TableScanCapability::from_receipt(wrong_action_receipt, table.clone()).is_err());

        let load_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:reader".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::TableLoad,
            table: Some(capability.table().clone()),
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        let load_capability =
            TableLoadCapability::from_receipt(load_receipt, capability.table().clone())
                .expect("matching load receipt should mint capability");
        assert_eq!(load_capability.table(), capability.table());

        let commit_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:writer".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::TableCommit,
            table: Some(capability.table().clone()),
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        let commit_capability =
            TableCommitCapability::from_receipt(commit_receipt, capability.table().clone())
                .expect("matching commit receipt should mint capability");
        assert_eq!(commit_capability.table(), capability.table());

        let drop_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:writer".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::TableDrop,
            table: Some(capability.table().clone()),
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        let drop_capability =
            TableDropCapability::from_receipt(drop_receipt, capability.table().clone())
                .expect("matching drop receipt should mint capability");
        assert_eq!(drop_capability.table(), capability.table());

        let create_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:writer".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::TableCreate,
            table: Some(capability.table().clone()),
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        let create_capability =
            TableCreateCapability::from_receipt(create_receipt, capability.table().clone())
                .expect("matching create receipt should mint capability");
        assert_eq!(create_capability.table(), capability.table());

        let credentials_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:reader".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::CredentialsVend,
            table: Some(capability.table().clone()),
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        let credentials_capability = CredentialsVendCapability::from_receipt(
            credentials_receipt,
            capability.table().clone(),
        )
        .expect("matching credential receipt should mint capability");
        assert_eq!(credentials_capability.table(), capability.table());

        let graph_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:querygraph".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::GraphRead,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        let graph_capability = GraphReadCapability::from_receipt(graph_receipt.clone())
            .expect("matching graph-read receipt should mint capability");
        assert_eq!(graph_capability.receipt(), &graph_receipt);

        let mut table_scoped_graph_receipt = graph_receipt;
        table_scoped_graph_receipt.table = Some(capability.table().clone());
        assert!(GraphReadCapability::from_receipt(table_scoped_graph_receipt).is_err());

        let config_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:catalog".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::CatalogConfig,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(CatalogConfigCapability::from_receipt(config_receipt).is_ok());

        let namespace_create_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:catalog".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::NamespaceCreate,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(NamespaceCreateCapability::from_receipt(namespace_create_receipt).is_ok());

        let namespace_list_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:catalog".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::NamespaceList,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(NamespaceListCapability::from_receipt(namespace_list_receipt).is_ok());

        let namespace_load_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:catalog".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::NamespaceLoad,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(NamespaceLoadCapability::from_receipt(namespace_load_receipt).is_ok());

        let namespace_drop_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:catalog".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::NamespaceDrop,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(NamespaceDropCapability::from_receipt(namespace_drop_receipt).is_ok());

        let server_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:operator".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::ServerManage,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(ServerManageCapability::from_receipt(server_receipt).is_ok());

        let project_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:operator".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::ProjectManage,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(ProjectManageCapability::from_receipt(project_receipt).is_ok());

        let warehouse_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:operator".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::WarehouseManage,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(WarehouseManageCapability::from_receipt(warehouse_receipt).is_ok());

        let storage_profile_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:operator".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::StorageProfileManage,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(StorageProfileManageCapability::from_receipt(storage_profile_receipt).is_ok());

        let view_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:operator".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::ViewManage,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(ViewManageCapability::from_receipt(view_receipt).is_ok());

        let view_load_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:reader".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::ViewLoad,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(ViewLoadCapability::from_receipt(view_load_receipt).is_ok());

        let view_drop_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:operator".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::ViewDrop,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(ViewDropCapability::from_receipt(view_drop_receipt).is_ok());

        let policy_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:operator".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::PolicyManage,
            table: None,
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(PolicyManageCapability::from_receipt(policy_receipt).is_ok());

        let restore_receipt = AuthorizationReceipt {
            principal: Principal {
                subject: "agent:operator".to_string(),
                kind: PrincipalKind::Agent,
            },
            action: CatalogAction::TableRestore,
            table: Some(table.clone()),
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: serde_json::json!({}),
            checked_at: Utc::now(),
        };
        assert!(TableRestoreCapability::from_receipt(restore_receipt, table).is_ok());
    }

    #[test]
    fn authorization_receipt_hashes_enforced_read_restriction_policies() {
        let table = TableIdent::new(
            WarehouseName::new("local").unwrap(),
            "default".parse::<Namespace>().unwrap(),
            TableName::new("events").unwrap(),
        );
        let policy_hash = content_hash_json(&json!({
            "uid": "policy:agent-read",
            "lakecat:read-restriction": {"allowed-columns": ["event_id"]}
        }))
        .unwrap();
        let receipt = AuthorizationReceipt {
            principal: Principal::anonymous(),
            action: CatalogAction::TablePlanScan,
            table: Some(table),
            allowed: true,
            engine: "test".to_string(),
            policy_hash: None,
            context: json!({
                "read-restriction": {
                    "allowed-columns": ["event_id"],
                    "policy-hashes": [policy_hash.clone()]
                }
            }),
            checked_at: Utc::now(),
        }
        .with_read_restriction_policy_hash()
        .unwrap();

        let receipt_policy_hash = receipt.policy_hash.as_deref().expect("receipt policy hash");
        assert_ne!(receipt_policy_hash, policy_hash);
        assert_eq!(
            receipt.context["read-restriction"]["policy-hashes"][0],
            policy_hash
        );
    }

    #[tokio::test]
    async fn allow_all_governance_receipt_names_local_engine() {
        let engine = AllowAllGovernanceEngine::new();
        let receipt = engine
            .authorize(AuthorizationRequest {
                principal: Principal::anonymous(),
                action: CatalogAction::CatalogConfig,
                table: None,
                context: json!({"warehouse": "local"}),
            })
            .await
            .expect("allow-all governance should authorize");

        assert!(receipt.allowed);
        assert_eq!(receipt.engine, ALLOW_ALL_LOCAL_ENGINE);
        assert!(!receipt.engine.contains("placeholder"));
        assert_eq!(receipt.context["warehouse"], "local");
    }
}

#[derive(Debug, Default)]
pub struct AllowAllGovernanceEngine;

impl AllowAllGovernanceEngine {
    pub fn new() -> Arc<Self> {
        Arc::new(Self)
    }
}

#[async_trait]
impl GovernanceEngine for AllowAllGovernanceEngine {
    async fn authorize(
        &self,
        request: AuthorizationRequest,
    ) -> LakeCatResult<AuthorizationReceipt> {
        Ok(AuthorizationReceipt {
            principal: request.principal,
            action: request.action,
            table: request.table,
            allowed: true,
            engine: ALLOW_ALL_LOCAL_ENGINE.to_string(),
            policy_hash: None,
            context: request.context,
            checked_at: Utc::now(),
        })
    }
}

#[cfg(feature = "typesec-local")]
pub mod typesec_integration {
    use std::sync::Arc;

    use async_trait::async_trait;
    use chrono::Utc;
    use lakecat_core::{LakeCatResult, content_hash_json};
    use typesec::{
        CombineStrategy, ComposedEngine, PolicyEngine, PolicyResult, ResourceId, SubjectId,
    };

    use crate::{
        AuthorizationReceipt, AuthorizationRequest, GovernanceEngine, action_name, resource_name,
    };

    pub struct TypeSecGovernanceEngine {
        engine: Arc<dyn PolicyEngine>,
    }

    impl TypeSecGovernanceEngine {
        pub fn new(engine: Arc<dyn PolicyEngine>) -> Arc<Self> {
            Arc::new(Self { engine })
        }

        pub fn with_fallback(
            primary: Arc<dyn PolicyEngine>,
            fallback: Arc<dyn PolicyEngine>,
        ) -> Arc<Self> {
            Arc::new(Self {
                engine: Arc::new(ComposedEngine::new(
                    vec![primary, fallback],
                    CombineStrategy::PriorityOrder,
                )),
            })
        }

        pub fn allow_all() -> Arc<Self> {
            Arc::new(Self {
                engine: Arc::new(AllowAllPolicy),
            })
        }

        pub fn rbac_from_yaml(yaml: &str) -> LakeCatResult<Arc<Self>> {
            let engine = typesec::RbacEngine::from_yaml(yaml).map_err(|err| {
                lakecat_core::LakeCatError::InvalidArgument(format!(
                    "failed to load TypeSec RBAC policy: {err}"
                ))
            })?;
            Ok(Self::new(Arc::new(engine)))
        }
    }

    struct AllowAllPolicy;

    impl PolicyEngine for AllowAllPolicy {
        fn check(
            &self,
            _subject: &SubjectId,
            _action: &str,
            _resource: &ResourceId,
        ) -> PolicyResult {
            PolicyResult::Allow
        }
    }

    #[async_trait]
    impl GovernanceEngine for TypeSecGovernanceEngine {
        async fn authorize(
            &self,
            request: AuthorizationRequest,
        ) -> LakeCatResult<AuthorizationReceipt> {
            let subject = SubjectId::from(request.principal.subject.clone());
            let action = action_name(&request.action);
            let resource = ResourceId::from(resource_name(&request));
            let decision = self.engine.check(&subject, action, &resource);
            let allowed = matches!(decision, PolicyResult::Allow);
            Ok(AuthorizationReceipt {
                principal: request.principal,
                action: request.action,
                table: request.table,
                allowed,
                engine: "typesec".to_string(),
                policy_hash: Some(content_hash_json(&serde_json::json!({
                    "engine": "typesec",
                    "subject": subject.as_str(),
                    "action": action,
                    "resource": resource.as_str(),
                    "decision": policy_result_name(&decision),
                    "context-hash": content_hash_json(&request.context)?,
                }))?),
                context: request.context,
                checked_at: Utc::now(),
            })
        }
    }

    fn policy_result_name(result: &PolicyResult) -> &'static str {
        match result {
            PolicyResult::Allow => "allow",
            PolicyResult::Deny(_) => "deny",
            PolicyResult::Delegate(_) => "delegate",
            _ => "unknown",
        }
    }

    #[cfg(test)]
    mod tests {
        use std::sync::Arc;

        use lakecat_core::{Principal, PrincipalKind};
        use typesec::{PolicyEngine, PolicyResult, ResourceId, SubjectId};

        use super::*;
        use crate::{AuthorizationRequest, CatalogAction};

        struct AllowRead;
        struct DelegateToRbac;

        impl PolicyEngine for AllowRead {
            fn check(
                &self,
                _subject: &SubjectId,
                action: &str,
                _resource: &ResourceId,
            ) -> PolicyResult {
                if action == "table.load" {
                    PolicyResult::Allow
                } else {
                    PolicyResult::Deny("not granted".to_string())
                }
            }
        }

        impl PolicyEngine for DelegateToRbac {
            fn check(
                &self,
                _subject: &SubjectId,
                _action: &str,
                _resource: &ResourceId,
            ) -> PolicyResult {
                PolicyResult::delegate("odrl", "rbac decides base access")
            }
        }

        #[tokio::test]
        async fn delegates_authorization_to_typesec_policy_engine() {
            let engine = TypeSecGovernanceEngine::new(Arc::new(AllowRead));
            let receipt = engine
                .authorize(AuthorizationRequest {
                    principal: Principal {
                        subject: "agent:reader".to_string(),
                        kind: PrincipalKind::Agent,
                    },
                    action: CatalogAction::TableLoad,
                    table: None,
                    context: serde_json::json!({}),
                })
                .await
                .expect("authorization should run");
            assert!(receipt.allowed);
            assert_eq!(receipt.engine, "typesec");
            assert!(receipt.policy_hash.is_some());
        }

        #[tokio::test]
        async fn delegates_to_typesec_fallback_policy_engine() {
            let engine = TypeSecGovernanceEngine::with_fallback(
                Arc::new(DelegateToRbac),
                Arc::new(AllowRead),
            );
            let receipt = engine
                .authorize(AuthorizationRequest {
                    principal: Principal {
                        subject: "agent:reader".to_string(),
                        kind: PrincipalKind::Agent,
                    },
                    action: CatalogAction::TableLoad,
                    table: None,
                    context: serde_json::json!({"read-restriction": {"allowed-columns": ["id"]}}),
                })
                .await
                .expect("authorization should run through TypeSec fallback");
            assert!(receipt.allowed);
            assert_eq!(receipt.engine, "typesec");
            assert!(receipt.policy_hash.is_some());
            assert_eq!(
                receipt.context["read-restriction"]["allowed-columns"][0],
                serde_json::json!("id")
            );
        }

        #[tokio::test]
        async fn loads_rbac_policy_yaml_for_authorization() {
            let engine = TypeSecGovernanceEngine::rbac_from_yaml(
                r#"
roles:
  - name: scanner
    permissions: ["table.plan_scan"]
    resources: ["lakecat:table:local:default:events"]
assignments:
  - subject: "agent:scanner"
    roles: [scanner]
"#,
            )
            .expect("rbac policy should load");
            let table = lakecat_core::TableIdent::new(
                lakecat_core::WarehouseName::new("local").unwrap(),
                "default".parse::<lakecat_core::Namespace>().unwrap(),
                lakecat_core::TableName::new("events").unwrap(),
            );
            let receipt = engine
                .authorize(AuthorizationRequest {
                    principal: lakecat_core::Principal::new(
                        "agent:scanner",
                        lakecat_core::PrincipalKind::Agent,
                    )
                    .unwrap(),
                    action: CatalogAction::TablePlanScan,
                    table: Some(table),
                    context: serde_json::json!({}),
                })
                .await
                .unwrap();

            assert!(receipt.allowed);
            assert_eq!(receipt.engine, "typesec");
        }

        #[test]
        fn rejects_invalid_rbac_policy_yaml() {
            let error = match TypeSecGovernanceEngine::rbac_from_yaml(
                r#"
roles:
  - name: broken
    inherits: [missing]
"#,
            ) {
                Ok(_) => panic!("invalid rbac policy should fail closed"),
                Err(error) => error,
            };

            assert!(
                error
                    .to_string()
                    .contains("failed to load TypeSec RBAC policy")
            );
        }
    }
}

pub fn action_name(action: &CatalogAction) -> &'static str {
    match action {
        CatalogAction::CatalogConfig => "catalog.config",
        CatalogAction::NamespaceCreate => "namespace.create",
        CatalogAction::NamespaceList => "namespace.list",
        CatalogAction::NamespaceLoad => "namespace.load",
        CatalogAction::NamespaceDrop => "namespace.drop",
        CatalogAction::TableCreate => "table.create",
        CatalogAction::TableRegister => "table.register",
        CatalogAction::TableLoad => "table.load",
        CatalogAction::TablePlanScan => "table.plan_scan",
        CatalogAction::TableCommit => "table.commit",
        CatalogAction::TableDrop => "table.drop",
        CatalogAction::TableRestore => "table.restore",
        CatalogAction::CredentialsVend => "credentials.vend",
        CatalogAction::ViewLoad => "view.load",
        CatalogAction::ViewDrop => "view.drop",
        CatalogAction::ServerManage => "server.manage",
        CatalogAction::ProjectManage => "project.manage",
        CatalogAction::WarehouseManage => "warehouse.manage",
        CatalogAction::StorageProfileManage => "storage_profile.manage",
        CatalogAction::ViewManage => "view.manage",
        CatalogAction::PolicyManage => "policy.manage",
        CatalogAction::GraphRead => "graph.read",
        CatalogAction::LineageRead => "lineage.read",
    }
}

pub fn resource_name(request: &AuthorizationRequest) -> String {
    request
        .table
        .as_ref()
        .map(TableIdent::stable_id)
        .unwrap_or_else(|| "lakecat:catalog".to_string())
}
