UHD_Media_Tool/apps/server/src/modules/auth-settings/engine-business.ts

import type { EngineBusinessKind, EngineRunBusiness, EngineRunQuery, EngineTriggeredBy } from "@umt/contracts";
import { type DB, type Kysely, sql } from "@umt/db";
import type { RawBuilder } from "kysely";

// 引擎 runs 的业务字段(W9-I):run → 关联业务对象 {kind, refId, title, triggeredBy}。
// kind/refId 用一条 SQL 表达式推断(列表筛选与展示共用,保证一致),按优先级取第一个命中:
//   1. run 输入(信封 input.input)里的业务 id 键:agentRunId / recordId / downloadTaskId / acquisitionPlanId /
//      subscriptionId / acquisitionSlotId / feedSourceId / siteId / accountId / decisionId / mediaEntityId
//   2. 业务任务行 task(engine_run_id = run.id)的 subject_type / subject_id
//   3. 无业务 id 的定时任务(purgeTrash 等,非手动触发)→ schedule + 定时任务 key
// title 与 triggeredBy 在 TS 里按页批量查(agent_run / task / tasks_run_meta / outbox_event / 各业务表)。
// 表达式引用外层 "<engine schema>".workflow_runs 的列,必须在以 workflow_runs 为 FROM 的查询里使用。

const I = sql.raw(`(workflow_runs.input->'input')`);
const RUN_ID = sql.raw(`workflow_runs.id`);
const RUN_NAME = sql.raw(`workflow_runs.workflow_name`);

/** 输入里的文本值转 uuid(不是 uuid 时为 null,避免 cast 报错) */
const uuidOf = (key: string) =>
  sql`(case when (${I}->>${key}) ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then (${I}->>${key})::uuid end)`;
const obj = (kind: EngineBusinessKind, ref: RawBuilder<unknown>) =>
  sql`jsonb_build_object('kind', ${sql.lit(kind)}::text, 'refId', (${ref})::text)`;
const whenKey = (key: string, kind: EngineBusinessKind, ref: RawBuilder<unknown> = sql`${I}->>${key}`) =>
  sql`(case when jsonb_typeof(${I}) = 'object' and ${I} ? ${key} and (${ref}) is not null then ${obj(kind, ref)} end)`;

/** 下载:优先指向下载任务(plan 已提交时),否则指向收录计划 id */
const planRef = sql`coalesce((select dt.id from download_task dt where dt.acquisition_plan_id = ${uuidOf("acquisitionPlanId")} order by dt.created_at limit 1), ${uuidOf("acquisitionPlanId")})`;
const slotSubRef = sql`(select s.subscription_id from acquisition_slot s where s.id = ${uuidOf("acquisitionSlotId")})`;
const accountSiteRef = sql`(select a.site_id from site_account a where a.id = ${uuidOf("accountId")})`;

/** 无业务 id 的任务 → 定时任务 key(只在非手动 / 非重跑时认作定时触发) */
export const SCHEDULE_TASK_JOBS: Record<string, string> = {
  purgeTrash: "trash.purge",
  refreshPiTokens: "pi.token_refresh",
  cleanupDownloadCache: "cache.threshold",
  reconcileLibrary: "library.reconcile",
};

const taskSubject = sql`(select case t.subject_type
    when 'agent_run' then jsonb_build_object('kind', 'agent', 'refId', t.subject_id::text)
    when 'organization_record' then jsonb_build_object('kind', 'organization', 'refId', t.subject_id::text)
    when 'download_task' then jsonb_build_object('kind', 'download', 'refId', t.subject_id::text)
    when 'acquisition_plan' then jsonb_build_object('kind', 'download', 'refId',
      coalesce((select dt.id::text from download_task dt where dt.acquisition_plan_id::text = t.subject_id::text
                 order by dt.created_at limit 1), t.subject_id::text))
    when 'acquisition_slot' then (select jsonb_build_object('kind', 'subscription', 'refId', s.subscription_id::text)
      from acquisition_slot s where s.id::text = t.subject_id::text and s.subscription_id is not null)
    when 'media_entity' then jsonb_build_object('kind', 'media', 'refId', t.subject_id::text)
  end
  from task t where t.engine_run_id = ${RUN_ID} and t.subject_id is not null order by t.created_at desc limit 1)`;

const manualKey = sql`coalesce((select m.idempotency_key ~ '^retry:|(^|:)manual(:|$)' from tasks_run_meta m where m.run_id = ${RUN_ID}), false)`;
const scheduleBranch = sql`(case when ${RUN_NAME} in (${sql.join(
  Object.keys(SCHEDULE_TASK_JOBS).map((k) => sql.lit(k)),
)}) and not ${manualKey} then jsonb_build_object('kind', 'schedule', 'refId', case ${RUN_NAME} ${sql.join(
  Object.entries(SCHEDULE_TASK_JOBS).map(([t, j]) => sql`when ${sql.lit(t)} then ${sql.lit(j)}`),
  sql` `,
)} end) end)`;

