Function createExecutionEventWriter

apps/sim/lib/execution/event-buffer.ts:720–1014

(
  executionId: string,
  context: ExecutionEventWriterContext = {}
)

}

export function createExecutionEventWriter(
  executionId: string,
  context: ExecutionEventWriterContext = {}
): ExecutionEventWriter {
  const redis = getRedisClient()
  if (!redis) {
    if (canUseMemoryEventBuffer()) {
      logger.info('createExecutionEventWriter: using in-memory event buffer', { executionId })
      return createMemoryExecutionEventWriter(executionId, context)
    }
    logger.warn(
      'createExecutionEventWriter: Redis client unavailable, events will not be buffered',
      {
        executionId,
      }
    )
    return {
      write: async (event) => ({ eventId: 0, executionId, event }),
      writeTerminal: async () => {
        throw new Error(`Execution event buffer unavailable for ${executionId}`)
      },
      flush: async () => {},
      close: async () => {},
    }
  }

  let pending: ExecutionEventEntry[] = []
  let nextEventId = 0
  let maxReservedId = 0
  let flushTimer: ReturnType<typeof setTimeout> | null = null
  let consecutiveFlushFailures = 0

  const getFlushDelayMs = () => {
    if (consecutiveFlushFailures === 0) return FLUSH_INTERVAL_MS
    const backoff = Math.min(
      FLUSH_INTERVAL_MS * 2 ** Math.min(consecutiveFlushFailures, 6),
      FLUSH_MAX_RETRY_INTERVAL_MS
    )
    return backoff + randomInt(0, FLUSH_INTERVAL_MS)
  }

  const scheduleFlush = (delayMs = FLUSH_INTERVAL_MS) => {
    if (flushTimer) return
    flushTimer = setTimeout(() => {
      flushTimer = null
      void flushPending()
    }, delayMs)
  }

  const reserveIds = async (minCount: number) => {
    const reserveCount = Math.max(RESERVE_BATCH, minCount)
    const newMax = await redis.incrby(getSeqKey(executionId), reserveCount)
    const startId = newMax - reserveCount + 1
    if (nextEventId === 0 || nextEventId > maxReservedId) {
      nextEventId = startId
      maxReservedId = newMax
    }
  }