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
  /** ms since 1970 (NaN or absent: unknown): tells a re-upload's copy of a file from the first. */
  postedAt?: number
}

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

interface OpenFile<S> {
  parts: Map<number, S>
  totalParts: number
  /** Batch in which a segment last arrived. */
  touched: number
  /** Posting times seen (NaN: none). */
  earliest: number
  latest: number
}

export class FileAggregator<S extends AggregatableSegment> {
  /** Per file key its open copies, oldest first: a re-upload of a file is a copy of its own. */
  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[] = []

  /**
   * `uploadGapMs`: a part already in every open copy, posted further than
   * this from all of them, starts another copy (a re-upload). A missing part
   * always fills a copy, the nearest in time (a poster re-sending what was
   * missing), and a part posted near a copy that has it is a re-send.
   */
  constructor(private uploadGapMs = 3 * 60 * 60 * 1000) {}

  /** How far `t` is from a copy's posting times (0 when unknown or within them). */
  private distance(file: OpenFile<S>, t: number | undefined): number {
    if (t === undefined || !Number.isFinite(t) || !Number.isFinite(file.earliest)) return 0
    return Math.max(0, file.earliest - t, t - file.latest)
  }

  /** 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 copies = this.open.get(segment.fileKey)
      if (!copies) {
        copies = []
        this.open.set(segment.fileKey, copies)
      }
      const t = segment.postedAt
      let file: OpenFile<S> | undefined
      let best = Number.POSITIVE_INFINITY
      for (const copy of copies) {
        if (copy.parts.has(segment.number)) continue
        const d = this.distance(copy, t)
        if (d < best) {
          best = d
          file = copy
        }
      }
      if (!file) {
        const near = copies.find((copy) => this.distance(copy, t) <= this.uploadGapMs)
        if (near) {
          // A re-send of a part this copy has: the first article stays.
          near.touched = this.batch
          touched.add(segment.fileKey)
          continue
        }
        file = { parts: new Map(), totalParts: segment.totalParts, touched: this.batch, earliest: Number.NaN, latest: Number.NaN }
        copies.push(file)
      }
      file.parts.set(segment.number, segment)
      if (t !== undefined && Number.isFinite(t)) {
        if (!(t >= file.earliest)) file.earliest = t
        if (!(t <= file.latest)) file.latest = t
      }
      file.touched = this.batch
      touched.add(segment.fileKey)
    }

    const complete: AggregatedFile<S>[] = []
    for (const key of touched) {
      const copies = this.open.get(key)!
      for (const file of [...copies]) {
        if (file.parts.size < file.totalParts) continue
        copies.splice(copies.indexOf(file), 1)
        complete.push({ fileKey: key, segments: [...file.parts.values()].sort((a, b) => a.number - b.number) })
      }
      if (copies.length === 0) this.open.delete(key)
    }
    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, copies] of this.open) {
      for (const file of [...copies]) {
        if (idleBatches > 0 && this.batch - file.touched < idleBatches) continue
        copies.splice(copies.indexOf(file), 1)
        out.push(...file.parts.values())
      }
      if (copies.length === 0) this.open.delete(key)
    }
    return out
  }

  /** Files still waiting for parts. */
  get openFiles(): number {
    let n = 0
    for (const copies of this.open.values()) n += copies.length
    return n
  }
}