ttrpg-tools/src/components/stores/journalStream.ts

/**
 * 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 link 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;
  /** Human-readable name for the current session (from manifest) */
  sessionName: string | null;
  /** Full message log, oldest-first */
  messages: StreamMessage[];
  /** Last sequence number per sender */
  senderSeq: Record<string, number>;
  /**
   * Paths and sections revealed by link messages.
   * Key: normalized path (no .md). Value: set of revealed section slugs.
   * An empty set means the whole article is revealed.
   * Populated during hydration and live receipt via the type's reducer.
   */
  revealedPaths: Record<string, Set<string>>;
  /** MQTT connection status */
  connected: boolean;
  /** Granular connection state for UI indicators */
  connectionStatus: "disconnected" | "connecting" | "connected" | "error";
  /** Last connection error message, if any */
  connectionError: string | null;
  /** This client's identity */
  myName: string;
  /** Role: gm | player | observer. Immutable while connected. */
  myRole: "gm" | "player" | "observer";
  /** Broker URL, set after connect */
  brokerUrl: string | null;
  /** Active player list (keyed by player name) */
  players: Record<string, { role: string }>;
}

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";
const LS_PLAYER_ROLE = "ttrpg.playerRole";

function loadPersisted(): {
  myName: string;
  brokerUrl: string | null;
  lastSessionId: string | null;
  myRole: string;
} {
  if (typeof localStorage === "undefined") {
    return { myName: "gm", brokerUrl: null, lastSessionId: null, myRole: "gm" };
  }
  return {
    myName: localStorage.getItem(LS_PLAYER_NAME) || "gm",
    brokerUrl: localStorage.getItem(LS_BROKER_URL),
    lastSessionId: localStorage.getItem(LS_LAST_SESSION),
    myRole: localStorage.getItem(LS_PLAYER_ROLE) || "gm",
  };
}

// ---------------------------------------------------------------------------
// Store
// ---------------------------------------------------------------------------

const persisted = loadPersisted();
const urlParams = readUrlParams();

// URL params override localStorage if present
const initialName = urlParams.playerName ?? persisted.myName;
const initialSession = urlParams.sessionId ?? persisted.lastSessionId;

const [state, setState] = createStore<JournalStreamState>({
  sessionId: initialSession,
  sessionName: null,
  messages: [],
  senderSeq: {},
  revealedPaths: {},
  connected: false,
  connectionStatus: "disconnected",
  connectionError: null,
  myName: initialName,
  myRole: (persisted.myRole as "gm" | "player" | "observer") || "gm",
  brokerUrl: persisted.brokerUrl,
  players: {},
});

// Sync initial URL params if they came from localStorage (not URL)
if (initialName && !urlParams.playerName) syncUrlParam("player", initialName);
if (initialSession && !urlParams.sessionId)
  syncUrlParam("session", initialSession);

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. Also syncs to URL search param.
 */
export function setMyName(name: string): void {
  setState("myName", name);
  if (typeof localStorage !== "undefined") {
    localStorage.setItem(LS_PLAYER_NAME, name);
  }
  syncUrlParam("player", name);
}

/**
 * Change the current player's role. Persisted to localStorage.
 * Only callable when disconnected.
 */
export function setMyRole(role: "gm" | "player" | "observer"): void {
  setState("myRole", role);
  if (typeof localStorage !== "undefined") {
    localStorage.setItem(LS_PLAYER_ROLE, role);
  }
}

/**
 * Set the active session ID and sync to URL. Also resolves the human-readable
 * session name from the cached manifest.
 */
export function setSessionId(id: string | null): void {
  setState("sessionId", id);
  if (typeof localStorage !== "undefined" && id) {
    localStorage.setItem(LS_LAST_SESSION, id);
  }
  if (id) {
    syncUrlParam("session", id);
    // Resolve session name from current manifest
    const manifest = sessionList();
    const name = manifest.sessions[id]?.name ?? null;
    setState("sessionName", name);
  } else {
    removeUrlParam("session");
    setState("sessionName", null);
  }
}

// ---------------------------------------------------------------------------
// URL param sync (session & player for bookmarkable links)
// ---------------------------------------------------------------------------

function syncUrlParam(key: string, value: string): void {
  if (typeof window === "undefined") return;
  const url = new URL(window.location.href);
  url.searchParams.set(key, value);
  window.history.replaceState(null, "", url.toString());
}

function removeUrlParam(key: string): void {
  if (typeof window === "undefined") return;
  const url = new URL(window.location.href);
  url.searchParams.delete(key);
  window.history.replaceState(null, "", url.toString());
}

/**
 * Read initial session/player from URL search params on store init.
 * Called once at module load time.
 */
function readUrlParams(): {
  sessionId: string | null;
  playerName: string | null;
} {
  if (typeof window === "undefined")
    return { sessionId: null, playerName: null };
  const params = new URL(window.location.href).searchParams;
  return {
    sessionId: params.get("session"),
    playerName: params.get("player"),
  };
}

