usenet-indexer/app/services/RemoteContentService.ts

import { randomUUID } from 'node:crypto'
import { Job, Worker } from 'bullmq'
import logger from '@adonisjs/core/services/logger'
import queueConfig from '#config/queue'
import nntpConfig from '#config/nntp'
import QueueService from '#services/QueueService'
import NntpService from '#services/NntpService'
import ContentService, { describeContent } from '#services/ContentService'
import ArchiveWalkService from '#services/ArchiveWalkService'
import type { ContentTask, InspectProbe, InspectTask, WalkProbe } from '#lib/content/probe'

/** Inspection first; archive walks (D40) only when it has nothing waiting. */
const QUEUES = ['collection-queue', 'archive-walk-queue'] as const
type QueueName = (typeof QUEUES)[number]

/** A node has this long to post a job's result; after that the job goes back to the queue (BullMQ's stalled check). */
const LOCK_MS = 10 * 60 * 1000

export interface RemoteJob {
  queue: QueueName
  jobId: string
  /** The job's lock: whoever holds it may complete the job. */
  token: string
  task: ContentTask
}

export class JobLost extends Error {}
export class ArticleRelayOff extends Error {}

/**
 * Content inspection and archive walks on seeding nodes (CONCEPT.md D41).
 * Nodes take jobs from the same queues worker:collection works on: a lease
 * locks a job (BullMQ's manual processing) and gives the node a task that
 * was prepared from the database; the node probes it, fetching articles
 * through the relay (articles()), and posts the result, which is applied
 * here. A job whose node goes quiet is unlocked after LOCK_MS and picked up
 * again by whoever comes next.
 */
class RemoteContentService {
  private readonly workers = new Map<QueueName, Worker>()

  /** Up to `max` jobs for a node, each locked for LOCK_MS. */
  async lease(nodeId: number, max: number): Promise<RemoteJob[]> {
    const jobs: RemoteJob[] = []
    for (const queue of QUEUES) {
      while (jobs.length < max) {
        const token = `node-${nodeId}:${randomUUID()}`
        const job = await this.worker(queue).getNextJob(token, { block: false })
        if (!job) break
        try {
          const task = await this.prepare(queue, job)
          if (!task) {
            await job.moveToCompleted(null, token, false)
            continue
          }
          // What applying needs, kept with the job: the node's copy of the task isn't trusted.
          if (task.kind === 'inspect') await job.updateData({ ...job.data, remote: { fileId: task.fileId, name: task.name } })
          jobs.push({ queue, jobId: job.id!, token, task })
        } catch (error: any) {
          await job.moveToFailed(error, token, false).catch(() => {})
        }
      }
    }
    return jobs
  }

  /** Applies a node's result and completes the job. */
  async finish(queue: QueueName, jobId: string, token: string, result: InspectProbe | WalkProbe) {
    const job = await this.job(queue, jobId)
    let summary: unknown
    if (queue === 'collection-queue') {
      const remote = job.data.remote
      if (!remote) throw new JobLost(`Job ${jobId} was not leased to a node`)
      const task: InspectTask = { kind: 'inspect', fileId: Number(remote.fileId), name: String(remote.name), segments: [], sevenZip: null }
      const content = await ContentService.apply(task, result as InspectProbe)
      for (const line of describeContent(task.name, content)) logger.info(`[node] ${line}`)
      summary = content.contentType
    } else {
      const outcome = await ArchiveWalkService.apply(Number(job.data.releaseId), result as WalkProbe)
      logger.info(
        `[node] Archive walk of release ${outcome.releaseId}: ${outcome.before} → ${outcome.entries} entries from ${outcome.segments} segments` +
          (outcome.complete ? '' : ` (stopped: ${outcome.stopped})`) +
          (outcome.renamed ? `, renamed to "${outcome.renamed}"` : '')
      )
      summary = outcome
    }
    await this.locked(() => job.moveToCompleted(summary, token, false))
  }

  /** The probe failed on the node (e.g. a missing article): retried like a local failure. */
  async fail(queue: QueueName, jobId: string, token: string, error: string) {
    const job = await this.job(queue, jobId)
    await this.locked(() => job.moveToFailed(new Error(`node: ${error}`), token, false))
  }

  /** A stopping node gives its jobs back, without using up an attempt. */
  async release(queue: QueueName, jobId: string, token: string) {
    const job = await this.job(queue, jobId)
    await this.locked(() => job.moveToWait(token))
  }

  /** One article body for a relay node, on this process's NNTP pool. Null when the server doesn't have it. */
  async article(messageId: string): Promise<Buffer | null> {
    // NNTP_POOL_SIZE 0 (SEED_ARTICLE_CONNECTIONS=0): no connections for this; acquire() would wait forever.
    if (nntpConfig.poolSize <= 0) throw new ArticleRelayOff('Article relay is off (SEED_ARTICLE_CONNECTIONS=0)')
    const conn = await NntpService.acquire()
    let failed = false
    try {
      return await NntpService.timed(NntpService.fetchBody(conn, messageId), `BODY ${messageId}`)
    } catch (error: any) {
      // 430 no such article: a fact about the article, the connection is fine.
      if (/^4\d\d/.test(String(error.message))) return null
      failed = true
      throw error
    } finally {
      NntpService.release(conn, failed)
    }
  }

  /** Closes the lease workers' Redis connections (tests; a server just exits). */
  async close() {
    await Promise.all([...this.workers.values()].map((worker) => worker.close()))
    this.workers.clear()
  }

  isQueue(name: unknown): name is QueueName {
    return QUEUES.includes(name as QueueName)
  }

  private async prepare(queue: QueueName, job: Job): Promise<ContentTask | null> {
    if (queue === 'collection-queue') return ContentService.prepare(Number(job.data.fileId))
    const prepared = await ArchiveWalkService.prepare(Number(job.data.releaseId))
    return 'task' in prepared ? prepared.task : null
  }

  private async job(queue: QueueName, jobId: string): Promise<Job> {
    const job = await Job.fromId(queue === 'collection-queue' ? QueueService.collectionQueue : QueueService.archiveWalkQueue, jobId)
    if (!job) throw new JobLost(`Job ${jobId} is gone`)
    return job
  }

  /** BullMQ refuses a job whose lock another holder has now: the node's result came too late. */
  private async locked(action: () => Promise<unknown>) {
    try {
      await action()
    } catch (error: any) {
      if (/lock|not in the active state|Missing key/i.test(String(error.message))) throw new JobLost(error.message)
      throw error
    }
  }

  /** A worker that only hands out jobs (no processor, never runs). */
  private worker(queue: QueueName): Worker {
    let worker = this.workers.get(queue)
    if (!worker) {
      worker = new Worker(queue, null, { connection: queueConfig.connection, lockDuration: LOCK_MS, autorun: false })
      this.workers.set(queue, worker)
    }
    return worker
  }
}

export default new RemoteContentService()