//! Foreign-key enforcement for the apply path: outbound reference checks on
//! writes, referencing-row handling (restrict / cascade / set-null) on deletes
//! and key updates, cascade depth limits, and FK definition prevalidation.

use super::{
    apply_delete_internal_encoded, extract_primary_key_from_row, split_namespace,
    validate_row_constraints,
};
use crate::catalog::schema::{Constraint, ForeignKeyAction, TableSchema};
use crate::catalog::types::{Row, Value};
use crate::catalog::{Catalog, namespace_key};
use crate::error::AedbError;
use crate::error::ResourceType as ErrorResourceType;
use crate::storage::encoded_key::EncodedKey;
use crate::storage::keyspace::Keyspace;

const MAX_CASCADE_DELETE_DEPTH: usize = 8;

pub(super) fn ensure_cascade_delete_depth(cascade_depth: usize) -> Result<(), AedbError> {
    if cascade_depth > MAX_CASCADE_DELETE_DEPTH {
        return Err(AedbError::Validation(format!(
            "cascade delete depth exceeded: depth={}, max_depth={}",
            cascade_depth, MAX_CASCADE_DELETE_DEPTH
        )));
    }
    Ok(())
}

pub(super) fn validate_foreign_keys(
    catalog: &Catalog,
    keyspace: &Keyspace,
    schema: &TableSchema,
    row: &Row,
    _old_row: Option<&Row>,
) -> Result<(), AedbError> {
    for fk in &schema.foreign_keys {
        let Some(ref_schema) = catalog.tables.get(&(
            namespace_key(&fk.references_project_id, &fk.references_scope_id),
            fk.references_table.clone(),
        )) else {
            return Err(AedbError::NotFound {
                resource_type: ErrorResourceType::Table,
                resource_id: format!(
                    "{}.{}.{}",
                    fk.references_project_id, fk.references_scope_id, fk.references_table
                ),
            });
        };
        let mut values = Vec::with_capacity(fk.columns.len());
        let mut any_null = false;
        for col in &fk.columns {
            let column_index = schema
                .columns
                .iter()
                .position(|c| c.name == *col)
                .ok_or_else(|| AedbError::UnknownColumn {
                    table: schema.table_name.clone(),
                    column: col.clone(),
                })?;
            let value = row.values.get(column_index).cloned().unwrap_or(Value::Null);
            if matches!(value, Value::Null) {
                any_null = true;
            }
            values.push(value);
        }
        if any_null {
            continue;
        }

        if ref_schema.primary_key == fk.references_columns {
            let exists = keyspace
                .row_slot(
                    &fk.references_project_id,
                    &fk.references_scope_id,
                    &fk.references_table,
                    &EncodedKey::from_values(&values),
                )
                .is_some();
            if !exists {
                return Err(AedbError::ForeignKeyViolation {
                    fk_name: fk.name.clone(),
                    table: schema.table_name.clone(),
                    ref_table: fk.references_table.clone(),
                    ref_key: format!("{values:?}"),
                });
            }
        } else {
            let ref_ns = namespace_key(&fk.references_project_id, &fk.references_scope_id);
            let lookup_key = EncodedKey::from_values(&values);
            let index_name = catalog
                .indexes
                .iter()
                .find(|((idx_ns, idx_table, _), idx_def)| {
                    idx_ns == &ref_ns
                        && idx_table == &fk.references_table
                        && idx_def.columns == fk.references_columns
                        && matches!(
                            idx_def.index_type,
                            crate::catalog::schema::IndexType::UniqueHash
                        )
                })
                .map(|((_, _, idx_name), _)| idx_name.clone())
                .ok_or_else(|| AedbError::IntegrityError {
                    message: format!(
                        "non-primary-key foreign key {} references {} without a unique runtime index",
                        fk.name, fk.references_table
                    ),
                })?;
            let ref_table = keyspace
                .table_by_namespace_key(&ref_ns, &fk.references_table)
                .ok_or_else(|| AedbError::NotFound {
                    resource_type: ErrorResourceType::Table,
                    resource_id: format!(
                        "{}.{}.{}",
                        fk.references_project_id, fk.references_scope_id, fk.references_table
                    ),
                })?;
            let reference_unique_index = ref_table
                .indexes
                .get(&index_name)
                .ok_or_else(|| AedbError::IntegrityError {
                    message: format!(
                        "missing runtime unique index {index_name} for non-primary-key foreign key {}",
                        fk.name
                    ),
                })?;
            if reference_unique_index
                .tier_unique_existing(&lookup_key, keyspace.kv_segment_store.as_deref())?
                .is_none()
            {
                return Err(AedbError::ForeignKeyViolation {
                    fk_name: fk.name.clone(),
                    table: schema.table_name.clone(),
                    ref_table: fk.references_table.clone(),
                    ref_key: format!("{values:?}"),
                });
            }
            continue;
        }
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
pub(super) fn handle_referencing_foreign_keys(
    catalog: &Catalog,
    keyspace: &mut Keyspace,
    ref_project_id: &str,
    ref_scope_id: &str,
    ref_table_name: &str,
    ref_row: &Row,
    commit_seq: u64,
    cascade_depth: usize,
) -> Result<(), AedbError> {
    ensure_cascade_delete_depth(cascade_depth)?;
    let target_ns = namespace_key(ref_project_id, ref_scope_id);
    for ((dep_ns, dep_table_name), dep_schema) in &catalog.tables {
        for fk in &dep_schema.foreign_keys {
            if namespace_key(&fk.references_project_id, &fk.references_scope_id) != target_ns
                || fk.references_table != ref_table_name
            {
                continue;
            }
            let mut referenced_vals = Vec::with_capacity(fk.references_columns.len());
            for ref_col in &fk.references_columns {
                let reference_column_index = catalog
                    .tables
                    .get(&(target_ns.clone(), ref_table_name.to_string()))
                    .and_then(|s| s.columns.iter().position(|c| c.name == *ref_col))
                    .ok_or_else(|| {
                        AedbError::Validation(format!(
                            "referenced column missing for fk {}",
                            fk.name
                        ))
                    })?;
                referenced_vals.push(ref_row.values[reference_column_index].clone());
            }
            let Some(dep_table) = keyspace
                .table_by_namespace_key(dep_ns, dep_table_name)
                .cloned()
            else {
                continue;
            };
            let mut matched_pks: Vec<EncodedKey> = Vec::new();
            for (pk, stored) in dep_table.rows {
                let row = keyspace.materialize_row(&stored)?;
                let mut matched = true;
                for (i, fk_col) in fk.columns.iter().enumerate() {
                    let fk_idx = dep_schema
                        .columns
                        .iter()
                        .position(|c| c.name == *fk_col)
                        .ok_or_else(|| AedbError::Validation("fk column missing".into()))?;
                    if row.values.get(fk_idx) != referenced_vals.get(i) {
                        matched = false;
                        break;
                    }
                }
                if matched {
                    matched_pks.push(pk);
                }
            }
            if matched_pks.is_empty() {
                continue;
            }
            match fk.on_delete {
                ForeignKeyAction::Restrict | ForeignKeyAction::NoAction => {
                    return Err(AedbError::ForeignKeyViolation {
                        fk_name: fk.name.clone(),
                        table: dep_table_name.clone(),
                        ref_table: ref_table_name.to_string(),
                        ref_key: "restrict".into(),
                    });
                }
                ForeignKeyAction::Cascade => {
                    for pk_encoded in matched_pks {
                        let (dep_project_id, dep_scope_id) = split_namespace(dep_ns)
                            .ok_or_else(|| AedbError::Validation("invalid namespace key".into()))?;
                        apply_delete_internal_encoded(
                            catalog,
                            keyspace,
                            &dep_project_id,
                            &dep_scope_id,
                            dep_table_name,
                            &pk_encoded,
                            commit_seq,
                            cascade_depth.saturating_add(1),
                        )?;
                    }
                }
                ForeignKeyAction::SetNull | ForeignKeyAction::SetDefault => {
                    for pk_encoded in matched_pks {
                        let (dep_project_id, dep_scope_id) = split_namespace(dep_ns)
                            .ok_or_else(|| AedbError::Validation("invalid namespace key".into()))?;
                        let dep_ns_key = namespace_key(&dep_project_id, &dep_scope_id);
                        let dep_schema = catalog
                            .tables
                            .get(&(dep_ns_key.clone(), dep_table_name.clone()))
                            .ok_or_else(|| {
                                AedbError::Validation("dependent table missing".into())
                            })?;
                        let mut row = keyspace
                            .get_row_by_encoded(
                                &dep_project_id,
                                &dep_scope_id,
                                dep_table_name,
                                &pk_encoded,
                            )?
                            .map(|row| row.into_owned())
                            .ok_or_else(|| AedbError::Validation("dependent row missing".into()))?;
                        for fk_col in &fk.columns {
                            if let Some(column_index) =
                                dep_schema.columns.iter().position(|c| c.name == *fk_col)
                            {
                                match fk.on_delete {
                                    ForeignKeyAction::SetNull => {
                                        row.values[column_index] = Value::Null
                                    }
                                    ForeignKeyAction::SetDefault => {
                                        if let Some(default_value) =
                                            dep_schema.constraints.iter().find_map(|c| match c {
                                                Constraint::Default { column, value }
                                                    if column == fk_col =>
                                                {
                                                    Some(value.clone())
                                                }
                                                _ => None,
                                            })
                                        {
                                            row.values[column_index] = default_value;
                                        } else {
                                            row.values[column_index] = Value::Null;
                                        }
                                    }
                                    _ => {}
                                }
                            }
                        }
                        validate_row_constraints(
                            catalog,
                            keyspace,
                            dep_schema,
                            &dep_project_id,
                            &dep_scope_id,
                            dep_table_name,
                            &extract_primary_key_from_row(dep_schema, &row)?,
                            &row,
                            None,
                        )?;
                        keyspace.upsert_row_by_encoded_pk(
                            &dep_project_id,
                            &dep_scope_id,
                            dep_table_name,
                            pk_encoded.clone(),
                            row.clone(),
                            commit_seq,
                        );
                    }
                }
            }
        }
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
pub(super) fn handle_referencing_foreign_keys_on_update(
    catalog: &Catalog,
    keyspace: &mut Keyspace,
    ref_project_id: &str,
    ref_scope_id: &str,
    ref_table_name: &str,
    ref_schema: &TableSchema,
    old_row: &Row,
    new_row: &Row,
    commit_seq: u64,
) -> Result<(), AedbError> {
    let target_ns = namespace_key(ref_project_id, ref_scope_id);
    for ((dep_ns, dep_table_name), dep_schema) in &catalog.tables {
        for fk in &dep_schema.foreign_keys {
            if namespace_key(&fk.references_project_id, &fk.references_scope_id) != target_ns
                || fk.references_table != ref_table_name
            {
                continue;
            }
            let mut old_values = Vec::with_capacity(fk.references_columns.len());
            let mut new_values = Vec::with_capacity(fk.references_columns.len());
            for ref_col in &fk.references_columns {
                let reference_column_index = ref_schema
                    .columns
                    .iter()
                    .position(|c| c.name == *ref_col)
                    .ok_or_else(|| AedbError::Validation("referenced column missing".into()))?;
                old_values.push(old_row.values[reference_column_index].clone());
                new_values.push(new_row.values[reference_column_index].clone());
            }
            if old_values == new_values {
                continue;
            }
            let Some(dep_table) = keyspace
                .table_by_namespace_key(dep_ns, dep_table_name)
                .cloned()
            else {
                continue;
            };
            let mut matched_pks: Vec<EncodedKey> = Vec::new();
            for (pk, stored) in dep_table.rows {
                let row = keyspace.materialize_row(&stored)?;
                let mut matched = true;
                for (i, fk_col) in fk.columns.iter().enumerate() {
                    let fk_idx = dep_schema
                        .columns
                        .iter()
                        .position(|c| c.name == *fk_col)
                        .ok_or_else(|| AedbError::Validation("fk column missing".into()))?;
                    if row.values.get(fk_idx) != old_values.get(i) {
                        matched = false;
                        break;
                    }
                }
                if matched {
                    matched_pks.push(pk);
                }
            }
            if matched_pks.is_empty() {
                continue;
            }
            match fk.on_update {
                ForeignKeyAction::Restrict | ForeignKeyAction::NoAction => {
                    return Err(AedbError::Validation(format!(
                        "foreign key update restrict violation: {}",
                        fk.name
                    )));
                }
                ForeignKeyAction::Cascade
                | ForeignKeyAction::SetNull
                | ForeignKeyAction::SetDefault => {
                    for pk_encoded in matched_pks {
                        let (dep_project_id, dep_scope_id) = split_namespace(dep_ns)
                            .ok_or_else(|| AedbError::Validation("invalid namespace key".into()))?;
                        let dep_schema = catalog
                            .tables
                            .get(&(
                                namespace_key(&dep_project_id, &dep_scope_id),
                                dep_table_name.clone(),
                            ))
                            .ok_or_else(|| {
                                AedbError::Validation("dependent table missing".into())
                            })?;
                        let mut row = keyspace
                            .get_row_by_encoded(
                                &dep_project_id,
                                &dep_scope_id,
                                dep_table_name,
                                &pk_encoded,
                            )?
                            .map(|row| row.into_owned())
                            .ok_or_else(|| AedbError::Validation("dependent row missing".into()))?;
                        for (i, fk_col) in fk.columns.iter().enumerate() {
                            if let Some(column_index) =
                                dep_schema.columns.iter().position(|c| c.name == *fk_col)
                            {
                                match fk.on_update {
                                    ForeignKeyAction::Cascade => {
                                        row.values[column_index] = new_values[i].clone()
                                    }
                                    ForeignKeyAction::SetNull => {
                                        row.values[column_index] = Value::Null
                                    }
                                    ForeignKeyAction::SetDefault => {
                                        row.values[column_index] = dep_schema
                                            .constraints
                                            .iter()
                                            .find_map(|c| match c {
                                                Constraint::Default { column, value }
                                                    if column == fk_col =>
                                                {
                                                    Some(value.clone())
                                                }
                                                _ => None,
                                            })
                                            .unwrap_or(Value::Null);
                                    }
                                    _ => {}
                                }
                            }
                        }
                        validate_row_constraints(
                            catalog,
                            keyspace,
                            dep_schema,
                            &dep_project_id,
                            &dep_scope_id,
                            dep_table_name,
                            &extract_primary_key_from_row(dep_schema, &row)?,
                            &row,
                            None,
                        )?;
                        keyspace.upsert_row_by_encoded_pk(
                            &dep_project_id,
                            &dep_scope_id,
                            dep_table_name,
                            pk_encoded,
                            row,
                            commit_seq,
                        );
                    }
                }
            }
        }
    }
    Ok(())
}

pub(super) fn validate_foreign_key_definition(
    catalog: &Catalog,
    project_id: &str,
    fk: &crate::catalog::schema::ForeignKey,
) -> Result<(), AedbError> {
    if fk.columns.is_empty() || fk.references_columns.is_empty() {
        return Err(AedbError::Validation(
            "foreign key columns cannot be empty".into(),
        ));
    }
    if fk.columns.len() != fk.references_columns.len() {
        return Err(AedbError::Validation(
            "foreign key columns length mismatch".into(),
        ));
    }
    if fk.references_project_id != project_id {
        if fk.references_project_id != "_global" {
            return Err(AedbError::Validation(
                "project foreign keys may only reference _global or same project".into(),
            ));
        }
        if matches!(fk.on_delete, ForeignKeyAction::Cascade) {
            return Err(AedbError::Validation(
                "project -> _global foreign keys cannot use ON DELETE CASCADE".into(),
            ));
        }
    }
    if project_id == "_global" && fk.references_project_id != "_global" {
        return Err(AedbError::Validation(
            "_global tables cannot reference project tables".into(),
        ));
    }
    let ref_key = (
        namespace_key(&fk.references_project_id, &fk.references_scope_id),
        fk.references_table.clone(),
    );
    let ref_table = catalog
        .tables
        .get(&ref_key)
        .ok_or_else(|| AedbError::NotFound {
            resource_type: ErrorResourceType::Table,
            resource_id: format!(
                "{}.{}.{}",
                fk.references_project_id, fk.references_scope_id, fk.references_table
            ),
        })?;
    for col in &fk.references_columns {
        if !ref_table.columns.iter().any(|c| c.name == *col) {
            return Err(AedbError::UnknownColumn {
                table: fk.references_table.clone(),
                column: col.clone(),
            });
        }
    }
    let refs_pk = ref_table.primary_key == fk.references_columns;
    let refs_unique = catalog.indexes.values().any(|idx| {
        idx.project_id == fk.references_project_id
            && idx.scope_id == fk.references_scope_id
            && idx.table_name == fk.references_table
            && idx.columns == fk.references_columns
            && matches!(
                idx.index_type,
                crate::catalog::schema::IndexType::UniqueHash
            )
    });
    if !refs_pk && !refs_unique {
        return Err(AedbError::Validation(
            "foreign key references must target PK or unique index".into(),
        ));
    }
    Ok(())
}
