From bc9dd5c85902872a9e8b8ca5647bcff341304819 Mon Sep 17 00:00:00 2001 From: Viren070 Date: Wed, 20 Aug 2025 20:47:14 +0100 Subject: [PATCH] feat: add redis support --- .env.sample | 9 + package-lock.json | 86 ++++++ packages/core/package.json | 1 + packages/core/src/builtins/gdrive/api.ts | 4 +- .../builtins/torbox-search/debrid-service.ts | 2 +- .../builtins/torbox-search/source-handlers.ts | 16 +- packages/core/src/db/db.ts | 1 + packages/core/src/debrid/interface.ts | 18 +- packages/core/src/main.ts | 18 +- packages/core/src/metadata/imdb.ts | 7 +- packages/core/src/metadata/tmdb.ts | 6 +- packages/core/src/presets/aiCompanion.ts | 8 +- packages/core/src/presets/aiSearch.ts | 4 +- packages/core/src/presets/moreLikeThis.ts | 2 +- packages/core/src/proxy/base.ts | 4 +- packages/core/src/utils/cache-adapter.ts | 268 ++++++++++++++++++ packages/core/src/utils/cache.ts | 246 +++++++++------- packages/core/src/utils/env.ts | 4 + packages/core/src/utils/http.ts | 14 +- packages/core/src/utils/rpdb.ts | 9 +- packages/core/src/utils/startup.ts | 7 +- packages/core/src/wrapper.ts | 4 +- packages/server/src/server.ts | 9 +- 23 files changed, 600 insertions(+), 147 deletions(-) create mode 100644 packages/core/src/utils/cache-adapter.ts diff --git a/.env.sample b/.env.sample index 7442cc9e..81ce35e9 100644 --- a/.env.sample +++ b/.env.sample @@ -52,6 +52,15 @@ ADDON_PASSWORD= # (e.g., postgresql://postgres:password@localhost:5432/aiostreams) DATABASE_URI=sqlite://./data/db.sqlite +# --- Redis (Optional, for Scaled Deployments) ---- +# Optionally provide a Redis URI to use for caching. This is only recommended for +# users running multiple instances (e.g., via Docker Swarm). +# +# For standard, single-container setups, leave this blank. It will default to +# a simpler internal memory cache that is more than sufficient and faster too. +# +# REDIS_URI=redis://your_redis_hostname:6379 + # ============================================================================== # BUILT-IN ADDON CONFIGURATION diff --git a/package-lock.json b/package-lock.json index 76558436..fc306113 100644 --- a/package-lock.json +++ b/package-lock.json @@ -2778,6 +2778,66 @@ "react": "^16.8.0 || ^17.0.0-rc.1 || ^18.0.0 || ^19.0.0-rc.1" } }, + "node_modules/@redis/bloom": { + "version": "5.8.2", + "resolved": "https://registry.npmjs.org/@redis/bloom/-/bloom-5.8.2.tgz", + "integrity": "sha512-855DR0ChetZLarblio5eM0yLwxA9Dqq50t8StXKp5bAtLT0G+rZ+eRzzqxl37sPqQKjUudSYypz55o6nNhbz0A==", + "license": "MIT", + "engines": { + "node": ">= 18" + }, + "peerDependencies": { + "@redis/client": "^5.8.2" + } + }, + "node_modules/@redis/client": { + "version": "5.8.2", + "resolved": "https://registry.npmjs.org/@redis/client/-/client-5.8.2.tgz", + "integrity": "sha512-WtMScno3+eBpTac1Uav2zugXEoXqaU23YznwvFgkPwBQVwEHTDgOG7uEAObtZ/Nyn8SmAMbqkEubJaMOvnqdsQ==", + "license": "MIT", + "dependencies": { + "cluster-key-slot": "1.1.2" + }, + "engines": { + "node": ">= 18" + } + }, + "node_modules/@redis/json": { + "version": "5.8.2", + "resolved": "https://registry.npmjs.org/@redis/json/-/json-5.8.2.tgz", + "integrity": "sha512-uxpVfas3I0LccBX9rIfDgJ0dBrUa3+0Gc8sEwmQQH0vHi7C1Rx1Qn8Nv1QWz5bohoeIXMICFZRcyDONvum2l/w==", + "license": "MIT", + "engines": { + "node": ">= 18" + }, + "peerDependencies": { + "@redis/client": "^5.8.2" + } + }, + "node_modules/@redis/search": { + "version": "5.8.2", + "resolved": "https://registry.npmjs.org/@redis/search/-/search-5.8.2.tgz", + "integrity": "sha512-cNv7HlgayavCBXqPXgaS97DRPVWFznuzsAmmuemi2TMCx5scwLiP50TeZvUS06h/MG96YNPe6A0Zt57yayfxwA==", + "license": "MIT", + "engines": { + "node": ">= 18" + }, + "peerDependencies": { + "@redis/client": "^5.8.2" + } + }, + "node_modules/@redis/time-series": { + "version": "5.8.2", + "resolved": "https://registry.npmjs.org/@redis/time-series/-/time-series-5.8.2.tgz", + "integrity": "sha512-g2NlHM07fK8H4k+613NBsk3y70R2JIM2dPMSkhIjl2Z17SYvaYKdusz85d7VYOrZBWtDrHV/WD2E3vGu+zni8A==", + "license": "MIT", + "engines": { + "node": ">= 18" + }, + "peerDependencies": { + "@redis/client": "^5.8.2" + } + }, "node_modules/@rollup/rollup-android-arm-eabi": { "version": "4.46.2", "resolved": "https://registry.npmjs.org/@rollup/rollup-android-arm-eabi/-/rollup-android-arm-eabi-4.46.2.tgz", @@ -5544,6 +5604,15 @@ "node": ">=6" } }, + "node_modules/cluster-key-slot": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz", + "integrity": "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/cmdk": { "version": "1.1.1", "resolved": "https://registry.npmjs.org/cmdk/-/cmdk-1.1.1.tgz", @@ -11516,6 +11585,22 @@ "node": ">=8.10.0" } }, + "node_modules/redis": { + "version": "5.8.2", + "resolved": "https://registry.npmjs.org/redis/-/redis-5.8.2.tgz", + "integrity": "sha512-31vunZj07++Y1vcFGcnNWEf5jPoTkGARgfWI4+Tk55vdwHxhAvug8VEtW7Cx+/h47NuJTEg/JL77zAwC6E0OeA==", + "license": "MIT", + "dependencies": { + "@redis/bloom": "5.8.2", + "@redis/client": "5.8.2", + "@redis/json": "5.8.2", + "@redis/search": "5.8.2", + "@redis/time-series": "5.8.2" + }, + "engines": { + "node": ">= 18" + } + }, "node_modules/reflect.getprototypeof": { "version": "1.0.10", "resolved": "https://registry.npmjs.org/reflect.getprototypeof/-/reflect.getprototypeof-1.0.10.tgz", @@ -14725,6 +14810,7 @@ "moment-timezone": "^0.5.48", "parse-torrent-title": "github:TheBeastLT/parse-torrent-title", "pg": "^8.16.0", + "redis": "^5.8.2", "sqlite": "^5.1.1", "sqlite3": "^5.1.7", "stremthru": "^0.8.0", diff --git a/packages/core/package.json b/packages/core/package.json index ad9f2a2f..6b5ab96d 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -19,6 +19,7 @@ "moment-timezone": "^0.5.48", "parse-torrent-title": "github:TheBeastLT/parse-torrent-title", "pg": "^8.16.0", + "redis": "^5.8.2", "sqlite": "^5.1.1", "sqlite3": "^5.1.7", "stremthru": "^0.8.0", diff --git a/packages/core/src/builtins/gdrive/api.ts b/packages/core/src/builtins/gdrive/api.ts index 89d9c429..70ce1e8f 100644 --- a/packages/core/src/builtins/gdrive/api.ts +++ b/packages/core/src/builtins/gdrive/api.ts @@ -76,7 +76,7 @@ export class GoogleOAuth { throw new Error('Builtin GDrive client ID and secret are not set'); } this.refreshToken = refreshToken; - this.accessToken = accessTokenCache.get(this.refreshToken); + this.accessToken = undefined; } private static get clientId(): string { @@ -160,7 +160,7 @@ export class GoogleOAuth { } private async refreshAccessToken(): Promise { - const cachedToken = accessTokenCache.get(this.refreshToken); + const cachedToken = await accessTokenCache.get(this.refreshToken); if (cachedToken) { this.accessToken = cachedToken; return; diff --git a/packages/core/src/builtins/torbox-search/debrid-service.ts b/packages/core/src/builtins/torbox-search/debrid-service.ts index cc072559..7ff69fb6 100644 --- a/packages/core/src/builtins/torbox-search/debrid-service.ts +++ b/packages/core/src/builtins/torbox-search/debrid-service.ts @@ -76,7 +76,7 @@ export class DebridService { const cacheKey = getSimpleTextHash( `${this.serviceConfig.id}:${torrent.hash}` ); - const cached = this.debridCache.get(cacheKey); + const cached = await this.debridCache.get(cacheKey); if (cached && cached.length > 0) { cachedResults.push(...cached); diff --git a/packages/core/src/builtins/torbox-search/source-handlers.ts b/packages/core/src/builtins/torbox-search/source-handlers.ts index ddc0a27c..091ace2d 100644 --- a/packages/core/src/builtins/torbox-search/source-handlers.ts +++ b/packages/core/src/builtins/torbox-search/source-handlers.ts @@ -179,7 +179,9 @@ export class TorrentSourceHandler extends SourceHandler { } } - const titleMetadata = this.metadataCache.get(`metadata:${type}:${id}`); + const titleMetadata = await this.metadataCache.get( + `metadata:${type}:${id}` + ); const filesByHash = await this.getAvailableFilesFromDebrid( torrents, parsedId, @@ -229,7 +231,7 @@ export class TorrentSourceHandler extends SourceHandler { ) { cacheKey += `:${this.searchApi.apiKey}`; } - const cachedTorrents = this.searchCache.get(cacheKey); + const cachedTorrents = await this.searchCache.get(cacheKey); if ( cachedTorrents && @@ -294,7 +296,7 @@ export class TorrentSourceHandler extends SourceHandler { return []; } - this.searchCache.set( + await this.searchCache.set( cacheKey, torrents.filter( (torrent) => @@ -364,7 +366,7 @@ export class UsenetSourceHandler extends SourceHandler { const cacheKey = `usenet:${type}:${id}:${season}:${episode}`; let usingCachedSearch = false; - let torrents = this.searchCache.get(cacheKey); + let torrents = await this.searchCache.get(cacheKey); if (!torrents) { const start = Date.now(); @@ -380,7 +382,7 @@ export class UsenetSourceHandler extends SourceHandler { if (torrents.length === 0) { return []; } - this.searchCache.set( + await this.searchCache.set( cacheKey, torrents, Env.BUILTIN_TORBOX_SEARCH_SEARCH_API_CACHE_TTL @@ -463,7 +465,9 @@ export class UsenetSourceHandler extends SourceHandler { const torrentsToCheck: Torrent[] = []; for (const torrent of torrents) { - const cachedStatus = this.instantAvailabilityCache.get(torrent.hash); + const cachedStatus = await this.instantAvailabilityCache.get( + torrent.hash + ); if (cachedStatus !== undefined) { instantAvailability.set(torrent.hash, cachedStatus); } else { diff --git a/packages/core/src/db/db.ts b/packages/core/src/db/db.ts index 09ff4a3a..8f250008 100644 --- a/packages/core/src/db/db.ts +++ b/packages/core/src/db/db.ts @@ -82,6 +82,7 @@ export class DB { DB.initialised = true; DB.dialect = this.uri.dialect; + logger.info('Database initialised'); } catch (error) { logger.error('Failed to initialize database:', error); throw error; diff --git a/packages/core/src/debrid/interface.ts b/packages/core/src/debrid/interface.ts index e17afc7b..c37a4481 100644 --- a/packages/core/src/debrid/interface.ts +++ b/packages/core/src/debrid/interface.ts @@ -116,10 +116,10 @@ export class DebridInterface { private async resolveTorrent( playbackInfo: PlaybackInfo & { type: 'torrent' }, filename: string - ) { + ): Promise { let { hash, index, parsedId } = playbackInfo; const cacheKey = `${this.storeAuth.storeName}:${hash}:${index ?? 'undefined'}:${this.storeAuth.storeCredential}:${this.clientIp}`; - const cachedLink = DebridInterface.playbackLinkCache.get(cacheKey); + const cachedLink = await DebridInterface.playbackLinkCache.get(cacheKey); if (cachedLink) { logger.debug(`Using cached link for ${hash}`); return cachedLink; @@ -181,7 +181,11 @@ export class DebridInterface { }); const playbackLink = link.data.link; - DebridInterface.playbackLinkCache.set(cacheKey, playbackLink, 60 * 30); + await DebridInterface.playbackLinkCache.set( + cacheKey, + playbackLink, + 60 * 30 + ); return playbackLink; } @@ -191,7 +195,7 @@ export class DebridInterface { ): Promise { const { nzb } = playbackInfo; const cacheKey = `${this.storeAuth.storeName}:${this.storeAuth.storeCredential}:${nzb}:${this.clientIp}`; - const cachedLink = DebridInterface.playbackLinkCache.get(cacheKey); + const cachedLink = await DebridInterface.playbackLinkCache.get(cacheKey); if (cachedLink) { logger.debug(`Using cached link for ${nzb}`); return cachedLink; @@ -236,7 +240,11 @@ export class DebridInterface { const playbackLink = link.data?.data; if (playbackLink) { - DebridInterface.playbackLinkCache.set(cacheKey, playbackLink, 60 * 30); + await DebridInterface.playbackLinkCache.set( + cacheKey, + playbackLink, + 60 * 30 + ); } return playbackLink; } diff --git a/packages/core/src/main.ts b/packages/core/src/main.ts index 129f5906..6dd0cb88 100644 --- a/packages/core/src/main.ts +++ b/packages/core/src/main.ts @@ -44,7 +44,11 @@ import { TMDBMetadata, TMDBMetadataResponse } from './metadata/tmdb'; const logger = createLogger('core'); const shuffleCache = Cache.getInstance('shuffle'); -const precacheCache = Cache.getInstance('precache'); +const precacheCache = Cache.getInstance( + 'precache', + undefined, + true +); export interface AIOStreamsError { title?: string; @@ -182,7 +186,7 @@ export class AIOStreams { // within the last 24 hours (Env.PRECACHE_NEXT_EPISODE_MIN_INTERVAL) let precache = false; const cacheKey = `precache-${type}-${id}-${this.userData.uuid}`; - const cachedNextEpisode = precacheCache.get(cacheKey, false); + const cachedNextEpisode = await precacheCache.get(cacheKey, false); if (cachedNextEpisode) { logger.info( `The current request for ${type} ${id} has already had the next episode precached within the last ${Env.PRECACHE_NEXT_EPISODE_MIN_INTERVAL} seconds (${precacheCache.getTTL(cacheKey)} seconds left). Skipping precaching.` @@ -302,7 +306,7 @@ export class AIOStreams { if (modification?.shuffle && !(extras && extras.includes('search'))) { // shuffle the catalog array if it is not a search const cacheKey = `shuffle-${type}-${actualCatalogId}-${extras}-${this.userData.uuid}`; - const cachedShuffle = shuffleCache.get(cacheKey); + const cachedShuffle = await shuffleCache.get(cacheKey); if (cachedShuffle) { catalog = cachedShuffle; } else { @@ -311,7 +315,7 @@ export class AIOStreams { [catalog[i], catalog[j]] = [catalog[j], catalog[i]]; } if (modification.persistShuffleFor) { - shuffleCache.set( + await shuffleCache.set( cacheKey, catalog, modification.persistShuffleFor * 3600 @@ -1344,7 +1348,11 @@ export class AIOStreams { ); } const cacheKey = `precache-${type}-${id}-${this.userData.uuid}`; - precacheCache.set(cacheKey, true, Env.PRECACHE_NEXT_EPISODE_MIN_INTERVAL); + await precacheCache.set( + cacheKey, + true, + Env.PRECACHE_NEXT_EPISODE_MIN_INTERVAL + ); logger.info(`Successfully precached a stream for ${id} (${type})`); } catch (error) { logger.error(`Error pinging url of first uncached stream`, { diff --git a/packages/core/src/metadata/imdb.ts b/packages/core/src/metadata/imdb.ts index 06484817..520526d1 100644 --- a/packages/core/src/metadata/imdb.ts +++ b/packages/core/src/metadata/imdb.ts @@ -54,9 +54,12 @@ export class IMDBMetadata { } } - private async getImdbSuggestionData(id: string, type: string) { + private async getImdbSuggestionData( + id: string, + type: string + ): Promise { const key = `${id}:${type}`; - const cached = this.titleCache.get(key); + const cached = await this.titleCache.get(key); if (cached) { return cached; } diff --git a/packages/core/src/metadata/tmdb.ts b/packages/core/src/metadata/tmdb.ts index 17bcc095..276ccb1b 100644 --- a/packages/core/src/metadata/tmdb.ts +++ b/packages/core/src/metadata/tmdb.ts @@ -92,7 +92,7 @@ export class TMDBMetadata { // Check cache first const cacheKey = `${id.type}:${id.value}:${type}`; - const cachedId = TMDBMetadata.idCache.get(cacheKey); + const cachedId = await TMDBMetadata.idCache.get(cacheKey); if (cachedId) { return cachedId; } @@ -147,7 +147,7 @@ export class TMDBMetadata { // Check cache first const cacheKey = `${tmdbId}:${type}`; - const cachedMetadata = TMDBMetadata.metadataCache.get(cacheKey); + const cachedMetadata = await TMDBMetadata.metadataCache.get(cacheKey); if (cachedMetadata) { return cachedMetadata; } @@ -231,7 +231,7 @@ export class TMDBMetadata { if (!cacheKey) { throw new Error('TMDB Access Token or API Key is not set'); } - const cachedResult = TMDBMetadata.validationCache.get(cacheKey); + const cachedResult = await TMDBMetadata.validationCache.get(cacheKey); if (cachedResult) { return cachedResult; } diff --git a/packages/core/src/presets/aiCompanion.ts b/packages/core/src/presets/aiCompanion.ts index 885e3091..cd1a539d 100644 --- a/packages/core/src/presets/aiCompanion.ts +++ b/packages/core/src/presets/aiCompanion.ts @@ -285,7 +285,7 @@ export class AICompanionPreset extends Preset { form.append('include_catalogs_movies', movieFeedCatalogs.join(',')); form.append('include_catalogs_series', seriesFeedCatalogs.join(',')); - let manifestUrl: string | undefined = manifestCache.get(cacheKey); + let manifestUrl: string | undefined = await manifestCache.get(cacheKey); if (manifestUrl) { return manifestUrl; } @@ -327,7 +327,11 @@ export class AICompanionPreset extends Preset { if (!result.success || !result.manifest_url) { throw new Error(result.detail || 'Unknown error'); } - manifestCache.set(cacheKey, result.manifest_url, 365 * 24 * 60 * 60); + await manifestCache.set( + cacheKey, + result.manifest_url, + 365 * 24 * 60 * 60 + ); return result.manifest_url; } catch (error) { throw new Error( diff --git a/packages/core/src/presets/aiSearch.ts b/packages/core/src/presets/aiSearch.ts index da9fd882..c8c32502 100644 --- a/packages/core/src/presets/aiSearch.ts +++ b/packages/core/src/presets/aiSearch.ts @@ -265,7 +265,7 @@ export class AISearchPreset extends Preset { // request to /aisearch/encrypt with cache and config as body const cacheKey = `${JSON.stringify(config)}`; - let configId: string | undefined = configCache.get(cacheKey); + let configId: string | undefined = await configCache.get(cacheKey); if (configId) { return `${url}/aisearch/${configId}/manifest.json`; } @@ -289,7 +289,7 @@ export class AISearchPreset extends Preset { }); const result = schema.parse(data); configId = result.encryptedConfig; - configCache.set(cacheKey, configId, 365 * 24 * 60 * 60); + await configCache.set(cacheKey, configId, 365 * 24 * 60 * 60); return `${url}/aisearch/${configId}/manifest.json`; } catch (error) { throw new Error( diff --git a/packages/core/src/presets/moreLikeThis.ts b/packages/core/src/presets/moreLikeThis.ts index 63f3afd2..3921e2f5 100644 --- a/packages/core/src/presets/moreLikeThis.ts +++ b/packages/core/src/presets/moreLikeThis.ts @@ -443,7 +443,7 @@ export class MoreLikeThisPreset extends Preset { forCopy: 'true', }; - const cachedManifest = moreLikeThisManifests.get( + const cachedManifest = await moreLikeThisManifests.get( getSimpleTextHash(`${url}?${JSON.stringify(config)}`) ); if (cachedManifest) { diff --git a/packages/core/src/proxy/base.ts b/packages/core/src/proxy/base.ts index 76b1c697..551a57bb 100644 --- a/packages/core/src/proxy/base.ts +++ b/packages/core/src/proxy/base.ts @@ -69,7 +69,7 @@ export abstract class BaseProxy { } const cacheKey = `${this.config.id}:${this.config.url}:${this.config.credentials}`; - const cachedPublicIp = cache ? cache.get(cacheKey) : null; + const cachedPublicIp = cache ? await cache.get(cacheKey) : null; if (cachedPublicIp) { logger.debug('Returning cached public IP'); return cachedPublicIp; @@ -99,7 +99,7 @@ export abstract class BaseProxy { const publicIp = this.getPublicIpFromResponse(data); if (publicIp && cache) { - cache.set(cacheKey, publicIp, Env.PROXY_IP_CACHE_TTL); + await cache.set(cacheKey, publicIp, Env.PROXY_IP_CACHE_TTL); } else { logger.error( `Proxy did not respond with a public IP. Response: ${JSON.stringify(data)}` diff --git a/packages/core/src/utils/cache-adapter.ts b/packages/core/src/utils/cache-adapter.ts new file mode 100644 index 00000000..db7571e0 --- /dev/null +++ b/packages/core/src/utils/cache-adapter.ts @@ -0,0 +1,268 @@ +import { createLogger } from './logger'; +import { Env } from './env'; +import { RedisClientType, RedisClientOptions, AbortError } from 'redis'; + +const logger = createLogger('cache'); + +const REDIS_TIMEOUT = 500; + +// 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): Promise; + update(key: K, value: V): Promise; + clear(): Promise; + getTTL(key: K): 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): 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 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; + } +} + +// Redis cache implementation with timeout handling +export class RedisCacheBackend implements CacheBackend { + private client: RedisClientType; + private prefix: string; + private maxSize: number; + private timeout: number; + + constructor( + redisClient: RedisClientType, + prefix: string = 'aiostreams:', + maxSize: number = Env.DEFAULT_MAX_CACHE_SIZE, + timeout: number = REDIS_TIMEOUT + ) { + this.client = redisClient; + this.prefix = prefix; + this.maxSize = maxSize; + this.timeout = timeout; + } + + private getKey(key: K): string { + return `${this.prefix}${String(key)}`; + } + + /** + * Execute Redis operation with timeout + * @param operation Function that performs the Redis operation + * @param fallback Value to return if operation times out or fails + * @param errorMessage Message to log if operation fails + */ + private async withTimeout( + operation: () => Promise, + fallback: T, + errorMessage: string + ): Promise { + // check if the client is connected + if (!this.client.isOpen) { + logger.error(`${errorMessage}: Redis client is not open`); + return fallback; + } + + try { + // Create a promise that rejects after timeout + const timeoutPromise = new Promise((_, reject) => { + const id = setTimeout(() => { + clearTimeout(id); + reject( + new Error(`Redis operation timed out after ${this.timeout}ms`) + ); + }, this.timeout); + }); + + // Race the operation against the timeout + return await Promise.race([operation(), timeoutPromise]); + } catch (err) { + logger.error(`${errorMessage}: ${err}`); + return fallback; + } + } + + async get(key: K, updateTTL: boolean = false): Promise { + const redisKey = this.getKey(key); + + return this.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, + `Error getting key ${String(key)} from Redis` + ); + } + + async set(key: K, value: V, ttl: number): Promise { + const redisKey = this.getKey(key); + + await this.withTimeout( + async () => { + await this.client.set(redisKey, JSON.stringify(value), { + EX: ttl, + }); + return true; + }, + false, + `Error setting key ${String(key)} in Redis` + ); + } + + async update(key: K, value: V): Promise { + const redisKey = this.getKey(key); + + await this.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, + `Error updating key ${String(key)} in Redis` + ); + } + + async clear(): Promise { + await this.withTimeout( + async () => { + // Delete all keys with this prefix + const keys = await this.client.keys(`${this.prefix}*`); + if (keys && keys.length > 0) { + await this.client.del(keys); + } + return true; + }, + false, + `Error clearing Redis cache` + ); + } + + async getTTL(key: K): Promise { + return this.withTimeout( + async () => { + const ttl = await this.client.ttl(this.getKey(key)); + return ttl > 0 ? ttl : 0; + }, + 0, + `Error getting TTL for key ${String(key)} from Redis` + ); + } +} + +// Item stored in memory cache +class CacheItem { + constructor( + public value: T, + public lastAccessed: number, + public createdAt: number, + public ttl: number // Time-To-Live in milliseconds + ) {} +} diff --git a/packages/core/src/utils/cache.ts b/packages/core/src/utils/cache.ts index 406832f8..3dd5c103 100644 --- a/packages/core/src/utils/cache.ts +++ b/packages/core/src/utils/cache.ts @@ -1,5 +1,11 @@ import { createLogger } from './logger'; import { Env } from './env'; +import { + CacheBackend, + MemoryCacheBackend, + RedisCacheBackend, +} from './cache-adapter'; +import { createClient, RedisClientType } from 'redis'; const logger = createLogger('cache'); @@ -15,26 +21,113 @@ function formatBytes(bytes: number, decimals: number = 2): string { return `${parseFloat((bytes / Math.pow(k, i)).toFixed(dm))} ${sizes[i]}`; } -class CacheItem { - constructor( - public value: T, - public lastAccessed: number, - public createdAt: number, - public ttl: number // Time-To-Live in milliseconds - ) {} -} - export class Cache { private static instances: Map = new Map(); - private cache: Map>; - private maxSize: number; private static isStatsLoopRunning: boolean = false; + private backend: CacheBackend; + private maxSize: number; + private name: string; - private constructor(maxSize: number) { - this.cache = new Map>(); + // Redis client singleton + private static redisClient: RedisClientType | null = null; + + private constructor(name: string, maxSize: number, forceMemory: boolean) { + this.name = name; this.maxSize = maxSize; - Cache.startStatsLoop(); + + // Initialize the appropriate backend based on environment configuration + if (Env.REDIS_URI && !forceMemory) { + this.backend = new RedisCacheBackend( + Cache.getRedisClient(), + `${name}:`, + maxSize + ); + logger.debug(`Created Redis cache backend for ${name}`); + } else { + this.backend = new MemoryCacheBackend(maxSize); + Cache.startStatsLoop(); + logger.debug(`Created Memory cache backend for ${name}`); + } } + + private static getRedisClient(): RedisClientType { + if (!this.redisClient) { + logger.info(`Initialising Redis client connection to ${Env.REDIS_URI}`); + this.redisClient = createClient({ + url: Env.REDIS_URI, + }); + this.redisClient.on('connect', () => { + logger.info('Connected to Redis server'); + }); + this.redisClient + .connect() + .then(() => { + if (!this.redisClient) { + throw new Error('Redis client not initialized'); + } + + this.redisClient.on('reconnecting', () => { + logger.warn('Reconnecting to Redis server'); + }); + + this.redisClient.on('error', (err: any) => { + logger.error(`Redis client error: ${err}`); + }); + }) + .catch((err: any) => { + throw new Error(`Failed to connect to Redis server: ${err}`); + }); + } + + return this.redisClient; + } + + /** + * Tests the Redis connection by attempting to set and get a test value + * @throws Error if Redis connection test fails + */ + public static async testRedisConnection(): Promise { + if (!Env.REDIS_URI) { + return; + } + + try { + const client = this.getRedisClient(); + const startTime = Date.now(); + while (Date.now() - startTime < 10000) { + if (client.isReady) { + break; + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + if (!client.isReady) { + throw new Error('Redis connection test timed out'); + } + + const testKey = 'redis:connection:test'; + const testValue = 'test-' + Date.now(); + + await client.set(testKey, testValue, { + expiration: { + type: 'EX', + value: 10, + }, + }); + + const retrievedValue = await client.get(testKey); + + if (retrievedValue !== testValue) { + throw new Error('Redis get/set test failed: values do not match'); + } + + await client.del(testKey); + + logger.info('Redis connection test successful'); + } catch (err: any) { + throw new Error(`Redis connection test failed: ${err.message}`); + } + } + private static startStatsLoop() { if (Cache.isStatsLoopRunning) { return; @@ -58,11 +151,12 @@ export class Cache { */ public static getInstance( name: string, - maxSize: number = Env.DEFAULT_MAX_CACHE_SIZE + maxSize: number = Env.DEFAULT_MAX_CACHE_SIZE, + forceMemory: boolean = false ): Cache { if (!this.instances.has(name)) { logger.debug(`Creating new cache instance: ${name}`); - this.instances.set(name, new Cache(maxSize)); + this.instances.set(name, new Cache(name, maxSize, forceMemory)); } return this.instances.get(name) as Cache; } @@ -71,7 +165,7 @@ export class Cache { * Gets the statistics of the cache in use by the program. returns a formatted string containing a list of all cache instances * and their currently held items, max items */ - public static stats() { + public static async stats() { if (!this.instances || this.instances.size === 0) { return; } @@ -85,31 +179,29 @@ export class Cache { '╠══════════════════════╪══════════╪═════════════════╪═════════════════╣', ]; - const bodyLines = Array.from(this.instances.entries()).map( - ([name, cache]) => { - let instanceSize = 0; - for (const item of cache.cache.values()) { - try { - // Estimate object size by getting the byte length of its JSON string representation. - // This is an approximation but is effective for many use cases. - instanceSize += Buffer.byteLength(JSON.stringify(item), 'utf8'); - } catch (e) { - // Could fail on circular references. In that case, we add 0. - instanceSize += 0; - } - } + const bodyLines = []; - grandTotalItems += cache.cache.size; - grandTotalSize += instanceSize; + for (const [name, cache] of this.instances.entries()) { + let itemCount = 0; + let instanceSize = 0; - const nameStr = name.padEnd(20); - const itemsStr = String(cache.cache.size).padEnd(8); - const maxSizeStr = String(cache.maxSize ?? '-').padEnd(15); - const estSizeStr = formatBytes(instanceSize).padEnd(15); - - return `║ ${nameStr} │ ${itemsStr} │ ${maxSizeStr} │ ${estSizeStr} ║`; + // Get stats differently depending on the backend type + if (cache.backend instanceof MemoryCacheBackend) { + itemCount = cache.backend.getSize(); + instanceSize = cache.backend.getMemoryUsageEstimate(); } - ); + grandTotalItems += itemCount; + grandTotalSize += instanceSize; + + const nameStr = name.padEnd(20); + const itemsStr = String(itemCount).padEnd(8); + const maxSizeStr = String(cache.maxSize ?? '-').padEnd(15); + const estSizeStr = formatBytes(instanceSize).padEnd(15); + + bodyLines.push( + `║ ${nameStr} │ ${itemsStr} │ ${maxSizeStr} │ ${estSizeStr} ║` + ); + } const footer = [ '╚══════════════════════╧══════════╧═════════════════╧═════════════════╝', @@ -133,31 +225,17 @@ export class Cache { ttl: number, ...args: Parameters ): Promise> { - const cachedValue = this.get(key); + const cachedValue = await this.get(key); if (cachedValue !== undefined) { return cachedValue as ReturnType; } const result = await fn(...args); - this.set(key, result, ttl); + await this.set(key, result, ttl); return result; } - get(key: K, updateTTL: boolean = false): V | undefined { - 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 get(key: K, updateTTL: boolean = false): Promise { + return this.backend.get(key, updateTTL); } /** @@ -166,19 +244,8 @@ export class Cache { * @param value The value to set * @param ttl The TTL in seconds */ - set(key: K, value: V, ttl: number): void { - if (this.cache.size >= this.maxSize) { - this.evict(); - } - this.cache.set( - key, - new CacheItem( - structuredClone(value), - Date.now(), - Date.now(), - ttl * 1000 - ) - ); + async set(key: K, value: V, ttl: number): Promise { + return this.backend.set(key, value, ttl); } /** @@ -186,42 +253,15 @@ export class Cache { * @param key The key to update * @param value The new value */ - update(key: K, value: V): void { - const item = this.cache.get(key); - if (item) { - item.value = value; - } + async update(key: K, value: V): Promise { + return this.backend.update(key, value); } - clear(): void { - this.cache.clear(); + async clear(): Promise { + return this.backend.clear(); } - getTTL(key: K): number { - // return the time left in seconds until the item expires - 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); - } + async getTTL(key: K): Promise { + return this.backend.getTTL(key); } } diff --git a/packages/core/src/utils/env.ts b/packages/core/src/utils/env.ts index b0492e2f..f84bc911 100644 --- a/packages/core/src/utils/env.ts +++ b/packages/core/src/utils/env.ts @@ -253,6 +253,10 @@ export const Env = cleanEnv(process.env, { default: 'sqlite://./data/db.sqlite', desc: 'Database URI for the addon', }), + REDIS_URI: str({ + default: undefined, + desc: 'Redis URI for the addon', + }), ADDON_PROXY: url({ default: undefined, desc: 'Proxy URL for the addon', diff --git a/packages/core/src/utils/http.ts b/packages/core/src/utils/http.ts index ac018ab0..f2ffb3f4 100644 --- a/packages/core/src/utils/http.ts +++ b/packages/core/src/utils/http.ts @@ -13,7 +13,11 @@ import { import { socksDispatcher } from 'fetch-socks'; const logger = createLogger('http'); -const urlCount = Cache.getInstance('url-count'); +const urlCount = Cache.getInstance( + 'url-count', + undefined, + true +); export class PossibleRecursiveRequestError extends Error { constructor(message: string) { @@ -47,7 +51,7 @@ export interface RequestOptions { rawOptions?: RequestInit; } -export function makeRequest(url: string, options: RequestOptions) { +export async function makeRequest(url: string, options: RequestOptions) { const urlObj = new URL(url); const useProxy = shouldProxy(urlObj); const headers = new Headers(options.headers); @@ -79,7 +83,7 @@ export function makeRequest(url: string, options: RequestOptions) { // block recursive requests const key = `${urlObj.toString()}-${options.forwardIp}`; - const currentCount = urlCount.get(key) ?? 0; + const currentCount = (await urlCount.get(key)) ?? 0; if ( currentCount > Env.RECURSION_THRESHOLD_LIMIT && !options.ignoreRecursion @@ -92,9 +96,9 @@ export function makeRequest(url: string, options: RequestOptions) { ); } if (currentCount > 0) { - urlCount.update(key, currentCount + 1); + await urlCount.update(key, currentCount + 1); } else { - urlCount.set(key, 1, Env.RECURSION_THRESHOLD_WINDOW); + await urlCount.set(key, 1, Env.RECURSION_THRESHOLD_WINDOW); } logger.debug( `Making a ${useProxy ? 'proxied' : 'direct'} request to ${makeUrlLogSafe( diff --git a/packages/core/src/utils/rpdb.ts b/packages/core/src/utils/rpdb.ts index 6850defb..35e763b1 100644 --- a/packages/core/src/utils/rpdb.ts +++ b/packages/core/src/utils/rpdb.ts @@ -9,7 +9,7 @@ interface Id { value: string; } -const apiKeyValidationCache = Cache.getInstance('rpdbApiKey'); +const apiKeyValidationCache = Cache.getInstance('rpdbApiKey'); const posterCheckCache = Cache.getInstance('rpdbPosterCheck'); export class RPDB { @@ -24,8 +24,8 @@ export class RPDB { } } - public async validateApiKey() { - const cached = apiKeyValidationCache.get(this.apiKey); + public async validateApiKey(): Promise { + const cached = await apiKeyValidationCache.get(this.apiKey); if (cached) { return cached; } @@ -53,6 +53,7 @@ export class RPDB { data.valid, Env.RPDB_API_KEY_VALIDITY_CACHE_TTL ); + return data.valid; } /** * @@ -65,7 +66,7 @@ export class RPDB { ): Promise { const parsedId = this.getParsedId(id, type); const cacheKey = `${type}-${id}`; - const cached = posterCheckCache.get(cacheKey); + const cached = await posterCheckCache.get(cacheKey); if (cached) { return cached; } diff --git a/packages/core/src/utils/startup.ts b/packages/core/src/utils/startup.ts index 7dbca058..f52b7f4d 100644 --- a/packages/core/src/utils/startup.ts +++ b/packages/core/src/utils/startup.ts @@ -95,8 +95,9 @@ const logStartupInfo = () => { }); // Database & Storage - logSection('DATABASE & STORAGE', '💾', () => { + logSection('STORAGE & CACHE', '💾', () => { const dbType = Env.DATABASE_URI.split('://')[0].toUpperCase(); + const cacheType = Env.REDIS_URI ? 'Redis' : 'Memory'; logKeyValue('Database Type:', dbType); if (Env.DATABASE_URI.includes('sqlite')) { const dbPath = @@ -108,6 +109,10 @@ const logStartupInfo = () => { Env.DATABASE_URI.replace(/:\/\/.*@/, '://***@') ); // Hide credentials } + logKeyValue('Cache Type:', cacheType); + if (Env.REDIS_URI) { + logKeyValue('Redis URI:', Env.REDIS_URI); + } }); // Logging Configuration diff --git a/packages/core/src/wrapper.ts b/packages/core/src/wrapper.ts index 47223b2e..a3f1ba66 100644 --- a/packages/core/src/wrapper.ts +++ b/packages/core/src/wrapper.ts @@ -367,7 +367,7 @@ export class Wrapper { } = options; if (cacher) { - const cached = cacher.get(cacheKey); + const cached = await cacher.get(cacheKey); if (cached) { logger.info( `Returning cached ${resourceName} for ${this.getAddonName(this.addon)}` @@ -380,7 +380,7 @@ export class Wrapper { const result = await requestFn(); const doCache = shouldCache ? shouldCache(result) : true; if (cacher && doCache) { - cacher.set(cacheKey, result, cacheTtl); + await cacher.set(cacheKey, result, cacheTtl); } return result; }; diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index eb3beb17..b2db49bb 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -6,6 +6,7 @@ import { DB, UserRepository, logStartupInfo, + Cache, } from '@aiostreams/core'; const logger = createLogger('server'); @@ -13,7 +14,6 @@ const logger = createLogger('server'); async function initialiseDatabase() { try { await DB.getInstance().initialise(Env.DATABASE_URI, []); - logger.info('Database initialised'); } catch (error) { logger.error('Failed to initialise database:', error); throw error; @@ -30,9 +30,16 @@ async function startAutoPrune() { setTimeout(startAutoPrune, Env.PRUNE_INTERVAL * 1000); } +async function initialiseRedis() { + if (Env.REDIS_URI) { + await Cache.testRedisConnection(); + } +} + async function start() { try { await initialiseDatabase(); + await initialiseRedis(); if (Env.PRUNE_MAX_DAYS >= 0) { startAutoPrune(); }