"""replace notification attempt stub with a durable M12 outbox

Revision ID: 0008_notification_outbox
Revises: 0007_repository_data_key_epochs
"""

from __future__ import annotations

import json
from datetime import UTC, datetime
from typing import Any

import sqlalchemy as sa
from alembic import op
from backup_tool.ids import new_uuid7

revision = "0008_notification_outbox"
down_revision = "0007_repository_data_key_epochs"
branch_labels = None
depends_on = None


def _now() -> datetime:
    return datetime.now(UTC)


def _parameterized_execute(connection: sa.Connection, statement: object) -> Any:
    """Execute only SQLAlchemy Core statements, never dynamic SQL strings."""
    database = connection.execution_options()
    return database.execute(statement)  # type: ignore[arg-type]


def upgrade() -> None:
    connection = op.get_bind()
    # SQLite batch-rebuilds notification_subscriptions. Its old delivery table
    # references this table, so enforcement must be suspended for this migration
    # only while the legacy rows are copied into the replacement outbox shape.
    if connection.dialect.name == "sqlite":
        connection.exec_driver_sql("PRAGMA foreign_keys=OFF")
    # Extend the existing subscription rows first: pre-M12 rows stay disabled until
    # an operator explicitly configures a new credential/key.
    with op.batch_alter_table("notification_subscriptions") as batch:
        batch.add_column(
            sa.Column("rate_limit_per_minute", sa.Integer(), nullable=False, server_default="60")
        )
        batch.add_column(sa.Column("rate_tokens", sa.Float(), nullable=False, server_default="60"))
        batch.add_column(sa.Column("rate_updated_at", sa.DateTime(timezone=True), nullable=True))
        batch.add_column(sa.Column("revision", sa.Integer(), nullable=False, server_default="1"))
        batch.create_check_constraint("rate_positive", "rate_limit_per_minute > 0")
        batch.create_check_constraint("rate_tokens_nonnegative", "rate_tokens >= 0")
        batch.create_check_constraint("revision_positive", "revision > 0")

    op.create_table(
        "notification_events",
        sa.Column("type", sa.String(96), nullable=False),
        sa.Column("schema_version", sa.Integer(), nullable=False, server_default="1"),
        sa.Column("occurred_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("correlation_id", sa.String(36), nullable=False),
        sa.Column("severity", sa.String(16), nullable=False),
        sa.Column("resource_refs", sa.JSON(), nullable=False),
        sa.Column("payload", sa.JSON(), nullable=False),
        sa.Column("canonical_envelope", sa.Text(), nullable=False),
        sa.Column("deduplication_key", sa.String(255), nullable=True),
        sa.Column("id", sa.String(36), primary_key=True),
        sa.CheckConstraint("schema_version = 1", name="ck_notification_events_schema_version"),
        sa.CheckConstraint(
            "severity IN ('info','warning','error','security')",
            name="ck_notification_events_severity",
        ),
        sa.UniqueConstraint("deduplication_key", name="uq_notification_events_deduplication_key"),
    )
    op.create_index(
        "ix_notification_events_type_occurred", "notification_events", ["type", "occurred_at"]
    )

    # Preserve unexpected rows from the unused baseline shape.  Renaming first
    # keeps the original data intact if an upgrade is interrupted before copy.
    op.rename_table("notification_deliveries", "notification_deliveries_legacy")
    op.create_table(
        "notification_deliveries",
        sa.Column(
            "event_id",
            sa.String(36),
            sa.ForeignKey("notification_events.id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column(
            "subscription_id",
            sa.String(36),
            sa.ForeignKey("notification_subscriptions.id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("state", sa.String(32), nullable=False, server_default="pending"),
        sa.Column("due_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("lease_owner", sa.String(255), nullable=True),
        sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("terminal_reason", sa.String(96), nullable=True),
        sa.Column("response_class", sa.String(64), nullable=True),
        sa.Column("response_summary", sa.String(512), nullable=True),
        sa.Column("id", sa.String(36), primary_key=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
        sa.CheckConstraint(
            "attempt_count >= 0", name="ck_notification_deliveries_attempt_count_nonnegative"
        ),
        sa.CheckConstraint(
            "state IN ('pending','leased','delivered','retry','failed')",
            name="ck_notification_deliveries_state",
        ),
        sa.UniqueConstraint(
            "event_id", "subscription_id", name="uq_notification_deliveries_event_subscription"
        ),
    )
    op.create_index(
        "ix_notification_deliveries_due", "notification_deliveries", ["state", "due_at"]
    )
    op.create_index(
        "ix_notification_deliveries_lease", "notification_deliveries", ["state", "lease_expires_at"]
    )
    op.create_table(
        "notification_delivery_attempts",
        sa.Column(
            "delivery_id",
            sa.String(36),
            sa.ForeignKey("notification_deliveries.id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("number", sa.Integer(), nullable=False),
        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("outcome", sa.String(32), nullable=False, server_default="started"),
        sa.Column("response_class", sa.String(64), nullable=True),
        sa.Column("diagnostic", sa.String(512), nullable=True),
        sa.Column("id", sa.String(36), primary_key=True),
        sa.CheckConstraint("number > 0", name="ck_notification_delivery_attempts_number_positive"),
        sa.CheckConstraint(
            "outcome IN ('started','delivered','retry','failed')",
            name="ck_notification_delivery_attempts_outcome",
        ),
        sa.UniqueConstraint(
            "delivery_id", "number", name="uq_notification_delivery_attempts_delivery_number"
        ),
    )
    op.create_index(
        "ix_notification_attempts_delivery",
        "notification_delivery_attempts",
        ["delivery_id", "number"],
    )

    legacy = sa.table(
        "notification_deliveries_legacy",
        sa.column("id"),
        sa.column("event_id"),
        sa.column("subscription_id"),
        sa.column("attempt"),
        sa.column("state"),
        sa.column("response_class"),
        sa.column("next_attempt_at"),
        sa.column("created_at"),
        sa.column("updated_at"),
    )
    rows = _parameterized_execute(connection, sa.select(legacy)).mappings().all()
    events = sa.table(
        "notification_events",
        sa.column("id"),
        sa.column("type"),
        sa.column("schema_version"),
        sa.column("occurred_at"),
        sa.column("correlation_id"),
        sa.column("severity"),
        sa.column("resource_refs", sa.JSON()),
        sa.column("payload", sa.JSON()),
        sa.column("canonical_envelope"),
        sa.column("deduplication_key"),
    )
    deliveries = sa.table(
        "notification_deliveries",
        *[
            sa.column(name)
            for name in (
                "id",
                "event_id",
                "subscription_id",
                "state",
                "due_at",
                "attempt_count",
                "terminal_reason",
                "response_class",
                "response_summary",
                "created_at",
                "updated_at",
            )
        ],
    )
    attempts = sa.table(
        "notification_delivery_attempts",
        *[
            sa.column(name)
            for name in (
                "id",
                "delivery_id",
                "number",
                "started_at",
                "completed_at",
                "outcome",
                "response_class",
                "diagnostic",
            )
        ],
    )
    for row in rows:
        occurred = row["created_at"] or _now()
        event_id, delivery_id, attempt_id = (str(new_uuid7()), str(new_uuid7()), str(new_uuid7()))
        payload = {"legacy_event_id": str(row["event_id"]), "legacy_delivery_id": str(row["id"])}
        envelope = {
            "event_schema_version": 1,
            "id": event_id,
            "type": "notification.legacy",
            "occurred_at": occurred.isoformat()
            if hasattr(occurred, "isoformat")
            else str(occurred),
            "correlation_id": event_id,
            "severity": "warning",
            "resource": {"subscription_id": str(row["subscription_id"])},
            "payload": payload,
        }
        _parameterized_execute(
            connection,
            events.insert().values(
                id=event_id,
                type="notification.legacy",
                schema_version=1,
                occurred_at=occurred,
                correlation_id=event_id,
                severity="warning",
                resource_refs=envelope["resource"],
                payload=payload,
                canonical_envelope=json.dumps(envelope, sort_keys=True, separators=(",", ":")),
                deduplication_key=f"legacy:{row['id']}",
            ),
        )
        old_state = str(row["state"])
        new_state = "retry" if old_state == "pending" else old_state
        try:
            legacy_attempt = max(1, int(row["attempt"]))
        except (TypeError, ValueError) as error:
            raise RuntimeError("legacy notification delivery has an invalid attempt") from error
        _parameterized_execute(
            connection,
            deliveries.insert().values(
                id=delivery_id,
                event_id=event_id,
                subscription_id=row["subscription_id"],
                state=new_state,
                due_at=row["next_attempt_at"] or occurred,
                attempt_count=legacy_attempt,
                terminal_reason="legacy_migrated" if new_state == "failed" else None,
                response_class=row["response_class"],
                response_summary="legacy delivery migrated",
                created_at=occurred,
                updated_at=row["updated_at"] or occurred,
            ),
        )
        _parameterized_execute(
            connection,
            attempts.insert().values(
                id=attempt_id,
                delivery_id=delivery_id,
                number=legacy_attempt,
                started_at=occurred,
                completed_at=row["updated_at"] if new_state in {"delivered", "failed"} else None,
                outcome="delivered"
                if new_state == "delivered"
                else ("failed" if new_state == "failed" else "retry"),
                response_class=row["response_class"],
                diagnostic="legacy delivery migrated",
            ),
        )
    op.drop_table("notification_deliveries_legacy")

    op.create_table(
        "notification_signing_keys",
        sa.Column(
            "subscription_id",
            sa.String(36),
            sa.ForeignKey("notification_subscriptions.id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("version", sa.Integer(), nullable=False),
        sa.Column(
            "secret_id",
            sa.String(36),
            sa.ForeignKey("secrets.id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("state", sa.String(16), nullable=False, server_default="active"),
        sa.Column("overlap_expires_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("id", sa.String(36), primary_key=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
        sa.CheckConstraint("version > 0", name="ck_notification_signing_keys_version_positive"),
        sa.CheckConstraint(
            "state IN ('active','overlap','retired')", name="ck_notification_signing_keys_state"
        ),
        sa.UniqueConstraint(
            "subscription_id", "version", name="uq_notification_signing_keys_subscription_version"
        ),
    )
    op.create_index(
        "ix_notification_signing_keys_subscription",
        "notification_signing_keys",
        ["subscription_id", "state"],
    )
    op.create_index(
        "uq_notification_signing_keys_active",
        "notification_signing_keys",
        ["subscription_id"],
        unique=True,
        sqlite_where=sa.text("state = 'active'"),
    )
    op.create_index(
        "uq_notification_signing_keys_overlap",
        "notification_signing_keys",
        ["subscription_id"],
        unique=True,
        sqlite_where=sa.text("state = 'overlap'"),
    )
    op.create_table(
        "notification_email_settings",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("host", sa.String(255), nullable=False),
        sa.Column("port", sa.Integer(), nullable=False, server_default="587"),
        sa.Column("username", sa.String(255), nullable=False),
        sa.Column(
            "password_secret_id",
            sa.String(36),
            sa.ForeignKey("secrets.id", ondelete="RESTRICT"),
            nullable=False,
        ),
        sa.Column("sender", sa.String(320), nullable=False),
        sa.Column("max_attempts", sa.Integer(), nullable=False, server_default="5"),
        sa.Column("rate_limit_per_minute", sa.Integer(), nullable=False, server_default="60"),
        sa.CheckConstraint("id = 1", name="ck_notification_email_settings_singleton"),
        sa.CheckConstraint("port > 0 AND port < 65536", name="ck_notification_email_settings_port"),
        sa.CheckConstraint("max_attempts > 0", name="ck_notification_email_settings_max_attempts"),
        sa.CheckConstraint(
            "rate_limit_per_minute > 0", name="ck_notification_email_settings_rate_positive"
        ),
    )
    if connection.dialect.name == "sqlite":
        connection.exec_driver_sql("PRAGMA foreign_keys=ON")


def downgrade() -> None:
    bind = op.get_bind()
    for table in (
        "notification_events",
        "notification_delivery_attempts",
        "notification_signing_keys",
        "notification_email_settings",
    ):
        if bind.scalar(sa.text(f"SELECT count(*) FROM {table}")):
            raise RuntimeError(
                "cannot downgrade while M12 notification history or configuration exists"
            )
    # A fresh M12 schema can safely return to the historical stub shape.
    op.drop_table("notification_email_settings")
    op.drop_index("uq_notification_signing_keys_overlap", table_name="notification_signing_keys")
    op.drop_index("uq_notification_signing_keys_active", table_name="notification_signing_keys")
    op.drop_index(
        "ix_notification_signing_keys_subscription", table_name="notification_signing_keys"
    )
    op.drop_table("notification_signing_keys")
    op.drop_index("ix_notification_attempts_delivery", table_name="notification_delivery_attempts")
    op.drop_table("notification_delivery_attempts")
    op.drop_index("ix_notification_deliveries_lease", table_name="notification_deliveries")
    op.drop_index("ix_notification_deliveries_due", table_name="notification_deliveries")
    op.drop_table("notification_deliveries")
    op.drop_index("ix_notification_events_type_occurred", table_name="notification_events")
    op.drop_table("notification_events")
    op.create_table(
        "notification_deliveries",
        sa.Column("event_id", sa.String(36), nullable=False),
        sa.Column("subscription_id", sa.String(36), nullable=False),
        sa.Column("attempt", sa.Integer(), nullable=False),
        sa.Column("state", sa.String(32), nullable=False),
        sa.Column("response_class", sa.String(64)),
        sa.Column("next_attempt_at", sa.DateTime(timezone=True)),
        sa.Column("id", sa.String(36), primary_key=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("CURRENT_TIMESTAMP"),
        ),
        sa.CheckConstraint("attempt > 0", name="ck_notification_deliveries_attempt_positive"),
        sa.CheckConstraint(
            "state IN ('pending','delivered','retry','failed')",
            name="ck_notification_deliveries_state",
        ),
        sa.ForeignKeyConstraint(
            ["subscription_id"], ["notification_subscriptions.id"], ondelete="RESTRICT"
        ),
        sa.UniqueConstraint("event_id", "subscription_id", "attempt", name="uq_delivery_attempt"),
    )
    op.create_index(
        "ix_notification_deliveries_state_next",
        "notification_deliveries",
        ["state", "next_attempt_at"],
    )
    with op.batch_alter_table("notification_subscriptions") as batch:
        batch.drop_constraint("revision_positive", type_="check")
        batch.drop_constraint("rate_tokens_nonnegative", type_="check")
        batch.drop_constraint("rate_positive", type_="check")
        batch.drop_column("revision")
        batch.drop_column("rate_updated_at")
        batch.drop_column("rate_tokens")
        batch.drop_column("rate_limit_per_minute")