"""Initial schema - all P2HS tables with p2hs_ prefixed enums.

Revision ID: 001
Revises:
Create Date: 2025-03-03

"""
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql

revision: str = "001"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def _create_enum_if_not_exists(enum_name: str, values: str) -> None:
    """Create enum idempotently - safe to re-run after partial failure."""
    op.execute(f"""
        DO $$ BEGIN
            CREATE TYPE {enum_name} AS ENUM ({values});
        EXCEPTION WHEN duplicate_object THEN NULL;
        END $$;
    """)


def _cleanup_partial_run() -> None:
    """Drop tables and enums from a prior partial run - allows re-running 001."""
    tables = [
        "user_backup_codes", "trusted_devices", "webauthn_credentials", "user_totp_secrets",
        "job_runs", "scheduled_jobs", "email_outbox", "api_keys",
        "usage_ledger", "stripe_events", "invoices", "subscriptions", "stripe_customers",
        "project_domains", "project_deploys", "routes_manifests", "preview_environments",
        "project_builds", "codegen_runs", "revision_tags", "project_revisions",
        "prompts", "prompt_sessions", "project_instructions", "project_instruction_versions",
        "project_env_vars", "project_settings", "project_repos", "projects",
        "workspace_settings", "workspace_invites", "workspace_memberships", "roles", "workspaces",
        "audit_logs", "magic_link_tokens", "sessions", "user_identities", "users",
    ]
    for t in tables:
        op.execute(f"DROP TABLE IF EXISTS {t} CASCADE")
    enums = [
        "p2hs_subscription_status", "p2hs_deploy_status", "p2hs_deploy_environment",
        "p2hs_preview_status", "p2hs_build_status", "p2hs_codegen_run_status",
        "p2hs_project_status", "p2hs_project_type",
    ]
    for e in enums:
        op.execute(f"DROP TYPE IF EXISTS {e} CASCADE")


