UHD_Media_Tool/apps/server/test/logs/legacy-query.ts

import type { LogSource, LogType } from "@umt/contracts";
import { type DB, type Kysely, sql } from "@umt/db";
import type { RawBuilder } from "kysely";
import { BUSINESS_EXPR } from "../../src/modules/auth-settings/engine-business.ts";
import {
  BUSINESS_OBJECT_TYPE,
  decodeCursor,
  encodeCursor,
  type LogQueries,
  toEntry,
} from "../../src/modules/logs/query.ts";

// 改写前的日志查询(先构造带派生列的子查询、外层再筛选),只作对照口径:新查询在同一份数据上的结果必须与它一致。
// 在大数据量上极慢,只用于小数据集。

type Row = Parameters<typeof toEntry>[0];

const SOURCE_TYPES: Record<LogSource, LogType[]> = {
  engine_run: ["task"],
  schedule_run: ["auto"],
  agent_run: ["auto"],
  audit_event: ["auth", "audit"],
};

const escapeLike = (s: string) => s.replace(/[\\%_]/g, (c) => `\\${c}`);

const UUID_RE = "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$";

function engineSource(schema: string): RawBuilder<unknown> {
  if (!/^[a-z_][a-z0-9_]*$/.test(schema)) throw new Error(`非法 schema 名 ${schema}`);
  const t = sql.raw(`"${schema}".workflow_runs`);
  const kind = sql`(biz.b->>'kind')`;
  const ref = sql`(biz.b->>'refId')`;
  const uuidRef = sql`(case when ${ref} ~* ${UUID_RE} then ${ref}::uuid end)`;
  const objectType = sql`(case ${kind} ${sql.join(
    Object.entries(BUSINESS_OBJECT_TYPE).map(([k, v]) => sql`when ${sql.lit(k)} then ${sql.lit(v)}`),
    sql` `,
  )} end)`;
  const title = sql`(case ${kind}
    when 'schedule' then (select j.name from scheduled_job j where j.id = ${ref})
    when 'agent' then (select a.goal from agent_run a where a.id = ${uuidRef})
    when 'download' then (select d.name from download_task d where d.id = ${uuidRef})
    when 'organization' then (select coalesce(e.title, r.target_path, r.source_path) from organization_record r
      left join media_entity e on e.id = r.media_entity_id where r.id = ${uuidRef})
    when 'subscription' then (select coalesce(e.title, s.title_query) from subscription s
      left join media_entity e on e.id = s.media_entity_id where s.id = ${uuidRef})
    when 'feed' then (select f.name from feed_source f where f.id = ${uuidRef})
    when 'site' then (select s.name from site s where s.id = ${uuidRef})
    when 'decision' then (select d.title from decision d where d.id = ${uuidRef})
    when 'media' then (select e.title from media_entity e where e.id = ${uuidRef})
  end)`;
  const agent = sql`(case when ${kind} = 'agent' then
    (select jsonb_build_object('status', a.status, 'trigger', a.trigger) from agent_run a where a.id = ${uuidRef}) end)`;
  const st = sql.raw("workflow_runs.status");
  return sql`select
      'engine_run:' || workflow_runs.id as id,
      'task'::text as type,
      'engine_run'::text as source,
      workflow_runs.id::text as ref_id,
      workflow_runs.created_at as at,
      workflow_runs.finished_at as finished_at,
      (case when workflow_runs.finished_at is not null and workflow_runs.started_at is not null
        then extract(epoch from workflow_runs.finished_at - workflow_runs.started_at) * 1000 end)::float8 as duration_ms,
      (case
        when ${st} = 'failed' then 'error'
        when ${st} = 'canceled' then 'warn'
        when ${st} in ('completed', 'succeeded') then (case ag.a->>'status'
          when 'failed' then 'error' when 'budget_exceeded' then 'warn' when 'cancelled' then 'warn' else 'success' end)
        else 'running' end) as level,
      ${st}::text as status,
      coalesce(bt.title, workflow_runs.workflow_name) as title,
      (case when ${st} = 'failed' then left(coalesce(workflow_runs.error->>'message', workflow_runs.error::text), 500) end)
        as message,
      ${objectType} as object_type,
      ${ref} as object_id,
      bt.title as object_label,
      null::text as actor_type,
      null::text as actor_id,
      jsonb_build_object('workflowName', workflow_runs.workflow_name, 'businessKind', ${kind},
        'businessRefId', ${ref}, 'businessTitle', bt.title, 'outcome', ag.a->>'status',
        'agentTrigger', ag.a->>'trigger', 'attempts', workflow_runs.attempts) as extra,
      (biz.b is null
        or coalesce(${kind} = 'schedule' and ${st} in ('completed', 'succeeded'), false)
        or coalesce((ag.a->>'trigger') in ('event', 'schedule'), false)) as routine,
      concat_ws(' ', workflow_runs.workflow_name, workflow_runs.id, bt.title, ${ref},
        left(workflow_runs.error::text, 500)) as search
    from ${t}
    cross join lateral (select ${BUSINESS_EXPR} as b) biz
    cross join lateral (select ${title} as title) bt
    cross join lateral (select ${agent} as a) ag
    where workflow_runs.namespace_id = 'default'`;
}