/** run → {kind, refId} 的 jsonb 表达式(可能为 null) */
export const BUSINESS_EXPR = sql<{ kind: EngineBusinessKind; refId: string } | null>`coalesce(
  ${whenKey("agentRunId", "agent")},
  ${whenKey("recordId", "organization")},
  ${whenKey("downloadTaskId", "download")},
  ${whenKey("acquisitionPlanId", "download", planRef)},
  ${whenKey("subscriptionId", "subscription")},
  ${whenKey("acquisitionSlotId", "subscription", slotSubRef)},
  ${whenKey("feedSourceId", "feed")},
  ${whenKey("siteId", "site")},
  ${whenKey("accountId", "site", accountSiteRef)},
  ${whenKey("decisionId", "decision")},
  ${whenKey("mediaEntityId", "media")},
  ${taskSubject},
  ${scheduleBranch}
)`;

/** EngineRunQuery 的业务筛选条件(加到 workflow_runs 查询的 where) */
export function businessConds(q: Pick<EngineRunQuery, "businessKind" | "businessRefId">): RawBuilder<unknown>[] {
  const out: RawBuilder<unknown>[] = [];
  if (q.businessKind) out.push(sql`(${BUSINESS_EXPR})->>'kind' = ${q.businessKind}`);
  if (q.businessRefId) out.push(sql`(${BUSINESS_EXPR})->>'refId' = ${q.businessRefId}`);
  return out;
}

// ── triggeredBy ──

/** task.triggered_by / agent_run.created_by 这类 "<kind>:<id>" 串 → 触发来源;认不出返回 null */
export function triggeredByFromActor(s: string | null | undefined): EngineTriggeredBy | null {
  if (!s) return null;
  const head = s.split(":")[0]!.toLowerCase();
  if (head === "api_key") return "api_key";
  if (["admin_session", "dev_admin", "user", "admin", "console"].includes(head)) return "user";
  if (head === "agent" || head === "subagent") return "agent";
  if (head === "schedule" || head === "scheduler") return "schedule";
  if (head === "event") return "event";
  return null;
}

/** 调度器按时间槽入队的幂等键前缀(见各模块 scheduler / start) */
const SCHEDULE_KEY =
  /^(snapshotSite:(?!manual:)|refreshSiteAccount:(relogin|userData):|slot-due:|refreshMetadata:tracking:|refreshMetadata:subscription:|search-eval:|refreshPiTokens:|purgeTrash:|reconcileLibrary:auto:|cleanupDownloadCache:threshold:|auto:|threshold:)/;

export interface TriggerFacts {
  kind: EngineBusinessKind;
  idempotencyKey: string | null;
  taskTriggeredBy: string | null;
  agentTrigger: string | null;
  agentCreatedBy: string | null;
  fromOutbox: boolean;
}

/** 触发来源推断(纯函数,便于测试):agent_run.trigger → task.triggered_by → 幂等键(手动/重跑)→ outbox → 调度槽 */
export function inferTriggeredBy(f: TriggerFacts): EngineTriggeredBy | null {
  if (f.agentTrigger) {
    if (f.agentTrigger === "subagent") return "agent";
    if (f.agentTrigger === "schedule") return "schedule";
    if (f.agentTrigger === "event") return "event";
    if (f.agentTrigger === "user") return triggeredByFromActor(f.agentCreatedBy) === "api_key" ? "api_key" : "user";
  }
  const byTask = triggeredByFromActor(f.taskTriggeredBy);
  if (byTask) return byTask;
  const key = f.idempotencyKey ?? "";
  if (/^retry:|(^|:)manual(:|$)/.test(key)) return "user";
  if (f.fromOutbox) return "event";
  if (f.kind === "schedule" || SCHEDULE_KEY.test(key)) return "schedule";
  return null;
}

// ── 批量补字段 ──

type Row = { id: string };
type Enriched<T> = T & { business: EngineRunBusiness | null };

const TITLE_SQL: Record<EngineBusinessKind, (ids: string[]) => RawBuilder<{ id: string; title: string | null }>> = {
  agent: (ids) => sql`select id::text as id, goal as title from agent_run where id::text = any(${ids}::text[])`,
  subscription: (ids) => sql`select s.id::text as id, coalesce(e.title, s.title_query) as title
    from subscription s left join media_entity e on e.id = s.media_entity_id where s.id::text = any(${ids}::text[])`,
  download: (ids) => sql`select id::text as id, name as title from download_task where id::text = any(${ids}::text[])`,
  organization: (ids) => sql`select r.id::text as id, coalesce(e.title, r.target_path, r.source_path) as title
    from organization_record r left join media_entity e on e.id = r.media_entity_id where r.id::text = any(${ids}::text[])`,
  feed: (ids) => sql`select id::text as id, name as title from feed_source where id::text = any(${ids}::text[])`,
  site: (ids) => sql`select id::text as id, name as title from site where id::text = any(${ids}::text[])`,
  decision: (ids) => sql`select id::text as id, title from decision where id::text = any(${ids}::text[])`,
  media: (ids) => sql`select id::text as id, title from media_entity where id::text = any(${ids}::text[])`,
  schedule: (ids) => sql`select id, name as title from scheduled_job where id = any(${ids}::text[])`,
};

