From 8194438816be444950c74151c70b7ff03a9a4f12 Mon Sep 17 00:00:00 2001 From: hypercross Date: Mon, 6 Jul 2026 14:48:18 +0800 Subject: [PATCH] feat(journal): implement message registry and stream store Implement a robust journal system featuring a type-safe message registry and an MQTT-backed stream store. - Add `registry.ts` to manage message type definitions, Zod schemas, and custom reducers. - Implement `journalStream.ts` to manage MQTT connections, session lifecycle, and reactive message state. - Add built-in message types for `narrative`, `intent`, `roll`, and `article`. - Provide utilities for message validation, optimistic local updates, and message reversion. --- src/components/journal/index.ts | 35 ++ src/components/journal/registry.ts | 126 +++++++ src/components/journal/types/article.ts | 37 ++ src/components/journal/types/index.ts | 11 + src/components/journal/types/intent.ts | 58 +++ src/components/journal/types/narrative.ts | 26 ++ src/components/journal/types/roll.ts | 60 +++ src/components/stores/journalStream.ts | 441 ++++++++++++++++++++++ 8 files changed, 794 insertions(+) create mode 100644 src/components/journal/index.ts create mode 100644 src/components/journal/registry.ts create mode 100644 src/components/journal/types/article.ts create mode 100644 src/components/journal/types/index.ts create mode 100644 src/components/journal/types/intent.ts create mode 100644 src/components/journal/types/narrative.ts create mode 100644 src/components/journal/types/roll.ts create mode 100644 src/components/stores/journalStream.ts diff --git a/src/components/journal/index.ts b/src/components/journal/index.ts new file mode 100644 index 0000000..c1737f3 --- /dev/null +++ b/src/components/journal/index.ts @@ -0,0 +1,35 @@ +/** + * Journal module — barrel export + */ + +// Trigger message type registration +import "./types"; + +export { + registerMessageType, + getMessageType, + registeredTypes, + validatePayload, + canEmit, +} from "./registry"; +export type { MessageTypeDef, StreamMessage } from "./registry"; +export { + connectStream, + disconnectStream, + hydrateFromServer, + sendMessage, + revertLatest, + canRevert, + visibleMessages, + useJournalStream, + journalStreamState, + createSession, + deleteSession, + sessions, + setMyName, +} from "../stores/journalStream"; +export type { + JournalStreamState, + SessionMeta, + SessionManifest, +} from "../stores/journalStream"; diff --git a/src/components/journal/registry.ts b/src/components/journal/registry.ts new file mode 100644 index 0000000..ba0d9d5 --- /dev/null +++ b/src/components/journal/registry.ts @@ -0,0 +1,126 @@ +/** + * Journal Stream — Message Type Registry + * + * An open, add-only registry of message types. Game-specific widgets + * call `registerMessageType()` at module import time. The stream core + * uses this to validate, render, and reduce messages. + */ + +import type { ZodType } from "zod"; +import type { JSX } from "solid-js"; + +// --------------------------------------------------------------------------- +// StreamMessage +// --------------------------------------------------------------------------- + +/** + * The typed envelope carried on the stream topic. + * `P` is narrowed by the `type` key at the receiving side. + */ +export interface StreamMessage

{ + /** Deterministic id: `{sender}-{seq}` */ + id: string; + /** Sender identifier — "gm" or a player name */ + sender: string; + /** Monotonic sequence number, per sender */ + seq: number; + /** Registered type key — determines which payload schema & widget are used */ + type: string; + /** Type-specific data, validated against the registry's Zod schema */ + payload: P; + /** Epoch millis, set by the sender at publish time */ + timestamp: number; + /** + * Soft-delete. Only the sender's latest (highest seq) message can be + * reverted. Sending a new message locks the previous one permanently. + */ + reverted: boolean; +} + +// --------------------------------------------------------------------------- +// MessageTypeDef +// --------------------------------------------------------------------------- + +export interface MessageTypeDef

