company-ai-platform/app/modules/feishu/services/inbox.py

from collections.abc import Callable
from copy import deepcopy
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any
from uuid import uuid4

from fastapi import HTTPException, status
from sqlalchemy import and_, func, or_, select, update
from sqlalchemy.engine import Connection, Engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core.http.pagination import bounded_limit
from app.core.utils.time import utc_now
from app.modules.feishu.constants import (
    FEISHU_INBOUND_BATCH_SIZE,
    FEISHU_INBOUND_LEASE_SECONDS,
    FEISHU_INBOUND_MAX_ATTEMPTS,
    FEISHU_INBOUND_RETRY_DELAYS_SECONDS,
    FeishuCommandResultKey,
    FeishuEventReceiptKey,
    FeishuInboundStatus,
    FeishuResponseKey,
)
from app.modules.feishu.errors import FeishuAPIError
from app.modules.feishu.models import FeishuEventReceipt
from app.modules.feishu.services.reply_outbox import (
    bind_reply_outbox,
    decode_image,
    identity_fence_required,
    operations_copy,
    payload_identity as reply_payload_identity,
    pending_image_indexes,
    resolved_message_operation,
)

InboundHandler = Callable[[dict[str, Any], str, bool, str], dict[str, Any]]
InboundHandlerFactory = Callable[[Session], InboundHandler]
_TERMINAL_STATUSES = frozenset(
    {
        FeishuInboundStatus.SUCCEEDED,
        FeishuInboundStatus.FAILED,
    }
)
_SENSITIVE_TRANSPORT_KEYS = frozenset(
    {
        "access_token",
        "app_access_token",
        "app_secret",
        "app_ticket",
        "authorization",
        "encrypt",
        "refresh_token",
        "secret",
        "tenant_access_token",
        "token",
    }
)


@dataclass(frozen=True, slots=True)
class FeishuInboundAcceptance:
    """Result of durably accepting one verified Feishu event."""

    record: FeishuEventReceipt
    created: bool

    @property
    def should_dispatch(self) -> bool:
        return self.record.status not in _TERMINAL_STATUSES


@dataclass(frozen=True, slots=True)
class FeishuInboundProcessResult:
    """Persistent state and the transient command result from one attempt."""

    record: FeishuEventReceipt
    handler_result: dict[str, Any] | None = None


