usenet-indexer/lib/edge/range_indexer.ts

import { hash } from 'node:crypto'
import type { Overview } from '../nntp/index.js'
import { FileAggregator, type AggregatedFile, isArchiveStart, isLaterVolume, sevenZipVolume } from '../files/index.js'
import { RuleSet, isRandomFile, isScrambledFile, uploadTool, type Rules } from '../rules/index.js'
import { classKey, isUnreadableName, pendingClass } from '../release/index.js'
import { explored } from '../rules/upload_tools.js'
import { SegmentScanner, scanOverviewText, type EdgeSegment } from './scan.js'
import { encodeChunk, type ChunkFile } from '../chunks/index.js'

/** A release's files from one batch (REBUILD.md, chunk store): what the catalog adds up, and the encoded chunk. */
export interface ReleaseChunk {
  key: string
  /** Which upload of the key in this batch (0, 1, …; uploadsOf): the catalog keeps uploads apart. */
  upload: number
  name: string
  source: string
  fileTotal: number | null
  poster: string
  groups: string[]
  postedAt: number | null
  /** The latest article's date: with postedAt, the upload's posting window. */
  postedLast: number | null
  files: number
  segments: number
  bytes: number
  firstArticle: number
  lastArticle: number
  /** A file with a readable name; a par2 file. */
  readable: boolean
  hasPar2: boolean
  /**
   * Voyager 3: the window closed on the release (default), or it is a tail release sent provisionally, finished by a
   * later chunk of the same release key (the totals then are the release's; the payload only the files not sent yet).
   */
  finished?: boolean
  /** Still open at its block's end, or posted within the window of its start: the block merger joins it with the neighbour block's part. */
  edge?: boolean
  /** A file of the release misses parts (the window closed on it): not complete, whatever the counter says. */
  partial?: boolean
  /**
   * Voyager 3, an unnamed release (readable false): the upload tool from its first article's message-id, its
   * signature with the group; the payload is empty (no chunk until a probe names it: pending_releases).
   */
  tool?: string
  payload: Buffer
}

export interface DroppedChunk extends ReleaseChunk {
  rule: string
  ruleKey: string
}

export type CandidateKind = 'archive' | 'par2' | 'lone' | '7z-tail'

/** An article phase 2 should read (REBUILD.md, probe queue). */
export interface ProbeCandidate {
  releaseKey: string
  /** Its chunk's upload (ReleaseChunk.upload). */
  upload: number
  kind: CandidateKind
  messageId: string
  file: string
  /** The target article's bytes: the cheapest probe first within a class (LOG.md (130)). */
  bytes?: number
}

/** What a rule dropped in a batch (REBUILD.md, rules and recovery). */
export interface RuleCount {
  rule: 'crosspost' | 'obfuscated' | 'poster' | 'executable' | 'garbage' | 'per-article' | 'stub-archive' | 'nonsense-piece' | 'tiny-archive' | 'incomplete'
  key: string
  articles: number
  files: number
}

/**
 * A file cut off by the range's edges, for the neighbouring range: its
 * details once, and the segments this range had, as
 * [part, message-id, bytes, article number].
 */
export interface Fragment {
  fileKey: string
  name: string
  poster: string
  subject: string
  totalParts: number
  groups: string[]
  release: { key: string; name: string; source: string; fileTotal?: number } | null
  postedAt: number | null
  segments: [number, string, number, number][]
}

export interface EdgeBatch {
  /** The range's resume point this batch starts at. */
  from: number
  /** First article number not covered yet. */
  next: number
  headers: number
  /**
   * The batch's posting times without its earliest and latest 1% (ms; null without a dated article): the range's
   * time span is their min and max (headers dated years off don't stretch it). The settler's final completeness
   * reads it (LOG.md 2026-10-16 (67)).
   */
  postedLow: number | null
  postedHigh: number | null
  releases: ReleaseChunk[]
  samples: DroppedChunk[]
  candidates: ProbeCandidate[]
  counts: RuleCount[]
  /** The range path's cut-off files (the TypeScript indexer); the Rust edge's window has none, the store ignores them. */
  fragments?: Fragment[]
}

