feat: introduce lazy extract for all non-HLS streams

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';

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<string, Mutex>();
+
+  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);
+    }
+  };
+}

src/controller/index.ts

@@ -1,3 +1,4 @@
+export * from './ExtractController';
 export * from './ConfigureController';
 export * from './ManifestController';
 export * from './StreamController';

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<UrlResult[]> {
+  public async handle(ctx: Context, url: URL, meta?: Meta, allowLazy?: boolean): Promise<UrlResult[]> {
     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<UrlResult[]>(normalizedUrl.href) ?? [];
 
-    const cachedMeta = await this.metaCache.get<Meta>(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<UrlResult[]>(cacheKey, urlResults, 43200000); // 12h
     } else if (!urlResults.some(urlResult => urlResult.error) && extractor.ttl) {
       await this.urlResultCache.set<UrlResult[]>(cacheKey, urlResults, extractor.ttl);
-    }
-
-    if (urlResults.length) {
-      await this.metaCache.set<Meta>(normalizedUrl.href, urlResults[0]?.meta as Meta, 2628000); // 1 month
+      await this.lazyUrlResultCache.set<UrlResult[]>(normalizedUrl.href, urlResults, 2628000); // 1 month
+    } else {
+      await this.lazyUrlResultCache.delete(normalizedUrl.href);
     }
 
     return urlResults;

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);
 

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());