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[0]; const SOURCE_TYPES: Record = { 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 { 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, 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][] = []; 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[] = []; 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`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, }; }, }; }