usenet-indexer/app/services/pending_namer.ts

import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import { setTimeout as sleep } from 'node:timers/promises'
import { promisify } from 'node:util'
import { unzip } from 'node:zlib'
import env from '#start/env'
import coordinatorConfig from '#config/coordinator'
import { XoverClient } from '#lib/nntp/index'
import { RangeIndexer, type ReleaseChunk } from '#lib/edge/index'
import { identityKey } from '#lib/release/keys'
import BatchStore from '#services/batch_store'
import ChunkPartitions from '#services/chunk_partitions'
import Corpus from '#services/corpus'
import RuleSets from '#services/rule_sets'
import { PENDING } from '#services/pending_ladder'

const inflate = promisify(unzip)

interface NamedRow {
  release_key: Buffer
  identity_key: Buffer
  group_id: number
  named: string
  first_article: number
  last_article: number
  file_total: number | null
  tries: number
}

/** The namer gives a release up after this many failed moves (its articles gone from the server and the corpus). */
const MAX_MOVES = 3

/**
 * The widest article span the namer reads again (a busy group posts 50M articles a day, a large upload among
 * them spans a million or two): one wider is a name the edge saw again much later ("v-lgpc.3ds"), and its span
 * of segments would not fit the indexer's memory (LOG.md (133)).
 */
const MAX_SPAN = 3_000_000

/**
 * The move into the catalog (REBUILD.md decision f): a pending release a probe named (status 4) gets its segments
 * read again, from the corpus when its range is stored, else from the server (XOVER over its article span, one
 * connection of its own, opened for it and closed after), through the TypeScript indexer, and is written as a
 * release with the name the probe found (source weak-stem, the name as inner name), its chunk and its probe
 * candidates, which the prober answers from probe_cache. Runs inside the prober.
 */
export class PendingNamer {
  private client: XoverClient | null = null
  private usedAt = 0

  constructor(private corpusDir: string | undefined = coordinatorConfig.corpus.dir) {}

  async run(stopped: () => boolean) {
    while (!stopped()) {
      let rows: NamedRow[] = []
      try {
        rows = await this.take(20)
        for (const row of rows) {
          if (stopped()) break
          try {
            await this.move(row)
          } catch (error: any) {
            logger.warn(
              `Naming ${row.release_key.toString('hex')} "${row.named}" failed: ${error.message}`
            )
            await db.rawQuery(
              `UPDATE pending_releases SET tries = tries + 1, probed_at = now() WHERE release_key = ?`,
              [row.release_key]
            )
          }
        }
      } catch (error: any) {
        logger.error(`Naming: ${error.message}`)
      }
      if (this.client && Date.now() - this.usedAt > coordinatorConfig.nntpIdleCloseMs) {
        this.client.close()
        this.client = null
      }
      if (rows.length === 0)
        for (let waited = 0; waited < 30_000 && !stopped(); waited += 1000) await sleep(1000)
    }
    this.client?.close()
    this.client = null
  }

  /** Named releases awaiting their move, the longest waiting first; those failed MAX_MOVES times stay as they are. */
  async take(n: number): Promise<NamedRow[]> {
    const { rows } = await db.rawQuery(
      `SELECT release_key, identity_key, group_id, named, first_article, last_article, file_total, tries
       FROM pending_releases WHERE status = ? AND tries < ? ORDER BY probed_at LIMIT ?`,
      [PENDING.naming, MAX_MOVES, n]
    )
    return rows.map((r: any) => ({
      ...r,
      group_id: Number(r.group_id),
      first_article: Number(r.first_article),
      last_article: Number(r.last_article),
      tries: Number(r.tries),
    }))
  }

