safe-migrate 0.8.0

use super::{AnalysisState, CascadeResult, MutationResult, ObjectLookup, RelationOverlay};
use crate::_internal::analysis::evidence::{EvidenceCode, EvidenceScope};
use crate::_internal::analysis::graph::{DependencyEdge, DependencyKind};
use crate::_internal::analysis::mutations::{
    CreateIndex, CreateMaterializedView, CreateView, DropIndex, DropMaterializedViewMutation,
    DropViewMutation, RefreshMaterializedViewMutation,
};
use crate::_internal::ast::identifiers::ObjectId;
use crate::_internal::model::relation::{Persistence, RelationKind, RelationState};
use std::collections::HashSet;

type RelationLookup = ObjectLookup;
type IndexLookup = ObjectLookup;

impl AnalysisState {
    fn validate_view_dependencies(
        &mut self,
        dependent: &ObjectId,
        dependencies: &[ObjectId],
    ) -> Result<(), MutationResult> {
        for dependency in dependencies {
            // Recursive views are represented by a self-reference in some
            // PostgreSQL catalog versions. It does not identify an external
            // object that must already exist.
            if dependency == dependent {
                continue;
            }
            self.ensure_relation_target(
                dependency,
                |_| true,
                format!("view dependency relation '{}' does not exist", dependency),
                format!("view dependency '{}' is not a relation", dependency),
            )?;
        }
        Ok(())
    }

    pub(crate) fn cascade_for_relations(&self, roots: &[ObjectId]) -> CascadeResult {
        let mut cascade = CascadeResult::default();
        for root in roots {
            let closure = self.get_cascade_closure(root);
            cascade.dropped_relations.extend(closure.dropped_relations);
            cascade.dropped_indexes.extend(closure.dropped_indexes);
            cascade
                .dropped_constraints
                .extend(closure.dropped_constraints);
        }
        cascade
    }

    fn remove_dropped_relation_edges(
        &mut self,
        dropped_relations: &HashSet<ObjectId>,
        dropped_indexes: &HashSet<ObjectId>,
    ) {
        self.snapshot_graph_full();
        let resolution_graph = self.local.graph.clone();
        self.local.graph.retain_edges(|edge| {
            let dependent = resolution_graph.resolve_rename(&edge.dependent);
            let referenced = resolution_graph.resolve_rename(&edge.referenced);
            if dropped_indexes.contains(dependent) {
                return false;
            }
            if dropped_relations.contains(dependent) {
                return false;
            }
            if dropped_relations.contains(referenced) {
                return false;
            }
            true
        });
    }

    fn has_external_view_dependents(&self, roots: &HashSet<ObjectId>) -> bool {
        self.local.graph.edges().iter().any(|edge| {
            self.dependency_edge_is_current(edge)
                && matches!(edge.kind, DependencyKind::ViewDependency { .. })
                && roots.contains(self.local.graph.resolve_rename(&edge.referenced))
                && !roots.contains(self.local.graph.resolve_rename(&edge.dependent))
        })
    }

