usenet-indexer/commands/workers/File.ts

import { BaseCommand } from '@adonisjs/core/ace'
import { CommandOptions } from '@adonisjs/core/types/ace'
import { Worker } from 'bullmq'
import db from '@adonisjs/lucid/services/db'
import env from '#start/env'
import FileStore from '#services/FileStore'
import SfvService from '#services/SfvService'
import queueConfig from '#config/queue'
import Shutdown from '#services/Shutdown'
import QueueService from '#services/QueueService'
import ContentService from '#services/ContentService'

export default class FileWorker extends BaseCommand {
  public static commandName = 'worker:file'
  public static description = 'Starts a worker that turns completed binaries into files.'

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

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

    const collectionQueue = QueueService.collectionQueue

    const worker = new Worker('file-queue', async (job) => {
      const { binaryId } = job.data

      const binary = await db.from('binaries').where('id', binaryId).first()
      if (!binary) {
        this.logger.warning(`Binary ${binaryId} no longer exists (garbage collected?).`)
        return
      }

      if (binary.conflicted_at) {
        this.logger.warning(`Binary ${binaryId} ("${binary.name}") has conflicting parts, skipping.`)
        return
      }

      const parts = await db
        .from('parts')
        .where('binary_id', binaryId)
        .orderBy('number')
        .select('number', 'message_id', 'bytes')
      if (parts.length < binary.total_parts) {
        throw new Error(`Binary ${binaryId} has ${parts.length}/${binary.total_parts} parts.`)
      }

      const segments = parts.map((part) => ({ number: Number(part.number), id: part.message_id, size: Number(part.bytes) }))

      // Already known as a file, e.g. added from a posted NZB before its own articles completed.
      const existing = await db.from('files').where('first_message_id', segments[0]?.id ?? '').first()
      if (existing) {
        await db.from('binaries').where('id', binaryId).update({ file_id: existing.id })
        await dropCopiedParts(binary)
        this.logger.debug(`File "${binary.name}" already exists as ID ${existing.id} (same first segment).`)
        return
      }

      // file_key is unique: a second completion of the same upload inserts nothing.
      const now = new Date()
      const [inserted] = await FileStore.insert([
        {
          file_key: binary.file_key,
          filename: binary.name,
          poster: binary.poster,
          date: binary.posted_at ? new Date(binary.posted_at).getTime() : now.getTime(),
          parts: parts.length,
          segments,
          groups: JSON.stringify(binary.groups),
          release_id: binary.release_id,
          real_name: binary.real_name,
          size: parts.reduce((sum, part) => sum + Number(part.bytes), 0),
          created_at: now,
          updated_at: now,
        },
      ])

      const fileId =
        inserted?.id ?? (await db.from('files').where('file_key', binary.file_key).first())?.id
      await db.from('binaries').where('id', binaryId).update({ file_id: fileId })
      await dropCopiedParts(binary)

      if (!inserted) {
        this.logger.debug(`File "${binary.name}" (${binary.file_key}) already exists as ID ${fileId}.`)
        return
      }

      this.logger.info(`Saved file "${binary.name}" (${parts.length} parts) as ID ${fileId}.`)
      // An SFV posted earlier may already list it (D34).
      await SfvService.claim({
        id: Number(fileId),
        name: binary.real_name ?? binary.name,
        poster: binary.poster,
        date: binary.posted_at ? new Date(binary.posted_at).getTime() : null,
        release_id: binary.release_id,
      })
      // Files inspection would skip anyway (later volumes, readable media names) don't take a queue slot.
      if (ContentService.isCandidate(binary.name)) {
        await collectionQueue.add('process-collection', { fileId }, { jobId: `file-${fileId}` })
      }
    }, {
      connection: queueConfig.connection,
      // One job at a time could not keep up with the backfill (130k waiting); jobs are per binary and
      // the file insert ignores duplicates, so they can run side by side.
      concurrency: env.get('FILE_CONCURRENCY', 8),
    })

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

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

/**
 * The file holds every message id now. Binaries identified by their subject
 * (no size: header-derived) don't need their parts any more; body-derived
 * ones keep them, so articles of an equal-sized next volume are recognised
 * as a conflict and split off (CONCEPT.md D20). LOG.md 2026-09-28 (17).
 */
/**
 * The file holds the segment list now. Header-derived parts (no per-part
 * CRC) aren't needed any more; body-derived ones are kept for split-volume
 * verification (D20). Going by the CRC, not the binary's size: content
 * inspection gives header-derived binaries a size too, which kept ~20M
 * parts around (LOG.md 2026-10-02).
 */
async function dropCopiedParts(binary: { id: number }) {
  await db.from('parts').where('binary_id', binary.id).whereNull('pcrc32').delete()
}