usenet-indexer/lib/files/aggregator.ts

/**
 * Edge aggregation (CONCEPT.md D37): groups header segments into files in
 * memory, so a seeding node writes one row per file instead of one per
 * article. A file is done once every part number from 1 to its total has
 * an article; what never completes within the node's window (an upload cut
 * by the range edges, a random subject per article) leaves as fragments
 * for the regular ingest path, where it can meet its other parts.
 */

/** What the aggregator needs of a segment (IngestSegment has all of it). */
export interface AggregatableSegment {
  fileKey: string
  number: number
  totalParts: number
  messageId: string
  bytes: number
}

export interface AggregatedFile<S extends AggregatableSegment> {
  fileKey: string
  /** One segment per part number, in part order; the first article seen wins (a later one is a repost). */
  segments: S[]
}

interface OpenFile<S> {
  parts: Map<number, S>
  totalParts: number
  /** Batch in which a segment last arrived. */
  touched: number
}

export class FileAggregator<S extends AggregatableSegment> {
  private open = new Map<string, OpenFile<S>>()
  private batch = 0
  /** Segments with a part number outside 1..total: never part of a complete file. */
  private stray: S[] = []

  /** Adds one batch of segments; returns the files they completed. */
  add(segments: S[]): AggregatedFile<S>[] {
    this.batch++
    const touched = new Set<string>()
    for (const segment of segments) {
      if (segment.number < 1 || segment.number > segment.totalParts) {
        this.stray.push(segment)
        continue
      }
      let file = this.open.get(segment.fileKey)
      if (!file) {
        file = { parts: new Map(), totalParts: segment.totalParts, touched: this.batch }
        this.open.set(segment.fileKey, file)
      }
      if (!file.parts.has(segment.number)) file.parts.set(segment.number, segment)
      file.touched = this.batch
      touched.add(segment.fileKey)
    }

    const complete: AggregatedFile<S>[] = []
    for (const key of touched) {
      const file = this.open.get(key)!
      if (file.parts.size < file.totalParts) continue
      this.open.delete(key)
      complete.push({ fileKey: key, segments: [...file.parts.values()].sort((a, b) => a.number - b.number) })
    }
    return complete
  }

  /**
   * Gives up on files that got no segment in the last `idleBatches` batches
   * (all open files when 0) and returns their segments, plus stray ones.
   */
  evict(idleBatches = 0): S[] {
    const out = this.stray
    this.stray = []
    for (const [key, file] of this.open) {
      if (idleBatches > 0 && this.batch - file.touched < idleBatches) continue
      this.open.delete(key)
      out.push(...file.parts.values())
    }
    return out
  }

  /** Files still waiting for parts. */
  get openFiles(): number {
    return this.open.size
  }
}