"""The decisions ledger, the proposals store, places, and the columns every later stream
reads (Stream K2; the catalogue refactor plan W7, W15, W18, W19; the escalation-2 spec
`.logs/planning/decisions-model-spec-2026-09-16.md` §1 and §9, read through K1's names).

Schema-only. New: `proposal_passes`, `decision_batches`, `proposals`, `decisions` (append-only,
held by the Postgres trigger `decisions_append_only`; `decision_batches_guard` allows exactly
one update, the close), `redirects` (replaces the spec's `line_slugs`: brands, product lines and
places share it), `places` and `shop_places`. Added: `uid` on brands, product lines, product
variants and suggestions (born at insert, the identity a decision names across hosts);
`hidden` and `indexed` on brands and product lines with partial indexes; `closed_reason` and
`decision_id` on suggestions with the `separate` partial index; `decision_id`, `batch_id`,
`reversed_by_id` and `reversed_at` on merges; `product_lines.key` widened so an approved line's
`decided:<slug>` key fits. Dropped: `overrides`, after asserting it holds 0 rows (0 on every host;
the downgrade recreates it exactly as migration #4 defined it). No CHECK on
`suggestions.decision` in this revision: the launch chain runs in one pass and K1's backfill
runs after it. Revises `c1d2e3f4a5b6`, the single head at the time of writing.

Revision ID: d2e3f4a5b6c7
Revises: c1d2e3f4a5b6
Create Date: 2026-09-17
"""

import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import JSONB

from alembic import op

revision = "d2e3f4a5b6c7"
down_revision = "c1d2e3f4a5b6"
branch_labels = None
depends_on = None

JSON = sa.JSON().with_variant(JSONB(), "postgresql")
UUID = sa.Uuid(as_uuid=True)
TS = sa.DateTime(timezone=True)
ENTITY_TYPES = ("brand", "product_line", "product_variant", "listing", "attribute_wording", "suggestion", "place")
FIELD_RE = r"^[a-z_]+(:[a-z0-9_-]{1,40}){0,2}$"

UID_TABLES = ("brands", "product_lines", "product_variants", "suggestions")
FLAG_TABLES = ("brands", "product_lines")


def _in(col: str, values: tuple[str, ...]) -> str:
    return f"{col} IN ({', '.join(repr(v) for v in values)})"


def _postgres() -> bool:
    return op.get_bind().dialect.name == "postgresql"


