feat(table): live cell updates via SSE + per-table event buffer

Replaces the polling-based row refetch with a push-based SSE stream that
patches the React Query cache directly as cell-state events arrive.

Architecture:
- New per-table event buffer in apps/sim/lib/table/events.ts. Redis sorted-set
  with monotonic eventId, 1h TTL, 5000-event cap, in-memory fallback. Modeled
  after apps/sim/lib/execution/event-buffer.ts but stripped of complexity
  tables don't need (no per-execution lifecycle, no id-batching, no write
  queue serialization). ~150 lines instead of 700.
- writeWorkflowGroupState appends a fat event after each successful 'wrote'.
  Status transitions carry executionId + jobId; terminal/partial transitions
  also include the new output values inline so the client can patch row data
  without a follow-up refetch.
- New SSE route at /api/table/[tableId]/events/stream?from=<lastEventId>.
  Replays from buffer on connect, polls at 500ms (mirrors workflow execution
  stream), heartbeat every 15s, signals 'pruned' if the caller fell off the
  back of the buffer.
- Client hook useTableEventStream subscribes via EventSource. Reconnect-resume
  with last-seen eventId. On 'pruned', invalidates the rows query and resumes
  from the new earliest. Cache patches walk every cached query under
  rowsRoot(tableId) so filter/sort variants all stay live.
- Removes refetchInterval from useTableRows and the per-page polling effect
  from useInfiniteTableRows. React Query's refetchOnWindowFocus +
  refetchOnReconnect cover the durability gap if any push is dropped.

Out of scope:
- Bulk-cancel events (cancellation path is being redesigned separately).
- Generalizing the workflow event-buffer module to a shared primitive (defer
  until a third use case appears; for now the table buffer is the simpler
  cousin of the workflow one).

Author: TheodoreSpeaks

apps/sim/app/api/table/[tableId]/events/stream/route.ts

@@ -0,0 +1,161 @@
+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 { SSE_HEADERS } from '@/lib/core/utils/sse'
+import { generateRequestId } from '@/lib/core/utils/request'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { readTableEventsSince, type TableEventEntry } 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'
+
+interface RouteContext {
+  params: Promise<{ tableId: string }>
+}
+
+/** GET /api/table/[tableId]/events/stream?from=<lastEventId>
+ *
+ *  SSE stream of cell-state transitions. Replay-on-reconnect via `from`.
+ *  Pruning (buffer cap exceeded or TTL expired) sends a `pruned` event and
+ *  closes; the client responds with a full row-query refetch and reconnects
+ *  from the new earliest. */
+export const GET = withRouteHandler(async (req: NextRequest, context: RouteContext) => {
+  const requestId = generateRequestId()
+  const parsed = await parseRequest(tableEventStreamContract, req, context)
+  if (!parsed.success) return parsed.response
+  const { tableId } = parsed.data.params
+  const { from: fromEventId } = parsed.data.query
+
+  const auth = await checkSessionOrInternalAuth(req, { requireWorkflowId: false })
+  if (!auth.success || !auth.userId) {
+    return NextResponse.json({ error: 'Authentication required' }, { status: 401 })
+  }
+
+  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<Uint8Array>({
+    async start(controller) {
+      let lastEventId = fromEventId
+      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 {
+        // 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 },
+  })
+})

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/index.ts

@@ -1,2 +1,3 @@
 export * from './use-context-menu'
 export * from './use-table'
+export * from './use-table-event-stream'