usenet-indexer/lib/nntp/xover_client.ts

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<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).
 */
/** The server doesn't have the article (430): taken down or expired. */
export class MissingArticle extends Error {}

export class XoverClient {
  private chunks: Buffer[] = []
  private length = 0
  private wake?: () => void
  private failure?: Error
  compressed = false
  /**
   * The last xoverRaw's timing, ms: until the status line came (the server
   * thinking), from there to the end (the transfer), and the longest silence
   * during the transfer.
   */
  lastTiming = { replyMs: 0, bodyMs: 0, longestGapMs: 0 }
  private measuring = false
  /** Bytes received and not consumed yet: during a reply, how much of it came in. */
  get buffered() {
    return this.length
  }
  private lastDataAt = 0
  private longestGap = 0

  private constructor(
    private sock: net.Socket,
    private timeoutMs: number
  ) {
    sock.on('data', (chunk: Buffer) => {
      if (this.measuring) {
        const now = performance.now()
        this.longestGap = Math.max(this.longestGap, now - this.lastDataAt)
        this.lastDataAt = now
      }
      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)
    // 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<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())`.
   */
  /** LIST ACTIVE <pattern>: 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'])
      // With XFEATURE COMPRESS GZIP on, Eweka compresses this list too ("215 NewsGroups Follow [COMPRESS=GZIP]").
      const body = reply.includes('COMPRESS=GZIP') ? (await this.compressedRaw(true)).text : await this.plainBody()
      return body
        .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<Buffer> {
    return this.timed(async () => {
      const sentAt = performance.now()
      const reply = await this.command(`XOVER ${from}-${to}`)
      const repliedAt = performance.now()
      if (reply.startsWith('420') || reply.startsWith('423')) return deflateSync(Buffer.alloc(0))
      await this.expect(reply, ['224'])
      this.measuring = true
      this.lastDataAt = repliedAt
      this.longestGap = 0
      try {
        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 })
      } finally {
        this.measuring = false
        this.lastTiming = { replyMs: repliedAt - sentAt, bodyMs: performance.now() - repliedAt, longestGapMs: this.longestGap }
      }
    })
  }

  /** 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()
    }
  }

  /**
   * An article's body by message-id, raw (dot-stuffed, "\r\n.\r\n" cut off):
   * yEnc decoding undoes the stuffing. A connection without COMPRESS GZIP.
   * 430 (no such article) throws MissingArticle.
   */
  async body(messageId: string): Promise<Buffer> {
    return this.timed(async () => {
      const id = messageId.startsWith('<') ? messageId : `<${messageId}>`
      const reply = await this.command(`BODY ${id}`)
      if (reply.startsWith('430') || reply.startsWith('423')) throw new MissingArticle(reply)
      await this.expect(reply, ['222'])
      return this.rawBody()
    })
  }

  /** A multi-line body up to "\r\n.\r\n", as bytes, the terminator cut off. */
  private async rawBody(): Promise<Buffer> {
    let searched = 0
    while (true) {
      const all = this.buffer()
      if (all.length >= 3 && searched === 0 && all.subarray(0, 3).equals(DOT_LINE)) {
        this.take(3)
        return Buffer.alloc(0)
      }
      const at = all.indexOf(TERMINATOR, Math.max(0, searched - TERMINATOR.length))
      if (at !== -1) return Buffer.from(this.take(at + TERMINATOR.length).subarray(0, at + 2))
      searched = all.length
      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(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<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
}