import { config as appConfig } from '../config/index.js'; import { RedisClientType } from 'redis'; import { REDIS_PREFIX } from './index.js'; import { createLogger } from '../logging/logger.js'; import { getTimeTakenSincePoint } from './time.js'; import { getDb } from '../db/db.js'; import type { DbDriver } from '../db/driver/types.js'; import { sql } from '../db/sql.js'; import { withTimeout } from './general.js'; const logger = createLogger('cache'); const REDIS_TIMEOUT = appConfig.bootstrap.redisTimeout; // Interface that both memory and Redis cache will implement export interface CacheBackend { get(key: K, updateTTL?: boolean): Promise; set(key: K, value: V, ttl: number, forceWrite?: boolean): Promise; flush(): Promise; delete(key: K): Promise; update(key: K, value: V): Promise; clear(): Promise; getTTL(key: K): Promise; waitUntilReady(): Promise; } // Memory cache implementation export class MemoryCacheBackend implements CacheBackend { private cache: Map>; private maxSize: number; constructor(maxSize: number) { this.cache = new Map>(); this.maxSize = maxSize; } async get(key: K, updateTTL: boolean = false): Promise { const item = this.cache.get(key); if (item) { const now = Date.now(); item.lastAccessed = now; if (now - item.createdAt > item.ttl) { this.cache.delete(key); return undefined; } if (updateTTL) { item.createdAt = now; } return structuredClone(item.value); } return undefined; } async set( key: K, value: V, ttl: number, forceWrite?: boolean ): Promise { if (this.cache.size >= this.maxSize) { this.evict(); } this.cache.set( key, new CacheItem( structuredClone(value), Date.now(), Date.now(), ttl * 1000 ) ); } async update(key: K, value: V): Promise { const item = this.cache.get(key); if (item) { item.value = value; } } async delete(key: K): Promise { return this.cache.delete(key); } async clear(): Promise { this.cache.clear(); } async getTTL(key: K): Promise { const item = this.cache.get(key); if (item) { return Math.max( 0, Math.floor((item.createdAt + item.ttl - Date.now()) / 1000) ); } return 0; } private evict(): void { let oldestKey: K | undefined; let oldestTime = Infinity; for (const [key, item] of this.cache.entries()) { if (item.lastAccessed < oldestTime) { oldestTime = item.lastAccessed; oldestKey = key; } } if (oldestKey !== undefined) { this.cache.delete(oldestKey); } } getSize(): number { return this.cache.size; } getMemoryUsageEstimate(): number { let totalSize = 0; for (const item of this.cache.values()) { try { totalSize += Buffer.byteLength(JSON.stringify(item), 'utf8'); } catch (e) { // In case of circular references } } return totalSize; } async waitUntilReady(): Promise { return Promise.resolve(); } async flush(): Promise { // Memory writes are synchronous — nothing to flush } } // Redis cache implementation with timeout handling export class RedisCacheBackend implements CacheBackend { private client: RedisClientType; private prefix: string; private maxSize: number; private timeout: number; private static writeBuffer: Map = new Map(); private static flushInterval: NodeJS.Timeout | null = null; private static isFlushing: boolean = false; private static batchSize: number = 100; private static flushIntervalTime: number = 2000; private static clientRef: RedisClientType | null = null; private static timeoutRef: number = REDIS_TIMEOUT; constructor( redisClient: RedisClientType, prefix: string = REDIS_PREFIX, maxSize: number = appConfig.resources.cache.defaultMaxSize, timeout: number = REDIS_TIMEOUT ) { this.client = redisClient; this.prefix = prefix; this.maxSize = maxSize; this.timeout = timeout; // Store client reference for static operations RedisCacheBackend.clientRef = redisClient; RedisCacheBackend.timeoutRef = timeout; RedisCacheBackend.startFlushInterval(); } private getKey(key: K): string { return `${REDIS_PREFIX}${this.prefix}${String(key)}`; } private static startFlushInterval() { if (RedisCacheBackend.flushInterval !== null) return; RedisCacheBackend.flushInterval = setInterval(() => { RedisCacheBackend.flushWriteBuffer(); }, RedisCacheBackend.flushIntervalTime); } async get(key: K, updateTTL: boolean = false): Promise { const redisKey = this.getKey(key); return withTimeout( async () => { const data = await this.client.get(redisKey); if (!data) return undefined; if (updateTTL) { // Update TTL if requested const ttl = await this.client.ttl(redisKey); if (ttl > 0) { await this.client.expire(redisKey, ttl); } } return JSON.parse(data) as V; }, undefined, { timeout: this.timeout, shouldProceed: () => this.client.isOpen, getContext: () => `getting key ${String(key)} from Redis`, } ); } async set( key: K, value: V, ttl: number, forceWrite?: boolean ): Promise { if (ttl === 0) return; const redisKey = this.getKey(key); RedisCacheBackend.writeBuffer.set(redisKey, { value: JSON.stringify(value), ttl, }); if (RedisCacheBackend.writeBuffer.size >= RedisCacheBackend.batchSize) { RedisCacheBackend.flushWriteBuffer(); } else if (forceWrite) { await RedisCacheBackend.flushWriteBuffer(); } } private static async flushWriteBuffer(): Promise { if ( RedisCacheBackend.isFlushing || RedisCacheBackend.writeBuffer.size === 0 ) return; RedisCacheBackend.isFlushing = true; const bufferToFlush = new Map(RedisCacheBackend.writeBuffer); RedisCacheBackend.writeBuffer.clear(); if (!RedisCacheBackend.clientRef) { logger.error( 'Cannot flush Redis write buffer - no client reference available' ); RedisCacheBackend.isFlushing = false; return; } const start = Date.now(); const pipeline = RedisCacheBackend.clientRef.multi(); for (const [key, item] of bufferToFlush.entries()) { pipeline.set(key, item.value, { EX: item.ttl }); } try { await withTimeout( async () => { await pipeline.exec(); }, undefined, { timeout: RedisCacheBackend.timeoutRef, shouldProceed: () => RedisCacheBackend.clientRef?.isOpen ?? false, getContext: () => 'flushing Redis write buffer', } ); logger.debug('Flushed Redis write buffer', { items: bufferToFlush.size, timeTaken: getTimeTakenSincePoint(start), }); } catch (err) { logger.error(`Error flushing Redis write buffer: ${err}`); } finally { RedisCacheBackend.isFlushing = false; } } async update(key: K, value: V): Promise { const redisKey = this.getKey(key); await withTimeout( async () => { // Get current TTL const ttl = await this.client.ttl(redisKey); if (ttl <= 0) return false; // Key doesn't exist or has no TTL // Update value but keep the same TTL await this.client.set(redisKey, JSON.stringify(value), { EX: ttl, }); return true; }, false, { timeout: this.timeout, shouldProceed: () => this.client.isOpen, getContext: () => `updating key ${String(key)} in Redis`, } ); } async delete(key: K): Promise { const redisKey = this.getKey(key); return withTimeout( async () => { const result = await this.client.del(redisKey); return result > 0; }, false, { timeout: this.timeout, shouldProceed: () => this.client.isOpen, getContext: () => `deleting key ${String(key)} from Redis`, } ); } async clear(): Promise { await withTimeout( async () => { // Delete all keys with this cache's prefix. Must include the global // `REDIS_PREFIX` because `getKey()` writes both, otherwise nothing // matches when REDIS_PREFIX is non-empty (the previous bug). // // SCAN instead of KEYS so a large keyspace doesn't block the Redis // event loop for the duration of the wipe — Redis serves other // commands between iterations. const pattern = `${REDIS_PREFIX}${this.prefix}*`; const batch: string[] = []; for await (const key of this.client.scanIterator({ MATCH: pattern, COUNT: 500, })) { // node-redis v4 yields one key at a time; v5 yields chunks. Both // are handled by flattening Array.isArray to support either. if (Array.isArray(key)) batch.push(...key); else batch.push(key); if (batch.length >= 1000) { await this.client.del(batch.splice(0, batch.length)); } } if (batch.length > 0) { await this.client.del(batch); } return true; }, false, { timeout: this.timeout, shouldProceed: () => this.client.isOpen, getContext: () => 'clearing Redis cache', } ); } async getTTL(key: K): Promise { return withTimeout( async () => { const ttl = await this.client.ttl(this.getKey(key)); return ttl > 0 ? ttl : 0; }, 0, { timeout: this.timeout, shouldProceed: () => this.client.isOpen, getContext: () => `getting TTL for key ${String(key)} from Redis`, } ); } async waitUntilReady(): Promise { while (!this.client.isOpen) { await new Promise((resolve) => setTimeout(resolve, 100)); } } async flush(): Promise { await RedisCacheBackend.flushWriteBuffer(); } } // SQL cache implementation export class SQLCacheBackend implements CacheBackend { private prefix: string; static maintenanceStarted: boolean = false; private static writeBuffer: Map = new Map(); private static flushInterval: NodeJS.Timeout | null = null; private static isFlushing: boolean = false; private static batchSize: number = 100; private static flushIntervalTime: number = 2000; constructor(prefix: string = '', _: number) { this.prefix = prefix; this.startMaintenance(); SQLCacheBackend.startFlushInterval(); } private get db(): DbDriver { return getDb(); } private static startFlushInterval() { if (SQLCacheBackend.flushInterval !== null) return; SQLCacheBackend.flushInterval = setInterval(() => { SQLCacheBackend.flushWriteBuffer(); }, SQLCacheBackend.flushIntervalTime); } private static async flushWriteBuffer() { if (SQLCacheBackend.isFlushing || SQLCacheBackend.writeBuffer.size === 0) return; SQLCacheBackend.isFlushing = true; const bufferToFlush = new Map(SQLCacheBackend.writeBuffer); SQLCacheBackend.writeBuffer.clear(); let db: DbDriver; try { db = getDb(); } catch (err) { // DB not yet ready — put items back and bail. for (const [key, value] of bufferToFlush.entries()) { SQLCacheBackend.writeBuffer.set(key, value); } SQLCacheBackend.isFlushing = false; return; } const start = Date.now(); try { let currentSize = await db.count( sql`SELECT COUNT(*) AS count FROM cache` ); let overflow = currentSize + bufferToFlush.size - appConfig.resources.cache.sqlMaxSize; if (overflow > 0) { const removed = await SQLCacheBackend.flushStaleEntries(db); logger.debug( `Removed ${removed} stale entries from SQL cache during flush.` ); currentSize -= removed; overflow -= removed; } if (overflow > 0) { logger.debug(`Cache overflow detected. Evicting ${overflow} items.`); const limit = Math.ceil(overflow); // Works identically on SQLite and Postgres. await db.exec( sql`DELETE FROM cache WHERE key IN ( SELECT key FROM cache ORDER BY last_accessed ASC LIMIT ${limit} )` ); } if (bufferToFlush.size === 0) return; // Build a multi-row VALUES list and upsert. `ON CONFLICT ... DO // UPDATE` with `EXCLUDED` works identically on SQLite (3.24+) and // Postgres, so one query handles both dialects. const values: unknown[] = []; const placeholders: string[] = []; const now = Date.now(); for (const [key, item] of bufferToFlush.entries()) { placeholders.push('(?, ?, ?)'); values.push(key, JSON.stringify(item.value), now + item.ttl * 1000); } const valuesClause = placeholders.join(', '); await db.exec( `INSERT INTO cache (key, value, expires_at) VALUES ${valuesClause} ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, expires_at = EXCLUDED.expires_at, last_accessed = CURRENT_TIMESTAMP`, values ); logger.debug('Flushed SQL write buffer', { items: bufferToFlush.size, timeTaken: getTimeTakenSincePoint(start), }); } catch (err) { logger.error(`Error flushing SQL cache write buffer: ${err}`); for (const [key, value] of bufferToFlush.entries()) { this.writeBuffer.set(key, value); } } finally { this.isFlushing = false; } } private static async flushStaleEntries(db: DbDriver): Promise { try { const result = await db.exec( sql`DELETE FROM cache WHERE expires_at < ${Date.now()}` ); return result.rowCount; } catch { return 0; } } private startMaintenance() { if (SQLCacheBackend.maintenanceStarted) return; logger.debug('Starting SQL cache maintenance'); SQLCacheBackend.maintenanceStarted = true; setInterval( () => { try { const db = getDb(); SQLCacheBackend.flushStaleEntries(db) .then((removed) => logger.debug(`${removed} stale entries removed from SQL cache`) ) .catch((err) => { logger.error(`Error during SQL cache maintenance: ${err}`); }); } catch { // DB not yet initialised — skip this tick. } }, 1 * 60 * 60 * 1000 // hourly ); } private getKey(key: K): string { return `${this.prefix}${String(key)}`; } async get(key: K, updateTTL: boolean = false): Promise { const sqlKey = this.getKey(key); const now = Date.now(); try { const row = await this.db.maybeOne<{ value: string; expires_at: number | string; }>(sql`SELECT value, expires_at FROM cache WHERE key = ${sqlKey}`); if (!row) return undefined; const expiresAt = Number(row.expires_at); if (now > expiresAt) { await this.db.exec(sql`DELETE FROM cache WHERE key = ${sqlKey}`); return undefined; } if (updateTTL) { const ttl = Math.max(0, expiresAt - now); await this.db.exec( sql`UPDATE cache SET expires_at = ${now + ttl}, last_accessed = CURRENT_TIMESTAMP WHERE key = ${sqlKey}` ); } else { await this.db.exec( sql`UPDATE cache SET last_accessed = CURRENT_TIMESTAMP WHERE key = ${sqlKey}` ); } return JSON.parse(row.value) as V; } catch (err) { logger.error(`Error getting key ${String(key)} from SQL cache: ${err}`); return undefined; } } async set( key: K, value: V, ttl: number, forceWrite?: boolean ): Promise { if (ttl === 0) return; const sqlKey = this.getKey(key); SQLCacheBackend.writeBuffer.set(sqlKey, { value: structuredClone(value), ttl, }); if (SQLCacheBackend.writeBuffer.size >= SQLCacheBackend.batchSize) { SQLCacheBackend.flushWriteBuffer(); } else if (forceWrite) { await SQLCacheBackend.flushWriteBuffer(); } } async update(key: K, value: V): Promise { const sqlKey = this.getKey(key); try { const row = await this.db.maybeOne<{ expires_at: number | string }>( sql`SELECT expires_at FROM cache WHERE key = ${sqlKey}` ); if (!row) return; if (Date.now() > Number(row.expires_at)) { await this.db.exec(sql`DELETE FROM cache WHERE key = ${sqlKey}`); return; } await this.db.exec( sql`UPDATE cache SET value = ${JSON.stringify(value)}, last_accessed = CURRENT_TIMESTAMP WHERE key = ${sqlKey}` ); } catch (err) { logger.error(`Error updating key ${String(key)} in SQL cache: ${err}`); } } async delete(key: K): Promise { const sqlKey = this.getKey(key); try { const result = await this.db.exec( sql`DELETE FROM cache WHERE key = ${sqlKey}` ); return result.rowCount > 0; } catch (err) { logger.error(`Error deleting key ${String(key)} from SQL cache: ${err}`); return false; } } async clear(): Promise { try { if (this.prefix) { await this.db.exec( sql`DELETE FROM cache WHERE key LIKE ${`${this.prefix}%`}` ); } else { await this.db.exec(sql`DELETE FROM cache`); } } catch (err) { logger.error(`Error clearing SQL cache: ${err}`); } } async getTTL(key: K): Promise { const sqlKey = this.getKey(key); const now = Date.now(); try { const row = await this.db.maybeOne<{ expires_at: number | string }>( sql`SELECT expires_at FROM cache WHERE key = ${sqlKey}` ); if (!row) return 0; return Math.max(0, Math.floor((Number(row.expires_at) - now) / 1000)); } catch (err) { logger.error( `Error getting TTL for key ${String(key)} from SQL cache: ${err}` ); return 0; } } async waitUntilReady(): Promise { // getDb() throws if not initialised. Calling it here turns that // into an early failure for any caller that awaits readiness. getDb(); } async flush(): Promise { await SQLCacheBackend.flushWriteBuffer(); } } class CacheItem { constructor( public value: T, public lastAccessed: number, public createdAt: number, public ttl: number // Time-To-Live in milliseconds ) {} }