def upgrade() -> None:
    bind = op.get_bind()
    pg = _postgres()

    # 0. overrides goes, after the assertion (0 rows on every host).
    n = bind.execute(sa.text("SELECT count(*) FROM overrides")).scalar()
    if n:
        raise RuntimeError(f"overrides holds {n} row(s); the ledger migration expects 0 (export them first)")
    op.drop_table("overrides")

    # 1. uids born at insert on every row a decision can name.
    for table in UID_TABLES:
        op.add_column(table, sa.Column("uid", UUID, nullable=False,
                                       server_default=sa.text("gen_random_uuid()") if pg else None))
        op.create_unique_constraint(f"uq_{table}_uid", table, ["uid"])

    # 2. hidden and indexed, materialised for W18's two decisions.
    for table in FLAG_TABLES:
        op.add_column(table, sa.Column("hidden", sa.Boolean(), nullable=False, server_default=sa.text("false")))
        op.add_column(table, sa.Column("indexed", sa.Boolean(), nullable=False, server_default=sa.text("false")))
        op.create_index(f"ix_{table}_hidden", table, ["id"], postgresql_where=sa.text("hidden"), sqlite_where=sa.text("hidden"))
        op.create_index(f"ix_{table}_indexed", table, ["id"], postgresql_where=sa.text("indexed"), sqlite_where=sa.text("indexed"))

    # An approved line's key is 'decided:' || slug (spec §2); the slug is up to 240.
    op.alter_column("product_lines", "key", type_=sa.String(260), existing_type=sa.String(160))

    # 3. places and shop places (W19).
    op.create_table(
        "places",
        sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), primary_key=True),
        sa.Column("uid", UUID, nullable=False, server_default=sa.text("gen_random_uuid()") if pg else None),
        sa.Column("slug", sa.String(240), nullable=False),
        sa.Column("kind", sa.String(24), nullable=False),
        sa.Column("name", sa.String(200), nullable=False),
        sa.Column("parent_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("places.id"), nullable=True),
        sa.Column("identifiers", JSON, nullable=False, server_default="[]"),
        sa.Column("attributes", JSON, nullable=False, server_default="{}"),
        sa.Column("hidden", sa.Boolean(), nullable=False, server_default=sa.text("false")),
        sa.Column("indexed", sa.Boolean(), nullable=False, server_default=sa.text("false")),
        sa.Column("created_at", TS, nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", TS, nullable=False, server_default=sa.func.now()),
        sa.UniqueConstraint("slug", name="uq_places_slug"),
        sa.UniqueConstraint("uid", name="uq_places_uid"),
    )
    op.create_index("ix_places_kind", "places", ["kind"])
    op.create_index("ix_places_parent_id", "places", ["parent_id"])
    op.create_index("ix_places_hidden", "places", ["id"], postgresql_where=sa.text("hidden"), sqlite_where=sa.text("hidden"))
    op.create_index("ix_places_indexed", "places", ["id"], postgresql_where=sa.text("indexed"), sqlite_where=sa.text("indexed"))
    op.create_table(
        "shop_places",
        sa.Column("shop_id", sa.Integer(), sa.ForeignKey("shops.id"), nullable=False),
        sa.Column("place_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("places.id"), nullable=False),
        sa.Column("role", sa.String(16), nullable=False),
        sa.PrimaryKeyConstraint("shop_id", "place_id", name="pk_shop_places"),
        sa.CheckConstraint("role IN ('primary', 'also')", name="ck_shop_places_role"),
    )
    op.create_index("ix_shop_places_place", "shop_places", ["place_id"])
    op.create_index("uq_shop_places_primary", "shop_places", ["shop_id"], unique=True,
                    postgresql_where=sa.text("role = 'primary'"), sqlite_where=sa.text("role = 'primary'"))

    # 4. the proposals store and the ledger, in FK order (spec §1.1 to §1.4).
    op.create_table(
        "proposal_passes",
        sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), primary_key=True),
        sa.Column("name", sa.String(120), nullable=False),
        sa.Column("kind", sa.String(12), nullable=False),
        sa.Column("process_version", sa.String(20), nullable=False),
        sa.Column("rules_version", sa.String(8), nullable=False),
        sa.Column("generator", sa.String(80), nullable=False),
        sa.Column("scope_brand_slug", sa.String(160), nullable=True),
        sa.Column("loaded_at", TS, nullable=False, server_default=sa.func.now()),
        sa.Column("loaded_by", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=True),
        sa.Column("loaded_by_username", sa.String(64), nullable=True),
        sa.Column("source_file", sa.Text(), nullable=True),
        sa.Column("file_sha256", sa.String(64), nullable=True),
        sa.Column("counts", JSON, nullable=False, server_default="{}"),
        sa.Column("note", sa.Text(), nullable=True),
        sa.Column("withdrawn_at", TS, nullable=True),
        sa.Column("withdrawn_by", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=True),
        sa.Column("withdrawn_reason", sa.Text(), nullable=True),
        sa.UniqueConstraint("name", name="uq_proposal_passes_name"),
        sa.CheckConstraint(_in("kind", ("session", "rule", "arrival")), name="ck_proposal_passes_kind"),
        sa.CheckConstraint("(withdrawn_at IS NULL) = (withdrawn_by IS NULL)", name="ck_proposal_passes_withdrawn"),
    )
    op.create_index("ix_proposal_passes_kind_loaded", "proposal_passes", ["kind", "loaded_at"])
    op.create_index("ix_proposal_passes_scope", "proposal_passes", ["scope_brand_slug"])

    op.create_table(
        "decision_batches",
        sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), primary_key=True),
        sa.Column("uid", UUID, nullable=False),
        sa.Column("opened_at", TS, nullable=False, server_default=sa.func.now()),
        sa.Column("closed_at", TS, nullable=True),
        sa.Column("origin_host", sa.String(20), nullable=False),
        sa.Column("kind", sa.String(12), nullable=False),
        sa.Column("mode", sa.String(10), nullable=False),
        sa.Column("by_account_id", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=True),
        sa.Column("by_username", sa.String(64), nullable=True),
        sa.Column("acting_as_id", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=True),
        sa.Column("pass_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("proposal_passes.id"), nullable=True),
        sa.Column("scope", JSON, nullable=True),
        sa.Column("reverses_batch_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("decision_batches.id"), nullable=True),
        sa.Column("replayed_from", sa.String(60), nullable=True),
        sa.Column("note", sa.Text(), nullable=True),
        sa.Column("summary", JSON, nullable=False, server_default="{}"),
        sa.UniqueConstraint("uid", name="uq_decision_batches_uid"),
        sa.CheckConstraint(_in("kind", ("sheet", "desk", "route", "cli", "undo", "replay")), name="ck_decision_batches_kind"),
        sa.CheckConstraint(_in("mode", ("individual", "bulk")), name="ck_decision_batches_mode"),
        sa.CheckConstraint("(kind = 'undo') = (reverses_batch_id IS NOT NULL)", name="ck_decision_batches_undo"),
    )
    op.create_index("ix_decision_batches_opened", "decision_batches", ["opened_at"])
    op.create_index("ix_decision_batches_pass", "decision_batches", ["pass_id"], postgresql_where=sa.text("pass_id IS NOT NULL"))
    op.create_index("ix_decision_batches_reverses", "decision_batches", ["reverses_batch_id"], postgresql_where=sa.text("reverses_batch_id IS NOT NULL"))
    op.create_index("ix_decision_batches_by", "decision_batches", ["by_account_id", "opened_at"])

    op.create_table(
        "proposals",
        sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), primary_key=True),
        sa.Column("uid", UUID, nullable=False),
        sa.Column("pass_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("proposal_passes.id"), nullable=False),
        sa.Column("brand_slug", sa.String(160), nullable=False),
        sa.Column("sheet_line_ref", sa.String(500), nullable=True),
        sa.Column("position", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("entity_type", sa.String(20), nullable=False),
        sa.Column("natural_key", sa.String(500), nullable=False),
        sa.Column("natural_key_detail", JSON, nullable=True),
        sa.Column("entity_id", sa.BigInteger(), nullable=True),
        sa.Column("resolution", sa.String(12), nullable=False, server_default="unresolved"),
        sa.Column("field", sa.String(80), nullable=False),
        sa.Column("value", JSON, nullable=False),
        sa.Column("rule_value", JSON, nullable=True),
        sa.Column("current_value", JSON, nullable=True),
        sa.Column("against_decision_id", sa.BigInteger(), nullable=True),
        sa.Column("generator", sa.String(80), nullable=False),
        sa.Column("reason", sa.Text(), nullable=True),
        sa.Column("evidence", JSON, nullable=False, server_default="[]"),
        sa.Column("confidence", sa.Numeric(4, 3), nullable=True),
        sa.Column("spot_check", sa.Boolean(), nullable=False, server_default=sa.text("false")),
        sa.Column("status", sa.String(12), nullable=False, server_default="open"),
        sa.Column("corrected_value", JSON, nullable=True),
        sa.Column("decision_id", sa.BigInteger(), nullable=True),
        sa.Column("resolution_note", sa.Text(), nullable=True),
        sa.Column("resolved_at", TS, nullable=True),
        sa.Column("resolved_by", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=True),
        sa.Column("detail", JSON, nullable=False, server_default="{}"),
        sa.Column("created_at", TS, nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", TS, nullable=False, server_default=sa.func.now()),
        sa.UniqueConstraint("uid", name="uq_proposals_uid"),
        sa.UniqueConstraint("pass_id", "entity_type", "natural_key", "field", name="uq_proposals_pass_target"),
        sa.CheckConstraint(_in("entity_type", ENTITY_TYPES), name="ck_proposals_entity_type"),
        sa.CheckConstraint(_in("resolution", ("resolved", "creatable", "unresolved")), name="ck_proposals_resolution"),
        sa.CheckConstraint(_in("status", ("open", "approved", "rejected", "withdrawn", "stale", "parked")), name="ck_proposals_status"),
        sa.CheckConstraint("confidence IS NULL OR (confidence >= 0 AND confidence <= 1)", name="ck_proposals_confidence"),
        sa.CheckConstraint("(status = 'approved') = (decision_id IS NOT NULL) OR status = 'rejected'", name="ck_proposals_decided"),
    )
    op.create_index("ix_proposals_sheet", "proposals", ["brand_slug", "status", "sheet_line_ref", "position"])
    op.create_index("ix_proposals_pass_status", "proposals", ["pass_id", "status"])
    op.create_index("ix_proposals_target", "proposals", ["entity_type", "natural_key", "field"])
    op.create_index("ix_proposals_entity", "proposals", ["entity_type", "entity_id", "field"], postgresql_where=sa.text("entity_id IS NOT NULL"))
    op.create_index("ix_proposals_spot", "proposals", ["brand_slug"], postgresql_where=sa.text("spot_check AND status = 'open'"))

    op.create_table(
        "decisions",
        sa.Column("id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), primary_key=True),
        sa.Column("uid", UUID, nullable=False),
        sa.Column("decided_at", TS, nullable=False),
        sa.Column("recorded_at", TS, nullable=False, server_default=sa.func.now()),
        sa.Column("origin_host", sa.String(20), nullable=False),
        sa.Column("entity_type", sa.String(20), nullable=False),
        sa.Column("entity_id", sa.BigInteger(), nullable=False),
        sa.Column("natural_key", sa.String(500), nullable=False),
        sa.Column("natural_key_detail", JSON, nullable=True),
        sa.Column("field", sa.String(80), nullable=False),
        sa.Column("effect", sa.String(8), nullable=False, server_default="set"),
        sa.Column("value", JSON, nullable=True),
        sa.Column("value_ref_id", sa.BigInteger(), nullable=True),
        sa.Column("rule_value", JSON, nullable=True),
        sa.Column("prior_value", JSON, nullable=True),
        sa.Column("supersedes_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("decisions.id"), nullable=True),
        sa.Column("reverses_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("decisions.id"), nullable=True),
        sa.Column("restores_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("decisions.id"), nullable=True),
        sa.Column("caused_by_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("decisions.id"), nullable=True),
        sa.Column("rules_version", sa.String(8), nullable=False),
        sa.Column("origin", sa.String(10), nullable=False),
        sa.Column("proposal_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), nullable=True),
        sa.Column("pass_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("proposal_passes.id"), nullable=True),
        sa.Column("batch_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("decision_batches.id"), nullable=False),
        sa.Column("mode", sa.String(10), nullable=False),
        sa.Column("decided_by", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=True),
        sa.Column("decided_by_username", sa.String(64), nullable=True),
        sa.Column("reason", sa.Text(), nullable=True),
        sa.Column("replayed_from", sa.String(60), nullable=True),
        sa.Column("detail", JSON, nullable=True),
        sa.UniqueConstraint("uid", name="uq_decisions_uid"),
        sa.CheckConstraint(_in("entity_type", ENTITY_TYPES), name="ck_decisions_entity_type"),
        sa.CheckConstraint(_in("effect", ("set", "release")), name="ck_decisions_effect"),
        sa.CheckConstraint(_in("origin", ("person", "proposal")), name="ck_decisions_origin"),
        sa.CheckConstraint(_in("mode", ("individual", "bulk")), name="ck_decisions_mode"),
        sa.CheckConstraint("(effect = 'release') = (value IS NULL)", name="ck_decisions_release_value"),
        sa.CheckConstraint("reverses_id IS NULL OR origin = 'person'", name="ck_decisions_undo_is_a_person"),
        sa.CheckConstraint("restores_id IS NULL OR reverses_id IS NOT NULL", name="ck_decisions_restores"),
        sa.CheckConstraint("origin <> 'proposal' OR pass_id IS NOT NULL", name="ck_decisions_proposal_pass"),
    )
    if pg:
        op.create_check_constraint("ck_decisions_field", "decisions", sa.text(f"field ~ '{FIELD_RE}'"))
        op.create_check_constraint("ck_proposals_field", "proposals", sa.text(f"field ~ '{FIELD_RE}'"))
    op.create_index("ix_decisions_effective", "decisions",
                    ["entity_type", "entity_id", "field", sa.text("decided_at DESC"), sa.text("id DESC")])
    op.create_index("ix_decisions_natural", "decisions", ["entity_type", "natural_key"])
    op.create_index("ix_decisions_batch", "decisions", ["batch_id", "id"])
    op.create_index("ix_decisions_pass", "decisions", ["pass_id", "id"], postgresql_where=sa.text("pass_id IS NOT NULL"))
    op.create_index("ix_decisions_proposal", "decisions", ["proposal_id"], postgresql_where=sa.text("proposal_id IS NOT NULL"))
    op.create_index("ix_decisions_caused_by", "decisions", ["caused_by_id"], postgresql_where=sa.text("caused_by_id IS NOT NULL"))
    op.create_index("ix_decisions_reverses", "decisions", ["reverses_id"], unique=True, postgresql_where=sa.text("reverses_id IS NOT NULL"))
    op.create_index("ix_decisions_decided_at", "decisions", ["decided_at"])
    op.create_index("ix_decisions_by", "decisions", ["decided_by", "decided_at"])
    # The cycle, closed here by name in both directions so the downgrade can open it first.
    op.create_foreign_key("fk_decisions_proposal", "decisions", "proposals", ["proposal_id"], ["id"])
    op.create_foreign_key("fk_proposals_decision", "proposals", "decisions", ["decision_id"], ["id"])
    op.create_foreign_key("fk_proposals_against_decision", "proposals", "decisions", ["against_decision_id"], ["id"])

    # 5. redirects (brief decision 7): the slug history for brands, product lines and places.
    op.create_table(
        "redirects",
        sa.Column("from_slug", sa.String(240), primary_key=True),
        sa.Column("kind", sa.String(16), nullable=False),
        sa.Column("to_slug", sa.String(240), nullable=False),
        sa.Column("since", TS, nullable=False, server_default=sa.func.now()),
        sa.Column("decision_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), sa.ForeignKey("decisions.id"), nullable=True),
        sa.CheckConstraint(_in("kind", ("brand", "product_line", "place")), name="ck_redirects_kind"),
    )
    op.create_index("ix_redirects_kind_to", "redirects", ["kind", "to_slug"])

    # 6. suggestions and merges beside the ledger (spec §1.6, §5).
    op.add_column("suggestions", sa.Column("closed_reason", sa.String(16), nullable=True))
    op.add_column("suggestions", sa.Column("decision_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"),
                                           sa.ForeignKey("decisions.id", name="fk_suggestions_decision"), nullable=True))
    op.create_index("ix_suggestions_separate", "suggestions", ["level", "left_id", "right_id"],
                    postgresql_where=sa.text("decision = 'separate'"), sqlite_where=sa.text("decision = 'separate'"))
    op.add_column("merges", sa.Column("decision_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"),
                                      sa.ForeignKey("decisions.id", name="fk_merges_decision"), nullable=True))
    op.add_column("merges", sa.Column("batch_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"),
                                      sa.ForeignKey("decision_batches.id", name="fk_merges_batch"), nullable=True))
    op.add_column("merges", sa.Column("reversed_by_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"),
                                      sa.ForeignKey("decisions.id", name="fk_merges_reversed_by"), nullable=True))
    op.add_column("merges", sa.Column("reversed_at", TS, nullable=True))
    op.create_index("ix_merges_decision", "merges", ["decision_id"], postgresql_where=sa.text("decision_id IS NOT NULL"))
    op.create_index("ix_merges_batch", "merges", ["batch_id"], postgresql_where=sa.text("batch_id IS NOT NULL"))

    # 7. the two triggers: the ledger is append-only; a batch closes once and changes nothing else.
    if pg:
        op.execute("""
            CREATE FUNCTION decisions_append_only() RETURNS trigger AS $$
            BEGIN
                RAISE EXCEPTION 'decisions is append-only: % refused on decision %', TG_OP, OLD.uid;
            END $$ LANGUAGE plpgsql;
            CREATE TRIGGER decisions_append_only BEFORE UPDATE OR DELETE ON decisions
                FOR EACH ROW EXECUTE FUNCTION decisions_append_only();
            CREATE FUNCTION decision_batches_guard() RETURNS trigger AS $$
            BEGIN
                IF TG_OP = 'DELETE' THEN
                    RAISE EXCEPTION 'decision_batches is append-only: DELETE refused on batch %', OLD.uid;
                END IF;
                IF OLD.closed_at IS NOT NULL THEN
                    RAISE EXCEPTION 'decision_batches: batch % is closed', OLD.uid;
                END IF;
                IF NEW.closed_at IS NULL THEN
                    RAISE EXCEPTION 'decision_batches: the only update is the close (batch %)', OLD.uid;
                END IF;
                IF (to_jsonb(OLD) - 'closed_at' - 'summary') <> (to_jsonb(NEW) - 'closed_at' - 'summary') THEN
                    RAISE EXCEPTION 'decision_batches: only closed_at and summary change at close (batch %)', OLD.uid;
                END IF;
                RETURN NEW;
            END $$ LANGUAGE plpgsql;
            CREATE TRIGGER decision_batches_guard BEFORE UPDATE OR DELETE ON decision_batches
                FOR EACH ROW EXECUTE FUNCTION decision_batches_guard();
        """)


def downgrade() -> None:
    pg = _postgres()
    if pg:
        op.execute("""
            DROP TRIGGER IF EXISTS decision_batches_guard ON decision_batches;
            DROP FUNCTION IF EXISTS decision_batches_guard();
            DROP TRIGGER IF EXISTS decisions_append_only ON decisions;
            DROP FUNCTION IF EXISTS decisions_append_only();
        """)
    op.drop_index("ix_merges_batch", table_name="merges")
    op.drop_index("ix_merges_decision", table_name="merges")
    for name in ("fk_merges_reversed_by", "fk_merges_batch", "fk_merges_decision"):
        op.drop_constraint(name, "merges", type_="foreignkey")
    for col in ("reversed_at", "reversed_by_id", "batch_id", "decision_id"):
        op.drop_column("merges", col)
    op.drop_index("ix_suggestions_separate", table_name="suggestions")
    op.drop_constraint("fk_suggestions_decision", "suggestions", type_="foreignkey")
    op.drop_column("suggestions", "decision_id")
    op.drop_column("suggestions", "closed_reason")
    op.drop_table("redirects")
    # The cycle first.
    op.drop_constraint("fk_proposals_against_decision", "proposals", type_="foreignkey")
    op.drop_constraint("fk_proposals_decision", "proposals", type_="foreignkey")
    op.drop_constraint("fk_decisions_proposal", "decisions", type_="foreignkey")
    op.drop_table("decisions")
    op.drop_table("proposals")
    op.drop_table("decision_batches")
    op.drop_table("proposal_passes")
    op.drop_table("shop_places")
    op.drop_table("places")
    op.alter_column("product_lines", "key", type_=sa.String(160), existing_type=sa.String(260))
    for table in FLAG_TABLES:
        op.drop_index(f"ix_{table}_indexed", table_name=table)
        op.drop_index(f"ix_{table}_hidden", table_name=table)
        op.drop_column(table, "indexed")
        op.drop_column(table, "hidden")
    for table in UID_TABLES:
        op.drop_constraint(f"uq_{table}_uid", table, type_="unique")
        op.drop_column(table, "uid")
    # overrides, exactly as migration #4 (b5c6d7e8f9a0) defined it.
    op.create_table(
        "overrides",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("entity_type", sa.String(40), nullable=False),
        sa.Column("entity_key", sa.Text(), nullable=False),
        sa.Column("field", sa.String(60), nullable=False),
        sa.Column("value", JSONB, nullable=True),
        sa.Column("collected_value", JSONB, nullable=True),
        sa.Column("set_by", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=True),
        sa.Column("set_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
        sa.Column("reason", sa.Text(), nullable=True),
        sa.Column("collector_disagrees_since", sa.DateTime(timezone=True), nullable=True),
        sa.UniqueConstraint("entity_type", "entity_key", "field", name="uq_overrides_entity_field"),
    )