MoviePilot/app/db/oper/transferpending.py

"""
持久化整理任务、检查点与租约的数据访问入口，适配调用方会话及兼容事务执行器。
"""

from typing import Any, Optional

from app.db.base import DbOper
from app.db.models.transferpending import TransferPending


class TransferPendingOper(DbOper):
    """
    待整理文件登记管理。

    保存稳定任务身份、存储、源文件路径和准入状态，用于在进程重启后把没走完
    整理链的文件重新送回去，避免挂载故障重启后永久漏件。
    """

    def stage_isolate_unhandled_failure(
            self, *, task_id: str, lease_token: str, error: str,
            now_utc: str, updated_at: str,
    ) -> int:
        """
        在调用方事务中暂存异常隔离，失败计数与租约释放使用同一 CAS。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.isolate_unhandled_failure(
                session, task_id=task_id, lease_token=lease_token,
                error=error, now_utc=now_utc, updated_at=updated_at,
            )
        )

    def stage_admit(self, *, task_id: str, storage: str, src_path: str,
                    state: str, now_time: str, input_version: int,
                    planning_input: dict[str, Any],
                    input_fingerprint: str) -> Optional[TransferPending]:
        """
        在当前会话中暂存一条持久接纳记录。

        适配器应传入显式 Session，使提交与回滚仍由应用用例对应的 UoW 管理。

        :param task_id: 任务标识
        :param storage: 存储
        :param src_path: 源文件路径
        :param state: 持久状态
        :param now_time: 当前时间
        :param input_version: 规划输入格式版本
        :param planning_input: 版本化规划输入 JSON
        :param input_fingerprint: 规划输入规范 JSON 指纹
        :return: 接纳记录
        """
        return self._execute_sync_write(
            lambda session: TransferPending.stage_admit(
                session,
                task_id=task_id,
                storage=storage,
                src_path=src_path,
                state=state,
                now_time=now_time,
                input_version=input_version,
                planning_input=planning_input,
                input_fingerprint=input_fingerprint,
            )
        )

    def get_by_identity(self, *, storage: str,
                        src_path: str) -> Optional[TransferPending]:
        """
        使用当前会话按存储与源路径查询接纳记录。

        :param storage: 存储
        :param src_path: 源文件路径
        :return: 接纳记录
        """
        return self._execute_sync_query(
            lambda session: TransferPending.get_by_identity(
                session,
                storage=storage,
                src_path=src_path,
            )
        )

    def stage_delete_inactive_for_replacement(
            self,
            *,
            task_id: str,
            storage: str,
            src_path: str,
            now_time: str,
    ) -> int:
        """
        在调用方准入事务中 CAS 删除无有效租约且无真实历史绑定的旧任务。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.delete_inactive_for_replacement(
                session,
                task_id=task_id,
                storage=storage,
                src_path=src_path,
                now_time=now_time,
            )
        )

    def get_task_id_by_identity(self, *, storage: str, src_path: str) -> Optional[str]:
        """
        只读稳定任务身份，避免损坏规划 JSON 阻断用户显式重做。
        """
        return self._execute_sync_query(
            lambda session: TransferPending.get_task_id_by_identity(
                session, storage=storage, src_path=src_path,
            )
        )

    def get_by_task_id(self, *, task_id: str) -> Optional[TransferPending]:
        """
        使用当前会话按稳定任务标识查询接纳记录。

        :param task_id: 稳定任务标识
        :return: 接纳记录
        """
        return self._execute_sync_query(
            lambda session: TransferPending.get_by_task_id(
                session,
                task_id=task_id,
            )
        )

    def stage_checkpoint_plan(
            self,
            *,
            task_id: str,
            input_fingerprint: str,
            checkpoint_version: int,
            checkpoint_payload: dict[str, Any],
            source_states: tuple[str, ...],
            target_state: str,
            lease_token: str,
            now_time: str,
            updated_at: str,
    ) -> int:
        """
        在当前会话中以输入指纹为条件暂存完整计划检查点。

        :param task_id: 稳定任务标识
        :param input_fingerprint: 规划输入规范 JSON 指纹
        :param checkpoint_version: 检查点格式版本
        :param checkpoint_payload: 完整有序计划 JSON
        :param source_states: 允许执行 CAS 的起始状态
        :param target_state: 检查点提交后的目标状态
        :param lease_token: 当前且未过期的租约令牌
        :param now_time: 用于租约 fencing 的当前 UTC 时间
        :param updated_at: 与既有业务审计字段一致的宿主本地时间
        :return: 更新的记录数
        """
        return self._execute_sync_write(
            lambda session: TransferPending.checkpoint_plan(
                session,
                task_id=task_id,
                input_fingerprint=input_fingerprint,
                checkpoint_version=checkpoint_version,
                checkpoint_payload=checkpoint_payload,
                source_states=source_states,
                target_state=target_state,
                lease_token=lease_token,
                now_time=now_time,
                updated_at=updated_at,
            )
        )

    def stage_record_planning_failure(self, *, task_id: str, lease_token: str,
                                      error: str, now_time: str,
                                      updated_at: str) -> int:
        """
        在当前会话中记录规划失败并保持任务处于接纳态。

        :param task_id: 稳定任务标识
        :param lease_token: 当前且未过期的租约令牌
        :param error: 失败原因
        :param now_time: 用于租约 fencing 的当前 UTC 时间
        :param updated_at: 与既有业务审计字段一致的宿主本地时间
        :return: 更新的记录数
        """
        return self._execute_sync_write(
            lambda session: TransferPending.record_planning_failure(
                session,
                task_id=task_id,
                lease_token=lease_token,
                error=error,
                now_time=now_time,
                updated_at=updated_at,
            )
        )

    def list_claimable_candidates(
            self,
            *,
            states: tuple[str, ...],
            now_time: str,
            limit: int,
            after_cursor: Optional[tuple[str, int]] = None,
    ) -> list[tuple[str, str, int]]:
        """
        使用当前会话按稳定游标读取未租用或已过期的恢复候选。

        :param states: 可恢复业务状态
        :param now_time: 当前 UTC 时间
        :param limit: 候选数量上限
        :param after_cursor: 上一页最后一条的规范登记时间与主键
        :return: 任务标识、规范登记时间与主键组成的稳定游标列表
        """
        return self._execute_sync_query(
            lambda session: TransferPending.list_claimable_candidates(
                session,
                states=states,
                now_time=now_time,
                limit=limit,
                after_cursor=after_cursor,
            )
        ) or []

    def stage_claim_task(
            self,
            *,
            task_id: str,
            states: tuple[str, ...],
            owner_id: str,
            lease_token: str,
            now_time: str,
            lease_expires_at: str,
            updated_at: str,
    ) -> int:
        """
        使用当前会话以未租或过期条件竞争一个新租约。

        :param task_id: 稳定任务标识
        :param states: 可恢复业务状态
        :param owner_id: 新租约拥有者
        :param lease_token: 新租约唯一令牌
        :param now_time: 当前 UTC 时间
        :param lease_expires_at: 新租约到期时间
        :param updated_at: 与既有业务审计字段一致的宿主本地时间
        :return: 更新的记录数
        """
        return self._execute_sync_write(
            lambda session: TransferPending.claim_task(
                session,
                task_id=task_id,
                states=states,
                owner_id=owner_id,
                lease_token=lease_token,
                now_time=now_time,
                lease_expires_at=lease_expires_at,
                updated_at=updated_at,
            )
        )

    def stage_record_projection_failure(
            self,
            *,
            task_id: str,
            states: tuple[str, ...],
            error: str,
            now_time: str,
            updated_at: str,
    ) -> int:
        """
        使用当前会话按无有效租约和诊断变化条件记录投影损坏。

        :param task_id: 稳定任务标识
        :param states: 可恢复业务状态
        :param error: 可持久化的稳定诊断文本
        :param now_time: 当前 UTC 租约时间
        :param updated_at: 宿主本地业务审计时间
        :return: 更新的记录数
        """
        return self._execute_sync_write(
            lambda session: TransferPending.record_projection_failure(
                session,
                task_id=task_id,
                states=states,
                error=error,
                now_time=now_time,
                updated_at=updated_at,
            )
        )

    def stage_heartbeat(
            self,
            *,
            task_id: str,
            lease_token: str,
            now_time: str,
            lease_expires_at: str,
    ) -> int:
        """
        使用当前会话以当前未过期 token 延长租约。

        :param task_id: 稳定任务标识
        :param lease_token: 当前租约令牌
        :param now_time: 当前 UTC 时间
        :param lease_expires_at: 新租约到期时间
        :return: 更新的记录数
        """
        return self._execute_sync_write(
            lambda session: TransferPending.heartbeat(
                session,
                task_id=task_id,
                lease_token=lease_token,
                now_time=now_time,
                lease_expires_at=lease_expires_at,
            )
        )

    def stage_defer_planning(self, *, task_id: str, lease_token: str, error: str,
                            retry_due_at: str, now_time: str, updated_at: str, max_retries: int) -> int:
        """
        在调用方事务内按租约和空操作证据约束暂存规划退避。
        """
        return self._execute_sync_write(lambda session: TransferPending.defer_planning(
            session, task_id=task_id, lease_token=lease_token, error=error,
            retry_due_at=retry_due_at, now_time=now_time, updated_at=updated_at, max_retries=max_retries,
        ))

    def stage_release_claim(
            self,
            *,
            task_id: str,
            lease_token: str,
            error: Optional[str],
            now_time: str,
            updated_at: str,
    ) -> int:
        """
        使用当前会话按未过期 token 释放租约并记录本次错误。

        :param task_id: 稳定任务标识
        :param lease_token: 当前租约令牌
        :param error: 本次执行错误，成功释放时为空
        :param now_time: 当前 UTC 时间
        :param updated_at: 与既有业务审计字段一致的宿主本地时间
        :return: 更新的记录数
        """
        return self._execute_sync_write(
            lambda session: TransferPending.release_claim(
                session,
                task_id=task_id,
                lease_token=lease_token,
                error=error,
                now_time=now_time,
                updated_at=updated_at,
            )
        )

    def stage_abandon_unstarted(
            self,
            *,
            task_id: str,
            lease_token: str,
            now_time: str,
    ) -> int:
        """
        使用当前会话按当前未过期 token 删除确认从未执行的缺失源任务。

        :param task_id: 稳定任务标识
        :param lease_token: 当前租约令牌
        :param now_time: 当前 UTC 时间
        :return: 删除的记录数
        """
        return self._execute_sync_write(
            lambda session: TransferPending.abandon_unstarted(
                session,
                task_id=task_id,
                lease_token=lease_token,
                now_time=now_time,
            )
        )

    def stage_record_enqueue_failure(self, *, task_id: str, error: str,
                                     now_time: str) -> int:
        """
        在当前会话中暂存最近一次入队失败。

        :param task_id: 任务标识
        :param error: 失败原因
        :param now_time: 当前时间
        :return: 更新的记录数
        """
        return self._execute_sync_write(
            lambda session: TransferPending.record_enqueue_failure(
                session,
                task_id=task_id,
                error=error,
                now_time=now_time,
            )
        )

    def stage_execution_running(
            self,
            *,
            task_id: str,
            lease_token: str,
            admission_state: str,
            checkpoint_version: int,
            checkpoint_payload: dict[str, Any],
            now_utc: str,
            updated_at: str,
    ) -> int:
        """
        以有效租约暂存任务执行状态为 running。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.stage_execution_running(
                session,
                task_id=task_id,
                lease_token=lease_token,
                admission_state=admission_state,
                checkpoint_version=checkpoint_version,
                checkpoint_payload=checkpoint_payload,
                now_utc=now_utc,
                updated_at=updated_at,
            )
        )

    def stage_defer_execution(
            self,
            *,
            task_id: str,
            lease_token: str,
            error: str,
            retry_due_at: str,
            now_utc: str,
            updated_at: str,
    ) -> int:
        """
        暂存重试世代和到期时间，并原子释放当前租约。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.defer_execution(
                session,
                task_id=task_id,
                lease_token=lease_token,
                error=error,
                retry_due_at=retry_due_at,
                now_utc=now_utc,
                updated_at=updated_at,
            )
        )

    def stage_mark_execution_manual_review(
            self,
            *,
            task_id: str,
            lease_token: str,
            error: str,
            now_utc: str,
            updated_at: str,
    ) -> int:
        """
        暂存人工复核隔离状态，并原子释放当前租约。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.mark_execution_manual_review(
                session,
                task_id=task_id,
                lease_token=lease_token,
                error=error,
                now_utc=now_utc,
                updated_at=updated_at,
            )
        )

    def stage_checkpoint_execution(
            self,
            *,
            task_id: str,
            lease_token: str,
            execution_version: int,
            execution_payload: dict[str, Any],
            execution_fingerprint: str,
            now_utc: str,
            updated_at: str,
    ) -> int:
        """
        以有效租约暂存聚合执行检查点并进入 settling。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.checkpoint_execution(
                session,
                task_id=task_id,
                lease_token=lease_token,
                execution_version=execution_version,
                execution_payload=execution_payload,
                execution_fingerprint=execution_fingerprint,
                now_utc=now_utc,
                updated_at=updated_at,
            )
        )

    def stage_checkpoint_exhausted_failure(
            self,
            *,
            task_id: str,
            lease_token: str,
            execution_version: int,
            execution_payload: dict[str, Any],
            execution_fingerprint: str,
            error: str,
            now_utc: str,
            updated_at: str,
    ) -> int:
        """
        暂存预算耗尽失败检查点，并保留有效 lease 进入 settling。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.checkpoint_exhausted_failure(
                session,
                task_id=task_id,
                lease_token=lease_token,
                execution_version=execution_version,
                execution_payload=execution_payload,
                execution_fingerprint=execution_fingerprint,
                error=error,
                now_utc=now_utc,
                updated_at=updated_at,
            )
        )

    def stage_request_execution_retry(
            self,
            *,
            task_id: str,
            reason: str,
            requested_by: str,
            retry_due_at: str,
            updated_at: str,
    ) -> int:
        """
        仅将 FAILED 任务暂存为立即到期的 retry_wait。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.request_execution_retry(
                session,
                task_id=task_id,
                reason=reason,
                requested_by=requested_by,
                retry_due_at=retry_due_at,
                updated_at=updated_at,
            )
        )

    def stage_resolve_manual_review(
            self,
            *,
            task_id: str,
            decision: str,
            actor: str,
            reason: str,
            retry_due_at: str,
            updated_at: str,
    ) -> int:
        """
        无 lease 地暂存人工判定审计并交回 retry_wait 调度。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.resolve_manual_review(
                session,
                task_id=task_id,
                decision=decision,
                actor=actor,
                reason=reason,
                retry_due_at=retry_due_at,
                updated_at=updated_at,
            )
        )

    def stage_terminal_failure(
            self,
            *,
            task_id: str,
            lease_token: str,
            execution_fingerprint: str,
            expected_revision: int,
            history_id: int,
            error: Optional[str],
            now_utc: str,
            updated_at: str,
    ) -> int:
        """
        以执行指纹和结算版本暂存失败终态。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.stage_terminal_failure(
                session,
                task_id=task_id,
                lease_token=lease_token,
                execution_fingerprint=execution_fingerprint,
                expected_revision=expected_revision,
                history_id=history_id,
                error=error,
                now_utc=now_utc,
                updated_at=updated_at,
            )
        )

    def stage_delete_terminal_failure(
            self,
            *,
            task_id: str,
            history_id: int,
            settlement_revision: int,
    ) -> int:
        """
        暂存与指定失败回执精确匹配的无租约 pending 删除。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.delete_terminal_failure(
                session,
                task_id=task_id,
                history_id=history_id,
                settlement_revision=settlement_revision,
            )
        )

    def stage_delete_terminal_success(
            self,
            *,
            task_id: str,
            lease_token: str,
            execution_fingerprint: str,
            expected_revision: int,
            now_utc: str,
    ) -> int:
        """
        以执行指纹和结算版本暂存成功 pending 删除。
        """
        return self._execute_sync_write(
            lambda session: TransferPending.delete_terminal_success(
                session,
                task_id=task_id,
                lease_token=lease_token,
                execution_fingerprint=execution_fingerprint,
                expected_revision=expected_revision,
                now_utc=now_utc,
            )
        )