    pub(super) fn apply_drop_relation_family(
        &mut self,
        present: &[ObjectId],
        cascade: bool,
        kind_name: &str,
    ) -> MutationResult {
        let roots = present.iter().cloned().collect::<HashSet<_>>();
        if present.iter().any(|id| {
            self.baseline_scoped_family_object(
                id,
                crate::_internal::db::cache::CatalogFamily::Relations,
            )
        }) {
            // View dependency rows are scope-aware, but expression and
            // extension-dependent objects outside the selected schemas are
            // not yet represented as a complete catalog contract.
            self.taint(
                EvidenceCode::CatalogCoverageIncomplete,
                EvidenceScope::Chain,
            );
            return MutationResult::Skipped;
        }
        if !cascade && self.has_external_view_dependents(&roots) {
            return MutationResult::Conflict {
                reason: format!(
                    "relation '{}' still has dependent views; use CASCADE",
                    present
                        .first()
                        .map(ToString::to_string)
                        .unwrap_or_else(|| kind_name.to_string())
                ),
            };
        }

        let cascade_result = cascade.then(|| self.cascade_for_relations(present));
        if let Some(result) = &cascade_result
            && result
                .dropped_relations
                .iter()
                .any(|id| !self.relation_is_present(id))
        {
            // A scoped cache can expose a dependency edge for a relation whose
            // metadata was intentionally omitted. CASCADE will remove that
            // relation in PostgreSQL, but the simulator cannot reproduce its
            // full state, so the result is necessarily tainted.
            self.taint(
                EvidenceCode::CatalogCoverageIncomplete,
                EvidenceScope::Chain,
            );
        }
        let all_dropped_relations = cascade_result
            .as_ref()
            .map(|result| result.dropped_relations.clone())
            .unwrap_or_else(|| roots.clone());
        let dropped_relations = all_dropped_relations
            .iter()
            .filter(|id| self.relation_is_present(id))
            .cloned()
            .collect::<HashSet<_>>();
        let dropped_indexes = cascade_result
            .as_ref()
            .map(|result| result.dropped_indexes.clone())
            .unwrap_or_default();
        let dropped_constraints = cascade_result
            .as_ref()
            .map(|result| result.dropped_constraints.clone())
            .unwrap_or_default();

        for id in &dropped_relations {
            self.snapshot_relation(id);
            self.local
                .relations
                .insert(id.clone(), RelationOverlay::Dropped);
        }
        let resolution_graph = self.local.graph.clone();
        let triggers_to_drop = self
            .local
            .triggers
            .iter()
            .filter_map(|(id, overlay)| {
                let crate::_internal::model::trigger::TriggerOverlay::Present(trigger) = overlay
                else {
                    return None;
                };
                dropped_relations
                    .contains(resolution_graph.resolve_rename(&trigger.table_id))
                    .then(|| id.clone())
            })
            .collect::<Vec<_>>();
        for trigger_id in triggers_to_drop {
            self.snapshot_trigger(&trigger_id);
            self.local.triggers.insert(
                trigger_id,
                crate::_internal::model::trigger::TriggerOverlay::Dropped,
            );
        }
        // Even when a scoped cache omitted a dependent view, its dependency
        // edge is known and PostgreSQL CASCADE removes that edge. Use the full
        // closure for topology cleanup while only marking modeled overlays.
        self.remove_dropped_constraints(&all_dropped_relations, &dropped_constraints);
        self.remove_dropped_relation_edges(&all_dropped_relations, &dropped_indexes);
        MutationResult::Applied
    }

    fn index_lookup(&self, id: &ObjectId) -> IndexLookup {
        if self.local.graph.edges().iter().any(|edge| {
            matches!(edge.kind, DependencyKind::IndexOnRelation { .. }) && edge.dependent == *id
        }) {
            IndexLookup::Present
        } else if self
            .baseline_covers_family_object(id, crate::_internal::db::cache::CatalogFamily::Indexes)
        {
            IndexLookup::AuthoritativelyAbsent
        } else {
            IndexLookup::Unknown
        }
    }

