"""api_key_assignments — ``transferred_from`` ownership-history column.

Revision ID: 190
Revises: 189
Create Date: 2026-05-19

Why this exists
---------------
PR2 of the API-Key lifecycle closure: ownership transfer becomes a
first-class self-service flow (employee picks a recipient → recipient
accepts via the existing :class:`AssetTransferRequest` machinery → the
``_apply_transfer`` api_key branch flips ``assigned_to``). Per
:meth:`OffboardingService._apply_transfer` the api_key branch already
rewrites ``assigned_to`` in-place, which destroys the audit trail of
prior ownership — a forensic gap when the same key changes hands twice
("who owned this last quarter?").

This column captures the *most recent* prior owner (NULL on a row that
has never been transferred). Single column, not a join table — multi-
hop transfer history lives in ``audit_logs`` under
``apikey.transfer.complete`` rows, which is the canonical source for
the full chain. The denormalised column is purely a "show the prior
owner at a glance" affordance for the admin table and the assignee's
self-service view, so the user / admin can answer "where did this key
come from?" without an audit-log drill-down.

Why a single previous-owner column rather than time-windowed temporal
table: this matches the same shape as ``revoked_by`` in migration 183
— forensic metadata is denormalised onto the row for fast scans; the
full timeline lives in ``audit_logs`` (the only place that's actually
authoritative for multi-step history). Adding a sub-table for transfer
history would duplicate AssetTransferRequest (which already records
every transfer with full source_payload).

``ON DELETE SET NULL`` so a hard-deleted person (extremely rare —
soft-delete is the norm) doesn't cascade-delete the assignment.

Idempotency
-----------
Column existence probed via ``information_schema.COLUMNS`` (same shape
as migration 183 / 181 / 178).
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "190"
down_revision = "189"
branch_labels = None
depends_on = None


def _column_exists(bind, table: str, column: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA = DATABASE() "
            "  AND TABLE_NAME = :t AND COLUMN_NAME = :c LIMIT 1"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    if not _column_exists(bind, "api_key_assignments", "transferred_from"):
        op.add_column(
            "api_key_assignments",
            sa.Column("transferred_from", sa.CHAR(36), nullable=True),
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _column_exists(bind, "api_key_assignments", "transferred_from"):
        op.drop_column("api_key_assignments", "transferred_from")