export interface RangeIndexerOptions {
  group: string
  /** Inclusive article numbers. */
  start: number
  end: number
  /** Where this worker starts: the range's resume point (a range taken over midway); its first batch starts here. */
  resumeAt?: number
  rules: Rules
  /** Chunks without a new part before an open file is given up (as a fragment near an edge). */
  idleChunks?: number
  /**
   * A file whose first part seen is above 1 and comes within this many
   * articles of the range's start is kept as a fragment (its earlier parts
   * are in the previous range). Default: the idle window (idleChunks of the
   * chunks seen so far), as uploads in busy groups spread over minutes.
   */
  edgeArticles?: number
  /**
   * A file given up short of parts is a file of its release all the same (partial), as the Rust edge's window writes
   * it, not a fragment: the pending namer reads such a release again (LOG.md 2026-10-29 (121)).
   */
  partialFiles?: boolean
}

const MAX_CANDIDATES = 4

/**
 * One leased range on an edge worker (REBUILD.md, part 2): overview lines
 * in, chunk by chunk and in order; releases, probe candidates, rule counts
 * and edge fragments out, per batch (`take`). Holds only the files still
 * waiting for parts.
 */
export class RangeIndexer {
  private aggregator = new FileAggregator<EdgeSegment>(UPLOAD_GAP_MS)
  private scanner = new SegmentScanner()
  private segments: EdgeSegment[] = []
  /** Posting times of this batch's segments (postedSpan at take). */
  private dates: number[] = []
  private rules: RuleSet
  private batch: EdgeBatch
  private counts = new Map<string, RuleCount>()
  private readonly idle: number
  private readonly edge: number
  /** Articles in the first chunk (for the default start window). */
  private chunkArticles = 0

  constructor(private options: RangeIndexerOptions) {
    this.rules = new RuleSet(options.rules)
    this.idle = options.idleChunks ?? 5
    this.edge = options.edgeArticles ?? 0
    this.batch = this.empty(options.resumeAt ?? options.start)
  }

  /** One chunk of overview lines (parsed); `next` is the first article after it. */
  add(overviews: Overview[], next: number) {
    for (const [n, h] of overviews) this.line(n, h.subject ?? '', h.from ?? '', h.date ?? '', h['message-id'] ?? '', h[':bytes'] ?? '', h.xref ?? '')
    this.endChunk(next)
  }

  /** One chunk as the XOVER reply's text (the worker's path: no object per line). */
  addText(text: string, next: number) {
    scanOverviewText(text, this.handleLine)
    this.endChunk(next)
  }

  private handleLine = (n: number, subject: string, from: string, date: string, messageId: string, bytes: string, xref: string) =>
    this.line(n, subject, from, date, messageId, bytes, xref)

  private line(articleNumber: number, subject: string, from: string, date: string, messageId: string, bytes: string, xref: string) {
    if (!Number.isFinite(articleNumber) || !messageId) return
    const { group, rules } = this.options
    this.batch.headers++
    const owner = xref ? this.scanner.owner(xref, group, rules.priorities) : group
    if (owner !== group) {
      this.count('crosspost', owner, 1, 0)
      return
    }
    const segment = this.scanner.segment(articleNumber, subject, from, date, messageId, bytes, xref, group)
    if (segment) {
      this.segments.push(segment)
      if (Number.isFinite(segment.postedAt)) this.dates.push(segment.postedAt)
    } else this.count('obfuscated', '', 1, 0)
  }

  private endChunk(next: number) {
    this.files(this.aggregator.add(this.segments))
    this.segments = []
    if (this.chunkArticles === 0) this.chunkArticles = next - this.batch.from
    this.cutOff(this.aggregator.evict(this.idle), false)
    this.batch.next = next
  }

  /** The end of the range: every open file is given up. */
  finish() {
    this.cutOff(this.aggregator.evict(0), true)
    this.batch.next = this.options.end + 1
  }

  /** What was found since the last take, and a fresh batch. */
  take(): EdgeBatch {
    this.judgeReleases()
    this.scanner.clear()
    const out = this.batch
    out.counts = [...this.counts.values()]
    this.counts.clear()
    ;[out.postedLow, out.postedHigh] = postedSpan(this.dates)
    this.dates = []
    this.batch = this.empty(out.next)
    return out
  }

  get openFiles() {
    return this.aggregator.openFiles
  }

  /** Files completed and fragments cut since the last take: how big the next batch is. */
  get pending(): number {
    let files = this.batch.fragments!.length
    for (const list of this.completed.values()) files += list.length
    return files
  }