    pub(super) fn apply_create_view(&mut self, create: &CreateView) -> MutationResult {
        if let Err(result) = self.ensure_schema_target(&create.id.schema) {
            return result;
        }
        let existing_view = matches!(
            self.relation_lookup(&create.id, |kind| *kind == RelationKind::View),
            RelationLookup::Present
        );
        if self.relation_namespace_is_taken(&create.id) && (!create.or_replace || !existing_view) {
            return MutationResult::Conflict {
                reason: format!("relation '{}' already exists", create.id),
            };
        }
        if let Err(result) = self.validate_view_dependencies(&create.id, &create.depends_on) {
            return result;
        }
        let owner = self
            .local
            .relations
            .get(&create.id)
            .and_then(|overlay| match overlay {
                RelationOverlay::Present(relation) => Some(relation.owner.clone()),
                RelationOverlay::Dropped => None,
            })
            .unwrap_or_else(|| ObjectId::new("", &self.local.current_role));
        self.snapshot_relation(&create.id);
        self.snapshot_generation_counter();
        self.local.generation_counter += 1;
        let generation = self.local.generation_counter;
        let mut relation = if existing_view {
            match self.local.relations.get(&create.id) {
                Some(RelationOverlay::Present(existing)) => existing.clone(),
                _ => unreachable!("existing view lookup established presence"),
            }
        } else {
            RelationState::new(
                create.id.clone(),
                owner,
                generation,
                None,
                RelationKind::View,
                Persistence::Permanent,
                self.local.transactions.len(),
            )
        };
        // CREATE OR REPLACE VIEW keeps the relation identity, ownership,
        // privileges, triggers, and other relation metadata. Only the
        // generation records the replacement in the simulated state.
        relation.id = create.id.clone();
        relation.generation = generation;
        relation.kind = RelationKind::View;
        relation.persistence = Persistence::Permanent;
        self.local
            .relations
            .insert(create.id.clone(), RelationOverlay::Present(relation));
        self.snapshot_graph_full();
        if existing_view {
            self.local.graph.retain_edges(|edge| {
                !(matches!(edge.kind, DependencyKind::ViewDependency { .. })
                    && edge.dependent == create.id)
            });
        }
        for dependency in &create.depends_on {
            self.local.graph.add_edge(DependencyEdge::new(
                create.id.clone(),
                dependency.clone(),
                DependencyKind::ViewDependency {
                    view_generation: generation,
                    referenced_column: None,
                },
            ));
        }
        MutationResult::Applied
    }

    pub(super) fn apply_create_materialized_view(
        &mut self,
        create: &CreateMaterializedView,
    ) -> MutationResult {
        if let Err(result) = self.ensure_schema_target(&create.id.schema) {
            return result;
        }
        if self.relation_namespace_is_taken(&create.id) {
            return MutationResult::Conflict {
                reason: format!("relation '{}' already exists", create.id),
            };
        }
        if let Err(result) = self.validate_view_dependencies(&create.id, &create.depends_on) {
            return result;
        }
        self.snapshot_relation(&create.id);
        self.snapshot_generation_counter();
        self.local.generation_counter += 1;
        let generation = self.local.generation_counter;
        let mut relation = RelationState::new(
            create.id.clone(),
            ObjectId::new("", &self.local.current_role),
            generation,
            None,
            RelationKind::MaterializedView,
            Persistence::Permanent,
            self.local.transactions.len(),
        );
        relation.is_populated = Some(true);
        self.local
            .relations
            .insert(create.id.clone(), RelationOverlay::Present(relation));
        self.snapshot_graph_full();
        for dependency in &create.depends_on {
            self.local.graph.add_edge(DependencyEdge::new(
                create.id.clone(),
                dependency.clone(),
                DependencyKind::ViewDependency {
                    view_generation: generation,
                    referenced_column: None,
                },
            ));
        }
        MutationResult::Applied
    }

