81 lines
2.0 KiB
TypeScript
81 lines
2.0 KiB
TypeScript
import { Job, Worker, type ConnectionOptions } from 'bullmq';
|
|
import IORedis from 'ioredis';
|
|
|
|
import { validateConfig } from '@hexahost/config';
|
|
import { prisma } from '@hexahost/database';
|
|
|
|
import { startHealthServer } from './health';
|
|
import { logger } from './logger';
|
|
import { WORKER_QUEUES, type WorkerQueueName } from './queues';
|
|
|
|
function createRedisConnection(redisUrl: string): IORedis {
|
|
return new IORedis(redisUrl, {
|
|
maxRetriesPerRequest: null,
|
|
});
|
|
}
|
|
|
|
function createQueueWorker(
|
|
queueName: WorkerQueueName,
|
|
connection: ConnectionOptions,
|
|
): Worker {
|
|
return new Worker(
|
|
queueName,
|
|
async (job: Job) => {
|
|
logger.info(
|
|
{ queue: queueName, jobId: job.id, jobName: job.name },
|
|
'Processing job',
|
|
);
|
|
|
|
return { acknowledged: true, queue: queueName };
|
|
},
|
|
{ connection },
|
|
);
|
|
}
|
|
|
|
async function bootstrap(): Promise<void> {
|
|
const config = validateConfig();
|
|
|
|
const healthServer = startHealthServer();
|
|
const redis = createRedisConnection(config.REDIS_URL);
|
|
const workers = WORKER_QUEUES.map((queueName) =>
|
|
createQueueWorker(queueName, redis),
|
|
);
|
|
|
|
for (const worker of workers) {
|
|
worker.on('completed', (job: Job | undefined) => {
|
|
logger.info({ queue: worker.name, jobId: job?.id }, 'Job completed');
|
|
});
|
|
|
|
worker.on('failed', (job: Job | undefined, error: Error) => {
|
|
logger.error(
|
|
{ queue: worker.name, jobId: job?.id, err: error },
|
|
'Job failed',
|
|
);
|
|
});
|
|
}
|
|
|
|
logger.info({ queues: WORKER_QUEUES }, 'Worker started');
|
|
|
|
const shutdown = async (signal: string): Promise<void> => {
|
|
logger.info({ signal }, 'Shutting down worker');
|
|
|
|
await Promise.all(workers.map((worker) => worker.close()));
|
|
await redis.quit();
|
|
await prisma.$disconnect();
|
|
|
|
healthServer.close(() => {
|
|
process.exit(0);
|
|
});
|
|
};
|
|
|
|
process.on('SIGINT', () => {
|
|
void shutdown('SIGINT');
|
|
});
|
|
|
|
process.on('SIGTERM', () => {
|
|
void shutdown('SIGTERM');
|
|
});
|
|
}
|
|
|
|
void bootstrap();
|