260 lines
8.9 KiB
Python
260 lines
8.9 KiB
Python
"""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")
|