import { mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises' import { join } from 'node:path' import { unzip } from 'node:zlib' import { promisify } from 'node:util' import coordinatorConfig from '#config/coordinator' import type { ChunkSpan } from '#services/spool' const inflate = promisify(unzip) export interface CorpusRange { group: string start: number end: number } /** * Stored XOVER chunks of the ranges marked `capture` (REBUILD.md, corpus): * `//-/--.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. */ class Corpus { get dir(): string | undefined { return coordinatorConfig.corpus.dir } private rangeDir(range: CorpusRange) { return join(this.dir!, range.group, `${range.start}-${range.end}`) } private path(range: CorpusRange, span: ChunkSpan) { const index = Math.floor((span.from - range.start) / coordinatorConfig.chunkArticles) return join(this.rangeDir(range), `${String(index).padStart(4, '0')}-${span.from}-${span.to}.gz`) } /** A stored chunk, or null. */ async read(range: CorpusRange, span: ChunkSpan): Promise { if (!this.dir) return null return readFile(this.path(range, span)).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). */ async write(range: CorpusRange, span: ChunkSpan, data: Buffer, total: number) { if (!this.dir) return try { await inflate(data) } catch { return } const dir = this.rangeDir(range) const file = this.path(range, span) await mkdir(dir, { recursive: true }) await writeFile(`${file}.part`, data) 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(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(join(this.rangeDir(range), 'range.json'), { force: true }) } } export default new Corpus()