JiuContinent
eb8267ed18
feat(feishu): 添加飞书入站事件inbox和混合数据库协调功能 - 实现飞书入站事件持久化inbox机制，支持状态管理、租约锁定和重试退避 - 添加混合数据库基线协调工具，确保平台PostgreSQL结构安全对齐 - 增加运行组件心跳检测和readiness就绪检查机制 - 实现app_ticket事件的安全轮换和验证处理 - 添加生产环境运行编排和fail-closed安全机制 - 支持webhook快速确认和长连接独立进程处理 - 完善个人数据擦除时的待处理事件清理功能 ```
53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
from socket import gethostname
|
|
from typing import Any
|
|
|
|
from app.application.feishu.events import FeishuEventService
|
|
from app.core.constants import ActorValue
|
|
from app.core.database import SessionLocal
|
|
from app.modules.feishu.constants import FEISHU_INBOUND_BATCH_SIZE
|
|
|
|
|
|
def process_feishu_inbound_event(
|
|
event_key: str,
|
|
*,
|
|
actor: str = ActorValue.WORKER,
|
|
) -> dict[str, Any]:
|
|
"""Process one durable Feishu inbox row in an isolated session."""
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
outcome = FeishuEventService(db).process_inbound_event(
|
|
event_key,
|
|
worker_id=f"{actor}:{gethostname()}",
|
|
)
|
|
return _serialize_outcome(outcome)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def process_due_feishu_inbound_events(
|
|
*,
|
|
limit: int = FEISHU_INBOUND_BATCH_SIZE,
|
|
actor: str = ActorValue.WORKER,
|
|
) -> list[dict[str, Any]]:
|
|
"""Recover due retries and expired Feishu inbox leases."""
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
outcomes = FeishuEventService(db).process_due_inbound_events(
|
|
limit=limit,
|
|
worker_id=f"{actor}:{gethostname()}",
|
|
)
|
|
return [_serialize_outcome(outcome) for outcome in outcomes]
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _serialize_outcome(outcome: Any) -> dict[str, Any]:
|
|
return {
|
|
"event_key": outcome.record.event_key,
|
|
"status": outcome.record.status,
|
|
"attempt_count": outcome.record.attempt_count,
|
|
"handled": outcome.handler_result is not None,
|
|
}
|