UHD_Media_Tool/apps/web/app/api/tasks.ts

import { useQuery } from "@tanstack/react-query";
import {
  type AcquisitionPlan,
  cursorPage,
  Decision,
  type DownloadTask,
  EngineRun,
  EngineRunDetail,
  type PublicOrganizationRecord,
  SourceRelease,
} from "@umt/contracts";
import { PLAN_STATUS_LABEL } from "./downloads.ts";
import { apiParsed, cleanParams } from "./parse.ts";
import { useRecordsForEntity } from "./records.ts";

// 业务任务视图:引擎 run 关联 + 按 trace 汇总的业务时间线(参考 MovieClaw task-activity)
// 关联键:plan → download_task(instanceId + hash)→ organization_record → decision/agent_run。

const EnginePage = cursorPage(EngineRun);
const DecisionPage = cursorPage(Decision);

const RESOLUTION_LABEL = {
  unresolved: "未识别",
  resolved: "已识别",
  ambiguous: "有歧义,待确认",
  unidentifiable: "无法识别",
  rejected: "已排除",
} as const;

/** 时间线阶段:observation → plan → download → organization → agent */
export type TimelineStage = "observation" | "plan" | "download" | "organization" | "agent" | "engine";
export type TimelineTone = "success" | "warning" | "destructive" | "neutral";

export type TimelineEvent = {
  key: string;
  stage: TimelineStage;
  at: string;
  title: string;
  detail?: string | null;
  tone: TimelineTone;
  href?: string;
  /** 进行中的阶段(不是终态) */
  ongoing?: boolean;
};

export function useSourceRelease(id: string | null | undefined) {
  return useQuery({
    queryKey: ["source-release", id ?? ""],
    enabled: !!id,
    queryFn: ({ signal }) =>
      apiParsed(SourceRelease, "/api/v1/source-releases/:id", { pathParams: { id: id ?? "" }, signal }),
  });
}

/** 与该下载有关的待处理事项(subject = 下载任务或计划),含 Agent 调查 run */
export function useSubjectDecisions(subjectIds: string[]) {
  const ids = subjectIds.filter(Boolean);
  return useQuery({
    queryKey: ["decisions", "subjects", ids],
    enabled: ids.length > 0,
    queryFn: ({ signal }) => apiParsed(DecisionPage, "/api/v1/decisions", { params: { limit: 200 }, signal }),
    select: (d) => d.items.filter((x) => x.subjectId && ids.includes(x.subjectId)),
  });
}

/**
 * 引擎 run 关联:submitDownload 的 input 带 acquisitionPlanId,verifyAndImport 的 input 带 downloadTaskId。
 * 服务端没有"按业务对象查 run"的端点,这里在时间窗内取少量候选 run 读详情后按 input 精确匹配。
 */
export function useTraceEngineRuns(opts: { planId?: string | null; downloadTaskId?: string; since?: string }) {
  const { planId, downloadTaskId, since } = opts;
  return useQuery({
    queryKey: ["engine", "trace", planId ?? "", downloadTaskId ?? "", since ?? ""],
    enabled: !!since && (!!planId || !!downloadTaskId),
    staleTime: 60_000,
    queryFn: async ({ signal }) => {
      const from = since ? new Date(new Date(since).getTime() - 10 * 60_000).toISOString() : undefined;
      const lists = await Promise.all(
        ["submitDownload", "verifyAndImport"].map((workflowName) =>
          apiParsed(EnginePage, "/api/v1/engine/runs", {
            params: cleanParams({ workflowName, from, limit: 15 }),
            signal,
          }).catch(() => ({ items: [], nextCursor: null })),
        ),
      );
      const candidates = lists.flatMap((l) => l.items);
      const details = await Promise.all(
        candidates.map((r) =>
          apiParsed(EngineRunDetail, "/api/v1/engine/runs/:id", { pathParams: { id: r.id }, signal }).catch(() => null),
        ),
      );
      return details.filter((d): d is EngineRunDetail => {
        if (!d) return false;
        const input = (d.input ?? {}) as Record<string, unknown>;
        return (
          (!!planId && input.acquisitionPlanId === planId) ||
          (!!downloadTaskId && input.downloadTaskId === downloadTaskId)
        );
      });
    },
  });
}

