Rebuild: phase 1 modules from the current app, ordered by pipeline stage
lib/nntp (XOVER client, Xref, GROUP), lib/headers (subjects, file keys, segments from overview lines; the yEnc path stays behind for phase 2), lib/release (identity, random names, now in lib), lib/files (aggregator, volume names), lib/rules (posting tools, owner by priority, release verdicts), lib/chunks (release chunk codec). Their unit tests came along; new ones for codec, rules and volumes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
96f0ade10e
commit
d773c91ad0
26 files changed
+1486
No files matched your search
@@ -0,0 +1,23 @@
|
||||
export interface GroupInfo {
|
||||
count: bigint
|
||||
first: bigint
|
||||
last: bigint
|
||||
name: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Parses a `211 count first last name` GROUP reply (RFC 3977 §6.1.1).
|
||||
*
|
||||
* nntp-js's own `group()` destructures this reply without skipping the status
|
||||
* code, so its `first`/`last` are off by one field. Pass it `result.resp`.
|
||||
*/
|
||||
export function parseGroupResponse(resp: string): GroupInfo {
|
||||
const match = /^211\s+(\d+)\s+(\d+)\s+(\d+)\s+(\S+)/.exec(resp.trim())
|
||||
if (!match) throw new Error(`Unexpected GROUP response: ${resp}`)
|
||||
return {
|
||||
count: BigInt(match[1]),
|
||||
first: BigInt(match[2]),
|
||||
last: BigInt(match[3]),
|
||||
name: match[4],
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
export { XoverClient, parseOverviews, type Overview, type XoverClientOptions } from './xover_client.js'
|
||||
export { parseXrefGroups } from './xref.js'
|
||||
export { parseGroupResponse, type GroupInfo } from './group.js'
|
||||
@@ -0,0 +1,238 @@
|
||||
import { promisify } from 'node:util'
|
||||
import net from 'node:net'
|
||||
import tls from 'node:tls'
|
||||
import { deflate, deflateSync, unzipSync } from 'node:zlib'
|
||||
|
||||
const deflateAsync = promisify(deflate)
|
||||
|
||||
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<string, string>]
|
||||
|
||||
/** 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<XoverClient> {
|
||||
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)
|
||||
sock.setTimeout(client.timeoutMs)
|
||||
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<Overview[]> {
|
||||
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())`.
|
||||
*/
|
||||
async xoverRaw(from: number, to: number): Promise<Buffer> {
|
||||
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()).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<T>(exchange: () => Promise<T>): Promise<T> {
|
||||
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<Overview[]> {
|
||||
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<string> {
|
||||
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<void>((resolve) => (this.wake = resolve))
|
||||
this.wake = undefined
|
||||
if (this.failure && this.length === 0) throw this.failure
|
||||
}
|
||||
|
||||
private async line(): Promise<string> {
|
||||
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<string> {
|
||||
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<string> {
|
||||
return (await this.compressedRaw()).text
|
||||
}
|
||||
|
||||
/** A zlib/gzip stream followed by a "." line: complete once it ends in ".\r\n" and inflates. */
|
||||
private async compressedRaw(): Promise<{ raw: Buffer; text: string }> {
|
||||
while (true) {
|
||||
const all = this.buffer()
|
||||
if (all.length > DOT_LINE.length && all.subarray(all.length - DOT_LINE.length).equals(DOT_LINE)) {
|
||||
const raw = all.subarray(0, all.length - DOT_LINE.length)
|
||||
try {
|
||||
const text = unzipSync(raw).toString('utf8')
|
||||
this.take(all.length)
|
||||
return { raw: Buffer.from(raw), text }
|
||||
} 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<string, string> = {}
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
/**
|
||||
* Returns the group names from an Xref header,
|
||||
* e.g. `news.eweka.nl alt.binaries.test:123 alt.binaries.x:456`.
|
||||
*/
|
||||
export function parseXrefGroups(xref: string | undefined): string[] {
|
||||
if (!xref) return []
|
||||
return xref
|
||||
.trim()
|
||||
.split(/\s+/)
|
||||
.slice(1)
|
||||
.map((entry) => entry.slice(0, entry.lastIndexOf(':')))
|
||||
.filter(Boolean)
|
||||
}
|
||||
Reference in new issue
Block a user