"""Split offboarding_blocker error storage out of free-text ``notes``.

Revision ID: 148
Revises: 147
Create Date: 2026-05-12

Why:

* Pre-this-migration, ``ProcessAdvanceService._force_transfer_all``
  caught every per-blocker exception and appended a ``[T+0 自动转让
  失败] {type}: {message}[:500]`` line to ``offboarding_blockers.notes``.
  Notes ALSO carry human-authored comments (skip reason, manual
  revoke note) and the success markers ``[离职] 自动 revoke ...``
  / ``[系统强制转让] ...``. Every failed tick appended another error
  line, so a blocker that had bounced 9 times since 2026-04-21
  carried a 4 KB notes column packed with stale ``PendingRollbackError``
  text — exactly the cascade-failure mode round-13 fixed at the
  code level (per-blocker SAVEPOINT, asset_label NOT NULL),
  but the historical lines remained even after the underlying
  cause was patched.
* On the offboarding admin page this rendered as a wall of dead
  error text under every advisory blocker, making it impossible to
  tell which rows were currently failing vs which were stale.
* The fix is to give errors their own structured columns and
  treat ``notes`` as purely human/audit prose. ``advance_service``
  is rewritten in the same PR to OVERWRITE last_error_message /
  last_error_at and INCREMENT error_count on each failure, instead
  of appending to notes. Backfill below extracts the most recent
  ``[T+0 自动转让失败]`` line from each row into the new columns
  and strips ALL such lines from notes so the historical bloat is
  gone after one deploy.
* ``error_count >= 1`` is the signal the new triage_bucket logic
  uses to promote a still-pending blocker out of the
  ``auto_in_progress`` bucket into ``needs_it`` (admin must act).

The backfill is done in Python rather than via MySQL/TiDB
``REGEXP_REPLACE`` for portability — TiDB's regex group capture
parses differently across builds and the multiline-aware modes
diverge from MySQL 8.0. Python's ``re`` is identical on both
backends and the row count we touch is small (offboarding_blockers
is a few thousand rows max).

Idempotent: re-running this migration on already-processed rows is
a no-op because every error line is stripped from ``notes`` on the
first pass; the SELECT predicate (``notes LIKE '%[T+0 自动转让失败]%'``)
filters them out next time. Post-deploy, ``advance_service`` writes
failures to ``last_error_*`` directly — it no longer touches
``notes`` with the ``[T+0 自动转让失败]`` marker — so a fresh
failure between deploy and a hypothetical migration re-run would
land in the new columns only, never re-introducing the legacy
substring. Re-running this backfill is therefore safe and bounded
even under that race.
"""
from __future__ import annotations

import re

import sqlalchemy as sa

from alembic import op

# revision identifiers, used by Alembic.
revision = "148"
down_revision = "147"
branch_labels = None
depends_on = None


# Matches a single ``[T+0 自动转让失败] {type}: {message}`` line.
# The error_summary truncation in advance_service is 500 chars but
# can contain non-newline characters; we cap at 1000 here to
# leave headroom while still protecting the column from runaway
# stack-trace dumps.
_LINE_PATTERN = re.compile(r"\[T\+0 自动转让失败\][^\n]*")
# Variant that ALSO consumes an optional leading ``\n`` so stripping
# leaves no orphaned blank lines.
_STRIP_PATTERN = re.compile(r"\n?\[T\+0 自动转让失败\][^\n]*")


def upgrade() -> None:
    op.add_column(
        "offboarding_blockers",
        sa.Column("last_error_message", sa.Text(), nullable=True),
    )
    op.add_column(
        "offboarding_blockers",
        sa.Column(
            "last_error_at",
            sa.DateTime(timezone=True),
            nullable=True,
        ),
    )
    op.add_column(
        "offboarding_blockers",
        sa.Column(
            "error_count",
            sa.Integer(),
            nullable=False,
            server_default="0",
        ),
    )

    conn = op.get_bind()
    rows = conn.execute(
        sa.text(
            "SELECT id, notes, updated_at "
            "FROM offboarding_blockers "
            "WHERE notes LIKE '%[T+0 自动转让失败]%'"
        )
    ).fetchall()

    # Batch the per-row UPDATEs so a long retry storm (e.g. round-13
    # pre-fix, where every blocker accumulated 9+ lines) doesn't
    # serialize thousands of synchronous round-trips. ``executemany``
    # via SQLAlchemy core sends ~500 rows per batch to MySQL / TiDB
    # which is well under the default ``max_allowed_packet``.
    BATCH_SIZE = 500
    update_stmt = sa.text(
        "UPDATE offboarding_blockers SET "
        "last_error_message = :em, "
        "last_error_at = updated_at, "
        "error_count = :ec, "
        "notes = :n "
        "WHERE id = :id"
    )
    pending_params: list[dict[str, object]] = []
    for row in rows:
        notes_value = row.notes or ""
        matches = _LINE_PATTERN.findall(notes_value)
        if not matches:
            # Defensive: predicate matched but regex did not. Skip.
            continue
        last_err = matches[-1].strip()[:1000]
        err_count = len(matches)
        cleaned = _STRIP_PATTERN.sub("", notes_value).strip() or None
        pending_params.append({
            "em": last_err,
            "ec": err_count,
            "n": cleaned,
            "id": row.id,
        })
        if len(pending_params) >= BATCH_SIZE:
            conn.execute(update_stmt, pending_params)
            pending_params.clear()
    if pending_params:
        conn.execute(update_stmt, pending_params)


def downgrade() -> None:
    op.drop_column("offboarding_blockers", "error_count")
    op.drop_column("offboarding_blockers", "last_error_at")
    op.drop_column("offboarding_blockers", "last_error_message")