class FeishuInboundService:
    """Persist, claim, retry, and finalize verified Feishu inbound events."""

    def __init__(
        self,
        db: Session,
        *,
        lease_seconds: int = FEISHU_INBOUND_LEASE_SECONDS,
    ) -> None:
        self.db = db
        self.lease_seconds = lease_seconds

    def accept(
        self,
        payload: dict[str, Any],
        event_identity: dict[str, str | None],
        *,
        event_type: str | None,
        auto_reply: bool,
        persist_payload: bool = True,
        now: datetime | None = None,
    ) -> FeishuInboundAcceptance:
        """Commit an inbox row before transport acknowledgement."""

        current = now or utc_now()
        event_key = str(event_identity[FeishuEventReceiptKey.EVENT_KEY])
        receipt = FeishuEventReceipt(
            event_key=event_key,
            source=str(event_identity[FeishuEventReceiptKey.SOURCE]),
            event_id=event_identity.get(FeishuEventReceiptKey.EVENT_ID),
            message_id=event_identity.get(FeishuEventReceiptKey.MESSAGE_ID),
            event_type=event_type or None,
            payload=_sanitize_payload(payload) if persist_payload else None,
            auto_reply=auto_reply,
            status=FeishuInboundStatus.PENDING,
            attempt_count=0,
            max_attempts=FEISHU_INBOUND_MAX_ATTEMPTS,
            next_attempt_at=current,
            received_at=current,
        )
        try:
            with self.db.begin_nested():
                self.db.add(receipt)
                self.db.flush()
        except IntegrityError:
            self.db.rollback()
            existing = self._get(event_key)
            return FeishuInboundAcceptance(record=existing, created=False)
        self.db.commit()
        self.db.refresh(receipt)
        return FeishuInboundAcceptance(record=receipt, created=True)

    def process(
        self,
        event_key: str,
        *,
        handler_factory: InboundHandlerFactory,
        payload_override: dict[str, Any] | None = None,
        now: datetime | None = None,
        worker_id: str = "feishu-inbound",
    ) -> FeishuInboundProcessResult:
        """Claim and execute one event, fencing concurrent duplicate workers."""

        current = now or utc_now()
        existing = self._get(event_key)
        if existing.status in _TERMINAL_STATUSES:
            return FeishuInboundProcessResult(record=existing)
        if existing.attempt_count >= existing.max_attempts:
            return FeishuInboundProcessResult(
                record=self._fail_exhausted(existing.id, current)
            )

        lock_owner = f"{worker_id}:{uuid4().hex}"
        if not self._claim(
            existing.id,
            lock_owner,
            current,
            max_attempts=existing.max_attempts,
            expected_attempt_count=existing.attempt_count,
            allow_pending_immediate=(
                payload_override is not None and existing.payload is None
            ),
        ):
            self.db.rollback()
            return FeishuInboundProcessResult(record=self._get(event_key))
        record = self._get(event_key)
        effective_payload = (
            payload_override
            if payload_override is not None
            else record.payload
        )
        if not isinstance(effective_payload, dict):
            return FeishuInboundProcessResult(
                record=self._finish_failure(
                    event_key,
                    lock_owner,
                    current,
                    ValueError("Feishu inbound payload is unavailable"),
                    retryable=False,
                )
            )

        source = str(record.source)
        auto_reply = bool(record.auto_reply)
        self.db.rollback()
        try:
            result = self._execute_atomically(
                event_key=event_key,
                lock_owner=lock_owner,
                current=current,
                payload=dict(effective_payload),
                source=source,
                auto_reply=auto_reply,
                handler_factory=handler_factory,
            )
        except Exception as exc:
            self.db.rollback()
            return FeishuInboundProcessResult(
                record=self._finish_failure(
                    event_key,
                    lock_owner,
                    current,
                    exc,
                    retryable=_is_retryable(exc),
                )
            )
        reply_result = self.dispatch_reply(
            event_key,
            now=current,
            worker_id=f"{worker_id}:reply",
        )
        _attach_provider_response(result, reply_result)
        self.db.expire_all()
        return FeishuInboundProcessResult(
            record=self._get(event_key),
            handler_result=result,
        )

    def process_due(
        self,
        *,
        handler_factory: InboundHandlerFactory,
        now: datetime | None = None,
        limit: int = FEISHU_INBOUND_BATCH_SIZE,
        worker_id: str = "feishu-inbound",
    ) -> list[FeishuInboundProcessResult]:
        """Process due pending/retry rows and reclaim expired processing leases."""

        current = now or utc_now()
        stmt = (
            select(FeishuEventReceipt.event_key)
            .where(
                _claimable(current),
                FeishuEventReceipt.payload.is_not(None),
            )
            .order_by(
                FeishuEventReceipt.next_attempt_at.asc(),
                FeishuEventReceipt.id.asc(),
            )
            .limit(bounded_limit(limit))
        )
        event_keys = list(self.db.execute(stmt).scalars())
        outcomes = [
            self.process(
                event_key,
                handler_factory=handler_factory,
                now=current,
                worker_id=worker_id,
            )
            for event_key in event_keys
        ]
        self.process_due_replies(
            now=current,
            limit=limit,
            worker_id=f"{worker_id}:reply",
        )
        return outcomes

    def process_due_replies(
        self,
        *,
        now: datetime | None = None,
        limit: int = FEISHU_INBOUND_BATCH_SIZE,
        worker_id: str = "feishu-inbound-reply",
    ) -> int:
        """Retry committed reply outboxes without re-running their commands."""

        current = now or utc_now()
        event_keys = list(
            self.db.execute(
                select(FeishuEventReceipt.event_key)
                .where(
                    FeishuEventReceipt.status == FeishuInboundStatus.SUCCEEDED,
                    FeishuEventReceipt.reply_payload.is_not(None),
                    _reply_claimable(current),
                )
                .order_by(
                    FeishuEventReceipt.reply_next_attempt_at.asc(),
                    FeishuEventReceipt.id.asc(),
                )
                .limit(bounded_limit(limit))
            ).scalars()
        )
        for event_key in event_keys:
            self.dispatch_reply(
                event_key,
                now=current,
                worker_id=worker_id,
            )
        return len(event_keys)

    def dispatch_reply(
        self,
        event_key: str,
        *,
        now: datetime | None = None,
        worker_id: str = "feishu-inbound-reply",
    ) -> dict[str, Any] | None:
        """Send one committed reply intent with a stable provider UUID."""

        current = now or utc_now()
        record = self._get(event_key)
        if record.reply_status in _TERMINAL_STATUSES or record.reply_payload is None:
            return None
        if record.reply_attempt_count >= record.max_attempts:
            self._fail_reply_exhausted(record.id, current)
            return None

        lock_owner = f"{worker_id}:{uuid4().hex}"
        if not self._claim_reply(record.id, lock_owner, current):
            self.db.rollback()
            return None
        try:
            self._prepare_reply_images(event_key, lock_owner)
            return self._send_prepared_reply(
                event_key,
                lock_owner,
                current,
            )
        except Exception as exc:
            self.db.rollback()
            self._finish_reply_failure(
                event_key,
                lock_owner,
                current,
                exc,
                retryable=_is_retryable(exc),
            )
            return None

    def erase_identity_payloads(
        self,
        *,
        tenant_key: str,
        open_id: str,
        exclude_event_key: str | None = None,
        now: datetime | None = None,
    ) -> int:
        """Clear replayable events for an identity being permanently erased."""

        current = now or utc_now()
        candidates = list(
            self.db.execute(
                select(FeishuEventReceipt).where(
                    FeishuEventReceipt.payload.is_not(None),
                    FeishuEventReceipt.status.in_(
                        [
                            FeishuInboundStatus.PENDING,
                            FeishuInboundStatus.PROCESSING,
                            FeishuInboundStatus.RETRY,
                        ]
                    ),
                )
            ).scalars()
        )
        matching_ids = [
            record.id
            for record in candidates
            if record.event_key != exclude_event_key
            and _matches_identity(record.payload, tenant_key, open_id)
        ]
        if matching_ids:
            self.db.execute(
                update(FeishuEventReceipt)
                .where(
                    FeishuEventReceipt.id.in_(matching_ids),
                    FeishuEventReceipt.status.in_(
                        [
                            FeishuInboundStatus.PENDING,
                            FeishuInboundStatus.PROCESSING,
                            FeishuInboundStatus.RETRY,
                        ]
                    ),
                )
                .values(
                    status=FeishuInboundStatus.FAILED,
                    payload=None,
                    last_error="personal_data_erased",
                    next_attempt_at=None,
                    locked_by=None,
                    locked_until=None,
                    processed_at=current,
                )
                .execution_options(synchronize_session=False)
            )

        reply_candidates = list(
            self.db.execute(
                select(FeishuEventReceipt).where(
                    FeishuEventReceipt.reply_payload.is_not(None),
                    FeishuEventReceipt.reply_status.in_(
                        [
                            FeishuInboundStatus.PENDING,
                            FeishuInboundStatus.PROCESSING,
                            FeishuInboundStatus.RETRY,
                        ]
                    ),
                )
            ).scalars()
        )
        reply_matching_ids = [
            record.id
            for record in reply_candidates
            if record.event_key != exclude_event_key
            and reply_payload_identity(record.reply_payload)
            == (tenant_key, open_id)
        ]
        if reply_matching_ids:
            self.db.execute(
                update(FeishuEventReceipt)
                .where(
                    FeishuEventReceipt.id.in_(reply_matching_ids),
                    FeishuEventReceipt.reply_status.in_(
                        [
                            FeishuInboundStatus.PENDING,
                            FeishuInboundStatus.PROCESSING,
                            FeishuInboundStatus.RETRY,
                        ]
                    ),
                )
                .values(
                    reply_status=FeishuInboundStatus.FAILED,
                    reply_payload=None,
                    reply_last_error="personal_data_erased",
                    reply_next_attempt_at=None,
                    reply_locked_by=None,
                    reply_locked_until=None,
                )
                .execution_options(synchronize_session=False)
            )
        self.db.flush()
        self.db.expire_all()
        return len(set(matching_ids) | set(reply_matching_ids))

    def status_counts(self) -> dict[str, int]:
        """Return operational inbox counts without loading event payloads."""

        tracked = (
            FeishuInboundStatus.PENDING,
            FeishuInboundStatus.RETRY,
            FeishuInboundStatus.PROCESSING,
            FeishuInboundStatus.FAILED,
        )
        rows = self.db.execute(
            select(
                FeishuEventReceipt.status,
                func.count(FeishuEventReceipt.id),
            )
            .where(FeishuEventReceipt.status.in_(tracked))
            .group_by(FeishuEventReceipt.status)
        )
        counts = {str(status_value): 0 for status_value in tracked}
        counts.update(
            {
                str(status_value): int(count)
                for status_value, count in rows
            }
        )
        reply_rows = self.db.execute(
            select(
                FeishuEventReceipt.reply_status,
                func.count(FeishuEventReceipt.id),
            )
            .where(FeishuEventReceipt.reply_status.in_(tracked))
            .group_by(FeishuEventReceipt.reply_status)
        )
        counts.update(
            {
                f"reply_{status_value}": 0
                for status_value in tracked
            }
        )
        counts.update(
            {
                f"reply_{status_value}": int(count)
                for status_value, count in reply_rows
            }
        )
        return counts

    def _execute_atomically(
        self,
        *,
        event_key: str,
        lock_owner: str,
        current: datetime,
        payload: dict[str, Any],
        source: str,
        auto_reply: bool,
        handler_factory: InboundHandlerFactory,
    ) -> dict[str, Any]:
        """Commit command writes and inbox success as one database transaction."""

        bind = self.db.get_bind()
        engine = bind.engine if isinstance(bind, Connection) else bind
        if not isinstance(engine, Engine):
            raise RuntimeError("Feishu inbound processing requires a SQLAlchemy engine")

        with engine.connect() as connection:
            transaction = connection.begin()
            try:
                if connection.dialect.name == "sqlite":
                    # SQLite has no SELECT FOR UPDATE. BEGIN IMMEDIATE provides
                    # the single-process test substitute while also ensuring a
                    # released handler SAVEPOINT cannot escape the outer rollback.
                    connection.exec_driver_sql("BEGIN IMMEDIATE")
                with Session(
                    bind=connection,
                    expire_on_commit=False,
                    join_transaction_mode="create_savepoint",
                ) as atomic_db:
                    _lock_identity_fence(atomic_db, payload)
                    receipt = atomic_db.scalar(
                        select(FeishuEventReceipt)
                        .where(FeishuEventReceipt.event_key == event_key)
                        .with_for_update()
                    )
                    if (
                        receipt is None
                        or receipt.status != FeishuInboundStatus.PROCESSING
                        or receipt.locked_by != lock_owner
                    ):
                        raise HTTPException(
                            status_code=status.HTTP_409_CONFLICT,
                            detail="Feishu inbound event lease was lost",
                        )
                    with bind_reply_outbox() as reply_outbox:
                        result = handler_factory(atomic_db)(
                            payload,
                            source,
                            auto_reply,
                            event_key,
                        )
                    identity = _payload_identity(payload)
                    reply_payload = reply_outbox.as_payload(
                        identity=identity,
                        identity_fence_required=_identity_exists(
                            atomic_db,
                            identity,
                        ),
                    )
                    if reply_payload is not None:
                        receipt.reply_payload = reply_payload
                        receipt.reply_status = FeishuInboundStatus.PENDING
                        receipt.reply_attempt_count = 0
                        receipt.reply_last_error = None
                        receipt.reply_next_attempt_at = current
                        receipt.reply_locked_by = None
                        receipt.reply_locked_until = None
                        receipt.reply_sent_at = None
                    self._mark_success(
                        atomic_db,
                        event_key,
                        lock_owner,
                        current,
                    )
                    atomic_db.commit()
                transaction.commit()
                return result
            except Exception:
                if transaction.is_active:
                    transaction.rollback()
                raise

    def _claim_reply(
        self,
        receipt_id: int,
        lock_owner: str,
        current: datetime,
    ) -> bool:
        result = self.db.execute(
            update(FeishuEventReceipt)
            .where(
                FeishuEventReceipt.id == receipt_id,
                FeishuEventReceipt.status == FeishuInboundStatus.SUCCEEDED,
                FeishuEventReceipt.reply_payload.is_not(None),
                _reply_claimable(current),
                FeishuEventReceipt.reply_attempt_count
                < FeishuEventReceipt.max_attempts,
            )
            .values(
                reply_status=FeishuInboundStatus.PROCESSING,
                reply_attempt_count=(
                    FeishuEventReceipt.reply_attempt_count + 1
                ),
                reply_locked_by=lock_owner,
                reply_locked_until=current + timedelta(seconds=self.lease_seconds),
            )
            .execution_options(synchronize_session=False)
        )
        self.db.commit()
        self.db.expire_all()
        return result.rowcount == 1

    def _prepare_reply_images(
        self,
        event_key: str,
        lock_owner: str,
    ) -> None:
        while True:
            self.db.expire_all()
            record = self._get(event_key)
            payload = record.reply_payload
            if not isinstance(payload, dict):
                raise ValueError("Feishu reply outbox payload is unavailable")
            indexes = pending_image_indexes(payload)
            self.db.rollback()
            if not indexes:
                return
            self._prepare_reply_image(
                event_key,
                lock_owner,
                indexes[0],
            )

    def _prepare_reply_image(
        self,
        event_key: str,
        lock_owner: str,
        operation_index: int,
    ) -> None:
        bind = self.db.get_bind()
        engine = bind.engine if isinstance(bind, Connection) else bind
        if not isinstance(engine, Engine):
            raise RuntimeError("Feishu reply processing requires a SQLAlchemy engine")

        with engine.connect() as connection:
            transaction = connection.begin()
            try:
                if connection.dialect.name == "sqlite":
                    connection.exec_driver_sql("BEGIN IMMEDIATE")
                with Session(
                    bind=connection,
                    expire_on_commit=False,
                    join_transaction_mode="create_savepoint",
                ) as atomic_db:
                    receipt = self._locked_reply(
                        atomic_db,
                        event_key,
                        lock_owner,
                    )
                    payload = receipt.reply_payload
                    if not isinstance(payload, dict):
                        raise ValueError(
                            "Feishu reply outbox payload is unavailable"
                        )
                    _lock_reply_identity(atomic_db, payload)
                    operations = operations_copy(payload)
                    if operation_index >= len(operations):
                        raise ValueError(
                            "Feishu reply image operation is unavailable"
                        )
                    operation = operations[operation_index]
                    if operation.get("kind") != "image":
                        raise ValueError(
                            "Feishu reply image operation is invalid"
                        )
                    if operation.get("image_key"):
                        atomic_db.commit()
                        transaction.commit()
                        return

                    from app.modules.feishu.service import FeishuService

                    result = FeishuService(atomic_db).upload_image(
                        decode_image(operation),
                        actor=str(operation.get("actor") or "feishu"),
                        tenant_key=_optional_text(
                            operation.get("tenant_key")
                        ),
                    )
                    image_key = str(
                        (result.get("data") or {}).get("image_key") or ""
                    ).strip()
                    if not image_key:
                        raise ValueError(
                            "Feishu image upload did not return image_key"
                        )
                    operation["image_key"] = image_key
                    operation.pop("image_base64", None)
                    updated_payload = deepcopy(payload)
                    updated_payload["operations"] = operations
                    receipt.reply_payload = updated_payload
                    atomic_db.commit()
                transaction.commit()
            except Exception:
                if transaction.is_active:
                    transaction.rollback()
                raise

    def _send_prepared_reply(
        self,
        event_key: str,
        lock_owner: str,
        current: datetime,
    ) -> dict[str, Any]:
        bind = self.db.get_bind()
        engine = bind.engine if isinstance(bind, Connection) else bind
        if not isinstance(engine, Engine):
            raise RuntimeError("Feishu reply processing requires a SQLAlchemy engine")

        with engine.connect() as connection:
            transaction = connection.begin()
            try:
                if connection.dialect.name == "sqlite":
                    connection.exec_driver_sql("BEGIN IMMEDIATE")
                with Session(
                    bind=connection,
                    expire_on_commit=False,
                    join_transaction_mode="create_savepoint",
                ) as atomic_db:
                    receipt = self._locked_reply(
                        atomic_db,
                        event_key,
                        lock_owner,
                    )
                    payload = receipt.reply_payload
                    if not isinstance(payload, dict):
                        raise ValueError(
                            "Feishu reply outbox payload is unavailable"
                        )
                    _lock_reply_identity(atomic_db, payload)
                    operation = resolved_message_operation(payload)

                    from app.modules.feishu.service import FeishuService

                    feishu = FeishuService(
                        atomic_db,
                        tenant_key=_optional_text(
                            operation.get("tenant_key")
                        ),
                    )
                    receive_id = _required_text(
                        operation.get("receive_id"),
                        "Feishu reply receive_id is unavailable",
                    )
                    receive_id_type = _required_text(
                        operation.get("receive_id_type"),
                        "Feishu reply receive_id_type is unavailable",
                    )
                    actor = str(operation.get("actor") or "feishu")
                    message_uuid = _optional_text(
                        operation.get("message_uuid")
                    )
                    if operation.get("kind") == "text":
                        response = feishu.send_text(
                            str(operation.get("text") or ""),
                            receive_id=receive_id,
                            receive_id_type=receive_id_type,
                            actor=actor,
                            uuid=message_uuid,
                            tenant_key=_optional_text(
                                operation.get("tenant_key")
                            ),
                            record_audit=bool(
                                operation.get("record_audit", True)
                            ),
                        )
                    else:
                        card = operation.get("card")
                        if not isinstance(card, dict):
                            raise ValueError(
                                "Feishu reply card payload is invalid"
                            )
                        response = feishu.send_card(
                            card,
                            receive_id=receive_id,
                            receive_id_type=receive_id_type,
                            actor=actor,
                            uuid=message_uuid,
                            tenant_key=_optional_text(
                                operation.get("tenant_key")
                            ),
                        )
                    self._mark_reply_success(
                        atomic_db,
                        event_key,
                        lock_owner,
                        current,
                    )
                    atomic_db.commit()
                transaction.commit()
                return response
            except Exception:
                if transaction.is_active:
                    transaction.rollback()
                raise

    @staticmethod
    def _locked_reply(
        db: Session,
        event_key: str,
        lock_owner: str,
    ) -> FeishuEventReceipt:
        receipt = db.scalar(
            select(FeishuEventReceipt)
            .where(FeishuEventReceipt.event_key == event_key)
            .with_for_update()
        )
        if (
            receipt is None
            or receipt.status != FeishuInboundStatus.SUCCEEDED
            or receipt.reply_status != FeishuInboundStatus.PROCESSING
            or receipt.reply_locked_by != lock_owner
        ):
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Feishu inbound reply lease was lost",
            )
        return receipt

    @staticmethod
    def _mark_reply_success(
        db: Session,
        event_key: str,
        lock_owner: str,
        current: datetime,
    ) -> None:
        result = db.execute(
            update(FeishuEventReceipt)
            .where(
                FeishuEventReceipt.event_key == event_key,
                FeishuEventReceipt.reply_status
                == FeishuInboundStatus.PROCESSING,
                FeishuEventReceipt.reply_locked_by == lock_owner,
            )
            .values(
                reply_status=FeishuInboundStatus.SUCCEEDED,
                reply_payload=None,
                reply_last_error=None,
                reply_next_attempt_at=None,
                reply_locked_by=None,
                reply_locked_until=None,
                reply_sent_at=current,
            )
            .execution_options(synchronize_session=False)
        )
        if result.rowcount != 1:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Feishu inbound reply lease was lost",
            )

    def _finish_reply_failure(
        self,
        event_key: str,
        lock_owner: str,
        current: datetime,
        exc: Exception,
        *,
        retryable: bool,
    ) -> None:
        self.db.expire_all()
        record = self._get(event_key)
        retry_index = record.reply_attempt_count - 1
        will_retry = (
            retryable
            and 0 <= retry_index < len(FEISHU_INBOUND_RETRY_DELAYS_SECONDS)
            and record.reply_attempt_count < record.max_attempts
        )
        next_attempt_at = (
            current
            + timedelta(seconds=FEISHU_INBOUND_RETRY_DELAYS_SECONDS[retry_index])
            if will_retry
            else None
        )
        self.db.execute(
            update(FeishuEventReceipt)
            .where(
                FeishuEventReceipt.event_key == event_key,
                FeishuEventReceipt.reply_status
                == FeishuInboundStatus.PROCESSING,
                FeishuEventReceipt.reply_locked_by == lock_owner,
            )
            .values(
                reply_status=(
                    FeishuInboundStatus.RETRY
                    if will_retry
                    else FeishuInboundStatus.FAILED
                ),
                reply_payload=record.reply_payload if will_retry else None,
                reply_last_error=_error_summary(exc),
                reply_next_attempt_at=next_attempt_at,
                reply_locked_by=None,
                reply_locked_until=None,
            )
            .execution_options(synchronize_session=False)
        )
        self.db.commit()
        self.db.expire_all()

    def _fail_reply_exhausted(
        self,
        receipt_id: int,
        current: datetime,
    ) -> None:
        self.db.execute(
            update(FeishuEventReceipt)
            .where(
                FeishuEventReceipt.id == receipt_id,
                FeishuEventReceipt.reply_status.not_in(_TERMINAL_STATUSES),
                or_(
                    FeishuEventReceipt.reply_locked_until.is_(None),
                    FeishuEventReceipt.reply_locked_until <= current,
                ),
            )
            .values(
                reply_status=FeishuInboundStatus.FAILED,
                reply_payload=None,
                reply_last_error="attempts_exhausted",
                reply_next_attempt_at=None,
                reply_locked_by=None,
                reply_locked_until=None,
            )
            .execution_options(synchronize_session=False)
        )
        self.db.commit()
        self.db.expire_all()

    def _claim(
        self,
        receipt_id: int,
        lock_owner: str,
        current: datetime,
        *,
        max_attempts: int,
        expected_attempt_count: int,
        allow_pending_immediate: bool = False,
    ) -> bool:
        if allow_pending_immediate:
            record = self.db.execute(
                select(FeishuEventReceipt)
                .where(FeishuEventReceipt.id == receipt_id)
                .with_for_update(skip_locked=True)
            ).scalar_one_or_none()
            if (
                record is None
                or record.attempt_count != expected_attempt_count
                or record.attempt_count >= max_attempts
                or record.locked_by is not None
            ):
                self.db.commit()
                return False
            record.status = FeishuInboundStatus.PROCESSING
            record.attempt_count += 1
            record.locked_by = lock_owner
            record.locked_until = current + timedelta(seconds=self.lease_seconds)
            self.db.commit()
            self.db.expire_all()
            return True
        result = self.db.execute(
            update(FeishuEventReceipt)
            .where(
                FeishuEventReceipt.id == receipt_id,
                _claimable(current),
                FeishuEventReceipt.attempt_count < max_attempts,
            )
            .values(
                status=FeishuInboundStatus.PROCESSING,
                attempt_count=FeishuEventReceipt.attempt_count + 1,
                locked_by=lock_owner,
                locked_until=current + timedelta(seconds=self.lease_seconds),
            )
            .execution_options(synchronize_session=False)
        )
        self.db.commit()
        self.db.expire_all()
        return result.rowcount == 1

    @staticmethod
    def _mark_success(
        db: Session,
        event_key: str,
        lock_owner: str,
        current: datetime,
    ) -> None:
        result = db.execute(
            update(FeishuEventReceipt)
            .where(
                FeishuEventReceipt.event_key == event_key,
                FeishuEventReceipt.status == FeishuInboundStatus.PROCESSING,
                FeishuEventReceipt.locked_by == lock_owner,
            )
            .values(
                status=FeishuInboundStatus.SUCCEEDED,
                payload=None,
                last_error=None,
                next_attempt_at=None,
                locked_by=None,
                locked_until=None,
                processed_at=current,
            )
            .execution_options(synchronize_session=False)
        )
        if result.rowcount != 1:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Feishu inbound event lease was lost",
            )

    def _finish_failure(
        self,
        event_key: str,
        lock_owner: str,
        current: datetime,
        exc: Exception,
        *,
        retryable: bool,
    ) -> FeishuEventReceipt:
        self.db.expire_all()
        record = self._get(event_key)
        retry_index = record.attempt_count - 1
        will_retry = (
            retryable
            and 0 <= retry_index < len(FEISHU_INBOUND_RETRY_DELAYS_SECONDS)
            and record.attempt_count < record.max_attempts
        )
        next_attempt_at = (
            current
            + timedelta(seconds=FEISHU_INBOUND_RETRY_DELAYS_SECONDS[retry_index])
            if will_retry
            else None
        )
        result = self.db.execute(
            update(FeishuEventReceipt)
            .where(
                FeishuEventReceipt.event_key == event_key,
                FeishuEventReceipt.status == FeishuInboundStatus.PROCESSING,
                FeishuEventReceipt.locked_by == lock_owner,
            )
            .values(
                status=(
                    FeishuInboundStatus.RETRY
                    if will_retry
                    else FeishuInboundStatus.FAILED
                ),
                payload=record.payload if will_retry else None,
                last_error=_error_summary(exc),
                next_attempt_at=next_attempt_at,
                locked_by=None,
                locked_until=None,
                processed_at=None if will_retry else current,
            )
            .execution_options(synchronize_session=False)
        )
        if result.rowcount != 1:
            self.db.rollback()
            return self._get(event_key)
        self.db.commit()
        self.db.expire_all()
        return self._get(event_key)

    def _fail_exhausted(
        self,
        receipt_id: int,
        current: datetime,
    ) -> FeishuEventReceipt:
        self.db.execute(
            update(FeishuEventReceipt)
            .where(
                FeishuEventReceipt.id == receipt_id,
                FeishuEventReceipt.status.not_in(_TERMINAL_STATUSES),
                or_(
                    FeishuEventReceipt.locked_until.is_(None),
                    FeishuEventReceipt.locked_until <= current,
                ),
            )
            .values(
                status=FeishuInboundStatus.FAILED,
                payload=None,
                last_error="attempts_exhausted",
                next_attempt_at=None,
                locked_by=None,
                locked_until=None,
                processed_at=current,
            )
            .execution_options(synchronize_session=False)
        )
        self.db.commit()
        self.db.expire_all()
        record = self.db.get(FeishuEventReceipt, receipt_id)
        if record is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Feishu inbound event was not found",
            )
        return record

    def _get(self, event_key: str) -> FeishuEventReceipt:
        record = self.db.scalar(
            select(FeishuEventReceipt).where(
                FeishuEventReceipt.event_key == event_key
            )
        )
        if record is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Feishu inbound event was not found",
            )
        return record


