feat(server): W5-A file manager module (/files)

Roots/list/stat with inline business status, path guard (realpath within registered roots, symlink escape, offline root 503), mkdir/rename syncing asset_location and path_binding revisions, two-phase move/copy/organize (planId + source facts), trash with reference checks and purgeTrash task, identify/probe/scrape, audit + idempotency keys. Runbook docs/runbook/files.md. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-09-27 23:02:20 +08:00
1 parent 09f6e3a3e5
commit 29cd10b27e
@@ -0,0 +1,159 @@
+import { sql } from "@umt/db";
+import type { Principal } from "../../http/auth.ts";
+import { actorOfP, FilesError, type FmDeps, hashJson, jsonb } from "./common.ts";
+
+// 计划冻结(artifact kind=plan, title=files_plan)、幂等键(artifact title=files_idem)、审计与 SSE。
+
+export interface StoredPlan<C> {
+  id: string;
+  content: C & { op: string; requestHash: string; actor: string };
+  contentHash: string;
+  expiresAt: Date;
+}
+
+/** 冻结 dryRun 计划;content.request 与 organize 路由同形时,一次性审批(plan_type=organize_plan)可直接绑定 */
+export async function savePlan<C extends object>(
+  d: FmDeps,
+  op: string,
+  requestHash: string,
+  principal: Principal | undefined,
+  content: C,
+): Promise<{ id: string; expiresAt: Date; contentHash: string }> {
+  const expiresAt = new Date(d.ao.ports.now().getTime() + d.options.planTtlMs);
+  const full = { ...content, op, requestHash, actor: actorOfP(principal) };
+  const contentHash = hashJson(full);
+  const row = await d.ctx.db
+    .insertInto("artifact")
+    .values({
+      kind: "plan",
+      subject_type: `files_${op}`,
+      title: "files_plan",
+      content: jsonb(full),
+      content_hash: contentHash,
+      expires_at: expiresAt,
+    })
+    .returning("id")
+    .executeTakeFirstOrThrow();
+  return { id: row.id, expiresAt, contentHash };
+}
+
+/** 取执行用计划:不存在 / 过期 / 操作或请求不符 → 409 plan_stale */
+export async function loadPlan<C>(d: FmDeps, planId: string, op: string, requestHash: string): Promise<StoredPlan<C>> {
+  const row = await d.ctx.db
+    .selectFrom("artifact")
+    .select(["id", "content", "content_hash", "expires_at", "title", "kind"])
+    .where("id", "=", planId)
+    .executeTakeFirst();
+  if (row?.kind !== "plan" || row.title !== "files_plan")
+    throw new FilesError("plan_stale", "planId 不存在,请重新预览");
+  const content = row.content as StoredPlan<C>["content"];
+  if (content.op !== op) throw new FilesError("plan_stale", `planId 属于 ${content.op} 操作,不能用于 ${op}`);
+  if (!row.expires_at || row.expires_at.getTime() <= d.ao.ports.now().getTime())
+    throw new FilesError("plan_stale", "计划已过期,请重新预览");
+  if (content.requestHash !== requestHash) throw new FilesError("plan_stale", "请求内容与预览时不一致,请重新预览");
+  return { id: row.id, content, contentHash: row.content_hash, expiresAt: row.expires_at };
+}
+
+/** 计划用完即作废(防重放执行) */
+export async function retirePlan(d: FmDeps, planId: string): Promise<void> {
+  await d.ctx.db.updateTable("artifact").set({ expires_at: d.ao.ports.now() }).where("id", "=", planId).execute();
+}
+
+/** API Key 的一次性审批:绑定该计划 id 与内容哈希,消费后标 consumed */
+export async function consumeApproval(d: FmDeps, planId: string, contentHash: string): Promise<string | null> {
+  const r = await sql<{ id: string }>`
+    update approval set status = 'consumed', updated_at = now()
+     where id = (
+       select id from approval
+        where plan_id = ${planId} and status = 'approved' and payload_hash = ${contentHash}
+          and plan_type in ('files_plan', 'organize_plan')
+          and (expires_at is null or expires_at > now())
+        order by created_at desc limit 1)
+       and status = 'approved'
+    returning id`.execute(d.ctx.db);
+  return r.rows[0]?.id ?? null;
+}
+
+// ── 幂等 ──
+
+/**
+ * 同一调用者 + 操作 + 幂等键:首次执行并存结果;重放返回首次结果;
+ * 同键不同请求体 → 409;首次仍在执行 → 409(稍后重试)。执行失败不占键。
+ */
+export async function idempotent<T>(
+  d: FmDeps,
+  principal: Principal | undefined,
+  op: string,
+  key: string,
+  request: unknown,
+  fn: () => Promise<T>,
+): Promise<T> {
+  const db = d.ctx.db;
+  const k = `${op}:${actorOfP(principal)}:${key}`;
+  const reqHash = hashJson(request);
+  const replay = async (): Promise<T | undefined> => {
+    const row = await db
+      .selectFrom("artifact")
+      .select(["content"])
+      .where("title", "=", "files_idem")
+      .where(sql<boolean>`content->>'key' = ${k}`)
+      .executeTakeFirst();
+    if (!row) return undefined;
+    const c = row.content as { reqHash: string; state: string; result?: T };
+    if (c.reqHash !== reqHash) throw new FilesError("exists", "idempotencyKey 已用于不同的请求");
+    if (c.state !== "done") throw new FilesError("exists", "同一 idempotencyKey 的请求仍在执行,请稍后重试");
+    return c.result as T;
+  };
+  const prev = await replay();
+  if (prev !== undefined) return prev;
+  const ins = await sql<{ id: string }>`
+    insert into artifact (kind, subject_type, title, content, content_hash, expires_at)
+    values ('log', ${`files_${op}`}, 'files_idem', ${jsonb({ key: k, reqHash, state: "pending" })}::jsonb,
+            ${hashJson(k)}, now() + interval '7 days')
+    on conflict ((content->>'key')) where title = 'files_idem' do nothing
+    returning id`.execute(db);
+  const id = ins.rows[0]?.id;
+  if (!id) {
+    const again = await replay();
+    if (again !== undefined) return again;
+    throw new FilesError("exists", "同一 idempotencyKey 的请求仍在执行,请稍后重试");
+  }
+  try {
+    const result = await fn();
+    await db
+      .updateTable("artifact")
+      .set({ content: jsonb({ key: k, reqHash, state: "done", result }) })
+      .where("id", "=", id)
+      .execute();
+    return result;
+  } catch (err) {
+    await db.deleteFrom("artifact").where("id", "=", id).execute();
+    throw err;
+  }
+}
+
+// ── 审计 / 事件 ──
+
+export async function audit(
+  d: FmDeps,
+  principal: Principal | undefined,
+  op: string,
+  payload: Record<string, unknown>,
+): Promise<void> {
+  await d.ctx.db
+    .insertInto("audit_event")
+    .values({
+      actor_type: principal?.kind === "api_key" ? "api_key" : "user",
+      actor_id: actorOfP(principal),
+      action: `files.${op}`,
+      target_type: "file",
+      target_id: null,
+      payload: jsonb(payload),
+    })
+    .execute();
+}
+
+/** SSE `files.changed`:受影响的根与目录(相对路径),前端据此刷新 */
+export function filesChanged(d: FmDeps, rootId: string, dirs: string[]): void {
+  void d.ctx.events.publish("files.changed", { rootId, dirs: [...new Set(dirs)] });
+}