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 D1 batch. 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>(
  items: T[],
  toStatements: (item: T) => D1PreparedStatement[],
  isolate: boolean,
): Promise<Poison<T>[]> {
  if (items.length === 0) return [];
  try {
    await batch(items.flatMap(toStatements));
    return [];
  } catch (error) {
    if (!isolate) throw error;
  }
  const poison: Poison<T>[] = [];
  for (const item of items) {
    try {
      await batch(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;
}