  /** Completed files, gathered per release until the batch is taken: one chunk per release per batch. */
  private completed = new Map<string, AggregatedFile<EdgeSegment>[]>()

  private files(files: AggregatedFile<EdgeSegment>[]) {
    for (const file of files) {
      const key = releaseKeyOf(file.segments[0])
      const list = this.completed.get(key)
      if (list) list.push(file)
      else this.completed.set(key, [file])
    }
  }

  /** Judges and encodes the releases completed in this batch: each upload of a key on its own. */
  private judgeReleases() {
    const byRelease = this.completed
    this.completed = new Map()
    for (const [key, all] of byRelease) {
      uploadsOf(all).forEach((list, upload) => {
        // Settled as far as this indexer knows: it judges a release's files as each batch completes them (the Rust
        // window passes whether the window closed on the release and no block edge cut it).
        const verdict = this.rules.judge(
          key,
          list.map((f) => ({
            name: f.segments[0].file.name,
            totalParts: f.segments[0].totalParts,
            groups: f.segments[0].file.groups,
            firstMessageId: f.segments[0].messageId,
            poster: f.segments[0].file.poster,
            segments: f.segments.length,
            bytes: f.segments.reduce((n, s) => n + s.bytes, 0),
          })),
          true
        )
        if (verdict.rule === null) {
          const chunk = chunkOf(key, list, upload)
          if (!chunk.readable) {
            // Unnamed: a class the ladder found never names anything stays out of the pending store, its explored
            // sample excepted (so the rate keeps being measured); counted as garbage (LOG.md (130)).
            const tool = uploadTool(list[0].segments[0].messageId)
            const klass = pendingClass(chunk.name, chunk.bytes, chunk.partial === true)
            if (this.rules.classes.has(classKey(tool, this.options.group, klass)) && !explored(key, this.options.rules.samplePercent)) {
              this.count('garbage', `${tool} ${this.options.group} #${klass}`, list.reduce((n, f) => n + f.segments.length, 0), list.length)
              return
            }
          }
          this.batch.releases.push(chunk)
          this.batch.candidates.push(...candidatesOf(key, list, upload))
          return
        }
        this.count(verdict.rule, verdict.key, list.reduce((n, f) => n + f.segments.length, 0), list.length)
        if (verdict.sample) this.batch.samples.push({ ...chunkOf(key, list, upload), rule: verdict.rule, ruleKey: verdict.key })
      })
    }
  }

  /**
   * Given-up files. Kept as fragments only when the range's edge can have
   * cut them: parts missing before the first one seen, which comes within
   * the start window, or after the last one seen while the file was still
   * open at the range's end. Any other file missing parts is incomplete on
   * the server, and only counted (in the first 31 ranges, 390k of 406k
   * fragment files never met a neighbour; LOG.md 2026-10-09).
   */
  private cutOff(segments: EdgeSegment[], atEnd: boolean) {
    const { start } = this.options
    const window = this.edge || this.idle * Math.max(1, this.chunkArticles)
    const byFile = new Map<string, EdgeSegment[]>()
    for (const s of segments) {
      const list = byFile.get(s.fileKey)
      if (list) list.push(s)
      else byFile.set(s.fileKey, [s])
    }
    for (const list of byFile.values()) {
      if (this.options.partialFiles) {
        // As the Rust window's files(): one article of a multi-part random file is no release; a scrambled name only
        // before the end, where the rest of the file can still come.
        const name = list[0].file.name
        if (list.length === 1 && (list[0].number !== 1 || list[0].totalParts > 1) && (isRandomFile(name) || (!atEnd && isScrambledFile(name)))) {
          this.count('per-article', uploadTool(list[0].messageId), 1, 0)
          continue
        }
        // One segment per part (two copies of a file share its key here), the first article seen.
        const parts = new Map<number, EdgeSegment>()
        for (const s of list.sort((a, b) => a.articleNumber - b.articleNumber)) if (!parts.has(s.number)) parts.set(s.number, s)
        this.files([{ fileKey: list[0].fileKey, segments: [...parts.values()].sort((a, b) => a.number - b.number) }])
        continue
      }
      // One article of a "file" of many parts under a random name: a random subject per article, which
      // never completes from headers (99.8% of the boneless fragments; LOG.md 2026-10-09).
      if (list.length === 1 && isRandomFile(list[0].file.name)) {
        this.count('per-article', uploadTool(list[0].messageId), 1, 0)
        continue
      }
      let minPart = Number.POSITIVE_INFINITY
      let maxPart = 0
      let firstArticle = Number.POSITIVE_INFINITY
      let lastArticle = 0
      for (const s of list) {
        if (s.number < minPart) minPart = s.number
        if (s.number > maxPart) maxPart = s.number
        if (s.articleNumber < firstArticle) firstArticle = s.articleNumber
        if (s.articleNumber > lastArticle) lastArticle = s.articleNumber
      }
      // Cut at the end: still open when the range ends (it had parts within the idle window), later parts missing.
      const cutAtStart = minPart > 1 && firstArticle < start + window
      const cutAtEnd = atEnd && maxPart < list[0].totalParts
      if (!cutAtStart && !cutAtEnd) {
        this.count('incomplete', '', list.length, 1)
        continue
      }
      const fragment = fragmentOf(list[0])
      for (const s of list) {
        fragment.segments.push([s.number, s.messageId, s.bytes, s.articleNumber])
        if (Number.isFinite(s.postedAt) && (fragment.postedAt === null || s.postedAt < fragment.postedAt)) fragment.postedAt = s.postedAt
      }
      this.batch.fragments!.push(fragment)
    }
  }

