fix(journal): make session dirs source of truth

Replaces the central manifest.json with per-session meta.json so the filesystem is the single source of truth. Metadata writes are now atomic (temp file + fsync + rename), startup adopts orphan dirs, corrupt meta is quarantined instead of silently wiped, truncated stream tails are repaired, and appends to unknown sessions are dropped so deleted sessions cannot be resurrected.

Existing manifests are migrated to per-session metadata on first run.

src/cli/journal.test.ts

@@ -0,0 +1,143 @@
+import {
+  existsSync,
+  mkdirSync,
+  mkdtempSync,
+  readFileSync,
+  readdirSync,
+  rmSync,
+  writeFileSync,
+} from "fs";
+import { tmpdir } from "os";
+import { join } from "path";
+import {
+  appendStream,
+  metaPath,
+  migrateLegacyManifest,
+  repairStreamTail,
+  scanSessions,
+  sessionDir,
+  streamPath,
+  writeFileAtomic,
+} from "./journal";
+
+function makeRoot(): string {
+  return mkdtempSync(join(tmpdir(), "ttrpg-journal-"));
+}
+
+function cleanup(root: string): void {
+  rmSync(root, { recursive: true, force: true });
+}
+
+describe("writeFileAtomic", () => {
+  test("writes content and leaves no temp file behind", () => {
+    const root = makeRoot();
+    const p = join(root, "meta.json");
+    writeFileAtomic(p, '{"a":1}');
+    expect(readFileSync(p, "utf-8")).toBe('{"a":1}');
+    expect(readdirSync(root)).toEqual(["meta.json"]);
+    cleanup(root);
+  });
+});
+
+describe("repairStreamTail", () => {
+  test("truncates a partial trailing line", () => {
+    const root = makeRoot();
+    const p = join(root, "stream.jsonl");
+    writeFileSync(p, '{"a":1}\n{"b":2');
+    repairStreamTail(p);
+    expect(readFileSync(p, "utf-8")).toBe('{"a":1}\n');
+    cleanup(root);
+  });
+
+  test("leaves a well-formed file untouched", () => {
+    const root = makeRoot();
+    const p = join(root, "stream.jsonl");
+    writeFileSync(p, '{"a":1}\n');
+    repairStreamTail(p);
+    expect(readFileSync(p, "utf-8")).toBe('{"a":1}\n');
+    cleanup(root);
+  });
+});
+
+describe("scanSessions", () => {
+  test("adopts a directory that has a stream but no meta", () => {
+    const root = makeRoot();
+    mkdirSync(sessionDir(root, "default"), { recursive: true });
+    writeFileSync(streamPath(root, "default"), '{"id":"gm-1"}\n');
+
+    const index = scanSessions(root);
+
+    expect(index["default"]).toMatchObject({ name: "default", players: [] });
+    expect(existsSync(metaPath(root, "default"))).toBe(true);
+    cleanup(root);
+  });
+
+  test("quarantines corrupt meta and adopts the session", () => {
+    const root = makeRoot();
+    const dir = sessionDir(root, "broken");
+    mkdirSync(dir, { recursive: true });
+    writeFileSync(metaPath(root, "broken"), "{ not json");
+    writeFileSync(streamPath(root, "broken"), '{"id":"gm-1"}\n');
+
+    const index = scanSessions(root);
+
+    expect(index["broken"]).toBeDefined();
+    const quarantined = readdirSync(dir).filter((f) =>
+      f.startsWith("meta.json.corrupt-"),
+    );
+    expect(quarantined).toHaveLength(1);
+    cleanup(root);
+  });
+
+  test("keeps a valid session that has no stream yet", () => {
+    const root = makeRoot();
+    mkdirSync(sessionDir(root, "test"), { recursive: true });
+    writeFileAtomic(
+      metaPath(root, "test"),
+      JSON.stringify({ name: "test", created: 1, players: [] }),
+    );
+
+    const index = scanSessions(root);
+
+    expect(index["test"]).toEqual({ name: "test", created: 1, players: [] });
+    cleanup(root);
+  });
+});
+
+describe("migrateLegacyManifest", () => {
+  test("splits a legacy manifest into per-session meta and renames it", () => {
+    const root = makeRoot();
+    writeFileSync(
+      join(root, "manifest.json"),
+      JSON.stringify({
+        sessions: {
+          aaa: { name: "aaa", created: 1, players: [] },
+          bbb: { name: "bbb", created: 2, players: [] },
+        },
+      }),
+    );
+
+    migrateLegacyManifest(root);
+
+    expect(existsSync(join(root, "manifest.json"))).toBe(false);
+    expect(existsSync(join(root, "manifest.json.migrated"))).toBe(true);
+    expect(existsSync(metaPath(root, "aaa"))).toBe(true);
+    expect(existsSync(metaPath(root, "bbb"))).toBe(true);
+    cleanup(root);
+  });
+});
+
+describe("appendStream", () => {
+  test("drops unknown sessions and appends known ones", () => {
+    const root = makeRoot();
+    const index = { known: { name: "known", created: 1, players: [] } };
+    mkdirSync(sessionDir(root, "known"), { recursive: true });
+
+    appendStream(root, "ghost", index, '{"x":1}');
+    expect(existsSync(streamPath(root, "ghost"))).toBe(false);
+
+    appendStream(root, "known", index, '{"x":1}');
+    expect(readFileSync(streamPath(root, "known"), "utf-8")).toBe('{"x":1}\n');
+    cleanup(root);
+  });
+});

