implemented a lot of features

commands/workers/Fetch.ts

@@ -2,12 +2,20 @@ 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'
 
 export default class FetchWorker extends BaseCommand {
   public static commandName = 'worker:fetch'
-  public static description = 'Starts a worker to fetch headers from the NNTP server.'
+  public static description = 'Starts a worker that fetches headers (XOVER) and stores them as binaries/parts.'
 
   public static options: CommandOptions = {
     startApp: true,
@@ -17,53 +25,86 @@ export default class FetchWorker extends BaseCommand {
     this.logger.info('Starting fetch worker...')
 
     const pool = NntpService
-    const headerQueue = QueueService.headerQueue
+    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 conn.group(groupName)
-        const overview: any = await conn.xover(startId, endId)
-        this.logger.info(`Fetched ${overview.overviews.length} headers from ${groupName}.`)
-
-        if (overview.overviews.length > 0) {
-          const jobs = overview.overviews.map(([id, header]: [number, any]) => {
-            if (!header) {
-              this.logger.warning(`Header is undefined for job ${job.id}`)
-              return null
-            }
-            return {
-              name: 'process-header',
-              data: { header, group: groupName },
-              opts: { jobId: `${groupName}-${id}` },
-            }
-          }).filter(Boolean)
-
-          await headerQueue.addBulk(jobs)
-          this.logger.info(`Added ${jobs.length} header jobs to the queue for ${groupName}.`)
-        }
+        overviews = (await conn.xover(startId, endId)).overviews
       } catch (error: any) {
+        failed = true
         this.logger.error(`Error fetching headers for ${groupName}: ${error.message}`)
         throw error
       } finally {
         if (conn) {
-          pool.release(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()
+      const fetchBodies = Boolean(group?.fetch_bodies)
+
+      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 })
+      }
+
+      const completed = await IngestService.ingest(segments)
+      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, ${segments.length} stored, ` +
+          `${obfuscated.length} sent to body-queue${skipped ? ` (${skipped} skipped)` : ''}` +
+          `${deferred.length ? `, ${deferred.length} deferred (${deferredFiles} files)` : ''}, ${completed.length} files completed.`
+      )
+    }, {
       connection: queueConfig.connection,
-      concurrency: 5,
+      // 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', (job, err) => {
+    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 new Promise(() => {}) // Keep command running
+    await Shutdown.wait()
+    this.logger.info('Shutting down: finishing active jobs...')
+    await worker.close()
+    await NntpService.shutdown()
+    await QueueService.closeAll()
   }
 }