    pub(super) fn apply_refresh_materialized_view(
        &mut self,
        refresh: &RefreshMaterializedViewMutation,
    ) -> MutationResult {
        if refresh.concurrently && self.in_transaction() {
            return MutationResult::Conflict {
                reason: "REFRESH MATERIALIZED VIEW CONCURRENTLY cannot run inside a transaction"
                    .to_string(),
            };
        }
        match self.relation_lookup(&refresh.id, |kind| *kind == RelationKind::MaterializedView) {
            RelationLookup::Present => {
                if refresh.concurrently {
                    match self.local.relations.get(&refresh.id) {
                        Some(RelationOverlay::Present(relation)) => match relation.is_populated {
                            Some(false) => {
                                return MutationResult::Conflict {
                                    reason: format!(
                                        "materialized view '{}' must be populated before a concurrent refresh",
                                        refresh.id
                                    ),
                                };
                            }
                            None => {
                                self.taint(
                                    EvidenceCode::CatalogCoverageIncomplete,
                                    EvidenceScope::Statement,
                                );
                                return MutationResult::Skipped;
                            }
                            Some(true) => {}
                        },
                        _ => unreachable!("materialized-view lookup established presence"),
                    }
                }
                if let Some(RelationOverlay::Present(relation)) =
                    self.local.relations.get_mut(&refresh.id)
                {
                    relation.is_populated = Some(true);
                }
                MutationResult::Applied
            }
            RelationLookup::WrongKind => MutationResult::Conflict {
                reason: format!("'{}' is not a materialized view", refresh.id),
            },
            RelationLookup::AuthoritativelyAbsent | RelationLookup::Tombstone => {
                MutationResult::Conflict {
                    reason: format!("materialized view '{}' does not exist", refresh.id),
                }
            }
            RelationLookup::Unknown => {
                self.taint(EvidenceCode::UnknownObjectState, EvidenceScope::Chain);
                MutationResult::Skipped
            }
        }
    }

    pub(super) fn apply_create_index(&mut self, create: &CreateIndex) -> MutationResult {
        if create.concurrently && self.in_transaction() {
            return MutationResult::Conflict {
                reason: "CREATE INDEX CONCURRENTLY cannot run inside a transaction".to_string(),
            };
        }
        if let Err(result) = self.ensure_schema_target(&create.id.schema) {
            return result;
        }
        if create.if_not_exists && self.index_lookup(&create.id) == IndexLookup::Present {
            return MutationResult::Skipped;
        }
        if self.relation_namespace_is_taken(&create.id) {
            return MutationResult::Conflict {
                reason: format!("relation '{}' already exists", create.id),
            };
        }
        if let Err(result) = self.ensure_relation_target(
            &create.table,
            |kind| matches!(kind, RelationKind::Table | RelationKind::MaterializedView),
            format!("index target relation '{}' does not exist", create.table),
            format!("index target '{}' cannot be indexed", create.table),
        ) {
            return result;
        }
        if create.concurrently
            && self
                .local
                .relations
                .get(&create.table)
                .and_then(|overlay| match overlay {
                    RelationOverlay::Present(relation) => relation.partition_type.as_deref(),
                    RelationOverlay::Dropped => None,
                })
                .is_some()
        {
            return MutationResult::Conflict {
                reason: format!(
                    "CREATE INDEX CONCURRENTLY cannot run on partitioned table '{}'",
                    create.table
                ),
            };
        }
        self.snapshot_graph();
        self.local.graph.add_edge(DependencyEdge::new(
            create.id.clone(),
            create.table.clone(),
            DependencyKind::IndexOnRelation {
                using_method: create
                    .using_method
                    .clone()
                    .or_else(|| Some("btree".to_string())),
                key_columns: create.key_columns.clone(),
                included_columns: create.included_columns.clone(),
                dependency_columns: create
                    .key_columns
                    .iter()
                    .chain(&create.included_columns)
                    .cloned()
                    .collect(),
                dependency_columns_known: !create.has_expression_keys && !create.has_predicate,
                has_expression_keys: create.has_expression_keys,
                has_predicate: create.has_predicate,
                is_concurrent: create.concurrently,
                is_unique: create.unique,
                is_valid: true,
                is_ready: true,
                is_live: true,
                has_default_sort_order: create.has_default_sort_order,
                has_default_opclasses: create.has_default_opclasses,
                has_default_collations: create.has_default_collations,
                // Eligibility facts are known even when dependency-column
                // proof is incomplete: expression and partial indexes are
                // deterministically ineligible for USING INDEX. Keep that
                // distinction separate from dependency_columns_known, which
                // governs exact column-drop cleanup.
                eligibility_known: true,
            },
        ));
        MutationResult::Applied
    }

