← back to Watches
database/cache/RedisCache.js
310 lines
/**
* Redis Caching Layer
* Implements intelligent caching with TTL, cache invalidation, and warming
*/
import Redis from 'ioredis';
import dotenv from 'dotenv';
dotenv.config();
class RedisCache {
constructor() {
// Redis configuration
this.client = new Redis({
host: process.env.REDIS_HOST || 'localhost',
port: parseInt(process.env.REDIS_PORT || '6379'),
password: process.env.REDIS_PASSWORD || undefined,
db: parseInt(process.env.REDIS_DB || '0'),
retryStrategy: (times) => {
const delay = Math.min(times * 50, 2000);
return delay;
},
maxRetriesPerRequest: 3,
enableReadyCheck: true,
lazyConnect: false
});
// Default TTL values (in seconds)
this.TTL = {
SHORT: 60, // 1 minute
MEDIUM: 300, // 5 minutes
LONG: 1800, // 30 minutes
VERY_LONG: 3600, // 1 hour
DAY: 86400 // 24 hours
};
// Event handlers
this.client.on('connect', () => {
console.log('✅ Redis connected');
});
this.client.on('ready', () => {
console.log('✅ Redis ready');
});
this.client.on('error', (err) => {
console.error('❌ Redis error:', err);
});
this.client.on('close', () => {
console.log('⚠️ Redis connection closed');
});
this.client.on('reconnecting', () => {
console.log('🔄 Redis reconnecting...');
});
}
/**
* Generate cache key
*/
generateKey(prefix, identifier) {
return `omega:${prefix}:${identifier}`;
}
/**
* Get cached value
*/
async get(key) {
try {
const value = await this.client.get(key);
if (value) {
return JSON.parse(value);
}
return null;
} catch (error) {
console.error(`Cache get error for key ${key}:`, error);
return null;
}
}
/**
* Set cached value with TTL
*/
async set(key, value, ttl = this.TTL.MEDIUM) {
try {
const serialized = JSON.stringify(value);
await this.client.setex(key, ttl, serialized);
return true;
} catch (error) {
console.error(`Cache set error for key ${key}:`, error);
return false;
}
}
/**
* Delete cached value
*/
async delete(key) {
try {
await this.client.del(key);
return true;
} catch (error) {
console.error(`Cache delete error for key ${key}:`, error);
return false;
}
}
/**
* Delete multiple keys by pattern
*/
async deletePattern(pattern) {
try {
const keys = await this.client.keys(pattern);
if (keys.length > 0) {
await this.client.del(...keys);
}
return keys.length;
} catch (error) {
console.error(`Cache delete pattern error for ${pattern}:`, error);
return 0;
}
}
/**
* Cache-aside pattern: Get or compute
*/
async getOrSet(key, computeFn, ttl = this.TTL.MEDIUM) {
try {
// Try to get from cache
const cached = await this.get(key);
if (cached !== null) {
return { value: cached, fromCache: true };
}
// Compute value
const computed = await computeFn();
// Store in cache
await this.set(key, computed, ttl);
return { value: computed, fromCache: false };
} catch (error) {
console.error(`Cache getOrSet error for key ${key}:`, error);
// Fallback to computing without caching
const computed = await computeFn();
return { value: computed, fromCache: false };
}
}
/**
* Increment counter (for rate limiting, view counts, etc.)
*/
async increment(key, ttl = null) {
try {
const newValue = await this.client.incr(key);
if (ttl && newValue === 1) {
await this.client.expire(key, ttl);
}
return newValue;
} catch (error) {
console.error(`Cache increment error for key ${key}:`, error);
return null;
}
}
/**
* Get multiple keys at once
*/
async mget(keys) {
try {
const values = await this.client.mget(keys);
return values.map(v => v ? JSON.parse(v) : null);
} catch (error) {
console.error('Cache mget error:', error);
return Array(keys.length).fill(null);
}
}
/**
* Set multiple keys at once
*/
async mset(keyValuePairs, ttl = this.TTL.MEDIUM) {
try {
const pipeline = this.client.pipeline();
for (const [key, value] of Object.entries(keyValuePairs)) {
const serialized = JSON.stringify(value);
pipeline.setex(key, ttl, serialized);
}
await pipeline.exec();
return true;
} catch (error) {
console.error('Cache mset error:', error);
return false;
}
}
/**
* Cache warming: Pre-populate cache with frequently accessed data
*/
async warmCache(data, prefix, ttl = this.TTL.LONG) {
try {
const pipeline = this.client.pipeline();
for (const [key, value] of Object.entries(data)) {
const cacheKey = this.generateKey(prefix, key);
const serialized = JSON.stringify(value);
pipeline.setex(cacheKey, ttl, serialized);
}
await pipeline.exec();
console.log(`✅ Warmed ${Object.keys(data).length} cache entries with prefix: ${prefix}`);
return true;
} catch (error) {
console.error('Cache warming error:', error);
return false;
}
}
/**
* Invalidate cache by watch ID
*/
async invalidateWatch(watchId) {
const patterns = [
`omega:watch:${watchId}*`,
`omega:watch-stats:${watchId}*`,
`omega:price-history:${watchId}*`
];
let totalDeleted = 0;
for (const pattern of patterns) {
const deleted = await this.deletePattern(pattern);
totalDeleted += deleted;
}
console.log(`🗑️ Invalidated ${totalDeleted} cache entries for watch ${watchId}`);
return totalDeleted;
}
/**
* Invalidate all watch-related cache
*/
async invalidateAllWatches() {
const deleted = await this.deletePattern('omega:watch*');
console.log(`🗑️ Invalidated ${deleted} watch cache entries`);
return deleted;
}
/**
* Get cache statistics
*/
async getStats() {
try {
const info = await this.client.info('stats');
const keyspace = await this.client.info('keyspace');
const memory = await this.client.info('memory');
return {
info,
keyspace,
memory
};
} catch (error) {
console.error('Cache stats error:', error);
return null;
}
}
/**
* Health check
*/
async healthCheck() {
try {
const pong = await this.client.ping();
return pong === 'PONG';
} catch (error) {
return false;
}
}
/**
* Flush all cache (use with caution!)
*/
async flushAll() {
try {
await this.client.flushdb();
console.log('⚠️ Cache flushed');
return true;
} catch (error) {
console.error('Cache flush error:', error);
return false;
}
}
/**
* Close connection
*/
async close() {
await this.client.quit();
console.log('✅ Redis connection closed');
}
}
// Singleton instance
const cache = new RedisCache();
export default cache;