JiuContinent
eb8267ed18
feat(feishu): 添加飞书入站事件inbox和混合数据库协调功能 - 实现飞书入站事件持久化inbox机制，支持状态管理、租约锁定和重试退避 - 添加混合数据库基线协调工具，确保平台PostgreSQL结构安全对齐 - 增加运行组件心跳检测和readiness就绪检查机制 - 实现app_ticket事件的安全轮换和验证处理 - 添加生产环境运行编排和fail-closed安全机制 - 支持webhook快速确认和长连接独立进程处理 - 完善个人数据擦除时的待处理事件清理功能 ```
91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
"""Fail-closed checks required before starting managed runtime processes."""
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.core.database.migrations import (
|
|
current_alembic_revisions,
|
|
expected_alembic_heads,
|
|
)
|
|
from app.core.database.safety import validate_platform_migration_target
|
|
from app.modules.feishu_users.constants import parse_admin_identities
|
|
|
|
|
|
class RuntimePreflightError(RuntimeError):
|
|
"""Raised when the configured runtime is not safe to start."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RuntimePreflightResult:
|
|
revisions: tuple[str, ...]
|
|
transport: str
|
|
|
|
|
|
def run_preflight(settings: Settings | None = None) -> RuntimePreflightResult:
|
|
"""Validate the platform target, migration head, and enabled transport."""
|
|
|
|
runtime_settings = settings or get_settings()
|
|
validate_platform_migration_target(
|
|
runtime_settings.database_url,
|
|
runtime_settings.legacy_database_url,
|
|
)
|
|
if runtime_settings.feishu_event_transport == "long_connection" and (
|
|
not runtime_settings.feishu_app_id or not runtime_settings.feishu_app_secret
|
|
):
|
|
raise RuntimePreflightError(
|
|
"FEISHU_APP_ID and FEISHU_APP_SECRET are required for long_connection"
|
|
)
|
|
if runtime_settings.feishu_admin_identities:
|
|
try:
|
|
parse_admin_identities(runtime_settings.feishu_admin_identities)
|
|
except ValueError as exc:
|
|
raise RuntimePreflightError(str(exc)) from exc
|
|
|
|
expected = expected_alembic_heads()
|
|
if len(expected) != 1:
|
|
raise RuntimePreflightError("Runtime requires exactly one Alembic head")
|
|
|
|
engine = None
|
|
try:
|
|
engine = create_engine(runtime_settings.database_url, poolclass=NullPool)
|
|
with engine.connect() as connection:
|
|
current = current_alembic_revisions(connection)
|
|
except Exception as exc:
|
|
raise RuntimePreflightError(
|
|
"Platform database schema could not be inspected"
|
|
) from exc
|
|
finally:
|
|
if engine is not None:
|
|
engine.dispose()
|
|
|
|
if current is None:
|
|
raise RuntimePreflightError(
|
|
"Platform database is not Alembic-versioned; run the approved migration first"
|
|
)
|
|
if current != expected:
|
|
raise RuntimePreflightError(
|
|
"Platform database is not at the current Alembic head"
|
|
)
|
|
return RuntimePreflightResult(
|
|
revisions=current,
|
|
transport=runtime_settings.feishu_event_transport,
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
try:
|
|
result = run_preflight()
|
|
except (RuntimeError, ValueError) as exc:
|
|
raise SystemExit(f"Runtime preflight failed: {exc}") from exc
|
|
print(
|
|
"Runtime preflight passed: "
|
|
f"schema={result.revisions[0]}, transport={result.transport}"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|