const SCHEDULE_SOURCE = sql`select
    'schedule_run:' || r.id::text as id,
    'auto'::text as type,
    'schedule_run'::text as source,
    r.id::text as ref_id,
    r.started_at as at,
    r.finished_at as finished_at,
    r.duration_ms::float8 as duration_ms,
    (case
      when r.status = 'failed' or r.agent_status = 'failed' then 'error'
      when r.agent_status in ('budget_exceeded', 'cancelled') then 'warn'
      when r.status = 'ok' then 'success'
      when r.status = 'skipped' then 'info'
      else 'running' end) as level,
    r.status::text as status,
    coalesce(j.name, r.job_id) as title,
    coalesce(r.error, r.summary) as message,
    'scheduled_job'::text as object_type,
    r.job_id as object_id,
    j.name as object_label,
    r.trigger::text as actor_type,
    null::text as actor_id,
    jsonb_build_object('jobId', r.job_id, 'jobKind', j.kind, 'trigger', r.trigger, 'agentRunId', r.agent_run_id,
      'agentStatus', r.agent_status, 'summary', r.summary, 'error', r.error) as extra,
    (r.status in ('ok', 'skipped') and coalesce(j.kind::text, 'task') <> 'agent_check'
      and r.agent_status is distinct from 'failed') as routine,
    concat_ws(' ', r.job_id, j.name, r.summary, r.error, r.agent_run_id::text) as search
  from scheduled_job_run r left join scheduled_job j on j.id = r.job_id`;

const AGENT_SOURCE = sql`select
    'agent_run:' || a.id::text as id,
    'auto'::text as type,
    'agent_run'::text as source,
    a.id::text as ref_id,
    a.created_at as at,
    a.finished_at as finished_at,
    (case when a.finished_at is not null and a.started_at is not null
      then extract(epoch from a.finished_at - a.started_at) * 1000 end)::float8 as duration_ms,
    (case a.status
      when 'succeeded' then 'success' when 'failed' then 'error'
      when 'cancelled' then 'warn' when 'budget_exceeded' then 'warn' else 'running' end) as level,
    a.status::text as status,
    a.goal as title,
    coalesce(a.summary, a.error->>'message', a.spawn_reason) as message,
    'agent_run'::text as object_type,
    a.id::text as object_id,
    null::text as object_label,
    'event'::text as actor_type,
    a.created_by as actor_id,
    jsonb_build_object('trigger', a.trigger, 'mode', a.mode, 'toolCalls', a.tool_call_count,
      'spawnReason', a.spawn_reason, 'model', a.model) as extra,
    false as routine,
    concat_ws(' ', a.goal, a.summary, a.id::text, left(a.error::text, 500)) as search
  from agent_run a where a.trigger = 'event' and a.parent_run_id is null`;