  /** One named release into the catalog. */
  async move(row: NamedRow): Promise<void> {
    if (row.last_article - row.first_article > MAX_SPAN)
      throw new Error(
        `its articles ${row.first_article}–${row.last_article} span more than ${MAX_SPAN}`
      )
    const group = await db.from('groups').where('id', row.group_id).firstOrFail()
    const range = await db
      .from('ranges')
      .where('group_id', row.group_id)
      .where('start_id', '<=', row.first_article)
      .where('end_id', '>=', row.first_article)
      .orderBy('id')
      .first()
    if (!range) throw new Error(`no range of ${group.name} holds article ${row.first_article}`)
    const rules = await RuleSets.current()
    const ix = new RangeIndexer({
      group: group.name,
      start: row.first_article,
      end: row.last_article,
      rules: { ...rules, samplePercent: 0 },
      // No file given up before the end: the span is read chunk by chunk (one text of it topped V8's string).
      idleChunks: Number.MAX_SAFE_INTEGER,
      edgeArticles: 0,
      // The edge wrote it partial when its file is short of parts on the server: so must the re-read.
      partialFiles: true,
    })
    for await (const piece of this.overviewPieces(group.name, row))
      ix.addText(piece.text, piece.next)
    ix.finish()
    const batch = ix.take()
    const found = batch.releases.find((r) => identityKey(r.key).equals(row.identity_key))
    if (!found)
      throw new Error(
        `its articles ${row.first_article}–${row.last_article} of ${group.name} no longer show it`
      )
    // The catalog row under the pending key (the identity, the group, the first article the edge saw).
    const chunk: ReleaseChunk = {
      ...found,
      name: row.named,
      source: 'weak-stem',
      readable: true,
      finished: true,
      edge: false,
      firstArticle: row.first_article,
      fileTotal: row.file_total ?? found.fileTotal,
      tool: undefined,
    }
    const candidates = batch.candidates
      .filter((c) => c.releaseKey === found.key)
      .map((c) => ({ ...c, upload: found.upload }))
    // The chunk's month partition, as the batch store makes sure of before its writes.
    await ChunkPartitions.ensure()
    await db.transaction(async (trx) => {
      const blockId = await BatchStore.blockOf(trx, range)
      await BatchStore.writeReleases(
        trx,
        { groupId: row.group_id, blockId },
        row.first_article,
        [chunk],
        candidates
      )
      await trx.rawQuery(
        `UPDATE releases SET inner_name = ?, inspected_at = now() WHERE release_key = ? AND inner_name IS NULL`,
        [row.named.slice(0, 255), row.release_key]
      )
      await trx.rawQuery(
        `UPDATE pending_releases SET status = ?, probed_at = now() WHERE release_key = ?`,
        [PENDING.named, row.release_key]
      )
    })
  }

  /**
   * The overview lines of the release's article span, a chunk of articles at a time with the first article after
   * it: the stored corpus chunks when its ranges are stored and contiguous, else XOVER in windows of a chunk.
   */
  private async *overviewPieces(
    groupName: string,
    row: NamedRow
  ): AsyncGenerator<{ text: string; next: number }> {
    const size = coordinatorConfig.chunkArticles
    const stored = await this.storedChunks(groupName, row)
    if (stored) {
      for (const { raw, next } of stored)
        yield { text: (await inflate(raw)).toString('latin1'), next }
      return
    }
    this.client ??= await XoverClient.connect({
      host: env.get('NNTP_HOST')!,
      port: env.get('NNTP_PORT', 563),
      user: env.get('NNTP_USER'),
      password: env.get('NNTP_PASSWORD'),
      secure: env.get('NNTP_SECURE', true),
      compress: true,
      timeoutMs: coordinatorConfig.nntpTimeoutMs,
    })
    this.usedAt = Date.now()
    try {
      await this.client.group(groupName)
      for (let from = row.first_article; from <= row.last_article; from += size) {
        const to = Math.min(from + size - 1, row.last_article)
        // One zlib/gzip stream, as the server sent it or deflated by the client (the corpus stores it so).
        const raw = await this.client.xoverRaw(from, to)
        yield { text: (await inflate(raw)).toString('latin1'), next: to + 1 }
      }
    } catch (error) {
      this.client.close()
      this.client = null
      throw error
    }
  }

  /** The span's stored corpus chunks (compressed, with the first article after each), or null when any is missing. */
  private async storedChunks(
    groupName: string,
    row: NamedRow
  ): Promise<{ raw: Buffer; next: number }[] | null> {
    if (!this.corpusDir) return null
    const stored = await db
      .from('ranges')
      .where('group_id', row.group_id)
      .whereNotNull('stored_at')
      .where('start_id', '<=', row.last_article)
      .where('end_id', '>=', row.first_article)
      .orderBy('start_id')
    if (
      stored.length === 0 ||
      Number(stored[0].start_id) > row.first_article ||
      Number(stored[stored.length - 1].end_id) < row.last_article
    )
      return null
    const size = coordinatorConfig.chunkArticles
    const chunks: { raw: Buffer; next: number }[] = []
    for (const r of stored) {
      const range = { group: groupName, start: Number(r.start_id), end: Number(r.end_id) }
      for (
        let from =
          range.start +
          Math.floor((Math.max(row.first_article, range.start) - range.start) / size) * size;
        from <= Math.min(row.last_article, range.end);
        from += size
      ) {
        const to = Math.min(from + size - 1, range.end)
        const raw = await Corpus.read(range, { from, to }, this.corpusDir)
        if (!raw) return null
        chunks.push({ raw, next: to + 1 })
      }
    }
    return chunks
  }
}