const tableOf = (schema: string) => {
  if (!/^[a-z_][a-z0-9_]*$/.test(schema)) throw new Error(`非法 schema 名 ${schema}`);
  return sql.raw(`"${schema}".workflow_runs`);
};

/** 给一页 run 补 business 字段(缺则 null);查询失败时整页降级为 null,不影响原响应 */
export async function attachBusiness<T extends Row>(
  db: Kysely<DB>,
  schema: string,
  items: T[],
): Promise<Enriched<T>[]> {
  if (!items.length) return [];
  try {
    const map = await resolveBusiness(
      db,
      schema,
      items.map((i) => i.id),
    );
    return items.map((i) => ({ ...i, business: map.get(i.id) ?? null }));
  } catch {
    return items.map((i) => ({ ...i, business: null }));
  }
}

export async function resolveBusiness(
  db: Kysely<DB>,
  schema: string,
  runIds: string[],
): Promise<Map<string, EngineRunBusiness>> {
  const out = new Map<string, EngineRunBusiness>();
  if (!runIds.length) return out;
  const rows = await sql<{
    id: string;
    b: { kind: EngineBusinessKind; refId: string } | null;
    idempotency_key: string | null;
    task_triggered_by: string | null;
  }>`select workflow_runs.id, ${BUSINESS_EXPR} as b,
        (select m.idempotency_key from tasks_run_meta m where m.run_id = workflow_runs.id) as idempotency_key,
        (select t.triggered_by from task t where t.engine_run_id = workflow_runs.id order by t.created_at desc limit 1)
          as task_triggered_by
      from ${tableOf(schema)} where workflow_runs.id = any(${runIds}::text[])`.execute(db);

  const found = rows.rows.filter((r) => r.b?.kind && r.b.refId);
  if (!found.length) return out;

  // outbox 来源:引擎幂等键 = outbox_event.idempotency_key 或 outbox:<event id>
  const keys = found.map((r) => r.idempotency_key).filter((k): k is string => !!k);
  const outboxKeys = new Set<string>();
  if (keys.length) {
    const ev = await sql<{ k: string }>`
      select idempotency_key as k from outbox_event where idempotency_key = any(${keys}::text[])
      union all
      select 'outbox:' || id::text as k from outbox_event
       where id::text = any(${keys.filter((k) => k.startsWith("outbox:")).map((k) => k.slice(7))}::text[])`.execute(db);
    for (const r of ev.rows) outboxKeys.add(r.k);
  }

  // 标题 + agent 触发信息:按 kind 分组批量查
  const byKind = new Map<EngineBusinessKind, Set<string>>();
  for (const r of found) {
    const set = byKind.get(r.b!.kind) ?? new Set<string>();
    set.add(r.b!.refId);
    byKind.set(r.b!.kind, set);
  }
  const titles = new Map<string, string | null>();
  await Promise.all(
    [...byKind].map(async ([kind, ids]) => {
      const q = TITLE_SQL[kind];
      const r = await q([...ids]).execute(db);
      for (const t of r.rows) titles.set(`${kind}:${t.id}`, t.title);
    }),
  );
  const agentIds = [...(byKind.get("agent") ?? [])];
  const agents = new Map<string, { trigger: string; created_by: string | null; status: string }>();
  if (agentIds.length) {
    const r = await sql<{ id: string; trigger: string; created_by: string | null; status: string }>`
      select id::text as id, trigger, created_by, status from agent_run where id::text = any(${agentIds}::text[])`.execute(
      db,
    );
    for (const a of r.rows) agents.set(a.id, a);
  }

  for (const r of found) {
    const b = r.b!;
    const agent = b.kind === "agent" ? agents.get(b.refId) : undefined;
    out.set(r.id, {
      kind: b.kind,
      refId: b.refId,
      title: titles.get(`${b.kind}:${b.refId}`) ?? null,
      triggeredBy: inferTriggeredBy({
        kind: b.kind,
        idempotencyKey: r.idempotency_key,
        taskTriggeredBy: r.task_triggered_by,
        agentTrigger: agent?.trigger ?? null,
        agentCreatedBy: agent?.created_by ?? null,
        fromOutbox: !!r.idempotency_key && outboxKeys.has(r.idempotency_key),
      }),
      ...(agent ? { outcome: agent.status } : {}),
    });
  }
  return out;
}