"""``asset_items.purchase_cost_source`` — where a purchase amount came from.

Why (2026-09-05). 资产财务完整度 read 27.6% (30 points, the worst
coverage band). Production's 760-row gap is almost entirely one event:
**758 of them are the 2026-04 易盘点 legacy import**, and the missing
fields are `purchase_cost` (718) and `currency` (718) — `purchase_date`
is missing on ZERO rows and `purchase_entity` on three. Assets Helix
created itself since May are 19 complete vs 2 incomplete, so the intake
form is fine; the gap is historical stock.

Closing it means writing amounts nobody recorded at the time, from three
sources of very different authority:

* the amount an operator actually recorded  → ``recorded`` (the default,
  and what every existing row is backfilled to);
* the median of same-model devices that DO have a price → ``estimated_model``;
* a match against the finance fixed-asset register by serial number →
  ``finance_register`` (the authoritative one).

**A single ``purchase_cost`` column cannot tell those apart**, and
depreciation, BvA and cost attribution all read it. Without this column
an estimate is indistinguishable from a recorded invoice amount the
moment it lands, which is how a convenience backfill turns into a
number finance cannot defend. The read surfaces label
``estimated_model`` explicitly and the health metric reports the
estimated subset separately, so "90% complete" can never quietly mean
"half of it is guessed".

Deliberately a plain VARCHAR with an app-level vocabulary, not a MySQL
ENUM: TiDB rejects ``MODIFY COLUMN`` on enums, so adding a fourth source
later would need a whole expand-contract cycle for no benefit.

Revision ID: 604
Revises: 603

⚠️ 603 was claimed concurrently by the access-lifecycle series
(``603_access_revocation_landing.py``, PR #3440) — that session flagged
it before either branch pushed, so this one moved up. Two branches that
both take a number land as two heads and break every backend shard.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "604"
down_revision = "603"
branch_labels = None
depends_on = None

T_ASSETS = "asset_items"
COL = "purchase_cost_source"
DEFAULT = "recorded"


def _has_table(bind, name: str) -> bool:
    """Existence guard scoped to the CURRENT database.

    ``DATABASE()`` matters: without it this matches a same-named table in
    any schema on the instance, which on a shared test server is a false
    positive that skips the DDL.
    """
    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 _has_column(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"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    if not _has_table(bind, T_ASSETS):
        return
    if _has_column(bind, T_ASSETS, COL):
        return

    # NOT NULL with a server default so every existing row lands on
    # ``recorded`` in one DDL pass. Backfilling a nullable column after
    # the fact would leave a window where a NULL means "unknown source",
    # a fourth value the vocabulary does not have.
    op.add_column(
        T_ASSETS,
        sa.Column(
            COL,
            sa.String(24),
            nullable=False,
            server_default=DEFAULT,
        ),
    )
    # Explicit re-assert: MySQL/TiDB apply the server default to existing
    # rows during ADD COLUMN, but stating it makes the invariant testable
    # and survives a future engine that does not.
    bind.execute(
        sa.text(
            f"UPDATE {T_ASSETS} SET {COL} = :d "
            f"WHERE {COL} IS NULL OR {COL} = ''"
        ),
        {"d": DEFAULT},
    )


def downgrade() -> None:
    bind = op.get_bind()
    if _has_table(bind, T_ASSETS) and _has_column(bind, T_ASSETS, COL):
        op.drop_column(T_ASSETS, COL)