usenet-indexer/commands/ReleaseRebuild.ts

import { BaseCommand, flags } from '@adonisjs/core/ace'
import { CommandOptions } from '@adonisjs/core/types/ace'
import db from '@adonisjs/lucid/services/db'
import IngestService from '#services/IngestService'
import NameService from '#services/NameService'
import QueueService from '#services/QueueService'
import { rarReleaseIdentity, releaseIdentity } from '#lib/release/index'

const BATCH = 5000

/**
 * Recomputes the release of every binary (and its file) with the current
 * rules in lib/release. Used after changing those rules.
 */
export default class ReleaseRebuild extends BaseCommand {
  public static commandName = 'release:rebuild'
  public static description = 'Reassigns all binaries and files to releases using the current grouping rules.'

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

  @flags.boolean({
    description:
      'Also drop everything derived from file contents (par2 lists, par2/RAR groupings, real names), ' +
      'to rebuild it with content:backfill --all',
  })
  declare resetContent: boolean

  public async run() {
    if (this.resetContent) await this.dropContentDerived()

    let lastId = 0
    let assigned = 0
    let unassigned = 0
    try {
      while (true) {
        const binaries = await db
          .from('binaries')
          .where('id', '>', lastId)
          .orderBy('id')
          .limit(BATCH)
          .select('id', 'poster', 'subject', 'name', 'groups', 'posted_at')
        if (binaries.length === 0) break
        lastId = Number(binaries.at(-1).id)

        const segments = binaries.map((binary) => {
          // Body-derived binaries store a synthetic subject, see segmentFromYenc.
          const weak = /^"[^"]*" yEnc$/.test(binary.subject)
          return {
            id: Number(binary.id),
            poster: binary.poster,
            groups: binary.groups,
            postedAt: binary.posted_at ? new Date(binary.posted_at) : undefined,
            weakIdentity: weak,
            release: releaseIdentity({ poster: binary.poster, subject: binary.subject, name: binary.name, weak }),
          }
        })

        await db.transaction(async (trx) => {
          const releaseIds = await IngestService.upsertReleases(trx.knexClient, segments)
          const ids = segments.map((s) => s.id)
          const releaseIdsPerBinary = segments.map((s) =>
            s.release ? releaseIds.get(s.release.key)! : null
          )
          await trx.rawQuery(
            `UPDATE binaries b SET release_id = v.release_id
             FROM unnest(?::bigint[], ?::bigint[]) AS v(id, release_id)
             WHERE b.id = v.id AND b.release_id IS DISTINCT FROM v.release_id
               AND NOT (v.release_id IS NULL
                        AND b.release_id IN (SELECT id FROM releases WHERE source IN ('par2', 'rar')))
               -- A posted NZB decided this grouping (NzbImportService); name rules don't override it.
               AND NOT EXISTS (SELECT 1 FROM files n WHERE n.release_id = b.release_id AND n.content_type = 'nzb')`,
            [ids, releaseIdsPerBinary] as any
          )
        })
        assigned += segments.filter((s) => s.release).length
        unassigned += segments.filter((s) => !s.release).length
      }

      // Files grouped by their RAR headers: regroup from the stored listing (no downloads).
      let rarGrouped = 0
      let rarCursor = 0
      while (true) {
        const rows = await db
          .from('files as f')
          .join('binaries as b', 'b.file_id', 'f.id')
          .whereNotNull('f.archive')
          .whereNull('b.release_id')
          .where('f.id', '>', rarCursor)
          .orderBy('f.id')
          .limit(BATCH)
          .select('f.id as file_id', 'f.archive', 'b.id as binary_id', 'b.poster', 'b.groups', 'b.posted_at')
        if (rows.length === 0) break
        rarCursor = Number(rows.at(-1).file_id)
        const segments = rows.map((r) => ({
          fileId: Number(r.file_id),
          binaryId: Number(r.binary_id),
          poster: r.poster,
          groups: r.groups,
          postedAt: r.posted_at ? new Date(r.posted_at) : undefined,
          weakIdentity: true,
          release: rarReleaseIdentity(r.archive?.entries ?? []),
        }))
        const ids = await IngestService.upsertReleases(db.connection().getWriteClient(), segments)
        for (const s of segments.filter((x) => x.release)) {
          const releaseId = ids.get(s.release!.key)
          await db.from('binaries').where('id', s.binaryId).update({ release_id: releaseId })
          await db.from('files').where('id', s.fileId).update({ release_id: releaseId })
          rarGrouped++
        }
      }

      // Real names from par2 for binaries with a known size (no downloads).
      let named = 0
      let nameCursor = 0
      while (true) {
        const rows = await db.from('binaries').whereNotNull('size').where('id', '>', nameCursor).orderBy('id').limit(BATCH).select('id')
        if (rows.length === 0) break
        nameCursor = Number(rows.at(-1).id)
        named += await NameService.resolve(rows.map((r) => Number(r.id)))
      }

      const files = await db.rawQuery(
        `UPDATE files f SET release_id = b.release_id FROM binaries b
         WHERE b.file_id = f.id AND f.release_id IS DISTINCT FROM b.release_id`
      )
      const orphans = await db.rawQuery(
        `DELETE FROM releases r WHERE NOT EXISTS (SELECT 1 FROM binaries b WHERE b.release_id = r.id)
           AND NOT EXISTS (SELECT 1 FROM files f WHERE f.release_id = r.id)`
      )
      if (this.resetContent) {
        this.logger.info('Content-derived data dropped; run `node ace content:backfill --all` to rebuild it.')
      }
      this.logger.info(
        `Binaries: ${assigned} in a release, ${unassigned} without, ${rarGrouped} regrouped from RAR headers, ` +
          `${named} named from par2. ` +
          `Files updated: ${files.rowCount}. Empty releases removed: ${orphans.rowCount}.`
      )
    } finally {
      await QueueService.closeAll()
    }
  }

  /** par2 lists, par2/RAR release assignments and recovered names. */
  private async dropContentDerived() {
    await db.transaction(async (trx) => {
      await trx.rawQuery('DELETE FROM release_files')
      await trx.rawQuery(`UPDATE binaries SET real_name = NULL WHERE real_name IS NOT NULL`)
      await trx.rawQuery(`UPDATE files SET real_name = NULL, archive = NULL WHERE real_name IS NOT NULL OR archive IS NOT NULL`)
      await trx.rawQuery(`UPDATE releases SET contents = NULL, encrypted = NULL WHERE contents IS NOT NULL OR encrypted IS NOT NULL`)
      // Binaries whose name rules find no release were grouped from contents (or wrongly by size).
      await trx.rawQuery(`
        UPDATE binaries SET release_id = NULL
        WHERE subject ~ '^"[^"]*" yEnc$' AND release_id IS NOT NULL
          AND (name !~ '\\.[A-Za-z0-9]{1,5}$' OR release_id IN (SELECT id FROM releases WHERE source IN ('par2', 'rar')))`)
      await trx.rawQuery(`
        UPDATE files f SET release_id = b.release_id FROM binaries b
        WHERE b.file_id = f.id AND f.release_id IS DISTINCT FROM b.release_id`)
    })
    this.logger.info('Dropped par2 lists, content-derived groupings and recovered names.')
  }
}