Function readNodeStreamToBufferWithLimit

apps/sim/lib/core/utils/stream-limits.ts:147–221

(
  stream: NodeJS.ReadableStream | null | undefined,
  options: ReadStreamWithLimitOptions
)

}

export async function readNodeStreamToBufferWithLimit(
  stream: NodeJS.ReadableStream | null | undefined,
  options: ReadStreamWithLimitOptions
): Promise<Buffer> {
  if (!stream) return Buffer.alloc(0)

  return new Promise((resolve, reject) => {
    const chunks: Buffer[] = []
    let totalBytes = 0
    let settled = false

    const finish = (callback: () => void) => {
      if (settled) return
      settled = true
      cleanup()
      callback()
    }

    const cleanup = () => {
      stream.off('data', onData)
      stream.off('end', onEnd)
      stream.off('error', onError)
      options.signal?.removeEventListener('abort', onAbort)
    }

    const onAbort = () => {
      if ('destroy' in stream && typeof stream.destroy === 'function') {
        stream.destroy(toError(options.signal?.reason ?? new Error('Aborted')))
      }
      finish(() => reject(toError(options.signal?.reason ?? new Error('Aborted'))))
    }

    const onData = (chunk: Buffer | Uint8Array | string) => {
      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
      totalBytes += buffer.byteLength

      if (totalBytes > options.maxBytes) {
        if ('destroy' in stream && typeof stream.destroy === 'function') {
          stream.destroy()
        }
        finish(() =>
          reject(
            new PayloadSizeLimitError({
              label: options.label,
              maxBytes: options.maxBytes,
              observedBytes: totalBytes,
            })
          )
        )
        return
      }

      void options.onChunk?.(buffer, totalBytes)
      chunks.push(buffer)
    }

    const onEnd = () => {
      finish(() => resolve(Buffer.concat(chunks, totalBytes)))