usenet-indexer/app/services/ArchiveWalkService.ts

import db from '@adonisjs/lucid/services/db'
import logger from '@adonisjs/core/services/logger'
import env from '#start/env'
import NntpService from '#services/NntpService'
import QueueService from '#services/QueueService'
import { segmentsOf } from '#services/FileStore'
import { looksRandom, releaseNameFromContents } from '#services/ArchiveService'
import { decodeSegment, YencCrcError, YencError, type DecodedSegment } from '#lib/yenc/index'
import { rarVolume } from '#lib/content/index'
import { walkRar, type RarInfo, type VolumeReader } from '#lib/rar/index'

/** Stored per release, as for a first-volume listing (ArchiveService). */
const MAX_ENTRIES = 500

/** Listed sizes below this share of the release's size: the first segment listed only part of it. */
const COVERED = 0.7

/** Inspection queues a walk this long after the first volume, when the release has had time to complete. */
const WALK_DELAY_MS = 30 * 60 * 1000

export interface WalkOutcome {
  releaseId: number
  /** Nothing downloaded: why. */
  skipped?: 'missing' | 'covered' | 'volumes'
  entries?: number
  before?: number
  complete?: boolean
  stopped?: string
  segments?: number
  renamed?: string
}

/**
 * Complete archive listings by walking RAR headers (CONCEPT.md D40). The
 * first segment of a first volume lists the entries that start in it: for
 * a set of many small files (an audiobook's MP3s, a music album) that is
 * the first one or two. The walk reads one segment per further entry and
 * volume, at low priority: its own queue, one job at a time per collection
 * worker, and NNTP connections only when inspection isn't waiting for one.
 */
class ArchiveWalkService {
  /** Segments one walk may download (~0.7 MB each). */
  readonly maxSegments = env.get('ARCHIVE_WALK_MAX_SEGMENTS', 400)

  /** A listing that leaves most of the release unaccounted for. */
  worthWalking(entries: number, listedBytes: number, releaseSize: number): boolean {
    if (entries === 0 || entries >= MAX_ENTRIES) return false
    return listedBytes < COVERED * releaseSize
  }

  /** After inspection listed a first RAR volume whose listing ends inside the set. */
  async queueAfterInspection(releaseId: number | null, info: RarInfo) {
    if (!releaseId || info.format === '7z' || info.firstVolume === false) return
    if (!info.truncated || info.headersEncrypted || info.entries.length === 0) return
    await QueueService.archiveWalkQueue.add('walk', { releaseId }, { jobId: `walk-${releaseId}`, delay: WALK_DELAY_MS })
  }

  async walk(releaseId: number): Promise<WalkOutcome> {
    const release = await db.from('releases').where('id', releaseId).select('id', 'name', 'source', 'size', 'contents', 'encrypted').first()
    if (!release) return { releaseId, skipped: 'missing' }
    const before = Array.isArray(release.contents) ? release.contents : []
    const listed = before.reduce((sum: number, e: any) => sum + Number(e.size ?? 0), 0)
    if (!this.worthWalking(before.length, listed, Number(release.size))) return this.done(releaseId, { releaseId, skipped: 'covered' })

    const volumes = await this.volumes(releaseId)
    if (!volumes) return this.done(releaseId, { releaseId, skipped: 'volumes' })

    const reader = new SegmentReader(volumes, this.maxSegments)
    const walked = await walkRar(reader)
    const outcome: WalkOutcome = {
      releaseId,
      before: before.length,
      entries: walked.entries.length,
      complete: walked.complete,
      stopped: walked.stopped,
      segments: reader.fetched,
    }
    if (walked.entries.length > before.length) {
      const contents = walked.entries.slice(0, MAX_ENTRIES).map(({ stored: _s, dataOffset: _d, ...entry }) => entry)
      const update: Record<string, unknown> = {
        contents: JSON.stringify(contents),
        encrypted: Boolean(release.encrypted) || walked.headersEncrypted || walked.entries.some((e) => e.encrypted),
      }
      const better = releaseNameFromContents({ entries: walked.entries })
      // The rule of ArchiveService.record: random and post-subject names give way, a spot's title stays.
      if (better && better !== release.name && release.source !== 'spot' && (looksRandom(release.name) || release.source === 'post')) {
        update.name = better
        outcome.renamed = better
      }
      await db.from('releases').where('id', releaseId).update(update)
    }
    return this.done(releaseId, outcome)
  }

