Spaces:
Running
Running
File size: 5,564 Bytes
e8a6607 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 | // src/redis/client.ts
import IORedis from 'ioredis';
import { Queue, Job } from 'bullmq';
import dotenv from 'dotenv';
import logger from '../logger';
const envFile: string = process.env.NODE_ENV === 'production' ? '.env.production' : '.env.development';
dotenv.config({ path: envFile });
export class RedisClientInfra {
public readonly redis: IORedis;
// public readonly apiUserActionsQueue: Queue;
private initialized: boolean;
private redisName: string;
constructor(redisName: string) {
this.redis = new IORedis({
host: process.env.REDIS_HOST1 || 'localhost',
port: parseInt(process.env.REDIS_PORT || '6379'),
// Connection options
maxRetriesPerRequest: null,
enableReadyCheck: false,
enableOfflineQueue: true,
// Socket options (CRITICAL for preventing ECONNABORTED)
retryStrategy: (times: number) => {
logger.info(`Redis connection attempt(${redisName}) ${times}`);
if (times > 10) {
logger.info(`Too many redis connection attempts(${redisName})`);
return null;
}
const delay = Math.min(times * 50, 2000);
return delay;
},
// Timeout settings
connectTimeout: 10000, // 10 seconds to connect
commandTimeout: 30000, // 30 seconds for commands
reconnectOnError: (err: any) => {
// Reconnect on network errors but not on command errors
const targetErrors = ['READONLY', 'ECONNRESET', 'ETIMEDOUT'];
return targetErrors.some(error => err.message.includes(error));
},
});
this.initialized = false;
this.redisName = redisName;
/*
// Create a new queue for API service
this.apiUserActionsQueue = new Queue('api-user-actions-queue', {
connection: this.apiRedis, // Pass connection object
defaultJobOptions: {
attempts: 3,
backoff: {
type: 'exponential', // Options: 'fixed' | 'exponential
delay: 3000 // Delay in ms
},
removeOnComplete: {
count: 1000, // Keep last 1000 completed jobs
age: 24 * 3600 // OR keep for 24 hours (optional)
},
removeOnFail: {
count: 6000, // Keep last 5000 failed jobs
age: 72 * 3600 // OR keep for 72 hours (optional)
},
}
}); */
}
async init() {
if (this.initialized) return;
this.initialized = true;
this.setupRedisEventListeners();
// this.setupUserActionsQueueEventListeners();
}
// Handle connection events
private setupRedisEventListeners(): void {
this.redis.on('connect', () => {
logger.info(`β
Redis Connected (${this.redisName})`);
});
this.redis.on('ready', () => {
logger.info(`π’ Redis ready (${this.redisName})`);
});
this.redis.on('error', (err: any) => {
logger.error({ err }, `β Redis connection error (${this.redisName}): ${err.message}`);
});
this.redis.on('close', () => {
logger.info(`β οΈ Redis connection closed (${this.redisName})`);
});
this.redis.on('reconnecting', (delay: number) => {
logger.info(`π Redis reconnecting in ${delay}ms (${this.redisName})`);
});
this.redis.on('end', () => {
logger.info(`π΄ Redis connection ended (${this.redisName})`);
});
}
/*private setupApiUserActionsQueueEventListeners(): void {
// The callback receives a Job object directly, not an args object
this.apiUserActionsQueue.on('waiting', (job: Job) => {
logger.info(`π₯ Job ${job.id} is waiting`);
logger.info(`Job data: ${job.data}`);
});
this.apiUserActionsQueue.on('error', (err: Error) => {
logger.error({ err }, `β Queue error: ${err.message}`);
});
this.apiUserActionsQueue.on('progress', (job: string, progress: any) => {
logger.debug(`π Job ${job} progress: ${progress}`);
});
// Other available events:
this.apiUserActionsQueue.on('paused', () => {
logger.info('βΈοΈ Queue paused');
});
this.apiUserActionsQueue.on('resumed', () => {
logger.info('βΆοΈ Queue resumed');
});
this.apiUserActionsQueue.on('cleaned', (jobs: string[], type: string) => {
logger.debug(`π§Ή Cleaned ${jobs.length} ${type} jobs`);
});
this.apiUserActionsQueue.on('removed', (job: string) => {
logger.debug(`ποΈ Job ${job} removed`);
});
// Redis-specific events
this.apiUserActionsQueue.on('ioredis:close', () => {
logger.info('π Redis connection closed(API redis)');
});
} */
// Test connection on startup
public async testRedisConnection() {
try {
await this.redis.ping();
logger.info(`β
Redis connection test passed(${this.redisName})`);
} catch (err: any) {
logger.error({err}, `β Redis connection test failed(${this.redisName}): ${err.message}`);
}
}
async redisGracefulShutdown(): Promise<void> {
logger.info(`β³ Closing Redis connection for ${this.redisName}...`);
if(!this.redis) {
logger.info(`β
Redis connection for ${this.redisName} is not active`);
return;
}
try {
await this.redis.quit(); // Graceful Redis shutdown
logger.info(`β
Redis connection closed for ${this.redisName}`);
} catch (err: any) {
logger.error({ err }, `β Failed to close redis connection for ${this.redisName}`);
throw err; // propagate to main shutdown
}
}
}
/*
If age exceeded β delete
OR
If count exceeded β delete
*/ |