usenet-indexer/lib/seed/runner.ts

import type { XoverClient, Overview } from '../nntp/xover_client.js'
import { segmentFromHeader, type IngestSegment, type OverviewHeader } from '../ingest/segment.js'
import { FileAggregator } from '../aggregate/index.js'
import { toFragment, toSeedFile, type SeedBatch } from './wire.js'

export interface RunRangeOptions {
  client: XoverClient
  group: string
  /** Inclusive article numbers. */
  start: number
  end: number
  /** Articles per XOVER. */
  chunk?: number
  /** Batches without a new part before an open file leaves as fragments. */
  idle?: number
  /** XOVER batches per SeedBatch handed to `sink`. */
  submitEvery?: number
  /** Headers left to the deferred-body path (counted, not indexed). */
  isDeferred?: (header: OverviewHeader) => boolean
  /** Receives batches in order; the next XOVER is already in flight meanwhile. */
  sink: (batch: SeedBatch) => Promise<void>
}

/**
 * Indexes one article range on a seeding node (CONCEPT.md D37): XOVER in
 * chunks over one connection (the next chunk in flight while the previous
 * one is handled), headers parsed with the live rules, grouped into files
 * in memory; complete files and given-up fragments go to `sink` in order.
 * Shared by seed:range (stores directly) and seed:node (submits over HTTPS).
 */
export async function runRange(options: RunRangeOptions): Promise<void> {
  const chunk = options.chunk ?? 20_000
  const idle = options.idle ?? 5
  const submitEvery = options.submitEvery ?? 5
  const aggregator = new FileAggregator<IngestSegment>()
  await options.client.group(options.group)

  const fetch = (from: number) => {
    const promise = options.client.xover(from, Math.min(from + chunk - 1, options.end))
    // Awaited after the current batch is handled; until then a failure must not count as unhandled.
    promise.catch(() => {})
    return promise
  }

  let batch = emptyBatch(options.start)
  let pending = 0
  let from = options.start
  let next: Promise<Overview[]> | null = fetch(from)
  while (next) {
    const overviews = await next
    from += chunk
    next = from <= options.end ? fetch(from) : null

    const segments: IngestSegment[] = []
    for (const [articleNumber, header] of overviews) {
      if (!Number.isFinite(articleNumber) || !header['message-id']) continue
      const segment = segmentFromHeader(header, options.group, articleNumber)
      if (segment) segments.push(segment)
      else if (options.isDeferred?.(header)) batch.deferred++
      else batch.obfuscated++
    }
    batch.headers += overviews.length
    batch.files.push(...aggregator.add(segments).map(toSeedFile))
    batch.fragments.push(...aggregator.evict(idle).map(toFragment))
    batch.next = Math.min(from, options.end + 1)

    if (++pending >= submitEvery || !next) {
      if (!next) batch.fragments.push(...aggregator.evict(0).map(toFragment))
      await options.sink(batch)
      batch = emptyBatch(batch.next)
      pending = 0
    }
  }
}

function emptyBatch(next: number): SeedBatch {
  return { next, headers: 0, obfuscated: 0, deferred: 0, files: [], fragments: [] }
}