rakuten-api/app/trading/worker/local_db.py

"""本地订单 SQLite：执行事实的权威记录

与网关的 tasks/task_reports 表对偶，但本地这张表才是「这单到底下没下、付没付」
的权威。网关失联或数据丢失时，本地这张表用于补报与对账。

三张表：
- orders：每笔任务一行，记录开始/结束时间与最终状态
- order_events：append-only 的状态迁移事件（与网关 task_reports 对账）
- evidence_index：每步证据文件的相对路径索引，便于事后翻查
"""
from __future__ import annotations

import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import aiosqlite

logger = logging.getLogger(__name__)

SCHEMA = """
CREATE TABLE IF NOT EXISTS orders (
    task_id      TEXT PRIMARY KEY,
    site         TEXT NOT NULL,
    intent_json  TEXT NOT NULL,
    final_state  TEXT,
    started_at   TEXT NOT NULL,
    finished_at  TEXT
);

CREATE TABLE IF NOT EXISTS order_events (
    task_id      TEXT NOT NULL,
    state        TEXT NOT NULL,
    detail       TEXT,
    evidence_ref TEXT,
    recorded_at  TEXT NOT NULL,
    PRIMARY KEY (task_id, state)
);

CREATE TABLE IF NOT EXISTS evidence_index (
    task_id      TEXT NOT NULL,
    step_no      INTEGER NOT NULL,
    step_name    TEXT NOT NULL,
    rel_path     TEXT NOT NULL,
    saved_at     TEXT NOT NULL,
    PRIMARY KEY (task_id, step_no)
);
"""


def _utcnow_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


@dataclass(slots=True)
class OrderRow:
    task_id: str
    site: str
    intent_json: str
    final_state: str | None
    started_at: str
    finished_at: str | None

    @property
    def intent(self) -> dict[str, Any]:
        return json.loads(self.intent_json)


class LocalDB:
    """本地订单 SQLite 访问对象"""

    def __init__(self, db_path: Path):
        self._db_path = db_path
        self._conn: aiosqlite.Connection | None = None

    async def start(self) -> None:
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        self._conn = await aiosqlite.connect(str(self._db_path))
        self._conn.row_factory = aiosqlite.Row
        await self._conn.executescript(SCHEMA)
        await self._conn.commit()
        logger.info("本地订单 DB 已就绪：%s", self._db_path)

    async def close(self) -> None:
        if self._conn is not None:
            await self._conn.close()
            self._conn = None

    @property
    def conn(self) -> aiosqlite.Connection:
        if self._conn is None:
            raise RuntimeError("LocalDB 未启动：先调用 start()")
        return self._conn

    # ---- orders ----

    async def ensure_started(self, task_id: str, site: str, intent: dict[str, Any]) -> None:
        """记录任务开始执行。已存在则忽略（同任务被重新领回时）"""
        await self.conn.execute(
            "INSERT OR IGNORE INTO orders (task_id, site, intent_json, started_at) "
            "VALUES (?, ?, ?, ?)",
            (task_id, site, json.dumps(intent, ensure_ascii=False), _utcnow_iso()),
        )
        await self.conn.commit()

    async def get_order(self, task_id: str) -> OrderRow | None:
        async with self.conn.execute(
            "SELECT * FROM orders WHERE task_id = ?",
            (task_id,),
        ) as cur:
            row = await cur.fetchone()
            if row is None:
                return None
            return OrderRow(
                task_id=row["task_id"],
                site=row["site"],
                intent_json=row["intent_json"],
                final_state=row["final_state"],
                started_at=row["started_at"],
                finished_at=row["finished_at"],
            )

    async def has_finished(self, task_id: str) -> bool:
        """是否已有终态记录。用于主循环里的本地幂等闸门"""
        async with self.conn.execute(
            "SELECT 1 FROM orders WHERE task_id = ? AND finished_at IS NOT NULL",
            (task_id,),
        ) as cur:
            return await cur.fetchone() is not None

    async def mark_finished(self, task_id: str, final_state: str) -> None:
        await self.conn.execute(
            "UPDATE orders SET final_state = ?, finished_at = ? WHERE task_id = ?",
            (final_state, _utcnow_iso(), task_id),
        )
        await self.conn.commit()

    async def final_state(self, task_id: str) -> str | None:
        async with self.conn.execute(
            "SELECT final_state FROM orders WHERE task_id = ?",
            (task_id,),
        ) as cur:
            row = await cur.fetchone()
            return row["final_state"] if row else None

    # ---- events ----

    async def record_event(
        self,
        task_id: str,
        state: str,
        *,
        detail: str = "",
        evidence_ref: str | None = None,
    ) -> None:
        """记录一次状态迁移事件（与网关 task_reports 对账）

        同一 (task_id, state) 重复记录视为同一次——INSERT OR IGNORE，与网关侧的
        INSERT OR REPLACE 不同：本地这张表是事件日志，不覆盖既有记录。
        """
        await self.conn.execute(
            "INSERT OR IGNORE INTO order_events "
            "(task_id, state, detail, evidence_ref, recorded_at) VALUES (?, ?, ?, ?, ?)",
            (task_id, state, detail, evidence_ref, _utcnow_iso()),
        )
        await self.conn.commit()

    # ---- evidence index ----

    async def index_evidence(
        self, task_id: str, step_no: int, step_name: str, rel_path: str
    ) -> None:
        await self.conn.execute(
            "INSERT OR REPLACE INTO evidence_index "
            "(task_id, step_no, step_name, rel_path, saved_at) VALUES (?, ?, ?, ?, ?)",
            (task_id, step_no, step_name, rel_path, _utcnow_iso()),
        )
        await self.conn.commit()