56 lines
1.6 KiB
JavaScript
56 lines
1.6 KiB
JavaScript
const path = require('path');
|
|
require('dotenv').config({ path: path.join(__dirname, '../.env') });
|
|
require('dotenv').config({ path: path.join(__dirname, '../../backend/.env') });
|
|
|
|
const { Worker } = require('bullmq');
|
|
const { connectDB } = require('../src/config/database');
|
|
const { getRedisConnection, verifyRedisConnection, QUEUE_PREFIX, INGEST_QUEUE } = require('../src/config/queue');
|
|
|
|
function attachWorkerErrors(worker, label) {
|
|
worker.on('error', (err) => {
|
|
console.error(`[${label}] Redis/BullMQ error:`, err.message);
|
|
if (err.code === 'ECONNRESET') {
|
|
console.error(
|
|
`[${label}] Connection reset — if using DigitalOcean Redis, ensure REDIS_URL uses rediss:// or REDIS_TLS=true`,
|
|
);
|
|
}
|
|
});
|
|
}
|
|
|
|
async function main() {
|
|
await connectDB();
|
|
await verifyRedisConnection();
|
|
|
|
const { runConciergeIngest } = require('../../backend/src/utils/conciergeIngest');
|
|
|
|
const worker = new Worker(
|
|
INGEST_QUEUE,
|
|
async (job) => {
|
|
const { sourceId } = job.data;
|
|
console.log('[ingest] processing', sourceId);
|
|
const result = await runConciergeIngest(sourceId);
|
|
console.log('[ingest] done', sourceId, result);
|
|
return result;
|
|
},
|
|
{
|
|
connection: getRedisConnection(),
|
|
prefix: QUEUE_PREFIX,
|
|
concurrency: 2,
|
|
lockDuration: 900000,
|
|
stalledInterval: 120000,
|
|
},
|
|
);
|
|
|
|
worker.on('failed', (job, err) => {
|
|
console.error('[ingest] failed', job?.id, err.message);
|
|
});
|
|
attachWorkerErrors(worker, 'ingest');
|
|
|
|
console.log('[ingest] worker started');
|
|
}
|
|
|
|
main().catch((e) => {
|
|
console.error(e);
|
|
process.exit(1);
|
|
});
|