"""What the researcher actually attached.

The write-up import read text out of the MIME tree and walked past every
attachment part. Measured against the real mailbox on 2026-08-26: **319
files, 143.1 MiB, across 54 of 149 reports** — 266 images, 25 text logs,
19 PDFs/archives and **9 PoC screen recordings**. Nothing was destroyed
(every byte is still in Gmail) but Helix held no record that any of it
existed, so an owner had no way to know there were 22 screenshots for
BB-2026-067 or a recording for BB-2026-133. For a good share of these
reports the screenshot IS the evidence and the prose is its caption.

One table. Bytes go to S3 through ``core.storage`` — the bucket and
region already sit in ``platform_settings`` and the auditor domain's
evidence files already travel that path, so there is no new
infrastructure here.

``gmail_message_id`` + ``gmail_attachment_id`` keep provenance: an
attachment id is only resolvable against the message it hung off, so
storing the pair is what makes a later re-fetch possible without
guessing. ``sha256`` is the dedupe key with ``report_id`` — researchers
resend threads and the same PNG arrives more than once.

FK: ``report_id`` → ``bounty_reports.id`` ON DELETE CASCADE, matching the
events table. That is the only FK, so one charset (taken from the parent)
satisfies TiDB's requirement that both sides agree.

Revision ID: 555
Revises: 554
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for, table_kw_for

revision = "555"
down_revision = "554"
branch_labels = None
depends_on = None

TABLE = "bounty_report_attachments"
PARENT = "bounty_reports"


def _table_exists(bind, name: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.tables "
            "WHERE table_schema = DATABASE() AND table_name = :n",
        ),
        {"n": name},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, TABLE):
        return
    if not _table_exists(bind, PARENT):
        # Nothing to hang off. Guarded rather than assumed so a partial
        # environment fails visibly at its own step instead of here.
        return

    # Charset/collation taken from the FK parent: TiDB rejects a foreign
    # key whose two sides disagree, and the parent is the side we cannot
    # change.
    pid = id_type_for(PARENT)
    tkw = table_kw_for(PARENT)

    op.create_table(
        TABLE,
        sa.Column("id", pid, nullable=False),
        sa.Column("report_id", pid, nullable=False),
        # FULL object key including the prefix active at upload time.
        # Never a URL — a presigned GET is minted per download, so a URL
        # stored here would be an expired string with an ACL baked in.
        sa.Column("object_key", sa.String(512), nullable=False),
        sa.Column("filename", sa.String(255), nullable=False),
        sa.Column("content_type", sa.String(128), nullable=False,
                  server_default="application/octet-stream"),
        sa.Column("size_bytes", sa.BigInteger),
        sa.Column("sha256", sa.String(64)),
        # A signature logo and a PoC screenshot are both image/png parts.
        # One report carries 90 files for 6 MiB, which is what a gallery
        # looks like when nobody separates them. Inline parts are kept —
        # sometimes the screenshot IS pasted inline — but collapse by
        # default in the UI.
        sa.Column("is_inline", sa.Boolean, nullable=False,
                  server_default="0"),
        sa.Column("source", sa.String(24), nullable=False,
                  server_default="researcher_mail"),
        sa.Column("gmail_message_id", sa.String(128)),
        # Gmail attachment ids run well past 255 characters.
        sa.Column("gmail_attachment_id", sa.Text),
        sa.Column("imported_at", sa.DateTime(timezone=True), nullable=False,
                  server_default=sa.func.now()),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False,
                  server_default=sa.func.now()),
        sa.PrimaryKeyConstraint("id"),
        sa.ForeignKeyConstraint(["report_id"], [f"{PARENT}.id"],
                                ondelete="CASCADE"),
        **tkw,
    )
    op.create_index("idx_bounty_attach_report", TABLE, ["report_id"])
    # Dedupe lookup, not a UNIQUE constraint: sha256 is nullable (an
    # unreadable part still deserves a row recording that it was there)
    # and MySQL treats NULLs as distinct, so UNIQUE would not hold the
    # invariant it appears to.
    op.create_index("idx_bounty_attach_dedupe", TABLE,
                    ["report_id", "sha256"])


def downgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, TABLE):
        return
    # Only the metadata rows go. The S3 objects are deliberately left in
    # place: a downgrade is a schema step, and deleting a researcher's
    # only copy of their proof-of-concept because a migration was rolled
    # back is not recoverable. Reclaiming them is a separate, deliberate
    # cleanup against ``object_key``.
    #
    # DROP TABLE only — no explicit DROP INDEX. ``idx_bounty_attach_report``
    # is the index backing the foreign key, and MySQL refuses to drop it
    # while the constraint stands (errno 1553), so the tidy-looking
    # index-then-table order fails on the first statement and leaves the
    # table behind. Caught by actually running the downgrade; the schema
    # alone reads fine either way. Dropping the table takes its indexes
    # with it.
    op.drop_table(TABLE)