AgentLab/backend/app/process/receipts.py

"""按幂等键去重的收件记录（docs/design/next-round/04-对外集成与触达.md A4）：入站钩子（hook_receipts）和访问密钥
（api_receipts）共用这一套状态机，各存各的表（Book 按表参数化）。

一条记录按（谁, 幂等键）唯一：钩子按钩子，访问密钥按拥有者（轮换了密钥，同一个键照样认得出是同一件事）。
- claim：先插一行 processing。插得进去就是头一回，交回 (None, False)；插不进去（撞了唯一约束）看已有的那一行：
  - accepted：交回上次的结果（duplicate: true），不再处理；
  - rejected：重新认领。被拒的（表单不对、没有在等的待办）可以用同一个键重试——外部系统多半会原样重发，拒过一次就永远拒，
    它只能换键，去重也就落空了；
  - processing 超过 STALE_S：处理一次请求用不了这么久，多半是中途服务停了，重新认领，交回 reclaimed=True。调用方先找这个键
    已经办成的事（已经发起的实例、已经解决的待办），找到就交回它，不重复办；
  - processing 没超时：正在处理，抛 ReceiptBusy（调用方回 409）。
  重新认领用条件 UPDATE（状态、收到时间都对得上才改）：并发的两次只有一次认领得到。
- finish：处理完写结果（accepted 带运行或待办，rejected 带原因）。

有指纹的表（api_receipts.fingerprint）另核对同一个键办的是不是同一件事：办成过的、还在办的，指纹对不上抛 ReceiptMismatch，
不把上一件事的结果当成这一件的交回去；被拒过的不算办过，换了内容重试照常认领（指纹换成这一次的）。
"""
from __future__ import annotations

from dataclasses import dataclass
from datetime import timedelta
from typing import Any

from sqlalchemy import select, update
from sqlalchemy.exc import IntegrityError

from app.db.base import SessionLocal, utcnow

#: 回执停在 processing 超过这么久（秒），同一个幂等键可以重新认领
STALE_S = 600
#: 幂等键最长存多少字
KEY_MAX = 200


class ReceiptBusy(Exception):
    """同一个幂等键的请求正在处理（调用方回 409）。"""


class ReceiptMismatch(Exception):
    """同一个幂等键已经用来办过另一件事（指纹对不上，调用方回 422）。"""


@dataclass(frozen=True)
class Book:
    """一张收件记录表：model 是表，owner 是「谁」那一列的名字（hook_id / owner_user_id）。表里要有 idem_key、status、
    detail、run_id、task_id、received_at；fingerprint 列可有可无。"""

    model: Any
    owner: str
    stale_s: int = STALE_S

    def _key(self, owner: str, idem_key: str) -> list[Any]:
        return [getattr(self.model, self.owner) == owner, self.model.idem_key == idem_key]

    async def claim(self, owner: str, idem_key: str, *, fingerprint: str | None = None,
                    extra: dict[str, Any] | None = None) -> tuple[dict[str, Any] | None, bool]:
        """按（谁, 幂等键）认领这一次，交回 (上次的结果, 是不是重新认领了一条停在处理中的)。规则见模块说明。extra 是
        认领时一并记下的列（访问密钥记哪一把密钥、哪个接口）。"""
        fields = {**(extra or {}), **({"fingerprint": fingerprint} if fingerprint is not None else {})}
        async with SessionLocal() as session:
            session.add(self.model(**{self.owner: owner}, idem_key=idem_key, status="processing", **fields))
            try:
                await session.commit()
                return None, False
            except IntegrityError:
                await session.rollback()
            row = await session.scalar(select(self.model).where(*self._key(owner, idem_key)))
            if row is None:
                # 撞了约束、回头又查不到：那一行刚被删掉（钩子被删除时连同记录一起删）。当成正在处理，让调用方稍后重发
                raise ReceiptBusy
            known = getattr(row, "fingerprint", None)
            if fingerprint is not None and known and known != fingerprint and row.status != "rejected":
                raise ReceiptMismatch
            if row.status == "accepted":
                return {"status": "accepted", "duplicate": True, "run_id": row.run_id, "task_id": row.task_id}, False
            stale = row.status == "processing" and row.received_at < utcnow() - timedelta(seconds=self.stale_s)
            if row.status == "rejected" or stale:
                again = await session.execute(update(self.model).where(
                    self.model.id == row.id, self.model.status == row.status,
                    self.model.received_at == row.received_at)
                    .values(status="processing", detail=None, received_at=utcnow(), **fields))
                await session.commit()
                if again.rowcount == 1:
                    return None, stale
        raise ReceiptBusy

    async def finish(self, owner: str, idem_key: str, *, status: str, detail: str | None = None,
                     run_id: str | None = None, task_id: str | None = None) -> None:
        async with SessionLocal() as session:
            await session.execute(update(self.model).where(*self._key(owner, idem_key))
                                  .values(status=status, detail=detail, run_id=run_id, task_id=task_id))
            await session.commit()