const RECORD_TONE: Record<PublicOrganizationRecord["status"], TimelineTone> = {
  queued: "neutral",
  running: "neutral",
  succeeded: "success",
  partial: "warning",
  failed: "destructive",
  cancelled: "neutral",
};

/** 整理记录与下载的关联:同实例 + 源路径落在下载内容路径下(合同里没有 downloadTaskId,只能按路径匹配) */
export function recordsForDownload(records: PublicOrganizationRecord[], task: DownloadTask) {
  const base = (task.contentPath ?? task.savePath ?? "").replace(/\/+$/, "");
  return records.filter(
    (r) =>
      r.source.downloaderInstanceId === task.instanceId &&
      !!base &&
      !!r.source.path &&
      (r.source.path === base || r.source.path.startsWith(`${base}/`)),
  );
}

/** 汇总一个下载 trace 的业务事件,按时间升序 */
export function buildTimeline(input: {
  task: DownloadTask;
  release?: SourceRelease | null;
  plan?: AcquisitionPlan | null;
  records?: PublicOrganizationRecord[];
  decisions?: Decision[];
  engineRuns?: EngineRunDetail[];
}): TimelineEvent[] {
  const { task, release, plan, records = [], decisions = [], engineRuns = [] } = input;
  const ev: TimelineEvent[] = [];
  if (release) {
    ev.push({
      key: "obs",
      stage: "observation",
      at: release.firstSeenAt,
      title: "首次发现资源",
      detail: release.title,
      tone: "neutral",
    });
    if (release.resolutionStatus !== "unresolved")
      ev.push({
        key: "resolve",
        stage: "observation",
        // 合同里没有识别完成时间,挂在首次发现之后,只表达结果
        at: release.firstSeenAt,
        title:
          release.resolutionStatus === "resolved"
            ? "识别结果:已关联本地作品"
            : `识别结果:${RESOLUTION_LABEL[release.resolutionStatus]}`,
        tone: release.resolutionStatus === "resolved" ? "success" : "warning",
      });
  }
  if (plan) {
    ev.push({
      key: "plan",
      stage: "plan",
      at: plan.createdAt,
      title: `创建下载计划(${plan.purpose === "fill" ? "补缺" : plan.purpose === "upgrade" ? "升级" : "换源"})`,
      detail: plan.statusReason,
      tone: "neutral",
    });
    if (plan.updatedAt !== plan.createdAt)
      ev.push({
        key: "plan-status",
        stage: "plan",
        at: plan.updatedAt,
        title: `计划状态:${PLAN_STATUS_LABEL[plan.status]}`,
        detail: plan.statusReason,
        tone: plan.status === "failed" ? "destructive" : plan.status === "expired" ? "warning" : "neutral",
      });
  }
  ev.push({
    key: "dl-submit",
    stage: "download",
    at: task.createdAt,
    title: task.origin === "adopted" ? "接管已有下载" : "提交下载",
    detail: task.name,
    tone: "neutral",
  });
  if (task.pauseReason)
    ev.push({
      key: "dl-pause",
      stage: "download",
      at: task.updatedAt,
      title: task.pauseReason === "threshold" ? "缓存达到硬阈值,系统暂停" : `已暂停(${task.pauseReason})`,
      tone: "warning",
      ongoing: task.status === "paused",
    });
  if (task.noProgressSince)
    ev.push({
      key: "dl-stall",
      stage: "download",
      at: task.noProgressSince,
      title: "无进展(只记录,不自动换源)",
      tone: "warning",
      ongoing: task.status === "stalled",
    });
  if (task.completedAt)
    ev.push({ key: "dl-done", stage: "download", at: task.completedAt, title: "下载完成", tone: "success" });
  else if (task.status === "downloading")
    ev.push({
      key: "dl-progress",
      stage: "download",
      at: task.lastProgressAt ?? task.updatedAt,
      title: `下载中 ${(task.progress * 100).toFixed(1)}%`,
      tone: "neutral",
      ongoing: true,
    });
  if (task.lastError)
    ev.push({
      key: "dl-err",
      stage: "download",
      at: task.updatedAt,
      title: "下载器报错",
      detail: String((task.lastError as { message?: unknown }).message ?? JSON.stringify(task.lastError)),
      tone: "destructive",
    });
  for (const r of records) {
    const c = r.counts;
    ev.push({
      key: `rec-${r.id}`,
      stage: "organization",
      at: r.startedAt ?? r.createdAt,
      title: `整理 ${r.action === "copy" ? "Copy" : r.action === "move" ? "Move" : r.action}:${r.status}`,
      detail: `${c.succeeded}/${c.total} 文件成功${c.failed ? `,${c.failed} 失败` : ""}${c.skipped ? `,${c.skipped} 跳过` : ""}${r.reason ? ` · ${r.reason.message}` : ""}`,
      tone: RECORD_TONE[r.status],
      href: `/console/organize/${r.id}`,
      ongoing: r.status === "running" || r.status === "queued",
    });
  }
  for (const d of decisions) {
    ev.push({
      key: `dec-${d.id}`,
      stage: "agent",
      at: d.createdAt,
      title: `待处理:${d.title}`,
      detail: d.summary,
      tone: d.severity === "critical" ? "destructive" : "warning",
      href: "/console/inbox",
    });
    if (d.agentRunId)
      ev.push({
        key: `agent-${d.id}`,
        stage: "agent",
        at: d.updatedAt,
        title: "Agent 已介入调查",
        tone: "neutral",
        href: `/console/agent/${d.agentRunId}`,
      });
  }
  for (const r of engineRuns) {
    ev.push({
      key: `eng-${r.id}`,
      stage: "engine",
      at: r.startedAt ?? r.createdAt,
      title: `引擎 ${r.workflowName}:${r.status}`,
      detail: `${r.stepAttempts.length} 个步骤尝试 · 第 ${r.attempts} 次`,
      tone:
        r.status === "failed"
          ? "destructive"
          : r.status === "completed" || r.status === "succeeded"
            ? "success"
            : "neutral",
      href: `/console/engine/${encodeURIComponent(r.id)}`,
      ongoing: r.status === "running" || r.status === "pending" || r.status === "sleeping",
    });
  }
  return ev.sort((a, b) => a.at.localeCompare(b.at));
}

/** 一站式取 trace 数据 */
export function useDownloadTrace(task: DownloadTask | undefined, plan: AcquisitionPlan | null | undefined) {
  const release = useSourceRelease(task?.sourceReleaseId ?? plan?.sourceReleaseId);
  const entityId = release.data?.mediaEntityId ?? null;
  const records = useRecordsForEntity(entityId);
  const decisions = useSubjectDecisions([task?.id ?? "", task?.acquisitionPlanId ?? ""]);
  const engine = useTraceEngineRuns({
    planId: task?.acquisitionPlanId,
    downloadTaskId: task?.id,
    since: plan?.createdAt ?? task?.createdAt,
  });
  const matched = task ? recordsForDownload(records.data?.items ?? [], task) : [];
  return {
    release,
    records,
    decisions,
    engine,
    events: task
      ? buildTimeline({
          task,
          release: release.data,
          plan,
          records: matched,
          decisions: decisions.data,
          engineRuns: engine.data,
        })
      : [],
    loading: release.isLoading || records.isLoading || decisions.isLoading || engine.isLoading,
  };
}