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 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 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> {
    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 text = await this.overviewText(group.name, row)
    const rules = await RuleSets.current()
    const ix = new RangeIndexer({
      group: group.name,
      start: row.first_article,
      end: row.last_article,
      rules: { ...rules, samplePercent: 0 },
      idleChunks: 0,
      edgeArticles: 0,
    })
    ix.addText(text, row.last_article + 1)
    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 }))
    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: the stored corpus chunks when its range is stored, else XOVER. */
  private async overviewText(groupName: string, row: NamedRow): Promise<string> {
    if (this.corpusDir) {
      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')
      const parts: string[] = []
      let covered = true
      for (const r of stored) {
        const range = { group: groupName, start: Number(r.start_id), end: Number(r.end_id) }
        const size = coordinatorConfig.chunkArticles
        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 raw = await Corpus.read(
            range,
            { from, to: Math.min(from + size - 1, range.end) },
            this.corpusDir
          )
          if (!raw) {
            covered = false
            break
          }
          parts.push((await inflate(raw)).toString('latin1'))
        }
        if (!covered) break
      }
      if (
        covered &&
        parts.length > 0 &&
        stored.length > 0 &&
        Number(stored[0].start_id) <= row.first_article &&
        Number(stored[stored.length - 1].end_id) >= row.last_article
      ) {
        return parts.join('')
      }
    }
    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)
      return (await this.client.xoverRaw(row.first_article, row.last_article)).toString('latin1')
    } catch (error) {
      this.client.close()
      this.client = null
      throw error
    }
  }
}