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]