From 744252b9699a18c4b70cd35b2b0fb33f021f895c Mon Sep 17 00:00:00 2001 From: root Date: Tue, 6 Oct 2026 17:58:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20=E6=97=A5=E5=BF=97=E6=A8=A1?= =?UTF-8?q?=E5=9D=97=E2=80=94=E2=80=94=E4=BB=BB=E5=8A=A1=20/=20=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E4=B8=8E=E8=87=AA=E5=8A=A8=20/=20=E7=99=BB=E5=BD=95?= =?UTF-8?q?=20/=20=E5=AE=A1=E8=AE=A1=E5=90=88=E5=B9=B6=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E4=B8=8E=20SSE=20=E6=B5=81=EF=BC=8C=E7=99=BB=E5=BD=95=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E7=95=99=E5=AE=A1=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../src/modules/auth-settings/auth-routes.ts | 18 +- apps/server/src/modules/index.ts | 2 + apps/server/src/modules/logs/index.ts | 130 ++++++++ apps/server/src/modules/logs/query.ts | 311 ++++++++++++++++++ apps/server/src/modules/logs/stream.ts | 56 ++++ apps/server/test/auth-settings/login.test.ts | 6 + apps/server/test/logs/logs.test.ts | 299 +++++++++++++++++ apps/server/test/modules.test.ts | 3 +- 8 files changed, 823 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/modules/logs/index.ts create mode 100644 apps/server/src/modules/logs/query.ts create mode 100644 apps/server/src/modules/logs/stream.ts create mode 100644 apps/server/test/logs/logs.test.ts diff --git a/apps/server/src/modules/auth-settings/auth-routes.ts b/apps/server/src/modules/auth-settings/auth-routes.ts index 73d03b56..59189159 100644 --- a/apps/server/src/modules/auth-settings/auth-routes.ts +++ b/apps/server/src/modules/auth-settings/auth-routes.ts @@ -135,7 +135,23 @@ export function registerAuthRoutes(app: ModuleApp, ctx: ServerContext) { }); endpoint(app, D.login, async ({ body, c }) => { - const r = await verifyLogin(deps(), body, clientIp(c)); + const r = await verifyLogin(deps(), body, clientIp(c)).catch(async (err) => { + // 日志模块的「登录」类:用户名 / 密码 / 验证码错误留一条(被限速拦下的不写,见 AuditAction 注释) + if (err instanceof ApiException && err.code === "unauthorized") { + await audit( + ctx, + undefined, + "admin.login_failed", + { type: "admin_user" }, + { + username: body.username, + ip: clientIp(c), + reason: err.message, + }, + ); + } + throw err; + }); if (r.status === "totp_required") return { status: "totp_required" as const }; const s = await startSession(c, r.admin.id); const actor = adminActor(r.admin); diff --git a/apps/server/src/modules/index.ts b/apps/server/src/modules/index.ts index bd0909b1..a2a432ac 100644 --- a/apps/server/src/modules/index.ts +++ b/apps/server/src/modules/index.ts @@ -5,6 +5,7 @@ import { browserModule } from "./browser/index.ts"; import { cacheModule } from "./cache/index.ts"; import { desktopModule } from "./desktop/index.ts"; import { fileManagerModule } from "./file-manager/index.ts"; +import { logsModule } from "./logs/index.ts"; import { mcpModule } from "./mcp/index.ts"; import { piModule } from "./pi/index.ts"; import { schedulesModule } from "./schedules/index.ts"; @@ -37,4 +38,5 @@ export const SERVER_MODULES: readonly ServerModule[] = [ webSearchModule, cacheModule, systemModule, + logsModule, ]; diff --git a/apps/server/src/modules/logs/index.ts b/apps/server/src/modules/logs/index.ts new file mode 100644 index 00000000..47ab3820 --- /dev/null +++ b/apps/server/src/modules/logs/index.ts @@ -0,0 +1,130 @@ +import { createRoute } from "@hono/zod-openapi"; +import { Api, apiResponse, LOG_STREAM_EVENT, type LogFilter, LogStreamQuery } from "@umt/contracts"; +import type { Context } from "hono"; +import { streamSSE } from "hono/streaming"; +import { requireScope } from "../../http/auth.ts"; +import { ApiException } from "../../http/errors.ts"; +import type { ServerModule } from "../types.ts"; +import { type LogQueries, logQueries } from "./query.ts"; +import { createLogStreamPoller, STREAM_POLL_MS } from "./stream.ts"; + +export { decodeCursor, encodeCursor, logQueries } from "./query.ts"; +export { createLogStreamPoller } from "./stream.ts"; + +/** + * 日志模块(界面第二轮 V 线,docs/runbook/logs.md): + * - GET /logs:任务 / 定时与自动 / 登录 / 审计合并成一份列表(查询时 union,不新建表),仅 admin; + * - GET /logs/stream:同一组筛选的 SSE 流,事件名 log.entry(新日志或状态变化,客户端按 id 覆盖)。 + */ +export interface LogsModuleOptions { + /** 引擎 schema(测试覆盖);缺省 openworkflow */ + engineSchema?: string; + /** 流的轮询间隔,缺省 2 秒 */ + pollMs?: number; + keepaliveMs?: number; +} + +export function createLogsModule(opts: LogsModuleOptions = {}): ServerModule { + return { + name: "logs", + registerRoutes(app, ctx) { + let q: LogQueries | null = null; + const queries = () => { + q ??= logQueries(ctx.db, { engineSchema: opts.engineSchema }); + return q; + }; + + const list = Api.logs.list; + app.openapi( + createRoute({ + method: list.method, + path: list.path, + summary: list.summary, + tags: ["logs"], + middleware: [requireScope(list.scope)] as never, + request: { query: list.query as never }, + responses: { + 200: { description: list.summary, content: { "application/json": { schema: apiResponse(list.response) } } }, + }, + }), + // biome-ignore lint/suspicious/noExplicitAny: 入参已由 zod-openapi 校验 + (async (c: any) => { + const query = c.req.valid("query"); + const data = list.response.parse(await queries().list(query)); + return c.json({ ok: true, data }, 200); + // biome-ignore lint/suspicious/noExplicitAny: 同上 + }) as any, + ); + + const stream = Api.logs.stream; + app.openapi( + createRoute({ + method: stream.method, + path: stream.path, + summary: stream.summary, + tags: ["logs"], + middleware: [requireScope(stream.scope)] as never, + responses: { + 200: { description: stream.summary, content: { "text/event-stream": { schema: { type: "string" } } } }, + }, + }), + // SSE 响应不走 JSON 信封 + (async (c: Context) => { + const parsed = LogStreamQuery.safeParse(c.req.query()); + if (!parsed.success) throw new ApiException("bad_request", parsed.error.issues[0]?.message ?? "筛选参数无效"); + const { since, ...filter } = parsed.data; + const poller = createLogStreamPoller(queries(), filter as LogFilter, { + ...(since ? { since: new Date(since) } : {}), + }); + const log = ctx.logger.child({ component: "logs-stream" }); + return streamSSE(c, async (s) => { + let closed = false; + let wake: (() => void) | null = null; + s.onAbort(() => { + closed = true; + wake?.(); + }); + const keepalive = setInterval(() => { + void s.write(": keepalive\n\n").catch(() => { + closed = true; + }); + }, opts.keepaliveMs ?? 15_000); + await s.writeSSE({ event: "ready", data: JSON.stringify({ at: new Date().toISOString() }) }); + try { + while (!closed) { + let items: Awaited> = []; + try { + items = await poller.poll(); + } catch (err) { + // 库暂时不可达:这一轮跳过,下一轮再查 + log.debug({ err }, "日志流查询失败"); + } + for (const e of items) { + if (closed) break; + await s.writeSSE({ id: e.id, event: LOG_STREAM_EVENT, data: JSON.stringify(e) }).catch(() => { + closed = true; + }); + } + if (closed) break; + await new Promise((r) => { + const t = setTimeout(r, opts.pollMs ?? STREAM_POLL_MS); + wake = () => { + clearTimeout(t); + r(); + }; + }); + wake = null; + } + } finally { + clearInterval(keepalive); + } + }); + // biome-ignore lint/suspicious/noExplicitAny: 同上 + }) as any, + ); + }, + registerTasks() {}, + }; +} + +export const logsModule = createLogsModule(); diff --git a/apps/server/src/modules/logs/query.ts b/apps/server/src/modules/logs/query.ts new file mode 100644 index 00000000..76bcd321 --- /dev/null +++ b/apps/server/src/modules/logs/query.ts @@ -0,0 +1,311 @@ +import type { LogEntry, LogFilter, LogLevel, LogPage, LogSource, LogType } from "@umt/contracts"; +import { type DB, type Kysely, sql } from "@umt/db"; +import type { RawBuilder } from "kysely"; +import { redact } from "../../logger.ts"; +import { BUSINESS_EXPR } from "../auth-settings/engine-business.ts"; + +// 日志聚合查询(界面第二轮 V 线):四个来源各自出同一组列,按 (at, id) 倒序各取 limit+1 条,union 后再取前 limit 条。 +// 每个来源单独 order + limit,PG 能沿各自的时间索引倒着扫,不需要把全部来源物化再排序。 +// +// 来源与口径(docs/runbook/logs.md): +// - engine_run → task:引擎一次运行;时间 = 创建时间;标题 = 业务对象名(没有时为工作流名);对象 = 业务对象 +// - schedule_run → auto:定时任务运行记录(含智能体巡检);时间 = 开始时间;对象 = 定时任务 +// - agent_run(trigger=event,非子会话)→ auto:智能体后台调查;对象 = 会话 +// - audit_event → auth(admin.* / api_key.*)或 audit(其余);对象 = 审计目标 +// 例行记录(routine,缺省不列,system=1 或按对象筛选时列出):没有业务对象的引擎运行、定时任务触发且成功的运行、 +// 巡检 / 后台调查对应的引擎运行(已在 auto 里各有一条)、非巡检定时任务的成功 / 跳过运行。 + +export type Cursor = { us: number; id: string }; + +export function encodeCursor(c: Cursor): string { + return btoa(JSON.stringify([c.us, c.id])).replace(/=+$/, ""); +} + +export function decodeCursor(s: string | undefined | null): Cursor | null { + if (!s) return null; + try { + const [us, id] = JSON.parse(atob(s)) as [unknown, unknown]; + if (typeof us === "number" && Number.isFinite(us) && typeof id === "string") return { us, id }; + } catch { + // 无效游标当作第一页 + } + return null; +} + +/** 业务对象 kind → 日志对象类型(与审计 target_type 同名,对象筛选两边通用) */ +export const BUSINESS_OBJECT_TYPE: Record = { + download: "download_task", + organization: "organization_record", + agent: "agent_run", + subscription: "subscription", + feed: "feed_source", + site: "site", + decision: "decision", + media: "media_entity", + schedule: "scheduled_job", +}; + +const SOURCE_TYPES: Record = { + engine_run: ["task"], + schedule_run: ["auto"], + agent_run: ["auto"], + audit_event: ["auth", "audit"], +}; + +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`; + +type Row = { + id: string; + type: LogType; + source: LogSource; + ref_id: string; + at: Date; + finished_at: Date | null; + duration_ms: number | null; + level: LogLevel; + status: string | null; + title: string | null; + message: string | null; + object_type: string | null; + object_id: string | null; + object_label: string | null; + actor_type: string | null; + actor_id: string | null; + extra: Record | null; + cursor_us: string | number; +}; + +export interface LogQueryInput extends LogFilter { + cursor?: string | null; + limit: number; +} + +const escapeLike = (s: string) => s.replace(/[\\%_]/g, (c) => `\\${c}`); + +export interface LogQueries { + list(q: LogQueryInput): Promise; +} + +export function logQueries(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})`, + ); + const rows = await 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(db); + 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, + }; + }, + }; +} + +const iso = (d: Date | string | null) => (d == null ? null : (d instanceof Date ? d : new Date(d)).toISOString()); + +export function toEntry(r: Row): LogEntry { + const extra = (r.extra ?? {}) as Record; + return { + id: r.id, + type: r.type, + source: r.source, + refId: r.ref_id, + at: iso(r.at)!, + finishedAt: iso(r.finished_at), + durationMs: r.duration_ms == null ? null : Math.max(0, Math.round(Number(r.duration_ms))), + level: r.level, + status: r.status, + title: r.title || r.ref_id, + message: r.message || null, + object: r.object_type ? { type: r.object_type, id: r.object_id, label: r.object_label } : null, + actor: r.actor_type ? { type: r.actor_type, id: r.actor_id } : null, + // 审计负载可能带敏感字段:与审计接口一样脱敏后再给 + extra: r.source === "audit_event" ? (redact(extra) as Record) : extra, + }; +} diff --git a/apps/server/src/modules/logs/stream.ts b/apps/server/src/modules/logs/stream.ts new file mode 100644 index 00000000..e1a0c6f7 --- /dev/null +++ b/apps/server/src/modules/logs/stream.ts @@ -0,0 +1,56 @@ +import type { LogEntry, LogFilter } from "@umt/contracts"; +import type { LogQueries } from "./query.ts"; + +// 日志流(GET /logs/stream):按同一组筛选每隔 pollMs 查一次「回看窗口」内的最新日志, +// 没推过的推出去(新日志),推过但级别 / 状态 / 摘要变了的再推一次(客户端按 id 覆盖,如任务从进行中变成完成)。 +// 回看窗口 = 最近 lookbackMs(首次从 since 起),避免事务晚提交的记录(时间早于已推的最新一条)被漏掉。 + +export const STREAM_POLL_MS = 2_000; +export const STREAM_LOOKBACK_MS = 10 * 60_000; +const STREAM_BATCH = 200; + +const signature = (e: LogEntry) => `${e.level}|${e.status ?? ""}|${e.message ?? ""}|${e.title}|${e.finishedAt ?? ""}`; + +export interface LogStreamPoller { + /** 查一次,返回需要推送的条目(按时间正序,新的在后) */ + poll(): Promise; +} + +export function createLogStreamPoller( + queries: LogQueries, + filter: LogFilter, + o: { since?: Date; now?: () => number; lookbackMs?: number } = {}, +): LogStreamPoller { + const now = o.now ?? Date.now; + const lookbackMs = o.lookbackMs ?? STREAM_LOOKBACK_MS; + const since = (o.since ?? new Date(now())).getTime(); + /** id → 上次推送时的签名;窗口外的定期丢掉 */ + const seen = new Map(); + let first = true; + + return { + async poll() { + // 首次从 since 起(客户端列表最新一条之后的都要补上),之后只看回看窗口 + const windowStart = first ? Math.min(since, now() - lookbackMs) : now() - lookbackMs; + const from = filter.from ? Math.max(Date.parse(filter.from), windowStart) : windowStart; + const page = await queries.list({ ...filter, from: new Date(from).toISOString(), limit: STREAM_BATCH }); + const out: LogEntry[] = []; + for (const e of page.items) { + const at = Date.parse(e.at); + const sig = signature(e); + const prev = seen.get(e.id); + seen.set(e.id, { sig, at }); + // 首次查询:连接之前(since 之前)的日志客户端已有,只记签名不推 + if (!prev) { + if (first && at <= since) continue; + out.push(e); + } else if (prev.sig !== sig) out.push(e); + } + first = false; + // 窗口外的不会再查到,签名丢掉 + const cutoff = now() - lookbackMs; + for (const [id, v] of seen) if (v.at < cutoff) seen.delete(id); + return out.reverse(); + }, + }; +} diff --git a/apps/server/test/auth-settings/login.test.ts b/apps/server/test/auth-settings/login.test.ts index e03e5fb3..8e0fc4e5 100644 --- a/apps/server/test/auth-settings/login.test.ts +++ b/apps/server/test/auth-settings/login.test.ts @@ -81,6 +81,12 @@ describe.skipIf(!hasPg)("管理员登录门禁(真实 PG)", () => { it("登录:错密码 401;正确密码下发 cookie;会话可访问 admin 端点;未登录 401", async () => { const b = new Browser(h.app); expect((await login(b, "wrong-password-1")).status).toBe(401); + // 日志模块「登录」类:失败也留一条审计(用户名、IP、原因,不含密码) + const failed = await sql<{ actor_type: string; payload: Record }>` + select actor_type, payload from audit_event where action = 'admin.login_failed'`.execute(h.db); + expect(failed.rows).toHaveLength(1); + expect(failed.rows[0]!.payload).toMatchObject({ username: "admin", ip: "203.0.113.10" }); + expect(JSON.stringify(failed.rows[0]!.payload)).not.toContain("wrong-password-1"); expect((await b.req("GET", "/settings")).status).toBe(401); const res = await login(b); expect(res.status).toBe(200); diff --git a/apps/server/test/logs/logs.test.ts b/apps/server/test/logs/logs.test.ts new file mode 100644 index 00000000..4ee01ada --- /dev/null +++ b/apps/server/test/logs/logs.test.ts @@ -0,0 +1,299 @@ +import { LOG_STREAM_EVENT, type LogEntry, LogPage, LogQuery } from "@umt/contracts"; +import { sql } from "@umt/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createLogStreamPoller, createLogsModule, decodeCursor, encodeCursor } from "../../src/modules/logs/index.ts"; +import { logQueries } from "../../src/modules/logs/query.ts"; +import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; +import { DEV_KEY, json } from "../helpers.ts"; + +// 日志模块:四个来源合并、筛选、游标分页、流(真实 PG + 独立引擎 schema)。 + +const hasPg = await pgAvailable(); + +describe("游标", () => { + it("编码往返;无效游标当第一页", () => { + const c = { us: 1_759_000_000_123_456, id: "engine_run:abc" }; + expect(decodeCursor(encodeCursor(c))).toEqual(c); + expect(decodeCursor("not-base64!!")).toBeNull(); + expect(decodeCursor(undefined)).toBeNull(); + }); + + it("type / level 接受逗号分隔多值,未知值 422", () => { + expect(LogQuery.parse({ type: "task,auth", level: "error" })).toMatchObject({ + type: ["task", "auth"], + level: ["error"], + limit: 50, + }); + expect(LogQuery.safeParse({ type: "nope" }).success).toBe(false); + }); +}); + +describe.skipIf(!hasPg)("日志聚合(真实 PG)", () => { + let h: Harness; + let admin: ReturnType; + const base = Date.now() - 3600_000; + const at = (min: number) => new Date(base + min * 60_000); + const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`); + const run = (id: string, status: string, name: string, input: unknown, created: Date, error: unknown = null) => + sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, + attempts, worker_id, available_at, started_at, finished_at, created_at, updated_at, error) + values ('default', ${id}, ${name}, 'v1', ${status}, '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1, + null, ${created}, ${created}, ${status === "running" ? null : new Date(created.getTime() + 2000)}, ${created}, + ${created}, ${error ? JSON.stringify(error) : null}::jsonb)`.execute(h.db); + const list = async (query: string) => { + const res = await admin("GET", `/logs${query}`); + expect(res.status, query).toBe(200); + return LogPage.parse((await json(res)).data).items; + }; + + let downloadId = ""; + + beforeAll(async () => { + h = await createHarness({ + engine: true, + extraModules: [], + }); + const { createApp } = await import("../../src/http/app.ts"); + const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts"); + h.app = createApp({ + config: h.ctx.config, + logger: h.ctx.logger, + bus: h.bus, + db: h.db, + role: "all", + version: "0.0.0-test", + modules: [ + createAuthSettingsModule({ setupTokenOnStart: false, engineSchema: h.engineSchema }), + createLogsModule({ engineSchema: h.engineSchema, pollMs: 50 }), + ], + ctx: h.ctx, + }) as Harness["app"]; + admin = bearer(h.app, DEV_KEY); + + // 业务对象:下载任务 + const inst = await sql<{ id: string }>`insert into downloader_instance (name, base_url) + values ('qb', 'http://qb') returning id`.execute(h.db); + const dl = await sql<{ id: string }>`insert into download_task (instance_id, torrent_hash, name) + values (${inst.rows[0]!.id}, ${"a".repeat(40)}, '沙丘2 2160p') returning id`.execute(h.db); + downloadId = dl.rows[0]!.id; + + // 任务:有业务对象的失败 / 完成;没有业务对象的(例行) + await run("r-dl", "failed", "submitDownload", { input: { downloadTaskId: downloadId } }, at(1), { + message: "qB 拒绝了种子", + }); + await run("r-dl2", "completed", "verifyAndImport", { input: { downloadTaskId: downloadId } }, at(2)); + await run("r-sys", "completed", "snapshotSite", { input: {} }, at(3)); + await run("r-running", "running", "submitDownload", { input: { downloadTaskId: downloadId } }, at(4)); + + // 定时任务:一条失败、一条成功(例行)、一条巡检 + await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule) + values ('feeds.poll', 'RSS 抓取', 'collect', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb, + '{"type":"interval","everyMs":60000}'::jsonb), + ('check.daily', '每日巡检', 'agent', 'agent_check', true, true, '{"type":"interval","everyMs":86400000}'::jsonb, + '{"type":"interval","everyMs":86400000}'::jsonb)`.execute(h.db); + await sql`insert into scheduled_job_run (job_id, trigger, status, error, started_at, finished_at, duration_ms) values + ('feeds.poll', 'schedule', 'failed', '站点超时', ${at(5)}, ${at(5)}, 1200), + ('feeds.poll', 'schedule', 'ok', null, ${at(6)}, ${at(6)}, 300), + ('check.daily', 'schedule', 'ok', null, ${at(7)}, ${at(7)}, 9000)`.execute(h.db); + + // 智能体后台调查(trigger=event)与用户会话(不进日志的 auto) + await sql`insert into agent_run (trigger, goal, status, summary, created_at, started_at, finished_at) values + ('event', '调查下载为什么失败', 'succeeded', '种子已被站点删除', ${at(8)}, ${at(8)}, ${at(9)}), + ('user', '帮我找沙丘', 'succeeded', null, ${at(8)}, ${at(8)}, ${at(9)})`.execute(h.db); + + // 审计:登录成功、登录失败、设置修改、智能体操作被拒 + await sql`insert into audit_event (at, actor_type, actor_id, action, target_type, target_id, payload) values + (${at(10)}, 'user', 'admin', 'admin.login_success', 'admin_session', 's1', '{"ip":"10.0.0.2"}'::jsonb), + (${at(11)}, 'system', null, 'admin.login_failed', 'admin_user', null, '{"username":"admin","ip":"10.0.0.9"}'::jsonb), + (${at(12)}, 'user', 'admin', 'settings.update', 'settings', 'cache', '{"keys":["cache"],"token":"sk-secret"}'::jsonb), + (${at(13)}, 'agent', 'run-1', 'capability.denied', 'download_task', ${downloadId}, '{}'::jsonb)`.execute(h.db); + }, 60_000); + afterAll(async () => { + await h?.close(); + }); + + it("四个来源合并、按时间倒序;例行记录缺省不列", async () => { + const items = await list("?limit=100"); + expect(items.map((e) => e.id.split(":")[0])).toEqual( + expect.arrayContaining(["engine_run", "schedule_run", "agent_run", "audit_event"]), + ); + const times = items.map((e) => Date.parse(e.at)); + expect([...times].sort((a, b) => b - a)).toEqual(times); + const refs = items.map((e) => e.refId); + // 没有业务对象的引擎运行、成功的普通定时任务运行、用户会话都不在 + expect(refs).not.toContain("r-sys"); + expect(items.filter((e) => e.source === "schedule_run").map((e) => e.status)).toEqual(["ok", "failed"]); + expect(items.filter((e) => e.source === "agent_run")).toHaveLength(1); + // system=1 时列出 + const all = await list("?limit=100&system=1"); + expect(all.map((e) => e.refId)).toContain("r-sys"); + expect(all.filter((e) => e.source === "schedule_run")).toHaveLength(3); + }); + + it("各来源的类型、级别、标题、对象", async () => { + const items = await list("?limit=100"); + const by = (ref: string) => items.find((e) => e.refId === ref)!; + expect(by("r-dl")).toMatchObject({ + type: "task", + level: "error", + title: "沙丘2 2160p", + message: "qB 拒绝了种子", + object: { type: "download_task", id: downloadId, label: "沙丘2 2160p" }, + extra: { workflowName: "submitDownload", businessKind: "download" }, + }); + expect(by("r-dl").durationMs).toBe(2000); + expect(by("r-running")).toMatchObject({ level: "running", finishedAt: null }); + const failedRun = items.find((e) => e.source === "schedule_run" && e.status === "failed")!; + expect(failedRun).toMatchObject({ + type: "auto", + level: "error", + title: "RSS 抓取", + message: "站点超时", + object: { type: "scheduled_job", id: "feeds.poll" }, + durationMs: 1200, + }); + const check = items.find((e) => e.source === "schedule_run" && e.status === "ok")!; + expect(check).toMatchObject({ title: "每日巡检", level: "success" }); + expect(items.find((e) => e.source === "agent_run")).toMatchObject({ + type: "auto", + level: "success", + title: "调查下载为什么失败", + message: "种子已被站点删除", + }); + const audits = items.filter((e) => e.source === "audit_event"); + expect(audits.map((e) => [e.status, e.type, e.level])).toEqual([ + ["capability.denied", "audit", "warn"], + ["settings.update", "audit", "info"], + ["admin.login_failed", "auth", "warn"], + ["admin.login_success", "auth", "success"], + ]); + expect(audits.find((e) => e.status === "admin.login_failed")!.message).toBe("admin · 10.0.0.9"); + // 审计负载脱敏 + const settings = audits.find((e) => e.status === "settings.update")!; + expect(JSON.stringify(settings.extra)).not.toContain("sk-secret"); + }); + + it("筛选:类型、级别、关键词、对象、时间", async () => { + expect((await list("?type=auth")).map((e) => e.status)).toEqual(["admin.login_failed", "admin.login_success"]); + expect((await list("?type=auto")).every((e) => e.type === "auto")).toBe(true); + const errors = await list("?level=error"); + expect(errors.map((e) => e.refId).sort()).toEqual( + ["r-dl", errors.find((e) => e.source === "schedule_run")!.refId].sort(), + ); + expect((await list("?level=warn,running")).map((e) => e.level).sort()).toEqual(["running", "warn", "warn"]); + // 关键词:任务按业务标题、审计按负载、定时任务按错误 + expect((await list("?q=沙丘")).map((e) => e.refId)).toEqual(expect.arrayContaining(["r-dl", "r-dl2"])); + expect((await list("?q=10.0.0.9")).map((e) => e.status)).toEqual(["admin.login_failed"]); + expect((await list("?q=超时")).map((e) => e.source)).toEqual(["schedule_run"]); + // %、_ 按字面匹配 + expect(await list("?q=%25")).toEqual([]); + // 对象:同一个下载任务的任务与审计都列出 + const obj = await list(`?objectType=download_task&objectId=${downloadId}`); + expect(obj.map((e) => e.refId)).toEqual(expect.arrayContaining(["r-dl", "r-dl2", "r-running"])); + expect(obj.some((e) => e.status === "capability.denied")).toBe(true); + // 按对象筛选时自动包含例行记录 + const job = await list("?objectType=scheduled_job&objectId=feeds.poll"); + expect(job.map((e) => e.status)).toEqual(["ok", "failed"]); + // 时间 + const recent = await list(`?from=${encodeURIComponent(at(10).toISOString())}`); + expect(recent.every((e) => Date.parse(e.at) >= at(10).getTime())).toBe(true); + expect(recent.map((e) => e.source)).toEqual(Array(4).fill("audit_event")); + const older = await list(`?to=${encodeURIComponent(at(2).toISOString())}`); + expect(older.map((e) => e.refId)).toEqual(["r-dl"]); + }); + + it("游标分页:逐页拿到与一次取全相同的序列,不重不漏", async () => { + const full = await list("?limit=200&system=1"); + const seen: string[] = []; + let cursor: string | null = null; + for (let i = 0; i < 50; i++) { + const res = await admin("GET", `/logs?limit=3&system=1${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ""}`); + const page = LogPage.parse((await json(res)).data); + seen.push(...page.items.map((e) => e.id)); + cursor = page.nextCursor; + if (!cursor) break; + } + expect(seen).toEqual(full.map((e) => e.id)); + expect(new Set(seen).size).toBe(seen.length); + }); + + it("流:首轮不推已有日志;新日志推出;状态变化再推一次", async () => { + let now = Date.now(); + const poller = createLogStreamPoller( + logQueries(h.db, { engineSchema: h.engineSchema }), + {}, + { + since: new Date(now), + now: () => now, + }, + ); + expect(await poller.poll()).toEqual([]); + const created = new Date(now + 1000); + now += 2000; + await run("r-new", "running", "submitDownload", { input: { downloadTaskId: downloadId } }, created); + await sql`insert into audit_event (at, actor_type, action, payload) values (${new Date(now - 500)}, 'user', + 'api_key.create', '{}'::jsonb)`.execute(h.db); + const pushed = await poller.poll(); + expect(pushed.map((e) => e.refId === "r-new" || e.status === "api_key.create")).toEqual([true, true]); + // 正序:旧的在前 + expect(Date.parse(pushed[0]!.at)).toBeLessThanOrEqual(Date.parse(pushed[1]!.at)); + expect(await poller.poll()).toEqual([]); + await sql`update ${t("workflow_runs")} set status = 'completed', finished_at = ${new Date(now)} where id = 'r-new'`.execute( + h.db, + ); + const updated = await poller.poll(); + expect(updated).toHaveLength(1); + expect(updated[0]).toMatchObject({ refId: "r-new", level: "success" }); + }); + + it("流接口:SSE 推 log.entry,筛选生效", async () => { + const ctl = new AbortController(); + const since = new Date().toISOString(); + const res = await h.app.request(`/api/v1/logs/stream?type=auth&since=${encodeURIComponent(since)}`, { + headers: { authorization: `Bearer ${DEV_KEY}` }, + signal: ctl.signal, + }); + expect(res.status).toBe(200); + expect(res.headers.get("content-type")).toContain("text/event-stream"); + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + const readUntil = async (pred: (s: string) => boolean) => { + const deadline = Date.now() + 10_000; + while (!pred(buf) && Date.now() < deadline) { + const { value, done } = await reader.read(); + if (done) break; + buf += decoder.decode(value); + } + }; + await readUntil((s) => s.includes("event: ready")); + // 一条不匹配(审计)、一条匹配(登录) + await sql`insert into audit_event (actor_type, action, payload) values ('user', 'settings.update', '{}'::jsonb), + ('system', 'admin.login_failed', '{"username":"x","ip":"1.2.3.4"}'::jsonb)`.execute(h.db); + await readUntil((s) => s.includes(LOG_STREAM_EVENT) && s.includes("1.2.3.4")); + ctl.abort(); + await reader.cancel().catch(() => {}); + const events = buf + .split("\n\n") + .filter((b) => b.includes(`event: ${LOG_STREAM_EVENT}`)) + .map( + (b) => + JSON.parse( + b + .split("\n") + .find((l) => l.startsWith("data: "))! + .slice(6), + ) as LogEntry, + ); + expect(events.length).toBeGreaterThanOrEqual(1); + expect(events.every((e) => e.type === "auth")).toBe(true); + expect(events.some((e) => e.message === "x · 1.2.3.4")).toBe(true); + }); + + it("没有引擎 schema 时只查其它来源", async () => { + const q = logQueries(h.db, { engineSchema: "no_such_engine" }); + const page = await q.list({ limit: 50 }); + expect(page.items.length).toBeGreaterThan(0); + expect(page.items.some((e) => e.source === "engine_run")).toBe(false); + }); +}); diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts index 47c45dcb..f8e9181d 100644 --- a/apps/server/test/modules.test.ts +++ b/apps/server/test/modules.test.ts @@ -43,7 +43,7 @@ const pingModule: ServerModule = { }; describe("ServerModule 骨架", () => { - it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 浏览器 + 桌面反代 + MCP + 网页搜索 + 缓存 + 进程与健康(末尾),名字唯一", () => { + it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 浏览器 + 桌面反代 + MCP + 网页搜索 + 缓存 + 进程与健康 + 日志(末尾),名字唯一", () => { expect(SERVER_MODULES.map((m) => m.name)).toEqual([ "auth-settings", "sources-identity", @@ -60,6 +60,7 @@ describe("ServerModule 骨架", () => { "websearch", "cache", "system", + "logs", ]); });