Function readTableEventsSince

apps/sim/lib/table/events.ts:285–340

(
  tableId: string,
  afterEventId: number
)

 * and resuming streaming from the new earliestEventId.
 */
export async function readTableEventsSince(
  tableId: string,
  afterEventId: number
): Promise<TableEventsReadResult> {
  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 }
  }
}