trying to fix build
This commit is contained in:
parent
269364fbc8
commit
2db7c0dc36
3 changed files with 119 additions and 30 deletions
|
|
@ -1,7 +1,8 @@
|
|||
import { Queue as BullQueue, Worker, QueueEvents, type Job } from 'bullmq';
|
||||
import { getLogger } from '@stock-bot/logger';
|
||||
import { processItems } from './batch-processor';
|
||||
import type { JobData, ProcessOptions, BatchResult } from './types';
|
||||
import { processItems, processBatchJob } from './batch-processor';
|
||||
import { providerRegistry } from './provider-registry';
|
||||
import type { JobData, ProcessOptions, BatchResult, BatchJobData } from './types';
|
||||
|
||||
const logger = getLogger('queue-instance');
|
||||
|
||||
|
|
@ -40,6 +41,9 @@ export class Queue {
|
|||
|
||||
// Initialize queue events
|
||||
this.queueEvents = new QueueEvents(`{${queueName}}`, { connection });
|
||||
|
||||
// Start a worker for this queue
|
||||
this.startWorker();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -177,6 +181,94 @@ export class Queue {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Start a worker for this queue
|
||||
*/
|
||||
private startWorker(): void {
|
||||
const connection = {
|
||||
host: this.redisConfig.host,
|
||||
port: this.redisConfig.port,
|
||||
password: this.redisConfig.password,
|
||||
db: this.redisConfig.db,
|
||||
};
|
||||
|
||||
const worker = new Worker(`{${this.queueName}}`, this.processJob.bind(this), {
|
||||
connection,
|
||||
concurrency: 20,
|
||||
});
|
||||
|
||||
worker.on('completed', job => {
|
||||
logger.debug('Job completed', {
|
||||
id: job.id,
|
||||
name: job.name,
|
||||
queue: this.queueName,
|
||||
});
|
||||
});
|
||||
|
||||
worker.on('failed', (job, err) => {
|
||||
logger.error('Job failed', {
|
||||
id: job?.id,
|
||||
name: job?.name,
|
||||
queue: this.queueName,
|
||||
error: err.message,
|
||||
});
|
||||
});
|
||||
|
||||
this.workers.push(worker);
|
||||
logger.info(`Started worker for queue: ${this.queueName}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a job
|
||||
*/
|
||||
private async processJob(job: Job) {
|
||||
const { provider, operation, payload }: JobData = job.data;
|
||||
|
||||
logger.info('Processing job', {
|
||||
id: job.id,
|
||||
provider,
|
||||
operation,
|
||||
queue: this.queueName,
|
||||
payloadKeys: Object.keys(payload || {}),
|
||||
});
|
||||
|
||||
try {
|
||||
let result;
|
||||
|
||||
if (operation === 'process-batch-items') {
|
||||
// Special handling for batch processing
|
||||
result = await processBatchJob(payload as BatchJobData, this);
|
||||
} else {
|
||||
// Regular handler lookup
|
||||
const handler = providerRegistry.getHandler(provider, operation);
|
||||
|
||||
if (!handler) {
|
||||
throw new Error(`No handler found for ${provider}:${operation}`);
|
||||
}
|
||||
|
||||
result = await handler(payload);
|
||||
}
|
||||
|
||||
logger.info('Job completed successfully', {
|
||||
id: job.id,
|
||||
provider,
|
||||
operation,
|
||||
queue: this.queueName,
|
||||
});
|
||||
|
||||
return result;
|
||||
} catch (error) {
|
||||
logger.error('Job processing failed', {
|
||||
id: job.id,
|
||||
provider,
|
||||
operation,
|
||||
queue: this.queueName,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the BullMQ queue instance (for advanced operations)
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -96,9 +96,7 @@ export class QueueManager {
|
|||
// Step 4: Setup event listeners
|
||||
this.setupEventListeners();
|
||||
|
||||
// Step 5: Initialize batch cache
|
||||
const { initializeBatchCache } = await import('./batch-processor');
|
||||
await initializeBatchCache(this);
|
||||
// Step 5: Batch cache will be initialized by individual Queue instances
|
||||
|
||||
// Step 6: Set up scheduled jobs
|
||||
if (this.enableScheduledJobs) {
|
||||
|
|
@ -373,20 +371,15 @@ export class QueueManager {
|
|||
try {
|
||||
let result;
|
||||
|
||||
if (operation === 'process-batch-items') {
|
||||
// Special handling for batch processing - requires queue manager instance
|
||||
result = await processBatchJob(payload, this);
|
||||
} else {
|
||||
// Regular handler lookup
|
||||
const handler = providerRegistry.getHandler(provider, operation);
|
||||
// Regular handler lookup
|
||||
const handler = providerRegistry.getHandler(provider, operation);
|
||||
|
||||
if (!handler) {
|
||||
throw new Error(`No handler found for ${provider}:${operation}`);
|
||||
}
|
||||
|
||||
result = await handler(payload);
|
||||
if (!handler) {
|
||||
throw new Error(`No handler found for ${provider}:${operation}`);
|
||||
}
|
||||
|
||||
result = await handler(payload);
|
||||
|
||||
logger.info('Job completed successfully', {
|
||||
id: job.id,
|
||||
provider,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue