usenet-indexer/commands/workers/Collection.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 Shutdown from '#services/Shutdown'
import QueueService from '#services/QueueService'
import NntpService from '#services/NntpService'
import ContentService from '#services/ContentService'
import ArchiveWalkService from '#services/ArchiveWalkService'
import File from '#models/file'

export default class CollectionWorker extends BaseCommand {
  public static commandName = 'worker:collection'
  public static description = 'Starts a worker to process file collections (e.g., RAR archives).'

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

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

    const worker = new Worker('collection-queue', async (job) => {
      const { fileId } = job.data
      this.logger.debug(`Processing file ID ${fileId} for collection.`)

      const file = await File.find(fileId)
      if (!file) {
        this.logger.error(`File with ID ${fileId} not found in the database.`)
        return
      }

      try {
        const content = await ContentService.inspect(fileId)
        if (content?.par2) {
          const { par2 } = content
          this.logger.info(`Par2 "${file.filename}": release "${par2.releaseName}", ${par2.listed} files listed.`)
        } else if (content?.archive) {
          const { info, releaseRenamed, grouped } = content.archive
          const flags = [info.headersEncrypted && 'encrypted headers', info.entries.some((e) => e.encrypted) && 'encrypted files']
            .filter(Boolean)
            .join(', ')
          this.logger.info(
            `Archive "${file.filename}": ${info.format}, ${info.entries.length} entries${flags ? ` (${flags})` : ''}` +
              (grouped ? ', grouped into a release' : '') +
              (releaseRenamed ? `, release renamed to "${releaseRenamed}"` : '')
          )
        }
        if (content?.named) this.logger.info(`"${file.filename}" got its real name from par2.`)
        if (content?.contentType === 'nzbe') this.logger.info(`"${file.filename}" is an encrypted NZB container.`)
        if (content?.contentType === 'rclone') this.logger.info(`"${file.filename}" is rclone-encrypted.`)
        if (content?.yencName) this.logger.info(`"${file.filename}" is "${content.yencName}" (yEnc name).`)
        if (content?.nicovideo) this.logger.info(`"${file.filename}" → "${content.nicovideo}".`)
        if (content?.sfvMoved) this.logger.info(`SFV "${file.filename}" moved ${content.sfvMoved} files into its release.`)
        if (content?.contentType === 'yencbin') this.logger.info(`"${file.filename}" is a yEncBin encrypted container.`)
        if (content?.nzb) {
          const n = content.nzb
          this.logger.info(`NZB "${file.filename}": ${n.listed} files (${n.known} known, ${n.added} added) → "${n.releaseName}".`)
        }
      } catch (error: any) {
        this.logger.warning(`Content inspection of "${file.filename}" failed: ${error.message}`)
      }
    }, {
      connection: queueConfig.connection,
      // Each job holds one NNTP connection while downloading; keep <= the pool size.
      concurrency: env.get('COLLECTION_CONCURRENCY', 1),
    })

    worker.on('failed', (job, err) => {
      this.logger.error(`Collection job ${job?.id} failed: ${err.message}`)
    })

    // Complete archive listings (D40): background work beside inspection, on the same NNTP pool.
    const walkConcurrency = env.get('ARCHIVE_WALK_CONCURRENCY', 1)
    const walker = walkConcurrency > 0
      ? new Worker('archive-walk-queue', async (job) => {
          const o = await ArchiveWalkService.walk(Number(job.data.releaseId))
          if (o.skipped) return o
          this.logger.info(
            `Archive walk of release ${o.releaseId}: ${o.before} → ${o.entries} entries from ${o.segments} segments` +
              (o.complete ? '' : ` (stopped: ${o.stopped})`) +
              (o.renamed ? `, renamed to "${o.renamed}"` : '')
          )
          return o
        }, { connection: queueConfig.connection, concurrency: walkConcurrency })
      : null
    walker?.on('failed', (job, err) => {
      this.logger.error(`Archive walk job ${job?.id} failed: ${err.message}`)
    })

    this.logger.info('Collection worker started and listening for jobs.')
    await Shutdown.wait()
    this.logger.info('Shutting down: finishing active jobs...')
    await Promise.all([worker.close(), walker?.close()])
    await NntpService.shutdown()
    await QueueService.closeAll()
  }
}