"""add feishu outbox and idempotency guards Revision ID: feishu_outbox_idempotency Revises: add_symbol_index_tables Create Date: 2026-03-05 """ from __future__ import annotations from typing import Any from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision = "feishu_outbox_idempotency" down_revision = "add_symbol_index_tables" branch_labels = None depends_on = None def _table_exists(bind, table_name: str) -> bool: inspector = sa.inspect(bind) return inspector.has_table(table_name) def _index_exists(bind, table_name: str, index_name: str) -> bool: inspector = sa.inspect(bind) indexes = inspector.get_indexes(table_name) return any(idx.get("name") == index_name for idx in indexes) def _create_index_if_not_exists( bind, *, index_name: str, table_name: str, columns: list[str], unique: bool = False, postgresql_where: Any = None, sqlite_where: Any = None, ) -> None: if _index_exists(bind, table_name, index_name): return kwargs: dict[str, Any] = {} if postgresql_where is not None: kwargs["postgresql_where"] = postgresql_where if sqlite_where is not None: kwargs["sqlite_where"] = sqlite_where op.create_index(index_name, table_name, columns, unique=unique, **kwargs) def _drop_index_if_exists(bind, index_name: str, table_name: str) -> None: if _index_exists(bind, table_name, index_name): op.drop_index(index_name, table_name=table_name) def _repoint_fk(bind, *, table_name: str, column_name: str, old_id: int, new_id: int) -> None: if not _table_exists(bind, table_name): return bind.execute( sa.text( f"UPDATE {table_name} " f"SET {column_name} = :new_id " f"WHERE {column_name} = :old_id" ), {"new_id": int(new_id), "old_id": int(old_id)}, ) def _dedupe_feishu_items(bind) -> None: if not _table_exists(bind, "feishu_items"): return groups = bind.execute( sa.text( """ SELECT tenant_id, feishu_message_id, split_from_item_id, MIN(id) AS keep_id FROM feishu_items WHERE feishu_message_id IS NOT NULL GROUP BY tenant_id, feishu_message_id, split_from_item_id HAVING COUNT(*) > 1 """ ) ).fetchall() for row in groups: tenant_id = int(row[0]) message_id = str(row[1]) split_from_item_id = row[2] keep_id = int(row[3]) if split_from_item_id is None: duplicate_rows = bind.execute( sa.text( """ SELECT id FROM feishu_items WHERE tenant_id = :tenant_id AND feishu_message_id = :message_id AND split_from_item_id IS NULL AND id <> :keep_id ORDER BY id ASC """ ), { "tenant_id": tenant_id, "message_id": message_id, "keep_id": keep_id, }, ).fetchall() else: duplicate_rows = bind.execute( sa.text( """ SELECT id FROM feishu_items WHERE tenant_id = :tenant_id AND feishu_message_id = :message_id AND split_from_item_id = :split_from_item_id AND id <> :keep_id ORDER BY id ASC """ ), { "tenant_id": tenant_id, "message_id": message_id, "split_from_item_id": int(split_from_item_id), "keep_id": keep_id, }, ).fetchall() for dup_row in duplicate_rows: duplicate_id = int(dup_row[0]) _repoint_fk( bind, table_name="feishu_items", column_name="split_from_item_id", old_id=duplicate_id, new_id=keep_id, ) _repoint_fk( bind, table_name="signals", column_name="feishu_item_id", old_id=duplicate_id, new_id=keep_id, ) _repoint_fk( bind, table_name="project_suggestions", column_name="feishu_item_id", old_id=duplicate_id, new_id=keep_id, ) _repoint_fk( bind, table_name="ai_processing_logs", column_name="feishu_item_id", old_id=duplicate_id, new_id=keep_id, ) bind.execute( sa.text("DELETE FROM feishu_items WHERE id = :id"), {"id": duplicate_id}, ) def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "feishu_reply_outbox"): op.create_table( "feishu_reply_outbox", sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), sa.Column("tenant_id", sa.BigInteger(), nullable=False, server_default="1"), sa.Column("created_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now()), sa.Column("updated_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now(), onupdate=sa.func.now()), sa.Column("trace_id", sa.String(length=64), nullable=True), sa.Column("chat_id", sa.String(length=128), nullable=False), sa.Column("message_text", sa.Text(), nullable=False), sa.Column("status", sa.String(length=20), nullable=False, server_default="pending"), sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"), sa.Column("next_retry_at", sa.TIMESTAMP(), nullable=True), sa.Column("sent_at", sa.TIMESTAMP(), nullable=True), sa.Column("last_error", sa.Text(), nullable=True), ) if _table_exists(bind, "feishu_reply_outbox"): _create_index_if_not_exists( bind, index_name="ix_feishu_reply_outbox_tenant_id", table_name="feishu_reply_outbox", columns=["tenant_id"], ) _create_index_if_not_exists( bind, index_name="ix_feishu_reply_outbox_trace_id", table_name="feishu_reply_outbox", columns=["trace_id"], ) _create_index_if_not_exists( bind, index_name="ix_feishu_reply_outbox_chat_id", table_name="feishu_reply_outbox", columns=["chat_id"], ) _create_index_if_not_exists( bind, index_name="ix_feishu_reply_outbox_status", table_name="feishu_reply_outbox", columns=["status"], ) _create_index_if_not_exists( bind, index_name="ix_feishu_reply_outbox_next_retry_at", table_name="feishu_reply_outbox", columns=["next_retry_at"], ) if _table_exists(bind, "feishu_items"): _dedupe_feishu_items(bind) where_root = sa.text("split_from_item_id IS NULL AND feishu_message_id IS NOT NULL") where_split = sa.text("split_from_item_id IS NOT NULL AND feishu_message_id IS NOT NULL") _create_index_if_not_exists( bind, index_name="uq_feishu_items_tenant_message_root", table_name="feishu_items", columns=["tenant_id", "feishu_message_id"], unique=True, postgresql_where=where_root, sqlite_where=where_root, ) _create_index_if_not_exists( bind, index_name="uq_feishu_items_tenant_message_split", table_name="feishu_items", columns=["tenant_id", "feishu_message_id", "split_from_item_id"], unique=True, postgresql_where=where_split, sqlite_where=where_split, ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "feishu_items"): _drop_index_if_exists(bind, "uq_feishu_items_tenant_message_split", "feishu_items") _drop_index_if_exists(bind, "uq_feishu_items_tenant_message_root", "feishu_items") if _table_exists(bind, "feishu_reply_outbox"): _drop_index_if_exists(bind, "ix_feishu_reply_outbox_next_retry_at", "feishu_reply_outbox") _drop_index_if_exists(bind, "ix_feishu_reply_outbox_status", "feishu_reply_outbox") _drop_index_if_exists(bind, "ix_feishu_reply_outbox_chat_id", "feishu_reply_outbox") _drop_index_if_exists(bind, "ix_feishu_reply_outbox_trace_id", "feishu_reply_outbox") _drop_index_if_exists(bind, "ix_feishu_reply_outbox_tenant_id", "feishu_reply_outbox") op.drop_table("feishu_reply_outbox")