Daan MeijerandClaude Opus 5.5
2dbca9b152
Inspection is split into prepare (database), probe (lib/content/probe: downloads, decoding and parsing, no database) and apply (database). worker:collection runs all three; seed:node with SEED_INSPECTORS leases jobs from the same queues through the seed API (BullMQ manual leases), fetches articles through the seed API and posts the probes back. A node now needs only SEED_URL and SEED_TOKEN: the settings the app requires but a node never uses get placeholders for seed:node. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
76 lines
3.0 KiB
TypeScript
76 lines
3.0 KiB
TypeScript
import { BaseCommand } from '@adonisjs/core/ace'
|
|
import { CommandOptions } from '@adonisjs/core/types/ace'
|
|
import { Worker } from 'bullmq'
|
|
import queueConfig from '#config/queue'
|
|
import env from '#start/env'
|
|
import Shutdown from '#services/Shutdown'
|
|
import QueueService from '#services/QueueService'
|
|
import NntpService from '#services/NntpService'
|
|
import ContentService, { describeContent } from '#services/ContentService'
|
|
import ArchiveWalkService from '#services/ArchiveWalkService'
|
|
import File from '#models/file'
|
|
|
|
export default class CollectionWorker extends BaseCommand {
|
|
public static commandName = 'worker:collection'
|
|
public static description = 'Starts a worker to process file collections (e.g., RAR archives).'
|
|
|
|
public static options: CommandOptions = {
|
|
startApp: true,
|
|
}
|
|
|
|
public async run() {
|
|
this.logger.info('Starting collection worker...')
|
|
|
|
const worker = new Worker('collection-queue', async (job) => {
|
|
const { fileId } = job.data
|
|
this.logger.debug(`Processing file ID ${fileId} for collection.`)
|
|
|
|
const file = await File.find(fileId)
|
|
if (!file) {
|
|
this.logger.error(`File with ID ${fileId} not found in the database.`)
|
|
return
|
|
}
|
|
|
|
try {
|
|
const content = await ContentService.inspect(fileId)
|
|
for (const line of describeContent(file.filename, content)) this.logger.info(line)
|
|
} catch (error: any) {
|
|
this.logger.warning(`Content inspection of "${file.filename}" failed: ${error.message}`)
|
|
}
|
|
}, {
|
|
connection: queueConfig.connection,
|
|
// Each job holds one NNTP connection while downloading; keep <= the pool size.
|
|
concurrency: env.get('COLLECTION_CONCURRENCY', 1),
|
|
})
|
|
|
|
worker.on('failed', (job, err) => {
|
|
this.logger.error(`Collection job ${job?.id} failed: ${err.message}`)
|
|
})
|
|
|
|
// Complete archive listings (D40): background work beside inspection, on the same NNTP pool.
|
|
const walkConcurrency = env.get('ARCHIVE_WALK_CONCURRENCY', 1)
|
|
const walker = walkConcurrency > 0
|
|
? new Worker('archive-walk-queue', async (job) => {
|
|
const o = await ArchiveWalkService.walk(Number(job.data.releaseId))
|
|
if (o.skipped) return o
|
|
this.logger.info(
|
|
`Archive walk of release ${o.releaseId}: ${o.before} → ${o.entries} entries from ${o.segments} segments` +
|
|
(o.complete ? '' : ` (stopped: ${o.stopped})`) +
|
|
(o.renamed ? `, renamed to "${o.renamed}"` : '')
|
|
)
|
|
return o
|
|
}, { connection: queueConfig.connection, concurrency: walkConcurrency })
|
|
: null
|
|
walker?.on('failed', (job, err) => {
|
|
this.logger.error(`Archive walk job ${job?.id} failed: ${err.message}`)
|
|
})
|
|
|
|
this.logger.info('Collection worker started and listening for jobs.')
|
|
await Shutdown.wait()
|
|
this.logger.info('Shutting down: finishing active jobs...')
|
|
await Promise.all([worker.close(), walker?.close()])
|
|
await NntpService.shutdown()
|
|
await QueueService.closeAll()
|
|
}
|
|
}
|