feat: add Valkey integration for task queue (closes #98)
- Add ioredis package dependency for Redis-compatible operations - Create ValkeyModule as global NestJS module - Implement ValkeyService with task queue operations: - enqueue(task): Add tasks to FIFO queue - dequeue(): Get next task and update to PROCESSING status - getStatus(taskId): Retrieve task metadata and status - updateStatus(taskId, status): Update task state (COMPLETED/FAILED) - getQueueLength(): Monitor queue depth - clearQueue(): Queue management utility - healthCheck(): Verify Valkey connectivity - Add TaskDto, EnqueueTaskDto, UpdateTaskStatusDto interfaces - Implement TaskStatus enum (PENDING/PROCESSING/COMPLETED/FAILED) - Add comprehensive test suite with in-memory Redis mock (20 tests) - Integrate ValkeyModule into app.module.ts - Valkey Docker Compose service already configured in docker-compose.yml - VALKEY_URL environment variable already in .env.example - Add detailed README with usage examples and API documentation Technical Details: - Uses FIFO queue (RPUSH/LPOP for strict ordering) - Task metadata stored with 24-hour TTL - Lifecycle hooks for connection management (onModuleInit/onModuleDestroy) - Automatic retry with exponential backoff on connection errors - Global module - no explicit imports needed Tests verify: - Connection initialization and health checks - FIFO enqueue/dequeue behavior - Status lifecycle transitions - Concurrent task handling - Queue management operations - Complete task processing workflows
This commit is contained in:
Jason Woltje
229
apps/api/src/valkey/valkey.service.ts
Normal file
229
apps/api/src/valkey/valkey.service.ts
Normal file
@@ -0,0 +1,229 @@
|
||||
import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
|
||||
import Redis from 'ioredis';
|
||||
import { TaskDto, TaskStatus, EnqueueTaskDto, UpdateTaskStatusDto, TaskMetadata } from './dto/task.dto';
|
||||
import { randomUUID } from 'crypto';
|
||||
|
||||
/**
|
||||
* ValkeyService - Task queue service using Valkey (Redis-compatible)
|
||||
*
|
||||
* Provides task queue operations:
|
||||
* - enqueue(task): Add task to queue
|
||||
* - dequeue(): Get next task from queue
|
||||
* - getStatus(taskId): Get task status and metadata
|
||||
* - updateStatus(taskId, status): Update task status
|
||||
*/
|
||||
@Injectable()
|
||||
export class ValkeyService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly logger = new Logger(ValkeyService.name);
|
||||
private client: Redis;
|
||||
private readonly QUEUE_KEY = 'mosaic:task:queue';
|
||||
private readonly TASK_PREFIX = 'mosaic:task:';
|
||||
private readonly TASK_TTL = 86400; // 24 hours in seconds
|
||||
|
||||
async onModuleInit() {
|
||||
const valkeyUrl = process.env.VALKEY_URL || 'redis://localhost:6379';
|
||||
|
||||
this.logger.log(`Connecting to Valkey at ${valkeyUrl}`);
|
||||
|
||||
this.client = new Redis(valkeyUrl, {
|
||||
maxRetriesPerRequest: 3,
|
||||
retryStrategy: (times) => {
|
||||
const delay = Math.min(times * 50, 2000);
|
||||
this.logger.warn(`Valkey connection retry attempt ${times}, waiting ${delay}ms`);
|
||||
return delay;
|
||||
},
|
||||
reconnectOnError: (err) => {
|
||||
this.logger.error('Valkey connection error:', err.message);
|
||||
return true;
|
||||
},
|
||||
});
|
||||
|
||||
this.client.on('connect', () => {
|
||||
this.logger.log('Valkey connected successfully');
|
||||
});
|
||||
|
||||
this.client.on('error', (err) => {
|
||||
this.logger.error('Valkey client error:', err.message);
|
||||
});
|
||||
|
||||
this.client.on('close', () => {
|
||||
this.logger.warn('Valkey connection closed');
|
||||
});
|
||||
|
||||
// Wait for connection
|
||||
try {
|
||||
await this.client.ping();
|
||||
this.logger.log('Valkey health check passed');
|
||||
} catch (error) {
|
||||
this.logger.error('Valkey health check failed:', error.message);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
this.logger.log('Disconnecting from Valkey');
|
||||
await this.client.quit();
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a task to the queue
|
||||
* @param task - Task to enqueue
|
||||
* @returns The created task with ID and metadata
|
||||
*/
|
||||
async enqueue(task: EnqueueTaskDto): Promise<TaskDto> {
|
||||
const taskId = randomUUID();
|
||||
const now = new Date();
|
||||
|
||||
const taskData: TaskDto = {
|
||||
id: taskId,
|
||||
type: task.type,
|
||||
data: task.data,
|
||||
status: TaskStatus.PENDING,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
};
|
||||
|
||||
// Store task metadata
|
||||
const taskKey = this.getTaskKey(taskId);
|
||||
await this.client.setex(
|
||||
taskKey,
|
||||
this.TASK_TTL,
|
||||
JSON.stringify(taskData)
|
||||
);
|
||||
|
||||
// Add to queue (RPUSH = add to tail, LPOP = remove from head => FIFO)
|
||||
await this.client.rpush(this.QUEUE_KEY, taskId);
|
||||
|
||||
this.logger.log(`Task enqueued: ${taskId} (type: ${task.type})`);
|
||||
return taskData;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the next task from the queue
|
||||
* @returns The next task or null if queue is empty
|
||||
*/
|
||||
async dequeue(): Promise<TaskDto | null> {
|
||||
// LPOP = remove from head (FIFO)
|
||||
const taskId = await this.client.lpop(this.QUEUE_KEY);
|
||||
|
||||
if (!taskId) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const task = await this.getStatus(taskId);
|
||||
|
||||
if (!task) {
|
||||
this.logger.warn(`Task ${taskId} not found in metadata store`);
|
||||
return null;
|
||||
}
|
||||
|
||||
// Update status to processing and return the updated task
|
||||
const updatedTask = await this.updateStatus(taskId, {
|
||||
status: TaskStatus.PROCESSING,
|
||||
});
|
||||
|
||||
this.logger.log(`Task dequeued: ${taskId} (type: ${task.type})`);
|
||||
return updatedTask;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get task status and metadata
|
||||
* @param taskId - Task ID
|
||||
* @returns Task data or null if not found
|
||||
*/
|
||||
async getStatus(taskId: string): Promise<TaskDto | null> {
|
||||
const taskKey = this.getTaskKey(taskId);
|
||||
const taskData = await this.client.get(taskKey);
|
||||
|
||||
if (!taskData) {
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
return JSON.parse(taskData) as TaskDto;
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to parse task data for ${taskId}:`, error.message);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Update task status and metadata
|
||||
* @param taskId - Task ID
|
||||
* @param update - Status update data
|
||||
* @returns Updated task or null if not found
|
||||
*/
|
||||
async updateStatus(taskId: string, update: UpdateTaskStatusDto): Promise<TaskDto | null> {
|
||||
const task = await this.getStatus(taskId);
|
||||
|
||||
if (!task) {
|
||||
this.logger.warn(`Cannot update status for non-existent task: ${taskId}`);
|
||||
return null;
|
||||
}
|
||||
|
||||
const now = new Date();
|
||||
const updatedTask: TaskDto = {
|
||||
...task,
|
||||
status: update.status,
|
||||
updatedAt: now,
|
||||
};
|
||||
|
||||
if (update.error) {
|
||||
updatedTask.error = update.error;
|
||||
}
|
||||
|
||||
if (update.status === TaskStatus.COMPLETED || update.status === TaskStatus.FAILED) {
|
||||
updatedTask.completedAt = now;
|
||||
}
|
||||
|
||||
if (update.result) {
|
||||
updatedTask.data = { ...task.data, ...update.result };
|
||||
}
|
||||
|
||||
const taskKey = this.getTaskKey(taskId);
|
||||
await this.client.setex(
|
||||
taskKey,
|
||||
this.TASK_TTL,
|
||||
JSON.stringify(updatedTask)
|
||||
);
|
||||
|
||||
this.logger.log(`Task status updated: ${taskId} => ${update.status}`);
|
||||
return updatedTask;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get queue length
|
||||
* @returns Number of tasks in queue
|
||||
*/
|
||||
async getQueueLength(): Promise<number> {
|
||||
return await this.client.llen(this.QUEUE_KEY);
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all tasks from queue (use with caution!)
|
||||
*/
|
||||
async clearQueue(): Promise<void> {
|
||||
await this.client.del(this.QUEUE_KEY);
|
||||
this.logger.warn('Queue cleared');
|
||||
}
|
||||
|
||||
/**
|
||||
* Get task key for Redis storage
|
||||
*/
|
||||
private getTaskKey(taskId: string): string {
|
||||
return `${this.TASK_PREFIX}${taskId}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Health check - ping Valkey
|
||||
*/
|
||||
async healthCheck(): Promise<boolean> {
|
||||
try {
|
||||
const result = await this.client.ping();
|
||||
return result === 'PONG';
|
||||
} catch (error) {
|
||||
this.logger.error('Valkey health check failed:', error.message);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user