"""The full R1 schema (02-domain-model.md as amended; 09-build-plan M2).

Tables: parties, tenants, engagements, compensation_terms (the
kind=cost|billout worker-terms model), projects (uniqueness
(tenant_id, client_party_id, lower(name)) — judgment #33),
rate_overrides (own table — judgment #25), import_batches, time_entries
(raw_* import payload + source_timezone + per-side currency stamps),
invoices + invoice_lines (R1 keeps today's invoice shape + party links —
judgment #24), cc_* tables, comments (tenant_id + nullable anchors),
audit_log (unified audit/undo — full before-images for bulk ops), and
migration_runs (the transform run-report table).

NO sets/tasks tables (judgment #32 — R3 creates them from the Airtable
spec). Also completes users.person_party_id: FK -> parties + unique
(deferred from M1 — ADR #002 #5).

Money: Numeric(12,2) + CHAR(3) currency, stamped as a pair (02 §3);
source_rate/source_amount keep 4dp (raw import provenance — rebuild
delta #2 decided explicitly). Entry times stay timezone-NAIVE (07 #17).

Revision ID: 0003
Revises: 0002
Create Date: 2026-07-07
"""

from collections.abc import Sequence

import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

from alembic import op

revision: str = "0003"
down_revision: str | None = "0002"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

ZERO_UUID = "00000000-0000-0000-0000-000000000000"


def _uuid_pk() -> sa.Column:
    return sa.Column(
        "id",
        postgresql.UUID(as_uuid=True),
        primary_key=True,
        server_default=sa.text("gen_random_uuid()"),
    )


def _created_at() -> sa.Column:
    return sa.Column(
        "created_at",
        sa.DateTime(timezone=True),
        nullable=False,
        server_default=sa.text("now()"),
    )


def _updated_at() -> sa.Column:
    return sa.Column(
        "updated_at",
        sa.DateTime(timezone=True),
        nullable=False,
        server_default=sa.text("now()"),
    )


