feat: add MQTT journal server for session persistence
Implement a journal server that uses an Aedes MQTT broker to persist stream messages to JSONL files and manage session metadata via a manifest. - Add `aedes`, `mqtt`, and `zod` dependencies - Implement `createJournalServer` in `src/cli/journal.ts` - Integrate journal server into the `serve` command - Add `--mqtt-port` option to the CLI - Add session lifecycle management (create, update, delete) via retained MQTT topics
This commit is contained in:
@@ -5,6 +5,7 @@ import { createReadStream } from "fs";
|
||||
import { join, resolve, extname, sep, relative, dirname } from "path";
|
||||
import { watch } from "chokidar";
|
||||
import { fileURLToPath } from "url";
|
||||
import { createJournalServer } from "../journal.js";
|
||||
|
||||
interface ContentIndex {
|
||||
[path: string]: string;
|
||||
@@ -63,7 +64,11 @@ export function scanDirectory(dir: string): ContentIndex {
|
||||
const stats = statSync(fullPath);
|
||||
if (stats.isDirectory()) {
|
||||
scan(fullPath, relPath);
|
||||
} else if (entry.endsWith(".md") || entry.endsWith(".csv") || entry.endsWith(".yarn")) {
|
||||
} else if (
|
||||
entry.endsWith(".md") ||
|
||||
entry.endsWith(".csv") ||
|
||||
entry.endsWith(".yarn")
|
||||
) {
|
||||
try {
|
||||
const content = readFileSync(fullPath, "utf-8");
|
||||
index[normalizedRelPath] = content;
|
||||
@@ -205,6 +210,7 @@ export function createContentServer(
|
||||
port: number,
|
||||
distPath: string = distDir,
|
||||
host: string = "0.0.0.0",
|
||||
mqttPort: number = 1883,
|
||||
): ContentServer {
|
||||
let contentIndex: ContentIndex = {};
|
||||
|
||||
@@ -216,14 +222,18 @@ export function createContentServer(
|
||||
// 监听文件变化
|
||||
console.log("监听文件变化...");
|
||||
const watcher = watch(contentDir, {
|
||||
ignored: /(^|[\/\\])\../,
|
||||
ignored: /(^|[\/\\])\./,
|
||||
persistent: true,
|
||||
ignoreInitial: true,
|
||||
});
|
||||
|
||||
watcher
|
||||
.on("add", (path) => {
|
||||
if (path.endsWith(".md") || path.endsWith(".csv") || path.endsWith(".yarn")) {
|
||||
if (
|
||||
path.endsWith(".md") ||
|
||||
path.endsWith(".csv") ||
|
||||
path.endsWith(".yarn")
|
||||
) {
|
||||
try {
|
||||
const content = readFileSync(path, "utf-8");
|
||||
const relPath = "/" + relative(contentDir, path).split(sep).join("/");
|
||||
@@ -235,7 +245,11 @@ export function createContentServer(
|
||||
}
|
||||
})
|
||||
.on("change", (path) => {
|
||||
if (path.endsWith(".md") || path.endsWith(".csv") || path.endsWith(".yarn")) {
|
||||
if (
|
||||
path.endsWith(".md") ||
|
||||
path.endsWith(".csv") ||
|
||||
path.endsWith(".yarn")
|
||||
) {
|
||||
try {
|
||||
const content = readFileSync(path, "utf-8");
|
||||
const relPath = "/" + relative(contentDir, path).split(sep).join("/");
|
||||
@@ -247,13 +261,24 @@ export function createContentServer(
|
||||
}
|
||||
})
|
||||
.on("unlink", (path) => {
|
||||
if (path.endsWith(".md") || path.endsWith(".csv") || path.endsWith(".yarn")) {
|
||||
if (
|
||||
path.endsWith(".md") ||
|
||||
path.endsWith(".csv") ||
|
||||
path.endsWith(".yarn")
|
||||
) {
|
||||
const relPath = "/" + relative(contentDir, path).split(sep).join("/");
|
||||
delete contentIndex[relPath];
|
||||
console.log(`[删除] ${relPath}`);
|
||||
}
|
||||
});
|
||||
|
||||
// ---- Journal / MQTT broker ----
|
||||
let journal: Awaited<ReturnType<typeof createJournalServer>> | null = null;
|
||||
|
||||
createJournalServer(contentDir, mqttPort).then((j) => {
|
||||
journal = j;
|
||||
});
|
||||
|
||||
// 创建请求处理器
|
||||
const handleRequest = createRequestHandler(
|
||||
contentDir,
|
||||
@@ -269,7 +294,9 @@ export function createContentServer(
|
||||
console.log(`\n开发服务器已启动：http://${displayHost}:${port}`);
|
||||
console.log(`内容目录：${contentDir}`);
|
||||
console.log(`静态资源目录：${distPath}`);
|
||||
console.log(`索引文件：http://${displayHost}:${port}/__CONTENT_INDEX.json\n`);
|
||||
console.log(
|
||||
`MQTT 代理端口：${mqttPort} (connect via tcp://${displayHost}:${mqttPort})\n`,
|
||||
);
|
||||
});
|
||||
|
||||
return {
|
||||
@@ -280,6 +307,7 @@ export function createContentServer(
|
||||
console.log("关闭内容服务器...");
|
||||
server.close();
|
||||
watcher.close();
|
||||
if (journal) journal.close();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -291,6 +319,7 @@ export const serveCommand: ServeCommandHandler = async (dir, options) => {
|
||||
const contentDir = resolve(dir);
|
||||
const port = parseInt(options.port, 10);
|
||||
const host = options.host || "0.0.0.0";
|
||||
const mqttPort = parseInt(options.mqttPort || "1883", 10);
|
||||
|
||||
createContentServer(contentDir, port, distDir, host);
|
||||
createContentServer(contentDir, port, distDir, host, mqttPort);
|
||||
};
|
||||
|
||||
+19
-19
@@ -1,33 +1,33 @@
|
||||
#!/usr/bin/env node
|
||||
import { Command } from 'commander';
|
||||
import { serveCommand } from './commands/serve.js';
|
||||
import { compileCommand } from './commands/compile.js';
|
||||
import { mcpCommand } from './commands/mcp.js';
|
||||
import type { ServeOptions, CompileOptions } from './types.js';
|
||||
import { Command } from "commander";
|
||||
import { serveCommand } from "./commands/serve.js";
|
||||
import { compileCommand } from "./commands/compile.js";
|
||||
import { mcpCommand } from "./commands/mcp.js";
|
||||
import type { ServeOptions, CompileOptions } from "./types.js";
|
||||
|
||||
const program = new Command();
|
||||
|
||||
program
|
||||
.name('ttrpg')
|
||||
.description('TTRPG 工具箱 - 用于编译和预览 TTRPG 文档')
|
||||
.version('0.0.1');
|
||||
.name("ttrpg")
|
||||
.description("TTRPG 工具箱 - 用于编译和预览 TTRPG 文档")
|
||||
.version("0.0.1");
|
||||
|
||||
program
|
||||
.command('serve')
|
||||
.description('运行一个 web 服务器预览目录中的内容，并实时监听更新')
|
||||
.argument('[dir]', '要预览的目录', '.')
|
||||
.option('-p, --port <port>', '端口号', '3000')
|
||||
.option('-h, --host <host>', '主机地址', '0.0.0.0')
|
||||
.command("serve")
|
||||
.description("运行一个 web 服务器预览目录中的内容，并实时监听更新")
|
||||
.argument("[dir]", "要预览的目录", ".")
|
||||
.option("-p, --port <port>", "HTTP 端口号", "3000")
|
||||
.option("--mqtt-port <port>", "MQTT 代理端口号", "1883")
|
||||
.option("-h, --host <host>", "主机地址", "0.0.0.0")
|
||||
.action(serveCommand);
|
||||
|
||||
program
|
||||
.command('compile')
|
||||
.description('将目录中的内容输出为带 hash 路由、单个 html 入口的 web 应用')
|
||||
.argument('[dir]', '要编译的目录', '.')
|
||||
.option('-o, --output <dir>', '输出目录', './dist/output')
|
||||
.command("compile")
|
||||
.description("将目录中的内容输出为带 hash 路由、单个 html 入口的 web 应用")
|
||||
.argument("[dir]", "要编译的目录", ".")
|
||||
.option("-o, --output <dir>", "输出目录", "./dist/output")
|
||||
.action(compileCommand);
|
||||
|
||||
program
|
||||
.addCommand(mcpCommand);
|
||||
program.addCommand(mcpCommand);
|
||||
|
||||
program.parse();
|
||||
|
||||
@@ -0,0 +1,227 @@
|
||||
/**
|
||||
* Journal stream persistence — MQTT broker + JSONL append + session manifest
|
||||
*
|
||||
* Starts an aedes MQTT broker and connects to itself to:
|
||||
* 1. Persist all stream messages to .ttrpg/sessions/{id}/stream.jsonl
|
||||
* 2. Manage session lifecycle via retained meta topics
|
||||
*
|
||||
* No HTTP routes — the JSONL files are served by the existing static server.
|
||||
*/
|
||||
|
||||
import { join, resolve } from "path";
|
||||
import {
|
||||
appendFileSync,
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
readFileSync,
|
||||
rmdirSync,
|
||||
unlinkSync,
|
||||
writeFileSync,
|
||||
} from "fs";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Types
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
interface SessionMeta {
|
||||
name: string;
|
||||
created: number;
|
||||
players: string[];
|
||||
}
|
||||
|
||||
interface SessionManifest {
|
||||
sessions: Record<string, SessionMeta>;
|
||||
}
|
||||
|
||||
export interface JournalServer {
|
||||
/** MQTT broker instance */
|
||||
broker: import("aedes").Aedes;
|
||||
/** TCP server the broker is attached to */
|
||||
tcpServer: import("net").Server;
|
||||
/** Persistence client (server subscribes to itself) */
|
||||
persistenceClient: import("mqtt").MqttClient | null;
|
||||
/** Clean shutdown */
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Path helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
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 {
|
||||
return join(dataRoot, "sessions", id);
|
||||
}
|
||||
|
||||
function streamPath(dataRoot: string, id: string): string {
|
||||
return join(sessionDir(dataRoot, id), "stream.jsonl");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Manifest
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function loadManifest(dataRoot: string): SessionManifest {
|
||||
const p = manifestPath(dataRoot);
|
||||
if (!existsSync(p)) return { sessions: {} };
|
||||
try {
|
||||
return JSON.parse(readFileSync(p, "utf-8"));
|
||||
} catch {
|
||||
return { sessions: {} };
|
||||
}
|
||||
}
|
||||
|
||||
function saveManifest(dataRoot: string, m: SessionManifest): void {
|
||||
writeFileSync(manifestPath(dataRoot), JSON.stringify(m, null, 2), "utf-8");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// JSONL persistence
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
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");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Server factory
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export async function createJournalServer(
|
||||
contentDir: string,
|
||||
mqttPort: number,
|
||||
): Promise<JournalServer> {
|
||||
const root = dataDir(contentDir);
|
||||
console.log(`[journal] data dir: ${root}`);
|
||||
|
||||
// ---- MQTT Broker ----
|
||||
const { Aedes: AedesFactory } = await import("aedes");
|
||||
const { createServer } = await import("net");
|
||||
|
||||
const broker = new AedesFactory();
|
||||
const tcpServer = createServer(broker.handle.bind(broker));
|
||||
|
||||
tcpServer.listen(mqttPort, "0.0.0.0", () => {
|
||||
console.log(`[journal] MQTT broker on port ${mqttPort}`);
|
||||
});
|
||||
|
||||
// ---- Persistence + Manifest client ----
|
||||
let client: import("mqtt").MqttClient | null = null;
|
||||
|
||||
const { default: mqtt } = await import("mqtt");
|
||||
client = mqtt.connect(`mqtt://127.0.0.1:${mqttPort}`, {
|
||||
clientId: "journal-server",
|
||||
});
|
||||
|
||||
client.on("connect", () => {
|
||||
// 1. Persistence: append every stream message to JSONL
|
||||
client!.subscribe("ttrpg/+/stream", { qos: 1 }, (err) => {
|
||||
if (err) console.error("[journal] stream sub err:", err);
|
||||
});
|
||||
|
||||
// 2. Session meta: manage manifest from retained meta topics
|
||||
client!.subscribe("ttrpg/+/meta", { qos: 1 }, (err) => {
|
||||
if (err) console.error("[journal] meta sub err:", err);
|
||||
});
|
||||
});
|
||||
|
||||
client.on("message", (topic, payload) => {
|
||||
const parts = topic.split("/");
|
||||
if (parts.length < 3) return;
|
||||
const sessionId = parts[1];
|
||||
const subtopic = parts[2];
|
||||
|
||||
if (subtopic === "stream") {
|
||||
// Persist every stream message
|
||||
try {
|
||||
appendStream(root, sessionId, payload.toString());
|
||||
} catch (e) {
|
||||
console.error(`[journal] stream append err for ${sessionId}:`, e);
|
||||
}
|
||||
} else if (subtopic === "meta") {
|
||||
// Session lifecycle
|
||||
handleMetaChange(root, sessionId, client!, payload.toString());
|
||||
}
|
||||
});
|
||||
|
||||
console.log("[journal] persistence listener active");
|
||||
|
||||
return {
|
||||
broker,
|
||||
tcpServer,
|
||||
persistenceClient: client,
|
||||
async close() {
|
||||
console.log("[journal] shutting down...");
|
||||
client?.end(true);
|
||||
await new Promise<void>((r) => broker.close(() => r()));
|
||||
await new Promise<void>((r) => tcpServer.close(() => r()));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Session lifecycle
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const $SESSIONS = "ttrpg/$SESSIONS";
|
||||
|
||||
function handleMetaChange(
|
||||
root: string,
|
||||
sessionId: string,
|
||||
client: import("mqtt").MqttClient,
|
||||
rawPayload: string,
|
||||
): void {
|
||||
const manifest = loadManifest(root);
|
||||
|
||||
try {
|
||||
const meta = JSON.parse(rawPayload) as SessionMeta | null;
|
||||
|
||||
if (!meta || !meta.name) {
|
||||
// Tombstone: empty or invalid payload → delete session
|
||||
if (manifest.sessions[sessionId]) {
|
||||
delete manifest.sessions[sessionId];
|
||||
|
||||
// Wipe files
|
||||
const sp = streamPath(root, sessionId);
|
||||
if (existsSync(sp)) unlinkSync(sp);
|
||||
const sd = sessionDir(root, sessionId);
|
||||
try {
|
||||
rmdirSync(sd);
|
||||
} catch {
|
||||
/* not empty, fine */
|
||||
}
|
||||
|
||||
console.log(`[journal] session deleted: ${sessionId}`);
|
||||
}
|
||||
} else {
|
||||
// Create or update
|
||||
manifest.sessions[sessionId] = {
|
||||
name: meta.name,
|
||||
created: meta.created || Date.now(),
|
||||
players: meta.players || [],
|
||||
};
|
||||
console.log(`[journal] session updated: ${sessionId} (${meta.name})`);
|
||||
}
|
||||
|
||||
saveManifest(root, manifest);
|
||||
|
||||
// Republish full manifest as a retained message
|
||||
client.publish($SESSIONS, JSON.stringify(manifest), {
|
||||
qos: 1,
|
||||
retain: true,
|
||||
});
|
||||
} catch (e) {
|
||||
console.error(`[journal] meta parse err for ${sessionId}:`, e);
|
||||
}
|
||||
}
|
||||
+9
-2
@@ -1,14 +1,21 @@
|
||||
export interface ServeOptions {
|
||||
port: string;
|
||||
host?: string;
|
||||
mqttPort?: string;
|
||||
}
|
||||
|
||||
export interface CompileOptions {
|
||||
output: string;
|
||||
}
|
||||
|
||||
export type ServeCommandHandler = (dir: string, options: ServeOptions) => Promise<void>;
|
||||
export type CompileCommandHandler = (dir: string, options: CompileOptions) => Promise<void>;
|
||||
export type ServeCommandHandler = (
|
||||
dir: string,
|
||||
options: ServeOptions,
|
||||
) => Promise<void>;
|
||||
export type CompileCommandHandler = (
|
||||
dir: string,
|
||||
options: CompileOptions,
|
||||
) => Promise<void>;
|
||||
|
||||
export interface MarkdownFile {
|
||||
path: string;
|
||||
|
||||
Reference in New Issue
Block a user