98 lines
3.6 KiB
TypeScript
98 lines
3.6 KiB
TypeScript
import { BaseCommand } from '@adonisjs/core/ace'
|
|
import { CommandOptions } from '@adonisjs/core/types/ace'
|
|
import { Worker } from 'bullmq'
|
|
import fs from 'node:fs/promises'
|
|
import path from 'node:path'
|
|
import queueConfig from '#config/queue'
|
|
import Shutdown from '#services/Shutdown'
|
|
import QueueService from '#services/QueueService'
|
|
import env from '#start/env'
|
|
import NntpService from '#services/NntpService'
|
|
import IngestService, { segmentFromYenc } from '#services/IngestService'
|
|
import { parseYencMeta, YencError } from '#lib/yenc/index'
|
|
import { firstChunkMd5 } from '#services/ContentService'
|
|
|
|
export default class BodyWorker extends BaseCommand {
|
|
public static commandName = 'worker:body'
|
|
public static description = 'Starts a worker to process article bodies for yEnc metadata.'
|
|
|
|
public static options: CommandOptions = {
|
|
startApp: true,
|
|
}
|
|
|
|
public async run() {
|
|
this.logger.info('Starting body worker...')
|
|
|
|
const pool = NntpService
|
|
|
|
const worker = new Worker('body-queue', async (job) => {
|
|
const { header, group, articleNumber } = job.data
|
|
const messageId = header['message-id']
|
|
this.logger.debug(`Processing header with unparsable subject: ${header.subject}`)
|
|
|
|
let conn
|
|
let failed = false
|
|
try {
|
|
conn = await pool.acquire()
|
|
const bodyBuffer = await pool.fetchBody(conn, messageId)
|
|
|
|
let meta
|
|
try {
|
|
meta = parseYencMeta(bodyBuffer)
|
|
} catch (parseError: any) {
|
|
if (parseError instanceof YencError && bodyBuffer.indexOf('=ybegin ') === -1) {
|
|
this.logger.debug(`Not a yEnc post, skipping: ${messageId}`)
|
|
return
|
|
}
|
|
this.logger.error(`Failed to parse yEnc data for message ID ${messageId}. Dumping buffer.`)
|
|
const debugDir = path.join(this.app.appRoot.pathname, 'debug')
|
|
await fs.mkdir(debugDir, { recursive: true })
|
|
const timestamp = new Date().toISOString().replace(/:/g, '-')
|
|
const dumpFile = path.join(debugDir, `body-error-${timestamp}-${messageId.replace(/[<>]/g, '')}.bin`)
|
|
await fs.writeFile(dumpFile, bodyBuffer)
|
|
this.logger.error(`Problematic body buffer saved to: ${dumpFile}`)
|
|
throw parseError
|
|
}
|
|
|
|
pool.release(conn)
|
|
conn = undefined
|
|
|
|
const segment = segmentFromYenc(header, group, meta, articleNumber)
|
|
if (segment?.number === 1) segment.md5_16k = firstChunkMd5(bodyBuffer, meta.begin.size)
|
|
if (!segment) {
|
|
this.logger.warning(`Inconsistent yEnc part numbering in ${messageId}, skipping.`)
|
|
return
|
|
}
|
|
const completed = await IngestService.ingest([segment])
|
|
this.logger.debug(
|
|
`Stored part ${segment.number}/${segment.totalParts} of "${segment.name}"` +
|
|
(completed.length > 0 ? ' (file complete)' : '')
|
|
)
|
|
} catch (error: any) {
|
|
failed = true
|
|
this.logger.error(`Error in body worker for message ID ${messageId}: ${error.message}`)
|
|
throw error
|
|
} finally {
|
|
if (conn) {
|
|
pool.release(conn, failed)
|
|
}
|
|
}
|
|
}, {
|
|
connection: queueConfig.connection,
|
|
// Each job holds one NNTP connection while downloading, so keep this <= the pool size.
|
|
concurrency: env.get('BODY_CONCURRENCY', 1),
|
|
})
|
|
|
|
worker.on('failed', (job, err) => {
|
|
this.logger.error(`Body job ${job?.id} failed: ${err.message}`)
|
|
})
|
|
|
|
this.logger.info('Body worker started and listening for jobs.')
|
|
await Shutdown.wait()
|
|
this.logger.info('Shutting down: finishing active jobs...')
|
|
await worker.close()
|
|
await NntpService.shutdown()
|
|
await QueueService.closeAll()
|
|
}
|
|
}
|