def _claimable(current: datetime) -> Any:
    return or_(
        and_(
            FeishuEventReceipt.status.in_(
                [
                    FeishuInboundStatus.PENDING,
                    FeishuInboundStatus.RETRY,
                ]
            ),
            FeishuEventReceipt.next_attempt_at.is_not(None),
            FeishuEventReceipt.next_attempt_at <= current,
            or_(
                FeishuEventReceipt.locked_until.is_(None),
                FeishuEventReceipt.locked_until <= current,
            ),
        ),
        and_(
            FeishuEventReceipt.status == FeishuInboundStatus.PROCESSING,
            FeishuEventReceipt.locked_until.is_not(None),
            FeishuEventReceipt.locked_until <= current,
        ),
    )


def _reply_claimable(current: datetime) -> Any:
    return or_(
        and_(
            FeishuEventReceipt.reply_status.in_(
                [
                    FeishuInboundStatus.PENDING,
                    FeishuInboundStatus.RETRY,
                ]
            ),
            FeishuEventReceipt.reply_next_attempt_at.is_not(None),
            FeishuEventReceipt.reply_next_attempt_at <= current,
            or_(
                FeishuEventReceipt.reply_locked_until.is_(None),
                FeishuEventReceipt.reply_locked_until <= current,
            ),
        ),
        and_(
            FeishuEventReceipt.reply_status
            == FeishuInboundStatus.PROCESSING,
            FeishuEventReceipt.reply_locked_until.is_not(None),
            FeishuEventReceipt.reply_locked_until <= current,
        ),
    )


