"""多厂商使用信号:`asset_seat_usage_signals` + 每日采集 cron(访问对象闭环 PR4)。

`software_seat.idle_reclaim` 此前只能回收 **Atlassian** 的闲置坐席 —— 因为 Atlassian
是唯一一个 Helix 会读活跃度的厂商(`integrations/atlassian_activity.py`)。Google
Workspace 与 Zoom 的付费坐席只能靠人注意到。

这张表是那道缝:一份**持久化**的 `(人 × 厂商 × 产品) → 最后活跃`,由各厂商采集器写、
由回收作业读,于是「加一个厂商」= 加一个采集器,而不是重写状态机。

**`state` 才是承重列,不是 `last_active_at`**:
* `active` —— 有真实时间戳
* `never` —— 厂商**明确**说从未登录(可回收,但闲置时钟从**发放坐席**那天起算,
  不是从一次没发生过的登录起算)
* `unknown` —— 厂商答了,但没给出可用的值(字段空、解析不了、名册里查无此人)
* `unsupported` —— 该厂商**根本答不了**(在 `NO_USAGE_SIGNAL` 里逐条写明理由)

`unknown` / `unsupported` 绝不能读成「闲置」。同一个错误这个仓库犯过两次:身份同步
零信任对账首跑 48 条里 36 条误报,就是把「快照作证不了」当成「账号还开着」;
2026-06-11 之前的闲置坐席尝试把 sandbox 冻结的时间戳当真,把日活用户判成两年没用。

`collected_at` 是新鲜度闸的依据:超期的行一律降级回 `unknown` —— 所以采集器坏掉会让
回收作业**停止回收**,而不是拿着过期证据开始回收。对一个会撤权的作业,只能往这个方向失败。
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for, table_kw_for

revision = "612"
down_revision = "611"
branch_labels = None
depends_on = None

TABLE = "asset_seat_usage_signals"
JOBS = "scheduled_jobs"
JOB_KEY = "software_seat.usage_signal_refresh"

_JOB_DESC = (
    "每日 02:40 UTC(排在 software_seat.idle_reclaim 之前)。对每个启用且绑定了活跃 "
    "software_connectors 的集成,若其 provider 有采集器(jira / google_workspace / zoom),"
    "读一次厂商名册,按 work_email 匹配到人,写入 asset_seat_usage_signals。"
    "四态:active(有时间戳)/ never(厂商明说从未登录,闲置时钟改从发放日起算)/ "
    "unknown(厂商没给可用值)/ unsupported(该厂商答不了,见 NO_USAGE_SIGNAL)。"
    "读不到的厂商**保留原有行不覆盖**(过期的真证据好过新鲜的假话,新鲜度闸自会退役它),"
    "失败按 provider 收集后统一抛出。"
)


def _has_table(bind, name: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.tables "
            "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1",
        ),
        {"t": name},
    ).scalar())


def upgrade() -> None:
    bind = op.get_bind()
    # ``sa.Uuid()`` renders CHAR(32) on MySQL. The ORM's ``Uuid`` is not
    # that type — ``app/domains/asset/models.py`` imports it as
    # ``from app.core.sql_types import GUID as Uuid``, i.e. CHAR(36) —
    # so a migration that reaches for the real ``sa.Uuid`` builds a
    # column the models can never match and an FK the parent rejects
    # outright (errno 3780, ``identity_persons.id`` being CHAR(36)).
    # ``id_type_for`` reads the parent and pins its collation too.
    pid = id_type_for("identity_persons")
    kw = table_kw_for("identity_persons")

    if not _has_table(bind, TABLE):
        op.create_table(
            TABLE,
            sa.Column("id", pid, nullable=False),
            sa.Column("person_id", pid, nullable=False),
            sa.Column("provider", sa.String(length=64), nullable=False),
            sa.Column("product_key", sa.String(length=128), nullable=False),
            sa.Column("state", sa.String(length=16), nullable=False),
            sa.Column("last_active_at", sa.DateTime(timezone=True), nullable=True),
            sa.Column("evidence", sa.JSON(), nullable=True),
            sa.Column("collected_at", sa.DateTime(timezone=True), nullable=False),
            sa.Column(
                "created_at", sa.DateTime(timezone=True),
                server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False,
            ),
            sa.Column(
                "updated_at", sa.DateTime(timezone=True),
                server_default=sa.text("CURRENT_TIMESTAMP"),
                server_onupdate=sa.text("CURRENT_TIMESTAMP"), nullable=False,
            ),
            sa.ForeignKeyConstraint(["person_id"], ["identity_persons.id"]),
            sa.PrimaryKeyConstraint("id"),
            sa.UniqueConstraint(
                "person_id", "provider", "product_key",
                name="uq_seat_usage_signal_person_provider_product",
            ),
            **kw,
        )
        op.create_index(
            "ix_asset_seat_usage_signals_person_id", TABLE, ["person_id"],
        )
        op.create_index(
            "ix_seat_usage_signal_provider_product", TABLE,
            ["provider", "product_key"],
        )
        op.create_index(
            "ix_seat_usage_signal_collected_at", TABLE, ["collected_at"],
        )

    if _has_table(bind, JOBS):
        bind.execute(
            sa.text(
                f"INSERT INTO {JOBS} "
                "(id, name, display_name, description, handler_key, "
                " trigger_type, cron_expr, timezone, enabled, config, "
                " run_count, failure_count, created_at, updated_at) "
                "SELECT :id, :name, :display, :desc, :handler, 'cron', "
                ":cron, 'UTC', 1, :config, 0, 0, NOW(), NOW() "
                "FROM DUAL WHERE NOT EXISTS "
                f"(SELECT 1 FROM {JOBS} WHERE handler_key = :handler)",
            ),
            {
                "id": str(uuid.uuid4()),
                "name": JOB_KEY,
                "display": "坐席使用信号 · 多厂商采集",
                "desc": _JOB_DESC,
                "handler": JOB_KEY,
                "cron": "40 2 * * *",
                "config": "{}",
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _has_table(bind, JOBS):
        bind.execute(
            sa.text(f"DELETE FROM {JOBS} WHERE handler_key = :h"), {"h": JOB_KEY},
        )
    if _has_table(bind, TABLE):
        op.drop_table(TABLE)