JiuContinent
eb8267ed18
feat(feishu): 添加飞书入站事件inbox和混合数据库协调功能 - 实现飞书入站事件持久化inbox机制，支持状态管理、租约锁定和重试退避 - 添加混合数据库基线协调工具，确保平台PostgreSQL结构安全对齐 - 增加运行组件心跳检测和readiness就绪检查机制 - 实现app_ticket事件的安全轮换和验证处理 - 添加生产环境运行编排和fail-closed安全机制 - 支持webhook快速确认和长连接独立进程处理 - 完善个人数据擦除时的待处理事件清理功能 ```
1221 lines
40 KiB
Python
1221 lines
40 KiB
Python
import json
|
||
import tempfile
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from datetime import datetime, timedelta
|
||
from hashlib import sha256
|
||
from pathlib import Path
|
||
from threading import Event, Lock
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from alembic import command
|
||
from alembic.config import Config
|
||
from fastapi import HTTPException
|
||
from sqlalchemy import create_engine, func, select, text
|
||
from sqlalchemy.orm import Session, sessionmaker
|
||
|
||
from app.application.feishu.commands import FeishuCommandService
|
||
from app.application.feishu.events import (
|
||
FeishuEventService,
|
||
_audit_event_metadata,
|
||
_identifier_digest,
|
||
)
|
||
from app.application.feishu.personal_data import FeishuPersonalDataService
|
||
from app.application.feishu.results import command_result
|
||
from app.core.config import get_settings
|
||
from app.core.database import Base
|
||
from app.core.utils.time import utc_now
|
||
from app.modules.audit.models import AuditLog
|
||
from app.modules.feishu.client import FeishuClient
|
||
from app.modules.feishu import long_connection
|
||
from app.modules.feishu.constants import (
|
||
FeishuCommandName,
|
||
FeishuEventSource,
|
||
FeishuInboundStatus,
|
||
FeishuReplyType,
|
||
)
|
||
from app.modules.feishu.models import FeishuEventReceipt
|
||
from app.modules.feishu.services import FeishuInboundService
|
||
from app.modules.feishu_users.models import FeishuUser
|
||
from app.modules.feishu_users.identifiers import feishu_audit_identity_hash
|
||
from app.modules.feishu_users.principal import FeishuPrincipal
|
||
from app.modules.feishu_users.services import FeishuIdentityService
|
||
|
||
|
||
@pytest.fixture
|
||
def session_factory(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> sessionmaker[Session]:
|
||
monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "false")
|
||
monkeypatch.setenv("FEISHU_VERIFICATION_TOKEN", "verified-token")
|
||
get_settings.cache_clear()
|
||
database_file = tempfile.NamedTemporaryFile(
|
||
prefix="feishu-inbound-",
|
||
suffix=".db",
|
||
delete=False,
|
||
)
|
||
database_file.close()
|
||
database_path = Path(database_file.name)
|
||
engine = create_engine(
|
||
f"sqlite:///{database_path.as_posix()}",
|
||
connect_args={"check_same_thread": False, "timeout": 5},
|
||
)
|
||
FeishuEventReceipt.__table__.create(engine)
|
||
AuditLog.__table__.create(engine)
|
||
FeishuUser.__table__.create(engine)
|
||
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
||
try:
|
||
yield factory
|
||
finally:
|
||
FeishuUser.__table__.drop(engine)
|
||
AuditLog.__table__.drop(engine)
|
||
FeishuEventReceipt.__table__.drop(engine)
|
||
engine.dispose()
|
||
database_path.unlink(missing_ok=True)
|
||
get_settings.cache_clear()
|
||
|
||
|
||
@pytest.fixture
|
||
def full_session_factory(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> sessionmaker[Session]:
|
||
monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "true")
|
||
monkeypatch.setenv("FEISHU_VERIFICATION_TOKEN", "verified-token")
|
||
get_settings.cache_clear()
|
||
database_file = tempfile.NamedTemporaryFile(
|
||
prefix="feishu-inbound-erasure-",
|
||
suffix=".db",
|
||
delete=False,
|
||
)
|
||
database_file.close()
|
||
database_path = Path(database_file.name)
|
||
engine = create_engine(
|
||
f"sqlite:///{database_path.as_posix()}",
|
||
connect_args={"check_same_thread": False, "timeout": 5},
|
||
)
|
||
Base.metadata.create_all(engine)
|
||
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
||
try:
|
||
yield factory
|
||
finally:
|
||
engine.dispose()
|
||
database_path.unlink(missing_ok=True)
|
||
get_settings.cache_clear()
|
||
|
||
|
||
def _message_event(
|
||
event_id: str,
|
||
*,
|
||
text: str = "hello",
|
||
tenant_key: str = "tenant-a",
|
||
open_id: str = "ou-user",
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"schema": "2.0",
|
||
"header": {
|
||
"event_id": event_id,
|
||
"event_type": "im.message.receive_v1",
|
||
"tenant_key": tenant_key,
|
||
"token": "verified-token",
|
||
},
|
||
"event": {
|
||
"sender": {"sender_id": {"open_id": open_id}},
|
||
"message": {
|
||
"chat_id": "oc-chat",
|
||
"chat_type": "p2p",
|
||
"message_id": f"om-{event_id}",
|
||
"message_type": "text",
|
||
"content": json.dumps({"text": text}),
|
||
},
|
||
},
|
||
}
|
||
|
||
|
||
def _accept(
|
||
db: Session,
|
||
payload: dict[str, Any],
|
||
) -> str:
|
||
accepted = FeishuEventService(db).accept_verified_event(
|
||
payload,
|
||
source=FeishuEventSource.WEBHOOK,
|
||
auto_reply=False,
|
||
)
|
||
assert accepted.event_key is not None
|
||
return accepted.event_key
|
||
|
||
|
||
def _expected_identifier_digest(value: str, domain: str) -> str:
|
||
digest = sha256(
|
||
f"company-ai-platform:feishu:{domain}:v1\0{value}".encode("utf-8")
|
||
).hexdigest()
|
||
return f"sha256-{digest}"
|
||
|
||
|
||
def _create_erasure_confirmation(
|
||
db: Session,
|
||
*,
|
||
tenant_key: str = "tenant-a",
|
||
open_id: str = "ou-user",
|
||
) -> tuple[FeishuPrincipal, str]:
|
||
user = FeishuUser(
|
||
code=f"FSU-ERASURE-{open_id}",
|
||
tenant_key=tenant_key,
|
||
open_id=open_id,
|
||
)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
principal = FeishuPrincipal.from_user(
|
||
user,
|
||
chat_id="oc-chat",
|
||
chat_type="p2p",
|
||
)
|
||
confirmation = FeishuPersonalDataService(db).request_confirmation(
|
||
principal
|
||
)
|
||
return principal, confirmation.confirmation_code
|
||
|
||
|
||
def test_identifier_metadata_uses_fixed_domain_separated_digests(
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
raw_identifier = "shared-raw-identifier"
|
||
payload = _message_event(
|
||
raw_identifier,
|
||
tenant_key=raw_identifier,
|
||
open_id=raw_identifier,
|
||
)
|
||
payload["header"]["app_id"] = raw_identifier
|
||
payload["event"]["message"]["chat_id"] = raw_identifier
|
||
payload["event"]["message"]["message_id"] = raw_identifier
|
||
|
||
with session_factory() as db:
|
||
_accept(db, payload)
|
||
receipt = db.scalar(select(FeishuEventReceipt))
|
||
|
||
assert receipt is not None
|
||
metadata = _audit_event_metadata(
|
||
payload,
|
||
include_open_id=True,
|
||
include_identity_context=True,
|
||
)
|
||
digests = {
|
||
receipt.event_key,
|
||
metadata["event_id"],
|
||
metadata["message_id"],
|
||
metadata["chat_id"],
|
||
metadata["open_id"],
|
||
metadata["tenant_key"],
|
||
metadata["app_id"],
|
||
}
|
||
assert len(digests) == 7
|
||
assert all(
|
||
isinstance(value, str)
|
||
and len(value) == 71
|
||
and value.startswith("sha256-")
|
||
for value in digests
|
||
)
|
||
assert raw_identifier not in json.dumps(metadata)
|
||
|
||
|
||
def test_verified_acceptance_audit_uses_erasable_identity_subject(
|
||
full_session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
with full_session_factory() as db:
|
||
_accept(
|
||
db,
|
||
_message_event(
|
||
"erasable-audit-subject",
|
||
tenant_key="tenant-audit",
|
||
open_id="ou-audit",
|
||
),
|
||
)
|
||
record = db.execute(
|
||
select(AuditLog).where(AuditLog.action == "webhook_event")
|
||
).scalar_one()
|
||
assert record.actor == feishu_audit_identity_hash(
|
||
"tenant-audit",
|
||
"ou-audit",
|
||
)
|
||
assert "tenant-audit" not in str(record.request_payload)
|
||
assert "ou-audit" not in str(record.request_payload)
|
||
|
||
|
||
def test_inbox_migration_digests_legacy_identifiers_once(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
database_path = tmp_path / "legacy-feishu-receipts.db"
|
||
database_url = f"sqlite:///{database_path.as_posix()}"
|
||
monkeypatch.setenv("DATABASE_URL", database_url)
|
||
get_settings.cache_clear()
|
||
config = Config("alembic.ini")
|
||
engine = create_engine(database_url)
|
||
received_at = datetime(2026, 7, 27, 1, 0)
|
||
raw_event_key = "tenant-a:im.message.receive_v1:legacy-event"
|
||
existing_event_key = _expected_identifier_digest(
|
||
"tenant-a:im.message.receive_v1:already-private",
|
||
"event-key",
|
||
)
|
||
existing_event_id = _expected_identifier_digest(
|
||
"already-private",
|
||
"event-id",
|
||
)
|
||
existing_message_id = _expected_identifier_digest(
|
||
"om-already-private",
|
||
"message-id",
|
||
)
|
||
|
||
try:
|
||
command.upgrade(config, "202607270001")
|
||
with engine.begin() as connection:
|
||
connection.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO feishu_event_receipts (
|
||
event_key,
|
||
source,
|
||
event_id,
|
||
message_id,
|
||
received_at
|
||
)
|
||
VALUES (
|
||
:event_key,
|
||
'webhook',
|
||
:event_id,
|
||
:message_id,
|
||
:received_at
|
||
)
|
||
"""
|
||
),
|
||
[
|
||
{
|
||
"event_key": raw_event_key,
|
||
"event_id": "legacy-event",
|
||
"message_id": "om-legacy-event",
|
||
"received_at": received_at,
|
||
},
|
||
{
|
||
"event_key": existing_event_key,
|
||
"event_id": existing_event_id,
|
||
"message_id": existing_message_id,
|
||
"received_at": received_at,
|
||
},
|
||
],
|
||
)
|
||
|
||
command.upgrade(config, "head")
|
||
|
||
with engine.connect() as connection:
|
||
rows = list(
|
||
connection.execute(
|
||
text(
|
||
"""
|
||
SELECT event_key, event_id, message_id, status, payload
|
||
FROM feishu_event_receipts
|
||
ORDER BY id
|
||
"""
|
||
)
|
||
).mappings()
|
||
)
|
||
|
||
assert rows[0]["event_key"] == _expected_identifier_digest(
|
||
raw_event_key,
|
||
"event-key",
|
||
)
|
||
assert rows[0]["event_id"] == _expected_identifier_digest(
|
||
"legacy-event",
|
||
"event-id",
|
||
)
|
||
assert rows[0]["message_id"] == _expected_identifier_digest(
|
||
"om-legacy-event",
|
||
"message-id",
|
||
)
|
||
assert rows[1]["event_key"] == existing_event_key
|
||
assert rows[1]["event_id"] == existing_event_id
|
||
assert rows[1]["message_id"] == existing_message_id
|
||
assert all(row["status"] == FeishuInboundStatus.SUCCEEDED for row in rows)
|
||
assert all(row["payload"] is None for row in rows)
|
||
assert "legacy-event" not in json.dumps(rows, default=str)
|
||
finally:
|
||
engine.dispose()
|
||
get_settings.cache_clear()
|
||
|
||
|
||
def test_verified_intake_is_durable_sanitized_and_does_not_execute_command(
|
||
session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setattr(
|
||
FeishuCommandService,
|
||
"handle_text",
|
||
lambda *_args, **_kwargs: pytest.fail("intake must not execute commands"),
|
||
)
|
||
payload = _message_event("durable-intake", text="private question")
|
||
payload["event"]["token"] = "nested-secret"
|
||
|
||
with session_factory() as db:
|
||
accepted = FeishuEventService(db).accept_verified_event(
|
||
payload,
|
||
source=FeishuEventSource.WEBHOOK,
|
||
auto_reply=False,
|
||
)
|
||
record = db.scalar(select(FeishuEventReceipt))
|
||
|
||
assert accepted.response["accepted"] is True
|
||
assert accepted.response["handled"] is False
|
||
assert accepted.should_dispatch is True
|
||
assert record is not None
|
||
assert record.status == FeishuInboundStatus.PENDING
|
||
assert record.next_attempt_at is not None
|
||
serialized = json.dumps(record.payload)
|
||
assert "verified-token" not in serialized
|
||
assert "nested-secret" not in serialized
|
||
assert "private question" in serialized
|
||
|
||
|
||
def test_concurrent_duplicate_workers_execute_only_once(
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
with session_factory() as db:
|
||
event_key = _accept(db, _message_event("concurrent-once"))
|
||
|
||
started = Event()
|
||
release = Event()
|
||
counter_lock = Lock()
|
||
execution_count = 0
|
||
|
||
def handler(
|
||
_payload: dict[str, Any],
|
||
_source: str,
|
||
_auto_reply: bool,
|
||
_event_key: str,
|
||
) -> dict[str, Any]:
|
||
nonlocal execution_count
|
||
with counter_lock:
|
||
execution_count += 1
|
||
started.set()
|
||
assert release.wait(timeout=5)
|
||
return {"ok": True}
|
||
|
||
def process(worker_id: str) -> str:
|
||
with session_factory() as db:
|
||
return FeishuInboundService(db).process(
|
||
event_key,
|
||
handler_factory=lambda _db: handler,
|
||
worker_id=worker_id,
|
||
).record.status
|
||
|
||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||
first = executor.submit(process, "worker-one")
|
||
assert started.wait(timeout=5)
|
||
second = executor.submit(process, "worker-two")
|
||
release.set()
|
||
assert first.result(timeout=5) == FeishuInboundStatus.SUCCEEDED
|
||
assert second.result(timeout=5) == FeishuInboundStatus.SUCCEEDED
|
||
|
||
assert execution_count == 1
|
||
with session_factory() as db:
|
||
record = db.scalar(select(FeishuEventReceipt))
|
||
assert record is not None
|
||
assert record.attempt_count == 1
|
||
assert record.payload is None
|
||
|
||
|
||
def test_handler_exceeding_lease_is_not_executed_concurrently(
|
||
session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
with session_factory() as db:
|
||
event_key = _accept(db, _message_event("slow-handler-lease"))
|
||
claimed_at = utc_now()
|
||
|
||
handler_started = Event()
|
||
release_handler = Event()
|
||
replacement_claim_started = Event()
|
||
counter_lock = Lock()
|
||
execution_count = 0
|
||
original_claim = FeishuInboundService._claim
|
||
|
||
def observed_claim(
|
||
self: FeishuInboundService,
|
||
receipt_id: int,
|
||
lock_owner: str,
|
||
current: datetime,
|
||
**kwargs: Any,
|
||
) -> bool:
|
||
if lock_owner.startswith("replacement:"):
|
||
replacement_claim_started.set()
|
||
return original_claim(
|
||
self,
|
||
receipt_id,
|
||
lock_owner,
|
||
current,
|
||
**kwargs,
|
||
)
|
||
|
||
def handler(
|
||
_payload: dict[str, Any],
|
||
_source: str,
|
||
_auto_reply: bool,
|
||
_event_key: str,
|
||
) -> dict[str, bool]:
|
||
nonlocal execution_count
|
||
with counter_lock:
|
||
execution_count += 1
|
||
handler_started.set()
|
||
assert release_handler.wait(timeout=5)
|
||
return {"ok": True}
|
||
|
||
def process(worker_id: str, current: datetime) -> str:
|
||
with session_factory() as db:
|
||
return FeishuInboundService(db, lease_seconds=1).process(
|
||
event_key,
|
||
handler_factory=lambda _db: handler,
|
||
now=current,
|
||
worker_id=worker_id,
|
||
).record.status
|
||
|
||
monkeypatch.setattr(FeishuInboundService, "_claim", observed_claim)
|
||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||
original = executor.submit(process, "original", claimed_at)
|
||
assert handler_started.wait(timeout=5)
|
||
replacement = executor.submit(
|
||
process,
|
||
"replacement",
|
||
claimed_at + timedelta(seconds=2),
|
||
)
|
||
assert replacement_claim_started.wait(timeout=5)
|
||
assert execution_count == 1
|
||
release_handler.set()
|
||
|
||
assert original.result(timeout=5) == FeishuInboundStatus.SUCCEEDED
|
||
assert replacement.result(timeout=5) == FeishuInboundStatus.SUCCEEDED
|
||
|
||
assert execution_count == 1
|
||
with session_factory() as db:
|
||
record = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_key == event_key
|
||
)
|
||
)
|
||
assert record is not None
|
||
assert record.attempt_count == 1
|
||
|
||
|
||
def test_failure_retries_with_same_reply_uuid_and_redacted_error(
|
||
session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
seen_uuids: list[str | None] = []
|
||
attempt_count = 0
|
||
|
||
def flaky_handle(
|
||
self: FeishuCommandService,
|
||
*_args: Any,
|
||
**_kwargs: Any,
|
||
) -> dict[str, Any]:
|
||
nonlocal attempt_count
|
||
attempt_count += 1
|
||
seen_uuids.append(self.feishu.message_uuid)
|
||
if attempt_count == 1:
|
||
raise RuntimeError("private prompt must never enter last_error")
|
||
return {"command": "fallback_ai"}
|
||
|
||
monkeypatch.setattr(FeishuCommandService, "handle_text", flaky_handle)
|
||
with session_factory() as db:
|
||
service = FeishuEventService(db)
|
||
event_key = _accept(db, _message_event("retry-stable-uuid"))
|
||
first = service.process_inbound_event(event_key, worker_id="first")
|
||
assert first.record.status == FeishuInboundStatus.RETRY
|
||
assert first.record.last_error == "RuntimeError"
|
||
assert "private prompt" not in str(first.record.last_error)
|
||
first.record.next_attempt_at = utc_now()
|
||
db.commit()
|
||
|
||
second = service.process_inbound_event(event_key, worker_id="second")
|
||
assert second.record.status == FeishuInboundStatus.SUCCEEDED
|
||
assert second.record.payload is None
|
||
|
||
duplicate = service.process_inbound_event(event_key, worker_id="duplicate")
|
||
assert duplicate.record.status == FeishuInboundStatus.SUCCEEDED
|
||
|
||
assert attempt_count == 2
|
||
assert len(seen_uuids) == 2
|
||
assert seen_uuids[0] == seen_uuids[1]
|
||
assert seen_uuids[0] is not None
|
||
assert seen_uuids[0].startswith("inbound-")
|
||
|
||
|
||
def test_command_writes_and_receipt_success_commit_atomically(
|
||
session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
with session_factory() as db:
|
||
event_key = _accept(db, _message_event("atomic-command"))
|
||
service = FeishuInboundService(db)
|
||
original_mark_success = FeishuInboundService._mark_success
|
||
fail_before_commit = True
|
||
|
||
def mark_success(
|
||
atomic_db: Session,
|
||
key: str,
|
||
lock_owner: str,
|
||
current: datetime,
|
||
) -> None:
|
||
nonlocal fail_before_commit
|
||
if fail_before_commit:
|
||
fail_before_commit = False
|
||
raise RuntimeError("simulated crash before atomic commit")
|
||
original_mark_success(atomic_db, key, lock_owner, current)
|
||
|
||
def handler_factory(atomic_db: Session) -> Any:
|
||
def handler(*_args: Any) -> dict[str, bool]:
|
||
atomic_db.add(
|
||
AuditLog(
|
||
actor="atomic-test",
|
||
source="test",
|
||
action="atomic-side-effect",
|
||
)
|
||
)
|
||
atomic_db.commit()
|
||
return {"ok": True}
|
||
|
||
return handler
|
||
|
||
monkeypatch.setattr(
|
||
FeishuInboundService,
|
||
"_mark_success",
|
||
staticmethod(mark_success),
|
||
)
|
||
first = service.process(
|
||
event_key,
|
||
handler_factory=handler_factory,
|
||
worker_id="first",
|
||
)
|
||
assert first.record.status == FeishuInboundStatus.RETRY
|
||
assert (
|
||
db.scalar(
|
||
select(func.count())
|
||
.select_from(AuditLog)
|
||
.where(AuditLog.action == "atomic-side-effect")
|
||
)
|
||
== 0
|
||
)
|
||
|
||
first.record.next_attempt_at = utc_now()
|
||
db.commit()
|
||
second = service.process(
|
||
event_key,
|
||
handler_factory=handler_factory,
|
||
worker_id="second",
|
||
)
|
||
assert second.record.status == FeishuInboundStatus.SUCCEEDED
|
||
assert (
|
||
db.scalar(
|
||
select(func.count())
|
||
.select_from(AuditLog)
|
||
.where(AuditLog.action == "atomic-side-effect")
|
||
)
|
||
== 1
|
||
)
|
||
|
||
|
||
def test_reply_outbox_does_not_rerun_command_after_post_send_crash(
|
||
session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setenv("FEISHU_APP_ID", "cli_test_app")
|
||
monkeypatch.setenv("FEISHU_APP_SECRET", "test-secret-value")
|
||
get_settings.cache_clear()
|
||
command_runs = 0
|
||
send_attempts: list[tuple[str, str | None]] = []
|
||
|
||
def handle_text(
|
||
self: FeishuCommandService,
|
||
_text: str,
|
||
chat_id: str | None = None,
|
||
actor: str = "feishu",
|
||
auto_reply: bool = True,
|
||
**_kwargs: Any,
|
||
) -> dict[str, Any]:
|
||
nonlocal command_runs
|
||
command_runs += 1
|
||
content = f"SUB-{command_runs}"
|
||
self.db.add(
|
||
AuditLog(
|
||
actor="reply-outbox-test",
|
||
source="test",
|
||
action="reply-outbox-command",
|
||
)
|
||
)
|
||
self.db.commit()
|
||
response = (
|
||
self.feishu.send_text(
|
||
content,
|
||
receive_id=chat_id,
|
||
actor=actor,
|
||
)
|
||
if auto_reply
|
||
else None
|
||
)
|
||
return command_result(
|
||
FeishuCommandName.SUBSCRIPTION_CREATE,
|
||
FeishuReplyType.TEXT,
|
||
"订阅",
|
||
content,
|
||
response,
|
||
)
|
||
|
||
def send_text(
|
||
_self: FeishuClient,
|
||
text_value: str,
|
||
_receive_id: str | None = None,
|
||
_receive_id_type: str = "chat_id",
|
||
uuid: str | None = None,
|
||
tenant_key: str | None = None,
|
||
) -> dict[str, Any]:
|
||
del tenant_key
|
||
send_attempts.append((text_value, uuid))
|
||
return {"code": 0, "data": {"message_id": "om-reply"}}
|
||
|
||
original_mark_reply_success = FeishuInboundService._mark_reply_success
|
||
crash_after_send = True
|
||
|
||
def mark_reply_success(
|
||
atomic_db: Session,
|
||
key: str,
|
||
lock_owner: str,
|
||
current: datetime,
|
||
) -> None:
|
||
nonlocal crash_after_send
|
||
if crash_after_send:
|
||
crash_after_send = False
|
||
raise RuntimeError("simulated crash after provider accepted reply")
|
||
original_mark_reply_success(
|
||
atomic_db,
|
||
key,
|
||
lock_owner,
|
||
current,
|
||
)
|
||
|
||
monkeypatch.setattr(FeishuCommandService, "handle_text", handle_text)
|
||
monkeypatch.setattr(FeishuClient, "send_text", send_text)
|
||
monkeypatch.setattr(
|
||
FeishuInboundService,
|
||
"_mark_reply_success",
|
||
staticmethod(mark_reply_success),
|
||
)
|
||
|
||
with session_factory() as db:
|
||
accepted = FeishuEventService(db).accept_verified_event(
|
||
_message_event("reply-post-send-crash", text="订阅 每天 09:00：提醒我"),
|
||
source=FeishuEventSource.WEBHOOK,
|
||
auto_reply=True,
|
||
)
|
||
assert accepted.event_key is not None
|
||
event_key = accepted.event_key
|
||
|
||
first = FeishuEventService(db).process_inbound_event(
|
||
event_key,
|
||
worker_id="first",
|
||
)
|
||
assert first.record.status == FeishuInboundStatus.SUCCEEDED
|
||
assert first.record.reply_status == FeishuInboundStatus.RETRY
|
||
assert first.record.reply_payload is not None
|
||
assert command_runs == 1
|
||
assert len(send_attempts) == 1
|
||
assert send_attempts[0][0] == "SUB-1"
|
||
|
||
first.record.reply_next_attempt_at = utc_now()
|
||
db.commit()
|
||
FeishuInboundService(db).dispatch_reply(
|
||
event_key,
|
||
worker_id="reply-retry",
|
||
)
|
||
db.expire_all()
|
||
completed = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_key == event_key
|
||
)
|
||
)
|
||
assert completed is not None
|
||
assert completed.reply_status == FeishuInboundStatus.SUCCEEDED
|
||
assert completed.reply_payload is None
|
||
assert command_runs == 1
|
||
assert len(send_attempts) == 2
|
||
assert send_attempts[0] == send_attempts[1]
|
||
assert send_attempts[0][1] is not None
|
||
assert send_attempts[0][1].startswith("inbound-")
|
||
assert (
|
||
db.scalar(
|
||
select(func.count())
|
||
.select_from(AuditLog)
|
||
.where(AuditLog.action == "reply-outbox-command")
|
||
)
|
||
== 1
|
||
)
|
||
|
||
|
||
def test_reply_outbox_defers_image_and_card_until_command_commit(
|
||
session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setenv("FEISHU_APP_ID", "cli_test_app")
|
||
monkeypatch.setenv("FEISHU_APP_SECRET", "test-secret-value")
|
||
get_settings.cache_clear()
|
||
image_uploads: list[bytes] = []
|
||
sent_cards: list[tuple[dict[str, Any], str | None]] = []
|
||
command_runs = 0
|
||
|
||
def handle_text(
|
||
self: FeishuCommandService,
|
||
_text: str,
|
||
chat_id: str | None = None,
|
||
actor: str = "feishu",
|
||
auto_reply: bool = True,
|
||
**_kwargs: Any,
|
||
) -> dict[str, Any]:
|
||
nonlocal command_runs
|
||
command_runs += 1
|
||
response = None
|
||
if auto_reply:
|
||
image = self.feishu.upload_image(b"stable-chart", actor=actor)
|
||
image_key = (image.get("data") or {}).get("image_key")
|
||
card = self.feishu.build_basic_card(
|
||
"可靠卡片",
|
||
["同一事务提交后发送"],
|
||
image_key=image_key,
|
||
)
|
||
response = self.feishu.send_card(
|
||
card,
|
||
receive_id=chat_id,
|
||
actor=actor,
|
||
)
|
||
return command_result(
|
||
FeishuCommandName.MARKET_OVERVIEW,
|
||
FeishuReplyType.CARD,
|
||
"可靠卡片",
|
||
"同一事务提交后发送",
|
||
response,
|
||
["同一事务提交后发送"],
|
||
)
|
||
|
||
def upload_image(
|
||
_self: FeishuClient,
|
||
image: bytes,
|
||
_filename: str = "lifecycle-report.png",
|
||
tenant_key: str | None = None,
|
||
) -> dict[str, Any]:
|
||
del tenant_key
|
||
image_uploads.append(image)
|
||
return {"code": 0, "data": {"image_key": "img-stable"}}
|
||
|
||
def send_card(
|
||
_self: FeishuClient,
|
||
card: dict[str, Any],
|
||
_receive_id: str | None = None,
|
||
_receive_id_type: str = "chat_id",
|
||
uuid: str | None = None,
|
||
tenant_key: str | None = None,
|
||
) -> dict[str, Any]:
|
||
del tenant_key
|
||
sent_cards.append((card, uuid))
|
||
return {"code": 0, "data": {"message_id": "om-card"}}
|
||
|
||
original_mark_success = FeishuInboundService._mark_success
|
||
fail_before_commit = True
|
||
|
||
def mark_success(
|
||
atomic_db: Session,
|
||
key: str,
|
||
lock_owner: str,
|
||
current: datetime,
|
||
) -> None:
|
||
nonlocal fail_before_commit
|
||
if fail_before_commit:
|
||
fail_before_commit = False
|
||
raise RuntimeError("simulated crash before command commit")
|
||
original_mark_success(atomic_db, key, lock_owner, current)
|
||
|
||
monkeypatch.setattr(FeishuCommandService, "handle_text", handle_text)
|
||
monkeypatch.setattr(FeishuClient, "upload_image", upload_image)
|
||
monkeypatch.setattr(FeishuClient, "send_card", send_card)
|
||
monkeypatch.setattr(
|
||
FeishuInboundService,
|
||
"_mark_success",
|
||
staticmethod(mark_success),
|
||
)
|
||
|
||
with session_factory() as db:
|
||
accepted = FeishuEventService(db).accept_verified_event(
|
||
_message_event("reply-card-commit"),
|
||
source=FeishuEventSource.WEBHOOK,
|
||
auto_reply=True,
|
||
)
|
||
assert accepted.event_key is not None
|
||
event_key = accepted.event_key
|
||
first = FeishuEventService(db).process_inbound_event(
|
||
event_key,
|
||
worker_id="first",
|
||
)
|
||
assert first.record.status == FeishuInboundStatus.RETRY
|
||
assert image_uploads == []
|
||
assert sent_cards == []
|
||
|
||
first.record.next_attempt_at = utc_now()
|
||
db.commit()
|
||
second = FeishuEventService(db).process_inbound_event(
|
||
event_key,
|
||
worker_id="second",
|
||
)
|
||
assert second.record.status == FeishuInboundStatus.SUCCEEDED
|
||
assert second.record.reply_status == FeishuInboundStatus.SUCCEEDED
|
||
assert command_runs == 2
|
||
assert image_uploads == [b"stable-chart"]
|
||
assert len(sent_cards) == 1
|
||
assert sent_cards[0][1] is not None
|
||
assert sent_cards[0][0]["elements"][0]["img_key"] == "img-stable"
|
||
|
||
|
||
def test_expired_lease_is_reclaimed_and_terminal_failure_clears_payload(
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
current = datetime(2026, 7, 27, 2, 0)
|
||
with session_factory() as db:
|
||
event_key = _accept(db, _message_event("expired-lease"))
|
||
record = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_key == event_key
|
||
)
|
||
)
|
||
assert record is not None
|
||
record.status = FeishuInboundStatus.PROCESSING
|
||
record.attempt_count = 1
|
||
record.locked_by = "dead-worker"
|
||
record.locked_until = current - timedelta(seconds=1)
|
||
db.commit()
|
||
|
||
recovered = FeishuInboundService(db).process(
|
||
event_key,
|
||
handler_factory=lambda _db: (
|
||
lambda *_args: {"ok": True}
|
||
),
|
||
now=current,
|
||
worker_id="replacement",
|
||
)
|
||
assert recovered.record.status == FeishuInboundStatus.SUCCEEDED
|
||
assert recovered.record.attempt_count == 2
|
||
|
||
failed_key = _accept(db, _message_event("terminal-failure"))
|
||
service = FeishuInboundService(db)
|
||
pending_failure = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_key == failed_key
|
||
)
|
||
)
|
||
assert pending_failure is not None
|
||
assert pending_failure.next_attempt_at is not None
|
||
next_time = pending_failure.next_attempt_at + timedelta(microseconds=1)
|
||
for _ in range(4):
|
||
failed = service.process(
|
||
failed_key,
|
||
handler_factory=lambda _db: (
|
||
lambda *_args: (_ for _ in ()).throw(
|
||
RuntimeError("secret")
|
||
)
|
||
),
|
||
now=next_time,
|
||
)
|
||
next_time = (
|
||
failed.record.next_attempt_at + timedelta(microseconds=1)
|
||
if failed.record.next_attempt_at is not None
|
||
else next_time
|
||
)
|
||
assert failed.record.status == FeishuInboundStatus.FAILED
|
||
assert failed.record.attempt_count == 4
|
||
assert failed.record.payload is None
|
||
assert failed.record.last_error == "RuntimeError"
|
||
|
||
|
||
def test_erasure_clears_other_pending_identity_payloads(
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
with session_factory() as db:
|
||
current_key = _accept(db, _message_event("erase-current"))
|
||
other_key = _accept(db, _message_event("erase-other"))
|
||
reply_key = _accept(db, _message_event("erase-pending-reply"))
|
||
untouched_key = _accept(
|
||
db,
|
||
_message_event(
|
||
"erase-unrelated",
|
||
tenant_key="tenant-b",
|
||
open_id="ou-other",
|
||
),
|
||
)
|
||
reply_record = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_key == reply_key
|
||
)
|
||
)
|
||
assert reply_record is not None
|
||
reply_record.status = FeishuInboundStatus.SUCCEEDED
|
||
reply_record.payload = None
|
||
reply_record.reply_status = FeishuInboundStatus.PENDING
|
||
reply_record.reply_next_attempt_at = utc_now()
|
||
reply_record.reply_payload = {
|
||
"version": 1,
|
||
"identity": {
|
||
"tenant_key": "tenant-a",
|
||
"open_id": "ou-user",
|
||
},
|
||
"identity_fence_required": False,
|
||
"operations": [
|
||
{
|
||
"kind": "text",
|
||
"text": "must not be sent after erasure",
|
||
"receive_id": "oc-chat",
|
||
"receive_id_type": "chat_id",
|
||
}
|
||
],
|
||
}
|
||
db.commit()
|
||
cleared = FeishuInboundService(db).erase_identity_payloads(
|
||
tenant_key="tenant-a",
|
||
open_id="ou-user",
|
||
exclude_event_key=current_key,
|
||
)
|
||
db.commit()
|
||
|
||
records = {
|
||
item.event_key: item
|
||
for item in db.execute(select(FeishuEventReceipt)).scalars()
|
||
}
|
||
assert cleared == 2
|
||
assert records[current_key].payload is not None
|
||
assert records[other_key].status == FeishuInboundStatus.FAILED
|
||
assert records[other_key].payload is None
|
||
assert records[reply_key].reply_status == FeishuInboundStatus.FAILED
|
||
assert records[reply_key].reply_payload is None
|
||
assert records[untouched_key].payload is not None
|
||
|
||
|
||
def test_erasure_fences_already_claimed_event_before_identity_deletion(
|
||
full_session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
lock_owner = "old-worker:claimed"
|
||
with full_session_factory() as db:
|
||
_principal, confirmation_code = _create_erasure_confirmation(db)
|
||
old_payload = _message_event("old-processing-event")
|
||
old_key = _accept(db, old_payload)
|
||
deletion_key = _accept(
|
||
db,
|
||
_message_event(
|
||
"delete-identity-event",
|
||
text=f"确认忘记我 {confirmation_code}",
|
||
),
|
||
)
|
||
old_record = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_key == old_key
|
||
)
|
||
)
|
||
assert old_record is not None
|
||
old_record.status = FeishuInboundStatus.PROCESSING
|
||
old_record.attempt_count = 1
|
||
old_record.locked_by = lock_owner
|
||
old_record.locked_until = utc_now() + timedelta(minutes=5)
|
||
db.commit()
|
||
|
||
erasure_has_identity_fence = Event()
|
||
allow_erasure = Event()
|
||
old_worker_attempting = Event()
|
||
old_handler_calls = 0
|
||
original_erase = FeishuInboundService.erase_identity_payloads
|
||
|
||
def blocking_erase(
|
||
self: FeishuInboundService,
|
||
**kwargs: Any,
|
||
) -> int:
|
||
erasure_has_identity_fence.set()
|
||
assert allow_erasure.wait(timeout=5)
|
||
return original_erase(self, **kwargs)
|
||
|
||
def process_deletion() -> str:
|
||
with full_session_factory() as db:
|
||
return FeishuEventService(db).process_inbound_event(
|
||
deletion_key,
|
||
worker_id="deletion",
|
||
).record.status
|
||
|
||
def resume_claimed_event() -> int:
|
||
nonlocal old_handler_calls
|
||
old_worker_attempting.set()
|
||
|
||
def handler_factory(atomic_db: Session) -> Any:
|
||
def handler(*_args: Any) -> dict[str, bool]:
|
||
nonlocal old_handler_calls
|
||
old_handler_calls += 1
|
||
FeishuIdentityService(atomic_db).resolve_or_register(
|
||
tenant_key="tenant-a",
|
||
open_id="ou-user",
|
||
)
|
||
return {"ok": True}
|
||
|
||
return handler
|
||
|
||
with full_session_factory() as db:
|
||
service = FeishuInboundService(db)
|
||
try:
|
||
service._execute_atomically(
|
||
event_key=old_key,
|
||
lock_owner=lock_owner,
|
||
current=utc_now(),
|
||
payload=old_payload,
|
||
source=FeishuEventSource.WEBHOOK,
|
||
auto_reply=False,
|
||
handler_factory=handler_factory,
|
||
)
|
||
except HTTPException as exc:
|
||
return exc.status_code
|
||
return 200
|
||
|
||
monkeypatch.setattr(
|
||
FeishuInboundService,
|
||
"erase_identity_payloads",
|
||
blocking_erase,
|
||
)
|
||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||
deletion = executor.submit(process_deletion)
|
||
assert erasure_has_identity_fence.wait(timeout=5)
|
||
old_worker = executor.submit(resume_claimed_event)
|
||
assert old_worker_attempting.wait(timeout=5)
|
||
assert not old_worker.done()
|
||
|
||
allow_erasure.set()
|
||
|
||
assert deletion.result(timeout=5) == FeishuInboundStatus.SUCCEEDED
|
||
assert old_worker.result(timeout=5) == 409
|
||
|
||
assert old_handler_calls == 0
|
||
with full_session_factory() as db:
|
||
assert db.scalar(select(FeishuUser)) is None
|
||
old_record = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_key == old_key
|
||
)
|
||
)
|
||
deletion_record = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_key == deletion_key
|
||
)
|
||
)
|
||
assert old_record is not None
|
||
assert old_record.status == FeishuInboundStatus.FAILED
|
||
assert old_record.payload is None
|
||
assert deletion_record is not None
|
||
assert deletion_record.status == FeishuInboundStatus.SUCCEEDED
|
||
|
||
monkeypatch.setattr(
|
||
FeishuIdentityService,
|
||
"resolve_or_register",
|
||
lambda *_args, **_kwargs: pytest.fail(
|
||
"a terminal erased event must not recreate its identity"
|
||
),
|
||
)
|
||
retry = FeishuEventService(db).process_inbound_event(
|
||
old_key,
|
||
worker_id="old-retry",
|
||
)
|
||
assert retry.record.status == FeishuInboundStatus.FAILED
|
||
assert db.scalar(select(FeishuUser)) is None
|
||
|
||
|
||
def test_completed_erasure_event_duplicate_does_not_recreate_identity(
|
||
full_session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
with full_session_factory() as db:
|
||
_principal, confirmation_code = _create_erasure_confirmation(db)
|
||
payload = _message_event(
|
||
"completed-delete-event",
|
||
text=f"确认忘记我 {confirmation_code}",
|
||
)
|
||
event_key = _accept(db, payload)
|
||
first = FeishuEventService(db).process_inbound_event(
|
||
event_key,
|
||
worker_id="first-delete",
|
||
)
|
||
|
||
assert first.record.status == FeishuInboundStatus.SUCCEEDED
|
||
assert db.scalar(select(FeishuUser)) is None
|
||
|
||
monkeypatch.setattr(
|
||
FeishuIdentityService,
|
||
"resolve_or_register",
|
||
lambda *_args, **_kwargs: pytest.fail(
|
||
"a completed deletion duplicate must not recreate its identity"
|
||
),
|
||
)
|
||
duplicate = FeishuEventService(db).process_inbound_event(
|
||
event_key,
|
||
worker_id="duplicate-delete",
|
||
)
|
||
accepted_duplicate = FeishuEventService(db).accept_verified_event(
|
||
payload,
|
||
source=FeishuEventSource.WEBHOOK,
|
||
auto_reply=False,
|
||
)
|
||
|
||
assert duplicate.record.status == FeishuInboundStatus.SUCCEEDED
|
||
assert accepted_duplicate.should_dispatch is False
|
||
assert db.scalar(select(FeishuUser)) is None
|
||
|
||
|
||
def test_long_connection_dispatches_only_when_durable_queue_is_enabled(
|
||
session_factory: sessionmaker[Session],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
payload = _message_event("long-connection-inbox")
|
||
queued: list[str] = []
|
||
monkeypatch.setattr(long_connection, "_sdk_event_to_payload", lambda _event: payload)
|
||
monkeypatch.setattr(long_connection, "SessionLocal", session_factory)
|
||
monkeypatch.setattr(
|
||
long_connection,
|
||
"enqueue_feishu_inbound_event",
|
||
lambda event_key, **_kwargs: queued.append(event_key),
|
||
)
|
||
|
||
long_connection._handle_verified_sdk_event(object())
|
||
|
||
with session_factory() as db:
|
||
record = db.scalar(select(FeishuEventReceipt))
|
||
assert record is not None
|
||
assert queued == []
|
||
assert record.source == FeishuEventSource.LONG_CONNECTION
|
||
assert record.status == FeishuInboundStatus.PENDING
|
||
|
||
queued_payload = _message_event("long-connection-queued")
|
||
monkeypatch.setattr(
|
||
long_connection,
|
||
"_sdk_event_to_payload",
|
||
lambda _event: queued_payload,
|
||
)
|
||
monkeypatch.setenv("TASK_QUEUE_ENABLED", "true")
|
||
get_settings.cache_clear()
|
||
long_connection._handle_verified_sdk_event(object())
|
||
|
||
with session_factory() as db:
|
||
queued_record = db.scalar(
|
||
select(FeishuEventReceipt).where(
|
||
FeishuEventReceipt.event_id
|
||
== _identifier_digest(
|
||
"long-connection-queued",
|
||
"event-id",
|
||
)
|
||
)
|
||
)
|
||
assert queued_record is not None
|
||
assert queued == [queued_record.event_key]
|