usenet-indexer/commands/pending_classes.ts

import { BaseCommand, flags } from '@adonisjs/core/ace'
import type { CommandOptions } from '@adonisjs/core/types/ace'
import db from '@adonisjs/lucid/services/db'
import coordinatorConfig from '#config/coordinator'
import { describeClass } from '#lib/release/index'

/**
 * The pending ladder's classes (LOG.md (130)) on the rows from before them: `klass` (the SQL pending_class) and
 * `target_bytes` (the first target's bytes, when its candidates carry them). The finished rows (named, exhausted,
 * naming) in one scan, seeding each class's record, which gets its verdict (dropped) as the ladder gives it; the
 * waiting rows in windows of created_at by their index (a scan per window starved the prober on the HDD); then
 * pending_classes' `waiting` counted afresh from the claim's index. --report shows the record.
 */
export default class PendingClasses extends BaseCommand {
  static commandName = 'pending:classes'
  static description =
    'Classes the pending releases from before the classes and recounts pending_classes; --report shows the record'
  static options: CommandOptions = { startApp: true }

  @flags.number({ description: 'Hours of created_at per window' })
  declare hours: number

  @flags.boolean({ description: 'Show the classes with the best and worst records, write nothing' })
  declare report: boolean

  async run() {
    if (this.report) return this.showReport()
    const hours = this.hours ?? 6
    const started = Date.now()
    // The finished rows (named, exhausted, naming; 6.5M on 2026-11-02) in one pass: a scan of the table, no index
    // reaches them by status; they seed each class's record.
    const { rows: seeded } = await db.rawQuery(
      `WITH classed AS (
         UPDATE pending_releases SET klass = pending_class(name, bytes, partial),
           target_bytes = (SELECT min((c->>'bytes')::bigint) FROM jsonb_array_elements(candidates) c WHERE c->>'bytes' IS NOT NULL)
         WHERE klass = 0 AND status IN (1, 3, 4)
         RETURNING tool, group_id, klass, status),
       upserted AS (
         INSERT INTO pending_classes (tool, group_id, klass, probed, named)
         SELECT tool, group_id, klass, count(*), count(*) FILTER (WHERE status IN (1, 4)) FROM classed GROUP BY 1, 2, 3
         ON CONFLICT (tool, group_id, klass) DO UPDATE SET probed = pending_classes.probed + excluded.probed, named = pending_classes.named + excluded.named
         RETURNING 1)
       SELECT (SELECT count(*) FROM classed) AS classed, (SELECT count(*) FROM upserted) AS classes`
    )
    this.logger.info(
      `Finished rows: ${seeded[0].classed} classed into ${seeded[0].classes} classes (${Math.round((Date.now() - started) / 1000)} s).`
    )
    const { pendingMinProbes, pendingMaxNamed } = coordinatorConfig.probe
    await db.rawQuery(
      `UPDATE pending_classes SET dropped = true WHERE NOT dropped AND probed >= ?::int AND named::float <= probed * ?::float`,
      [pendingMinProbes, pendingMaxNamed]
    )
    // The waiting rows by windows of created_at (pending_releases_waiting_idx reaches them).
    const { rows: span } = await db.rawQuery(
      `SELECT (SELECT min(created_at) FROM pending_releases WHERE status = 0) AS lo, (SELECT max(created_at) FROM pending_releases WHERE status = 0) AS hi`
    )
    if (span[0].lo) {
      let from = new Date(span[0].lo)
      const hi = new Date(span[0].hi)
      let classed = 0
      while (from <= hi) {
        const to = new Date(from.getTime() + hours * 3600_000)
        const result = await db.rawQuery(
          `UPDATE pending_releases SET klass = pending_class(name, bytes, partial),
             target_bytes = (SELECT min((c->>'bytes')::bigint) FROM jsonb_array_elements(candidates) c WHERE c->>'bytes' IS NOT NULL)
           WHERE status = 0 AND created_at >= ? AND created_at < ? AND klass = 0`,
          [from.toISOString(), to.toISOString()]
        )
        classed += result.rowCount ?? 0
        from = to
        this.logger.info(
          `Waiting rows up to ${to.toISOString().slice(0, 16)}: ${classed} classed (${Math.round((Date.now() - started) / 1000)} s).`
        )
      }
    }
    // The waiting counts from the rows themselves (the claim, the discard and the batch store keep them from here).
    await db.rawQuery(`
      INSERT INTO pending_classes (tool, group_id, klass, waiting)
      SELECT tool, group_id, klass, count(*) FROM pending_releases WHERE status = 0 AND finished GROUP BY 1, 2, 3
      ON CONFLICT (tool, group_id, klass) DO UPDATE SET waiting = excluded.waiting`)
    await db.rawQuery(
      `UPDATE pending_classes c SET waiting = 0 WHERE waiting > 0 AND NOT EXISTS (
         SELECT 1 FROM pending_releases p WHERE p.status = 0 AND p.finished AND p.tool = c.tool AND p.group_id = c.group_id AND p.klass = c.klass)`
    )
    const { rows: totals } = await db.rawQuery(
      `SELECT count(*) AS classes, coalesce(sum(waiting), 0) AS waiting, count(*) FILTER (WHERE dropped) AS dropped FROM pending_classes`
    )
    this.logger.info(
      `Done: ${totals[0].classes} classes, ${totals[0].waiting} waiting, ${totals[0].dropped} dropped.`
    )
  }

  private async showReport() {
    const { rows } = await db.rawQuery(
      `SELECT c.tool, g.name AS grp, c.klass, c.probed, c.named, c.waiting, c.dropped FROM pending_classes c JOIN groups g ON g.id = c.group_id
       WHERE c.probed >= 20 ORDER BY c.named::float / c.probed DESC, c.probed DESC`
    )
    const line = (r: any) =>
      `  ${r.tool.padEnd(10)} ${r.grp.padEnd(32)} ${describeClass(Number(r.klass)).padEnd(26)} probed ${String(r.probed).padStart(7)} named ${String(Math.round((100 * r.named) / r.probed)).padStart(3)}% waiting ${String(r.waiting).padStart(8)}${r.dropped ? ' dropped' : ''}`
    this.logger.info(`${rows.length} classes with 20+ climbs; the best:`)
    for (const r of rows.slice(0, 25)) this.logger.info(line(r))
    this.logger.info('The worst:')
    for (const r of rows.slice(-25)) this.logger.info(line(r))
  }
}