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.
This commit is contained in:
parent
21f8fbcd0b
commit
8194438816
|
|
@ -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";
|
||||
|
|
@ -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<P = unknown> {
|
||||
/** 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<P = unknown> {
|
||||
/** 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<P>;
|
||||
/** 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<P>) => 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<P>) => void;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Registry
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const registry = new Map<string, MessageTypeDef<any>>();
|
||||
|
||||
/**
|
||||
* 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<P>(def: MessageTypeDef<P>): 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<string> {
|
||||
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<P>(
|
||||
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");
|
||||
}
|
||||
|
|
@ -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<typeof schema>;
|
||||
|
||||
registerMessageType<ArticleRevealPayload>({
|
||||
type: "article.reveal",
|
||||
label: "Reveal Article",
|
||||
emitters: ["gm"],
|
||||
schema,
|
||||
defaultPayload: () => ({ path: "/" }),
|
||||
reducer: (p) => {
|
||||
journalSetState(
|
||||
produce((s) => {
|
||||
s.revealedPaths.add(p.path);
|
||||
}),
|
||||
);
|
||||
},
|
||||
});
|
||||
|
|
@ -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";
|
||||
|
|
@ -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<typeof intentSchema>;
|
||||
|
||||
registerMessageType<IntentPayload>({
|
||||
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<typeof resolutionSchema>;
|
||||
|
||||
registerMessageType<ResolutionPayload>({
|
||||
type: "resolution",
|
||||
label: "Resolve Intent",
|
||||
emitters: ["gm"],
|
||||
schema: resolutionSchema,
|
||||
});
|
||||
|
|
@ -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<typeof schema>;
|
||||
|
||||
registerMessageType<NarrativePayload>({
|
||||
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;
|
||||
},
|
||||
});
|
||||
|
|
@ -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<typeof rollRequestSchema>;
|
||||
|
||||
registerMessageType<RollRequestPayload>({
|
||||
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<typeof rollResultSchema>;
|
||||
|
||||
registerMessageType<RollResultPayload>({
|
||||
type: "roll.result",
|
||||
label: "Roll Result",
|
||||
emitters: ["player"],
|
||||
schema: rollResultSchema,
|
||||
});
|
||||
|
|
@ -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<string, number>;
|
||||
/**
|
||||
* Paths revealed by article.reveal messages.
|
||||
* Populated during hydration and live receipt via the type's reducer.
|
||||
*/
|
||||
revealedPaths: Set<string>;
|
||||
/** 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<string, SessionMeta>;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 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<JournalStreamState>({
|
||||
sessionId: persisted.lastSessionId,
|
||||
messages: [],
|
||||
senderSeq: {},
|
||||
revealedPaths: new Set(),
|
||||
connected: false,
|
||||
myName: persisted.myName,
|
||||
brokerUrl: persisted.brokerUrl,
|
||||
});
|
||||
|
||||
export { setState as journalSetState };
|
||||
|
||||
const [sessionList, setSessionList] = createSignal<SessionManifest>({
|
||||
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<void> {
|
||||
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<string, number> = {};
|
||||
|
||||
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<void> {
|
||||
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<T>(
|
||||
type: string,
|
||||
payload: T,
|
||||
):
|
||||
{ success: true; msg: StreamMessage<T> } | { 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<T> = {
|
||||
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 };
|
||||
Loading…
Reference in New Issue