usenet-indexer/commands/ContentBackfill.ts

import { BaseCommand, flags } from '@adonisjs/core/ace'
import { CommandOptions } from '@adonisjs/core/types/ace'
import db from '@adonisjs/lucid/services/db'
import QueueService from '#services/QueueService'
import ContentService from '#services/ContentService'
import { isArchiveStart } from '#lib/content/index'

const BATCH = 1000

/**
 * Queues completed files for content inspection (names and groupings from
 * par2 and RAR headers, CONCEPT.md D21) that were completed before it
 * existed. Costs one article download per file, done by worker:collection.
 */
export default class ContentBackfill extends BaseCommand {
  public static commandName = 'content:backfill'
  public static description = 'Queues existing files with random names for content inspection.'

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

  @flags.number({ description: 'Queue at most this many files' })
  declare limit?: number

  @flags.boolean({ description: 'Also files that were inspected before (after release:rebuild --reset-content)' })
  declare all: boolean

  @flags.boolean({ description: 'Only archives whose listing contains NZBs, to import those (CONCEPT.md D28)' })
  declare archiveNzbs: boolean

  @flags.boolean({ description: 'Only unlisted 7z archives (first volumes: .7z, .7z.001)' })
  declare sevenZip: boolean

  @flags.boolean({ description: 'Archive starts never inspected (RAR first volumes, 7z, .bin), also files from NZBs/spots' })
  declare archives: boolean

  @flags.boolean({ description: 'SFV files not read yet (their listings group the files they name, D34)' })
  declare sfv: boolean

  @flags.boolean({ description: 'niconico captures not read yet (their releases get the video title, D34)' })
  declare nicovideo: boolean

  public async run() {
    let lastId = 0
    let queued = 0
    try {
      while (!this.limit || queued < this.limit) {
        // Files never inspected: no fingerprint on their binary and no archive listing.
        const query = db.from('files as f').where('f.id', '>', lastId)
        // Files from NZBs (spots) have no binary.
        if (this.sevenZip || this.archives || this.sfv || this.nicovideo) query.leftJoin('binaries as b', 'b.file_id', 'f.id')
        else query.join('binaries as b', 'b.file_id', 'f.id')
        if (this.nicovideo) {
          query.whereNull('f.content_type').whereRaw(`f.filename ~* '^\\d{9,11}_nicovideo_jp_watch_sm\\d+\\.tar\\.zst$'`)
        } else if (this.sfv) {
          query.whereNull('f.content_type').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.sfv$'`)
        } else if (this.archives) {
          query.whereNull('f.content_type').whereNull('f.archive').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.(rar|7z|7z\\.0*1|bin)$'`)
        } else if (this.sevenZip) {
          query.whereNull('f.archive').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.7z(\\.0*1){0,1}$'`)
        } else if (this.archiveNzbs) query.whereRaw(`f.archive::text ~* '\\.nzb"'`)
        else if (!this.all) query.whereNull('b.md5_16k').whereNull('f.archive')
        const files = await query
          .orderBy('f.id')
          .limit(BATCH)
          .select('f.id', 'f.filename')
        if (files.length === 0) break
        lastId = Number(files.at(-1).id)

        const jobs = files
          .filter((f) => (this.archives ? isArchiveStart(f.filename) : this.sfv || this.nicovideo || ContentService.isCandidate(f.filename)))
          .slice(0, this.limit ? this.limit - queued : undefined)
          .map((f) => ({
            name: 'process-collection',
            data: { fileId: Number(f.id) },
            // A new id per run: BullMQ ignores an add() for an id it still knows.
            // Targeted backfills go ahead of whatever is waiting.
            opts: {
              jobId: `content-${f.id}-${Date.now()}`,
              lifo: this.archiveNzbs || this.sevenZip || this.archives || this.sfv,
              // niconico captures are many (~190k) and only cosmetic: behind everything else.
              ...(this.nicovideo ? { priority: 10 } : {}),
            },
          }))
        if (jobs.length > 0) await QueueService.collectionQueue.addBulk(jobs)
        queued += jobs.length
      }
      this.logger.info(`Queued ${queued} files for content inspection.`)
    } finally {
      await QueueService.closeAll()
    }
  }
}