"""Hold what arrived on security@ until a person has looked at it.

Today every white-hat report is pasted in by hand. That is why 118 of 149
never got an answer and the oldest waited 295 days: the work only happens
when somebody opens the mailbox.

This table is the landing area for a poller. It is deliberately NOT the
report table:

* ``security@`` receives vendor pitches, internal mail and automated scan
  output as well as researchers. Auto-creating reports from all of it would
  pollute the ledger, the budget attribution and every statistic built on
  them.
* The precedent is concrete: an HRIS predicate that was slightly too loose
  created 26 accounts for system users. Nothing asked a person first.

So a poll writes rows here with the AI's reading attached, and a human
accepts or ignores each one. Accepting creates the report; ignoring keeps
the row so the same mail is never re-offered.

Dedup keys, both unique-indexed:

* ``message_id`` — the RFC822 header. One mail is offered once, no matter
  how often the poller runs or how far its watermark rewinds.
* ``thread_id``  — used to ATTACH follow-ups to an existing report instead
  of creating a second one. Researchers chase their own threads, and 118
  unanswered reports means a lot of chasing.

Revision ID: 548
Revises: 547
"""
from __future__ import annotations

import json as _json
import uuid

import sqlalchemy as sa

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

revision = "548"
down_revision = "547"
branch_labels = None
depends_on = None

TABLE = "bounty_mail_intake"
HANDLER_KEY = "bounty.mail_intake_poll"
_DESC = (
    "轮询 security@ 收件箱,把新的白帽子来信落到「待确认收件」队列,"
    "由人一键接受或忽略;同一 thread 的后续来信挂到已有报告上。"
    "需要 Workspace 给服务账号加 gmail.readonly —— 未授权前保持停用,"
    "否则失败通知每 15 分钟响一次,人会学会忽略它。"
)


def upgrade() -> None:
    conn = op.get_bind()
    if sa.inspect(conn).has_table(TABLE):
        return

    op.create_table(
        TABLE,
        sa.Column("id", id_type_for(TABLE), primary_key=True),
        # RFC822 Message-ID. Unique so one mail can only ever be offered
        # once, however often the poller runs.
        sa.Column("message_id", sa.String(512), nullable=False),
        sa.Column("thread_id", sa.String(128), nullable=True),
        sa.Column("from_email", sa.String(320), nullable=False),
        sa.Column("from_name", sa.String(255), nullable=True),
        sa.Column("subject", sa.String(500), nullable=True),
        # A snippet, not the body. The body stays in the mailbox: the FY27
        # import established that researcher correspondence does not go in
        # the database, and researchers send tax forms and bank details
        # unprompted.
        sa.Column("snippet", sa.Text(), nullable=True),
        sa.Column("received_at", sa.DateTime(timezone=True), nullable=False),
        # pending | accepted | ignored | attached
        sa.Column(
            "status", sa.String(24), nullable=False,
            server_default="pending",
        ),
        # What the AI read out of the subject/snippet. Advisory only.
        sa.Column("parsed", sa.JSON(), nullable=True),
        sa.Column(
            "report_id", id_type_for("bounty_reports"), nullable=True,
            comment="Set when accepted (new report) or attached (follow-up "
                    "on an existing thread).",
        ),
        sa.Column("decided_by", id_type_for("identity_persons"), nullable=True),
        sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("note", sa.String(1000), nullable=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"),
            onupdate=sa.text("CURRENT_TIMESTAMP"),
        ),
        **table_kw_for(TABLE),
    )
    op.create_index(
        f"uq_{TABLE}_message", TABLE, ["message_id"], unique=True,
    )
    op.create_index(f"ix_{TABLE}_status", TABLE, ["status", "received_at"])
    op.create_index(f"ix_{TABLE}_thread", TABLE, ["thread_id"])
    # No FK to bounty_reports: this row must survive a report being
    # soft-deleted, or the same mail would come back around.

    # Seed the poller, disabled. The Google service account has only
    # ``gmail.send`` today; adding ``gmail.readonly`` is a Workspace admin
    # action, and a job that fails every fifteen minutes until somebody
    # does it would train people to ignore the failure notifier.
    conn.execute(
        sa.text(
            "INSERT INTO scheduled_jobs "
            "(id, name, display_name, description, handler_key, "
            " trigger_type, cron_expr, timezone, enabled, config, "
            " run_count, failure_count, created_at, updated_at) "
            "SELECT :id, :name, :display, :desc, :handler, 'cron', "
            ":cron, 'Asia/Shanghai', 0, :config, 0, 0, NOW(), NOW() "
            "FROM DUAL WHERE NOT EXISTS "
            "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)",
        ),
        {
            "id": str(uuid.uuid4()),
            "name": HANDLER_KEY,
            "display": "白帽子来信轮询(security@)",
            "desc": _DESC,
            "handler": HANDLER_KEY,
            # Every 15 minutes: a researcher who gets an answer the same
            # working day stops chasing, and chasing is a large part of
            # what 118 unanswered reports produced.
            "cron": "*/15 * * * *",
            "config": _json.dumps({"query": "in:inbox newer_than:30d"}),
        },
    )


def downgrade() -> None:
    conn = op.get_bind()
    conn.execute(
        sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"),
        {"k": HANDLER_KEY},
    )
    if not sa.inspect(conn).has_table(TABLE):
        return
    # Index → table. No FKs on this table by design, so no 1553 dance.
    op.drop_index(f"ix_{TABLE}_thread", table_name=TABLE)
    op.drop_index(f"ix_{TABLE}_status", table_name=TABLE)
    op.drop_index(f"uq_{TABLE}_message", table_name=TABLE)
    op.drop_table(TABLE)