  private count(rule: RuleCount['rule'], key: string, articles: number, files: number) {
    const id = `${rule}\n${key}`
    const c = this.counts.get(id)
    if (c) {
      c.articles += articles
      c.files += files
    } else {
      this.counts.set(id, { rule, key, articles, files })
    }
  }

  private empty(from: number): EdgeBatch {
    return { from, next: from, headers: 0, postedLow: null, postedHigh: null, releases: [], samples: [], candidates: [], counts: [], fragments: [] }
  }
}

/** Posting times without the earliest and latest 1% (floor(n / 100) at each end); nulls without any. */
export function postedSpan(dates: number[]): [number | null, number | null] {
  if (dates.length === 0) return [null, null]
  const sorted = Float64Array.from(dates).sort()
  const k = Math.floor(sorted.length / 100)
  return [sorted[k], sorted[sorted.length - 1 - k]]
}

/** The release key; a file without a recognisable release becomes a release of its own. */
export function releaseKeyOf(s: EdgeSegment): string {
  return s.file.release?.key ?? hash('sha1', `file\n${s.fileKey}`)
}

/**
 * A pause of more than this in posting starts another upload of the same
 * release key (a re-upload): 91% of uploads post within 3 h, 73% of
 * re-uploaded copies come later (LOG.md 2026-10-09 (10)).
 */
export const UPLOAD_GAP_MS = 3 * 60 * 60 * 1000

/**
 * A key's files split into uploads where posting pauses for more than
 * UPLOAD_GAP_MS: files in order of their first article's date; a file
 * starts another upload when it begins more than UPLOAD_GAP_MS after the
 * latest article posted so far (so a file that takes hours to post doesn't
 * split its upload). Earliest upload first, each in the files' own order;
 * files without a date go with the first.
 */
export function uploadsOf<F extends AggregatedFile<EdgeSegment>>(files: F[]): F[][] {
  if (files.length < 2) return [files]
  const first = files.map((f) => postedOf(f.segments))
  const last = files.map((f) => latestOf(f.segments))
  const dated = files.map((_, i) => i).filter((i) => first[i] !== null).sort((a, b) => first[a]! - first[b]! || a - b)
  if (dated.length < 2) return [files]
  const upload = new Array<number>(files.length).fill(0)
  let current = 0
  let end = last[dated[0]]!
  for (let k = 1; k < dated.length; k++) {
    const i = dated[k]
    if (first[i]! - end > UPLOAD_GAP_MS) {
      current++
      end = last[i]!
    } else if (last[i]! > end) {
      end = last[i]!
    }
    upload[i] = current
  }
  if (current === 0) return [files]
  return Array.from({ length: current + 1 }, (_, u) => files.filter((_, i) => upload[i] === u))
}

/** The latest article date of a file's segments, or null. */
function latestOf(segments: EdgeSegment[]): number | null {
  let max: number | null = null
  for (const s of segments) if (Number.isFinite(s.postedAt) && (max === null || s.postedAt > max)) max = s.postedAt
  return max
}

function postedOf(segments: EdgeSegment[]): number | null {
  let min: number | null = null
  for (const s of segments) {
    const t = s.postedAt
    if (Number.isFinite(t) && (min === null || t < min)) min = t
  }
  return min
}

