import { promisify } from 'node:util' import net from 'node:net' import tls from 'node:tls' import { deflate, deflateSync, unzip } from 'node:zlib' const deflateAsync = promisify(deflate) const unzipAsync = promisify(unzip) export interface XoverClientOptions { host: string port: number user?: string password?: string secure?: boolean /** Hostname the TLS certificate must match, when connecting through a tunnel (default: host). */ servername?: string /** A reply with no data for this long fails, and the connection with it (inactivity, not duration). */ timeoutMs?: number /** Ask for gzip-compressed overview data (XFEATURE COMPRESS GZIP), when the server has it. */ compress?: boolean } export type Overview = [number, Record] /** The overview fields in RFC 3977 order, named as nntp-js names them (segmentFromHeader reads these). */ const FIELDS = ['subject', 'from', 'date', 'message-id', 'references', ':bytes', ':lines', 'xref'] const TERMINATOR = Buffer.from('\r\n.\r\n') const DOT_LINE = Buffer.from('.\r\n') /** * A minimal NNTP client for seeding nodes (CONCEPT.md D37): one connection, * GROUP and XOVER only, with gzip-compressed overviews. nntp-js can't read * those: after "224 … [COMPRESS=GZIP]" the lines come as one zlib stream, * followed by a plain "." line; an empty range is a 420 without body. * About six times less traffic per header (LOG.md 2026-10-05). */ export class XoverClient { private chunks: Buffer[] = [] private length = 0 private wake?: () => void private failure?: Error compressed = false private constructor( private sock: net.Socket, private timeoutMs: number ) { sock.on('data', (chunk: Buffer) => { this.chunks.push(chunk) this.length += chunk.length this.wake?.() }) const fail = (error: Error) => { this.failure ??= error this.wake?.() } sock.on('error', fail) sock.on('close', () => fail(new Error('NNTP connection closed'))) sock.on('timeout', () => { fail(new Error('NNTP reply timed out')) sock.destroy() }) } static async connect(options: XoverClientOptions): Promise { const sock = options.secure ? tls.connect({ host: options.host, port: options.port, servername: options.servername || options.host }) : net.connect({ host: options.host, port: options.port }) const client = new XoverClient(sock, options.timeoutMs ?? 120_000) // The greeting and login are short: a server holding a connection back (too many open) fails it in 30 s, not timeoutMs. sock.setTimeout(Math.min(client.timeoutMs, 30_000)) try { await client.expect(await client.line(), ['200', '201']) if (options.user) { const reply = await client.command(`AUTHINFO USER ${options.user}`) if (reply.startsWith('381')) await client.expect(await client.command(`AUTHINFO PASS ${options.password ?? ''}`), ['281']) else await client.expect(reply, ['281']) } if (options.compress) client.compressed = (await client.command('XFEATURE COMPRESS GZIP')).startsWith('290') } catch (error) { client.close() throw error } // Idle between commands is fine (a node waiting for the seed API); only replies are timed. sock.setTimeout(0) return client } async group(name: string): Promise<{ count: number; low: number; high: number }> { return this.timed(() => this.selectGroup(name)) } /** Overviews of articles `from`..`to` in the current group; an empty range gives []. */ async xover(from: number, to: number): Promise { return this.timed(() => this.overviews(from, to)) } /** * Overviews of `from`..`to` as one zlib/gzip stream, as the server sent it when it * compresses (plain replies are deflated here): the seed API passes these to nodes * without unpacking them (CONCEPT.md D38). `parseOverviews(unzipSync(raw).toString())`. */ /** LIST ACTIVE : each group's name, high and low article (low is often just 2 on big servers; GROUP gives the real one). */ async listActive(pattern: string): Promise<{ name: string; high: number; low: number }[]> { return this.timed(async () => { const reply = await this.command(`LIST ACTIVE ${pattern}`) await this.expect(reply, ['215']) return (await this.plainBody()) .split('\r\n') .filter(Boolean) .map((line) => { const [name, high, low] = line.split(' ') return { name, high: Number(high), low: Number(low) } }) }) } async xoverRaw(from: number, to: number): Promise { return this.timed(async () => { const reply = await this.command(`XOVER ${from}-${to}`) if (reply.startsWith('420') || reply.startsWith('423')) return deflateSync(Buffer.alloc(0)) await this.expect(reply, ['224']) if (reply.includes('COMPRESS=GZIP')) return (await this.compressedRaw(false)).raw // Plain replies (the relay's default): compressed off the event loop, at the fastest level. return deflateAsync(Buffer.from(await this.plainBody()), { level: 1 }) }) } /** A reply that stalls for timeoutMs fails the connection; idle time between commands doesn't count. */ private async timed(exchange: () => Promise): Promise { this.sock.setTimeout(this.timeoutMs) try { return await exchange() } finally { this.sock.setTimeout(0) } } private async selectGroup(name: string): Promise<{ count: number; low: number; high: number }> { const reply = await this.command(`GROUP ${name}`) await this.expect(reply, ['211']) const [, count, low, high] = reply.split(' ') return { count: Number(count), low: Number(low), high: Number(high) } } private async overviews(from: number, to: number): Promise { const reply = await this.command(`XOVER ${from}-${to}`) if (reply.startsWith('420') || reply.startsWith('423')) return [] await this.expect(reply, ['224']) const data = reply.includes('COMPRESS=GZIP') ? await this.compressedBody() : await this.plainBody() return parseOverviews(data) } close() { try { this.sock.end('QUIT\r\n') } catch { // already closed } this.sock.destroy() } private async command(line: string): Promise { this.sock.write(`${line}\r\n`) return this.line() } private async expect(reply: string, codes: string[]) { if (!codes.some((code) => reply.startsWith(code))) throw new Error(`NNTP: unexpected reply "${reply}"`) } private buffer(): Buffer { if (this.chunks.length > 1) this.chunks = [Buffer.concat(this.chunks, this.length)] return this.chunks[0] ?? Buffer.alloc(0) } private take(n: number): Buffer { const all = this.buffer() const out = all.subarray(0, n) const rest = all.subarray(n) this.chunks = rest.length > 0 ? [rest] : [] this.length = rest.length return out } private async more() { if (this.failure) throw this.failure await new Promise((resolve) => (this.wake = resolve)) this.wake = undefined if (this.failure && this.length === 0) throw this.failure } private async line(): Promise { while (true) { const at = this.buffer().indexOf('\r\n') if (at !== -1) return this.take(at + 2).subarray(0, at).toString('latin1') await this.more() } } /** A plain multi-line body, up to "\r\n.\r\n", dot-unstuffed. */ private async plainBody(): Promise { let searched = 0 while (true) { const all = this.buffer() // An empty body is just ".\r\n". if (all.length >= 3 && searched === 0 && all.subarray(0, 3).equals(DOT_LINE)) { this.take(3) return '' } const at = all.indexOf(TERMINATOR, Math.max(0, searched - TERMINATOR.length)) if (at !== -1) return this.take(at + TERMINATOR.length).subarray(0, at + 2).toString('utf8').replace(/^\.\./gm, '.') searched = all.length await this.more() } } private async compressedBody(): Promise { return (await this.compressedRaw(true)).text } /** Whether what was received so far ends in `tail`, without joining the pieces. */ private endsWith(tail: Buffer): boolean { if (this.length < tail.length) return false let need = tail.length for (let i = this.chunks.length - 1; i >= 0 && need > 0; i--) { const chunk = this.chunks[i] const n = Math.min(need, chunk.length) if (!chunk.subarray(chunk.length - n).equals(tail.subarray(need - n, need))) return false need -= n } return need === 0 } /** * A zlib/gzip stream followed by a "." line. The pieces are joined only * once the data ends in ".\r\n": joining them on every arrival was * quadratic, ~15% of a prefetcher's time (LOG.md 2026-10-09). * * Raw (the coordinator's spool) it is complete then: inflating every reply * to be sure cost the coordinator ~1.5 cores at ~630 k articles/s (mostly * GC of the inflated text). ".\r\n" can also end a network read inside * the compressed data (~1 in 16 M reads): that chunk doesn't inflate on * the worker, which reports it and the chunk is fetched again (REBUILD.md, * prefetch), and the rest of the reply breaks this connection's next * command. With the text wanted, it must inflate. */ private async compressedRaw(wantText: boolean): Promise<{ raw: Buffer; text: string }> { while (true) { if (this.length > DOT_LINE.length && this.endsWith(DOT_LINE)) { const all = this.buffer() const raw = all.subarray(0, all.length - DOT_LINE.length) if (!wantText) { this.take(all.length) return { raw, text: '' } } try { const inflated = await unzipAsync(raw) this.take(all.length) return { raw, text: inflated.toString('utf8') } } catch { // ".\r\n" inside the compressed data: wait for more. } } await this.more() } } } /** Overview lines (tab-separated, after the article number) as nntp-js returns them. */ export function parseOverviews(data: string): Overview[] { const out: Overview[] = [] for (const line of data.split('\r\n')) { if (!line) continue const [number, ...tokens] = line.split('\t') const fields: Record = {} for (let i = 0; i < FIELDS.length && i < tokens.length; i++) { // Full fields after the defaults carry their name: "Xref: host group:number". fields[FIELDS[i]] = i >= 7 ? tokens[i].replace(/^[^:\s]+:\s*/, '') : tokens[i] } out.push([Number.parseInt(number, 10), fields]) } return out }