aedb 0.3.1

//! Post-commit projection maintenance: async secondary-index refresh and the
//! KV-projection tables (rebuild, per-commit delta application, structural
//! version tracking). Runs after the epoch loop publishes a visible head, so
//! everything here is replay-safe and derives purely from committed state.
use super::{ExecutorState, extract_pk_from_row, now_micros};
use crate::catalog::namespace_key;
use crate::catalog::schema::TableSchema;
use crate::commit::validation::Mutation;
use crate::error::AedbError;
use crate::storage::encoded_key::EncodedKey;
use crate::storage::keyspace::NamespaceId;
use crate::version_store::CommitDelta;
use tracing::warn;

pub(crate) fn refresh_async_indexes(
    state: &mut ExecutorState,
    pending_delta: Option<&CommitDelta>,
) -> Result<bool, AedbError> {
    let target_seq = state.visible_head_seq;
    let mut changed = false;
    for ((ns, table_name, index_name), def) in &state.catalog.async_indexes {
        let key = (
            NamespaceId::Project(ns.clone()),
            table_name.clone(),
            index_name.clone(),
        );
        let current = state
            .keyspace
            .async_indexes
            .get(&key)
            .map(|v| v.materialized_seq)
            .unwrap_or(0);
        if current >= target_seq {
            continue;
        }
        let schema = state
            .catalog
            .tables
            .get(&(ns.clone(), table_name.clone()))
            .ok_or_else(|| AedbError::Validation("table missing".into()))?;
        let existing_projection = state.keyspace.take_async_projection(&key.0, &key.1, &key.2);
        let (mut projection_rows, mut materialized_seq) = existing_projection
            .map(|projection| (projection.rows, projection.materialized_seq))
            .unwrap_or_default();

        if current == 0
            && projection_rows.is_empty()
            && let Some(table) = state.keyspace.table_by_namespace_key(ns, table_name)
        {
            for (pk, stored) in &table.rows {
                let row = state.keyspace.materialize_row(stored)?;
                let projected = project_row(&row, schema, &def.projected_columns)?;
                projection_rows.insert(pk.clone(), projected);
            }
            materialized_seq = target_seq;
        }

        if materialized_seq < target_seq {
            let Some(deltas) = state
                .version_store
                .deltas_since(materialized_seq, target_seq)
            else {
                warn!(
                    from_seq = materialized_seq,
                    to_seq = target_seq,
                    "async projection delta history evicted; rebuilding projection"
                );
                projection_rows.clear();
                if let Some(table) = state.keyspace.table_by_namespace_key(ns, table_name) {
                    for (pk, stored) in &table.rows {
                        let row = state.keyspace.materialize_row(stored)?;
                        let projected = project_row(&row, schema, &def.projected_columns)?;
                        projection_rows.insert(pk.clone(), projected);
                    }
                }
                materialized_seq = target_seq;
                state.keyspace.insert_async_projection(
                    key.0.clone(),
                    key.1.clone(),
                    key.2.clone(),
                    crate::storage::keyspace::AsyncProjectionData {
                        rows: projection_rows,
                        materialized_seq,
                    },
                );
                changed = true;
                continue;
            };

            let mut requires_rebuild = false;
            for delta in deltas {
                for mutation in &delta.mutations {
                    if matches!(
                        &mutation,
                        Mutation::TableIncU256 {
                            project_id,
                            scope_id,
                            table_name: mutation_table,
                            ..
                        } | Mutation::TableDecU256 {
                            project_id,
                            scope_id,
                            table_name: mutation_table,
                            ..
                        } | Mutation::DeleteWhere {
                            project_id,
                            scope_id,
                            table_name: mutation_table,
                            ..
                        } | Mutation::UpdateWhere {
                            project_id,
                            scope_id,
                            table_name: mutation_table,
                            ..
                        } | Mutation::UpdateWhereExpr {
                            project_id,
                            scope_id,
                            table_name: mutation_table,
                            ..
                        } | Mutation::UpdateFields {
                            project_id,
                            scope_id,
                            table_name: mutation_table,
                            ..
                        } if namespace_key(project_id, scope_id) == *ns
                            && mutation_table == table_name
                    ) {
                        requires_rebuild = true;
                        break;
                    }
                    apply_projection_delta(
                        mutation,
                        ns,
                        table_name,
                        schema,
                        &def.projected_columns,
                        &mut projection_rows,
                    )?;
                }
                materialized_seq = delta.seq;
            }

            if let Some(delta) = pending_delta
                && delta.seq > materialized_seq
                && delta.seq <= target_seq
            {
                for mutation in &delta.mutations {
                    apply_projection_delta(
                        mutation,
                        ns,
                        table_name,
                        schema,
                        &def.projected_columns,
                        &mut projection_rows,
                    )?;
                }
                if requires_rebuild {
                    break;
                }
                materialized_seq = delta.seq;
            }

            if requires_rebuild {
                projection_rows.clear();
                if let Some(table) = state.keyspace.table_by_namespace_key(ns, table_name) {
                    for (pk, stored) in &table.rows {
                        let row = state.keyspace.materialize_row(stored)?;
                        let projected = project_row(&row, schema, &def.projected_columns)?;
                        projection_rows.insert(pk.clone(), projected);
                    }
                }
                materialized_seq = target_seq;
            }
        }

        state.keyspace.insert_async_projection(
            key.0,
            key.1,
            key.2,
            crate::storage::keyspace::AsyncProjectionData {
                rows: projection_rows,
                materialized_seq,
            },
        );
        changed = true;
    }
    if refresh_kv_projection_tables(state, pending_delta)? {
        changed = true;
    }
    Ok(changed)
}

