Function processBatch

scripts/pipeline/batch-processor.ts:329–584

(config: BatchConfig)

// --- Main Batch Loop ---

async function processBatch(config: BatchConfig): Promise<BatchProgress> {
  const progress: BatchProgress = {
    total: 0,
    completed: 0,
    failed: 0,
    skipped: 0,
    tier0Count: 0,
    tier1Count: 0,
    totalCostCents: 0,
    startTime: Date.now(),
  };

  // Ensure jobs exist in queue
  await ensureJobsExist(config);

  // Count total pending
  const countResult = await db
    .select({ count: sql<number>`COUNT(*)::int` })
    .from(pipelineJobs)
    .where(
      and(
        eq(pipelineJobs.jobType, "ai_analysis"),
        eq(pipelineJobs.status, "pending"),
      )
    );
  progress.total = countResult[0]?.count ?? 0;

  if (progress.total === 0) {
    console.log("\nNo pending jobs to process.");
    return progress;
  }

  console.log(`\nTotal pending jobs: ${progress.total}`);

  // Check budget
  const { totalCents: monthlySpent, docCount: monthlyDocs } = await getMonthlySpend();
  const budgetRemaining = config.monthlyCapCents - monthlySpent;

  console.log(`Monthly budget: ${formatCents(config.monthlyCapCents)} (spent: ${formatCents(monthlySpent)} on ${monthlyDocs} docs, remaining: ${formatCents(budgetRemaining)})`);

  if (budgetRemaining <= 0 && config.forceTier !== 0) {
    console.log("Monthly budget exhausted. Tier 1 analysis disabled, falling back to Tier 0.");
  }

  let processedInSession = 0;
  const limit = config.limit ?? Infinity;
  let currentBudgetRemaining = budgetRemaining;

  while (processedInSession < limit) {
    const batch = await getNextBatch(Math.min(config.batchSize, limit - processedInSession));
    if (batch.length === 0) break;

    for (const job of batch) {
      if (processedInSession >= limit) break;

      const meta = (job.metadata as any) ?? {};
      const dataSet = meta.dataSet ?? "unknown";
      const eftaNumber = meta.eftaNumber ?? "";