{ + /** Unique type key, e.g. "roll.request", "article.reveal" */ + type: string; + /** Human-readable label shown in the compose panel dropdown */ + label: string; + /** Which roles can emit this type. Empty array = all roles allowed. */ + emitters?: ("gm" | "player")[]; + /** Zod schema used to validate payload before publish */ + schema: ZodType

; + /** Returns a default payload for compose panel initialization */ + defaultPayload?: () => P; + /** + * Render the message body in the stream. + * Receives the full message for context (sender, timestamp, reverted). + * Return undefined to suppress inline rendering (e.g. pure state mutations). + */ + render?: (payload: P, msg: StreamMessage

) => JSX.Element | undefined; + /** + * Apply side-effects to local state when this message is received. + * Runs once per message, at initial hydration and on live receipt. + */ + reducer?: (payload: P, msg: StreamMessage

) => void; +} + +// --------------------------------------------------------------------------- +// Registry +// --------------------------------------------------------------------------- + +const registry = new Map>(); + +/** + * Register a message type. Calling module should import and call this at + * top-level so the registry is populated before any stream operations. + */ +export function registerMessageType

(def: MessageTypeDef

): void { + if (registry.has(def.type)) { + throw new Error(`Message type "${def.type}" is already registered.`); + } + registry.set(def.type, def); +} + +/** Retrieve a registered type definition. */ +export function getMessageType(type: string): MessageTypeDef | undefined { + return registry.get(type); +} + +/** Iterate over all registered type keys. */ +export function registeredTypes(): IterableIterator { + return registry.keys(); +} + +/** + * Validate a payload against the registered schema for `type`. + * Returns the parsed (narrowed) payload on success, or a ZodError. + */ +export function validatePayload

