"""Align resource/decision/checkpoint schema gaps with specification DDL.

Adds four spec-parity changes:

1. ``resource_links.link_type`` with default ``'contains'``.
2. Foreign keys on ``checkpoint_metadata.decision_id`` and
   ``checkpoint_metadata.resource_id``.
3. Partial index ``idx_decisions_superseded`` on
   ``decisions.superseded_by`` where non-null.
4. SQLite trigger guards to enforce checkpoint FK semantics at runtime.

Revision ID: m4_004_schema_parity_resource_decision_checkpoint
Revises: m4_003_plan_env_columns
Create Date: 2026-03-26 00:00:00
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision: str = "m4_004_schema_parity_resource_decision_checkpoint"
down_revision: str | Sequence[str] | None = "m4_003_plan_env_columns"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def _inspector() -> sa.Inspector:
    return sa.inspect(op.get_bind())


def _create_sqlite_checkpoint_fk_triggers() -> None:
    # NOTE: These triggers guard INSERT and UPDATE on checkpoint_metadata only.
    # DELETE-direction enforcement (preventing deletion of a referenced decision
    # or resource) relies on PRAGMA foreign_keys=ON at the connection level.
    # This is a known limitation of trigger-based FK emulation on SQLite.
    op.execute(
        sa.text(
            """
            CREATE TRIGGER IF NOT EXISTS trg_checkpoint_metadata_decision_fk_insert
            BEFORE INSERT ON checkpoint_metadata
            FOR EACH ROW
            WHEN NEW.decision_id IS NOT NULL
              AND NOT EXISTS (
                  SELECT 1 FROM decisions WHERE decision_id = NEW.decision_id
              )
            BEGIN
                SELECT RAISE(ABORT, 'FOREIGN KEY constraint failed');
            END;
            """
        )
    )
    op.execute(
        sa.text(
            """
            CREATE TRIGGER IF NOT EXISTS trg_checkpoint_metadata_decision_fk_update
            BEFORE UPDATE OF decision_id ON checkpoint_metadata
            FOR EACH ROW
            WHEN NEW.decision_id IS NOT NULL
              AND NOT EXISTS (
                  SELECT 1 FROM decisions WHERE decision_id = NEW.decision_id
              )
            BEGIN
                SELECT RAISE(ABORT, 'FOREIGN KEY constraint failed');
            END;
            """
        )
    )
    op.execute(
        sa.text(
            """
            CREATE TRIGGER IF NOT EXISTS trg_checkpoint_metadata_resource_fk_insert
            BEFORE INSERT ON checkpoint_metadata
            FOR EACH ROW
            WHEN NEW.resource_id IS NOT NULL
              AND NOT EXISTS (
                  SELECT 1 FROM resources WHERE resource_id = NEW.resource_id
              )
            BEGIN
                SELECT RAISE(ABORT, 'FOREIGN KEY constraint failed');
            END;
            """
        )
    )
    op.execute(
        sa.text(
            """
            CREATE TRIGGER IF NOT EXISTS trg_checkpoint_metadata_resource_fk_update
            BEFORE UPDATE OF resource_id ON checkpoint_metadata
            FOR EACH ROW
            WHEN NEW.resource_id IS NOT NULL
              AND NOT EXISTS (
                  SELECT 1 FROM resources WHERE resource_id = NEW.resource_id
              )
            BEGIN
                SELECT RAISE(ABORT, 'FOREIGN KEY constraint failed');
            END;
            """
        )
    )


def _drop_sqlite_checkpoint_fk_triggers() -> None:
    op.execute(
        sa.text("DROP TRIGGER IF EXISTS trg_checkpoint_metadata_decision_fk_insert")
    )
    op.execute(
        sa.text("DROP TRIGGER IF EXISTS trg_checkpoint_metadata_decision_fk_update")
    )
    op.execute(
        sa.text("DROP TRIGGER IF EXISTS trg_checkpoint_metadata_resource_fk_insert")
    )
    op.execute(
        sa.text("DROP TRIGGER IF EXISTS trg_checkpoint_metadata_resource_fk_update")
    )


def upgrade() -> None:
    """Apply spec-parity schema updates."""
    inspector = _inspector()

    resource_link_columns = {
        column["name"] for column in inspector.get_columns("resource_links")
    }
    if "link_type" not in resource_link_columns:
        with op.batch_alter_table("resource_links") as batch_op:
            batch_op.add_column(
                sa.Column(
                    "link_type",
                    sa.Text(),
                    nullable=False,
                    server_default=sa.text("'contains'"),
                )
            )
            # NOTE: Spec DDL (line 45549) does not declare a CHECK on
            # link_type; we add it to match the sibling resource_edges
            # constraint and prevent invalid values at the DB level.
            batch_op.create_check_constraint(
                "ck_resource_links_link_type",
                "link_type IN ('contains', 'references', 'derived_from')",
            )
    else:
        # link_type column already exists — fix default and ensure CHECK
        # constraint is present (guards against partial prior migration).
        # NOTE: Spec DDL (line 45549) defines link_type as nullable
        # (TEXT DEFAULT 'contains' without NOT NULL).  We deliberately
        # enforce NOT NULL because a NULL link type is semantically
        # meaningless and the sibling resource_edges table also requires
        # a non-NULL link_type.
        # NOTE: Column TYPE is not verified here.  On SQLite all text
        # types are equivalent, so a prior VARCHAR(30) column works
        # identically to TEXT.  On PostgreSQL, VARCHAR(30) != TEXT; if
        # a future migration targets PostgreSQL this path should also
        # include an ALTER COLUMN TYPE to sa.Text().
        link_type_column = next(
            column
            for column in inspector.get_columns("resource_links")
            if column["name"] == "link_type"
        )
        current_default = str(link_type_column.get("default") or "").lower()
        needs_default_fix = "contains" not in current_default

        check_constraints = inspector.get_check_constraints("resource_links")
        has_link_type_ck = any(
            ck.get("name") == "ck_resource_links_link_type" for ck in check_constraints
        )

        if needs_default_fix or not has_link_type_ck:
            with op.batch_alter_table("resource_links") as batch_op:
                if needs_default_fix:
                    batch_op.alter_column(
                        "link_type",
                        server_default=sa.text("'contains'"),
                    )
                if not has_link_type_ck:
                    batch_op.create_check_constraint(
                        "ck_resource_links_link_type",
                        "link_type IN ('contains', 'references', 'derived_from')",
                    )

    decision_indexes = {index["name"] for index in inspector.get_indexes("decisions")}
    if "idx_decisions_superseded" not in decision_indexes:
        op.create_index(
            "idx_decisions_superseded",
            "decisions",
            ["superseded_by"],
            unique=False,
            postgresql_where=sa.text("superseded_by IS NOT NULL"),
            sqlite_where=sa.text("superseded_by IS NOT NULL"),
        )

    # Re-inspect after potential DDL changes above (link_type column, index).
    inspector = _inspector()
    checkpoint_fks = inspector.get_foreign_keys("checkpoint_metadata")
    fk_signatures = {
        (
            tuple(fk.get("constrained_columns") or []),
            fk.get("referred_table"),
            tuple(fk.get("referred_columns") or []),
        )
        for fk in checkpoint_fks
    }

    needs_decision_fk = (
        ("decision_id",),
        "decisions",
        ("decision_id",),
    ) not in fk_signatures
    needs_resource_fk = (
        ("resource_id",),
        "resources",
        ("resource_id",),
    ) not in fk_signatures

    if needs_decision_fk or needs_resource_fk:
        # Nullify orphan references before creating FK constraints so the
        # migration does not fail on existing data with dangling IDs.
        # Uses NOT EXISTS instead of NOT IN for better query-plan
        # performance on large checkpoint_metadata tables.
        op.execute(
            sa.text(
                """
                UPDATE checkpoint_metadata
                SET decision_id = NULL
                WHERE decision_id IS NOT NULL
                  AND NOT EXISTS (
                      SELECT 1 FROM decisions d
                      WHERE d.decision_id = checkpoint_metadata.decision_id
                  )
                """
            )
        )
        op.execute(
            sa.text(
                """
                UPDATE checkpoint_metadata
                SET resource_id = NULL
                WHERE resource_id IS NOT NULL
                  AND NOT EXISTS (
                      SELECT 1 FROM resources r
                      WHERE r.resource_id = checkpoint_metadata.resource_id
                  )
                """
            )
        )

        # NOTE: Spec DDL (lines 45569, 45571) uses bare REFERENCES without
        # an ON DELETE clause (default: NO ACTION / RESTRICT).  We use
        # ondelete="SET NULL" so that DecisionRepository.delete() and
        # ResourceRepository.delete() do not raise IntegrityError when a
        # parent decision or resource is removed while checkpoint rows
        # still reference it.
        with op.batch_alter_table("checkpoint_metadata") as batch_op:
            if needs_decision_fk:
                batch_op.create_foreign_key(
                    "fk_checkpoint_metadata_decision",
                    "decisions",
                    ["decision_id"],
                    ["decision_id"],
                    ondelete="SET NULL",
                )
            if needs_resource_fk:
                batch_op.create_foreign_key(
                    "fk_checkpoint_metadata_resource",
                    "resources",
                    ["resource_id"],
                    ["resource_id"],
                    ondelete="SET NULL",
                )

    if op.get_bind().dialect.name == "sqlite":
        _create_sqlite_checkpoint_fk_triggers()


def downgrade() -> None:
    """Revert spec-parity schema updates."""
    inspector = _inspector()

    if op.get_bind().dialect.name == "sqlite":
        _drop_sqlite_checkpoint_fk_triggers()

    decision_indexes = {index["name"] for index in inspector.get_indexes("decisions")}
    if "idx_decisions_superseded" in decision_indexes:
        op.drop_index("idx_decisions_superseded", table_name="decisions")

    checkpoint_fks = inspector.get_foreign_keys("checkpoint_metadata")
    fk_names = {fk.get("name") for fk in checkpoint_fks}
    if (
        "fk_checkpoint_metadata_decision" in fk_names
        or "fk_checkpoint_metadata_resource" in fk_names
    ):
        with op.batch_alter_table("checkpoint_metadata") as batch_op:
            if "fk_checkpoint_metadata_decision" in fk_names:
                batch_op.drop_constraint(
                    "fk_checkpoint_metadata_decision",
                    type_="foreignkey",
                )
            if "fk_checkpoint_metadata_resource" in fk_names:
                batch_op.drop_constraint(
                    "fk_checkpoint_metadata_resource",
                    type_="foreignkey",
                )

    resource_link_columns = {
        column["name"] for column in inspector.get_columns("resource_links")
    }
    if "link_type" in resource_link_columns:
        check_constraints = inspector.get_check_constraints("resource_links")
        has_link_type_ck = any(
            ck.get("name") == "ck_resource_links_link_type" for ck in check_constraints
        )
        with op.batch_alter_table("resource_links") as batch_op:
            if has_link_type_ck:
                batch_op.drop_constraint("ck_resource_links_link_type", type_="check")
            batch_op.drop_column("link_type")
