diff --git a/jest.setup.ts b/jest.setup.ts index ddf8cc8..2387c1c 100644 --- a/jest.setup.ts +++ b/jest.setup.ts @@ -1,4 +1,6 @@ process.env['CACHE_DIR'] = '/dev/null'; +process.env['HOST'] = 'example.test'; +process.env['PROTOCOL'] = 'https'; if (!process.env['TMDB_ACCESS_TOKEN']) { process.env['TMDB_ACCESS_TOKEN'] = 'some access token'; diff --git a/src/controller/ExtractController.ts b/src/controller/ExtractController.ts new file mode 100644 index 0000000..87abcf4 --- /dev/null +++ b/src/controller/ExtractController.ts @@ -0,0 +1,53 @@ +import { Mutex } from 'async-mutex'; +import { Request, Response, Router } from 'express'; +import winston from 'winston'; +import { ExtractorRegistry } from '../extractor'; +import { contextFromRequestAndResponse, Fetcher } from '../utils'; + +export class ExtractController { + public readonly router: Router; + + private readonly logger: winston.Logger; + private readonly extractorRegistry: ExtractorRegistry; + + private readonly locks = new Map(); + + public constructor(logger: winston.Logger, _fetcher: Fetcher, extractorRegistry: ExtractorRegistry) { + this.router = Router(); + + this.logger = logger; + this.extractorRegistry = extractorRegistry; + + this.router.get('/extract', this.extract.bind(this)); + } + + private async extract(req: Request, res: Response) { + if (req.method !== 'GET') { + res.status(405).send('Method Not Allowed'); + return; + } + + const ctx = contextFromRequestAndResponse(req, res); + + const index = parseInt(req.query['index'] as string); + const url = new URL(req.query['url'] as string); + + this.logger.info(`Lazy extract index ${index} of URL ${url} for ip ${ctx.ip}`, ctx); + + let mutex = this.locks.get(url.href); + if (!mutex) { + mutex = new Mutex(); + this.locks.set(url.href, mutex); + } + + await mutex.runExclusive(async () => { + const urlResults = await this.extractorRegistry.handle(ctx, url); + + res.redirect(urlResults[index]?.url.href as string); + }); + + if (!mutex.isLocked()) { + this.locks.delete(url.href); + } + }; +} diff --git a/src/controller/index.ts b/src/controller/index.ts index 95a75d3..905d4bd 100644 --- a/src/controller/index.ts +++ b/src/controller/index.ts @@ -1,3 +1,4 @@ +export * from './ExtractController'; export * from './ConfigureController'; export * from './ManifestController'; export * from './StreamController'; diff --git a/src/extractor/ExtractorRegistry.ts b/src/extractor/ExtractorRegistry.ts index be3e0df..f93bf38 100644 --- a/src/extractor/ExtractorRegistry.ts +++ b/src/extractor/ExtractorRegistry.ts @@ -2,32 +2,31 @@ import KeyvSqlite from '@keyv/sqlite'; import { Cacheable, CacheableMemory, Keyv } from 'cacheable'; import winston from 'winston'; -import { Context, Meta, UrlResult } from '../types'; -import { getCacheDir, isExtractorDisabled } from '../utils'; +import { Context, Format, Meta, UrlResult } from '../types'; +import { envGet, getCacheDir, isExtractorDisabled } from '../utils'; import { Extractor } from './Extractor'; export class ExtractorRegistry { private readonly logger: winston.Logger; private readonly extractors: Extractor[]; - private readonly metaCache: Cacheable; private readonly urlResultCache: Cacheable; + private readonly lazyUrlResultCache: Cacheable; public constructor(logger: winston.Logger, extractors: Extractor[]) { this.logger = logger; this.extractors = extractors; - this.metaCache = new Cacheable({ - nonBlocking: true, - primary: new Keyv({ store: new CacheableMemory({ lruSize: 16384 }) }), - secondary: new Keyv(new KeyvSqlite(`sqlite://${getCacheDir()}/webstreamr-meta-cache.sqlite`)), - }); this.urlResultCache = new Cacheable({ nonBlocking: true, primary: new Keyv({ store: new CacheableMemory({ lruSize: 1024 }) }), secondary: new Keyv(new KeyvSqlite(`sqlite://${getCacheDir()}/webstreamr-extractor-cache.sqlite`)), stats: true, }); + this.lazyUrlResultCache = new Cacheable({ + primary: new Keyv({ store: new CacheableMemory({ lruSize: 1024 }) }), + secondary: new Keyv(new KeyvSqlite(`sqlite://${getCacheDir()}/webstreamr-extractor-lazy-cache.sqlite`)), + }); } public stats() { @@ -36,7 +35,7 @@ export class ExtractorRegistry { }; }; - public async handle(ctx: Context, url: URL, meta?: Meta): Promise { + public async handle(ctx: Context, url: URL, meta?: Meta, allowLazy?: boolean): Promise { const extractor = this.extractors.find(extractor => !isExtractorDisabled(ctx.config, extractor) && extractor.supports(ctx, url)); if (!extractor) { return []; @@ -60,16 +59,32 @@ export class ExtractorRegistry { } } - this.logger.info(`Extract ${url} using ${extractor.id} extractor`, ctx); + const lazyUrlResults = await this.lazyUrlResultCache.get(normalizedUrl.href) ?? []; - const cachedMeta = await this.metaCache.get(normalizedUrl.href); + /* istanbul ignore next */ + if ( + lazyUrlResults.length && allowLazy && !extractor.viaMediaFlowProxy && extractor.id !== 'external' + && lazyUrlResults.every(urlResult => urlResult.format !== Format.hls) // related to Android issues, e.g. https://github.com/Stremio/stremio-bugs/issues/1574 or https://github.com/Stremio/stremio-bugs/issues/1579 + ) { + // generate lazy extract urls + return lazyUrlResults.map((urlResult, index) => { + const extractUrl = new URL(`${envGet('PROTOCOL')}:${envGet('HOST')}/extract/`); + + extractUrl.searchParams.set('index', `${index}`); + extractUrl.searchParams.set('url', url.href); + + return { ...urlResult, url: extractUrl }; + }); + } + + this.logger.info(`Extract ${url} using ${extractor.id} extractor`, ctx); const urlResults = await extractor.extract( ctx, normalizedUrl, { ...meta, - ...cachedMeta, + ...lazyUrlResults[0]?.meta, extractorId: meta?.extractorId ?? extractor.id, }, ); @@ -78,10 +93,9 @@ export class ExtractorRegistry { await this.urlResultCache.set(cacheKey, urlResults, 43200000); // 12h } else if (!urlResults.some(urlResult => urlResult.error) && extractor.ttl) { await this.urlResultCache.set(cacheKey, urlResults, extractor.ttl); - } - - if (urlResults.length) { - await this.metaCache.set(normalizedUrl.href, urlResults[0]?.meta as Meta, 2628000); // 1 month + await this.lazyUrlResultCache.set(normalizedUrl.href, urlResults, 2628000); // 1 month + } else { + await this.lazyUrlResultCache.delete(normalizedUrl.href); } return urlResults; diff --git a/src/index.ts b/src/index.ts index fca5989..290eae3 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,7 +4,7 @@ import { setupCache } from 'axios-cache-interceptor'; import axiosRetry from 'axios-retry'; import express, { NextFunction, Request, Response } from 'express'; import winston from 'winston'; -import { ConfigureController, ManifestController, StreamController } from './controller'; +import { ConfigureController, ExtractController, ManifestController, StreamController } from './controller'; import { BlockedError, logErrorAndReturnNiceString } from './error'; import { createExtractors, ExtractorRegistry } from './extractor'; import { createSources, Source } from './source'; @@ -47,7 +47,10 @@ const extractors = createExtractors(fetcher); const addon = express(); addon.set('trust proxy', true); -addon.use((_req: Request, res: Response, next: NextFunction) => { +addon.use((req: Request, res: Response, next: NextFunction) => { + process.env['HOST'] = req.host; + process.env['PROTOCOL'] = req.protocol; + res.setHeader('X-Request-ID', randomUUID()); res.setHeader('Access-Control-Allow-Origin', '*'); @@ -60,10 +63,12 @@ addon.use((_req: Request, res: Response, next: NextFunction) => { next(); }); +const extractorRegistry = new ExtractorRegistry(logger, extractors); + +addon.use('/', (new ExtractController(logger, fetcher, extractorRegistry)).router); addon.use('/', (new ConfigureController(sources, extractors)).router); addon.use('/', (new ManifestController(sources, extractors)).router); -const extractorRegistry = new ExtractorRegistry(logger, extractors); const streamResolver = new StreamResolver(logger, extractorRegistry); addon.use('/', (new StreamController(logger, sources, streamResolver)).router); diff --git a/src/utils/StreamResolver.ts b/src/utils/StreamResolver.ts index 83a01e9..8955c43 100644 --- a/src/utils/StreamResolver.ts +++ b/src/utils/StreamResolver.ts @@ -50,7 +50,7 @@ export class StreamResolver { const sourceResults = await source.handle(ctx, type, id); const sourceUrlResults = await Promise.all( - sourceResults.map(({ url, meta }) => this.extractorRegistry.handle(ctx, url, { ...meta, sourceLabel: source.label, sourceId: source.id })), + sourceResults.map(({ url, meta }) => this.extractorRegistry.handle(ctx, url, { ...meta, sourceLabel: source.label, sourceId: source.id }, true)), ); urlResults.push(...sourceUrlResults.flat());