Function appendEvents

apps/sim/lib/copilot/request/session/buffer.ts:128–154

(
  envelopes: PersistedStreamEventEnvelope[]
)

}

export async function appendEvents(
  envelopes: PersistedStreamEventEnvelope[]
): Promise<PersistedStreamEventEnvelope[]> {
  if (envelopes.length === 0) {
    return envelopes
  }

  const streamId = envelopes[0].stream.streamId
  const config = getStreamConfig()

  await withRedisRetry({ operation: 'append_event', streamId }, async (redis) => {
    const key = getEventsKey(streamId)
    const seqKey = getSeqKey(streamId)
    const pipeline = redis.pipeline()
    const zaddArgs: Array<number | string> = []
    for (const envelope of envelopes) {
      zaddArgs.push(envelope.seq, JSON.stringify(envelope))
    }
    pipeline.zadd(key, ...(zaddArgs as [number, string, ...Array<number | string>]))
    pipeline.zremrangebyrank(key, 0, -config.eventLimit - 1)
    pipeline.expire(key, config.ttlSeconds)
    pipeline.set(seqKey, String(envelopes[envelopes.length - 1].seq), 'EX', config.ttlSeconds)
    await pipeline.exec()
  })

  return envelopes
}

export async function appendEvent(
  envelope: PersistedStreamEventEnvelope