Database batch writes

Use chunks(100) to group at most 100 rows, then await each write before requesting the next batch. Use bufferTime to flush by time as well as count.

import { bufferTime, chunks, from } from "mogami-stream";

export async function writeBatches<T>(
  rows: AsyncIterable<T>,
  insertMany: (batch: T[], signal: AbortSignal) => Promise<void>,
  signal: AbortSignal,
) {
  await from(rows)
    .pipe(chunks(100))
    .forEach((batch, { signal }) => insertMany(batch, signal), { signal });
}

export async function writeTimedBatches<T>(
  rows: AsyncIterable<T>,
  insertMany: (batch: T[], signal: AbortSignal) => Promise<void>,
  signal: AbortSignal,
) {
  await from(rows)
    .pipe(bufferTime(1_000, { maxItems: 100 }))
    .forEach((batch, { signal }) => insertMany(batch, signal), { signal });
}

Normal completion flushes a final partial batch; failure or cancellation discards an incomplete batch.

chunks · bufferTime