"""Scrub legacy integration token prefixes from persisted telemetry.

Revision ID: 445
Revises: 444
Create Date: 2026-08-02

Older connection probes included the first few bearer-token characters in
their success message.  Runtime redaction prevents new writes, but existing
``integrations.last_test_message`` and matching audit entries would otherwise
remain readable through the admin APIs.
"""
from __future__ import annotations

import json
import re
from typing import Any

import sqlalchemy as sa

from alembic import op

revision = "445"
down_revision = "444"
branch_labels = None
depends_on = None

_TOKEN_PREFIX_RE = re.compile(
    r"([\"']?token_prefix[\"']?\s*[:=]\s*[\"']?)"
    r"[^\"'\s,;}\]]+",
    re.IGNORECASE,
)


def _scrub_text(value: str) -> str:
    return _TOKEN_PREFIX_RE.sub(r"\1<redacted>", value)


def _scrub_detail(value: Any) -> Any:
    """Remove prefix fields and scrub prefix fragments in nested JSON."""
    if isinstance(value, dict):
        return {
            key: _scrub_detail(item)
            for key, item in value.items()
            if not (isinstance(key, str) and key.lower() == "token_prefix")
        }
    if isinstance(value, list):
        return [_scrub_detail(item) for item in value]
    if isinstance(value, str):
        return _scrub_text(value)
    return value


def _decoded_json(value: Any) -> Any:
    if isinstance(value, (str, bytes, bytearray)):
        try:
            return json.loads(value)
        except (TypeError, ValueError):
            return value
    return value


def _scrub_existing_rows(connection: sa.Connection) -> None:
    integration_rows = connection.execute(
        sa.text(
            "SELECT id, last_test_message FROM integrations "
            "WHERE LOWER(last_test_message) LIKE '%token_prefix%'"
        )
    ).all()
    for integration_id, message in integration_rows:
        cleaned = _scrub_text(str(message))
        if cleaned != message:
            connection.execute(
                sa.text(
                    "UPDATE integrations SET last_test_message=:message "
                    "WHERE id=:integration_id"
                ),
                {"message": cleaned, "integration_id": integration_id},
            )

    audit_logs = sa.table(
        "audit_logs",
        sa.column("id"),
        sa.column("trace_id"),
        sa.column("actor_user_id"),
        sa.column("actor_type"),
        sa.column("action"),
        sa.column("resource_type"),
        sa.column("resource_id"),
        sa.column("status"),
        sa.column("detail", sa.JSON()),
        sa.column("ip_address"),
        sa.column("user_agent"),
        sa.column("created_at"),
    )
    audit_rows = connection.execute(
        sa.select(audit_logs).where(
            audit_logs.c.action == "integration.test_connection"
        )
    ).all()
    for row in audit_rows:
        values = dict(row._mapping)
        raw_detail = values["detail"]
        detail = _decoded_json(raw_detail)
        cleaned = _scrub_detail(detail)
        if cleaned != detail:
            values["detail"] = cleaned
            if connection.dialect.name == "mysql":
                # Migration 014 deliberately blocks UPDATE on audit_logs.
                # MySQL/TiDB REPLACE performs an atomic delete+insert, so the
                # row keeps its identity and all forensic metadata without
                # disabling that global protection even briefly.
                values["detail"] = json.dumps(cleaned, ensure_ascii=False)
                connection.execute(
                    sa.text(
                        "REPLACE INTO audit_logs "
                        "(id, trace_id, actor_user_id, actor_type, action, "
                        "resource_type, resource_id, status, detail, ip_address, "
                        "user_agent, created_at) VALUES "
                        "(:id, :trace_id, :actor_user_id, :actor_type, :action, "
                        ":resource_type, :resource_id, :status, :detail, "
                        ":ip_address, :user_agent, :created_at)"
                    ),
                    values,
                )
            else:
                # Keeps the pure migration regression portable to SQLite.
                connection.execute(
                    sa.delete(audit_logs).where(audit_logs.c.id == values["id"])
                )
                connection.execute(sa.insert(audit_logs).values(**values))


def upgrade() -> None:
    _scrub_existing_rows(op.get_bind())


def downgrade() -> None:
    # Redaction is intentionally irreversible: a rollback must never restore
    # bearer-token material after it has been removed.
    pass