def _sanitize_payload(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            str(key): _sanitize_payload(item)
            for key, item in value.items()
            if str(key).casefold() not in _SENSITIVE_TRANSPORT_KEYS
        }
    if isinstance(value, list):
        return [_sanitize_payload(item) for item in value]
    if isinstance(value, tuple):
        return [_sanitize_payload(item) for item in value]
    return value


def _matches_identity(
    payload: Any,
    tenant_key: str,
    open_id: str,
) -> bool:
    return _payload_identity(payload) == (tenant_key, open_id)


def _payload_identity(payload: Any) -> tuple[str, str] | None:
    if not isinstance(payload, dict):
        return None
    header = payload.get("header") or {}
    event = payload.get("event") or {}
    sender = event.get("sender") or {}
    sender_id = sender.get("sender_id") or {}
    if not isinstance(header, dict) or not isinstance(sender_id, dict):
        return None
    tenant_key = str(header.get("tenant_key") or "").strip()
    open_id = str(sender_id.get("open_id") or "").strip()
    if not tenant_key or not open_id:
        return None
    return tenant_key, open_id


def _lock_identity_fence(db: Session, payload: dict[str, Any]) -> None:
    """Serialize an identity's handler with erasure before locking its inbox row."""

    identity = _payload_identity(payload)
    if identity is None:
        return

    # Imported locally to keep the durable inbox usable for app-ticket events
    # without introducing an import cycle at module load time.
    from app.modules.feishu_users.models import FeishuUser

    tenant_key, open_id = identity
    db.scalar(
        select(FeishuUser.id)
        .where(
            FeishuUser.tenant_key == tenant_key,
            FeishuUser.open_id == open_id,
        )
        .with_for_update()
    )


