omgebouwd naar adonisjs, werkt nog niet echt goed
This commit is contained in:
Daan Meijer
@@ -0,0 +1,85 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import Group from '#models/group'
|
||||
import NntpService from '#services/NntpService'
|
||||
import QueueService from '#services/QueueService'
|
||||
|
||||
export default class IndexScheduler extends BaseCommand {
|
||||
public static commandName = 'index:scheduler'
|
||||
public static description = 'Periodically checks for new articles and schedules them for fetching.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
private pool = NntpService
|
||||
private fetchQueue = QueueService.nntpFetchQueue
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Scheduler started. Awaiting tasks...')
|
||||
|
||||
const schedule = async () => {
|
||||
this.logger.info('Checking for new headers...')
|
||||
const groups = await Group.query().where('active', true)
|
||||
|
||||
if (groups.length === 0) {
|
||||
this.logger.info('No active groups to index. Add some via `node ace db:seed` or manually.')
|
||||
return
|
||||
}
|
||||
|
||||
let conn
|
||||
try {
|
||||
conn = await this.pool.acquire()
|
||||
|
||||
for (const group of groups) {
|
||||
try {
|
||||
const groupInfo: any = await conn.group(group.name)
|
||||
|
||||
// nntp-js returns article numbers as strings. We must parse them to BigInts.
|
||||
const firstArticle = BigInt(groupInfo.first)
|
||||
const lastArticle = BigInt(groupInfo.last)
|
||||
|
||||
// lastIndexedId from the database should also be treated as a BigInt.
|
||||
const lastIndexed = group.lastIndexedId ? BigInt(group.lastIndexedId) : null
|
||||
|
||||
const startId = lastIndexed ? lastIndexed + 1n : firstArticle
|
||||
|
||||
if (startId > lastArticle) {
|
||||
this.logger.info(`No new headers for group ${group.name}.`)
|
||||
continue
|
||||
}
|
||||
|
||||
const BATCH_SIZE = 100000n
|
||||
const proposedEndId = startId + BATCH_SIZE - 1n
|
||||
const endId = proposedEndId < lastArticle ? proposedEndId : lastArticle
|
||||
|
||||
this.logger.info(`Queueing fetch job for ${group.name}: articles ${startId} to ${endId}`)
|
||||
await this.fetchQueue.add('fetch-headers', {
|
||||
groupName: group.name,
|
||||
startId: startId.toString(),
|
||||
endId: endId.toString(),
|
||||
})
|
||||
|
||||
group.lastIndexedId = endId
|
||||
await group.save()
|
||||
} catch (err: any) {
|
||||
this.logger.error(`Error processing group ${group.name}: ${err.message}`)
|
||||
}
|
||||
}
|
||||
} catch (err: any) {
|
||||
this.logger.error(`Error in scheduler main loop: ${err.message}`)
|
||||
} finally {
|
||||
if (conn) {
|
||||
this.pool.release(conn)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Run once immediately and then on an interval.
|
||||
schedule()
|
||||
setInterval(schedule, 60000) // 1 minute
|
||||
|
||||
// Keep the command running
|
||||
await new Promise(() => {})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
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 QueueService from '#services/QueueService'
|
||||
import NntpService from '#services/NntpService'
|
||||
import { parseYencMeta } from '#services/YencService'
|
||||
|
||||
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 headerQueue = QueueService.headerQueue
|
||||
|
||||
const worker = new Worker('body-queue', async (job) => {
|
||||
const { header, group } = job.data
|
||||
const messageId = header['message-id']
|
||||
this.logger.debug(`Processing header with unparsable subject: ${header.subject}`)
|
||||
|
||||
let conn
|
||||
try {
|
||||
conn = await pool.acquire()
|
||||
const bodyBuffer: Buffer = (await conn.body(messageId)).data
|
||||
|
||||
try {
|
||||
const meta = parseYencMeta(bodyBuffer)
|
||||
if (meta.header.name) {
|
||||
const { name, part, total } = meta.header
|
||||
const newSubject = `"${name}" yEnc (${part}/${total})`
|
||||
header.subject = newSubject
|
||||
|
||||
this.logger.info(`Found yEnc metadata in body. New subject: ${newSubject}`)
|
||||
await headerQueue.add('process-header', { header, group })
|
||||
} else {
|
||||
this.logger.warning(`Could not find yEnc metadata in body for header: ${header.subject}`)
|
||||
}
|
||||
} catch (parseError: any) {
|
||||
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
|
||||
}
|
||||
} catch (error: any) {
|
||||
this.logger.error(`Error in body worker for message ID ${messageId}: ${error.message}`)
|
||||
throw error
|
||||
} finally {
|
||||
if (conn) {
|
||||
pool.release(conn)
|
||||
}
|
||||
}
|
||||
}, { connection: queueConfig.connection })
|
||||
|
||||
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 new Promise(() => {})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { Worker } from 'bullmq'
|
||||
import queueConfig from '#config/queue'
|
||||
import NntpService from '#services/NntpService'
|
||||
import { YencFile } from '#services/YencFile'
|
||||
import File from '#models/file'
|
||||
import { createExtractorFromData } from 'node-unrar-js'
|
||||
|
||||
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 pool = NntpService
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
const RAR_REGEX = /\.part0*1\.rar$/
|
||||
if (RAR_REGEX.test(file.filename)) {
|
||||
this.logger.info(`File "${file.filename}" is the first part of a RAR set.`)
|
||||
|
||||
const firstPart = file.messageIds['1']
|
||||
if (!firstPart || !firstPart.id) {
|
||||
this.logger.error(`Could not find message ID for the first part of file "${file.filename}".`)
|
||||
return
|
||||
}
|
||||
|
||||
let conn
|
||||
try {
|
||||
conn = await pool.acquire()
|
||||
const bodyBuffer = (await conn.body(`<${firstPart.id}>`)).data
|
||||
|
||||
const yencFile = new YencFile()
|
||||
yencFile.processPart(bodyBuffer)
|
||||
const decodedBuffer = yencFile.getBuffer()
|
||||
|
||||
const extractor = await createExtractorFromData({ data: new Uint8Array(decodedBuffer).buffer })
|
||||
const fileList = extractor.getFileList()
|
||||
// In a real implementation, we would save this file list.
|
||||
this.logger.info(`Files in "${file.filename}": ${JSON.stringify(fileList)}`)
|
||||
|
||||
} catch (error: any) {
|
||||
if (error.code === 430) {
|
||||
this.logger.error(`Article not found for first part of RAR set (Message ID: ${firstPart.id})`)
|
||||
} else {
|
||||
this.logger.error(`Error processing RAR file: ${error.message}`)
|
||||
}
|
||||
} finally {
|
||||
if (conn) {
|
||||
pool.release(conn)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
this.logger.debug(`File "${file.filename}" is not the first part of a RAR set.`)
|
||||
}
|
||||
}, { connection: queueConfig.connection })
|
||||
|
||||
worker.on('failed', (job, err) => {
|
||||
this.logger.error(`Collection job ${job?.id} failed: ${err.message}`)
|
||||
})
|
||||
|
||||
this.logger.info('Collection worker started and listening for jobs.')
|
||||
await new Promise(() => {})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { Worker } from 'bullmq'
|
||||
import queueConfig from '#config/queue'
|
||||
import QueueService from '#services/QueueService'
|
||||
import NntpService from '#services/NntpService'
|
||||
|
||||
export default class FetchWorker extends BaseCommand {
|
||||
public static commandName = 'worker:fetch'
|
||||
public static description = 'Starts a worker to fetch headers from the NNTP server.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Starting fetch worker...')
|
||||
|
||||
const pool = NntpService
|
||||
const headerQueue = QueueService.headerQueue
|
||||
|
||||
const worker = new Worker('nntp-fetch-queue', async (job) => {
|
||||
const { groupName, startId, endId } = job.data
|
||||
this.logger.info(`Processing fetch job for ${groupName}, articles ${startId}-${endId}`)
|
||||
|
||||
let conn
|
||||
try {
|
||||
conn = await pool.acquire()
|
||||
await conn.group(groupName)
|
||||
const overview: any = await conn.xover(startId, endId)
|
||||
this.logger.info(`Fetched ${overview.overviews.length} headers from ${groupName}.`)
|
||||
|
||||
if (overview.overviews.length > 0) {
|
||||
const jobs = overview.overviews.map(([id, header]: [number, any]) => {
|
||||
if (!header) {
|
||||
this.logger.warning(`Header is undefined for job ${job.id}`)
|
||||
return null
|
||||
}
|
||||
return {
|
||||
name: 'process-header',
|
||||
data: { header, group: groupName },
|
||||
opts: { jobId: `${groupName}-${id}` },
|
||||
}
|
||||
}).filter(Boolean)
|
||||
|
||||
await headerQueue.addBulk(jobs)
|
||||
this.logger.info(`Added ${jobs.length} header jobs to the queue for ${groupName}.`)
|
||||
}
|
||||
} catch (error: any) {
|
||||
this.logger.error(`Error fetching headers for ${groupName}: ${error.message}`)
|
||||
throw error
|
||||
} finally {
|
||||
if (conn) {
|
||||
pool.release(conn)
|
||||
}
|
||||
}
|
||||
}, {
|
||||
connection: queueConfig.connection,
|
||||
concurrency: 5,
|
||||
})
|
||||
|
||||
worker.on('failed', (job, err) => {
|
||||
this.logger.error(`Fetch job ${job?.id} failed for group ${job?.data.groupName}: ${err.message}`)
|
||||
})
|
||||
|
||||
this.logger.info('Fetch worker started and listening for jobs.')
|
||||
await new Promise(() => {}) // Keep command running
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { Worker } from 'bullmq'
|
||||
import queueConfig from '#config/queue'
|
||||
import QueueService from '#services/QueueService'
|
||||
import File from '#models/file'
|
||||
|
||||
export default class FileWorker extends BaseCommand {
|
||||
public static commandName = 'worker:file'
|
||||
public static description = 'Starts a worker to process completed files.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Starting file worker...')
|
||||
|
||||
const collectionQueue = QueueService.collectionQueue
|
||||
|
||||
const worker = new Worker('file-queue', async (job) => {
|
||||
const { filename, parts, groups } = job.data
|
||||
const partCount = Object.keys(parts).length
|
||||
this.logger.debug(`Processing complete file: "${filename}" with ${partCount} parts.`)
|
||||
|
||||
const firstPart = JSON.parse(Object.values(parts)[0] as string)
|
||||
const poster = firstPart.from
|
||||
const date = new Date(firstPart.date).getTime()
|
||||
|
||||
const messageIds = Object.entries(parts).reduce((acc, [partNumber, partData]) => {
|
||||
const part = JSON.parse(partData as string)
|
||||
const messageId = part['message-id']
|
||||
if (messageId) {
|
||||
acc[partNumber] = {
|
||||
id: messageId.replace(/[<>]/g, ''),
|
||||
size: part[':bytes'],
|
||||
}
|
||||
} else {
|
||||
this.logger.warning(`Message ID not found for part ${partNumber} of file "${filename}"`)
|
||||
}
|
||||
return acc
|
||||
}, {} as Record<string, any>)
|
||||
|
||||
if (Object.keys(messageIds).length !== partCount) {
|
||||
throw new Error(`Could not process all parts for file "${filename}" due to missing message IDs.`)
|
||||
}
|
||||
|
||||
const file = await File.create({
|
||||
filename,
|
||||
poster,
|
||||
date,
|
||||
parts: partCount,
|
||||
messageIds,
|
||||
groups,
|
||||
})
|
||||
|
||||
this.logger.debug(`Saved file "${filename}" to database with ID: ${file.id}`)
|
||||
|
||||
await collectionQueue.add('process-collection', { fileId: file.id })
|
||||
this.logger.debug(`Added file ID ${file.id} to collection queue.`)
|
||||
}, { connection: queueConfig.connection })
|
||||
|
||||
worker.on('failed', (job, err) => {
|
||||
this.logger.error(`File job ${job?.id} failed: ${err.message}`)
|
||||
})
|
||||
|
||||
this.logger.info('File worker started and listening for jobs.')
|
||||
await new Promise(() => {})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import { Worker } from 'bullmq'
|
||||
import queueConfig from '#config/queue'
|
||||
import QueueService from '#services/QueueService'
|
||||
import RedisService from '#services/RedisService'
|
||||
|
||||
export default class HeaderWorker extends BaseCommand {
|
||||
public static commandName = 'worker:header'
|
||||
public static description = 'Starts a worker to process headers.'
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
this.logger.info('Starting header worker...')
|
||||
|
||||
const redis = RedisService.client
|
||||
const fileQueue = QueueService.fileQueue
|
||||
const bodyQueue = QueueService.bodyQueue
|
||||
|
||||
const worker = new Worker('header-queue', async (job) => {
|
||||
const { header, group } = job.data
|
||||
if (!header || !header.subject) {
|
||||
this.logger.warning(`Received job with invalid header data. JobID: ${job.id}`)
|
||||
return
|
||||
}
|
||||
|
||||
const subject = header.subject
|
||||
const SUBJECT_REGEX = /"(.+)"(?: yEnc)? \((\d+)\/(\d+)\)/
|
||||
const match = subject.match(SUBJECT_REGEX)
|
||||
|
||||
if (match) {
|
||||
const filename = match[1]
|
||||
const part = parseInt(match[2], 10)
|
||||
const total = parseInt(match[3], 10)
|
||||
|
||||
const fileKey = `file:${filename}`
|
||||
await redis.hset(fileKey, part, JSON.stringify(header))
|
||||
|
||||
const partCount = await redis.hlen(fileKey)
|
||||
|
||||
if (partCount === total) {
|
||||
const fileParts = await redis.hgetall(fileKey)
|
||||
await fileQueue.add('process-file', { filename, parts: fileParts, groups: [group] })
|
||||
await redis.del(fileKey)
|
||||
this.logger.info(`File "${filename}" is complete and moved to file-queue.`)
|
||||
} else {
|
||||
this.logger.info(`Stored part ${part}/${total} for file "${filename}"`)
|
||||
}
|
||||
} else {
|
||||
this.logger.warning(`Could not parse subject: "${subject}". Moving to body-queue.`)
|
||||
await bodyQueue.add('process-body', { header, group })
|
||||
}
|
||||
}, { connection: queueConfig.connection })
|
||||
|
||||
worker.on('failed', (job, err) => {
|
||||
this.logger.error(`Header job ${job?.id} failed: ${err.message}`)
|
||||
})
|
||||
|
||||
this.logger.info('Header worker started and listening for jobs.')
|
||||
await new Promise(() => {})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user