Add replay mode for tapes

apps/web/app/globals.css

@@ -9,6 +9,7 @@
   --accent: #2f6d4f;
   --accent-soft: rgba(47, 109, 79, 0.18);
   --warning: #c46f2a;
+  --replay: #1f4a7b;
   --grid: rgba(82, 64, 36, 0.12);
 }
 
@@ -69,7 +70,7 @@ h1 {
 
 .summary {
   display: grid;
-  gap: 8px;
+  gap: 12px;
   padding: 16px 20px;
   border-radius: 16px;
   border: 1px solid var(--panel-border);
@@ -88,6 +89,23 @@ h1 {
   font-size: 1rem;
 }
 
+.mode-button {
+  border: 1px solid rgba(31, 74, 123, 0.35);
+  border-radius: 999px;
+  padding: 8px 14px;
+  background: rgba(31, 74, 123, 0.12);
+  color: #1f4a7b;
+  font-size: 0.75rem;
+  letter-spacing: 0.12em;
+  text-transform: uppercase;
+  cursor: pointer;
+}
+
+.mode-button:focus-visible {
+  outline: 2px solid rgba(31, 74, 123, 0.4);
+  outline-offset: 2px;
+}
+
 .cards {
   display: grid;
   gap: 28px;
@@ -126,6 +144,11 @@ h1 {
   box-shadow: 0 0 0 4px var(--accent-soft);
 }
 
+.status-replay .status-dot {
+  background: var(--replay);
+  box-shadow: 0 0 0 4px rgba(31, 74, 123, 0.18);
+}
+
 .status-connecting .status-dot {
   animation: pulse 1.2s ease-in-out infinite;
 }

apps/web/app/page.tsx

@@ -8,6 +8,8 @@ const LOCAL_HOSTS = new Set(["localhost", "127.0.0.1"]);
 
 type WsStatus = "connecting" | "connected" | "disconnected";
 
+type TapeMode = "live" | "replay";
+
 type MessageType = "option-print" | "equity-print";
 
 type StreamMessage<T> = {
@@ -15,6 +17,16 @@ type StreamMessage<T> = {
   payload: T;
 };
 
+type ReplayCursor = {
+  ts: number;
+  seq: number;
+};
+
+type ReplayResponse<T> = {
+  data: T[];
+  next: ReplayCursor | null;
+};
+
 type TapeState<T> = {
   status: WsStatus;
   items: T[];
@@ -44,6 +56,27 @@ const buildWsUrl = (path: string): string => {
   return `${wsProtocol}://${host}${path}`;
 };
 
+const buildApiUrl = (path: string): string => {
+  const envBase = process.env.NEXT_PUBLIC_API_URL;
+
+  if (envBase) {
+    const url = new URL(envBase);
+    const secure = url.protocol === "https:" || url.protocol === "wss:";
+    url.protocol = secure ? "https:" : "http:";
+    url.pathname = path;
+    url.search = "";
+    url.hash = "";
+    return url.toString();
+  }
+
+  const { protocol, hostname } = window.location;
+  const httpProtocol = protocol === "https:" ? "https" : "http";
+  const isLocal = LOCAL_HOSTS.has(hostname);
+  const host = isLocal ? `${hostname}:4000` : window.location.host;
+
+  return `${httpProtocol}://${host}${path}`;
+};
+
 const formatPrice = (price: number): string => {
   return price.toFixed(2);
 };
@@ -56,11 +89,15 @@ const formatTime = (ts: number): string => {
   return new Date(ts).toLocaleTimeString();
 };
 
-const statusLabel = (status: WsStatus, paused: boolean): string => {
+const statusLabel = (status: WsStatus, paused: boolean, mode: TapeMode): string => {
   if (paused) {
     return "Paused";
   }
 
+  if (mode === "replay") {
+    return status === "disconnected" ? "Replay Down" : "Replay";
+  }
+
   switch (status) {
     case "connected":
       return "Live";
@@ -72,7 +109,21 @@ const statusLabel = (status: WsStatus, paused: boolean): string => {
   }
 };
 
-const useTape = <T,>(path: string, expectedType: MessageType): TapeState<T> => {
+type TapeConfig<T> = {
+  mode: TapeMode;
+  wsPath: string;
+  replayPath: string;
+  expectedType: MessageType;
+  batchSize?: number;
+  pollMs?: number;
+};
+
+const useTape = <T extends { ts: number; seq: number }>(
+  config: TapeConfig<T>
+): TapeState<T> => {
+  const { mode, wsPath, replayPath, expectedType } = config;
+  const batchSize = config.batchSize ?? 40;
+  const pollMs = config.pollMs ?? 1000;
   const [status, setStatus] = useState<WsStatus>("connecting");
   const [items, setItems] = useState<T[]>([]);
   const [lastUpdate, setLastUpdate] = useState<number | null>(null);
@@ -80,6 +131,12 @@ const useTape = <T,>(path: string, expectedType: MessageType): TapeState<T> => {
   const [dropped, setDropped] = useState<number>(0);
   const reconnectRef = useRef<number | null>(null);
   const socketRef = useRef<WebSocket | null>(null);
+  const cursorRef = useRef<ReplayCursor>({ ts: 0, seq: 0 });
+  const pausedRef = useRef(paused);
+
+  useEffect(() => {
+    pausedRef.current = paused;
+  }, [paused]);
 
   const togglePause = useCallback(() => {
     setPaused((prev) => {
@@ -92,6 +149,18 @@ const useTape = <T,>(path: string, expectedType: MessageType): TapeState<T> => {
   }, []);
 
   useEffect(() => {
+    setItems([]);
+    setLastUpdate(null);
+    setDropped(0);
+    setStatus("connecting");
+    cursorRef.current = { ts: 0, seq: 0 };
+  }, [mode]);
+
+  useEffect(() => {
+    if (mode !== "live") {
+      return;
+    }
+
     let active = true;
 
     const connect = () => {
@@ -101,7 +170,7 @@ const useTape = <T,>(path: string, expectedType: MessageType): TapeState<T> => {
 
       setStatus("connecting");
 
-      const socket = new WebSocket(buildWsUrl(path));
+      const socket = new WebSocket(buildWsUrl(wsPath));
       socketRef.current = socket;
 
       socket.onopen = () => {
@@ -122,7 +191,7 @@ const useTape = <T,>(path: string, expectedType: MessageType): TapeState<T> => {
             return;
           }
 
-          if (paused) {
+          if (pausedRef.current) {
             setDropped((prev) => prev + 1);
             setLastUpdate(Date.now());
             return;
@@ -170,7 +239,61 @@ const useTape = <T,>(path: string, expectedType: MessageType): TapeState<T> => {
         socketRef.current.close();
       }
     };
-  }, [path, expectedType, paused]);
+  }, [mode, wsPath, expectedType]);
+
+  useEffect(() => {
+    if (mode !== "replay") {
+      return;
+    }
+
+    let active = true;
+
+    const poll = async () => {
+      if (!active || pausedRef.current) {
+        return;
+      }
+
+      try {
+        const cursor = cursorRef.current;
+        const url = new URL(buildApiUrl(replayPath));
+        url.searchParams.set("after_ts", cursor.ts.toString());
+        url.searchParams.set("after_seq", cursor.seq.toString());
+        url.searchParams.set("limit", batchSize.toString());
+
+        const response = await fetch(url.toString());
+        if (!response.ok) {
+          throw new Error(`Replay request failed with ${response.status}`);
+        }
+
+        const payload = (await response.json()) as ReplayResponse<T>;
+        if (payload.data.length > 0) {
+          const nextItems = [...payload.data].reverse();
+          setItems((prev) => {
+            const next = [...nextItems, ...prev];
+            return next.slice(0, MAX_ITEMS);
+          });
+          setLastUpdate(Date.now());
+        }
+
+        if (payload.next) {
+          cursorRef.current = payload.next;
+        }
+
+        setStatus("connected");
+      } catch (error) {
+        console.warn("Replay poll failed", error);
+        setStatus("disconnected");
+      }
+    };
+
+    void poll();
+    const interval = window.setInterval(poll, pollMs);
+
+    return () => {
+      active = false;
+      window.clearInterval(interval);
+    };
+  }, [mode, replayPath, batchSize, pollMs]);
 
   return { status, items, lastUpdate, paused, dropped, togglePause };
 };
@@ -180,14 +303,18 @@ type TapeStatusProps = {
   lastUpdate: number | null;
   paused: boolean;
   dropped: number;
+  mode: TapeMode;
   onTogglePause: () => void;
 };
 
-const TapeStatus = ({ status, lastUpdate, paused, dropped, onTogglePause }: TapeStatusProps) => {
+const TapeStatus = ({ status, lastUpdate, paused, dropped, mode, onTogglePause }: TapeStatusProps) => {
+  const replayClass = mode === "replay" ? "status-replay" : "";
+  const pausedClass = paused ? "status-paused" : "";
+
   return (
-    <div className={`status status-${status} status-compact ${paused ? "status-paused" : ""}`}>
+    <div className={`status status-${status} status-compact ${replayClass} ${pausedClass}`.trim()}>
       <span className="status-dot" />
-      <span>{statusLabel(status, paused)}</span>
+      <span>{statusLabel(status, paused, mode)}</span>
       {lastUpdate ? (
         <span className="timestamp">Updated {formatTime(lastUpdate)}</span>
       ) : (
@@ -204,8 +331,21 @@ const TapeStatus = ({ status, lastUpdate, paused, dropped, onTogglePause }: Tape
 };
 
 export default function HomePage() {
-  const options = useTape<OptionPrint>("/ws/options", "option-print");
+  const [mode, setMode] = useState<TapeMode>("live");
-  const equities = useTape<EquityPrint>("/ws/equities", "equity-print");
+
+  const options = useTape<OptionPrint>({
+    mode,
+    wsPath: "/ws/options",
+    replayPath: "/replay/options",
+    expectedType: "option-print"
+  });
+
+  const equities = useTape<EquityPrint>({
+    mode,
+    wsPath: "/ws/equities",
+    replayPath: "/replay/equities",
+    expectedType: "equity-print"
+  });
 
   const lastSeen = useMemo(() => {
     return [options.lastUpdate, equities.lastUpdate]
@@ -213,6 +353,10 @@ export default function HomePage() {
       .sort((a, b) => b - a)[0] ?? null;
   }, [options.lastUpdate, equities.lastUpdate]);
 
+  const toggleMode = () => {
+    setMode((prev) => (prev === "live" ? "replay" : "live"));
+  };
+
   return (
     <main className="dashboard">
       <header className="header">
@@ -220,7 +364,7 @@ export default function HomePage() {
           <p className="eyebrow">Realtime flow workspace</p>
           <h1>Islandflow</h1>
           <p className="subtitle">
-            Options + equities streaming over WebSocket from the local API gateway.
+            Options + equities streaming over WebSocket or replayed from ClickHouse.
           </p>
         </div>
         <div className="summary">
@@ -228,6 +372,9 @@ export default function HomePage() {
           <span className="summary-value">
             {lastSeen ? formatTime(lastSeen) : "Waiting for data"}
           </span>
+          <button className="mode-button" type="button" onClick={toggleMode}>
+            Switch to {mode === "live" ? "Replay" : "Live"}
+          </button>
         </div>
       </header>
 
@@ -243,13 +390,18 @@ export default function HomePage() {
               lastUpdate={options.lastUpdate}
               paused={options.paused}
               dropped={options.dropped}
+              mode={mode}
               onTogglePause={options.togglePause}
             />
           </div>
 
           <div className="list">
             {options.items.length === 0 ? (
-              <div className="empty">No option prints yet. Start ingest-options.</div>
+              <div className="empty">
+                {mode === "live"
+                  ? "No option prints yet. Start ingest-options."
+                  : "Replay queue empty. Ensure ClickHouse has data."}
+              </div>
             ) : (
               options.items.map((print) => (
                 <div className="row" key={`${print.trace_id}-${print.seq}`}>
@@ -282,13 +434,18 @@ export default function HomePage() {
               lastUpdate={equities.lastUpdate}
               paused={equities.paused}
               dropped={equities.dropped}
+              mode={mode}
               onTogglePause={equities.togglePause}
             />
           </div>
 
           <div className="list">
             {equities.items.length === 0 ? (
-              <div className="empty">No equity prints yet. Start ingest-equities.</div>
+              <div className="empty">
+                {mode === "live"
+                  ? "No equity prints yet. Start ingest-equities."
+                  : "Replay queue empty. Ensure ClickHouse has data."}
+              </div>
             ) : (
               equities.items.map((print) => (
                 <div className="row" key={`${print.trace_id}-${print.seq}`}>

packages/storage/src/clickhouse.ts

@@ -76,6 +76,14 @@ const clampLimit = (limit: number): number => {
   return Math.max(1, Math.min(1000, Math.floor(limit)));
 };
 
+const clampCursor = (value: number): number => {
+  if (!Number.isFinite(value)) {
+    return 0;
+  }
+
+  return Math.max(0, Math.floor(value));
+};
+
 const coerceNumber = (value: unknown): unknown => {
   if (typeof value === "string") {
     const parsed = Number(value);
@@ -102,6 +110,45 @@ const normalizeNumericFields = (
   return record;
 };
 
+const normalizeOptionRow = (row: unknown): unknown => {
+  if (row && typeof row === "object") {
+    return normalizeNumericFields(row as Record<string, unknown>, [
+      "source_ts",
+      "ingest_ts",
+      "seq",
+      "ts",
+      "price",
+      "size"
+    ]);
+  }
+
+  return row;
+};
+
+const normalizeEquityRow = (row: unknown): unknown => {
+  if (row && typeof row === "object") {
+    const record = normalizeNumericFields(row as Record<string, unknown>, [
+      "source_ts",
+      "ingest_ts",
+      "seq",
+      "ts",
+      "price",
+      "size"
+    ]);
+
+    if ("offExchangeFlag" in record) {
+      return {
+        ...record,
+        offExchangeFlag: Boolean(record.offExchangeFlag)
+      };
+    }
+
+    return record;
+  }
+
+  return row;
+};
+
 export const fetchRecentOptionPrints = async (
   client: ClickHouseClient,
   limit: number
@@ -113,22 +160,7 @@ export const fetchRecentOptionPrints = async (
   });
 
   const rows = await result.json<unknown[]>();
-  const normalized = rows.map((row) => {
+  return OptionPrintSchema.array().parse(rows.map(normalizeOptionRow));
-    if (row && typeof row === "object") {
-      return normalizeNumericFields(row as Record<string, unknown>, [
-        "source_ts",
-        "ingest_ts",
-        "seq",
-        "ts",
-        "price",
-        "size"
-      ]);
-    }
-
-    return row;
-  });
-
-  return OptionPrintSchema.array().parse(normalized);
 };
 
 export const fetchRecentEquityPrints = async (
@@ -142,29 +174,43 @@ export const fetchRecentEquityPrints = async (
   });
 
   const rows = await result.json<unknown[]>();
-  const normalized = rows.map((row) => {
+  return EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
-    if (row && typeof row === "object") {
+};
-      const record = normalizeNumericFields(row as Record<string, unknown>, [
-        "source_ts",
-        "ingest_ts",
-        "seq",
-        "ts",
-        "price",
-        "size"
-      ]);
 
-      if ("offExchangeFlag" in record) {
+export const fetchOptionPrintsAfter = async (
-        return {
+  client: ClickHouseClient,
-          ...record,
+  afterTs: number,
-          offExchangeFlag: Boolean(record.offExchangeFlag)
+  afterSeq: number,
-        };
+  limit: number
-      }
+): Promise<OptionPrint[]> => {
+  const safeLimit = clampLimit(limit);
+  const safeAfterTs = clampCursor(afterTs);
+  const safeAfterSeq = clampCursor(afterSeq);
 
-      return record;
+  const result = await client.query({
-    }
+    query: `SELECT * FROM ${OPTION_PRINTS_TABLE} WHERE (ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
-
+    format: "JSONEachRow"
-    return row;
   });
 
-  return EquityPrintSchema.array().parse(normalized);
+  const rows = await result.json<unknown[]>();
+  return OptionPrintSchema.array().parse(rows.map(normalizeOptionRow));
+};
+
+export const fetchEquityPrintsAfter = async (
+  client: ClickHouseClient,
+  afterTs: number,
+  afterSeq: number,
+  limit: number
+): Promise<EquityPrint[]> => {
+  const safeLimit = clampLimit(limit);
+  const safeAfterTs = clampCursor(afterTs);
+  const safeAfterSeq = clampCursor(afterSeq);
+
+  const result = await client.query({
+    query: `SELECT * FROM ${EQUITY_PRINTS_TABLE} WHERE (ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  return EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
 };

services/api/src/index.ts

@@ -14,7 +14,9 @@ import {
   createClickHouseClient,
   ensureEquityPrintsTable,
   ensureOptionPrintsTable,
+  fetchEquityPrintsAfter,
   fetchRecentEquityPrints,
+  fetchOptionPrintsAfter,
   fetchRecentOptionPrints
 } from "@islandflow/storage";
 import { EquityPrintSchema, OptionPrintSchema } from "@islandflow/types";
@@ -34,6 +36,11 @@ const envSchema = z.object({
 const env = readEnv(envSchema);
 
 const limitSchema = z.coerce.number().int().positive().max(1000);
+const replayParamsSchema = z.object({
+  after_ts: z.coerce.number().int().nonnegative().default(0),
+  after_seq: z.coerce.number().int().nonnegative().default(0),
+  limit: z.coerce.number().int().positive().max(1000).default(200)
+});
 
 type Channel = "options" | "equities";
 
@@ -61,6 +68,20 @@ const parseLimit = (value: string | null): number => {
   return limitSchema.parse(value);
 };
 
+const parseReplayParams = (url: URL): { afterTs: number; afterSeq: number; limit: number } => {
+  const params = replayParamsSchema.parse({
+    after_ts: url.searchParams.get("after_ts") ?? undefined,
+    after_seq: url.searchParams.get("after_seq") ?? undefined,
+    limit: url.searchParams.get("limit") ?? undefined
+  });
+
+  return {
+    afterTs: params.after_ts,
+    afterSeq: params.after_seq,
+    limit: params.limit
+  };
+};
+
 const broadcast = (sockets: Set<WebSocket<WsData>>, payload: unknown): void => {
   const message = JSON.stringify(payload);
 
@@ -187,6 +208,22 @@ const run = async () => {
         return jsonResponse({ data });
       }
 
+      if (req.method === "GET" && url.pathname === "/replay/options") {
+        const { afterTs, afterSeq, limit } = parseReplayParams(url);
+        const data = await fetchOptionPrintsAfter(clickhouse, afterTs, afterSeq, limit);
+        const last = data.at(-1);
+        const next = last ? { ts: last.ts, seq: last.seq } : null;
+        return jsonResponse({ data, next });
+      }
+
+      if (req.method === "GET" && url.pathname === "/replay/equities") {
+        const { afterTs, afterSeq, limit } = parseReplayParams(url);
+        const data = await fetchEquityPrintsAfter(clickhouse, afterTs, afterSeq, limit);
+        const last = data.at(-1);
+        const next = last ? { ts: last.ts, seq: last.seq } : null;
+        return jsonResponse({ data, next });
+      }
+
       if (req.method === "GET" && url.pathname === "/ws/options") {
         if (serverRef.upgrade(req, { data: { channel: "options" } })) {
           return new Response(null, { status: 101 });