diff --git a/src/cli/journal.test.ts b/src/cli/journal.test.ts new file mode 100644 index 0000000..563b50d --- /dev/null +++ b/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); + }); +}); diff --git a/src/cli/journal.ts b/src/cli/journal.ts index c841487..2cafdcc 100644 --- a/src/cli/journal.ts +++ b/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//meta.json` 存在且可解析;不再维护中心化清单, + * 会话列表在内存中派生并通过保留主题 `ttrpg/$SESSIONS` 发布。 + * 元数据写入使用「临时文件 + fsync + 原子重命名」,避免崩溃时 + * 截断损坏;损坏的元数据会被隔离而非静默丢弃。 */ import { join, resolve } from "path"; import { appendFileSync, + closeSync, existsSync, + fsyncSync, mkdirSync, + openSync, readFileSync, - rmdirSync, - unlinkSync, - writeFileSync, + readdirSync, + readSync, + 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 { - name: string; - created: number; - players: string[]; -} +const SessionMetaSchema = z.object({ + name: z.string().min(1), + created: z.number().default(() => Date.now()), + players: z.array(z.string()).default([]), +}); + +type SessionMeta = z.infer; interface SessionManifest { sessions: Record; @@ -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 { - return join(dataRoot, "manifest.json"); -} - -function sessionDir(dataRoot: string, id: string): string { +export 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")); - } catch { - return { sessions: {} }; + 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 saveManifest(dataRoot: string, m: SessionManifest): void { - writeFileSync(manifestPath(dataRoot), JSON.stringify(m, null, 2), "utf-8"); +// --------------------------------------------------------------------------- +// 会话扫描 / 迁移 +// --------------------------------------------------------------------------- + +function safeMtime(path: string): number { + try { + return statSync(path).mtimeMs; + } catch { + return Date.now(); + } } -function appendStream(dataRoot: string, id: string, line: string): void { - const sd = sessionDir(dataRoot, id); - mkdirSync(sd, { recursive: true }); - appendFileSync(streamPath(dataRoot, id), line + "\n", "utf-8"); +/** + * 扫描 sessions/ 目录,构建内存索引。 + * + * - 有合法 meta.json 的目录 → 直接纳入索引; + * - meta.json 损坏 → 隔离为 meta.json.corrupt-,再尝试接管; + * - 无 meta.json 但有 stream.jsonl → 接管(用目录名与流文件 mtime 合成元数据); + * - 其余目录 → 忽略。 + */ +export function scanSessions(dataRoot: string): Record { + const sessionsRoot = join(dataRoot, "sessions"); + mkdirSync(sessionsRoot, { recursive: true }); + + const index: Record = {}; + + 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); + } + } + } + + if (existsSync(sp)) { + const meta: SessionMeta = { + name: id, + 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//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, +): 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); - broker.publish( - { - topic: $SESSIONS, - payload: Buffer.from(JSON.stringify(initialManifest)), - qos: 1, - retain: true, - cmd: "publish" as const, - dup: false, - }, - () => {}, - ); + publishManifest(broker, index); console.log("[日志] 持久化监听已启动"); console.log("[日志] MQTT 代理已就绪:ws://:"); @@ -191,50 +370,52 @@ function extractSessionId(topic: string): string | null { // 会话生命周期管理 // --------------------------------------------------------------------------- +/** + * 追加一条流记录。会话必须已在索引中(即存在合法 meta.json), + * 否则丢弃——这样已删除的会话不会被迟到的消息「复活」。 + * 目录由元数据处理函数创建,此处不再 mkdir。 + */ +export function appendStream( + root: string, + id: string, + index: Record, + 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, rawPayload: string, ): void { - const manifest = loadManifest(root); try { - const meta = JSON.parse(rawPayload) as SessionMeta | null; + const raw = JSON.parse(rawPayload) as Partial | null; - if (!meta || !meta.name) { - if (manifest.sessions[sessionId]) { - delete manifest.sessions[sessionId]; - const sp = streamPath(root, sessionId); - if (existsSync(sp)) unlinkSync(sp); - const sd = sessionDir(root, sessionId); - try { - rmdirSync(sd); - } catch { - /* */ - } - console.log(`[日志] 会话已删除:${sessionId}`); - } + if (!raw || !raw.name) { + // 空载荷 = 删除会话:递归删除目录,移除索引项 + delete index[sessionId]; + rmSync(sessionDir(root, sessionId), { recursive: true, force: true }); + console.log(`[日志] 会话已删除:${sessionId}`); } else { - manifest.sessions[sessionId] = { - name: meta.name, - created: meta.created || Date.now(), - players: meta.players || [], + const meta: SessionMeta = { + name: raw.name, + created: raw.created || Date.now(), + 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); - broker.publish( - { - topic: $SESSIONS, - payload: Buffer.from(JSON.stringify(manifest)), - qos: 1, - retain: true, - cmd: "publish" as const, - dup: false, - }, - () => {}, - ); + publishManifest(broker, index); } catch (e) { console.error(`[日志] 会话 ${sessionId} 元数据解析失败:`, e); }