pigway-frame-flow/pigway_frame_flow/api/history.py

"""Paginated history without removing operation receipts used by other services."""

from datetime import UTC, date, datetime, time, timedelta
from zoneinfo import ZoneInfo

from fastapi import Depends, Query, Request
from pydantic import BaseModel, Field

from pigway_frame_flow.core.auth import require
from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import Operation, OperationStatus


class HistoryPage(BaseModel):
    items: list[Operation]
    total: int
    page: int
    page_size: int


class HistoryDelete(BaseModel):
    operation_ids: list[str] = Field(min_length=1, max_length=200)


def install_history(app, service):
    db = service.db

    @app.get(
        "/api/v1/operation-history",
        response_model=HistoryPage,
        tags=["Operations"],
        dependencies=[Depends(require("history:read"))],
    )
    def list_history(
        request: Request,
        page: int = Query(1, ge=1),
        page_size: int = Query(20, ge=1, le=100),
        status: OperationStatus | None = None,
        action: str = Query("", max_length=80),
        device_id: str = Query("", max_length=120),
        q: str = Query("", max_length=200),
        start: date | None = None,
        end: date | None = None,
    ):
        if start and end and start > end:
            raise FlowError(
                E.INVALID_REQUEST, "Start date must not follow end date", 422
            )
        clauses = [
            "NOT EXISTS (SELECT 1 FROM hidden_history h WHERE h.operation_id=o.id)"
        ]
        args: list = []
        if "admin" not in request.state.permissions:
            clauses.append("o.client_id=?")
            args.append(request.state.client_id)
        for column, value in [
            ("status", status),
            ("action", action),
            ("device_id", device_id),
        ]:
            if value:
                clauses.append(f"o.{column}=?")
                args.append(value)
        if q.strip():
            clauses.append(
                "instr(lower(o.id||' '||o.request_id||' '||o.client_id||' '||coalesce(o.result,'')),lower(?))>0"
            )
            args.append(q.strip())
        for day, operator, offset in [(start, ">=", 0), (end, "<", 1)]:
            if day:
                boundary = datetime.combine(
                    day, time(), ZoneInfo("Asia/Shanghai")
                ) + timedelta(days=offset)
                clauses.append(f"o.created_at{operator}?")
                args.append(boundary.astimezone(UTC).isoformat())
        where = " AND ".join(clauses)
        # Keep count and rows consistent if a worker finishes or a deletion occurs.
        with db.lock:
            total = db.one(
                "SELECT count(*) AS total FROM operations o WHERE " + where, args
            )["total"]
            page = min(page, max(1, (total + page_size - 1) // page_size))
            rows = db.all(
                "SELECT o.id FROM operations o WHERE "
                + where
                + " ORDER BY o.created_at DESC,o.rowid DESC LIMIT ? OFFSET ?",
                [*args, page_size, (page - 1) * page_size],
            )
            return HistoryPage(
                items=[service.operation(row["id"]) for row in rows],
                total=total,
                page=page,
                page_size=page_size,
            )

    @app.delete(
        "/api/v1/operation-history",
        tags=["Operations"],
        dependencies=[Depends(require("admin"))],
    )
    def delete_history(payload: HistoryDelete):
        ids = list(dict.fromkeys(payload.operation_ids))
        placeholders = ",".join("?" for _ in ids)
        with db.transaction() as connection:
            rows = connection.execute(
                f"SELECT id,status FROM operations WHERE id IN ({placeholders})", ids
            ).fetchall()
            if len(rows) != len(ids):
                raise FlowError(E.NOT_FOUND, "History record not found", 404)
            if any(
                row["status"] not in {"COMPLETED", "FAILED", "UNKNOWN"} for row in rows
            ):
                raise FlowError(E.CONFLICT, "Running operations cannot be deleted", 409)
            before = connection.total_changes
            connection.executemany(
                "INSERT OR IGNORE INTO hidden_history(operation_id) VALUES(?)",
                [(key,) for key in ids],
            )
            deleted = connection.total_changes - before
        return {"deleted": deleted}