fn refresh_kv_projection_tables(
    state: &mut ExecutorState,
    pending_delta: Option<&CommitDelta>,
) -> Result<bool, AedbError> {
    let target_seq = state.visible_head_seq;
    let mut changed = false;
    let projections: Vec<(String, String, String)> = state
        .catalog
        .kv_projections
        .iter()
        .map(|((project_id, scope_id), def)| {
            (project_id.clone(), scope_id.clone(), def.table_name.clone())
        })
        .collect();
    for (project_id, scope_id, table_name) in projections {
        let ns = namespace_key(&project_id, &scope_id);
        let current = state
            .keyspace
            .table_structural_version(&project_id, &scope_id, &table_name);
        if current >= target_seq {
            continue;
        }

        if current == 0 {
            rebuild_kv_projection_rows(state, &project_id, &scope_id, &table_name, target_seq)?;
            changed = true;
            continue;
        }

        let Some(deltas) = state.version_store.deltas_since(current, target_seq) else {
            rebuild_kv_projection_rows(state, &project_id, &scope_id, &table_name, target_seq)?;
            changed = true;
            continue;
        };
        for delta in deltas {
            for mutation in &delta.mutations {
                apply_kv_projection_delta(
                    state,
                    mutation,
                    &ns,
                    &project_id,
                    &scope_id,
                    &table_name,
                )?;
            }
            set_kv_projection_structural_version(
                state,
                &project_id,
                &scope_id,
                &table_name,
                delta.seq,
            );
        }
        if let Some(delta) = pending_delta
            && delta.seq > current
            && delta.seq <= target_seq
        {
            for mutation in &delta.mutations {
                apply_kv_projection_delta(
                    state,
                    mutation,
                    &ns,
                    &project_id,
                    &scope_id,
                    &table_name,
                )?;
            }
            set_kv_projection_structural_version(
                state,
                &project_id,
                &scope_id,
                &table_name,
                delta.seq,
            );
        }
        changed = true;
    }
    Ok(changed)
}