  private async done(releaseId: number, outcome: WalkOutcome) {
    await db.from('releases').where('id', releaseId).update({ contents_walked_at: new Date() })
    return outcome
  }

  /**
   * The release's RAR volumes in order, from volume 1 up to the first gap.
   * Null without a first volume (e.g. volumes with random names).
   */
  private async volumes(releaseId: number): Promise<any[] | null> {
    const files = await db.from('files').where('release_id', releaseId).select('id', 'filename', 'real_name', 'segments')
    const sets = new Map<string, Map<number, any>>()
    for (const file of files) {
      const volume = rarVolume(String(file.real_name ?? file.filename))
      if (!volume || segmentsOf(file).length === 0) continue
      const set = sets.get(volume.base) ?? new Map()
      if (!set.has(volume.number)) set.set(volume.number, file)
      sets.set(volume.base, set)
    }
    // A release holds one set, now and then a second (a sample, subs): the biggest.
    const set = [...sets.values()].sort((a, b) => b.size - a.size)[0]
    if (!set?.has(1)) return null
    const ordered: any[] = []
    for (let n = 1; set.has(n); n++) ordered.push(set.get(n))
    return ordered
  }
}

/**
 * Reads volumes a segment at a time, finding the segment for an offset from
 * the part size (yEnc posters use one size per set) and correcting from the
 * `=ypart` offsets when a guess is off.
 */
export class SegmentReader implements VolumeReader {
  fetched = 0
  private partSize = 0
  private readonly sizes = new Map<number, number>()
  private readonly cache = new Map<string, DecodedSegment>()

  constructor(
    private readonly files: any[],
    private readonly maxSegments: number,
    private readonly fetchBody: (messageId: string) => Promise<Buffer> = fetchLow
  ) {}

  get volumes() {
    return this.files.length
  }

  size(v: number) {
    return this.sizes.get(v) ?? null
  }

  async read(v: number, offset: number, more: boolean): Promise<Buffer | null> {
    const count = segmentsOf(this.files[v]).length
    let index = this.partSize ? Math.min(count - 1, Math.floor(offset / this.partSize)) : 0
    for (let tries = 0; tries < 4; tries++) {
      const segment = await this.segment(v, index)
      if (!segment) return null
      const end = segment.offset + segment.data.length
      if (offset >= segment.offset && offset < end) {
        const data = segment.data.subarray(offset - segment.offset)
        if (!more || index + 1 >= count) return data
        const next = await this.segment(v, index + 1)
        return next ? Buffer.concat([data, next.data]) : null
      }
      const step = Math.floor((offset - segment.offset) / Math.max(1, segment.data.length))
      const moved = Math.max(0, Math.min(count - 1, index + (step === 0 ? (offset < segment.offset ? -1 : 1) : step)))
      if (moved === index) return null
      index = moved
    }
    return null
  }

  private async segment(v: number, index: number): Promise<DecodedSegment | null> {
    const key = `${v}:${index}`
    const cached = this.cache.get(key)
    if (cached) return cached
    const part = segmentsOf(this.files[v])[index]
    if (!part?.id || this.fetched >= this.maxSegments) return null
    this.fetched++
    let body: Buffer
    try {
      body = await this.fetchBody(part.id)
    } catch (error: any) {
      logger.debug(`Archive walk: segment ${index} of file ${this.files[v].id} unavailable: ${error.message}`)
      return null
    }
    let segment: DecodedSegment
    try {
      segment = decodeSegment(body)
    } catch (error) {
      if (!(error instanceof YencError) || error instanceof YencCrcError) return null
      segment = decodeSegment(body, { lenientSize: true })
    }
    this.sizes.set(v, segment.meta.begin.size)
    if (!this.partSize && segment.meta.part) this.partSize = segment.meta.part.end - segment.meta.part.begin + 1
    // Walks move forward: the last few segments are enough.
    if (this.cache.size >= 4) this.cache.delete(this.cache.keys().next().value!)
    this.cache.set(key, segment)
    return segment
  }
}

/** One article body on a pooled connection, behind inspection's waiting jobs. */
async function fetchLow(messageId: string): Promise<Buffer> {
  const conn = await NntpService.acquire(true)
  let failed = false
  try {
    return await NntpService.timed(NntpService.fetchBody(conn, messageId), 'BODY')
  } catch (error) {
    failed = true
    throw error
  } finally {
    NntpService.release(conn, failed)
  }
}

export default new ArchiveWalkService()