def upgrade() -> None:
    # Cleanup any partial prior run - allows re-running 001 from failed state
    _cleanup_partial_run()
    # Enums (p2hs_ prefix) - create idempotently for re-run after partial failure
    _create_enum_if_not_exists("p2hs_project_type", "'react_app','nuxt_app','html','shopify_theme','wordpress_theme'")
    _create_enum_if_not_exists("p2hs_project_status", "'creating','ready','error','archived'")
    _create_enum_if_not_exists("p2hs_codegen_run_status", "'pending','planning','generating','applying','completed','failed'")
    _create_enum_if_not_exists("p2hs_build_status", "'pending','running','success','failed','cancelled'")
    _create_enum_if_not_exists("p2hs_preview_status", "'provisioning','running','stopped','failed'")
    _create_enum_if_not_exists("p2hs_deploy_environment", "'staging','production'")
    _create_enum_if_not_exists("p2hs_deploy_status", "'pending','deploying','live','failed','rolled_back'")
    _create_enum_if_not_exists("p2hs_subscription_status", "'active','past_due','canceled','trialing','incomplete'")

    # 2.1 Identity and Auth
    op.create_table(
        "users",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("email", sa.String(255), unique=True, nullable=False),
        sa.Column("email_verified_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("display_name", sa.String(255)),
        sa.Column("avatar_url", sa.Text()),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "user_identities",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
        sa.Column("provider", sa.String(50), nullable=False),
        sa.Column("provider_user_id", sa.String(255)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.UniqueConstraint("provider", "provider_user_id", name="uq_user_identities_provider"),
    )
    op.create_table(
        "sessions",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
        sa.Column("refresh_token_hash", sa.String(255), nullable=False),
        sa.Column("device_info", postgresql.JSONB()),
        sa.Column("ip_address", postgresql.INET()),
        sa.Column("user_agent", sa.Text()),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("expires_at", sa.TIMESTAMP(timezone=True), nullable=False),
        sa.Column("revoked_at", sa.TIMESTAMP(timezone=True)),
    )
    op.create_table(
        "magic_link_tokens",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("email", sa.String(255), nullable=False),
        sa.Column("token_hash", sa.String(255), nullable=False),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("expires_at", sa.TIMESTAMP(timezone=True), nullable=False),
        sa.Column("used_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("ip_address", postgresql.INET()),
    )
    op.create_table(
        "audit_logs",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True)),
        sa.Column("action", sa.String(100), nullable=False),
        sa.Column("resource_type", sa.String(100)),
        sa.Column("resource_id", postgresql.UUID(as_uuid=True)),
        sa.Column("details", postgresql.JSONB()),
        sa.Column("ip_address", postgresql.INET()),
        sa.Column("user_agent", sa.Text()),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_index("idx_audit_logs_workspace", "audit_logs", ["workspace_id", "created_at"])

    # 2.2 Workspaces and RBAC
    op.create_table(
        "workspaces",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("slug", sa.String(100), unique=True, nullable=False),
        sa.Column("name", sa.String(255), nullable=False),
        sa.Column("owner_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("deleted_at", sa.TIMESTAMP(timezone=True)),
    )
    op.create_table(
        "roles",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id")),
        sa.Column("name", sa.String(100), nullable=False),
        sa.Column("permissions", postgresql.ARRAY(sa.Text()), nullable=False),
        sa.Column("is_system", sa.Boolean(), default=False),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "workspace_memberships",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False),
        sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
        sa.Column("role_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("roles.id")),
        sa.Column("invited_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("joined_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.UniqueConstraint("workspace_id", "user_id", name="uq_workspace_memberships"),
    )
    op.create_table(
        "workspace_invites",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False),
        sa.Column("email", sa.String(255), nullable=False),
        sa.Column("role_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("roles.id")),
        sa.Column("token_hash", sa.String(255), nullable=False),
        sa.Column("invited_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("expires_at", sa.TIMESTAMP(timezone=True), nullable=False),
        sa.Column("accepted_at", sa.TIMESTAMP(timezone=True)),
    )
    op.create_table(
        "workspace_settings",
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), primary_key=True),
        sa.Column("default_code_apply_policy", sa.String(20), server_default="commit"),
        sa.Column("github_installation_id", sa.BigInteger()),
        sa.Column("settings", postgresql.JSONB(), server_default="{}"),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )

    # 2.3 Projects
    op.create_table(
        "projects",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False),
        sa.Column("slug", sa.String(100), nullable=False),
        sa.Column("name", sa.String(255), nullable=False),
        sa.Column("description", sa.Text()),
        sa.Column("type", postgresql.ENUM("react_app", "nuxt_app", "html", "shopify_theme", "wordpress_theme", name="p2hs_project_type", create_type=False), nullable=False),
        sa.Column("status", postgresql.ENUM("creating", "ready", "error", "archived", name="p2hs_project_status", create_type=False), server_default="creating"),
        sa.Column("created_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("deleted_at", sa.TIMESTAMP(timezone=True)),
        sa.UniqueConstraint("workspace_id", "slug", name="uq_projects_workspace_slug"),
    )
    op.create_index("idx_projects_workspace", "projects", ["workspace_id"])
    op.create_table(
        "project_repos",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), unique=True),
        sa.Column("github_repo_id", sa.BigInteger()),
        sa.Column("repo_full_name", sa.String(255), nullable=False),
        sa.Column("default_branch", sa.String(100), server_default="main"),
        sa.Column("webhook_secret_hash", sa.String(255)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "project_settings",
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), primary_key=True),
        sa.Column("code_apply_policy", sa.String(20)),
        sa.Column("build_command", sa.Text()),
        sa.Column("output_directory", sa.String(255)),
        sa.Column("settings", postgresql.JSONB(), server_default="{}"),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "project_env_vars",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("key", sa.String(255), nullable=False),
        sa.Column("encrypted_value", sa.Text(), nullable=False),
        sa.Column("environment", sa.String(50), server_default="all"),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.UniqueConstraint("project_id", "key", "environment", name="uq_project_env_vars"),
    )

    # 2.4 Instructions and Prompts
    op.create_table(
        "project_instruction_versions",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("version_number", sa.Integer(), nullable=False),
        sa.Column("content", sa.Text(), nullable=False),
        sa.Column("created_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.UniqueConstraint("project_id", "version_number", name="uq_instruction_versions"),
    )
    op.create_table(
        "project_instructions",
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), primary_key=True),
        sa.Column("current_version_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_instruction_versions.id")),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "prompt_sessions",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("title", sa.String(255)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "prompts",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("session_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("prompt_sessions.id", ondelete="CASCADE"), nullable=False),
        sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("content", sa.Text(), nullable=False),
        sa.Column("context", postgresql.JSONB()),
        sa.Column("response", sa.Text()),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )

    # 2.5 Revisions
    op.create_table(
        "project_revisions",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("revision_number", sa.Integer(), nullable=False),
        sa.Column("git_sha", sa.String(40), nullable=False),
        sa.Column("commit_message", sa.Text()),
        sa.Column("created_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("source", sa.String(50)),
        sa.Column("source_id", postgresql.UUID(as_uuid=True)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.UniqueConstraint("project_id", "revision_number", name="uq_project_revisions"),
    )
    op.create_index("idx_revisions_project", "project_revisions", ["project_id", "revision_number"])
    op.create_table(
        "revision_tags",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("revision_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_revisions.id", ondelete="CASCADE"), nullable=False),
        sa.Column("tag", sa.String(100), nullable=False),
        sa.Column("created_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.UniqueConstraint("revision_id", "tag", name="uq_revision_tags"),
    )

    # 2.6 Codegen
    op.create_table(
        "codegen_runs",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("prompt_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("prompts.id")),
        sa.Column("status", postgresql.ENUM("pending", "planning", "generating", "applying", "completed", "failed", name="p2hs_codegen_run_status", create_type=False), server_default="pending"),
        sa.Column("plan", postgresql.JSONB()),
        sa.Column("input_revision_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_revisions.id")),
        sa.Column("output_revision_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_revisions.id")),
        sa.Column("apply_policy", sa.String(20)),
        sa.Column("pr_number", sa.Integer()),
        sa.Column("error_message", sa.Text()),
        sa.Column("ai_model", sa.String(100)),
        sa.Column("tokens_used", postgresql.JSONB()),
        sa.Column("started_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("completed_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_index("idx_codegen_runs_project", "codegen_runs", ["project_id", "created_at"])

    # 2.7 Build, Preview, Deploy
    op.create_table(
        "project_builds",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("revision_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_revisions.id")),
        sa.Column("status", postgresql.ENUM("pending", "running", "success", "failed", "cancelled", name="p2hs_build_status", create_type=False), server_default="pending"),
        sa.Column("build_command", sa.Text()),
        sa.Column("output_directory", sa.String(255)),
        sa.Column("artifact_url", sa.Text()),
        sa.Column("logs_url", sa.Text()),
        sa.Column("started_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("completed_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "preview_environments",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("revision_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_revisions.id")),
        sa.Column("build_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_builds.id")),
        sa.Column("status", postgresql.ENUM("provisioning", "running", "stopped", "failed", name="p2hs_preview_status", create_type=False), server_default="provisioning"),
        sa.Column("base_url", sa.Text()),
        sa.Column("adapter_type", sa.String(50)),
        sa.Column("expires_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("started_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("stopped_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "routes_manifests",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("revision_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_revisions.id")),
        sa.Column("routes", postgresql.JSONB(), nullable=False),
        sa.Column("generated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.UniqueConstraint("project_id", "revision_id", name="uq_routes_manifests"),
    )
    op.create_table(
        "project_deploys",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("revision_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_revisions.id")),
        sa.Column("build_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("project_builds.id")),
        sa.Column("environment", postgresql.ENUM("staging", "production", name="p2hs_deploy_environment", create_type=False), nullable=False),
        sa.Column("status", postgresql.ENUM("pending", "deploying", "live", "failed", "rolled_back", name="p2hs_deploy_status", create_type=False), server_default="pending"),
        sa.Column("deployment_url", sa.Text()),
        sa.Column("cloudflare_deployment_id", sa.String(255)),
        sa.Column("deployed_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("deployed_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("rolled_back_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("rolled_back_to", postgresql.UUID(as_uuid=True)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "project_domains",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
        sa.Column("domain", sa.String(255), nullable=False),
        sa.Column("environment", postgresql.ENUM("staging", "production", name="p2hs_deploy_environment", create_type=False)),
        sa.Column("is_primary", sa.Boolean(), server_default="false"),
        sa.Column("verification_token", sa.String(255)),
        sa.Column("verified_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("tls_provisioned_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("cloudflare_zone_id", sa.String(255)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.UniqueConstraint("project_id", "domain", name="uq_project_domains"),
    )

    # 2.8 Billing
    op.create_table(
        "stripe_customers",
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), primary_key=True),
        sa.Column("stripe_customer_id", sa.String(255), unique=True, nullable=False),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "subscriptions",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False),
        sa.Column("stripe_subscription_id", sa.String(255), unique=True, nullable=False),
        sa.Column("stripe_price_id", sa.String(255)),
        sa.Column("status", postgresql.ENUM("active", "past_due", "canceled", "trialing", "incomplete", name="p2hs_subscription_status", create_type=False), nullable=False),
        sa.Column("current_period_start", sa.TIMESTAMP(timezone=True)),
        sa.Column("current_period_end", sa.TIMESTAMP(timezone=True)),
        sa.Column("cancel_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("canceled_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "invoices",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False),
        sa.Column("stripe_invoice_id", sa.String(255), unique=True, nullable=False),
        sa.Column("stripe_subscription_id", sa.String(255)),
        sa.Column("amount_due", sa.Integer()),
        sa.Column("amount_paid", sa.Integer()),
        sa.Column("currency", sa.String(10)),
        sa.Column("status", sa.String(50)),
        sa.Column("invoice_url", sa.Text()),
        sa.Column("invoice_pdf", sa.Text()),
        sa.Column("period_start", sa.TIMESTAMP(timezone=True)),
        sa.Column("period_end", sa.TIMESTAMP(timezone=True)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "stripe_events",
        sa.Column("stripe_event_id", sa.String(255), primary_key=True),
        sa.Column("event_type", sa.String(100), nullable=False),
        sa.Column("processed_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "usage_ledger",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False),
        sa.Column("event_type", sa.String(100), nullable=False),
        sa.Column("quantity", sa.BigInteger(), nullable=False),
        sa.Column("metadata", postgresql.JSONB()),
        sa.Column("recorded_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_index("idx_usage_ledger_workspace", "usage_ledger", ["workspace_id", "recorded_at"])

    # 2.9 Public API and Jobs
    op.create_table(
        "api_keys",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False),
        sa.Column("name", sa.String(255), nullable=False),
        sa.Column("key_prefix", sa.String(10), nullable=False),
        sa.Column("key_hash", sa.String(255), nullable=False),
        sa.Column("permissions", postgresql.ARRAY(sa.Text())),
        sa.Column("last_used_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("expires_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("created_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id")),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("revoked_at", sa.TIMESTAMP(timezone=True)),
    )
    op.create_table(
        "email_outbox",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("category", sa.String(50), nullable=False),
        sa.Column("to_email", sa.String(255), nullable=False),
        sa.Column("subject", sa.String(500), nullable=False),
        sa.Column("body_html", sa.Text()),
        sa.Column("body_text", sa.Text()),
        sa.Column("metadata", postgresql.JSONB()),
        sa.Column("status", sa.String(20), server_default="pending"),
        sa.Column("attempts", sa.Integer(), server_default="0"),
        sa.Column("last_attempt_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("sent_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("error_message", sa.Text()),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "scheduled_jobs",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("name", sa.String(255), unique=True, nullable=False),
        sa.Column("cron_expression", sa.String(100), nullable=False),
        sa.Column("task_type", sa.String(100), nullable=False),
        sa.Column("task_params", postgresql.JSONB()),
        sa.Column("enabled", sa.Boolean(), server_default="true"),
        sa.Column("last_run_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("next_run_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )
    op.create_table(
        "job_runs",
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("job_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("scheduled_jobs.id")),
        sa.Column("task_type", sa.String(100), nullable=False),
        sa.Column("status", sa.String(20), nullable=False),
        sa.Column("started_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
        sa.Column("completed_at", sa.TIMESTAMP(timezone=True)),
        sa.Column("error_message", sa.Text()),
        sa.Column("result", postgresql.JSONB()),
    )


def downgrade() -> None:
    # Drop tables in reverse order
    tables = [
        "job_runs", "scheduled_jobs", "email_outbox", "api_keys",
        "usage_ledger", "stripe_events", "invoices", "subscriptions", "stripe_customers",
        "project_domains", "project_deploys", "routes_manifests", "preview_environments",
        "project_builds", "codegen_runs", "revision_tags", "project_revisions",
        "prompts", "prompt_sessions", "project_instructions", "project_instruction_versions",
        "project_env_vars", "project_settings", "project_repos", "projects",
        "workspace_settings", "workspace_invites", "workspace_memberships", "roles", "workspaces",
        "audit_logs", "magic_link_tokens", "sessions", "user_identities", "users",
    ]
    for t in tables:
        op.drop_table(t)

    # Drop enums (CASCADE for column dependencies)
    for enum_name in [
        "p2hs_subscription_status", "p2hs_deploy_status", "p2hs_deploy_environment",
        "p2hs_preview_status", "p2hs_build_status", "p2hs_codegen_run_status",
        "p2hs_project_status", "p2hs_project_type",
    ]:
        op.execute(f"DROP TYPE IF EXISTS {enum_name} CASCADE")