    pub(super) fn apply_drop_view(&mut self, drop: &DropViewMutation) -> MutationResult {
        let mut present = Vec::new();
        let mut unknown_target = false;
        for id in &drop.ids {
            match self.relation_lookup(id, |kind| *kind == RelationKind::View) {
                RelationLookup::Present => present.push(id.clone()),
                RelationLookup::WrongKind => {
                    return MutationResult::Conflict {
                        reason: format!("'{}' is not a view", id),
                    };
                }
                RelationLookup::AuthoritativelyAbsent if drop.if_exists => {}
                RelationLookup::AuthoritativelyAbsent => {
                    return MutationResult::Conflict {
                        reason: format!("view '{}' does not exist", id),
                    };
                }
                RelationLookup::Tombstone if drop.if_exists => {}
                RelationLookup::Tombstone => {
                    return MutationResult::Conflict {
                        reason: format!("view '{}' does not exist", id),
                    };
                }
                RelationLookup::Unknown => {
                    self.taint(EvidenceCode::UnknownObjectState, EvidenceScope::Chain);
                    unknown_target = true;
                }
            }
        }
        // PostgreSQL resolves all names in a multi-target DROP before
        // executing the statement.  Do not remove known targets when another
        // target is outside a scoped/incomplete baseline.
        if unknown_target {
            return MutationResult::Skipped;
        }
        if present.is_empty() {
            return MutationResult::Skipped;
        }
        self.apply_drop_relation_family(&present, drop.cascade, "view")
    }

    pub(super) fn apply_drop_materialized_view(
        &mut self,
        drop: &DropMaterializedViewMutation,
    ) -> MutationResult {
        let mut present = Vec::new();
        let mut unknown_target = false;
        for id in &drop.ids {
            match self.relation_lookup(id, |kind| *kind == RelationKind::MaterializedView) {
                RelationLookup::Present => present.push(id.clone()),
                RelationLookup::WrongKind => {
                    return MutationResult::Conflict {
                        reason: format!("'{}' is not a materialized view", id),
                    };
                }
                RelationLookup::AuthoritativelyAbsent if drop.if_exists => {}
                RelationLookup::AuthoritativelyAbsent => {
                    return MutationResult::Conflict {
                        reason: format!("materialized view '{}' does not exist", id),
                    };
                }
                RelationLookup::Tombstone if drop.if_exists => {}
                RelationLookup::Tombstone => {
                    return MutationResult::Conflict {
                        reason: format!("materialized view '{}' does not exist", id),
                    };
                }
                RelationLookup::Unknown => {
                    self.taint(EvidenceCode::UnknownObjectState, EvidenceScope::Chain);
                    unknown_target = true;
                }
            }
        }
        if unknown_target {
            return MutationResult::Skipped;
        }
        if present.is_empty() {
            return MutationResult::Skipped;
        }
        self.apply_drop_relation_family(&present, drop.cascade, "materialized view")
    }