src/cli/journal.ts

@@ -1,5 +1,5 @@
 /**
- * 日志流持久化 — MQTT 代理 + JSONL 追加写入 + 会话清单
+ * 日志流持久化 — MQTT 代理 + JSONL 追加写入 + 按会话元数据
  *
  * 在现有 HTTP 服务器上挂载 aedes MQTT 代理（WebSocket 传输），
  * 使用 ws v8 内置的 createWebSocketStream。
@@ -9,31 +9,46 @@
  *
  * 持久化使用 aedes 内部 subscribe/publish API，无需服务器进程
  * 以 MQTT 客户端身份连接自身。
+ *
+ * 存储模型：目录即事实来源。会话存在当且仅当
+ * `sessions/<id>/meta.json` 存在且可解析；不再维护中心化清单，
+ * 会话列表在内存中派生并通过保留主题 `ttrpg/$SESSIONS` 发布。
+ * 元数据写入使用「临时文件 + fsync + 原子重命名」，避免崩溃时
+ * 截断损坏；损坏的元数据会被隔离而非静默丢弃。
  */
 
 import { join, resolve } from "path";
 import {
   appendFileSync,
+  closeSync,
   existsSync,
+  fsyncSync,
   mkdirSync,
+  openSync,
   readFileSync,
-  rmdirSync,
+  readdirSync,
-  unlinkSync,
+  readSync,
-  writeFileSync,
+  renameSync,
+  rmSync,
+  statSync,
+  truncateSync,
+  writeSync,
 } from "fs";
 import type { Server as HttpServer } from "http";
-import type { AedesPublishPacket } from "aedes";
 import { WebSocketServer, createWebSocketStream } from "ws";
+import { z } from "zod";
 
 // ---------------------------------------------------------------------------
 // 类型定义
 // ---------------------------------------------------------------------------
 
