linxus fs fixes
This commit is contained in:
parent
ac23b70146
commit
0b7846fe67
292 changed files with 41947 additions and 41947 deletions
|
|
@ -1,478 +1,478 @@
|
|||
import { Queue, Worker, QueueEvents } from 'bullmq';
|
||||
import { getLogger } from '@stock-bot/logger';
|
||||
import { providerRegistry } from './provider-registry.service';
|
||||
|
||||
export interface JobData {
|
||||
type: string;
|
||||
service: string;
|
||||
provider: string;
|
||||
operation: string;
|
||||
payload: any;
|
||||
priority?: number;
|
||||
immediately?: boolean;
|
||||
}
|
||||
|
||||
export class QueueService {
|
||||
private logger = getLogger('queue-service');
|
||||
private queue!: Queue;
|
||||
private workers: Worker[] = [];
|
||||
private queueEvents!: QueueEvents;
|
||||
private isInitialized = false;
|
||||
|
||||
constructor() {
|
||||
// Don't initialize in constructor to allow for proper async initialization
|
||||
}
|
||||
|
||||
async initialize() {
|
||||
if (this.isInitialized) {
|
||||
this.logger.warn('Queue service already initialized');
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.info('Initializing queue service...');
|
||||
|
||||
// Register all providers first
|
||||
await this.registerProviders();
|
||||
|
||||
const connection = {
|
||||
host: process.env.DRAGONFLY_HOST || 'localhost',
|
||||
port: parseInt(process.env.DRAGONFLY_PORT || '6379'),
|
||||
// Add these Redis-specific options to fix the undeclared key issue
|
||||
maxRetriesPerRequest: null,
|
||||
retryDelayOnFailover: 100,
|
||||
enableReadyCheck: false,
|
||||
lazyConnect: true,
|
||||
// Disable Redis Cluster mode if you're using standalone Redis/Dragonfly
|
||||
enableOfflineQueue: false
|
||||
};
|
||||
|
||||
// Worker configuration
|
||||
const workerCount = parseInt(process.env.WORKER_COUNT || '5');
|
||||
const concurrencyPerWorker = parseInt(process.env.WORKER_CONCURRENCY || '20');
|
||||
|
||||
this.logger.info('Connecting to Redis/Dragonfly', connection);
|
||||
|
||||
try {
|
||||
this.queue = new Queue('{data-service-queue}', {
|
||||
connection,
|
||||
defaultJobOptions: {
|
||||
removeOnComplete: 10,
|
||||
removeOnFail: 5,
|
||||
attempts: 3,
|
||||
backoff: {
|
||||
type: 'exponential',
|
||||
delay: 1000,
|
||||
}
|
||||
}
|
||||
});
|
||||
// Create multiple workers
|
||||
for (let i = 0; i < workerCount; i++) {
|
||||
const worker = new Worker(
|
||||
'{data-service-queue}',
|
||||
this.processJob.bind(this),
|
||||
{
|
||||
connection: { ...connection }, // Each worker gets its own connection
|
||||
concurrency: concurrencyPerWorker,
|
||||
maxStalledCount: 1,
|
||||
stalledInterval: 30000,
|
||||
}
|
||||
);
|
||||
// Add worker-specific logging
|
||||
worker.on('ready', () => {
|
||||
this.logger.info(`Worker ${i + 1} ready`, { workerId: i + 1 });
|
||||
});
|
||||
|
||||
worker.on('error', (error) => {
|
||||
this.logger.error(`Worker ${i + 1} error`, { workerId: i + 1, error });
|
||||
});
|
||||
|
||||
this.workers.push(worker);
|
||||
}
|
||||
this.queueEvents = new QueueEvents('{data-service-queue}', { connection }); // Test connection
|
||||
|
||||
// Wait for all workers to be ready
|
||||
await this.queue.waitUntilReady();
|
||||
await Promise.all(this.workers.map(worker => worker.waitUntilReady()));
|
||||
await this.queueEvents.waitUntilReady();
|
||||
|
||||
this.setupEventListeners();
|
||||
this.isInitialized = true;
|
||||
this.logger.info('Queue service initialized successfully');
|
||||
|
||||
await this.setupScheduledTasks();
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to initialize queue service', { error });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// Update getTotalConcurrency method
|
||||
getTotalConcurrency() {
|
||||
if (!this.isInitialized) {
|
||||
return 0;
|
||||
}
|
||||
return this.workers.reduce((total, worker) => {
|
||||
return total + (worker.opts.concurrency || 1);
|
||||
}, 0);
|
||||
}
|
||||
|
||||
private async registerProviders() {
|
||||
this.logger.info('Registering providers...');
|
||||
|
||||
try {
|
||||
// Import and register all providers
|
||||
const { proxyProvider } = await import('../providers/proxy.provider');
|
||||
const { quotemediaProvider } = await import('../providers/quotemedia.provider');
|
||||
const { yahooProvider } = await import('../providers/yahoo.provider');
|
||||
|
||||
providerRegistry.registerProvider(proxyProvider);
|
||||
providerRegistry.registerProvider(quotemediaProvider);
|
||||
providerRegistry.registerProvider(yahooProvider);
|
||||
|
||||
this.logger.info('All providers registered successfully');
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to register providers', { error });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async processJob(job: any) {
|
||||
const { service, provider, operation, payload }: JobData = job.data;
|
||||
|
||||
this.logger.info('Processing job', {
|
||||
id: job.id,
|
||||
service,
|
||||
provider,
|
||||
operation,
|
||||
payloadKeys: Object.keys(payload || {})
|
||||
});
|
||||
|
||||
try {
|
||||
// Get handler from registry
|
||||
const handler = providerRegistry.getHandler(service, provider, operation);
|
||||
|
||||
if (!handler) {
|
||||
throw new Error(`No handler found for ${service}:${provider}:${operation}`);
|
||||
}
|
||||
|
||||
// Execute the handler
|
||||
const result = await handler(payload);
|
||||
|
||||
this.logger.info('Job completed successfully', {
|
||||
id: job.id,
|
||||
service,
|
||||
provider,
|
||||
operation
|
||||
});
|
||||
|
||||
return result;
|
||||
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
this.logger.error('Job failed', {
|
||||
id: job.id,
|
||||
service,
|
||||
provider,
|
||||
operation,
|
||||
error: errorMessage
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async addBulk(jobs: any[]) : Promise<any[]> {
|
||||
return await this.queue.addBulk(jobs)
|
||||
}
|
||||
private setupEventListeners() {
|
||||
this.queueEvents.on('completed', (job) => {
|
||||
this.logger.info('Job completed', { id: job.jobId });
|
||||
});
|
||||
|
||||
this.queueEvents.on('failed', (job) => {
|
||||
this.logger.error('Job failed', { id: job.jobId, error: job.failedReason });
|
||||
});
|
||||
|
||||
// Note: Worker-specific events are already set up during worker creation
|
||||
// No need for additional progress events since we handle them per-worker
|
||||
}
|
||||
private async setupScheduledTasks() {
|
||||
try {
|
||||
this.logger.info('Setting up scheduled tasks from providers...');
|
||||
|
||||
// Get all scheduled jobs from all providers
|
||||
const allScheduledJobs = providerRegistry.getAllScheduledJobs();
|
||||
|
||||
if (allScheduledJobs.length === 0) {
|
||||
this.logger.warn('No scheduled jobs found in providers');
|
||||
return;
|
||||
}
|
||||
|
||||
// Get existing repeatable jobs for comparison
|
||||
const existingJobs = await this.queue.getRepeatableJobs();
|
||||
this.logger.info(`Found ${existingJobs.length} existing repeatable jobs`);
|
||||
|
||||
let successCount = 0;
|
||||
let failureCount = 0;
|
||||
let updatedCount = 0;
|
||||
let newCount = 0;
|
||||
|
||||
// Process each scheduled job
|
||||
for (const { service, provider, job } of allScheduledJobs) {
|
||||
try {
|
||||
const jobKey = `${service}-${provider}-${job.operation}`;
|
||||
|
||||
// Check if this job already exists
|
||||
const existingJob = existingJobs.find(existing =>
|
||||
existing.key?.includes(jobKey) || existing.name === job.type
|
||||
);
|
||||
|
||||
if (existingJob) {
|
||||
// Check if the job needs updating (different cron pattern or config)
|
||||
const needsUpdate = existingJob.pattern !== job.cronPattern;
|
||||
|
||||
if (needsUpdate) {
|
||||
this.logger.info('Job configuration changed, updating', {
|
||||
jobKey,
|
||||
oldPattern: existingJob.pattern,
|
||||
newPattern: job.cronPattern
|
||||
});
|
||||
updatedCount++;
|
||||
} else {
|
||||
this.logger.debug('Job unchanged, skipping', { jobKey });
|
||||
successCount++;
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
newCount++;
|
||||
}
|
||||
|
||||
// Add delay between job registrations
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
|
||||
await this.addRecurringJob({
|
||||
type: job.type,
|
||||
service: service,
|
||||
provider: provider,
|
||||
operation: job.operation,
|
||||
payload: job.payload,
|
||||
priority: job.priority,
|
||||
immediately: job.immediately || false
|
||||
}, job.cronPattern);
|
||||
|
||||
this.logger.info('Scheduled job registered', {
|
||||
type: job.type,
|
||||
service,
|
||||
provider,
|
||||
operation: job.operation,
|
||||
cronPattern: job.cronPattern,
|
||||
description: job.description,
|
||||
immediately: job.immediately || false
|
||||
});
|
||||
|
||||
successCount++;
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to register scheduled job', {
|
||||
type: job.type,
|
||||
service,
|
||||
provider,
|
||||
error: error instanceof Error ? error.message : String(error)
|
||||
});
|
||||
failureCount++;
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Scheduled tasks setup complete`, {
|
||||
total: allScheduledJobs.length,
|
||||
successful: successCount,
|
||||
failed: failureCount,
|
||||
updated: updatedCount,
|
||||
new: newCount
|
||||
});
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to setup scheduled tasks', error);
|
||||
}
|
||||
}
|
||||
|
||||
async addJob(jobData: JobData, options?: any) {
|
||||
if (!this.isInitialized) {
|
||||
throw new Error('Queue service not initialized. Call initialize() first.');
|
||||
}
|
||||
return this.queue.add(jobData.type, jobData, {
|
||||
priority: jobData.priority || 0,
|
||||
removeOnComplete: 10,
|
||||
removeOnFail: 5,
|
||||
...options
|
||||
});
|
||||
}
|
||||
|
||||
async addRecurringJob(jobData: JobData, cronPattern: string, options?: any) {
|
||||
if (!this.isInitialized) {
|
||||
throw new Error('Queue service not initialized. Call initialize() first.');
|
||||
}
|
||||
|
||||
try {
|
||||
// Create a unique job key for this specific job
|
||||
const jobKey = `${jobData.service}-${jobData.provider}-${jobData.operation}`;
|
||||
|
||||
// Get all existing repeatable jobs
|
||||
const existingJobs = await this.queue.getRepeatableJobs();
|
||||
|
||||
// Find and remove the existing job with the same key if it exists
|
||||
const existingJob = existingJobs.find(job => {
|
||||
// Check if this is the same job by comparing key components
|
||||
return job.key?.includes(jobKey) || job.name === jobData.type;
|
||||
});
|
||||
|
||||
if (existingJob) {
|
||||
this.logger.info('Updating existing recurring job', {
|
||||
jobKey,
|
||||
existingPattern: existingJob.pattern,
|
||||
newPattern: cronPattern
|
||||
});
|
||||
|
||||
// Remove the existing job
|
||||
await this.queue.removeRepeatableByKey(existingJob.key);
|
||||
|
||||
// Small delay to ensure cleanup is complete
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
} else {
|
||||
this.logger.info('Creating new recurring job', { jobKey, cronPattern });
|
||||
}
|
||||
|
||||
// Add the new/updated recurring job
|
||||
const job = await this.queue.add(jobData.type, jobData, {
|
||||
repeat: {
|
||||
pattern: cronPattern,
|
||||
tz: 'UTC',
|
||||
immediately: jobData.immediately || false,
|
||||
},
|
||||
// Use a consistent jobId for this specific recurring job
|
||||
jobId: `recurring-${jobKey}`,
|
||||
removeOnComplete: 1,
|
||||
removeOnFail: 1,
|
||||
attempts: 2,
|
||||
backoff: {
|
||||
type: 'fixed',
|
||||
delay: 5000
|
||||
},
|
||||
...options
|
||||
});
|
||||
|
||||
this.logger.info('Recurring job added/updated successfully', {
|
||||
jobKey,
|
||||
type: jobData.type,
|
||||
cronPattern,
|
||||
immediately: jobData.immediately || false
|
||||
});
|
||||
|
||||
return job;
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to add/update recurring job', {
|
||||
jobData,
|
||||
cronPattern,
|
||||
error: error instanceof Error ? error.message : String(error)
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async getJobStats() {
|
||||
if (!this.isInitialized) {
|
||||
throw new Error('Queue service not initialized. Call initialize() first.');
|
||||
}
|
||||
const [waiting, active, completed, failed, delayed] = await Promise.all([
|
||||
this.queue.getWaiting(),
|
||||
this.queue.getActive(),
|
||||
this.queue.getCompleted(),
|
||||
this.queue.getFailed(),
|
||||
this.queue.getDelayed()
|
||||
]);
|
||||
|
||||
return {
|
||||
waiting: waiting.length,
|
||||
active: active.length,
|
||||
completed: completed.length,
|
||||
failed: failed.length,
|
||||
delayed: delayed.length
|
||||
};
|
||||
}
|
||||
|
||||
async drainQueue() {
|
||||
if (!this.isInitialized) {
|
||||
await this.queue.drain()
|
||||
}
|
||||
}
|
||||
|
||||
async getQueueStatus() {
|
||||
if (!this.isInitialized) {
|
||||
throw new Error('Queue service not initialized. Call initialize() first.');
|
||||
}
|
||||
const stats = await this.getJobStats();
|
||||
return {
|
||||
...stats,
|
||||
workers: this.getWorkerCount(),
|
||||
totalConcurrency: this.getTotalConcurrency(),
|
||||
queue: this.queue.name,
|
||||
connection: {
|
||||
host: process.env.DRAGONFLY_HOST || 'localhost',
|
||||
port: parseInt(process.env.DRAGONFLY_PORT || '6379')
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
getWorkerCount() {
|
||||
if (!this.isInitialized) {
|
||||
return 0;
|
||||
}
|
||||
return this.workers.length;
|
||||
}
|
||||
|
||||
getRegisteredProviders() {
|
||||
return providerRegistry.getProviders().map(({ key, config }) => ({
|
||||
key,
|
||||
name: config.name,
|
||||
service: config.service,
|
||||
operations: Object.keys(config.operations),
|
||||
scheduledJobs: config.scheduledJobs?.length || 0
|
||||
}));
|
||||
}
|
||||
|
||||
getScheduledJobsInfo() {
|
||||
return providerRegistry.getAllScheduledJobs().map(({ service, provider, job }) => ({
|
||||
id: `${service}-${provider}-${job.type}`,
|
||||
service,
|
||||
provider,
|
||||
type: job.type,
|
||||
operation: job.operation,
|
||||
cronPattern: job.cronPattern,
|
||||
priority: job.priority,
|
||||
description: job.description,
|
||||
immediately: job.immediately || false
|
||||
}));
|
||||
}
|
||||
async shutdown() {
|
||||
if (!this.isInitialized) {
|
||||
this.logger.warn('Queue service not initialized, nothing to shutdown');
|
||||
return;
|
||||
}
|
||||
this.logger.info('Shutting down queue service');
|
||||
|
||||
// Close all workers
|
||||
this.logger.info(`Closing ${this.workers.length} workers...`);
|
||||
await Promise.all(this.workers.map((worker, index) => {
|
||||
this.logger.debug(`Closing worker ${index + 1}`);
|
||||
return worker.close();
|
||||
}));
|
||||
|
||||
await this.queue.close();
|
||||
await this.queueEvents.close();
|
||||
this.isInitialized = false;
|
||||
this.logger.info('Queue service shutdown complete');
|
||||
}
|
||||
}
|
||||
|
||||
export const queueManager = new QueueService();
|
||||
import { Queue, Worker, QueueEvents } from 'bullmq';
|
||||
import { getLogger } from '@stock-bot/logger';
|
||||
import { providerRegistry } from './provider-registry.service';
|
||||
|
||||
export interface JobData {
|
||||
type: string;
|
||||
service: string;
|
||||
provider: string;
|
||||
operation: string;
|
||||
payload: any;
|
||||
priority?: number;
|
||||
immediately?: boolean;
|
||||
}
|
||||
|
||||
export class QueueService {
|
||||
private logger = getLogger('queue-service');
|
||||
private queue!: Queue;
|
||||
private workers: Worker[] = [];
|
||||
private queueEvents!: QueueEvents;
|
||||
private isInitialized = false;
|
||||
|
||||
constructor() {
|
||||
// Don't initialize in constructor to allow for proper async initialization
|
||||
}
|
||||
|
||||
async initialize() {
|
||||
if (this.isInitialized) {
|
||||
this.logger.warn('Queue service already initialized');
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.info('Initializing queue service...');
|
||||
|
||||
// Register all providers first
|
||||
await this.registerProviders();
|
||||
|
||||
const connection = {
|
||||
host: process.env.DRAGONFLY_HOST || 'localhost',
|
||||
port: parseInt(process.env.DRAGONFLY_PORT || '6379'),
|
||||
// Add these Redis-specific options to fix the undeclared key issue
|
||||
maxRetriesPerRequest: null,
|
||||
retryDelayOnFailover: 100,
|
||||
enableReadyCheck: false,
|
||||
lazyConnect: true,
|
||||
// Disable Redis Cluster mode if you're using standalone Redis/Dragonfly
|
||||
enableOfflineQueue: false
|
||||
};
|
||||
|
||||
// Worker configuration
|
||||
const workerCount = parseInt(process.env.WORKER_COUNT || '5');
|
||||
const concurrencyPerWorker = parseInt(process.env.WORKER_CONCURRENCY || '20');
|
||||
|
||||
this.logger.info('Connecting to Redis/Dragonfly', connection);
|
||||
|
||||
try {
|
||||
this.queue = new Queue('{data-service-queue}', {
|
||||
connection,
|
||||
defaultJobOptions: {
|
||||
removeOnComplete: 10,
|
||||
removeOnFail: 5,
|
||||
attempts: 3,
|
||||
backoff: {
|
||||
type: 'exponential',
|
||||
delay: 1000,
|
||||
}
|
||||
}
|
||||
});
|
||||
// Create multiple workers
|
||||
for (let i = 0; i < workerCount; i++) {
|
||||
const worker = new Worker(
|
||||
'{data-service-queue}',
|
||||
this.processJob.bind(this),
|
||||
{
|
||||
connection: { ...connection }, // Each worker gets its own connection
|
||||
concurrency: concurrencyPerWorker,
|
||||
maxStalledCount: 1,
|
||||
stalledInterval: 30000,
|
||||
}
|
||||
);
|
||||
// Add worker-specific logging
|
||||
worker.on('ready', () => {
|
||||
this.logger.info(`Worker ${i + 1} ready`, { workerId: i + 1 });
|
||||
});
|
||||
|
||||
worker.on('error', (error) => {
|
||||
this.logger.error(`Worker ${i + 1} error`, { workerId: i + 1, error });
|
||||
});
|
||||
|
||||
this.workers.push(worker);
|
||||
}
|
||||
this.queueEvents = new QueueEvents('{data-service-queue}', { connection }); // Test connection
|
||||
|
||||
// Wait for all workers to be ready
|
||||
await this.queue.waitUntilReady();
|
||||
await Promise.all(this.workers.map(worker => worker.waitUntilReady()));
|
||||
await this.queueEvents.waitUntilReady();
|
||||
|
||||
this.setupEventListeners();
|
||||
this.isInitialized = true;
|
||||
this.logger.info('Queue service initialized successfully');
|
||||
|
||||
await this.setupScheduledTasks();
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to initialize queue service', { error });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// Update getTotalConcurrency method
|
||||
getTotalConcurrency() {
|
||||
if (!this.isInitialized) {
|
||||
return 0;
|
||||
}
|
||||
return this.workers.reduce((total, worker) => {
|
||||
return total + (worker.opts.concurrency || 1);
|
||||
}, 0);
|
||||
}
|
||||
|
||||
private async registerProviders() {
|
||||
this.logger.info('Registering providers...');
|
||||
|
||||
try {
|
||||
// Import and register all providers
|
||||
const { proxyProvider } = await import('../providers/proxy.provider');
|
||||
const { quotemediaProvider } = await import('../providers/quotemedia.provider');
|
||||
const { yahooProvider } = await import('../providers/yahoo.provider');
|
||||
|
||||
providerRegistry.registerProvider(proxyProvider);
|
||||
providerRegistry.registerProvider(quotemediaProvider);
|
||||
providerRegistry.registerProvider(yahooProvider);
|
||||
|
||||
this.logger.info('All providers registered successfully');
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to register providers', { error });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async processJob(job: any) {
|
||||
const { service, provider, operation, payload }: JobData = job.data;
|
||||
|
||||
this.logger.info('Processing job', {
|
||||
id: job.id,
|
||||
service,
|
||||
provider,
|
||||
operation,
|
||||
payloadKeys: Object.keys(payload || {})
|
||||
});
|
||||
|
||||
try {
|
||||
// Get handler from registry
|
||||
const handler = providerRegistry.getHandler(service, provider, operation);
|
||||
|
||||
if (!handler) {
|
||||
throw new Error(`No handler found for ${service}:${provider}:${operation}`);
|
||||
}
|
||||
|
||||
// Execute the handler
|
||||
const result = await handler(payload);
|
||||
|
||||
this.logger.info('Job completed successfully', {
|
||||
id: job.id,
|
||||
service,
|
||||
provider,
|
||||
operation
|
||||
});
|
||||
|
||||
return result;
|
||||
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
this.logger.error('Job failed', {
|
||||
id: job.id,
|
||||
service,
|
||||
provider,
|
||||
operation,
|
||||
error: errorMessage
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async addBulk(jobs: any[]) : Promise<any[]> {
|
||||
return await this.queue.addBulk(jobs)
|
||||
}
|
||||
private setupEventListeners() {
|
||||
this.queueEvents.on('completed', (job) => {
|
||||
this.logger.info('Job completed', { id: job.jobId });
|
||||
});
|
||||
|
||||
this.queueEvents.on('failed', (job) => {
|
||||
this.logger.error('Job failed', { id: job.jobId, error: job.failedReason });
|
||||
});
|
||||
|
||||
// Note: Worker-specific events are already set up during worker creation
|
||||
// No need for additional progress events since we handle them per-worker
|
||||
}
|
||||
private async setupScheduledTasks() {
|
||||
try {
|
||||
this.logger.info('Setting up scheduled tasks from providers...');
|
||||
|
||||
// Get all scheduled jobs from all providers
|
||||
const allScheduledJobs = providerRegistry.getAllScheduledJobs();
|
||||
|
||||
if (allScheduledJobs.length === 0) {
|
||||
this.logger.warn('No scheduled jobs found in providers');
|
||||
return;
|
||||
}
|
||||
|
||||
// Get existing repeatable jobs for comparison
|
||||
const existingJobs = await this.queue.getRepeatableJobs();
|
||||
this.logger.info(`Found ${existingJobs.length} existing repeatable jobs`);
|
||||
|
||||
let successCount = 0;
|
||||
let failureCount = 0;
|
||||
let updatedCount = 0;
|
||||
let newCount = 0;
|
||||
|
||||
// Process each scheduled job
|
||||
for (const { service, provider, job } of allScheduledJobs) {
|
||||
try {
|
||||
const jobKey = `${service}-${provider}-${job.operation}`;
|
||||
|
||||
// Check if this job already exists
|
||||
const existingJob = existingJobs.find(existing =>
|
||||
existing.key?.includes(jobKey) || existing.name === job.type
|
||||
);
|
||||
|
||||
if (existingJob) {
|
||||
// Check if the job needs updating (different cron pattern or config)
|
||||
const needsUpdate = existingJob.pattern !== job.cronPattern;
|
||||
|
||||
if (needsUpdate) {
|
||||
this.logger.info('Job configuration changed, updating', {
|
||||
jobKey,
|
||||
oldPattern: existingJob.pattern,
|
||||
newPattern: job.cronPattern
|
||||
});
|
||||
updatedCount++;
|
||||
} else {
|
||||
this.logger.debug('Job unchanged, skipping', { jobKey });
|
||||
successCount++;
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
newCount++;
|
||||
}
|
||||
|
||||
// Add delay between job registrations
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
|
||||
await this.addRecurringJob({
|
||||
type: job.type,
|
||||
service: service,
|
||||
provider: provider,
|
||||
operation: job.operation,
|
||||
payload: job.payload,
|
||||
priority: job.priority,
|
||||
immediately: job.immediately || false
|
||||
}, job.cronPattern);
|
||||
|
||||
this.logger.info('Scheduled job registered', {
|
||||
type: job.type,
|
||||
service,
|
||||
provider,
|
||||
operation: job.operation,
|
||||
cronPattern: job.cronPattern,
|
||||
description: job.description,
|
||||
immediately: job.immediately || false
|
||||
});
|
||||
|
||||
successCount++;
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to register scheduled job', {
|
||||
type: job.type,
|
||||
service,
|
||||
provider,
|
||||
error: error instanceof Error ? error.message : String(error)
|
||||
});
|
||||
failureCount++;
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Scheduled tasks setup complete`, {
|
||||
total: allScheduledJobs.length,
|
||||
successful: successCount,
|
||||
failed: failureCount,
|
||||
updated: updatedCount,
|
||||
new: newCount
|
||||
});
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to setup scheduled tasks', error);
|
||||
}
|
||||
}
|
||||
|
||||
async addJob(jobData: JobData, options?: any) {
|
||||
if (!this.isInitialized) {
|
||||
throw new Error('Queue service not initialized. Call initialize() first.');
|
||||
}
|
||||
return this.queue.add(jobData.type, jobData, {
|
||||
priority: jobData.priority || 0,
|
||||
removeOnComplete: 10,
|
||||
removeOnFail: 5,
|
||||
...options
|
||||
});
|
||||
}
|
||||
|
||||
async addRecurringJob(jobData: JobData, cronPattern: string, options?: any) {
|
||||
if (!this.isInitialized) {
|
||||
throw new Error('Queue service not initialized. Call initialize() first.');
|
||||
}
|
||||
|
||||
try {
|
||||
// Create a unique job key for this specific job
|
||||
const jobKey = `${jobData.service}-${jobData.provider}-${jobData.operation}`;
|
||||
|
||||
// Get all existing repeatable jobs
|
||||
const existingJobs = await this.queue.getRepeatableJobs();
|
||||
|
||||
// Find and remove the existing job with the same key if it exists
|
||||
const existingJob = existingJobs.find(job => {
|
||||
// Check if this is the same job by comparing key components
|
||||
return job.key?.includes(jobKey) || job.name === jobData.type;
|
||||
});
|
||||
|
||||
if (existingJob) {
|
||||
this.logger.info('Updating existing recurring job', {
|
||||
jobKey,
|
||||
existingPattern: existingJob.pattern,
|
||||
newPattern: cronPattern
|
||||
});
|
||||
|
||||
// Remove the existing job
|
||||
await this.queue.removeRepeatableByKey(existingJob.key);
|
||||
|
||||
// Small delay to ensure cleanup is complete
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
} else {
|
||||
this.logger.info('Creating new recurring job', { jobKey, cronPattern });
|
||||
}
|
||||
|
||||
// Add the new/updated recurring job
|
||||
const job = await this.queue.add(jobData.type, jobData, {
|
||||
repeat: {
|
||||
pattern: cronPattern,
|
||||
tz: 'UTC',
|
||||
immediately: jobData.immediately || false,
|
||||
},
|
||||
// Use a consistent jobId for this specific recurring job
|
||||
jobId: `recurring-${jobKey}`,
|
||||
removeOnComplete: 1,
|
||||
removeOnFail: 1,
|
||||
attempts: 2,
|
||||
backoff: {
|
||||
type: 'fixed',
|
||||
delay: 5000
|
||||
},
|
||||
...options
|
||||
});
|
||||
|
||||
this.logger.info('Recurring job added/updated successfully', {
|
||||
jobKey,
|
||||
type: jobData.type,
|
||||
cronPattern,
|
||||
immediately: jobData.immediately || false
|
||||
});
|
||||
|
||||
return job;
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to add/update recurring job', {
|
||||
jobData,
|
||||
cronPattern,
|
||||
error: error instanceof Error ? error.message : String(error)
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async getJobStats() {
|
||||
if (!this.isInitialized) {
|
||||
throw new Error('Queue service not initialized. Call initialize() first.');
|
||||
}
|
||||
const [waiting, active, completed, failed, delayed] = await Promise.all([
|
||||
this.queue.getWaiting(),
|
||||
this.queue.getActive(),
|
||||
this.queue.getCompleted(),
|
||||
this.queue.getFailed(),
|
||||
this.queue.getDelayed()
|
||||
]);
|
||||
|
||||
return {
|
||||
waiting: waiting.length,
|
||||
active: active.length,
|
||||
completed: completed.length,
|
||||
failed: failed.length,
|
||||
delayed: delayed.length
|
||||
};
|
||||
}
|
||||
|
||||
async drainQueue() {
|
||||
if (!this.isInitialized) {
|
||||
await this.queue.drain()
|
||||
}
|
||||
}
|
||||
|
||||
async getQueueStatus() {
|
||||
if (!this.isInitialized) {
|
||||
throw new Error('Queue service not initialized. Call initialize() first.');
|
||||
}
|
||||
const stats = await this.getJobStats();
|
||||
return {
|
||||
...stats,
|
||||
workers: this.getWorkerCount(),
|
||||
totalConcurrency: this.getTotalConcurrency(),
|
||||
queue: this.queue.name,
|
||||
connection: {
|
||||
host: process.env.DRAGONFLY_HOST || 'localhost',
|
||||
port: parseInt(process.env.DRAGONFLY_PORT || '6379')
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
getWorkerCount() {
|
||||
if (!this.isInitialized) {
|
||||
return 0;
|
||||
}
|
||||
return this.workers.length;
|
||||
}
|
||||
|
||||
getRegisteredProviders() {
|
||||
return providerRegistry.getProviders().map(({ key, config }) => ({
|
||||
key,
|
||||
name: config.name,
|
||||
service: config.service,
|
||||
operations: Object.keys(config.operations),
|
||||
scheduledJobs: config.scheduledJobs?.length || 0
|
||||
}));
|
||||
}
|
||||
|
||||
getScheduledJobsInfo() {
|
||||
return providerRegistry.getAllScheduledJobs().map(({ service, provider, job }) => ({
|
||||
id: `${service}-${provider}-${job.type}`,
|
||||
service,
|
||||
provider,
|
||||
type: job.type,
|
||||
operation: job.operation,
|
||||
cronPattern: job.cronPattern,
|
||||
priority: job.priority,
|
||||
description: job.description,
|
||||
immediately: job.immediately || false
|
||||
}));
|
||||
}
|
||||
async shutdown() {
|
||||
if (!this.isInitialized) {
|
||||
this.logger.warn('Queue service not initialized, nothing to shutdown');
|
||||
return;
|
||||
}
|
||||
this.logger.info('Shutting down queue service');
|
||||
|
||||
// Close all workers
|
||||
this.logger.info(`Closing ${this.workers.length} workers...`);
|
||||
await Promise.all(this.workers.map((worker, index) => {
|
||||
this.logger.debug(`Closing worker ${index + 1}`);
|
||||
return worker.close();
|
||||
}));
|
||||
|
||||
await this.queue.close();
|
||||
await this.queueEvents.close();
|
||||
this.isInitialized = false;
|
||||
this.logger.info('Queue service shutdown complete');
|
||||
}
|
||||
}
|
||||
|
||||
export const queueManager = new QueueService();
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue