//! Caller permission checks for commits: which [`Permission`] each mutation
//! requires, whether a caller may administer a grant, and permission
//! validation for read assertions.

use super::Mutation;
use crate::catalog::{Catalog, DdlOperation, ResourceType};
use crate::error::AedbError;
use crate::permission::{CallerContext, Permission};

pub fn validate_permissions(
    catalog: &Catalog,
    caller: Option<&CallerContext>,
    mutation: &Mutation,
) -> Result<(), AedbError> {
    let Some(caller) = caller else {
        return Ok(());
    };
    if matches!(mutation, Mutation::OrderBookMatch { .. }) {
        if caller.is_internal_system() {
            return Ok(());
        }
        return Err(AedbError::PermissionDenied(
            "OrderBookMatch is system-only".into(),
        ));
    }
    if let Mutation::PostflightCheck { assertions } = mutation {
        if caller.is_internal_system() {
            return Ok(());
        }
        return validate_assertion_permissions(catalog, caller, assertions);
    }
    let is_admin = catalog.has_permission(&caller.caller_id, &Permission::GlobalAdmin);
    match mutation {
        Mutation::OrderBookNew { request, .. } => {
            if !is_admin && request.owner != caller.caller_id {
                return Err(AedbError::PermissionDenied(
                    "order owner must match caller".into(),
                ));
            }
        }
        Mutation::OrderBookCancel { owner, .. }
        | Mutation::OrderBookCancelReplace { owner, .. }
        | Mutation::OrderBookReduce { owner, .. }
        | Mutation::OrderBookMassCancel { owner, .. }
            if !is_admin && owner != &caller.caller_id =>
        {
            return Err(AedbError::PermissionDenied(
                "order owner must match caller".into(),
            ));
        }
        _ => {}
    }
    if let Some((project_id, scope_id, key)) = kv_write_target(mutation) {
        if catalog.has_kv_write_permission(&caller.caller_id, project_id, scope_id, key) {
            return Ok(());
        }
        return Err(AedbError::PermissionDenied("permission denied".into()));
    }
    if let Mutation::Ddl(ddl) = mutation {
        match ddl {
            DdlOperation::GrantPermission { permission, .. }
            | DdlOperation::RevokePermission { permission, .. } => {
                if can_administer_permission(catalog, &caller.caller_id, permission) {
                    return Ok(());
                }
                return Err(AedbError::PermissionDenied("permission denied".into()));
            }
            DdlOperation::TransferOwnership {
                resource_type,
                project_id,
                scope_id,
                table_name,
                ..
            } => {
                let allowed = catalog.has_permission(&caller.caller_id, &Permission::GlobalAdmin)
                    || match resource_type {
                        ResourceType::Project => {
                            catalog.is_owner_of_project(&caller.caller_id, project_id)
                        }
                        ResourceType::Scope => scope_id.as_ref().is_some_and(|scope| {
                            catalog.is_owner_of_scope(&caller.caller_id, project_id, scope)
                                || catalog.is_owner_of_project(&caller.caller_id, project_id)
                        }),
                        ResourceType::Table => match (scope_id.as_ref(), table_name.as_ref()) {
                            (Some(scope), Some(table)) => {
                                catalog.is_owner_of_table(
                                    &caller.caller_id,
                                    project_id,
                                    scope,
                                    table,
                                ) || catalog.is_owner_of_project(&caller.caller_id, project_id)
                            }
                            _ => false,
                        },
                    };
                if allowed {
                    return Ok(());
                }
                return Err(AedbError::PermissionDenied("permission denied".into()));
            }
            _ => {}
        }
    }
    let required = required_permission(mutation)?;
    if catalog.has_permission(&caller.caller_id, &required) {
        return Ok(());
    }
    Err(AedbError::PermissionDenied("permission denied".into()))
}