def _identity_exists(
    db: Session,
    identity: tuple[str, str] | None,
) -> bool:
    if identity is None:
        return False

    from app.modules.feishu_users.models import FeishuUser

    tenant_key, open_id = identity
    return (
        db.scalar(
            select(FeishuUser.id).where(
                FeishuUser.tenant_key == tenant_key,
                FeishuUser.open_id == open_id,
            )
        )
        is not None
    )


def _lock_reply_identity(db: Session, payload: dict[str, Any]) -> None:
    if not identity_fence_required(payload):
        return
    identity = reply_payload_identity(payload)
    if identity is None:
        raise ValueError("Feishu reply identity fence is unavailable")

    from app.modules.feishu_users.models import FeishuUser

    tenant_key, open_id = identity
    owner_id = db.scalar(
        select(FeishuUser.id)
        .where(
            FeishuUser.tenant_key == tenant_key,
            FeishuUser.open_id == open_id,
        )
        .with_for_update()
    )
    if owner_id is None:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Feishu reply identity no longer exists",
        )


def _is_retryable(exc: Exception) -> bool:
    if isinstance(exc, FeishuAPIError):
        return exc.retryable
    if isinstance(exc, HTTPException):
        return (
            exc.status_code == status.HTTP_429_TOO_MANY_REQUESTS
            or exc.status_code >= status.HTTP_500_INTERNAL_SERVER_ERROR
        )
    if isinstance(exc, (TypeError, ValueError)):
        return False
    return True


def _error_summary(exc: Exception) -> str:
    if isinstance(exc, FeishuAPIError):
        return (
            "FeishuAPIError:"
            f"http_status={exc.http_status}:provider_code={exc.provider_code}"
        )[:2000]
    if isinstance(exc, HTTPException):
        return f"HTTPException:status_code={exc.status_code}"
    return type(exc).__name__[:2000]


def _attach_provider_response(
    result: dict[str, Any],
    provider_response: dict[str, Any] | None,
) -> None:
    if provider_response is None:
        return
    command_result = result.get(FeishuResponseKey.RESULT)
    if not isinstance(command_result, dict):
        return
    if FeishuCommandResultKey.PROVIDER_RESPONSE in command_result:
        command_result[FeishuCommandResultKey.PROVIDER_RESPONSE] = provider_response


def _optional_text(value: Any) -> str | None:
    text = str(value or "").strip()
    return text or None


def _required_text(value: Any, message: str) -> str:
    text = _optional_text(value)
    if text is None:
        raise ValueError(message)
    return text