added new queue lib with batch processor and provider
This commit is contained in:
parent
ddcf94a587
commit
6c548416d1
19 changed files with 1939 additions and 35 deletions
345
libs/queue/src/batch-processor.ts
Normal file
345
libs/queue/src/batch-processor.ts
Normal file
|
|
@ -0,0 +1,345 @@
|
|||
import { CacheProvider, createCache } from '@stock-bot/cache';
|
||||
import { getLogger } from '@stock-bot/logger';
|
||||
import type { QueueManager } from './queue-manager';
|
||||
import type { BatchJobData, BatchResult, JobData, ProcessOptions } from './types';
|
||||
|
||||
const logger = getLogger('batch-processor');
|
||||
|
||||
const cacheProviders = new Map<string, CacheProvider>();
|
||||
|
||||
function getCache(queueName: string): CacheProvider {
|
||||
if (!cacheProviders.has(queueName)) {
|
||||
const cacheProvider = createCache({
|
||||
keyPrefix: `batch:${queueName}:`,
|
||||
ttl: 86400, // 24 hours default
|
||||
enableMetrics: true,
|
||||
});
|
||||
cacheProviders.set(queueName, cacheProvider);
|
||||
}
|
||||
return cacheProviders.get(queueName) as CacheProvider;
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the batch cache before any batch operations
|
||||
* This should be called during application startup
|
||||
*/
|
||||
export async function initializeBatchCache(queueManager: QueueManager): Promise<void> {
|
||||
const queueName = queueManager.getQueueName();
|
||||
logger.info('Initializing batch cache...', { queueName });
|
||||
|
||||
const cache = getCache(queueName);
|
||||
await cache.waitForReady(10000);
|
||||
logger.info('Batch cache initialized successfully', { queueName });
|
||||
}
|
||||
|
||||
/**
|
||||
* Main function - processes items either directly or in batches
|
||||
* Each item becomes payload: item (no processing needed)
|
||||
*/
|
||||
export async function processItems<T>(
|
||||
items: T[],
|
||||
queue: QueueManager,
|
||||
options: ProcessOptions
|
||||
): Promise<BatchResult> {
|
||||
const startTime = Date.now();
|
||||
|
||||
if (items.length === 0) {
|
||||
return {
|
||||
jobsCreated: 0,
|
||||
mode: 'direct',
|
||||
totalItems: 0,
|
||||
duration: 0,
|
||||
};
|
||||
}
|
||||
|
||||
logger.info('Starting batch processing', {
|
||||
totalItems: items.length,
|
||||
mode: options.useBatching ? 'batch' : 'direct',
|
||||
batchSize: options.batchSize,
|
||||
totalDelayMs: options.totalDelayMs,
|
||||
});
|
||||
|
||||
try {
|
||||
const result = options.useBatching
|
||||
? await processBatched(items, queue, options)
|
||||
: await processDirect(items, queue, options);
|
||||
|
||||
const duration = Date.now() - startTime;
|
||||
|
||||
logger.info('Batch processing completed', {
|
||||
...result,
|
||||
duration: `${(duration / 1000).toFixed(1)}s`,
|
||||
});
|
||||
|
||||
return { ...result, duration };
|
||||
} catch (error) {
|
||||
logger.error('Batch processing failed', error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process items directly - each item becomes a separate job
|
||||
*/
|
||||
async function processDirect<T>(
|
||||
items: T[],
|
||||
queue: QueueManager,
|
||||
options: ProcessOptions
|
||||
): Promise<Omit<BatchResult, 'duration'>> {
|
||||
const delayPerItem = options.totalDelayMs / items.length;
|
||||
|
||||
logger.info('Creating direct jobs', {
|
||||
totalItems: items.length,
|
||||
delayPerItem: `${(delayPerItem / 1000).toFixed(1)}s`,
|
||||
});
|
||||
|
||||
const jobs = items.map((item, index) => ({
|
||||
name: 'process-item',
|
||||
data: {
|
||||
type: 'process-item',
|
||||
provider: options.provider || 'generic',
|
||||
operation: options.operation || 'process-item',
|
||||
payload: item, // Just the item directly - no wrapper!
|
||||
priority: options.priority || undefined,
|
||||
},
|
||||
opts: {
|
||||
delay: index * delayPerItem,
|
||||
priority: options.priority || undefined,
|
||||
attempts: options.retries || 3,
|
||||
removeOnComplete: options.removeOnComplete || 10,
|
||||
removeOnFail: options.removeOnFail || 5,
|
||||
},
|
||||
}));
|
||||
|
||||
const createdJobs = await addJobsInChunks(queue, jobs);
|
||||
|
||||
return {
|
||||
totalItems: items.length,
|
||||
jobsCreated: createdJobs.length,
|
||||
mode: 'direct',
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Process items in batches - store items directly
|
||||
*/
|
||||
async function processBatched<T>(
|
||||
items: T[],
|
||||
queue: QueueManager,
|
||||
options: ProcessOptions
|
||||
): Promise<Omit<BatchResult, 'duration'>> {
|
||||
const batchSize = options.batchSize || 100;
|
||||
const batches = createBatches(items, batchSize);
|
||||
const delayPerBatch = options.totalDelayMs / batches.length;
|
||||
|
||||
logger.info('Creating batch jobs', {
|
||||
totalItems: items.length,
|
||||
batchSize,
|
||||
totalBatches: batches.length,
|
||||
delayPerBatch: `${(delayPerBatch / 1000 / 60).toFixed(2)} minutes`,
|
||||
});
|
||||
|
||||
const batchJobs = await Promise.all(
|
||||
batches.map(async (batch, batchIndex) => {
|
||||
// Just store the items directly - no processing needed
|
||||
const payloadKey = await storeItems(batch, queue, options);
|
||||
|
||||
return {
|
||||
name: 'process-batch',
|
||||
data: {
|
||||
type: 'process-batch',
|
||||
provider: options.provider || 'generic',
|
||||
operation: 'process-batch-items',
|
||||
payload: {
|
||||
payloadKey,
|
||||
batchIndex,
|
||||
totalBatches: batches.length,
|
||||
itemCount: batch.length,
|
||||
} as BatchJobData,
|
||||
priority: options.priority || undefined,
|
||||
},
|
||||
opts: {
|
||||
delay: batchIndex * delayPerBatch,
|
||||
priority: options.priority || undefined,
|
||||
attempts: options.retries || 3,
|
||||
removeOnComplete: options.removeOnComplete || 10,
|
||||
removeOnFail: options.removeOnFail || 5,
|
||||
},
|
||||
};
|
||||
})
|
||||
);
|
||||
|
||||
const createdJobs = await addJobsInChunks(queue, batchJobs);
|
||||
|
||||
return {
|
||||
totalItems: items.length,
|
||||
jobsCreated: createdJobs.length,
|
||||
batchesCreated: batches.length,
|
||||
mode: 'batch',
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a batch job - loads items and creates individual jobs
|
||||
*/
|
||||
export async function processBatchJob(
|
||||
jobData: BatchJobData,
|
||||
queue: QueueManager
|
||||
): Promise<unknown> {
|
||||
const { payloadKey, batchIndex, totalBatches, itemCount } = jobData;
|
||||
|
||||
logger.debug('Processing batch job', {
|
||||
batchIndex,
|
||||
totalBatches,
|
||||
itemCount,
|
||||
});
|
||||
|
||||
try {
|
||||
const payload = await loadPayload(payloadKey, queue);
|
||||
if (!payload || !payload.items || !payload.options) {
|
||||
logger.error('Invalid payload data', { payloadKey, payload });
|
||||
throw new Error(`Invalid payload data for key: ${payloadKey}`);
|
||||
}
|
||||
|
||||
const { items, options } = payload;
|
||||
|
||||
// Create jobs directly from items - each item becomes payload: item
|
||||
const jobs = items.map((item: unknown, index: number) => ({
|
||||
name: 'process-item',
|
||||
data: {
|
||||
type: 'process-item',
|
||||
provider: options.provider || 'generic',
|
||||
operation: options.operation || 'generic',
|
||||
payload: item, // Just the item directly!
|
||||
priority: options.priority || undefined,
|
||||
},
|
||||
opts: {
|
||||
delay: index * (options.delayPerItem || 1000),
|
||||
priority: options.priority || undefined,
|
||||
attempts: options.retries || 3,
|
||||
},
|
||||
}));
|
||||
|
||||
const createdJobs = await addJobsInChunks(queue, jobs);
|
||||
|
||||
// Cleanup payload after successful processing
|
||||
await cleanupPayload(payloadKey, queue);
|
||||
|
||||
return {
|
||||
batchIndex,
|
||||
itemsProcessed: items.length,
|
||||
jobsCreated: createdJobs.length,
|
||||
};
|
||||
} catch (error) {
|
||||
logger.error('Batch job processing failed', { batchIndex, error });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// Helper functions
|
||||
|
||||
function createBatches<T>(items: T[], batchSize: number): T[][] {
|
||||
const batches: T[][] = [];
|
||||
for (let i = 0; i < items.length; i += batchSize) {
|
||||
batches.push(items.slice(i, i + batchSize));
|
||||
}
|
||||
return batches;
|
||||
}
|
||||
|
||||
async function storeItems<T>(
|
||||
items: T[],
|
||||
queue: QueueManager,
|
||||
options: ProcessOptions
|
||||
): Promise<string> {
|
||||
if (!queue) {
|
||||
throw new Error('Batch cache not initialized. Call initializeBatchCache() first.');
|
||||
}
|
||||
|
||||
const cache = getCache(queue.getQueueName());
|
||||
const payloadKey = `payload:${Date.now()}:${Math.random().toString(36).substr(2, 9)}`;
|
||||
|
||||
const payload = {
|
||||
items, // Just store the items directly
|
||||
options: {
|
||||
delayPerItem: 1000,
|
||||
priority: options.priority || undefined,
|
||||
retries: options.retries || 3,
|
||||
provider: options.provider || 'generic',
|
||||
operation: options.operation || 'generic',
|
||||
},
|
||||
createdAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const ttlSeconds = options.ttl || 86400; // 24 hours default
|
||||
await cache.set(payloadKey, payload, ttlSeconds);
|
||||
|
||||
return payloadKey;
|
||||
}
|
||||
|
||||
async function loadPayload<T>(
|
||||
key: string,
|
||||
queue: QueueManager
|
||||
): Promise<{
|
||||
items: T[];
|
||||
options: {
|
||||
delayPerItem: number;
|
||||
priority?: number;
|
||||
retries: number;
|
||||
provider: string;
|
||||
operation: string;
|
||||
};
|
||||
} | null> {
|
||||
if (!queue) {
|
||||
throw new Error('Batch cache not initialized. Call initializeBatchCache() first.');
|
||||
}
|
||||
|
||||
const cache = getCache(queue.getQueueName());
|
||||
return (await cache.get(key)) as {
|
||||
items: T[];
|
||||
options: {
|
||||
delayPerItem: number;
|
||||
priority?: number;
|
||||
retries: number;
|
||||
provider: string;
|
||||
operation: string;
|
||||
};
|
||||
} | null;
|
||||
}
|
||||
|
||||
async function cleanupPayload(key: string, queue: QueueManager): Promise<void> {
|
||||
if (!queue) {
|
||||
throw new Error('Batch cache not initialized. Call initializeBatchCache() first.');
|
||||
}
|
||||
|
||||
const cache = getCache(queue.getQueueName());
|
||||
await cache.del(key);
|
||||
}
|
||||
|
||||
async function addJobsInChunks(
|
||||
queue: QueueManager,
|
||||
jobs: Array<{ name: string; data: JobData; opts?: Record<string, unknown> }>,
|
||||
chunkSize = 100
|
||||
): Promise<unknown[]> {
|
||||
const allCreatedJobs = [];
|
||||
|
||||
for (let i = 0; i < jobs.length; i += chunkSize) {
|
||||
const chunk = jobs.slice(i, i + chunkSize);
|
||||
try {
|
||||
const createdJobs = await queue.addBulk(chunk);
|
||||
allCreatedJobs.push(...createdJobs);
|
||||
|
||||
// Small delay between chunks to avoid overwhelming Redis
|
||||
if (i + chunkSize < jobs.length) {
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Failed to add job chunk', {
|
||||
startIndex: i,
|
||||
chunkSize: chunk.length,
|
||||
error,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return allCreatedJobs;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue