Function resetExecutionStreamBuffer

apps/sim/lib/execution/event-buffer.ts:489–541

(executionId: string)

}

export async function resetExecutionStreamBuffer(executionId: string): Promise<boolean> {
  const redis = getRedisClient()
  if (!redis) {
    if (!canUseMemoryEventBuffer()) {
      logger.warn('resetExecutionStreamBuffer: Redis client unavailable', { executionId })
      return false
    }
    const stream = getMemoryStream(executionId)
    stream.events = []
    stream.meta = {
      status: 'active',
      replayStartEventId: stream.nextEventId,
      updatedAt: new Date().toISOString(),
    }
    stream.expiresAt = Date.now() + TTL_SECONDS * 1000
    return true
  }

  try {
    const currentSequence = Number(await redis.get(getSeqKey(executionId)).catch(() => 0))
    const replayStartEventId = Number.isFinite(currentSequence) ? currentSequence + 1 : 1
    const metaKey = getMetaKey(executionId)
    const meta = (await redis.hgetall(metaKey).catch(() => ({}))) as Record<string, string>
    const userId = typeof meta.userId === 'string' ? meta.userId : undefined
    const budgetReservation: ExecutionRedisBudgetReservation = {
      executionId,
      userId,
      category: 'event_buffer',
      operation: 'reset_events',
      bytes: 0,
      logger,
    }
    const budgetKeys = getExecutionRedisBudgetKeys(budgetReservation)
    await redis.eval(
      RESET_STREAM_SCRIPT,
      2 + budgetKeys.length,
      getEventsKey(executionId),
      metaKey,
      ...budgetKeys,
      String(replayStartEventId),
      new Date().toISOString(),
      TTL_SECONDS,
      getExecutionRedisBudgetLimits().ttlSeconds
    )
    return true
  } catch (error) {
    logger.warn('Failed to reset execution stream buffer', {
      executionId,
      error: toError(error).message,
    })
    return false
  }
}

export async function setExecutionMeta(
  executionId: string,