From af27883a68de22557f854ad36055a35e80f9af48 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 12:34:11 -0700 Subject: [PATCH] refactor(tables): adopt shared durable event-log core --- .../table/[tableId]/events/stream/route.ts | 152 +-------- apps/sim/lib/realtime/event-log.test.ts | 86 +++++ apps/sim/lib/realtime/event-log.ts | 280 +++++++++++++++++ apps/sim/lib/realtime/event-stream-route.ts | 160 ++++++++++ apps/sim/lib/table/events.ts | 293 +++--------------- 5 files changed, 579 insertions(+), 392 deletions(-) create mode 100644 apps/sim/lib/realtime/event-log.test.ts create mode 100644 apps/sim/lib/realtime/event-log.ts create mode 100644 apps/sim/lib/realtime/event-stream-route.ts diff --git a/apps/sim/app/api/table/[tableId]/events/stream/route.ts b/apps/sim/app/api/table/[tableId]/events/stream/route.ts index 2e9cff2eb37..2142e4a1d93 100644 --- a/apps/sim/app/api/table/[tableId]/events/stream/route.ts +++ b/apps/sim/app/api/table/[tableId]/events/stream/route.ts @@ -1,26 +1,13 @@ -import { createLogger } from '@sim/logger' -import { toError } from '@sim/utils/errors' -import { sleep } from '@sim/utils/helpers' import { type NextRequest, NextResponse } from 'next/server' import { tableEventStreamContract } from '@/lib/api/contracts/tables' import { parseRequest } from '@/lib/api/server' import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid' import { generateRequestId } from '@/lib/core/utils/request' -import { SSE_HEADERS } from '@/lib/core/utils/sse' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' -import { - getLatestTableEventId, - readTableEventsSince, - type TableEventEntry, -} from '@/lib/table/events' +import { createEventStreamResponse } from '@/lib/realtime/event-stream-route' +import { getLatestTableEventId, readTableEventsSince } from '@/lib/table/events' import { accessError, checkAccess } from '@/app/api/table/utils' -const logger = createLogger('TableEventStreamAPI') - -const POLL_INTERVAL_MS = 500 -const HEARTBEAT_INTERVAL_MS = 15_000 -const MAX_STREAM_DURATION_MS = 4 * 60 * 60 * 1000 // 4 hours; client reconnects past this - export const runtime = 'nodejs' export const dynamic = 'force-dynamic' @@ -30,12 +17,9 @@ interface RouteContext { /** GET /api/table/[tableId]/events/stream?from= * - * SSE stream of cell-state transitions. Replay-on-reconnect via `from`; - * absent `from` tails from the latest event id (fresh mount — the client has - * just fetched current state, so replaying history would rewind it). - * Pruning (buffer cap exceeded or TTL expired) sends a `pruned` event and - * closes; the client responds with a full row-query refetch and reconnects - * tailing from latest. */ + * SSE stream of cell-state transitions over the shared durable event log. Auth + * and access are checked here; the replay/tail/poll/heartbeat/prune mechanics + * come from `createEventStreamResponse`. */ export const GET = withRouteHandler(async (req: NextRequest, context: RouteContext) => { const requestId = generateRequestId() const parsed = await parseRequest(tableEventStreamContract, req, context) @@ -51,123 +35,13 @@ export const GET = withRouteHandler(async (req: NextRequest, context: RouteConte const access = await checkAccess(tableId, auth.userId, 'read') if (!access.ok) return accessError(access, requestId, tableId) - logger.info(`[${requestId}] Table event stream opened`, { tableId, fromEventId }) - - const encoder = new TextEncoder() - let closed = false - - const stream = new ReadableStream({ - async start(controller) { - let lastEventId = fromEventId ?? 0 - const deadline = Date.now() + MAX_STREAM_DURATION_MS - let nextHeartbeatAt = Date.now() + HEARTBEAT_INTERVAL_MS - - const enqueue = (text: string) => { - if (closed) return - try { - controller.enqueue(encoder.encode(text)) - } catch { - closed = true - } - } - - const sendEvents = (events: TableEventEntry[]) => { - for (const entry of events) { - if (closed) return - enqueue(`data: ${JSON.stringify(entry)}\n\n`) - lastEventId = entry.eventId - } - } - - const sendPrunedAndClose = (earliestEventId: number | undefined) => { - enqueue( - `event: pruned\ndata: ${JSON.stringify({ earliestEventId: earliestEventId ?? null })}\n\n` - ) - if (!closed) { - closed = true - try { - controller.close() - } catch {} - } - } - - const sendHeartbeat = () => { - // SSE comment line — keeps proxies (ALB default 60s idle) from closing - // the connection during quiet periods. - enqueue(`: ping ${Date.now()}\n\n`) - } - - try { - // No replay cursor → tail from the latest event id. Resolved inside - // the try so a Redis failure errors the stream (client reconnects - // with backoff) rather than silently replaying the whole buffer. - if (fromEventId === undefined) { - lastEventId = await getLatestTableEventId(tableId) - } - // Initial replay from buffer. - const initial = await readTableEventsSince(tableId, lastEventId) - if (initial.status === 'pruned') { - sendPrunedAndClose(initial.earliestEventId) - return - } - if (initial.status === 'unavailable') { - throw new Error(`Table event buffer unavailable: ${initial.error}`) - } - sendEvents(initial.events) - - // Stream loop — poll the buffer and forward new events. Workflow - // execution stream uses the same shape; pub/sub wakeups are an - // optimization we can add later if 500ms polling becomes a problem. - while (!closed && Date.now() < deadline) { - await sleep(POLL_INTERVAL_MS) - if (closed) return - - const result = await readTableEventsSince(tableId, lastEventId) - if (result.status === 'pruned') { - sendPrunedAndClose(result.earliestEventId) - return - } - if (result.status === 'unavailable') { - throw new Error(`Table event buffer unavailable: ${result.error}`) - } - if (result.events.length > 0) { - sendEvents(result.events) - } - - if (Date.now() >= nextHeartbeatAt) { - sendHeartbeat() - nextHeartbeatAt = Date.now() + HEARTBEAT_INTERVAL_MS - } - } - - // Reached the defensive duration ceiling — close cleanly so the client - // reconnects with the latest lastEventId. - if (!closed) { - enqueue(`event: rotate\ndata: {}\n\n`) - closed = true - try { - controller.close() - } catch {} - } - } catch (error) { - logger.error(`[${requestId}] Table event stream error`, { - tableId, - error: toError(error).message, - }) - if (!closed) { - try { - controller.error(error) - } catch {} - } - } - }, - cancel() { - closed = true - logger.info(`[${requestId}] Client disconnected from table event stream`, { tableId }) - }, - }) - - return new NextResponse(stream, { - headers: { ...SSE_HEADERS, 'X-Table-Id': tableId }, + return createEventStreamResponse({ + requestId, + label: 'table', + streamId: tableId, + fromEventId, + getLatestEventId: getLatestTableEventId, + readEventsSince: readTableEventsSince, + extraHeaders: { 'X-Table-Id': tableId }, }) }) diff --git a/apps/sim/lib/realtime/event-log.test.ts b/apps/sim/lib/realtime/event-log.test.ts new file mode 100644 index 00000000000..34d113d79f5 --- /dev/null +++ b/apps/sim/lib/realtime/event-log.test.ts @@ -0,0 +1,86 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +vi.mock('@/lib/core/config/env', () => ({ env: { REDIS_URL: undefined } })) +vi.mock('@/lib/core/config/redis', () => ({ getRedisClient: () => null })) + +import { + appendEvent, + type EventLogConfig, + type EventLogEntry, + getLatestEventId, + readEventsSince, + resetEventLogMemoryForTesting, +} from '@/lib/realtime/event-log' + +interface TestEntry extends EventLogEntry { + eventId: number + streamId: string + value: string +} + +const config: EventLogConfig = { prefix: 'test:stream:', ttlSeconds: 3600, cap: 3, readChunk: 500 } + +function serializerFor(streamId: string, value: string) { + return { + entryPrefix: '{"eventId":', + entrySuffix: `,"streamId":${JSON.stringify(streamId)},"value":${JSON.stringify(value)}}`, + buildMemory: (eventId: number): TestEntry => ({ eventId, streamId, value }), + } +} + +describe('event-log (memory fallback)', () => { + beforeEach(() => resetEventLogMemoryForTesting()) + + it('assigns monotonically increasing event ids', async () => { + const first = await appendEvent(config, 's1', serializerFor('s1', 'a')) + const second = await appendEvent(config, 's1', serializerFor('s1', 'b')) + expect(first?.eventId).toBe(1) + expect(second?.eventId).toBe(2) + }) + + it('isolates streams by id', async () => { + await appendEvent(config, 's1', serializerFor('s1', 'a')) + const other = await appendEvent(config, 's2', serializerFor('s2', 'x')) + expect(other?.eventId).toBe(1) + expect(await getLatestEventId(config, 's1')).toBe(1) + expect(await getLatestEventId(config, 's2')).toBe(1) + }) + + it('reads only events after the cursor', async () => { + await appendEvent(config, 's1', serializerFor('s1', 'a')) + await appendEvent(config, 's1', serializerFor('s1', 'b')) + const result = await readEventsSince(config, 's1', 1) + expect(result.status).toBe('ok') + if (result.status === 'ok') { + expect(result.events).toHaveLength(1) + expect(result.events[0].eventId).toBe(2) + expect(result.events[0].value).toBe('b') + } + }) + + it('tails from the latest id and returns nothing for a fresh cursor', async () => { + await appendEvent(config, 's1', serializerFor('s1', 'a')) + await appendEvent(config, 's1', serializerFor('s1', 'b')) + const latest = await getLatestEventId(config, 's1') + const result = await readEventsSince(config, 's1', latest) + expect(result).toEqual({ status: 'ok', events: [] }) + }) + + it('reports pruned when the cursor falls behind the cap-trimmed buffer', async () => { + // cap = 3; append 5, so the earliest retained id is 3. + for (const v of ['a', 'b', 'c', 'd', 'e']) { + await appendEvent(config, 's1', serializerFor('s1', v)) + } + const result = await readEventsSince(config, 's1', 1) + expect(result.status).toBe('pruned') + if (result.status === 'pruned') expect(result.earliestEventId).toBe(3) + }) + + it('reports pruned for a non-zero cursor against a never-seen stream', async () => { + const result = await readEventsSince(config, 'missing', 5) + expect(result.status).toBe('pruned') + }) +}) diff --git a/apps/sim/lib/realtime/event-log.ts b/apps/sim/lib/realtime/event-log.ts new file mode 100644 index 00000000000..fd107a774a2 --- /dev/null +++ b/apps/sim/lib/realtime/event-log.ts @@ -0,0 +1,280 @@ +/** + * Generic durable event log over Redis (sorted set + monotonic id + TTL), with an + * in-memory fallback for dev/tests. This is the reusable core extracted from the + * Tables cell-event buffer; a domain adapter (e.g. `lib/table/events.ts`) supplies + * its Redis key prefix and how to serialize an entry, and gets append/read/tail + * semantics for free — including replay-on-reconnect and prune detection. + * + * The core is deliberately domain-neutral: it only knows an entry has a numeric + * `eventId`. Everything else in the entry is opaque bytes the adapter owns, so a + * domain can keep its exact wire shape (and its existing Redis keys) unchanged. + * + * Modeled after `apps/sim/lib/execution/event-buffer.ts` but stripped of what an + * always-on stream doesn't need (no id-reservation batching, no write-queue + * serialization, no per-entity terminal lifecycle, no byte budgeting). + */ + +import { createLogger } from '@sim/logger' +import { toError } from '@sim/utils/errors' +import { env } from '@/lib/core/config/env' +import { getRedisClient } from '@/lib/core/config/redis' + +const logger = createLogger('EventLog') + +/** + * Atomic append: INCR the seq counter to mint a new eventId, splice it into the + * adapter-supplied entry JSON, ZADD it, refresh TTLs, trim to cap, and record the + * resulting earliestEventId in meta — one round-trip. Without atomicity a slow + * reader could observe the trim before the meta update and miss the prune signal. + * + * KEYS: [events, seq, meta] + * ARGV: [ttlSec, cap, updatedAtIso, entryPrefix, entrySuffix] + * The new eventId is spliced between prefix/suffix to form the entry JSON. + * Returns the new eventId. + */ +const APPEND_EVENT_SCRIPT = ` +local eventId = redis.call('INCR', KEYS[2]) +local entry = ARGV[4] .. eventId .. ARGV[5] +redis.call('ZADD', KEYS[1], eventId, entry) +redis.call('EXPIRE', KEYS[1], tonumber(ARGV[1])) +redis.call('EXPIRE', KEYS[2], tonumber(ARGV[1])) +redis.call('ZREMRANGEBYRANK', KEYS[1], 0, -tonumber(ARGV[2]) - 1) +local oldest = redis.call('ZRANGE', KEYS[1], 0, 0, 'WITHSCORES') +if oldest[2] then + redis.call('HSET', KEYS[3], 'earliestEventId', tostring(math.floor(tonumber(oldest[2]))), 'updatedAt', ARGV[3]) + redis.call('EXPIRE', KEYS[3], tonumber(ARGV[1])) +end +return eventId +` + +/** Configuration for a durable event-log stream family (e.g. Tables). */ +export interface EventLogConfig { + /** + * Redis key prefix, e.g. `table:stream:`. STABLE per family — renaming it resets + * the seq counter, which silently strands live clients holding a higher + * in-memory `lastEventId` (their `?from=` never matches). Never change it. + */ + prefix: string + ttlSeconds: number + cap: number + /** Max entries returned by one read; the SSE route drains in chunks. */ + readChunk: number +} + +/** Minimal shape the core requires; adapters extend it with their own fields. */ +export interface EventLogEntry { + eventId: number +} + +/** + * How an adapter serializes one entry. `entryPrefix`/`entrySuffix` are spliced + * around the minted `eventId` in Lua (`prefix + eventId + suffix`); `buildMemory` + * must produce the byte-identical object for the in-memory fallback. The two MUST + * agree — a divergence makes dev/no-Redis behave differently from prod. + */ +export interface EntrySerializer { + entryPrefix: string + entrySuffix: string + buildMemory: (eventId: number) => E +} + +export type EventLogReadResult = + | { status: 'ok'; events: E[] } + | { status: 'pruned'; earliestEventId: number | undefined } + | { status: 'unavailable'; error: string } + +function eventsKey(config: EventLogConfig, streamId: string) { + return `${config.prefix}${streamId}:events` +} +function seqKey(config: EventLogConfig, streamId: string) { + return `${config.prefix}${streamId}:seq` +} +function metaKey(config: EventLogConfig, streamId: string) { + return `${config.prefix}${streamId}:meta` +} + +interface MemoryStream { + events: E[] + earliestEventId?: number + nextEventId: number + expiresAt: number +} + +/** In-memory fallback keyed by `${prefix}${streamId}`, shared across all families. */ +const memoryStreams = new Map>() + +function memoryKey(config: EventLogConfig, streamId: string) { + return `${config.prefix}${streamId}` +} + +function canUseMemoryBuffer(): boolean { + return typeof window === 'undefined' && !env.REDIS_URL +} + +function pruneExpiredMemoryStreams(now = Date.now()): void { + for (const [key, stream] of memoryStreams) { + if (stream.expiresAt <= now) memoryStreams.delete(key) + } +} + +function getMemoryStream(config: EventLogConfig, streamId: string): MemoryStream { + pruneExpiredMemoryStreams() + const key = memoryKey(config, streamId) + let stream = memoryStreams.get(key) + if (!stream) { + stream = { events: [], nextEventId: 1, expiresAt: Date.now() + config.ttlSeconds * 1000 } + memoryStreams.set(key, stream) + } + return stream +} + +/** + * Append an event. Fire-and-forget from the caller — never throws, returns null on + * failure. A Redis blip must not fail the originating mutation. + */ +export async function appendEvent( + config: EventLogConfig, + streamId: string, + serializer: EntrySerializer +): Promise { + const redis = getRedisClient() + if (!redis) { + if (canUseMemoryBuffer()) { + try { + const stream = getMemoryStream(config, streamId) + const entry = serializer.buildMemory(stream.nextEventId++) + stream.events.push(entry) + if (stream.events.length > config.cap) { + stream.events = stream.events.slice(-config.cap) + stream.earliestEventId = stream.events[0]?.eventId + } + stream.expiresAt = Date.now() + config.ttlSeconds * 1000 + return entry + } catch (error) { + logger.warn('appendEvent: memory append failed', { + streamId, + error: toError(error).message, + }) + return null + } + } + return null + } + try { + const result = await redis.eval( + APPEND_EVENT_SCRIPT, + 3, + eventsKey(config, streamId), + seqKey(config, streamId), + metaKey(config, streamId), + config.ttlSeconds, + config.cap, + new Date().toISOString(), + serializer.entryPrefix, + serializer.entrySuffix + ) + const eventId = typeof result === 'number' ? result : Number(result) + if (!Number.isFinite(eventId)) return null + return serializer.buildMemory(eventId) + } catch (error) { + logger.warn('appendEvent: Redis append failed', { streamId, error: toError(error).message }) + return null + } +} + +/** + * The latest eventId assigned for a stream, or 0 when empty/expired. Used by the + * stream route to tail from "now" when a client connects without a replay cursor. + * Redis errors propagate so the route errors the stream instead of replaying the + * whole buffer over freshly-fetched state. + */ +export async function getLatestEventId(config: EventLogConfig, streamId: string): Promise { + const redis = getRedisClient() + if (!redis) { + if (canUseMemoryBuffer()) { + const stream = memoryStreams.get(memoryKey(config, streamId)) + return stream ? stream.nextEventId - 1 : 0 + } + return 0 + } + const raw = await redis.get(seqKey(config, streamId)) + if (!raw) return 0 + const parsed = Number.parseInt(raw, 10) + return Number.isFinite(parsed) && parsed > 0 ? parsed : 0 +} + +/** + * Read events where eventId > afterEventId. Returns 'pruned' if the caller has + * fallen off the back of the buffer (TTL expired or cap rolled past their cursor); + * the caller should full-refetch and resume from the new earliest id. + */ +export async function readEventsSince( + config: EventLogConfig, + streamId: string, + afterEventId: number +): Promise> { + const redis = getRedisClient() + if (!redis) { + if (canUseMemoryBuffer()) { + pruneExpiredMemoryStreams() + const stream = memoryStreams.get(memoryKey(config, streamId)) + if (!stream) { + if (afterEventId > 0) return { status: 'pruned', earliestEventId: undefined } + return { status: 'ok', events: [] } + } + if (stream.earliestEventId !== undefined && afterEventId + 1 < stream.earliestEventId) { + return { status: 'pruned', earliestEventId: stream.earliestEventId } + } + return { + status: 'ok', + events: stream.events + .filter((entry) => entry.eventId > afterEventId) + .slice(0, config.readChunk) as E[], + } + } + return { status: 'unavailable', error: 'Redis client unavailable' } + } + try { + const meta = await redis.hgetall(metaKey(config, streamId)) + const earliestEventId = + meta?.earliestEventId !== undefined ? Number(meta.earliestEventId) : undefined + if (earliestEventId !== undefined && afterEventId + 1 < earliestEventId) { + return { status: 'pruned', earliestEventId } + } + const raw = await redis.zrangebyscore( + eventsKey(config, streamId), + afterEventId + 1, + '+inf', + 'LIMIT', + 0, + config.readChunk + ) + if (raw.length === 0 && afterEventId > 0) { + const seqExists = await redis.exists(seqKey(config, streamId)) + if (seqExists === 0) { + return { status: 'pruned', earliestEventId: undefined } + } + } + return { + status: 'ok', + events: raw + .map((entry) => { + try { + return JSON.parse(entry) as E + } catch { + return null + } + }) + .filter((entry): entry is E => entry !== null), + } + } catch (error) { + const message = toError(error).message + logger.warn('readEventsSince failed', { streamId, error: message }) + return { status: 'unavailable', error: message } + } +} + +/** Test-only: clear the in-memory streams between cases. */ +export function resetEventLogMemoryForTesting(): void { + memoryStreams.clear() +} diff --git a/apps/sim/lib/realtime/event-stream-route.ts b/apps/sim/lib/realtime/event-stream-route.ts new file mode 100644 index 00000000000..ad4fd2e8a82 --- /dev/null +++ b/apps/sim/lib/realtime/event-stream-route.ts @@ -0,0 +1,160 @@ +import { createLogger } from '@sim/logger' +import { toError } from '@sim/utils/errors' +import { sleep } from '@sim/utils/helpers' +import { NextResponse } from 'next/server' +import { SSE_HEADERS } from '@/lib/core/utils/sse' +import type { EventLogEntry, EventLogReadResult } from '@/lib/realtime/event-log' + +const logger = createLogger('EventStreamRoute') + +const POLL_INTERVAL_MS = 500 +const HEARTBEAT_INTERVAL_MS = 15_000 +/** Defensive ceiling; the client reconnects (resuming from lastEventId) past this. */ +const MAX_STREAM_DURATION_MS = 4 * 60 * 60 * 1000 + +export interface EventStreamResponseOptions { + requestId: string + /** The durable-log stream id (e.g. a tableId). */ + streamId: string + /** Replay cursor from `?from=`; `undefined` tails from the latest event id. */ + fromEventId: number | undefined + getLatestEventId: (streamId: string) => Promise + readEventsSince: (streamId: string, afterEventId: number) => Promise> + /** Extra response headers (e.g. `{ 'X-Table-Id': id }`). */ + extraHeaders?: Record + /** Short label for logs (e.g. 'table'). */ + label: string +} + +/** + * Shared SSE stream for any durable event log (`@/lib/realtime/event-log`). Handles + * replay-on-reconnect (`?from=`), tail-from-latest on a fresh mount, chunked poll + + * forward, heartbeats, graceful `pruned`/`rotate` close, and error propagation. + * + * Auth and contract parsing stay in the route (they are domain-specific); this + * owns only the streaming mechanics, so every durable-log surface streams + * identically. The poll loop mirrors the execution stream; pub/sub wakeups are an + * optimization that can replace the 500ms poll later without changing this shape. + */ +export function createEventStreamResponse( + options: EventStreamResponseOptions +): NextResponse { + const { requestId, streamId, fromEventId, getLatestEventId, readEventsSince, label } = options + + logger.info(`[${requestId}] ${label} event stream opened`, { streamId, fromEventId }) + + const encoder = new TextEncoder() + let closed = false + + const stream = new ReadableStream({ + async start(controller) { + let lastEventId = fromEventId ?? 0 + const deadline = Date.now() + MAX_STREAM_DURATION_MS + let nextHeartbeatAt = Date.now() + HEARTBEAT_INTERVAL_MS + + const enqueue = (text: string) => { + if (closed) return + try { + controller.enqueue(encoder.encode(text)) + } catch { + closed = true + } + } + + const sendEvents = (events: E[]) => { + for (const entry of events) { + if (closed) return + enqueue(`data: ${JSON.stringify(entry)}\n\n`) + lastEventId = entry.eventId + } + } + + const sendPrunedAndClose = (earliestEventId: number | undefined) => { + enqueue( + `event: pruned\ndata: ${JSON.stringify({ earliestEventId: earliestEventId ?? null })}\n\n` + ) + if (!closed) { + closed = true + try { + controller.close() + } catch {} + } + } + + const sendHeartbeat = () => { + // SSE comment line — keeps proxies (ALB default 60s idle) from closing + // the connection during quiet periods. + enqueue(`: ping ${Date.now()}\n\n`) + } + + try { + // No replay cursor → tail from the latest event id. Resolved inside the + // try so a Redis failure errors the stream (client reconnects with + // backoff) rather than silently replaying the whole buffer. + if (fromEventId === undefined) { + lastEventId = await getLatestEventId(streamId) + } + + const initial = await readEventsSince(streamId, lastEventId) + if (initial.status === 'pruned') { + sendPrunedAndClose(initial.earliestEventId) + return + } + if (initial.status === 'unavailable') { + throw new Error(`${label} event buffer unavailable: ${initial.error}`) + } + sendEvents(initial.events) + + while (!closed && Date.now() < deadline) { + await sleep(POLL_INTERVAL_MS) + if (closed) return + + const result = await readEventsSince(streamId, lastEventId) + if (result.status === 'pruned') { + sendPrunedAndClose(result.earliestEventId) + return + } + if (result.status === 'unavailable') { + throw new Error(`${label} event buffer unavailable: ${result.error}`) + } + if (result.events.length > 0) { + sendEvents(result.events) + } + + if (Date.now() >= nextHeartbeatAt) { + sendHeartbeat() + nextHeartbeatAt = Date.now() + HEARTBEAT_INTERVAL_MS + } + } + + // Reached the defensive duration ceiling — close cleanly so the client + // reconnects with the latest lastEventId. + if (!closed) { + enqueue(`event: rotate\ndata: {}\n\n`) + closed = true + try { + controller.close() + } catch {} + } + } catch (error) { + logger.error(`[${requestId}] ${label} event stream error`, { + streamId, + error: toError(error).message, + }) + if (!closed) { + try { + controller.error(error) + } catch {} + } + } + }, + cancel() { + closed = true + logger.info(`[${requestId}] Client disconnected from ${label} event stream`, { streamId }) + }, + }) + + return new NextResponse(stream, { + headers: { ...SSE_HEADERS, ...(options.extraHeaders ?? {}) }, + }) +} diff --git a/apps/sim/lib/table/events.ts b/apps/sim/lib/table/events.ts index 7d1c8135fe0..27d9b3bb0ec 100644 --- a/apps/sim/lib/table/events.ts +++ b/apps/sim/lib/table/events.ts @@ -1,69 +1,38 @@ /** * Per-table event buffer for live cell-state updates. * - * The grid subscribes to a per-table SSE stream and patches its React Query - * cache as events arrive. This buffer is the durable mid-tier between the - * cell-write paths (`writeWorkflowGroupState`, `cancelWorkflowGroupRuns`) and - * the SSE consumers — every status transition appends here with a monotonic - * eventId; SSE clients resume on reconnect via `?from=` and the - * server replays from this buffer. + * The grid subscribes to a per-table SSE stream and patches its React Query cache + * as events arrive. This is a thin domain adapter over the generic durable event + * log (`@/lib/realtime/event-log`): it owns the Redis key prefix (`table:stream:`) + * and the entry wire shape (`{ eventId, tableId, event }`), and gets append/read/ + * tail + replay + prune semantics from the core. Every status transition appends + * here with a monotonic eventId; SSE clients resume on reconnect via + * `?from=` and the server replays from this buffer. * - * Modeled after `apps/sim/lib/execution/event-buffer.ts` but stripped of - * complexity tables don't need: no per-execution lifecycle, no id reservation - * batching, no write-queue serialization. Tables are always-on; cell writes - * are sparse and independent. + * The `table:stream:` prefix and the entry shape are a wire contract — renaming + * the prefix resets the seq counter and silently strands connected clients (their + * in-memory `lastEventId` no longer matches), so both are intentionally fixed here. */ -import { createLogger } from '@sim/logger' -import { toError } from '@sim/utils/errors' -import { env } from '@/lib/core/config/env' -import { getRedisClient } from '@/lib/core/config/redis' +import { + appendEvent, + type EventLogConfig, + type EventLogReadResult, + getLatestEventId, + readEventsSince, +} from '@/lib/realtime/event-log' -const logger = createLogger('TableEventBuffer') - -const REDIS_PREFIX = 'table:stream:' export const TABLE_EVENT_TTL_SECONDS = 60 * 60 // 1 hour export const TABLE_EVENT_CAP = 5000 /** Max events returned by a single read; the SSE route drains in chunks. */ export const TABLE_EVENT_READ_CHUNK = 500 -/** - * Atomic append: INCR the seq counter to mint a new eventId, build the entry - * JSON inline, ZADD it, refresh TTL on events + seq + meta, trim to cap, then - * write the resulting earliestEventId to meta. Single round-trip per event. - * Without atomicity a slow reader could observe the trim before the meta - * update and miss the prune signal. - * - * KEYS: [events, seq, meta] - * ARGV: [ttlSec, cap, updatedAtIso, entryPrefix, entrySuffix] - * The new eventId is spliced between prefix/suffix to form the entry JSON. - * Returns the new eventId. - */ -const APPEND_EVENT_SCRIPT = ` -local eventId = redis.call('INCR', KEYS[2]) -local entry = ARGV[4] .. eventId .. ARGV[5] -redis.call('ZADD', KEYS[1], eventId, entry) -redis.call('EXPIRE', KEYS[1], tonumber(ARGV[1])) -redis.call('EXPIRE', KEYS[2], tonumber(ARGV[1])) -redis.call('ZREMRANGEBYRANK', KEYS[1], 0, -tonumber(ARGV[2]) - 1) -local oldest = redis.call('ZRANGE', KEYS[1], 0, 0, 'WITHSCORES') -if oldest[2] then - redis.call('HSET', KEYS[3], 'earliestEventId', tostring(math.floor(tonumber(oldest[2]))), 'updatedAt', ARGV[3]) - redis.call('EXPIRE', KEYS[3], tonumber(ARGV[1])) -end -return eventId -` - -function getEventsKey(tableId: string) { - return `${REDIS_PREFIX}${tableId}:events` -} - -function getSeqKey(tableId: string) { - return `${REDIS_PREFIX}${tableId}:seq` -} - -function getMetaKey(tableId: string) { - return `${REDIS_PREFIX}${tableId}:meta` +/** Wire contract — see the file header; the prefix must never change. */ +const TABLE_EVENT_LOG: EventLogConfig = { + prefix: 'table:stream:', + ttlSeconds: TABLE_EVENT_TTL_SECONDS, + cap: TABLE_EVENT_CAP, + readChunk: TABLE_EVENT_READ_CHUNK, } export type TableCellStatus = 'pending' | 'queued' | 'running' | 'completed' | 'cancelled' | 'error' @@ -148,220 +117,38 @@ export interface TableEventEntry { event: TableEvent } -export type TableEventsReadResult = - | { status: 'ok'; events: TableEventEntry[] } - | { status: 'pruned'; earliestEventId: number | undefined } - | { status: 'unavailable'; error: string } - -/** In-memory fallback for dev/tests when Redis isn't configured. */ -interface MemoryTableStream { - events: TableEventEntry[] - earliestEventId?: number - nextEventId: number - expiresAt: number -} - -const memoryTableStreams = new Map() - -function canUseMemoryBuffer(): boolean { - return typeof window === 'undefined' && !env.REDIS_URL -} - -function pruneExpiredMemoryStreams(now = Date.now()): void { - for (const [tableId, stream] of memoryTableStreams) { - if (stream.expiresAt <= now) { - memoryTableStreams.delete(tableId) - } - } -} - -function getMemoryStream(tableId: string): MemoryTableStream { - pruneExpiredMemoryStreams() - let stream = memoryTableStreams.get(tableId) - if (!stream) { - stream = { - events: [], - nextEventId: 1, - expiresAt: Date.now() + TABLE_EVENT_TTL_SECONDS * 1000, - } - memoryTableStreams.set(tableId, stream) - } - return stream -} - -function appendMemory(event: TableEvent): TableEventEntry { - const stream = getMemoryStream(event.tableId) - const entry: TableEventEntry = { - eventId: stream.nextEventId++, - tableId: event.tableId, - event, - } - stream.events.push(entry) - if (stream.events.length > TABLE_EVENT_CAP) { - stream.events = stream.events.slice(-TABLE_EVENT_CAP) - stream.earliestEventId = stream.events[0]?.eventId - } - stream.expiresAt = Date.now() + TABLE_EVENT_TTL_SECONDS * 1000 - return entry -} - -function readMemory(tableId: string, afterEventId: number): TableEventsReadResult { - pruneExpiredMemoryStreams() - const stream = memoryTableStreams.get(tableId) - if (!stream) { - // Mirror the Redis path: a non-zero afterEventId with no buffer at all - // means TTL expired or the stream never existed; either way the caller's - // cursor is stale. - if (afterEventId > 0) return { status: 'pruned', earliestEventId: undefined } - return { status: 'ok', events: [] } - } - if (stream.earliestEventId !== undefined && afterEventId + 1 < stream.earliestEventId) { - return { status: 'pruned', earliestEventId: stream.earliestEventId } - } - return { - status: 'ok', - events: stream.events - .filter((entry) => entry.eventId > afterEventId) - .slice(0, TABLE_EVENT_READ_CHUNK), - } -} +export type TableEventsReadResult = EventLogReadResult /** - * Append an event to the table's buffer. Fire-and-forget from the caller — - * this never throws, returns null on failure. A Redis blip must not fail a - * cell-write. + * Append an event to the table's buffer. Fire-and-forget — never throws, returns + * null on failure (a Redis blip must not fail a cell-write). The Redis (Lua splice) + * and in-memory paths are built to produce byte-identical entries. */ export async function appendTableEvent(event: TableEvent): Promise { - const redis = getRedisClient() - if (!redis) { - if (canUseMemoryBuffer()) { - try { - return appendMemory(event) - } catch (error) { - logger.warn('appendTableEvent: memory append failed', { - tableId: event.tableId, - error: toError(error).message, - }) - return null - } - } - return null - } - try { - // Build the entry JSON in two halves so Lua can splice the new eventId - // between them without us needing a round-trip just to mint the id first. - const tail = `,"tableId":${JSON.stringify(event.tableId)},"event":${JSON.stringify(event)}}` - const head = `{"eventId":` - const result = await redis.eval( - APPEND_EVENT_SCRIPT, - 3, - getEventsKey(event.tableId), - getSeqKey(event.tableId), - getMetaKey(event.tableId), - TABLE_EVENT_TTL_SECONDS, - TABLE_EVENT_CAP, - new Date().toISOString(), - head, - tail - ) - const eventId = typeof result === 'number' ? result : Number(result) - if (!Number.isFinite(eventId)) return null - return { eventId, tableId: event.tableId, event } - } catch (error) { - logger.warn('appendTableEvent: Redis append failed', { - tableId: event.tableId, - error: toError(error).message, - }) - return null - } + return appendEvent(TABLE_EVENT_LOG, event.tableId, { + entryPrefix: '{"eventId":', + entrySuffix: `,"tableId":${JSON.stringify(event.tableId)},"event":${JSON.stringify(event)}}`, + buildMemory: (eventId) => ({ eventId, tableId: event.tableId, event }), + }) } /** * The latest eventId assigned for a table, or 0 when the buffer is empty or * expired. Used by the stream route to tail from "now" when a client connects - * without a replay cursor (fresh mount — its caches were just fetched from - * the DB, so replaying history would only rewind them). - * - * Redis errors propagate: silently falling back to 0 would replay the whole - * buffer over fresh state — the exact churn tail-from-latest exists to avoid. - * The stream route errors the stream instead and the client reconnects with - * backoff. + * without a replay cursor. */ -export async function getLatestTableEventId(tableId: string): Promise { - const redis = getRedisClient() - if (!redis) { - if (canUseMemoryBuffer()) { - // Pure read — getMemoryStream() would allocate a stream as a side effect. - const stream = memoryTableStreams.get(tableId) - return stream ? stream.nextEventId - 1 : 0 - } - return 0 - } - const raw = await redis.get(getSeqKey(tableId)) - if (!raw) return 0 - const parsed = Number.parseInt(raw, 10) - return Number.isFinite(parsed) && parsed > 0 ? parsed : 0 +export function getLatestTableEventId(tableId: string): Promise { + return getLatestEventId(TABLE_EVENT_LOG, tableId) } /** - * Read events for a table where eventId > afterEventId. Returns 'pruned' if - * the caller has fallen off the back of the buffer (TTL expired or cap rolled - * past their lastEventId). Caller should respond by full-refetching from DB - * and resuming streaming from the new earliestEventId. + * Read events for a table where eventId > afterEventId. Returns 'pruned' if the + * caller has fallen off the back of the buffer (TTL expired or cap rolled past + * their lastEventId). */ -export async function readTableEventsSince( +export function readTableEventsSince( tableId: string, afterEventId: number ): Promise { - const redis = getRedisClient() - if (!redis) { - if (canUseMemoryBuffer()) { - return readMemory(tableId, afterEventId) - } - return { status: 'unavailable', error: 'Redis client unavailable' } - } - try { - const meta = await redis.hgetall(getMetaKey(tableId)) - const earliestEventId = - meta?.earliestEventId !== undefined ? Number(meta.earliestEventId) : undefined - if (earliestEventId !== undefined && afterEventId + 1 < earliestEventId) { - return { status: 'pruned', earliestEventId } - } - // Read in capped chunks so a 5000-event backlog doesn't materialize as one - // multi-MB Redis reply + JSON parse + SSE flush. The route loop drains - // chunks across ticks. - const raw = await redis.zrangebyscore( - getEventsKey(tableId), - afterEventId + 1, - '+inf', - 'LIMIT', - 0, - TABLE_EVENT_READ_CHUNK - ) - if (raw.length === 0 && afterEventId > 0) { - // Total TTL expiry: events + meta both gone. The seq counter has the - // same TTL — its absence means the buffer was wiped and the caller's - // `afterEventId` is stale. Signal pruned so the client refetches. - const seqExists = await redis.exists(getSeqKey(tableId)) - if (seqExists === 0) { - return { status: 'pruned', earliestEventId: undefined } - } - } - return { - status: 'ok', - events: raw - .map((entry) => { - try { - return JSON.parse(entry) as TableEventEntry - } catch { - return null - } - }) - .filter((entry): entry is TableEventEntry => Boolean(entry)), - } - } catch (error) { - const message = toError(error).message - logger.warn('readTableEventsSince failed', { tableId, error: message }) - return { status: 'unavailable', error: message } - } + return readEventsSince(TABLE_EVENT_LOG, tableId, afterEventId) }