fn rebuild_kv_projection_rows(
    state: &mut ExecutorState,
    project_id: &str,
    scope_id: &str,
    table_name: &str,
    materialized_seq: u64,
) -> Result<(), AedbError> {
    let entries = state
        .keyspace
        .try_kv_scan_prefix(project_id, scope_id, &[], usize::MAX)?;
    let ns = namespace_key(project_id, scope_id);
    if let Some(table) = state.keyspace.table_by_namespace_key_mut(&ns, table_name) {
        table.rows.clear();
        table.row_versions.clear();
        table.structural_version = materialized_seq;
    } else {
        state
            .keyspace
            .table_mut(project_id, scope_id, table_name)
            .structural_version = materialized_seq;
    }
    for (key, entry) in entries {
        upsert_kv_projection_row(
            state,
            project_id,
            scope_id,
            table_name,
            KvProjectionRow {
                key,
                value: entry.value,
                commit_seq: entry.version,
                updated_at: materialized_seq,
            },
        )?;
    }
    Ok(())
}

fn apply_kv_projection_delta(
    state: &mut ExecutorState,
    mutation: &Mutation,
    namespace: &str,
    project_id: &str,
    scope_id: &str,
    table_name: &str,
) -> Result<(), AedbError> {
    match mutation {
        Mutation::KvSet {
            project_id: p,
            scope_id: s,
            key,
            value,
        } if namespace_key(p, s) == namespace => {
            upsert_kv_projection_row(
                state,
                project_id,
                scope_id,
                table_name,
                KvProjectionRow {
                    key: key.clone(),
                    value: value.clone(),
                    commit_seq: state.keyspace.try_kv_version(project_id, scope_id, key)?,
                    updated_at: now_micros(),
                },
            )?;
        }
        Mutation::KvIncU256 {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvDecU256 {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvAddU256Ex {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvSubU256Ex {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvMaxU256 {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvMinU256 {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvMutateU256 {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvAddU64Ex {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvSubU64Ex {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvMaxU64 {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvMinU64 {
            project_id: p,
            scope_id: s,
            key,
            ..
        }
        | Mutation::KvMutateU64 {
            project_id: p,
            scope_id: s,
            key,
            ..
        } if namespace_key(p, s) == namespace => {
            if let Some(entry) = state.keyspace.try_kv_get(project_id, scope_id, key)? {
                upsert_kv_projection_row(
                    state,
                    project_id,
                    scope_id,
                    table_name,
                    KvProjectionRow {
                        key: key.clone(),
                        value: entry.value,
                        commit_seq: entry.version,
                        updated_at: now_micros(),
                    },
                )?;
            }
        }
        Mutation::KvDel {
            project_id: p,
            scope_id: s,
            key,
        } if namespace_key(p, s) == namespace => {
            let _ = state.keyspace.delete_row(
                project_id,
                scope_id,
                table_name,
                &[crate::catalog::types::Value::Blob(key.clone())],
                now_micros(),
            );
        }
        Mutation::Ddl(crate::catalog::DdlOperation::DropScope {
            project_id: p,
            scope_id: s,
            if_exists: true,
            ..
        }) if namespace_key(p, s) == namespace => {
            state.keyspace.drop_table(project_id, scope_id, table_name);
        }
        Mutation::Ddl(crate::catalog::DdlOperation::DropProject { project_id: p, .. })
            if namespace.starts_with(&(p.to_string() + "::")) =>
        {
            state.keyspace.drop_table(project_id, scope_id, table_name);
        }
        Mutation::Ddl(crate::catalog::DdlOperation::DisableKvProjection {
            project_id: p,
            scope_id: s,
        }) if namespace_key(p, s) == namespace => {
            state.keyspace.drop_table(project_id, scope_id, table_name);
        }
        _ => {}
    }
    Ok(())
}

struct KvProjectionRow {
    key: Vec<u8>,
    value: Vec<u8>,
    commit_seq: u64,
    updated_at: u64,
}

fn upsert_kv_projection_row(
    state: &mut ExecutorState,
    project_id: &str,
    scope_id: &str,
    table_name: &str,
    row: KvProjectionRow,
) -> Result<(), AedbError> {
    let commit_seq_i64 = i64::try_from(row.commit_seq)
        .map_err(|_| AedbError::Validation("commit_seq overflow".into()))?;
    let updated_at = i64::try_from(row.updated_at)
        .map_err(|_| AedbError::Validation("updated_at overflow".into()))?;
    state.keyspace.upsert_row(
        project_id,
        scope_id,
        table_name,
        vec![crate::catalog::types::Value::Blob(row.key.clone())],
        crate::catalog::types::Row {
            values: vec![
                crate::catalog::types::Value::Text(project_id.into()),
                crate::catalog::types::Value::Text(scope_id.into()),
                crate::catalog::types::Value::Blob(row.key),
                crate::catalog::types::Value::Blob(row.value),
                crate::catalog::types::Value::Integer(commit_seq_i64),
                crate::catalog::types::Value::Timestamp(updated_at),
            ],
        },
        row.commit_seq,
    );
    Ok(())
}

fn set_kv_projection_structural_version(
    state: &mut ExecutorState,
    project_id: &str,
    scope_id: &str,
    table_name: &str,
    seq: u64,
) {
    let ns = namespace_key(project_id, scope_id);
    if let Some(table) = state.keyspace.table_by_namespace_key_mut(&ns, table_name) {
        table.structural_version = seq;
    }
}

fn project_row(
    row: &crate::catalog::types::Row,
    schema: &TableSchema,
    projected_columns: &[String],
) -> Result<crate::catalog::types::Row, AedbError> {
    let mut values = Vec::with_capacity(projected_columns.len());
    for col in projected_columns {
        let column_index = schema
            .columns
            .iter()
            .position(|c| c.name == *col)
            .ok_or_else(|| AedbError::Validation(format!("projection column missing: {col}")))?;
        values.push(row.values[column_index].clone());
    }
    Ok(crate::catalog::types::Row { values })
}

fn apply_projection_delta(
    mutation: &Mutation,
    ns: &str,
    table_name: &str,
    schema: &TableSchema,
    projected_columns: &[String],
    projection_rows: &mut im::OrdMap<EncodedKey, crate::catalog::types::Row>,
) -> Result<(), AedbError> {
    match mutation {
        Mutation::Insert {
            project_id,
            scope_id,
            table_name: mutation_table,
            primary_key,
            row,
        }
        | Mutation::Upsert {
            project_id,
            scope_id,
            table_name: mutation_table,
            primary_key,
            row,
        } => {
            if namespace_key(project_id, scope_id) == ns && mutation_table == table_name {
                let projected = project_row(row, schema, projected_columns)?;
                projection_rows.insert(EncodedKey::from_values(primary_key), projected);
            }
        }
        Mutation::InsertBatch {
            project_id,
            scope_id,
            table_name: mutation_table,
            rows,
        } => {
            if namespace_key(project_id, scope_id) == ns && mutation_table == table_name {
                for row in rows {
                    let pk = extract_pk_from_row(schema, row)?;
                    let projected = project_row(row, schema, projected_columns)?;
                    projection_rows.insert(EncodedKey::from_values(&pk), projected);
                }
            }
        }
        Mutation::Delete {
            project_id,
            scope_id,
            table_name: mutation_table,
            primary_key,
        } => {
            if namespace_key(project_id, scope_id) == ns && mutation_table == table_name {
                projection_rows.remove(&EncodedKey::from_values(primary_key));
            }
        }
        Mutation::Ddl(crate::catalog::DdlOperation::DropTable {
            project_id,
            scope_id,
            table_name: dropped_table,
            ..
        }) => {
            if namespace_key(project_id, scope_id) == ns && dropped_table == table_name {
                projection_rows.clear();
            }
        }
        Mutation::Ddl(crate::catalog::DdlOperation::DropScope {
            project_id,
            scope_id,
            ..
        }) => {
            if namespace_key(project_id, scope_id) == ns {
                projection_rows.clear();
            }
        }
        Mutation::Ddl(crate::catalog::DdlOperation::DropProject { project_id, .. })
            if ns.starts_with(&(project_id.to_owned() + "::")) =>
        {
            projection_rows.clear();
        }
        _ => {}
    }
    Ok(())
}