( + type: string, + payload: P, +): { success: true; data: P } | { success: false; error: string } { + const def = registry.get(type); + if (!def) { + return { success: false, error: `Unknown message type: "${type}"` }; + } + const result = def.schema.safeParse(payload); + if (result.success) { + return { success: true, data: result.data as P }; + } + return { + success: false, + error: result.error.issues.map((i) => `${i.path.join(".")}: ${i.message}`).join("; "), + }; +} + +/** + * Check whether a role (sender) is allowed to emit a given message type. + */ +export function canEmit(type: string, role: string): boolean { + const def = registry.get(type); + if (!def) return false; + const emitters = def.emitters; + if (!emitters || emitters.length === 0) return true; + return emitters.includes(role as "gm" | "player"); +} diff --git a/src/components/journal/types/article.ts b/src/components/journal/types/article.ts new file mode 100644 index 0000000..b6a1d62 --- /dev/null +++ b/src/components/journal/types/article.ts @@ -0,0 +1,37 @@ +/** + * Built-in message type: article.reveal + * + * GM reveals a document/article path to players. The reducer populates + * the revealedPaths set so the Article component can gate visibility. + */ + +import { z } from "zod"; +import { registerMessageType } from "../registry"; +import { journalSetState } from "../../stores/journalStream"; +import { produce } from "solid-js/store"; + +const schema = z.object({ + /** Path relative to content root, e.g. "/adventures/dungeon.md" */ + path: z.string().min(1), + /** Optional display label, defaults to the path */ + label: z.string().optional(), + /** Optional section heading to deep-link */ + section: z.string().optional(), +}); + +export type ArticleRevealPayload = z.infer; + +registerMessageType({ + type: "article.reveal", + label: "Reveal Article", + emitters: ["gm"], + schema, + defaultPayload: () => ({ path: "/" }), + reducer: (p) => { + journalSetState( + produce((s) => { + s.revealedPaths.add(p.path); + }), + ); + }, +}); diff --git a/src/components/journal/types/index.ts b/src/components/journal/types/index.ts new file mode 100644 index 0000000..3ff6a0e --- /dev/null +++ b/src/components/journal/types/index.ts @@ -0,0 +1,11 @@ +/** + * Journal message types — barrel export + * + * Importing this module registers all built-in message types. + * Game-specific widgets should create their own barrel and import it. + */ + +import "./narrative"; +import "./roll"; +import "./article"; +import "./intent"; diff --git a/src/components/journal/types/intent.ts b/src/components/journal/types/intent.ts new file mode 100644 index 0000000..15772d2 --- /dev/null +++ b/src/components/journal/types/intent.ts @@ -0,0 +1,58 @@ +/** + * Built-in message types: intent / resolution + * + * intent — Player declares an action or resource usage for GM to resolve + * resolution — GM resolves a player intent (outcome, stat changes, etc.) + */ + +import { z } from "zod"; +import { registerMessageType } from "../registry"; + +// --------------------------------------------------------------------------- +// intent +// --------------------------------------------------------------------------- + +const intentSchema = z.object({ + /** Plain-text description of what the player wants to do */ + description: z.string().min(1), + /** + * Optional JSON-serializable context (item name, skill name, target, etc.) + * Parsed by game-specific widgets. + */ + context: z.record(z.string(), z.unknown()).optional(), +}); + +export type IntentPayload = z.infer; + +registerMessageType({ + type: "intent", + label: "Declare Intent", + emitters: ["player"], + schema: intentSchema, + defaultPayload: () => ({ description: "" }), +}); + +// --------------------------------------------------------------------------- +// resolution +// --------------------------------------------------------------------------- + +const resolutionSchema = z.object({ + /** The id of the intent message being resolved */ + intentId: z.string().min(1), + /** Freeform outcome text */ + outcome: z.string().min(1), + /** + * Stat changes to apply. Game-specific widgets interpret these keys. + * Example: { "hp": -3, "stress": 1, "status.add": "bleeding" } + */ + statChanges: z.record(z.string(), z.unknown()).optional(), +}); + +export type ResolutionPayload = z.infer; + +registerMessageType({ + type: "resolution", + label: "Resolve Intent", + emitters: ["gm"], + schema: resolutionSchema, +}); diff --git a/src/components/journal/types/narrative.ts b/src/components/journal/types/narrative.ts new file mode 100644 index 0000000..26b48b5 --- /dev/null +++ b/src/components/journal/types/narrative.ts @@ -0,0 +1,26 @@ +/** + * Built-in message type: narrative + * + * Freeform text from GM or players. Renders as a chat bubble. + */ + +import { z } from "zod"; +import { registerMessageType } from "../registry"; + +const schema = z.object({ + text: z.string().min(1, "Message cannot be empty").max(2000, "Message too long"), +}); + +export type NarrativePayload = z.infer; + +registerMessageType({ + type: "narrative", + label: "Narrative", + schema, + defaultPayload: () => ({ text: "" }), + render: (p, msg) => { + // Placeholder — real render will be in StreamMessage component + // Returning a simple representation for now + return undefined; + }, +}); diff --git a/src/components/journal/types/roll.ts b/src/components/journal/types/roll.ts new file mode 100644 index 0000000..f02ee3a --- /dev/null +++ b/src/components/journal/types/roll.ts @@ -0,0 +1,60 @@ +/** + * Built-in message types: roll.request / roll.result + * + * roll.request — GM asks for a roll, renders a prompt with dice roller + * roll.result — Player responds with their roll outcome + */ + +import { z } from "zod"; +import { registerMessageType } from "../registry"; + +// --------------------------------------------------------------------------- +// roll.request +// --------------------------------------------------------------------------- + +const rollRequestSchema = z.object({ + /** Dice notation, e.g. "2d6", "1d20+5", "3d6kh1" */ + notation: z.string().min(1), + /** Human label, e.g. "Fear save", "Attack roll" */ + label: z.string().min(1), + /** Optional target number to compare against */ + target: z.number().int().optional(), + /** Optional extra context */ + description: z.string().optional(), +}); + +export type RollRequestPayload = z.infer; + +registerMessageType({ + type: "roll.request", + label: "Request Roll", + emitters: ["gm"], + schema: rollRequestSchema, + defaultPayload: () => ({ notation: "1d20", label: "" }), +}); + +// --------------------------------------------------------------------------- +// roll.result +// --------------------------------------------------------------------------- + +const rollResultSchema = z.object({ + /** The dice notation that was rolled */ + notation: z.string().min(1), + /** The raw dice results, e.g. [6, 4] for 2d6 */ + dice: z.array(z.number().int().min(1)), + /** Modifiers, e.g. +5 */ + modifier: z.number().int().default(0), + /** Final total after modifiers and keep/drop rules */ + total: z.number(), + /** Optional: which roll.request message this is responding to (by id) */ + requestId: z.string().optional(), +}); + +export type RollResultPayload = z.infer; + +registerMessageType({ + type: "roll.result", + label: "Roll Result", + emitters: ["player"], + schema: rollResultSchema, +}); diff --git a/src/components/stores/journalStream.ts b/src/components/stores/journalStream.ts new file mode 100644 index 0000000..bb44ad4 --- /dev/null +++ b/src/components/stores/journalStream.ts @@ -0,0 +1,441 @@ +/** + * Journal Stream — Client Store + * + * Reactive state for a single session's message stream. Manages MQTT + * connection, local message log, per-sender sequence tracking, and the + * revealed-paths set (populated by the article.reveal reducer). + * + * Session lifecycle (create/list/delete) is handled via MQTT retained + * topics — ttrpg/$SESSIONS for the manifest and ttrpg/{id}/meta per session. + * + * No persistence here — that's the CLI server's job via JSONL append. + */ + +import { createStore, produce } from "solid-js/store"; +import { createSignal } from "solid-js"; +import type { StreamMessage } from "../journal/registry"; +import { getMessageType, validatePayload } from "../journal/registry"; + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +export interface JournalStreamState { + sessionId: string | null; + /** Full message log, oldest-first */ + messages: StreamMessage[]; + /** Last sequence number per sender */ + senderSeq: Record; + /** + * Paths revealed by article.reveal messages. + * Populated during hydration and live receipt via the type's reducer. + */ + revealedPaths: Set; + /** MQTT connection status */ + connected: boolean; + /** This client's identity */ + myName: string; + /** Broker URL, set after connect */ + brokerUrl: string | null; +} + +export interface SessionMeta { + name: string; + created: number; + players: string[]; +} + +export interface SessionManifest { + sessions: Record; +} + +// --------------------------------------------------------------------------- +// localStorage keys +// --------------------------------------------------------------------------- + +const LS_PLAYER_NAME = "ttrpg.playerName"; +const LS_BROKER_URL = "ttrpg.brokerUrl"; +const LS_LAST_SESSION = "ttrpg.lastSessionId"; + +function loadPersisted(): { + myName: string; + brokerUrl: string | null; + lastSessionId: string | null; +} { + if (typeof localStorage === "undefined") { + return { myName: "gm", brokerUrl: null, lastSessionId: null }; + } + return { + myName: localStorage.getItem(LS_PLAYER_NAME) || "gm", + brokerUrl: localStorage.getItem(LS_BROKER_URL), + lastSessionId: localStorage.getItem(LS_LAST_SESSION), + }; +} + +// --------------------------------------------------------------------------- +// Store +// --------------------------------------------------------------------------- + +const persisted = loadPersisted(); + +const [state, setState] = createStore({ + sessionId: persisted.lastSessionId, + messages: [], + senderSeq: {}, + revealedPaths: new Set(), + connected: false, + myName: persisted.myName, + brokerUrl: persisted.brokerUrl, +}); + +export { setState as journalSetState }; + +const [sessionList, setSessionList] = createSignal({ + sessions: {}, +}); + +export { sessionList as sessions }; + +/** + * Change the current player's name. Persisted to localStorage so it + * survives page reloads. + */ +export function setMyName(name: string): void { + setState("myName", name); + if (typeof localStorage !== "undefined") { + localStorage.setItem(LS_PLAYER_NAME, name); + } +} + +// Will hold the MQTT client instance after connect() +let _mqttClient: import("mqtt").MqttClient | null = null; +let _mqttConnected = false; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeMessageId(sender: string, seq: number): string { + return `${sender}-${seq}`; +} + +function runReducer(msg: StreamMessage): void { + const def = getMessageType(msg.type); + if (def?.reducer) { + def.reducer(msg.payload, msg); + } +} + +const $SESSIONS = "ttrpg/$SESSIONS"; + +// --------------------------------------------------------------------------- +// Hydration (initial load from server) +// --------------------------------------------------------------------------- + +/** + * Load the full message history from the static server's JSONL file. + * The file is served from the same HTTP origin as the web app. + * Runs all reducers in order. + */ +export async function hydrateFromServer(sessionId: string): Promise { + const response = await fetch( + `/.ttrpg/sessions/${encodeURIComponent(sessionId)}/stream.jsonl`, + ); + + if (!response.ok) { + if (response.status === 404) { + return; // fresh session, no file yet + } + throw new Error(`Failed to load session: ${response.statusText}`); + } + + const text = await response.text(); + const lines = text.split("\n").filter((l) => l.trim()); + + const messages: StreamMessage[] = []; + const senderSeq: Record = {}; + + for (const line of lines) { + try { + const msg: StreamMessage = JSON.parse(line); + messages.push(msg); + senderSeq[msg.sender] = Math.max(senderSeq[msg.sender] ?? 0, msg.seq); + runReducer(msg); + } catch { + /* skip corrupt */ + } + } + + setState( + produce((s) => { + s.messages = messages; + s.senderSeq = senderSeq; + s.sessionId = sessionId; + }), + ); +} + +// --------------------------------------------------------------------------- +// MQTT Connect +// --------------------------------------------------------------------------- + +/** + * Connect to the MQTT broker, subscribe to the session stream, session + * list, and session meta. Must be called after `hydrateFromServer`. + */ +export async function connectStream( + sessionId: string, + brokerUrl: string, +): Promise { + const { default: mqtt } = await import("mqtt"); + + const client = mqtt.connect(brokerUrl, { + clientId: `${state.myName}-${Date.now()}`, + protocol: brokerUrl.startsWith("wss") ? "wss" : "ws", + reconnectPeriod: 2000, + }); + + _mqttClient = client; + + client.on("connect", () => { + _mqttConnected = true; + setState("connected", true); + setState("brokerUrl", brokerUrl); + + // Persist connection info for next time + if (typeof localStorage !== "undefined") { + localStorage.setItem(LS_BROKER_URL, brokerUrl); + localStorage.setItem(LS_LAST_SESSION, sessionId); + } + + client.subscribe(`ttrpg/${sessionId}/stream`, { qos: 1 }, (err) => { + if (err) console.error("[stream] stream sub err:", err); + }); + client.subscribe($SESSIONS, { qos: 1 }, (err) => { + if (err) console.error("[stream] sessions sub err:", err); + }); + client.subscribe(`ttrpg/${sessionId}/meta`, { qos: 1 }); + }); + + client.on("message", (topic, payload) => { + const raw = payload.toString(); + const parts = topic.split("/"); + + if (topic === $SESSIONS) { + // Session manifest update + try { + const manifest: SessionManifest = JSON.parse(raw); + setSessionList(manifest); + } catch (e) { + console.error("[stream] manifest parse err:", e); + } + return; + } + + if (parts.length >= 3 && parts[0] === "ttrpg" && parts[2] === "meta") { + // Session meta update — manifest will be republished by server, + // handled via $SESSIONS subscription above. + return; + } + + if (parts.length >= 3 && parts[0] === "ttrpg" && parts[2] === "stream") { + try { + const msg: StreamMessage = JSON.parse(raw); + receiveMessage(msg); + } catch (e) { + console.error("[stream] malformed message:", e); + } + } + }); + + client.on("close", () => { + _mqttConnected = false; + setState("connected", false); + }); + + client.on("error", (err) => { + console.error("[stream] mqtt error:", err); + }); +} + +// --------------------------------------------------------------------------- +// Session lifecycle +// --------------------------------------------------------------------------- + +/** + * Create a new session by publishing retained metadata to its meta topic. + * The server picks it up and adds it to the $SESSIONS manifest. + */ +export function createSession(name: string, players: string[] = []): void { + if (!_mqttClient || !_mqttConnected) return; + + const id = + name + .toLowerCase() + .replace(/[^a-z0-9]+/g, "-") + .replace(/^-|-$/g, "") || "session"; + + const meta: SessionMeta = { name, created: Date.now(), players }; + + _mqttClient.publish(`ttrpg/${id}/meta`, JSON.stringify(meta), { + qos: 1, + retain: true, + }); +} + +/** + * Delete a session: publish tombstone (empty payload) to its meta topic. + */ +export function deleteSession(sessionId: string): void { + if (!_mqttClient || !_mqttConnected) return; + + _mqttClient.publish(`ttrpg/${sessionId}/meta`, "", { qos: 1, retain: true }); +} + +// --------------------------------------------------------------------------- +// Send +// --------------------------------------------------------------------------- + +/** + * Publish a message to the stream. Validates the payload against the + * registered Zod schema before sending. Auto-increments the sender's seq. + */ +export function sendMessage( + type: string, + payload: T, +): + { success: true; msg: StreamMessage } | { success: false; error: string } { + if (!_mqttClient || !_mqttConnected) { + return { success: false, error: "Not connected to stream" }; + } + + const sessionId = state.sessionId; + if (!sessionId) { + return { success: false, error: "No active session" }; + } + + const validation = validatePayload(type, payload); + if (!validation.success) { + return { success: false, error: validation.error }; + } + + const sender = state.myName; + const seq = (state.senderSeq[sender] ?? 0) + 1; + const id = makeMessageId(sender, seq); + + const msg: StreamMessage = { + id, + sender, + seq, + type, + payload: validation.data, + timestamp: Date.now(), + reverted: false, + }; + + const topic = `ttrpg/${sessionId}/stream`; + _mqttClient.publish(topic, JSON.stringify(msg), { qos: 1 }, (err) => { + if (err) console.error("[stream] publish error:", err); + }); + + // Optimistic local insert + receiveMessage(msg as StreamMessage); + + return { success: true, msg }; +} + +// --------------------------------------------------------------------------- +// Receive +// --------------------------------------------------------------------------- + +function receiveMessage(msg: StreamMessage): void { + const existing = state.messages.find((m) => m.id === msg.id); + if (existing) { + setState( + produce((s) => { + const idx = s.messages.findIndex((m) => m.id === msg.id); + if (idx !== -1) s.messages[idx] = msg; + }), + ); + return; + } + + setState( + produce((s) => { + s.messages.push(msg); + s.senderSeq[msg.sender] = Math.max(s.senderSeq[msg.sender] ?? 0, msg.seq); + }), + ); + + runReducer(msg); +} + +// --------------------------------------------------------------------------- +// Revert +// --------------------------------------------------------------------------- + +/** + * Revert the current sender's latest (highest seq) message. + * Only works if it's still the latest — a subsequent message locks it. + */ +export function revertLatest(): + { success: true } | { success: false; error: string } { + if (!_mqttClient || !_mqttConnected) { + return { success: false, error: "Not connected to stream" }; + } + + const sessionId = state.sessionId; + if (!sessionId) return { success: false, error: "No active session" }; + + const sender = state.myName; + const latestSeq = state.senderSeq[sender]; + if (!latestSeq) return { success: false, error: "No messages to revert" }; + + const id = makeMessageId(sender, latestSeq); + const original = state.messages.find((m) => m.id === id); + if (!original) return { success: false, error: "Message not found" }; + if (original.reverted) return { success: false, error: "Already reverted" }; + + const reverted: StreamMessage = { ...original, reverted: true }; + const topic = `ttrpg/${sessionId}/stream`; + _mqttClient.publish(topic, JSON.stringify(reverted), { qos: 1 }, (err) => { + if (err) console.error("[stream] revert publish error:", err); + }); + + return { success: true }; +} + +// --------------------------------------------------------------------------- +// Disconnect +// --------------------------------------------------------------------------- + +export function disconnectStream(): void { + if (_mqttClient) { + _mqttClient.end(true); + _mqttClient = null; + _mqttConnected = false; + } + setState("connected", false); +} + +// --------------------------------------------------------------------------- +// Derived / helpers +// --------------------------------------------------------------------------- + +export function canRevert(): boolean { + const sender = state.myName; + const seq = state.senderSeq[sender]; + if (!seq) return false; + const msg = state.messages.find((m) => m.id === makeMessageId(sender, seq)); + return msg !== undefined && !msg.reverted; +} + +export function visibleMessages(): StreamMessage[] { + return state.messages.filter((m) => !m.reverted); +} + +export function useJournalStream() { + return state; +} + +export { state as journalStreamState };