const AUDIT_SOURCE = sql`select
    'audit_event:' || e.id::text as id,
    (case when e.action like 'admin.%' or e.action like 'api_key.%' then 'auth' else 'audit' end) as type,
    'audit_event'::text as source,
    e.id::text as ref_id,
    e.at as at,
    null::timestamptz as finished_at,
    null::float8 as duration_ms,
    (case
      when e.action in ('admin.login_failed', 'capability.denied') then 'warn'
      when e.action in ('admin.login_success', 'admin.passkey_login', 'admin.setup') then 'success'
      else 'info' end) as level,
    e.action as status,
    e.action as title,
    (case when e.action like 'admin.%' then concat_ws(' · ', e.payload->>'username', e.payload->>'ip')
      else nullif(concat_ws(' · ', e.payload->>'reason', e.payload->>'error'), '') end) as message,
    e.target_type as object_type,
    e.target_id as object_id,
    null::text as object_label,
    e.actor_type as actor_type,
    e.actor_id as actor_id,
    jsonb_build_object('action', e.action, 'traceId', e.trace_id, 'payload', e.payload) as extra,
    false as routine,
    concat_ws(' ', e.action, e.target_type, e.target_id, e.actor_id, left(e.payload::text, 2000)) as search
  from audit_event e`;

export function legacyLogQueries(db: Kysely<DB>, opts: { engineSchema?: string } = {}): LogQueries {
  const engineSchema = opts.engineSchema ?? "openworkflow";
  let engineReady = false;
  const engineExists = async () => {
    if (engineReady) return true;
    const r = await sql<{
      t: string | null;
    }>`select to_regclass(${`"${engineSchema}".workflow_runs`})::text as t`.execute(db);
    engineReady = !!r.rows[0]?.t;
    return engineReady;
  };

  return {
    async list(q) {
      const types = q.type?.length ? new Set(q.type) : null;
      const sources: [LogSource, RawBuilder<unknown>][] = [];
      const want = (s: LogSource) => !types || SOURCE_TYPES[s].some((t) => types.has(t));
      if (want("engine_run") && (await engineExists())) sources.push(["engine_run", engineSource(engineSchema)]);
      if (want("schedule_run")) sources.push(["schedule_run", SCHEDULE_SOURCE]);
      if (want("agent_run")) sources.push(["agent_run", AGENT_SOURCE]);
      if (want("audit_event")) sources.push(["audit_event", AUDIT_SOURCE]);
      if (!sources.length) return { items: [], nextCursor: null };

      const conds: RawBuilder<unknown>[] = [];
      if (types) conds.push(sql`s.type = any(${[...types]}::text[])`);
      if (q.level?.length) conds.push(sql`s.level = any(${q.level}::text[])`);
      if (q.from) conds.push(sql`s.at >= ${new Date(q.from)}`);
      if (q.to) conds.push(sql`s.at < ${new Date(q.to)}`);
      if (q.q) conds.push(sql`s.search ilike ${`%${escapeLike(q.q)}%`}`);
      if (q.objectType) conds.push(sql`s.object_type = ${q.objectType}`);
      if (q.objectId) conds.push(sql`s.object_id = ${q.objectId}`);
      if (q.system !== "1" && !q.objectId) conds.push(sql`not s.routine`);
      const cur = decodeCursor(q.cursor);
      if (cur) {
        conds.push(sql`(s.at, s.id) < (timestamptz 'epoch' + ${cur.us} * interval '1 microsecond', ${cur.id})`);
      }
      const where = conds.length ? sql`where ${sql.join(conds, sql` and `)}` : sql``;
      const n = q.limit + 1;
      const parts = sources.map(
        ([, src]) => sql`(select s.* from (${src}) s ${where} order by s.at desc, s.id desc limit ${n})`,
      );
      // 对照只关心结果;关掉 JIT 省掉每条查询上秒的编译
      const rows = await db.transaction().execute(async (trx) => {
        await sql`set local jit = off`.execute(trx);
        return sql<Row & { cursor_us: string }>`select u.*, (extract(epoch from u.at) * 1000000)::bigint as cursor_us
          from (${sql.join(parts, sql` union all `)}) u
          order by u.at desc, u.id desc limit ${n}`.execute(trx);
      });
      const items = rows.rows.slice(0, q.limit);
      const last = items[items.length - 1];
      return {
        items: items.map(toEntry),
        nextCursor:
          rows.rows.length > q.limit && last ? encodeCursor({ us: Number(last.cursor_us), id: last.id }) : null,
      };
    },
  };
}