    pub(super) fn apply_drop_index(&mut self, drop: &DropIndex) -> MutationResult {
        if drop.concurrently && self.in_transaction() {
            return MutationResult::Conflict {
                reason: "DROP INDEX CONCURRENTLY cannot run inside a transaction".to_string(),
            };
        }
        if drop.concurrently && drop.cascade {
            return MutationResult::Conflict {
                reason: "DROP INDEX CONCURRENTLY cannot use CASCADE".to_string(),
            };
        }
        // PostgreSQL resolves every target before it applies a multi-index
        // DROP. Preflight the complete statement so a later invalid target
        // cannot leave an earlier index removed from simulated state.
        let mut targets = Vec::new();
        for id in &drop.ids {
            match self.index_lookup(id) {
                IndexLookup::Present => {
                    if !targets.contains(id) {
                        targets.push(id.clone());
                    }
                }
                IndexLookup::AuthoritativelyAbsent if drop.if_exists => {}
                IndexLookup::AuthoritativelyAbsent => {
                    return MutationResult::Conflict {
                        reason: format!("index '{}' does not exist", id),
                    };
                }
                IndexLookup::Unknown => {
                    self.taint(EvidenceCode::UnknownObjectState, EvidenceScope::Chain);
                    return MutationResult::Skipped;
                }
                IndexLookup::WrongKind | IndexLookup::Tombstone => {
                    unreachable!("indexes have no overlay kind or tombstone")
                }
            }
        }

        if targets.is_empty() {
            return MutationResult::Skipped;
        }

        if targets.iter().any(|id| {
            self.baseline_scoped_family_object(
                id,
                crate::_internal::db::cache::CatalogFamily::Indexes,
            )
        }) {
            // Scoped index rows do not yet carry a complete backing-constraint
            // identity or every external dependency, so PostgreSQL's DROP
            // INDEX conflict semantics cannot be proven from the partial
            // graph. Leave the baseline unchanged until V7 index ownership
            // coverage is object-complete.
            self.taint(
                EvidenceCode::CatalogCoverageIncomplete,
                EvidenceScope::Chain,
            );
            return MutationResult::Skipped;
        }

        for id in &targets {
            let Some(index_edge) = self.local.graph.edges().iter().find(|edge| {
                matches!(edge.kind, DependencyKind::IndexOnRelation { .. }) && edge.dependent == *id
            }) else {
                self.taint(
                    EvidenceCode::CatalogCoverageIncomplete,
                    EvidenceScope::Chain,
                );
                return MutationResult::Skipped;
            };
            let referenced_table = self.local.graph.resolve_rename(&index_edge.referenced);
            let resolved_index = self.local.graph.resolve_rename(id);
            if drop.concurrently
                && self
                    .local
                    .relations
                    .get(referenced_table)
                    .and_then(|overlay| match overlay {
                        RelationOverlay::Present(relation) => relation.partition_type.as_deref(),
                        RelationOverlay::Dropped => None,
                    })
                    .is_some()
            {
                return MutationResult::Conflict {
                    reason: format!(
                        "DROP INDEX CONCURRENTLY cannot run on partitioned index '{}'",
                        id
                    ),
                };
            }
            let unresolved_constraint = self.baseline_indexes.contains(id)
                && self
                    .local
                    .constraints
                    .iter()
                    .any(|((table, _), constraint)| {
                        self.local.graph.resolve_rename(table) == referenced_table
                            && matches!(
                                constraint.kind,
                                crate::_internal::model::constraint::ConstraintKind::PrimaryKey
                                    | crate::_internal::model::constraint::ConstraintKind::Unique
                                    | crate::_internal::model::constraint::ConstraintKind::Exclusion
                            )
                            && constraint.backing_index.is_none()
                    });
            if unresolved_constraint {
                self.taint(
                    EvidenceCode::CatalogCoverageIncomplete,
                    EvidenceScope::Chain,
                );
                return MutationResult::Skipped;
            }
            let backs_constraint = self
                .local
                .constraints
                .iter()
                .any(|((table, _), constraint)| {
                    self.local.graph.resolve_rename(table) == referenced_table
                        && constraint.backing_index.as_ref() == Some(resolved_index)
                });
            if backs_constraint {
                return MutationResult::Conflict {
                    reason: format!(
                        "cannot drop index '{}' because a constraint requires it",
                        id
                    ),
                };
            }
            if matches!(
                index_edge.kind,
                DependencyKind::IndexOnRelation {
                    eligibility_known: false,
                    ..
                }
            ) {
                self.taint(
                    EvidenceCode::CatalogCoverageIncomplete,
                    EvidenceScope::Chain,
                );
                return MutationResult::Skipped;
            }
        }
        self.snapshot_graph();
        self.local.graph.retain_edges(|edge| {
            !(matches!(edge.kind, DependencyKind::IndexOnRelation { .. })
                && targets.contains(&edge.dependent))
        });
        MutationResult::Applied
    }
}