"""飞书早就知道谁管这条流程,是我们没问。

The register (553) published 「还没人认领 —— 找 IT」 on 32 of the 49 flows
this company runs. The Feishu approval console showed a 流程管理员 for
almost every one of them. Both statements were true at once because
``GET /approval/v4/approvals/{code}`` omits ``approval_admin_ids``
unless the request carries ``?with_admin_id=true`` — the definition
sweep never sent it, read the absent key as proof the concept did not
exist, and wrote that conclusion into three docstrings where it went
unchallenged for a release.

Re-measured against this tenant on 2026-08-27 with the param: **42 of
49 definitions return at least one admin, 37 distinct people, 33 of
them resolving to an active ``identity_persons`` row.**

Two columns, and the shape of them is the whole design decision:

* ``feishu_admins`` is a **list**, not a foreign key. Seven flows name
  three or more admins (Business Card Application names seven). A
  register that collapsed that to one column would have to pick, and
  picking one of seven is inventing an owner, not reading one. The
  list is what gets published as the answer to 「找谁」.
* ``feishu_admins_synced_at`` exists separately from ``last_synced_at``
  because the old sweep DID sync these definitions — without asking
  for admins. Without its own timestamp, NULL admins would be
  indistinguishable from never-asked, which is precisely the confusion
  that produced this migration.

``owner_person_id`` is untouched here and only ever adopts an admin
when the definition names exactly one and they still work here — as
``owner_source='feishu_admin'``, which renders 待确认, not 已确认. A
declaration by the Feishu console is better evidence than a statistic
over who happens to click Approve, and still not the same thing as the
person saying yes.

Revision ID: 559
Revises: 558
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "559"
down_revision = "558"
branch_labels = None
depends_on = None

TABLE = "registry_approval_flows"
COLUMNS = ("feishu_admins", "feishu_admins_synced_at")


def _has_column(conn: sa.engine.Connection, column: str) -> bool:
    # DATABASE()-qualified: an unqualified information_schema probe
    # matches a same-named table in ANY schema, and this box runs
    # hundreds of helix_test_* schemas side by side.
    return bool(
        conn.execute(
            sa.text(
                "SELECT COUNT(*) FROM information_schema.columns "
                "WHERE table_schema = DATABASE() "
                "AND table_name = :t AND column_name = :c",
            ),
            {"t": TABLE, "c": column},
        ).scalar()
    )


def upgrade() -> None:
    conn = op.get_bind()
    if not sa.inspect(conn).has_table(TABLE):
        return
    if not _has_column(conn, "feishu_admins"):
        op.add_column(
            TABLE,
            sa.Column(
                "feishu_admins",
                sa.JSON(),
                nullable=True,
                comment=(
                    "Feishu 流程管理员 resolved to people: "
                    "[{user_id, name, email, person_id, active}]. A list "
                    "because the upstream fact is a list — collapsing it "
                    "would force the register to invent a single owner."
                ),
            ),
        )
    if not _has_column(conn, "feishu_admins_synced_at"):
        op.add_column(
            TABLE,
            sa.Column(
                "feishu_admins_synced_at",
                sa.DateTime(timezone=True),
                nullable=True,
                comment=(
                    "Last read of the admin list. Separate from "
                    "last_synced_at: earlier sweeps synced the definition "
                    "without asking for admins, so NULL here means "
                    "never-asked, not none-exist."
                ),
            ),
        )


def downgrade() -> None:
    conn = op.get_bind()
    if not sa.inspect(conn).has_table(TABLE):
        return
    # Neither column is indexed, so a plain drop is safe on TiDB too.
    for column in reversed(COLUMNS):
        if _has_column(conn, column):
            op.drop_column(TABLE, column)