def upgrade() -> None:
    # --- identity and actors (02 §1) -----------------------------------
    op.create_table(
        "parties",
        _uuid_pk(),
        sa.Column("kind", sa.Text(), nullable=False),
        sa.Column("name", sa.Text(), nullable=False),
        sa.Column("slug", sa.Text(), nullable=False),
        sa.Column("email", sa.Text(), nullable=True),
        sa.Column("phone", sa.Text(), nullable=True),
        sa.Column("address", sa.Text(), nullable=True),
        sa.Column("notes", sa.Text(), nullable=True),
        sa.Column(
            "is_active", sa.Boolean(), nullable=False, server_default=sa.text("true")
        ),
        _created_at(),
        _updated_at(),
        sa.CheckConstraint("kind IN ('person', 'org')", name="ck_parties_kind"),
        sa.CheckConstraint("length(trim(name)) > 0", name="ck_parties_name_nonempty"),
        sa.UniqueConstraint("slug", name="uq_parties_slug"),
    )

    op.create_table(
        "tenants",
        _uuid_pk(),
        sa.Column(
            "party_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id"),
            nullable=False,
        ),
        sa.Column(
            "settings",
            postgresql.JSONB(),
            nullable=False,
            server_default=sa.text("'{}'::jsonb"),
        ),
        _created_at(),
        sa.UniqueConstraint("party_id", name="uq_tenants_party_id"),
    )

    # users.person_party_id becomes a real FK + unique (every user IS a
    # person) — deferred from M1 (ADR #002 #5).
    op.create_foreign_key(
        "fk_users_person_party_id", "users", "parties", ["person_party_id"], ["id"]
    )
    op.create_unique_constraint("uq_users_person_party_id", "users", ["person_party_id"])

    # --- engagements — the relationship graph (02 §2) -------------------
    op.create_table(
        "engagements",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column("type", sa.Text(), nullable=False),
        sa.Column(
            "party_a_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id"),
            nullable=False,
        ),
        sa.Column(
            "party_b_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id"),
            nullable=False,
        ),
        sa.Column("role_a", sa.Text(), nullable=False),
        sa.Column("role_b", sa.Text(), nullable=False),
        sa.Column(
            "transparency",
            sa.Text(),
            nullable=False,
            server_default=sa.text("'transparent'"),
        ),
        sa.Column("started_on", sa.Date(), nullable=True),
        sa.Column("ended_on", sa.Date(), nullable=True),
        sa.Column(
            "can_manage_imports",
            sa.Boolean(),
            nullable=False,
            server_default=sa.text("false"),
        ),
        sa.Column("consolidate_as", sa.Text(), nullable=True),
        sa.Column("notes", sa.Text(), nullable=True),
        _created_at(),
        _updated_at(),
        sa.CheckConstraint(
            "type IN ('subcontract', 'freelance', 'partnership', 'referral',"
            " 'referral_pass_through', 'direct_client', 'employment',"
            " 'official_partnership')",
            name="ck_engagements_type",
        ),
        sa.CheckConstraint(
            "transparency IN ('transparent', 'opaque')",
            name="ck_engagements_transparency",
        ),
    )
    op.create_index("ix_engagements_tenant", "engagements", ["tenant_id"])
    op.create_index("ix_engagements_party_a", "engagements", ["party_a_id"])
    op.create_index("ix_engagements_party_b", "engagements", ["party_b_id"])

    # --- projects (02 §3; judgment #33 uniqueness) -----------------------
    op.create_table(
        "projects",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column(
            "operator_party_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id"),
            nullable=False,
        ),
        sa.Column(
            "client_party_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id"),
            nullable=False,
        ),
        sa.Column("name", sa.Text(), nullable=False),
        sa.Column(
            "status", sa.Text(), nullable=False, server_default=sa.text("'active'")
        ),
        sa.Column("income", sa.Numeric(12, 2), nullable=True),
        sa.Column("income_currency", sa.CHAR(3), nullable=True),
        sa.Column("billout_adjustment_pct", sa.Numeric(7, 4), nullable=True),
        sa.Column("billout_adjustment_amount", sa.Numeric(12, 2), nullable=True),
        sa.Column("notes", sa.Text(), nullable=True),
        _created_at(),
        _updated_at(),
        sa.CheckConstraint("length(trim(name)) > 0", name="ck_projects_name_nonempty"),
        sa.CheckConstraint("status IN ('active', 'archived')", name="ck_projects_status"),
        sa.CheckConstraint(
            "(income IS NULL) = (income_currency IS NULL)",
            name="ck_projects_income_currency_pair",
        ),
    )
    op.create_index(
        "uq_projects_tenant_client_lower_name",
        "projects",
        ["tenant_id", "client_party_id", sa.text("lower(name)")],
        unique=True,
    )
    op.create_index("ix_projects_operator_party", "projects", ["operator_party_id"])
    op.create_index("ix_projects_client_party", "projects", ["client_party_id"])

    # --- compensation terms + rate overrides (02 §2) ---------------------
    op.create_table(
        "compensation_terms",
        _uuid_pk(),
        sa.Column(
            "engagement_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("engagements.id", ondelete="CASCADE"),
            nullable=False,
        ),
        sa.Column("kind", sa.Text(), nullable=False),
        sa.Column("model", sa.Text(), nullable=False),
        sa.Column("rate_amount", sa.Numeric(12, 2), nullable=True),
        sa.Column("currency", sa.CHAR(3), nullable=True),
        sa.Column("percentage", sa.Numeric(7, 4), nullable=True),
        sa.Column(
            "project_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("projects.id", ondelete="CASCADE"),
            nullable=True,
        ),
        sa.Column("effective_from", sa.Date(), nullable=False),
        sa.Column("effective_to", sa.Date(), nullable=True),
        _created_at(),
        _updated_at(),
        sa.CheckConstraint("kind IN ('cost', 'billout')", name="ck_compensation_terms_kind"),
        sa.CheckConstraint(
            "model IN ('hourly', 'per_unit', 'fixed_price', 'prepaid_block',"
            " 'profit_share', 'commission', 'gross_percentage', 'salary')",
            name="ck_compensation_terms_model",
        ),
        sa.CheckConstraint(
            "(rate_amount IS NULL) = (currency IS NULL)",
            name="ck_compensation_terms_currency_pair",
        ),
    )
    op.create_index(
        "ix_compensation_terms_engagement", "compensation_terms", ["engagement_id"]
    )
    op.create_index("ix_compensation_terms_project", "compensation_terms", ["project_id"])

    op.create_table(
        "rate_overrides",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column(
            "project_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("projects.id", ondelete="CASCADE"),
            nullable=False,
        ),
        sa.Column(
            "worker_party_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id", ondelete="CASCADE"),
            nullable=True,
        ),
        sa.Column("override_rate", sa.Numeric(12, 2), nullable=True),
        sa.Column("currency", sa.CHAR(3), nullable=True),
        sa.Column("override_pct", sa.Numeric(7, 4), nullable=True),
        sa.Column("notes", sa.Text(), nullable=True),
        _created_at(),
        _updated_at(),
        sa.CheckConstraint(
            "(override_rate IS NOT NULL AND override_pct IS NULL)"
            " OR (override_rate IS NULL AND override_pct IS NOT NULL)",
            name="ck_rate_overrides_one_kind",
        ),
        sa.CheckConstraint(
            "override_rate IS NULL OR override_rate >= 0",
            name="ck_rate_overrides_rate_nonnegative",
        ),
        sa.CheckConstraint(
            "(override_rate IS NULL) = (currency IS NULL)",
            name="ck_rate_overrides_currency_pair",
        ),
    )
    op.create_index(
        "uq_rate_overrides_project_worker",
        "rate_overrides",
        ["project_id", sa.text(f"coalesce(worker_party_id, '{ZERO_UUID}'::uuid)")],
        unique=True,
    )
    op.create_index("ix_rate_overrides_project", "rate_overrides", ["project_id"])
    op.create_index("ix_rate_overrides_worker", "rate_overrides", ["worker_party_id"])

    # --- imports (02 §3; ADR #041 two-step flow lands at M4) -------------
    op.create_table(
        "import_batches",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column("source", sa.Text(), nullable=True),
        sa.Column("filename", sa.Text(), nullable=True),
        sa.Column("name", sa.Text(), nullable=True),
        sa.Column("notes", sa.Text(), nullable=True),
        sa.Column("row_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
        sa.Column(
            "imported_by",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column(
            "imported_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("now()"),
        ),
        sa.CheckConstraint(
            "source IS NULL OR source IN ('clockify', 'toggl', 'manual', 'backfill')",
            name="ck_import_batches_source",
        ),
    )
    op.create_index(
        "ix_import_batches_tenant_date", "import_batches", ["tenant_id", "imported_at"]
    )

    # --- invoices + lines (02 §3; judgments #9, #24) ----------------------
    op.create_table(
        "invoices",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column("name", sa.Text(), nullable=False),
        sa.Column(
            "status", sa.Text(), nullable=False, server_default=sa.text("'open'")
        ),
        sa.Column("currency", sa.CHAR(3), nullable=False, server_default=sa.text("'CAD'")),
        sa.Column("invoice_date", sa.Date(), nullable=True),
        sa.Column("sent_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("paid_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column(
            "from_party_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id"),
            nullable=True,
        ),
        sa.Column(
            "to_party_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id"),
            nullable=True,
        ),
        sa.Column("manual_total", sa.Numeric(12, 2), nullable=True),
        sa.Column("notes", sa.Text(), nullable=True),
        sa.Column(
            "created_by",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="SET NULL"),
            nullable=True,
        ),
        _created_at(),
        sa.CheckConstraint("status IN ('open', 'sent', 'paid')", name="ck_invoices_status"),
    )
    op.create_index(
        "uq_invoices_tenant_lower_name",
        "invoices",
        ["tenant_id", sa.text("lower(name)")],
        unique=True,
    )
    op.create_index("ix_invoices_tenant_status", "invoices", ["tenant_id", "status"])
    op.create_index("ix_invoices_from_party", "invoices", ["from_party_id"])
    op.create_index("ix_invoices_to_party", "invoices", ["to_party_id"])

    op.create_table(
        "invoice_lines",
        _uuid_pk(),
        sa.Column(
            "invoice_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("invoices.id", ondelete="CASCADE"),
            nullable=False,
        ),
        sa.Column("kind", sa.Text(), nullable=False),
        sa.Column("description", sa.Text(), nullable=True),
        sa.Column("quantity", sa.Numeric(12, 2), nullable=True),
        sa.Column("rate", sa.Numeric(12, 2), nullable=True),
        sa.Column("amount", sa.Numeric(12, 2), nullable=True),
        sa.Column("snapshot_client_name", sa.Text(), nullable=True),
        sa.Column("snapshot_project_name", sa.Text(), nullable=True),
        sa.Column("attached_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column(
            "attached_by",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="SET NULL"),
            nullable=True,
        ),
        _created_at(),
        sa.CheckConstraint(
            "kind IN ('time', 'expense', 'adjustment', 'manual')",
            name="ck_invoice_lines_kind",
        ),
    )
    op.create_index("ix_invoice_lines_invoice", "invoice_lines", ["invoice_id"])

    # --- time entries (02 §3) --------------------------------------------
    op.create_table(
        "time_entries",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column(
            "project_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("projects.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column(
            "worker_party_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("parties.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column(
            "via_engagement_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("engagements.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column("source", sa.Text(), nullable=False),
        sa.Column("source_timezone", sa.Text(), nullable=True),
        sa.Column(
            "import_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("import_batches.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column("source_user_name", sa.Text(), nullable=True),
        sa.Column("source_user_email", sa.Text(), nullable=True),
        sa.Column("raw_operator", sa.Text(), nullable=True),
        sa.Column("raw_client", sa.Text(), nullable=True),
        sa.Column("raw_project", sa.Text(), nullable=True),
        sa.Column("description", sa.Text(), nullable=True),
        sa.Column("billable", sa.Boolean(), nullable=True),
        # deliberately timezone-naive (07 #17)
        sa.Column("start_at", sa.DateTime(timezone=False), nullable=False),
        sa.Column("end_at", sa.DateTime(timezone=False), nullable=True),
        sa.Column("duration_seconds", sa.Integer(), nullable=True),
        sa.Column("source_rate", sa.Numeric(12, 4), nullable=True),
        sa.Column("source_amount", sa.Numeric(12, 4), nullable=True),
        sa.Column("cost_amount", sa.Numeric(12, 2), nullable=True),
        sa.Column("cost_currency", sa.CHAR(3), nullable=True),
        sa.Column("billout_amount", sa.Numeric(12, 2), nullable=True),
        sa.Column("billout_currency", sa.CHAR(3), nullable=True),
        sa.Column(
            "invoice_line_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("invoice_lines.id", ondelete="SET NULL"),
            nullable=True,
        ),
        _created_at(),
        sa.CheckConstraint(
            "source IN ('clockify', 'toggl', 'manual', 'backfill')",
            name="ck_time_entries_source",
        ),
        sa.CheckConstraint(
            "(cost_amount IS NULL) = (cost_currency IS NULL)",
            name="ck_time_entries_cost_currency_pair",
        ),
        sa.CheckConstraint(
            "(billout_amount IS NULL) = (billout_currency IS NULL)",
            name="ck_time_entries_billout_currency_pair",
        ),
    )
    op.create_index(
        "ix_time_entries_tenant_start", "time_entries", ["tenant_id", "start_at"]
    )
    op.create_index(
        "ix_time_entries_tenant_project_start",
        "time_entries",
        ["tenant_id", "project_id", "start_at"],
    )
    op.create_index(
        "ix_time_entries_tenant_source_email",
        "time_entries",
        ["tenant_id", "source_user_email"],
    )
    op.create_index("ix_time_entries_project", "time_entries", ["project_id"])
    op.create_index("ix_time_entries_worker_party", "time_entries", ["worker_party_id"])
    op.create_index("ix_time_entries_import", "time_entries", ["import_id"])
    op.create_index(
        "ix_time_entries_invoice_line",
        "time_entries",
        ["invoice_line_id"],
        postgresql_where=sa.text("invoice_line_id IS NOT NULL"),
    )

    # --- CC expense pass-through (02 §3) ----------------------------------
    op.create_table(
        "cc_expense_batches",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column("name", sa.Text(), nullable=False),
        sa.Column("currency", sa.CHAR(3), nullable=False, server_default=sa.text("'CAD'")),
        sa.Column(
            "created_by",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="SET NULL"),
            nullable=True,
        ),
        _created_at(),
    )
    op.create_index("ix_cc_expense_batches_tenant", "cc_expense_batches", ["tenant_id"])

    op.create_table(
        "cc_expense_lines",
        _uuid_pk(),
        sa.Column(
            "batch_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("cc_expense_batches.id", ondelete="CASCADE"),
            nullable=False,
        ),
        sa.Column("line_index", sa.Integer(), nullable=False),
        sa.Column("raw_line", sa.Text(), nullable=False),
        sa.Column("expense_date", sa.Date(), nullable=True),
        sa.Column("description", sa.Text(), nullable=True),
        sa.Column("amount", sa.Numeric(12, 2), nullable=True),
        sa.Column("balance", sa.Numeric(12, 2), nullable=True),
        sa.Column(
            "project_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("projects.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column("assignment_description", sa.Text(), nullable=True),
        sa.Column("category", sa.Text(), nullable=True),
        sa.Column(
            "auto_assigned", sa.Boolean(), nullable=False, server_default=sa.text("false")
        ),
        sa.Column(
            "invoice_line_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("invoice_lines.id", ondelete="SET NULL"),
            nullable=True,
        ),
        _created_at(),
        # rebuild delta #8: uniqueness the legacy table lacked
        sa.UniqueConstraint("batch_id", "line_index", name="uq_cc_expense_lines_batch_index"),
    )
    op.create_index(
        "ix_cc_expense_lines_project",
        "cc_expense_lines",
        ["project_id"],
        postgresql_where=sa.text("project_id IS NOT NULL"),
    )

    op.create_table(
        "cc_highlight_keywords",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column("keyword", sa.Text(), nullable=False),
        _created_at(),
    )
    op.create_index(
        "uq_cc_highlight_keywords_tenant_kw",
        "cc_highlight_keywords",
        ["tenant_id", sa.text("lower(keyword)")],
        unique=True,
    )

    op.create_table(
        "cc_auto_rules",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column("keyword", sa.Text(), nullable=False),
        sa.Column(
            "project_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("projects.id", ondelete="CASCADE"),
            nullable=False,
        ),
        sa.Column("assignment_description", sa.Text(), nullable=True),
        sa.Column("category", sa.Text(), nullable=True),
        _created_at(),
    )
    op.create_index(
        "uq_cc_auto_rules_tenant_kw",
        "cc_auto_rules",
        ["tenant_id", sa.text("lower(keyword)")],
        unique=True,
    )
    op.create_index("ix_cc_auto_rules_project", "cc_auto_rules", ["project_id"])

    # --- comments (02 §3) --------------------------------------------------
    op.create_table(
        "comments",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=False,
        ),
        sa.Column(
            "author_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column("body", sa.Text(), nullable=False),
        sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column(
            "resolved_by",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column("entity_type", sa.Text(), nullable=True),
        sa.Column("entity_id", postgresql.UUID(as_uuid=True), nullable=True),
        _created_at(),
        sa.CheckConstraint("length(trim(body)) > 0", name="ck_comments_body_nonempty"),
        sa.CheckConstraint(
            "(entity_type IS NULL) = (entity_id IS NULL)", name="ck_comments_anchor_pair"
        ),
    )
    op.create_index("ix_comments_tenant_created", "comments", ["tenant_id", "created_at"])
    op.create_index(
        "ix_comments_tenant_unresolved",
        "comments",
        ["tenant_id", "created_at"],
        postgresql_where=sa.text("resolved_at IS NULL"),
    )

    # --- audit/undo (04-architecture) ---------------------------------------
    op.create_table(
        "audit_log",
        _uuid_pk(),
        sa.Column(
            "tenant_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("tenants.id"),
            nullable=True,
        ),
        sa.Column(
            "actor_user_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column(
            "impersonated_by",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="SET NULL"),
            nullable=True,
        ),
        sa.Column("action", sa.Text(), nullable=False),
        sa.Column("entity_type", sa.Text(), nullable=True),
        sa.Column("entity_id", postgresql.UUID(as_uuid=True), nullable=True),
        sa.Column("before", postgresql.JSONB(), nullable=True),
        sa.Column("after", postgresql.JSONB(), nullable=True),
        _created_at(),
    )
    op.create_index("ix_audit_log_tenant_created", "audit_log", ["tenant_id", "created_at"])
    op.create_index("ix_audit_log_entity", "audit_log", ["entity_type", "entity_id"])

    # --- migration run reports (05 step 2 logging) ---------------------------
    op.create_table(
        "migration_runs",
        _uuid_pk(),
        sa.Column(
            "started_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("now()"),
        ),
        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("ok", sa.Boolean(), nullable=False, server_default=sa.text("false")),
        sa.Column(
            "report",
            postgresql.JSONB(),
            nullable=False,
            server_default=sa.text("'{}'::jsonb"),
        ),
    )


def downgrade() -> None:
    op.drop_table("migration_runs")
    op.drop_table("audit_log")
    op.drop_table("comments")
    op.drop_table("cc_auto_rules")
    op.drop_table("cc_highlight_keywords")
    op.drop_table("cc_expense_lines")
    op.drop_table("cc_expense_batches")
    op.drop_table("time_entries")
    op.drop_table("invoice_lines")
    op.drop_table("invoices")
    op.drop_table("import_batches")
    op.drop_table("rate_overrides")
    op.drop_table("compensation_terms")
    op.drop_table("projects")
    op.drop_table("engagements")
    op.drop_constraint("uq_users_person_party_id", "users")
    op.drop_constraint("fk_users_person_party_id", "users", type_="foreignkey")
    op.drop_table("tenants")
    op.drop_table("parties")