usenet-indexer/commands/workers/Fetch.ts

import { BaseCommand } from '@adonisjs/core/ace'
import { CommandOptions } from '@adonisjs/core/types/ace'
import { Worker } from 'bullmq'
import queueConfig from '#config/queue'
import env from '#start/env'
import db from '@adonisjs/lucid/services/db'
import indexerConfig from '#config/indexer'
import Shutdown from '#services/Shutdown'
import QueueService from '#services/QueueService'
import NntpService from '#services/NntpService'
import FetchRangeService from '#services/FetchRangeService'
import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService'
import DeferredService, { bodyJob, isDeferred, type DeferredArticle } from '#services/DeferredService'
import SpotService, { isSpotGroup } from '#services/SpotService'
import StagingService from '#services/StagingService'

export default class FetchWorker extends BaseCommand {
  public static commandName = 'worker:fetch'
  public static description = 'Starts a worker that fetches headers (XOVER) and stores them as binaries/parts.'

  public static options: CommandOptions = {
    startApp: true,
  }

  public async run() {
    this.logger.info('Starting fetch worker...')

    const pool = NntpService
    const bodyQueue = QueueService.bodyQueue

    const worker = new Worker('nntp-fetch-queue', async (job) => {
      const { groupName, startId, endId } = job.data
      this.logger.info(`Processing fetch job for ${groupName}, articles ${startId}-${endId}`)

      let overviews: [number, Record<string, string>][]
      let conn
      let failed = false
      try {
        conn = await pool.acquire()
        await pool.timed(conn.group(groupName), `GROUP ${groupName}`)
        const reply = await pool.timed<any>(conn.xover(startId, endId), `XOVER ${groupName} ${startId}-${endId}`)
        // An empty reply parses as one line without fields ([null, {}]).
        overviews = reply.overviews.filter(([number, header]: [number | null, Record<string, string>]) => number !== null && header?.['message-id'])
      } catch (error: any) {
        failed = true
        this.logger.error(`Error fetching headers for ${groupName}: ${error.message}`)
        throw error
      } finally {
        if (conn) {
          pool.release(conn, failed)
        }
      }

      if (isSpotGroup(groupName)) {
        const spots = await SpotService.recordHeaders(groupName, overviews)
        if (job.data.rangeId) await FetchRangeService.markDone(job.data.rangeId)
        this.logger.info(`${groupName} ${startId}-${endId}: ${overviews.length} headers, ${spots} spots queued.`)
        return
      }

      const group = await db.from('groups').where('name', groupName).first()
      // Backfill is header-only: old bodies would be far more than the link carries (CONCEPT.md D32).
      const fetchBodies = Boolean(group?.fetch_bodies) && !job.data.backfill

      const segments: IngestSegment[] = []
      const obfuscated: { articleNumber: number; header: Record<string, string> }[] = []
      const deferred: DeferredArticle[] = []
      let skipped = 0
      for (const [articleNumber, header] of overviews) {
        if (!header) continue
        const segment = segmentFromHeader(header, groupName, articleNumber)
        if (segment) segments.push(segment)
        // Recorded per file (or as article ranges) in every group, bodies or not; processed later if at all.
        else if (isDeferred(header)) deferred.push({ articleNumber, header })
        else if (!fetchBodies || indexerConfig.skipBodyMessageIds.some((re) => re.test(header['message-id'] ?? ''))) skipped++
        else obfuscated.push({ articleNumber, header })
      }

      // Lone segments of new files wait in staging until their file proves real (D35).
      const { ingest, staged } = StagingService.enabled(groupName)
        ? await StagingService.filter(segments, { counters: StagingService.countersEnabled(groupName) })
        : { ingest: segments, staged: 0 }
      const completed = await IngestService.ingest(ingest)
      const deferredFiles = await DeferredService.record(groupName, deferred)

      if (obfuscated.length > 0) {
        await bodyQueue.addBulk(obfuscated.map(({ articleNumber, header }) => bodyJob(header, groupName, articleNumber)))
      }

      if (job.data.rangeId) await FetchRangeService.markDone(job.data.rangeId)

      this.logger.info(
        `${groupName} ${startId}-${endId}: ${overviews.length} headers, ${ingest.length} stored${staged ? `, ${staged} staged` : ''}, ` +
          `${obfuscated.length} sent to body-queue${skipped ? ` (${skipped} skipped)` : ''}` +
          `${deferred.length ? `, ${deferred.length} deferred (${deferredFiles} files)` : ''}, ${completed.length} files completed.`
      )
    }, {
      connection: queueConfig.connection,
      // XOVER is quick; most of a job is storing. Each concurrent job uses one DB connection (DB_POOL_MAX).
      concurrency: env.get('FETCH_CONCURRENCY', 5),
    })

    worker.on('failed', async (job, err) => {
      this.logger.error(`Fetch job ${job?.id} failed for group ${job?.data.groupName}: ${err.message}`)
      // Only after the last attempt; the scheduler re-queues the range later.
      if (job?.data.rangeId && job.attemptsMade >= (job.opts.attempts ?? 1)) {
        await FetchRangeService.markFailed(job.data.rangeId, err.message).catch(() => {})
      }
    })

    this.logger.info('Fetch worker started and listening for jobs.')
    await Shutdown.wait()
    this.logger.info('Shutting down: finishing active jobs...')
    await worker.close()
    await NntpService.shutdown()
    await QueueService.closeAll()
  }
}