boocode/apps/server/src/services/inference/content-flusher.ts

// P5: the debounced DB content-flush timer, extracted from the verbatim copy
// that lived in executeStreamPhase + the three sentinel summaries (4 sites).
// Each site streamed deltas into a local `accumulated`/`state.accumulated`
// string and threw an UPDATE at the row at most once per DB_FLUSH_INTERVAL_MS
// to bound write rate under heavy streaming.
//
// The accumulated string stays owned by the caller (stream-phase keeps it on
// the shared StreamPhaseState; the summaries keep a local) — the flusher reads
// it through a `getContent` thunk at fire time, snapshotting the latest value
// exactly as the inline `const snapshot = accumulated` did. No final flush is
// performed on drain (matches the originals): every caller writes the full
// content itself in its terminal UPDATE, so drain only cancels the pending
// timer and awaits whatever write is already chained.

import type { Sql } from '../../db.js';
import { DB_FLUSH_INTERVAL_MS } from './types.js';

export interface ContentFlusher {
  // Arm a debounced flush. No-op if one is already pending (the in-flight timer
  // will pick up the latest content via getContent when it fires).
  scheduleFlush: () => void;
  // Cancel any pending timer and await the in-flight write chain. Does NOT
  // perform a final flush — the caller's terminal UPDATE owns the final write.
  drain: () => Promise<void>;
}

export function createContentFlusher(
  sql: Sql,
  messageId: string,
  getContent: () => string,
  intervalMs: number = DB_FLUSH_INTERVAL_MS,
): ContentFlusher {
  let pendingFlushTimer: NodeJS.Timeout | null = null;
  let flushPromise: Promise<unknown> = Promise.resolve();

  const flushNow = () => {
    if (pendingFlushTimer) {
      clearTimeout(pendingFlushTimer);
      pendingFlushTimer = null;
    }
    const snapshot = getContent();
    flushPromise = flushPromise.then(() =>
      sql`UPDATE messages SET content = ${snapshot} WHERE id = ${messageId}`
    );
  };

  const scheduleFlush = () => {
    if (pendingFlushTimer) return;
    pendingFlushTimer = setTimeout(() => {
      pendingFlushTimer = null;
      flushNow();
    }, intervalMs);
  };

  const drain = async () => {
    if (pendingFlushTimer) {
      clearTimeout(pendingFlushTimer);
      pendingFlushTimer = null;
    }
    await flushPromise;
  };

  return { scheduleFlush, drain };
}