usenet-indexer/app/services/corpus.ts

import { access, mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises'
import { join } from 'node:path'
import { deflate as deflateCb, gzip as gzipCb, unzip } from 'node:zlib'
import { promisify } from 'node:util'
import coordinatorConfig from '#config/coordinator'
import { filterOverviews } from '#lib/edge/index'
import type { ChunkSpan } from '#services/spool'

const inflate = promisify(unzip)
const deflate = promisify(deflateCb)
const gzip = promisify(gzipCb)
/** The largest inflated chunk the filter reads as one string (V8's limit is 0x1fffffe8 characters). */
const MAX_FILTER_BYTES = 256 * 1024 * 1024

/** What a filtered chunk was filtered with (CORPUS_FILTER): the rule set and the group priorities of its lease. */
export interface CorpusFilter {
  ruleVersion: number
  priorities: Record<string, number>
}
/** The rules a filtered chunk leaves out (lib/edge/corpus_filter.ts), in its range.json. */
export const FILTERED = ['crosspost', 'obfuscated']

export interface CorpusRange {
  group: string
  start: number
  end: number
}

/**
 * Stored XOVER chunks of the ranges marked `capture` (REBUILD.md, corpus):
 * `<dir>/<group>/<start>-<end>/<nnnn>-<from>-<to>.gz` with a range.json once
 * every chunk is there, the layout edge-bench reads (edge/fixtures). The
 * grouping sweeps replay them with each build's rules instead of waiting for
 * a re-index; the prefetcher reads a stored chunk instead of fetching it
 * again. Off without CORPUS_DIR. Filtered (CORPUS_FILTER; LOG.md 2026-10-23
 * (101)): the lines the edge worker only counts are left out, ~half of them,
 * and each chunk's `.drops` beside it (gzip JSON, ChunkDrops) lists their
 * article numbers by rule; its range.json says `filtered`.
 */
class Corpus {
  get dir(): string | undefined {
    return coordinatorConfig.corpus.dir
  }

  private rangeDir(range: CorpusRange, dir = this.dir!) {
    return join(dir, range.group, `${range.start}-${range.end}`)
  }

  private path(range: CorpusRange, span: ChunkSpan, dir = this.dir!) {
    const index = Math.floor((span.from - range.start) / coordinatorConfig.chunkArticles)
    return join(this.rangeDir(range, dir), `${String(index).padStart(4, '0')}-${span.from}-${span.to}.gz`)
  }

  /** Whether every chunk of the range is stored (its range.json). */
  async complete(range: CorpusRange): Promise<boolean> {
    if (!this.dir) return false
    return access(join(this.rangeDir(range), 'range.json')).then(
      () => true,
      () => false
    )
  }

  /** Whether one chunk is stored. */
  async has(range: CorpusRange, span: ChunkSpan): Promise<boolean> {
    if (!this.dir) return false
    return access(this.path(range, span)).then(
      () => true,
      () => false
    )
  }

  /** A stored chunk, or null (`dir`: another corpus than CORPUS_DIR's, the namer's tests). */
  async read(range: CorpusRange, span: ChunkSpan, dir = this.dir): Promise<Buffer | null> {
    if (!dir) return null
    return readFile(this.path(range, span, dir)).catch(() => null)
  }

  /**
   * Stores a chunk that inflates (a reply cut short isn't kept), and the
   * range.json once the range is complete (`total` chunks). With `filter`,
   * without the lines the edge worker drops on their own, compressed again
   * as the news server's (zlib), with its `.drops` written first.
   */
  async write(range: CorpusRange, span: ChunkSpan, data: Buffer, total: number, filter?: CorpusFilter) {
    if (!this.dir) return
    let raw: Buffer
    try {
      raw = await inflate(data)
    } catch {
      return
    }
    const dir = this.rangeDir(range)
    const file = this.path(range, span)
    await mkdir(dir, { recursive: true })
    let stored = data
    // Past V8's longest string (0x1fffffe8) the text can't be filtered; such a chunk is stored as it came (the
    // worker drops the same lines itself). Seen once: a chunk inflating past 512 MB (LOG.md 2026-10-24 (102)).
    if (filter && raw.length < MAX_FILTER_BYTES) {
      // Latin-1 both ways: every byte back as it was.
      const { kept, drops } = filterOverviews(raw.toString('latin1'), range.group, filter.priorities, filter.ruleVersion)
      stored = await deflate(Buffer.from(kept, 'latin1'), { level: 9 })
      await writeFile(`${file}.drops.part`, await gzip(JSON.stringify(drops)))
      await rename(`${file}.drops.part`, `${file}.drops`)
    }
    await writeFile(`${file}.part`, stored)
    await rename(`${file}.part`, file)
    const chunks = (await readdir(dir)).filter((f) => f.endsWith('.gz')).length
    if (chunks >= total) await writeFile(join(dir, 'range.json'), JSON.stringify(filter ? { ...range, filtered: FILTERED } : range))
  }

  /** A chunk a worker couldn't inflate after all: not served again, and the range not complete. */
  async removeChunk(range: CorpusRange, span: ChunkSpan) {
    if (!this.dir) return
    await rm(this.path(range, span), { force: true })
    await rm(`${this.path(range, span)}.drops`, { force: true })
    await rm(join(this.rangeDir(range), 'range.json'), { force: true })
  }
}

export default new Corpus()