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
*/