export function chunkOf(key: string, files: AggregatedFile<EdgeSegment>[], upload = 0): ReleaseChunk {
  const first = files[0].segments[0].file
  const chunkFiles: ChunkFile[] = files.map((f) => {
    const m = f.segments[0].file
    return { name: m.name, poster: m.poster, subject: m.subject, postedAt: postedOf(f.segments), groups: m.groups, segments: f.segments.map((x) => [x.number, x.messageId, x.bytes]) }
  })
  let segments = 0
  let bytes = 0
  let firstArticle = Number.POSITIVE_INFINITY
  let lastArticle = 0
  let postedAt: number | null = null
  let postedLast: number | null = null
  for (const f of files) {
    for (const s of f.segments) {
      segments++
      bytes += s.bytes
      const n = s.articleNumber
      if (n < firstArticle) firstArticle = n
      if (n > lastArticle) lastArticle = n
      if (Number.isFinite(s.postedAt) && (postedLast === null || s.postedAt > postedLast)) postedLast = s.postedAt
    }
    const t = postedOf(f.segments)
    if (t !== null && (postedAt === null || t < postedAt)) postedAt = t
  }
  return {
    key,
    upload,
    name: first.release?.name ?? first.name,
    source: first.release?.source ?? 'file',
    fileTotal: files.find((f) => f.segments[0].file.release?.fileTotal)?.segments[0].file.release?.fileTotal ?? null,
    poster: first.poster,
    groups: first.groups,
    postedAt,
    postedLast,
    files: files.length,
    segments,
    bytes,
    firstArticle,
    lastArticle,
    readable: chunkFiles.some((f) => !isUnreadableName(f.name)),
    hasPar2: chunkFiles.some((f) => /\.par2$/i.test(f.name)),
    partial: files.some((f) => f.segments.length < f.segments[0].totalParts),
    payload: encodeChunk(chunkFiles),
  }
}

/**
 * The articles that say what a release is: the first of its first archive
 * volume, of its par2 index, of a lone random-named file, and a 7z set's
 * last volume (its listing is in the tail). At most a few per release.
 */
export function candidatesOf(key: string, files: AggregatedFile<EdgeSegment>[], upload = 0): ProbeCandidate[] {
  const out: ProbeCandidate[] = []
  let tail: { volume: number; file: AggregatedFile<EdgeSegment> } | null = null
  for (const f of files) {
    const name = f.segments[0].file.name
    const volume = sevenZipVolume(name)
    if (volume !== null && volume > 1) {
      if (!tail || volume > tail.volume) tail = { volume, file: f }
      continue
    }
    // An unsplit 7z of more than one segment: its listing is in its last one.
    if (volume === 0 && f.segments.length > 1 && !tail) tail = { volume, file: f }
    if (isArchiveStart(name)) out.push({ releaseKey: key, upload, kind: 'archive', messageId: f.segments[0].messageId, file: name, bytes: f.segments[0].bytes })
    else if (/\.par2$/i.test(name) && !isLaterVolume(name)) out.push({ releaseKey: key, upload, kind: 'par2', messageId: f.segments[0].messageId, file: name, bytes: f.segments[0].bytes })
  }
  if (tail) out.push({ releaseKey: key, upload, kind: '7z-tail', messageId: tail.file.segments.at(-1)!.messageId, file: tail.file.segments[0].file.name, bytes: tail.file.segments.at(-1)!.bytes })
  if (out.length === 0 && files.length === 1 && isRandomFile(files[0].segments[0].file.name)) {
    out.push({ releaseKey: key, upload, kind: 'lone', messageId: files[0].segments[0].messageId, file: files[0].segments[0].file.name, bytes: files[0].segments[0].bytes })
  }
  return out.slice(0, MAX_CANDIDATES)
}

function fragmentOf(s: EdgeSegment): Fragment {
  const m = s.file
  return {
    fileKey: s.fileKey,
    name: m.name,
    poster: m.poster,
    subject: m.subject,
    totalParts: m.totalParts,
    groups: m.groups,
    release: m.release ? { key: m.release.key, name: m.release.name, source: m.release.source, ...(m.release.fileTotal ? { fileTotal: m.release.fileTotal } : {}) } : null,
    postedAt: null,
    segments: [],
  }
}