-interface SessionMeta {
+const SessionMetaSchema = z.object({
-  name: string;
+  name: z.string().min(1),
-  created: number;
+  created: z.number().default(() => Date.now()),
-  players: string[];
+  players: z.array(z.string()).default([]),
-}
+});
+
+type SessionMeta = z.infer<typeof SessionMetaSchema>;
 
 interface SessionManifest {
   sessions: Record<string, SessionMeta>;
@@ -49,42 +64,188 @@ export interface JournalServer {
 // 路径工具函数
 // ---------------------------------------------------------------------------
 
-function dataDir(contentDir: string): string {
+export function dataDir(contentDir: string): string {
   const d = resolve(contentDir, ".ttrpg");
   mkdirSync(d, { recursive: true });
   return d;
 }
 
-function manifestPath(dataRoot: string): string {
+export function sessionDir(dataRoot: string, id: string): string {
-  return join(dataRoot, "manifest.json");
-}
-
-function sessionDir(dataRoot: string, id: string): string {
   return join(dataRoot, "sessions", id);
 }
 
-function streamPath(dataRoot: string, id: string): string {
+export function metaPath(dataRoot: string, id: string): string {
+  return join(sessionDir(dataRoot, id), "meta.json");
+}
+
+export function streamPath(dataRoot: string, id: string): string {
   return join(sessionDir(dataRoot, id), "stream.jsonl");
 }
 
-function loadManifest(dataRoot: string): SessionManifest {
+// ---------------------------------------------------------------------------
-  const p = manifestPath(dataRoot);
+// 原子写入
-  if (!existsSync(p)) return { sessions: {} };
+// ---------------------------------------------------------------------------
+
+let _tmpCounter = 0;
+
+/**
+ * 原子写入：先写临时文件并 fsync，再重命名覆盖目标。
+ * rename 在同一文件系统上是原子操作（Windows 上由 libuv 以
+ * MOVEFILE_REPLACE_EXISTING 实现），因此崩溃时目标文件要么是旧内容，
+ * 要么是新内容，绝不会是截断的中间态。
+ */
+export function writeFileAtomic(path: string, data: string): void {
+  const tmp = `${path}.tmp-${process.pid}-${Date.now()}-${_tmpCounter++}`;
+  const fd = openSync(tmp, "w");
   try {
-    return JSON.parse(readFileSync(p, "utf-8"));
+    writeSync(fd, data);
+    fsyncSync(fd);
+  } finally {
+    closeSync(fd);
+  }
+  renameSync(tmp, path);
+}
+
+// ---------------------------------------------------------------------------
+// 流文件修复
+// ---------------------------------------------------------------------------
+
+/**
+ * 修复崩溃时被截断的尾行：若文件不以换行结尾，则截断到最后一个换行。
+ * 否则后续追加会与残行粘连，产生一条损坏记录。
+ */
+export function repairStreamTail(path: string): void {
+  if (!existsSync(path)) return;
+  try {
+    const size = statSync(path).size;
+    if (size === 0) return;
+
+    const fd = openSync(path, "r");
+    const last = Buffer.alloc(1);
+    try {
+      readSync(fd, last, 0, 1, size - 1);
+    } finally {
+      closeSync(fd);
+    }
+    if (last[0] === 0x0a) return;
+
+    const buf = readFileSync(path);
+    const lastNl = buf.lastIndexOf(0x0a);
+    truncateSync(path, lastNl === -1 ? 0 : lastNl + 1);
+    console.warn(`[日志] 已修复截断的流尾行：${path}`);
+  } catch (e) {
+    console.error(`[日志] 修复流文件失败：${path}`, e);
+  }
+}
+
+// ---------------------------------------------------------------------------
+// 会话扫描 / 迁移
+// ---------------------------------------------------------------------------
+
+function safeMtime(path: string): number {
+  try {
+    return statSync(path).mtimeMs;
   } catch {
-    return { sessions: {} };
+    return Date.now();
   }
 }
 
-function saveManifest(dataRoot: string, m: SessionManifest): void {
+/**
-  writeFileSync(manifestPath(dataRoot), JSON.stringify(m, null, 2), "utf-8");
+ * 扫描 sessions/ 目录，构建内存索引。
+ *
+ * - 有合法 meta.json 的目录 → 直接纳入索引；
+ * - meta.json 损坏 → 隔离为 meta.json.corrupt-<ts>，再尝试接管；
+ * - 无 meta.json 但有 stream.jsonl → 接管（用目录名与流文件 mtime 合成元数据）；
+ * - 其余目录 → 忽略。
+ */
+export function scanSessions(dataRoot: string): Record<string, SessionMeta> {
+  const sessionsRoot = join(dataRoot, "sessions");
+  mkdirSync(sessionsRoot, { recursive: true });
+
+  const index: Record<string, SessionMeta> = {};
+
+  for (const entry of readdirSync(sessionsRoot, { withFileTypes: true })) {
+    if (!entry.isDirectory()) continue;
+    const id = entry.name;
+    const mp = metaPath(dataRoot, id);
+    const sp = streamPath(dataRoot, id);
+
+    if (existsSync(mp)) {
+      try {
+        const meta = SessionMetaSchema.parse(
+          JSON.parse(readFileSync(mp, "utf-8")),
+        );
+        index[id] = meta;
+        repairStreamTail(sp);
+        continue;
+      } catch (e) {
+        const quarantine = `${mp}.corrupt-${Date.now()}`;
+        try {
+          renameSync(mp, quarantine);
+          console.error(
+            `[日志] 会话 ${id} 元数据损坏，已隔离到 ${quarantine}：`,
+            e,
+          );
+        } catch (renameErr) {
+          console.error(`[日志] 会话 ${id} 元数据损坏且隔离失败：`, renameErr);
+        }
+      }
     }
 
-function appendStream(dataRoot: string, id: string, line: string): void {
+    if (existsSync(sp)) {
-  const sd = sessionDir(dataRoot, id);
+      const meta: SessionMeta = {
-  mkdirSync(sd, { recursive: true });
+        name: id,
-  appendFileSync(streamPath(dataRoot, id), line + "\n", "utf-8");
+        created: safeMtime(sp),
+        players: [],
+      };
+      try {
+        writeFileAtomic(mp, JSON.stringify(meta, null, 2));
+        index[id] = meta;
+        repairStreamTail(sp);
+        console.log(`[日志] 已接管孤立会话：${id}`);
+      } catch (e) {
+        console.error(`[日志] 接管孤立会话失败：${id}`, e);
+      }
+    }
+  }
+
+  return index;
+}
+
+/**
+ * 迁移旧的中心化 manifest.json：把每个会话条目拆分为
+ * sessions/<id>/meta.json，然后把旧文件重命名为 manifest.json.migrated。
+ * 已存在 meta.json 的会话会被跳过（支持中断后重跑）。
+ */
+export function migrateLegacyManifest(dataRoot: string): void {
+  const legacy = join(dataRoot, "manifest.json");
+  if (!existsSync(legacy)) return;
+
+  try {
+    const parsed = JSON.parse(readFileSync(legacy, "utf-8")) as SessionManifest;
+    const sessions = parsed?.sessions ?? {};
+    let migrated = 0;
+
+    for (const [id, raw] of Object.entries(sessions)) {
+      const mp = metaPath(dataRoot, id);
+      if (existsSync(mp)) continue;
+      try {
+        const meta = SessionMetaSchema.parse(raw);
+        mkdirSync(sessionDir(dataRoot, id), { recursive: true });
+        writeFileAtomic(mp, JSON.stringify(meta, null, 2));
+        migrated++;
+      } catch (e) {
+        console.error(`[日志] 跳过无法迁移的会话 ${id}：`, e);
+      }
+    }
+
+    renameSync(legacy, `${legacy}.migrated`);
+    console.log(
+      `[日志] 已迁移 ${migrated} 个会话到按会话元数据；旧清单重命名为 manifest.json.migrated`,
+    );
+  } catch (e) {
+    console.error("[日志] 旧清单迁移失败，保留原文件：", e);
+  }
 }
 
 // ---------------------------------------------------------------------------
@@ -93,6 +254,24 @@ function appendStream(dataRoot: string, id: string, line: string): void {
 
 const $SESSIONS = "ttrpg/$SESSIONS";
 
+function publishManifest(
+  broker: import("aedes").Aedes,
+  index: Record<string, SessionMeta>,
+): void {
+  const manifest: SessionManifest = { sessions: index };
+  broker.publish(
+    {
+      topic: $SESSIONS,
+      payload: Buffer.from(JSON.stringify(manifest)),
+      qos: 1,
+      retain: true,
+      cmd: "publish" as const,
+      dup: false,
+    },
+    () => {},
+  );
+}
+
 export async function createJournalServer(
   contentDir: string,
   httpServer: HttpServer,
@@ -100,6 +279,11 @@ export async function createJournalServer(
   const root = dataDir(contentDir);
   console.log(`[日志] 数据目录：${root}`);
 
+  // 迁移旧清单并扫描现有会话，构建内存索引（目录即事实来源）
+  migrateLegacyManifest(root);
+  const index = scanSessions(root);
+  console.log(`[日志] 已加载 ${Object.keys(index).length} 个会话`);
+
   // ---- MQTT 代理 ----
   const { Aedes: AedesFactory } = await import("aedes");
   // aedes 需要先 listen() 初始化持久化，再处理连接
@@ -128,7 +312,7 @@ export async function createJournalServer(
       const sessionId = extractSessionId(packet.topic);
       if (sessionId) {
         try {
-          appendStream(root, sessionId, packet.payload.toString());
+          appendStream(root, sessionId, index, packet.payload.toString());
         } catch (e) {
           console.error(`[日志] 会话 ${sessionId} 流记录写入失败：`, e);
         }
@@ -143,7 +327,13 @@ export async function createJournalServer(
     (packet, cb) => {
       const sessionId = extractSessionId(packet.topic);
       if (sessionId) {
-        handleMetaChange(root, sessionId, broker, packet.payload.toString());
+        handleMetaChange(
+          root,
+          sessionId,
+          broker,
+          index,
+          packet.payload.toString(),
+        );
       }
       cb();
     },
@@ -151,18 +341,7 @@ export async function createJournalServer(
   );
 
   // 启动时发布初始清单，让客户端能看到已有会话
-  const initialManifest = loadManifest(root);
+  publishManifest(broker, index);
-  broker.publish(
-    {
-      topic: $SESSIONS,
-      payload: Buffer.from(JSON.stringify(initialManifest)),
-      qos: 1,
-      retain: true,
-      cmd: "publish" as const,
-      dup: false,
-    },
-    () => {},
-  );
 
   console.log("[日志] 持久化监听已启动");
   console.log("[日志] MQTT 代理已就绪：ws://<host>:<port>");
@@ -191,50 +370,52 @@ function extractSessionId(topic: string): string | null {
 // 会话生命周期管理
 // ---------------------------------------------------------------------------
 
+/**
+ * 追加一条流记录。会话必须已在索引中（即存在合法 meta.json），
+ * 否则丢弃——这样已删除的会话不会被迟到的消息「复活」。
+ * 目录由元数据处理函数创建，此处不再 mkdir。
+ */
+export function appendStream(
+  root: string,
+  id: string,
+  index: Record<string, SessionMeta>,
+  line: string,
+): void {
+  if (!index[id]) {
+    console.warn(`[日志] 丢弃未知会话 ${id} 的流记录`);
+    return;
+  }
+  appendFileSync(streamPath(root, id), line + "\n", "utf-8");
+}
+
 function handleMetaChange(
   root: string,
   sessionId: string,
   broker: import("aedes").Aedes,
+  index: Record<string, SessionMeta>,
   rawPayload: string,
 ): void {
-  const manifest = loadManifest(root);
   try {
-    const meta = JSON.parse(rawPayload) as SessionMeta | null;
+    const raw = JSON.parse(rawPayload) as Partial<SessionMeta> | null;
 
-    if (!meta || !meta.name) {
+    if (!raw || !raw.name) {
-      if (manifest.sessions[sessionId]) {
+      // 空载荷 = 删除会话：递归删除目录，移除索引项
-        delete manifest.sessions[sessionId];
+      delete index[sessionId];
-        const sp = streamPath(root, sessionId);
+      rmSync(sessionDir(root, sessionId), { recursive: true, force: true });
-        if (existsSync(sp)) unlinkSync(sp);
-        const sd = sessionDir(root, sessionId);
-        try {
-          rmdirSync(sd);
-        } catch {
-          /* */
-        }
       console.log(`[日志] 会话已删除：${sessionId}`);
-      }
     } else {
-      manifest.sessions[sessionId] = {
+      const meta: SessionMeta = {
-        name: meta.name,
+        name: raw.name,
-        created: meta.created || Date.now(),
+        created: raw.created || Date.now(),
-        players: meta.players || [],
+        players: raw.players || [],
       };
+      mkdirSync(sessionDir(root, sessionId), { recursive: true });
+      writeFileAtomic(metaPath(root, sessionId), JSON.stringify(meta, null, 2));
+      index[sessionId] = meta;
       console.log(`[日志] 会话已更新：${sessionId} (${meta.name})`);
     }
 
-    saveManifest(root, manifest);
+    publishManifest(broker, index);
-    broker.publish(
-      {
-        topic: $SESSIONS,
-        payload: Buffer.from(JSON.stringify(manifest)),
-        qos: 1,
-        retain: true,
-        cmd: "publish" as const,
-        dup: false,
-      },
-      () => {},
-    );
   } catch (e) {
     console.error(`[日志] 会话 ${sessionId} 元数据解析失败：`, e);
   }