// 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");

  setState("connectionStatus", "connecting");
  setState("connectionError", null);

  return new Promise<void>((resolve, reject) => {
    const client = mqtt.connect(brokerUrl, {
      clientId: `${state.myName}-${state.myRole}-${Date.now()}`,
      protocol: brokerUrl.startsWith("wss") ? "wss" : "ws",
      reconnectPeriod: 2000,
    });

    _mqttClient = client;

    client.on("connect", () => {
      _mqttConnected = true;
      setState("connected", true);
      setState("connectionStatus", "connected");
      setState("connectionError", null);
      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 });
      // Presence tracking
      client.subscribe(`ttrpg/${sessionId}/presence/+`, { qos: 1 });

      // Publish own presence (retained)
      const presenceData = JSON.stringify({
        name: state.myName,
        role: state.myRole,
      });
      client.publish(
        `ttrpg/${sessionId}/presence/${state.myName}`,
        presenceData,
        { qos: 1, retain: true },
      );

      resolve();
    });

    client.on("error", (err) => {
      console.error("[stream] mqtt error:", err);
      setState("connectionStatus", "error");
      setState("connectionError", err.message);
      reject(err);
    });

    client.on("close", () => {
      _mqttConnected = false;
      setState("connected", false);
      setState("connectionStatus", "disconnected");
    });

    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);
          // Refresh the sessionName if we're in a session now
          const currentId = state.sessionId;
          if (currentId && manifest.sessions[currentId]) {
            setState("sessionName", manifest.sessions[currentId].name);
          }
        } 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;
      }

      // Presence: ttrpg/{sessionId}/presence/{playerName}
      if (
        parts.length >= 4 &&
        parts[0] === "ttrpg" &&
        parts[2] === "presence"
      ) {
        const playerName = parts[3];
        if (raw) {
          try {
            const presence = JSON.parse(raw);
            setState("players", playerName, {
              role: presence.role || "player",
            });
          } catch {
            setState("players", playerName, { role: "player" });
          }
        } else {
          // Tombstone — player disconnected
          setState(
            produce((s) => {
              delete s.players[playerName];
            }),
          );
        }
        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);
        }
      }
    });
  });
}

// ---------------------------------------------------------------------------
// 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) {
    // Clear our presence before disconnecting (tombstone)
    const sessionId = state.sessionId;
    if (sessionId) {
      _mqttClient.publish(`ttrpg/${sessionId}/presence/${state.myName}`, "", {
        qos: 1,
        retain: true,
      });
    }
    // Force disconnect without reconnect
    _mqttClient.end(true, void 0, () => {
      // noop
    });
    _mqttClient = null;
    _mqttConnected = false;
  }
  setState("connected", false);
  setState("connectionStatus", "disconnected");
  setState("players", {});

  // Strip autojoin param so the dialog shows normally on next connect
  removeUrlParam("autojoin");
}

// ---------------------------------------------------------------------------
// 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);
}

/**
 * Check whether a path (and optionally section) is revealed.
 * GM and disconnected clients see everything.
 * Non-GM connected clients only see explicitly revealed content.
 */
export function isPathRevealed(path: string, section?: string): boolean {
  if (!state.connected) return true;
  if (state.myRole === "gm") return true;

  const normalized = normalizePath(path);
  const revealed = state.revealedPaths[normalized];
  if (!revealed) return false;

  // Whole article revealed (empty set)
  if (revealed.size === 0) return true;
  // Section-specific check
  if (section) return revealed.has(section);
  // No section asked — at least some sections are revealed, article is partially visible
  return revealed.size > 0;
}

/**
 * Walk every element in `root` and tag it as revealed or concealed.
 *
 * Two-pass approach:
 * 1. Collect all headings with their ancestor chain.
 * 2. Cascade revealed status upward — if a heading is revealed, all its
 *    parent headings are marked revealed as well.
 * 3. Walk the tree again, applying `revealed` or `concealed` classes to
 *    every element (headings included) based on whether any current
 *    heading ancestor is in the cascaded revealed set.
 */
export function addRevealedClasses(root: HTMLDivElement, path: string) {
  if (!state.connected) return;
  if (state.myRole === "gm") return;

  const normalized = normalizePath(path);
  const revealed = state.revealedPaths[normalized];
  if (!revealed) return;

  const HEADING_TAGS = new Set(["H1", "H2", "H3", "H4", "H5", "H6"]);

  // ---- Pass 1: collect heading hierarchy ----
  interface HeadingEntry {
    el: Element;
    level: number;
    text: string;
    /** Texts of all ancestor headings (shallowest first) */
    parents: string[];
  }
  const headings: HeadingEntry[] = [];
  const cur: Record<number, string> = {};

  const collect = (el: Element) => {
    const tag = el.tagName.toUpperCase();
    if (HEADING_TAGS.has(tag)) {
      const level = Number(tag.charAt(1));
      const text = el.id || el.textContent?.trim() || "";
      const parents: string[] = [];
      for (let l = 1; l < level; l++) {
        if (cur[l]) parents.push(cur[l]);
      }
      headings.push({ el, level, text, parents });
      cur[level] = text;
      for (let l = level + 1; l <= 6; l++) delete cur[l];
    }
    for (const child of el.children) collect(child);
  };
  for (const child of root.children) collect(child);

  // ---- Cascade: if a heading is revealed, its parents are too ----
  const revealedSet = new Set<string>();
  for (const h of headings) {
    if (revealed.has(h.text) || revealedSet.has(h.text)) {
      revealedSet.add(h.text);
      for (const p of h.parents) revealedSet.add(p);
    }
  }

  // ---- Pass 2: apply classes using the cascaded set ----
  const stack: string[] = [];
  const apply = (el: Element) => {
    let pushed = false;
    for (const child of el.children) {
      const tag = child.tagName.toUpperCase();
      if (HEADING_TAGS.has(tag)) {
        if (pushed) stack.pop();
        const text = child.id || child.textContent?.trim() || "";
        stack.push(text);
        pushed = true;
      }

      const current = stack.length > 0 ? stack[stack.length - 1] : null;
      const isRevealed = current !== null && revealedSet.has(current);
      child.classList.add(isRevealed ? "revealed" : "concealed");

      apply(child);
    }
    if (pushed) stack.pop();
  };
  apply(root);
}

/** Normalize a path for lookup: strip .md, leading ./ or / */
function normalizePath(p: string): string {
  return p.replace(/^\.?\//, "").replace(/\.md$/, "");
}

export function useJournalStream() {
  return state;
}

export { state as journalStreamState };