"""通知卡片记下自己在飞书里的 message_id —— 让「回复这张卡」有得可查。

平台每 30 天推出 **3,785** 条通知(生产实测:盘点提醒 641、云订阅负责人半年度
确认 296、年假到期 222、证书/域名到期 125…),而同期助手收到 **98** 条消息、
历史累计只有 **3 个人**用过。两个面之间没有任何连接。

而它们其实在同一个地方:通知卡是用
``POST /im/v1/messages?receive_id_type=open_id`` 发的,用的是**和助手同一个
飞书自建应用**,落在**同一个 1:1 会话**里。所以员工在卡片下面回一句话,
``im.message.receive_v1`` **今天就已经送到助手了** —— 助手只是不知道刚刚发过
一张卡,于是「这台我早就还了」里的「这台」解析不出来。

缺的就是一个 id。飞书的发送响应里带 ``data.message_id``,而
``NotificationService._send_feishu`` 把整个返回值丢掉了;入站回复自带
``parent_id`` 指向那张卡,而 ``IncomingMessage`` 没有解析这个字段。两头都在,
中间没有落库。

这个迁移加的就是中间那一格:出站卡片的飞书 ``message_id``,建索引,因为入站
回复要按它反查 —— 一次回复一次查,走不了全表。

`provider_delivery_keys`(JSON)里存的是**发送前**自己生成的幂等键,方向相反
且不可索引,不能顶替这一列。

Revision ID: 571
Revises: 570
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "571"
down_revision = "570"
branch_labels = None
depends_on = None

TABLE = "notification_outbox"
COLUMN = "feishu_message_id"
INDEX = "ix_notification_outbox_feishu_message_id"


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 _has_index(conn: sa.engine.Connection, name: str) -> bool:
    return bool(
        conn.execute(
            sa.text(
                "SELECT COUNT(*) FROM information_schema.statistics "
                "WHERE table_schema = DATABASE() "
                "AND table_name = :t AND index_name = :i",
            ),
            {"t": TABLE, "i": name},
        ).scalar()
    )


def upgrade() -> None:
    conn = op.get_bind()
    if not sa.inspect(conn).has_table(TABLE):
        return
    if not _has_column(conn, COLUMN):
        op.add_column(
            TABLE,
            sa.Column(
                COLUMN,
                sa.String(64),
                nullable=True,
                comment=(
                    "飞书返回的这张卡的 message_id。入站回复带 parent_id "
                    "指向它,是「员工回的是哪张卡」唯一可查的线索。NULL = "
                    "非飞书渠道、发送失败、或本列上线之前发的卡。"
                ),
            ),
        )
    if not _has_index(conn, INDEX):
        # The inbound path looks a reply's parent up by this value, once
        # per reply. Without the index that is a full scan of a table
        # that grows by ~3,800 rows a month.
        op.create_index(INDEX, TABLE, [COLUMN])


def downgrade() -> None:
    conn = op.get_bind()
    if not sa.inspect(conn).has_table(TABLE):
        return
    if _has_index(conn, INDEX):
        op.drop_index(INDEX, table_name=TABLE)
    if _has_column(conn, COLUMN):
        op.drop_column(TABLE, COLUMN)