dirtydishes/islandflow
1
0
Fork 0
Code Issues 2 Pull requests 4 Projects Releases Packages Wiki Activity Actions 1

Add replay for flow alerts and hits

Browse source
This commit is contained in:
dirtydishes 2026-01-09 17:09:05 -05:00
parent 2752025fbc
commit 6951dddfdf
3 changed files with 142 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

78
apps/web/app/page.tsx
View file

@ -1957,27 +1957,41 @@ export default function HomePage() {
getReplayKey: disableReplayGrouping
});
const flowHold = useCallback(() => !flowScroll.isAtTopRef.current, [flowScroll.isAtTopRef]);
const flow = useFlowStream(
mode === "live",
flowScroll.onNewItems,
flowAnchor.capture,
flowHold,
flowScroll.resumeTick
);
const alerts = useLiveStream<AlertEvent>({
enabled: mode === "live",
wsPath: "/ws/alerts",
expectedType: "alert",
onNewItems: alertsScroll.onNewItems,
captureScroll: alertsAnchor.capture
const flow = useTape<FlowPacket>({
mode,
wsPath: "/ws/flow",
replayPath: "/replay/flow",
latestPath: "/flow/packets",
expectedType: "flow-packet",
batchSize: mode === "replay" ? 120 : undefined,
pollMs: mode === "replay" ? 200 : undefined,
captureScroll: flowAnchor.capture,
onNewItems: flowScroll.onNewItems,
getReplayKey: disableReplayGrouping
});
const classifierHits = useLiveStream<ClassifierHitEvent>({
enabled: mode === "live",
const alerts = useTape<AlertEvent>({
mode,
wsPath: "/ws/alerts",
replayPath: "/replay/alerts",
latestPath: "/flow/alerts",
expectedType: "alert",
batchSize: mode === "replay" ? 120 : undefined,
pollMs: mode === "replay" ? 200 : undefined,
captureScroll: alertsAnchor.capture,
onNewItems: alertsScroll.onNewItems,
getReplayKey: disableReplayGrouping
});
const classifierHits = useTape<ClassifierHitEvent>({
mode,
wsPath: "/ws/classifier-hits",
replayPath: "/replay/classifier-hits",
latestPath: "/flow/classifier-hits",
expectedType: "classifier-hit",
batchSize: mode === "replay" ? 120 : undefined,
pollMs: mode === "replay" ? 200 : undefined,
captureScroll: classifierAnchor.capture,
onNewItems: classifierScroll.onNewItems,
captureScroll: classifierAnchor.capture
getReplayKey: disableReplayGrouping
});
useLayoutEffect(() => {
@ -2498,7 +2512,7 @@ export default function HomePage() {
<div className="card-header">
<div>
<h2>Flow Packets</h2>
<p className="card-subtitle">Deterministic clusters (live only).</p>
<p className="card-subtitle">Deterministic clusters.</p>
</div>
</div>
<div className="card-controls">
@ -2521,13 +2535,13 @@ export default function HomePage() {
<div className="card-body">
<div className="list" ref={flowScroll.listRef}>
{mode !== "live" ? (
<div className="empty">Flow packets are live-only in this build.</div>
) : filteredFlow.length === 0 ? (
{filteredFlow.length === 0 ? (
<div className="empty">
{tickerSet.size > 0
? "No flow packets match the current filter."
: "No flow packets yet. Start compute."}
: mode === "live"
? "No flow packets yet. Start compute."
: "Replay queue empty. Ensure ClickHouse has data."}
</div>
) : (
filteredFlow.map((packet) => {
@ -2640,7 +2654,7 @@ export default function HomePage() {
replayComplete={alerts.replayComplete}
paused={alerts.paused}
dropped={alerts.dropped}
mode="live"
mode={mode}
onTogglePause={alerts.togglePause}
/>
<TapeControls
@ -2653,13 +2667,13 @@ export default function HomePage() {
<div className="card-body">
<AlertSeverityStrip alerts={filteredAlerts} />
<div className="list" ref={alertsScroll.listRef}>
{mode !== "live" ? (
<div className="empty">Alerts are live-only in this build.</div>
) : filteredAlerts.length === 0 ? (
{filteredAlerts.length === 0 ? (
<div className="empty">
{tickerSet.size > 0
? "No alerts match the current filter."
: "No alerts yet. Start compute."}
: mode === "live"
? "No alerts yet. Start compute."
: "Replay queue empty. Ensure ClickHouse has data."}
</div>
) : (
filteredAlerts.map((alert) => {
@ -2716,7 +2730,7 @@ export default function HomePage() {
replayComplete={classifierHits.replayComplete}
paused={classifierHits.paused}
dropped={classifierHits.dropped}
mode="live"
mode={mode}
onTogglePause={classifierHits.togglePause}
/>
<TapeControls
@ -2728,13 +2742,13 @@ export default function HomePage() {
<div className="card-body">
<div className="list" ref={classifierScroll.listRef}>
{mode !== "live" ? (
<div className="empty">Classifier hits are live-only in this build.</div>
) : filteredClassifierHits.length === 0 ? (
{filteredClassifierHits.length === 0 ? (
<div className="empty">
{tickerSet.size > 0
? "No classifier hits match the current filter."
: "No classifier hits yet. Start compute."}
: mode === "live"
? "No classifier hits yet. Start compute."
: "Replay queue empty. Ensure ClickHouse has data."}
</div>
) : (
filteredClassifierHits.map((hit) => {

69
packages/storage/src/clickhouse.ts
View file

@ -860,3 +860,72 @@ export const fetchInferredDarkAfter = async (
const events = records.map(fromInferredDarkRecord);
return InferredDarkEventSchema.array().parse(events);
};
export const fetchFlowPacketsAfter = async (
client: ClickHouseClient,
afterTs: number,
afterSeq: number,
limit: number
): Promise<FlowPacket[]> => {
const safeLimit = clampLimit(limit);
const safeAfterTs = clampCursor(afterTs);
const safeAfterSeq = clampCursor(afterSeq);
const result = await client.query({
query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
format: "JSONEachRow"
});
const rows = await result.json<unknown[]>();
const records = rows
.map(normalizeFlowPacketRow)
.filter((record): record is FlowPacketRecord => record !== null);
const packets = records.map(fromFlowPacketRecord);
return FlowPacketSchema.array().parse(packets);
};
export const fetchClassifierHitsAfter = async (
client: ClickHouseClient,
afterTs: number,
afterSeq: number,
limit: number
): Promise<ClassifierHitEvent[]> => {
const safeLimit = clampLimit(limit);
const safeAfterTs = clampCursor(afterTs);
const safeAfterSeq = clampCursor(afterSeq);
const result = await client.query({
query: `SELECT * FROM ${CLASSIFIER_HITS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
format: "JSONEachRow"
});
const rows = await result.json<unknown[]>();
const records = rows
.map(normalizeClassifierHitRow)
.filter((record): record is ClassifierHitRecord => record !== null);
const hits = records.map(fromClassifierHitRecord);
return ClassifierHitEventSchema.array().parse(hits);
};
export const fetchAlertsAfter = async (
client: ClickHouseClient,
afterTs: number,
afterSeq: number,
limit: number
): Promise<AlertEvent[]> => {
const safeLimit = clampLimit(limit);
const safeAfterTs = clampCursor(afterTs);
const safeAfterSeq = clampCursor(afterSeq);
const result = await client.query({
query: `SELECT * FROM ${ALERTS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
format: "JSONEachRow"
});
const rows = await result.json<unknown[]>();
const records = rows
.map(normalizeAlertRow)
.filter((record): record is AlertRecord => record !== null);
const alerts = records.map(fromAlertRecord);
return AlertEventSchema.array().parse(alerts);
};

27
services/api/src/index.ts
View file

@ -38,6 +38,9 @@ import {
ensureFlowPacketsTable,
ensureOptionNBBOTable,
ensureOptionPrintsTable,
fetchAlertsAfter,
fetchClassifierHitsAfter,
fetchFlowPacketsAfter,
fetchRecentAlerts,
fetchRecentClassifierHits,
fetchRecentEquityPrintJoins,
@ -916,6 +919,30 @@ const run = async () => {
return jsonResponse({ data, next });
}
if (req.method === "GET" && url.pathname === "/replay/flow") {
const { afterTs, afterSeq, limit } = parseReplayParams(url);
const data = await fetchFlowPacketsAfter(clickhouse, afterTs, afterSeq, limit);
const last = data.at(-1);
const next = last ? { ts: last.source_ts, seq: last.seq } : null;
return jsonResponse({ data, next });
}
if (req.method === "GET" && url.pathname === "/replay/classifier-hits") {
const { afterTs, afterSeq, limit } = parseReplayParams(url);
const data = await fetchClassifierHitsAfter(clickhouse, afterTs, afterSeq, limit);
const last = data.at(-1);
const next = last ? { ts: last.source_ts, seq: last.seq } : null;
return jsonResponse({ data, next });
}
if (req.method === "GET" && url.pathname === "/replay/alerts") {
const { afterTs, afterSeq, limit } = parseReplayParams(url);
const data = await fetchAlertsAfter(clickhouse, afterTs, afterSeq, limit);
const last = data.at(-1);
const next = last ? { ts: last.source_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 });