gitfor/platform

import { batch } from "./db";

/** Consecutive failed flushes after which a batch is split to find the rows D1 rejects on their own. */
export const ISOLATE_AFTER = 3;
const MAX_BACKOFF_MS = 60 * 60 * 1000;

export function backoffMs(attempt: number, baseMs: number): number {
  return Math.min(MAX_BACKOFF_MS, baseMs * 2 ** Math.max(0, attempt - 1));
}

export type Poison<T> = { item: T; error: string };

/**
 * Flushes items as one batch (D1 unless `runBatch` routes it elsewhere, e.g. an account's object). Once `isolate` is set (the batch has failed repeatedly), items are retried
 * one by one and those that still fail are returned as poison so the caller can dead-letter them instead of
 * blocking every later flush. If several items all fail individually D1 itself is the problem, so it throws.
 */
export async function flushIsolating<T, S = D1PreparedStatement>(
  items: T[],
  toStatements: (item: T) => S[],
  isolate: boolean,
  runBatch: (statements: S[]) => Promise<unknown> = (statements) => batch(statements as D1PreparedStatement[]),
): Promise<Poison<T>[]> {
  if (items.length === 0) return [];
  try {
    await runBatch(items.flatMap(toStatements));
    return [];
  } catch (error) {
    if (!isolate) throw error;
  }
  const poison: Poison<T>[] = [];
  for (const item of items) {
    try {
      await runBatch(toStatements(item));
    } catch (error) {
      poison.push({ item, error: String(error) });
    }
  }
  if (items.length > 1 && poison.length === items.length) throw new Error(`All ${items.length} rows failed to flush`);
  return poison;