fn kv_write_target(mutation: &Mutation) -> Option<(&str, &str, &[u8])> {
    match mutation {
        Mutation::KvSet {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvDel {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvIncU256 {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvDecU256 {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvAddU256Ex {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvSubU256Ex {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvMaxU256 {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvMinU256 {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvMutateU256 {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvAddU64Ex {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvSubU64Ex {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvMaxU64 {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvMinU64 {
            project_id,
            scope_id,
            key,
            ..
        }
        | Mutation::KvMutateU64 {
            project_id,
            scope_id,
            key,
            ..
        } => Some((project_id.as_str(), scope_id.as_str(), key.as_slice())),
        Mutation::EmitEvent {
            project_id,
            scope_id,
            topic,
            ..
        } => Some((project_id.as_str(), scope_id.as_str(), topic.as_bytes())),
        Mutation::OrderBookNew {
            project_id,
            scope_id,
            request,
        } => Some((
            project_id.as_str(),
            scope_id.as_str(),
            request.instrument.as_bytes(),
        )),
        Mutation::OrderBookCancel {
            project_id,
            scope_id,
            instrument,
            ..
        }
        | Mutation::OrderBookCancelReplace {
            project_id,
            scope_id,
            instrument,
            ..
        }
        | Mutation::OrderBookMassCancel {
            project_id,
            scope_id,
            instrument,
            ..
        }
        | Mutation::OrderBookReduce {
            project_id,
            scope_id,
            instrument,
            ..
        }
        | Mutation::OrderBookMatch {
            project_id,
            scope_id,
            instrument,
            ..
        } => Some((
            project_id.as_str(),
            scope_id.as_str(),
            instrument.as_bytes(),
        )),
        Mutation::OrderBookDefineTable {
            project_id,
            scope_id,
            table_id,
            ..
        }
        | Mutation::OrderBookDropTable {
            project_id,
            scope_id,
            table_id,
            ..
        } => Some((project_id.as_str(), scope_id.as_str(), table_id.as_bytes())),
        Mutation::OrderBookSetInstrumentConfig {
            project_id,
            scope_id,
            instrument,
            ..
        }
        | Mutation::OrderBookSetInstrumentHalted {
            project_id,
            scope_id,
            instrument,
            ..
        } => Some((
            project_id.as_str(),
            scope_id.as_str(),
            instrument.as_bytes(),
        )),
        _ => None,
    }
}

pub fn required_permission(mutation: &Mutation) -> Result<Permission, AedbError> {
    match mutation {
        Mutation::Insert {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::Upsert {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::InsertBatch {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::UpsertBatch {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::UpsertOnConflict {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::UpsertBatchOnConflict {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::Delete {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::DeleteWhere {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::UpdateWhere {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::UpdateWhereExpr {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::TableIncU256 {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::TableDecU256 {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | Mutation::UpdateFields {
            project_id,
            scope_id,
            table_name,
            ..
        } => Ok(Permission::TableWrite {
            project_id: project_id.clone(),
            scope_id: scope_id.clone(),
            table_name: table_name.clone(),
        }),
        Mutation::Ddl(ddl) => match ddl {
            DdlOperation::CreateProject { .. } => Ok(Permission::GlobalAdmin),
            DdlOperation::DropProject { project_id, .. } => Ok(Permission::ProjectAdmin {
                project_id: project_id.clone(),
            }),
            DdlOperation::GrantPermission { .. } | DdlOperation::RevokePermission { .. } => {
                Ok(Permission::GlobalAdmin)
            }
            DdlOperation::CreateScope { project_id, .. } => Ok(Permission::ProjectAdmin {
                project_id: project_id.clone(),
            }),
            DdlOperation::DropScope {
                project_id,
                scope_id,
                ..
            } => Ok(Permission::ScopeAdmin {
                project_id: project_id.clone(),
                scope_id: scope_id.clone(),
            }),
            DdlOperation::CreateTable { project_id, .. }
            | DdlOperation::AlterTable { project_id, .. }
            | DdlOperation::DropTable { project_id, .. }
            | DdlOperation::CreateIndex { project_id, .. }
            | DdlOperation::DropIndex { project_id, .. }
            | DdlOperation::CreateAsyncIndex { project_id, .. }
            | DdlOperation::DropAsyncIndex { project_id, .. }
            | DdlOperation::EnableKvProjection { project_id, .. }
            | DdlOperation::DisableKvProjection { project_id, .. }
            | DdlOperation::SetReadPolicy { project_id, .. }
            | DdlOperation::ClearReadPolicy { project_id, .. } => Ok(Permission::TableDdl {
                project_id: project_id.clone(),
            }),
            DdlOperation::TransferOwnership {
                resource_type,
                project_id,
                scope_id,
                table_name: _,
                ..
            } => match resource_type {
                ResourceType::Project => Ok(Permission::ProjectAdmin {
                    project_id: project_id.clone(),
                }),
                ResourceType::Scope => Ok(Permission::ScopeAdmin {
                    project_id: project_id.clone(),
                    scope_id: scope_id.clone().ok_or_else(|| {
                        AedbError::Validation("scope transfer requires scope_id".into())
                    })?,
                }),
                ResourceType::Table => Ok(Permission::TableDdl {
                    project_id: project_id.clone(),
                }),
            },
        },
        Mutation::KvSet {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvDel {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvIncU256 {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvDecU256 {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvAddU256Ex {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvSubU256Ex {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvMaxU256 {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvMinU256 {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvMutateU256 {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvAddU64Ex {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvSubU64Ex {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvSubIntEx {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvAddI64Bounded {
            project_id,
            scope_id,
            ..
        }
        | Mutation::CounterAdd {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvMaxU64 {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvMinU64 {
            project_id,
            scope_id,
            ..
        }
        | Mutation::KvMutateU64 {
            project_id,
            scope_id,
            ..
        }
        | Mutation::EmitEvent {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookNew {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookCancel {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookCancelReplace {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookMassCancel {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookReduce {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookMatch {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookDefineTable {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookDropTable {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookSetInstrumentConfig {
            project_id,
            scope_id,
            ..
        }
        | Mutation::OrderBookSetInstrumentHalted {
            project_id,
            scope_id,
            ..
        } => Ok(Permission::KvWrite {
            project_id: project_id.clone(),
            scope_id: Some(scope_id.clone()),
            prefix: None,
        }),
        Mutation::PostflightCheck { .. } => Ok(Permission::GlobalAdmin),
    }
}

fn can_administer_permission(catalog: &Catalog, caller_id: &str, permission: &Permission) -> bool {
    if catalog.has_permission(caller_id, &Permission::GlobalAdmin) {
        return true;
    }
    match permission {
        Permission::GlobalAdmin => false,
        Permission::ProjectAdmin { project_id } => {
            catalog.is_owner_of_project(caller_id, project_id)
        }
        Permission::TableDdl { project_id }
        | Permission::KvRead {
            project_id,
            scope_id: None,
            ..
        }
        | Permission::KvWrite {
            project_id,
            scope_id: None,
            ..
        }
        | Permission::PolicyBypass {
            project_id,
            table_name: None,
        } => {
            catalog.is_owner_of_project(caller_id, project_id)
                || catalog.has_delegable_grant(caller_id, permission)
                || catalog.has_permission(
                    caller_id,
                    &Permission::ProjectAdmin {
                        project_id: project_id.clone(),
                    },
                )
        }
        Permission::ScopeAdmin {
            project_id,
            scope_id,
        }
        | Permission::KvRead {
            project_id,
            scope_id: Some(scope_id),
            ..
        }
        | Permission::KvWrite {
            project_id,
            scope_id: Some(scope_id),
            ..
        }
        | Permission::TableRead {
            project_id,
            scope_id,
            ..
        }
        | Permission::TableWrite {
            project_id,
            scope_id,
            ..
        }
        | Permission::IndexRead {
            project_id,
            scope_id,
            ..
        } => {
            let has_scope_owner = catalog.is_owner_of_scope(caller_id, project_id, scope_id);
            let has_table_owner = match permission {
                Permission::TableRead { table_name, .. }
                | Permission::TableWrite { table_name, .. }
                | Permission::IndexRead { table_name, .. }
                | Permission::PolicyBypass {
                    table_name: Some(table_name),
                    ..
                } => catalog.is_owner_of_table(caller_id, project_id, scope_id, table_name),
                _ => false,
            };
            catalog.has_permission(
                caller_id,
                &Permission::ProjectAdmin {
                    project_id: project_id.clone(),
                },
            ) || catalog.has_permission(
                caller_id,
                &Permission::ScopeAdmin {
                    project_id: project_id.clone(),
                    scope_id: scope_id.clone(),
                },
            ) || has_scope_owner
                || has_table_owner
                || catalog.has_delegable_grant(caller_id, permission)
        }
        Permission::PolicyBypass {
            project_id,
            table_name: Some(_),
        } => {
            catalog.has_permission(
                caller_id,
                &Permission::ProjectAdmin {
                    project_id: project_id.clone(),
                },
            ) || catalog.is_owner_of_project(caller_id, project_id)
                || catalog.has_delegable_grant(caller_id, permission)
        }
    }
}

pub(crate) fn validate_assertion_permissions(
    catalog: &Catalog,
    caller: &CallerContext,
    assertions: &[crate::commit::tx::ReadAssertion],
) -> Result<(), AedbError> {
    for assertion in assertions {
        validate_assertion_permission(catalog, caller, assertion)?;
    }
    Ok(())
}

fn validate_assertion_permission(
    catalog: &Catalog,
    caller: &CallerContext,
    assertion: &crate::commit::tx::ReadAssertion,
) -> Result<(), AedbError> {
    use crate::commit::tx::ReadAssertion;

    match assertion {
        ReadAssertion::KeyEquals {
            project_id,
            scope_id,
            key,
            ..
        }
        | ReadAssertion::KeyCompare {
            project_id,
            scope_id,
            key,
            ..
        }
        | ReadAssertion::KeyExists {
            project_id,
            scope_id,
            key,
            ..
        }
        | ReadAssertion::KeyVersion {
            project_id,
            scope_id,
            key,
            ..
        } => {
            if catalog.has_kv_read_permission(&caller.caller_id, project_id, scope_id, key) {
                Ok(())
            } else {
                Err(AedbError::PermissionDenied("permission denied".into()))
            }
        }
        ReadAssertion::RowVersion {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | ReadAssertion::RowExists {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | ReadAssertion::RowColumnCompare {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | ReadAssertion::CountCompare {
            project_id,
            scope_id,
            table_name,
            ..
        }
        | ReadAssertion::SumCompare {
            project_id,
            scope_id,
            table_name,
            ..
        } => {
            let required = Permission::TableRead {
                project_id: project_id.clone(),
                scope_id: scope_id.clone(),
                table_name: table_name.clone(),
            };
            if catalog.has_permission(&caller.caller_id, &required) {
                Ok(())
            } else {
                Err(AedbError::PermissionDenied("permission denied".into()))
            }
        }
        ReadAssertion::All(assertions) | ReadAssertion::Any(assertions) => {
            validate_assertion_permissions(catalog, caller, assertions)
        }
        ReadAssertion::Not(assertion) => validate_assertion_permission(catalog, caller, assertion),
    }
}
