diff --git a/packages/core/src/builtins/base/debrid.ts b/packages/core/src/builtins/base/debrid.ts index ed5e9723..4148e35f 100644 --- a/packages/core/src/builtins/base/debrid.ts +++ b/packages/core/src/builtins/base/debrid.ts @@ -3,9 +3,12 @@ import { z, ZodError } from 'zod'; import { IdParser, IdType, ParsedId } from '../../utils/id-parser.js'; import { AnimeDatabase, + BuiltinServiceId, constants, + encryptString, Env, formatZodError, + getSimpleTextHash, getTimeTakenSincePoint, SERVICE_DETAILS, } from '../../utils/index.js'; @@ -21,6 +24,9 @@ import { ServiceAuth, DebridError, generatePlaybackUrl, + TitleMetadata as DebridTitleMetadata, + metadataStore, + FileInfo, } from '../../debrid/index.js'; import { processTorrents, processNZBs } from '../utils/debrid.js'; import { calculateAbsoluteEpisode } from '../utils/general.js'; @@ -212,9 +218,34 @@ export abstract class BaseDebridAddon { ), ]); + const encryptedStoreAuths = this.userData.services.reduce( + (acc, service) => { + const auth = { + id: service.id, + credential: service.credential, + }; + acc[service.id] = encryptString(JSON.stringify(auth)).data ?? ''; + return acc; + }, + {} as Record + ); + const debridTitleMetadata: DebridTitleMetadata = { + titles: searchMetadata.titles, + year: searchMetadata.year, + season: searchMetadata.season, + episode: searchMetadata.episode, + absoluteEpisode: searchMetadata.absoluteEpisode, + }; + const metadataId = getSimpleTextHash(JSON.stringify(debridTitleMetadata)); + await metadataStore().set( + metadataId, + debridTitleMetadata, + Env.BUILTIN_PLAYBACK_LINK_VALIDITY + ); + const resultStreams = await Promise.all( [...processedTorrents.results, ...processedNzbs.results].map((result) => - this._createStream(result, this.userData, searchMetadata) + this._createStream(result, encryptedStoreAuths, metadataId) ) ); @@ -414,37 +445,27 @@ export abstract class BaseDebridAddon { protected _createStream( torrentOrNzb: TorrentWithSelectedFile | NZBWithSelectedFile, - userData: T, - titleMetadata?: TitleMetadata + encryptedStoreAuths: Record, + metadataId: string ): Stream { // Handle debrid streaming - const storeAuth: ServiceAuth | undefined = torrentOrNzb.service - ? { - id: torrentOrNzb.service!.id, - credential: - userData.services.find( - (service) => service.id === torrentOrNzb.service!.id - )?.credential ?? '', - } + const encryptedStoreAuth = torrentOrNzb.service + ? encryptedStoreAuths?.[torrentOrNzb.service?.id] : undefined; - const playbackInfo: PlaybackInfo | undefined = torrentOrNzb.service + const fileInfo: FileInfo | undefined = torrentOrNzb.service ? torrentOrNzb.type === 'torrent' ? { type: 'torrent', hash: torrentOrNzb.hash, sources: torrentOrNzb.sources, - title: torrentOrNzb.title, - file: torrentOrNzb.file, - metadata: titleMetadata, + index: torrentOrNzb.file.index, } : { type: 'usenet', nzb: torrentOrNzb.nzb, - title: torrentOrNzb.title, hash: torrentOrNzb.hash, - file: torrentOrNzb.file, - metadata: titleMetadata, + index: torrentOrNzb.file.index, } : undefined; @@ -471,9 +492,11 @@ export abstract class BaseDebridAddon { return { url: torrentOrNzb.service ? generatePlaybackUrl( - storeAuth!, - playbackInfo!, - torrentOrNzb.file.name || torrentOrNzb.title || 'unknown' + encryptedStoreAuth!, + metadataId!, + fileInfo!, + torrentOrNzb.title, + torrentOrNzb.file.name ) : undefined, name, diff --git a/packages/core/src/builtins/torbox-search/source-handlers.ts b/packages/core/src/builtins/torbox-search/source-handlers.ts index 8682479a..5d6444df 100644 --- a/packages/core/src/builtins/torbox-search/source-handlers.ts +++ b/packages/core/src/builtins/torbox-search/source-handlers.ts @@ -2,10 +2,13 @@ import { number, z } from 'zod'; import { Stream } from '../../db/index.js'; import { AnimeDatabase, + BuiltinServiceId, Cache, Env, SERVICE_DETAILS, createLogger, + encryptString, + getSimpleTextHash, getTimeTakenSincePoint, } from '../../utils/index.js'; // import { DebridService, DebridFile } from './debrid-service'; @@ -24,8 +27,9 @@ import { NZBWithSelectedFile, TorrentWithSelectedFile, generatePlaybackUrl, + metadataStore, } from '../../debrid/utils.js'; -import { DebridFile, PlaybackInfo } from '../../debrid/index.js'; +import { DebridFile, FileInfo, PlaybackInfo } from '../../debrid/index.js'; import { getTraktAliases } from '../../metadata/trakt.js'; const logger = createLogger('torbox-search'); @@ -75,70 +79,69 @@ abstract class SourceHandler { } protected createStream( - id: ParsedId, - torrentOrNZB: TorrentWithSelectedFile | NZBWithSelectedFile, - userData: z.infer, - titleMetadata?: TitleMetadata - ): Stream & { type: 'torrent' | 'usenet' } { - if (!torrentOrNZB.service) { - throw new Error('Torrent or NZB has no service'); - } - const storeAuth = { - id: torrentOrNZB.service.id, - credential: - userData.services.find( - (service) => service.id === torrentOrNZB.service!.id - )?.credential ?? '', - }; + torrentOrNzb: TorrentWithSelectedFile | NZBWithSelectedFile, + encryptedStoreAuths: Record, + metadataId: string + ): Stream { + // Handle debrid streaming + const encryptedStoreAuth = torrentOrNzb.service + ? encryptedStoreAuths?.[torrentOrNzb.service?.id] + : undefined; - // const playbackInfo: PlaybackInfo = { - // type: 'usenet', - // hash: torrent.hash, - // magnet: torrent.type === 'torrent' ? torrent.magnet : undefined, - // title: torrent.title, - // nzb: torrent.type === 'usenet' ? torrent.nzb : undefined, - // file: torrent.file, - // metadata: titleMetadata, - // }; - const playbackInfo: PlaybackInfo = - torrentOrNZB.type === 'torrent' + const fileInfo: FileInfo | undefined = torrentOrNzb.service + ? torrentOrNzb.type === 'torrent' ? { type: 'torrent', - hash: torrentOrNZB.hash, - sources: torrentOrNZB.sources, - // magnet: torrentOrNZB.magnet, - title: torrentOrNZB.title, - file: torrentOrNZB.file, - metadata: titleMetadata, + hash: torrentOrNzb.hash, + sources: torrentOrNzb.sources, + index: torrentOrNzb.file.index, } : { type: 'usenet', - nzb: torrentOrNZB.nzb, - title: torrentOrNZB.title, - hash: torrentOrNZB.hash, - file: torrentOrNZB.file, - metadata: titleMetadata, - }; + nzb: torrentOrNzb.nzb, + hash: torrentOrNzb.hash, + index: torrentOrNzb.file.index, + } + : undefined; - const svcMeta = SERVICE_DETAILS[torrentOrNZB.service.id]; - const name = `[${svcMeta.shortName} ${torrentOrNZB.service.cached ? '⚡' : '⏳'}${torrentOrNZB.service.owned ? ' ☁️' : ''}] TorBox Search`; - const description = `${torrentOrNZB.title}\n${torrentOrNZB.file.name}\n${torrentOrNZB.indexer ? `🔍 ${torrentOrNZB.indexer}` : ''} ${torrentOrNZB.seeders ? `👤 ${torrentOrNZB.seeders}` : ''} ${torrentOrNZB.age && torrentOrNZB.age !== '0d' ? `🕒 ${torrentOrNZB.age}` : ''}`; + const svcMeta = torrentOrNzb.service + ? SERVICE_DETAILS[torrentOrNzb.service.id] + : undefined; + // const svcMeta = SERVICE_DETAILS[torrentOrNzb.service.id]; + const shortCode = svcMeta?.shortName || 'P2P'; + const cacheIndicator = torrentOrNzb.service + ? torrentOrNzb.service.cached + ? '⚡' + : '⏳' + : ''; + + const name = `[${shortCode} ${cacheIndicator}${torrentOrNzb.service?.owned ? ' ☁️' : ''}] TorBox Search`; + const description = `${torrentOrNzb.title}\n${torrentOrNzb.file.name}\n${ + torrentOrNzb.indexer ? `🔍 ${torrentOrNzb.indexer}` : '' + } ${'seeders' in torrentOrNzb && torrentOrNzb.seeders ? `👤 ${torrentOrNzb.seeders}` : ''} ${ + torrentOrNzb.age && torrentOrNzb.age !== '0d' + ? `🕒 ${torrentOrNzb.age}` + : '' + }`; return { - url: torrentOrNZB.service + url: torrentOrNzb.service ? generatePlaybackUrl( - storeAuth!, - playbackInfo!, - torrentOrNZB.file.name || torrentOrNZB.title || 'unknown' + encryptedStoreAuth!, + metadataId!, + fileInfo!, + torrentOrNzb.title, + torrentOrNzb.file.name ) : undefined, name, description, - type: torrentOrNZB.type, - infoHash: torrentOrNZB.hash, + type: torrentOrNzb.type, + infoHash: torrentOrNzb.hash, + fileIdx: torrentOrNzb.file.index, behaviorHints: { - videoSize: torrentOrNZB.file.size, - filename: torrentOrNZB.file.name, + videoSize: torrentOrNzb.file.size, + filename: torrentOrNzb.file.name, }, }; } @@ -333,8 +336,29 @@ export class TorrentSourceHandler extends SourceHandler { ?.owned ?? false; }); + const encryptedStoreAuths = userData.services.reduce( + (acc, service) => { + const auth = { + id: service.id, + credential: service.credential, + }; + acc[service.id] = encryptString(JSON.stringify(auth)).data ?? ''; + return acc; + }, + {} as Record + ); + + const metadataId = getSimpleTextHash(JSON.stringify(fetchResult.metadata)); + if (fetchResult.metadata) { + await metadataStore().set( + metadataId, + fetchResult.metadata, + Env.BUILTIN_PLAYBACK_LINK_VALIDITY + ); + } + return results.map((result) => - this.createStream(parsedId, result, userData, fetchResult.metadata) + this.createStream(result, encryptedStoreAuths, metadataId) ); } @@ -440,7 +464,7 @@ export class UsenetSourceHandler extends SourceHandler { `metadata:${type}:${value}` ); - if (!torrents) { + if (!torrents || !titleMetadata) { const start = Date.now(); try { const data = await this.searchApi.getUsenetById( @@ -541,8 +565,29 @@ export class UsenetSourceHandler extends SourceHandler { nzbs.find((nzb) => nzb.hash === result.hash)?.owned ?? false; }); + const encryptedStoreAuths = userData.services.reduce( + (acc, service) => { + const auth = { + id: service.id, + credential: service.credential, + }; + acc[service.id] = encryptString(JSON.stringify(auth)).data ?? ''; + return acc; + }, + {} as Record + ); + + const metadataId = getSimpleTextHash(JSON.stringify(titleMetadata)); + if (titleMetadata) { + await metadataStore().set( + metadataId, + titleMetadata, + Env.BUILTIN_PLAYBACK_LINK_VALIDITY + ); + } + return results.map((result) => - this.createStream(parsedId, result, userData, titleMetadata) + this.createStream(result, encryptedStoreAuths, metadataId) ); } } diff --git a/packages/core/src/debrid/base.ts b/packages/core/src/debrid/base.ts index f388fe35..982cb371 100644 --- a/packages/core/src/debrid/base.ts +++ b/packages/core/src/debrid/base.ts @@ -102,39 +102,52 @@ export interface DebridDownload { files?: DebridFile[]; } -const BasePlaybackInfoSchema = z.object({ - // hash: z.string(), - title: z.string().optional(), - metadata: z - .object({ - titles: z.array(z.string()), - year: z.number().optional(), - season: z.number().optional(), - episode: z.number().optional(), - absoluteEpisode: z.number().optional(), - }) - .optional(), - file: DebridFileSchema.optional(), +const TitleMetadataSchema = z.object({ + titles: z.array(z.string()), + year: z.number().optional(), + season: z.number().optional(), + episode: z.number().optional(), + absoluteEpisode: z.number().optional(), }); -const TorrentPlaybackInfoSchema = BasePlaybackInfoSchema.extend({ +const BasePlaybackInfoSchema = z.object({ + // title: z.string().optional(), + metadata: TitleMetadataSchema.optional(), + filename: z.string().optional(), + index: z.number().optional(), +}); + +const BaseFileInfoSchema = z.object({ + index: z.number().optional(), +}); + +const TorrentInfoSchema = BaseFileInfoSchema.extend({ hash: z.string(), sources: z.array(z.string()), - // magnet: z.string().optional(), type: z.literal('torrent'), }); -const UsenetPlaybackInfoSchema = BasePlaybackInfoSchema.extend({ +const TorrentPlaybackInfoSchema = + BasePlaybackInfoSchema.merge(TorrentInfoSchema); + +const UsenetInfoSchema = BaseFileInfoSchema.extend({ hash: z.string(), nzb: z.string(), type: z.literal('usenet'), }); +const UsenetPlaybackInfoSchema = BasePlaybackInfoSchema.merge(UsenetInfoSchema); + export const PlaybackInfoSchema = z.discriminatedUnion('type', [ TorrentPlaybackInfoSchema, UsenetPlaybackInfoSchema, ]); +export const FileInfoSchema = z.discriminatedUnion('type', [ + TorrentInfoSchema, + UsenetInfoSchema, +]); + export const ServiceAuthSchema = z.object({ id: z.enum(constants.BUILTIN_SUPPORTED_SERVICES), credential: z.string(), @@ -142,6 +155,8 @@ export const ServiceAuthSchema = z.object({ export type ServiceAuth = z.infer; export type PlaybackInfo = z.infer; +export type FileInfo = z.infer; +export type TitleMetadata = z.infer; export interface DebridService { // Common methods diff --git a/packages/core/src/debrid/stremthru.ts b/packages/core/src/debrid/stremthru.ts index 6b3a5db9..73a23f07 100644 --- a/packages/core/src/debrid/stremthru.ts +++ b/packages/core/src/debrid/stremthru.ts @@ -209,7 +209,7 @@ export class StremThruInterface implements DebridService { }); } - const { hash, file: chosenFile, metadata } = playbackInfo; + const { hash, metadata } = playbackInfo; const cacheKey = `${this.serviceName}:${this.config.token}:${this.config.clientIp}:${playbackInfo.hash}:${playbackInfo.metadata?.season}:${playbackInfo.metadata?.episode}:${playbackInfo.metadata?.absoluteEpisode}`; const cachedLink = await StremThruInterface.playbackLinkCache.get(cacheKey); @@ -249,7 +249,7 @@ export class StremThruInterface implements DebridService { } const torrent: Torrent = { - title: magnetDownload.name || playbackInfo.title || '', + title: magnetDownload.name || '', hash: hash, size: magnetDownload.size || 0, type: 'torrent', @@ -273,8 +273,8 @@ export class StremThruInterface implements DebridService { parsedFiles, metadata, { - chosenFilename: chosenFile?.name, - chosenIndex: chosenFile?.index, + chosenFilename: playbackInfo.filename, + chosenIndex: playbackInfo.index, } ); diff --git a/packages/core/src/debrid/torbox.ts b/packages/core/src/debrid/torbox.ts index 6c698d32..5b151b41 100644 --- a/packages/core/src/debrid/torbox.ts +++ b/packages/core/src/debrid/torbox.ts @@ -263,7 +263,7 @@ export class TorboxDebridService implements DebridService { return this.stremthru.resolve(playbackInfo, filename); } - const { nzb, file: chosenFile, metadata, title, hash } = playbackInfo; + const { nzb, metadata, hash } = playbackInfo; const cacheKey = `${this.serviceName}:${this.config.token}:${this.config.clientIp}:${JSON.stringify(playbackInfo)}`; const cachedLink = await TorboxDebridService.playbackLinkCache.get(cacheKey); @@ -307,8 +307,8 @@ export class TorboxDebridService implements DebridService { type: 'usenet' as const, nzb: nzb, hash: hash, - title: title || usenetDownload.name, - file: chosenFile, + title: usenetDownload.name, + file: usenetDownload.files[playbackInfo.index ?? 0], metadata: metadata, size: usenetDownload.size || 0, }; @@ -330,8 +330,8 @@ export class TorboxDebridService implements DebridService { parsedFiles, metadata, { - chosenFilename: chosenFile?.name, - chosenIndex: chosenFile?.index, + chosenFilename: playbackInfo.filename, + chosenIndex: playbackInfo.index, } ); diff --git a/packages/core/src/debrid/utils.ts b/packages/core/src/debrid/utils.ts index 9d53cdd7..5f2c9ba7 100644 --- a/packages/core/src/debrid/utils.ts +++ b/packages/core/src/debrid/utils.ts @@ -13,6 +13,8 @@ import { DebridDownload, PlaybackInfo, ServiceAuth, + FileInfo, + TitleMetadata, } from './base.js'; import { normaliseTitle, titleMatch } from '../parser/utils.js'; @@ -269,9 +271,6 @@ export async function selectFileInTorrentOrNZB( // return undefined; // } } - logger.verbose( - `File selected with score ${bestMatch.score}: ${bestMatch.file.name}` - ); return bestMatch.file; } @@ -324,28 +323,33 @@ export function isVideoFile(file: DebridFile): boolean { ); } -export const pbiCache = () => { - const prefix = 'pbi'; - if (Env.REDIS_URI && Env.BUILTIN_PLAYBACK_LINK_STORE === 'redis') { - return Cache.getInstance( - prefix, - 1_000_000_000, - 'redis' - ); - } - return Cache.getInstance(prefix, 1_000_000_000, 'sql'); +export const metadataStore = () => { + const prefix = 'mds'; + const store: 'redis' | 'sql' | 'memory' = + Env.BUILTIN_DEBRID_METADATA_STORE || (Env.REDIS_URI ? 'redis' : 'sql'); + return Cache.getInstance(prefix, 1_000_000_000, store); }; +// export function generatePlaybackUrl( +// storeAuth: ServiceAuth, +// playbackInfo: MinimisedPlaybackInfo, +// filename: string +// ) { +// const encryptedStoreAuth = encryptString(JSON.stringify(storeAuth)); +// if (!encryptedStoreAuth.success) { +// throw new Error('Failed to encrypt store auth'); +// } +// const playbackId = getSimpleTextHash(JSON.stringify(playbackInfo)); +// pbiCache().set(playbackId, playbackInfo, Env.BUILTIN_PLAYBACK_LINK_VALIDITY); +// return `${Env.BASE_URL}/api/v1/debrid/playback/${encryptedStoreAuth.data}/${playbackId}/${encodeURIComponent(filename)}`; +// } + export function generatePlaybackUrl( - storeAuth: ServiceAuth, - playbackInfo: PlaybackInfo, - filename: string -) { - const encryptedStoreAuth = encryptString(JSON.stringify(storeAuth)); - if (!encryptedStoreAuth.success) { - throw new Error('Failed to encrypt store auth'); - } - const playbackId = getSimpleTextHash(JSON.stringify(playbackInfo)); - pbiCache().set(playbackId, playbackInfo, Env.BUILTIN_PLAYBACK_LINK_VALIDITY); - return `${Env.BASE_URL}/api/v1/debrid/playback/${encryptedStoreAuth.data}/${playbackId}/${encodeURIComponent(filename)}`; + encryptedStoreAuth: string, + metadataId: string, + fileInfo: FileInfo, + title?: string, + filename?: string +): string { + return `${Env.BASE_URL}/api/v1/debrid/playback/${encryptedStoreAuth}/${Buffer.from(JSON.stringify(fileInfo)).toString('base64')}/${metadataId}/${encodeURIComponent(filename ?? title ?? 'unknown')}`; } diff --git a/packages/core/src/utils/cache-adapter.ts b/packages/core/src/utils/cache-adapter.ts index de7250c6..bfbabe0e 100644 --- a/packages/core/src/utils/cache-adapter.ts +++ b/packages/core/src/utils/cache-adapter.ts @@ -1,7 +1,8 @@ import { RedisClientType } from 'redis'; import { REDIS_PREFIX, Env } from './index.js'; -import { createLogger } from './logger.js'; +import { createLogger, getTimeTakenSincePoint } from './logger.js'; import { DB } from '../db/db.js'; +import { withTimeout } from './general.js'; const logger = createLogger('cache'); @@ -126,6 +127,15 @@ export class RedisCacheBackend implements CacheBackend { 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, @@ -136,52 +146,29 @@ export class RedisCacheBackend implements CacheBackend { 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)}`; } - /** - * 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; - } + 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 this.withTimeout( + return withTimeout( async () => { const data = await this.client.get(redisKey); if (!data) return undefined; @@ -197,32 +184,81 @@ export class RedisCacheBackend implements CacheBackend { return JSON.parse(data) as V; }, undefined, - `Error getting key ${String(key)} from Redis` + { + timeout: this.timeout, + shouldProceed: () => this.client.isOpen, + getContext: () => `getting key ${String(key)} from Redis`, + } ); } async set(key: K, value: V, ttl: number): Promise { - if (ttl === 0) { + 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(); + } + } + + 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 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` - ); + 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, + time: 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 this.withTimeout( + await withTimeout( async () => { // Get current TTL const ttl = await this.client.ttl(redisKey); @@ -235,12 +271,16 @@ export class RedisCacheBackend implements CacheBackend { return true; }, false, - `Error updating key ${String(key)} in Redis` + { + timeout: this.timeout, + shouldProceed: () => this.client.isOpen, + getContext: () => `updating key ${String(key)} in Redis`, + } ); } async clear(): Promise { - await this.withTimeout( + await withTimeout( async () => { // Delete all keys with this prefix const keys = await this.client.keys(`${this.prefix}*`); @@ -250,18 +290,26 @@ export class RedisCacheBackend implements CacheBackend { return true; }, false, - `Error clearing Redis cache` + { + timeout: this.timeout, + shouldProceed: () => this.client.isOpen, + getContext: () => 'clearing Redis cache', + } ); } async getTTL(key: K): Promise { - return this.withTimeout( + return 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` + { + timeout: this.timeout, + shouldProceed: () => this.client.isOpen, + getContext: () => `getting TTL for key ${String(key)} from Redis`, + } ); } @@ -279,6 +327,13 @@ export class SQLCacheBackend implements CacheBackend { private maxSize: number; 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 = '', maxSize: number = Env.DEFAULT_MAX_CACHE_SIZE @@ -287,6 +342,84 @@ export class SQLCacheBackend implements CacheBackend { this.prefix = prefix; this.maxSize = maxSize; this.startMaintenance(); + SQLCacheBackend.startFlushInterval(); + } + + 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(); + + const db = DB.getInstance(); + + const start = Date.now(); + + try { + const countResult = await db.query('SELECT COUNT(*) as count FROM cache'); + let currentSize = countResult[0].count; + const overflow = + currentSize + bufferToFlush.size - Env.DEFAULT_MAX_CACHE_SIZE; + + if (overflow > 0) { + logger.debug(`Cache overflow detected. Evicting ${overflow} items.`); + const limit = Math.ceil(overflow); + if (db.isSQLite()) { + await db.execute( + `DELETE FROM cache WHERE key IN (SELECT key FROM cache ORDER BY last_accessed ASC LIMIT ${limit})` + ); + } else { + await db.execute( + `DELETE FROM cache WHERE ctid IN (SELECT ctid FROM cache ORDER BY last_accessed ASC LIMIT ${limit})` + ); + } + } + + // Prepare for batch upsert + const values: any[] = []; + const now = Date.now(); + for (const [key, item] of bufferToFlush.entries()) { + values.push(key, JSON.stringify(item.value), now + item.ttl * 1000); + } + + if (values.length === 0) return; + + if (db.isSQLite()) { + const placeholders = Array(bufferToFlush.size) + .fill('(?, ?, ?)') + .join(', '); + const sql = `INSERT OR REPLACE INTO cache (key, value, expires_at) VALUES ${placeholders}`; + await db.execute(sql, values); + } else { + const placeholders = Array(bufferToFlush.size) + .fill('(?, ?, ?)') + .join(', '); + const timestampFunc = 'NOW()'; + const sql = `INSERT INTO cache (key, value, expires_at) VALUES ${placeholders} ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, expires_at = EXCLUDED.expires_at, last_accessed = ${timestampFunc}`; + await db.execute(sql, values); + } + logger.debug('Flushed SQL write buffer', { + items: bufferToFlush.size, + time: 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 startMaintenance() { @@ -366,47 +499,13 @@ export class SQLCacheBackend implements CacheBackend { if (ttl === 0) return; const sqlKey = this.getKey(key); - const expiresAt = Date.now() + ttl * 1000; - const jsonValue = JSON.stringify(value); + SQLCacheBackend.writeBuffer.set(sqlKey, { + value: structuredClone(value), + ttl, + }); - try { - // Check current cache size - const countResult = await this.db.query( - 'SELECT COUNT(*) as count FROM cache' - ); - const currentSize = countResult[0].count; - - if (currentSize >= this.maxSize) { - // Remove oldest accessed entry - if (this.db.isSQLite()) { - await this.db.execute( - 'DELETE FROM cache WHERE key IN (SELECT key FROM cache ORDER BY last_accessed ASC LIMIT 1)' - ); - } else { - // PostgreSQL compatible version - await this.db.execute( - 'DELETE FROM cache WHERE key = (SELECT key FROM cache ORDER BY last_accessed ASC LIMIT 1)' - ); - } - } - - // Upsert the new value - if (this.db.isSQLite()) { - await this.db.execute( - 'INSERT OR REPLACE INTO cache (key, value, expires_at) VALUES (?, ?, ?)', - [sqlKey, jsonValue, expiresAt] - ); - } else { - const timestampFunc = this.db.isSQLite() - ? 'CURRENT_TIMESTAMP' - : 'NOW()'; - await this.db.execute( - `INSERT INTO cache (key, value, expires_at) VALUES (?, ?, ?) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, expires_at = EXCLUDED.expires_at, last_accessed = ${timestampFunc}`, - [sqlKey, jsonValue, expiresAt] - ); - } - } catch (err) { - logger.error(`Error setting key ${String(key)} in SQL cache: ${err}`); + if (SQLCacheBackend.writeBuffer.size >= SQLCacheBackend.batchSize) { + SQLCacheBackend.flushWriteBuffer(); } } diff --git a/packages/core/src/utils/env.ts b/packages/core/src/utils/env.ts index e72a1f09..01593cec 100644 --- a/packages/core/src/utils/env.ts +++ b/packages/core/src/utils/env.ts @@ -1601,10 +1601,10 @@ export const Env = cleanEnv(process.env, { default: 60 * 60, // 1 hour desc: 'Builtin Debrid playback link cache TTL', }), - BUILTIN_PLAYBACK_LINK_STORE: str({ - choices: ['redis', 'sql'], - default: 'sql', - desc: 'Builtin Debrid playback link store', + BUILTIN_DEBRID_METADATA_STORE: str({ + choices: ['redis', 'sql', 'memory'], + default: 'memory', + desc: 'Builtin Debrid metadata store', }), BUILTIN_PLAYBACK_LINK_VALIDITY: num({ default: 1 * 24 * 60 * 60, // 1 day diff --git a/packages/core/src/utils/general.ts b/packages/core/src/utils/general.ts index 87610302..f76bba41 100644 --- a/packages/core/src/utils/general.ts +++ b/packages/core/src/utils/general.ts @@ -84,6 +84,63 @@ export async function withRetry( throw new Error('Unexpected state in retry logic'); } +export interface TimeoutOptions { + /** + * Timeout duration in milliseconds + * @default 5000 + */ + timeout?: number; + /** + * Optional function to check if the operation should be allowed to proceed + * @returns true if operation should proceed, false otherwise + */ + shouldProceed?: () => boolean; + /** + * Optional function to get context for error logging + * @returns string context to include in error logs + */ + getContext?: () => string; +} + +/** + * Utility function to execute an async operation with timeout + * @param operation The async operation to execute + * @param fallback Value to return if operation times out or fails + * @param options Timeout configuration options + * @returns The result of the operation or fallback value + */ +export async function withTimeout( + operation: () => Promise, + fallback: T, + options: TimeoutOptions = {} +): Promise { + const { timeout = 5000, shouldProceed, getContext } = options; + + // Check if operation should proceed + if (shouldProceed && !shouldProceed()) { + const context = getContext ? ` for ${getContext()}` : ''; + logger.error(`Operation skipped${context}: Precondition check failed`); + return fallback; + } + + try { + // Create a promise that rejects after timeout + const timeoutPromise = new Promise((_, reject) => { + const id = setTimeout(() => { + clearTimeout(id); + reject(new Error(`Operation timed out after ${timeout}ms`)); + }, timeout); + }); + + // Race the operation against the timeout + return await Promise.race([operation(), timeoutPromise]); + } catch (err) { + const context = getContext ? ` for ${getContext()}` : ''; + logger.error(`Operation failed${context}: ${err}`); + return fallback; + } +} + /** * Base64 URL safe encoding * @param data - The data to encode diff --git a/packages/server/src/routes/api/debrid.ts b/packages/server/src/routes/api/debrid.ts index 1e750894..05e5ea2b 100644 --- a/packages/server/src/routes/api/debrid.ts +++ b/packages/server/src/routes/api/debrid.ts @@ -13,7 +13,10 @@ import { PlaybackInfo, ServiceAuth, decryptString, - pbiCache, + metadataStore, + TitleMetadata, + FileInfoSchema, + getSimpleTextHash, } from '@aiostreams/core'; import { ZodError } from 'zod'; import { StaticFiles } from '../../app.js'; @@ -30,18 +33,27 @@ router.use((req: Request, res: Response, next: NextFunction) => { }); router.get( - '/playback/:encryptedStoreAuth/:playbackId/:filename', + '/playback/:encryptedStoreAuth/:fileInfo/:metadataId/:filename', async (req: Request, res: Response, next: NextFunction) => { try { - const { encryptedStoreAuth, playbackId, filename } = req.params; - if (!playbackId || !filename) { + const { + encryptedStoreAuth, + fileInfo: encodedFileInfo, + metadataId, + filename, + } = req.params; + if (!encodedFileInfo || !metadataId || !filename) { throw new APIError( constants.ErrorCode.BAD_REQUEST, undefined, - 'Encrypted store auth, playback info and filename are required' + 'Encrypted store auth, file info, metadata id and filename are required' ); } + const fileInfo = FileInfoSchema.parse( + JSON.parse(Buffer.from(encodedFileInfo, 'base64').toString('utf-8')) + ); + const decryptedStoreAuth = decryptString(encryptedStoreAuth); if (!decryptedStoreAuth.success) { throw new APIError( @@ -54,15 +66,37 @@ router.get( const storeAuth = ServiceAuthSchema.parse( JSON.parse(decryptedStoreAuth.data) ); - const playbackInfo = await pbiCache().get(playbackId); - if (!playbackInfo) { + const metadata: TitleMetadata | undefined = + await metadataStore().get(metadataId); + if (!metadata) { throw new APIError( constants.ErrorCode.BAD_REQUEST, undefined, - 'Playback info not found' + 'Metadata not found' ); } + logger.verbose(`Got metadata: ${JSON.stringify(metadata)}`); + + const playbackInfo: PlaybackInfo = + fileInfo.type === 'torrent' + ? { + type: 'torrent', + metadata: metadata, + hash: fileInfo.hash, + sources: fileInfo.sources, + index: fileInfo.index, + filename: filename, + } + : { + type: 'usenet', + metadata: metadata, + hash: fileInfo.hash, + nzb: fileInfo.nzb, + index: fileInfo.index, + filename: filename, + }; + const debridInterface = getDebridService( storeAuth.id, storeAuth.credential,