diff --git a/apps/electron-backend-e2e/src/series-download-queue.e2e.ts b/apps/electron-backend-e2e/src/series-download-queue.e2e.ts index fe4c9fffa..8236bdcbd 100644 --- a/apps/electron-backend-e2e/src/series-download-queue.e2e.ts +++ b/apps/electron-backend-e2e/src/series-download-queue.e2e.ts @@ -26,10 +26,38 @@ const expectedEpisodeIds = [80_000, 80_001, 80_002, 80_003]; type EpisodeQueueRow = { episodeNumber: number | null; + seriesXtreamId: number | null; status: string; xtreamId: number; }; +const expectedEpisodeQueue: EpisodeQueueRow[] = [ + { + episodeNumber: 1, + seriesXtreamId: 30_000, + status: 'downloading', + xtreamId: expectedEpisodeIds[0], + }, + { + episodeNumber: 2, + seriesXtreamId: 30_000, + status: 'queued', + xtreamId: expectedEpisodeIds[1], + }, + { + episodeNumber: 3, + seriesXtreamId: 30_000, + status: 'queued', + xtreamId: expectedEpisodeIds[2], + }, + { + episodeNumber: 4, + seriesXtreamId: 30_000, + status: 'queued', + xtreamId: expectedEpisodeIds[3], + }, +]; + type SeriesFixture = Awaited>; function waitForQueueStabilityInterval(): Promise { @@ -49,28 +77,18 @@ function readFirstSeriesTitle(fixture: SeriesFixture): string { return title; } -async function readEpisodeQueue( - page: Page, - episodeIds: readonly number[] -): Promise { - return page.evaluate( - async (expectedIds) => { - const downloads = - (await window.electron?.downloadsGetList?.()) ?? []; - return downloads - .filter( - (download) => - download.contentType === 'episode' && - expectedIds.includes(download.xtreamId) - ) - .map((download) => ({ - episodeNumber: download.episodeNumber ?? null, - status: download.status, - xtreamId: download.xtreamId, - })); - }, - [...episodeIds] - ); +async function readEpisodeQueue(page: Page): Promise { + return page.evaluate(async () => { + const downloads = (await window.electron?.downloadsGetList?.()) ?? []; + return downloads + .filter((download) => download.contentType === 'episode') + .map((download) => ({ + episodeNumber: download.episodeNumber ?? null, + seriesXtreamId: download.seriesXtreamId ?? null, + status: download.status, + xtreamId: download.xtreamId, + })); + }); } test.describe('Electron Series Download Queue', () => { @@ -137,22 +155,10 @@ test.describe('Electron Series Download Queue', () => { await secondEpisodeAction.click(); await expect - .poll( - () => readEpisodeQueue(app.mainWindow, expectedEpisodeIds), - { timeout: 20_000 } - ) - .toEqual([ - { - episodeNumber: 1, - status: 'downloading', - xtreamId: expectedEpisodeIds[0], - }, - { - episodeNumber: 2, - status: 'queued', - xtreamId: expectedEpisodeIds[1], - }, - ]); + .poll(() => readEpisodeQueue(app.mainWindow), { + timeout: 20_000, + }) + .toEqual(expectedEpisodeQueue.slice(0, 2)); const seasonAction = app.mainWindow.getByTestId('download-season'); await expect(seasonAction).toBeVisible(); @@ -166,42 +172,17 @@ test.describe('Electron Series Download Queue', () => { ).toBeVisible({ timeout: 20_000 }); await expect - .poll( - async () => - ( - await readEpisodeQueue( - app.mainWindow, - expectedEpisodeIds - ) - ).map((row) => row.xtreamId), - { timeout: 20_000 } - ) - .toEqual(expectedEpisodeIds); + .poll(() => readEpisodeQueue(app.mainWindow), { + timeout: 20_000, + }) + .toEqual(expectedEpisodeQueue); - const queuedSeason = await readEpisodeQueue( - app.mainWindow, - expectedEpisodeIds - ); - expect(new Set(queuedSeason.map((row) => row.xtreamId)).size).toBe( - 4 - ); - expect(queuedSeason.map((row) => row.status)).toEqual([ - 'downloading', - 'queued', - 'queued', - 'queued', - ]); await expect(seasonAction).toBeDisabled(); await expect(seasonAction).toContainText('Download season (0)'); await waitForQueueStabilityInterval(); - const stableQueue = await readEpisodeQueue( - app.mainWindow, - expectedEpisodeIds - ); - expect(stableQueue).toHaveLength(4); - expect(new Set(stableQueue.map((row) => row.xtreamId)).size).toBe( - 4 + expect(await readEpisodeQueue(app.mainWindow)).toEqual( + expectedEpisodeQueue ); } finally { await closeElectronApp(app); diff --git a/apps/xtream-mock-server/src/app/server.spec.ts b/apps/xtream-mock-server/src/app/server.spec.ts index f38feccfe..209d53a21 100644 --- a/apps/xtream-mock-server/src/app/server.spec.ts +++ b/apps/xtream-mock-server/src/app/server.spec.ts @@ -1,9 +1,14 @@ import { connect } from 'node:net'; +import express from 'express'; import { resetAll } from './data-store.js'; import { createXtreamMockApp, parseXtreamMockServerEnvironment, } from './server.js'; +import { + type SlowSeriesDownloadOptions, + streamSlowSeriesDownload, +} from './slow-series-download.js'; import { startLoopbackServer } from './testing/http-server.fixture.js'; jest.mock('@faker-js/faker', () => { @@ -224,6 +229,64 @@ describe('Xtream mock server factory', () => { }); }); +describe('Slow series download stream', () => { + it('completes the configured byte count over a loopback response', async () => { + const options = { + chunkSize: 1_024, + intervalMs: 1, + totalBytes: 10 * 1_024 + 7, + } satisfies SlowSeriesDownloadOptions; + const running = await startLoopbackServer( + createSlowSeriesDownloadApp(options) + ); + let closed = false; + + try { + const response = await fetch(`${running.origin}/slow-series`); + const body = await response.arrayBuffer(); + + expect(response.status).toBe(200); + expect(response.headers.get('content-type')).toContain('video/mp4'); + expect(Number(response.headers.get('content-length'))).toBe( + options.totalBytes + ); + expect(body.byteLength).toBe(options.totalBytes); + + await within(running.close(), 1_000); + closed = true; + } finally { + if (!closed) await running.close().catch(() => undefined); + } + }); + + it('stops a longer recursive stream after the response body is cancelled', async () => { + const options = { + chunkSize: 1_024, + intervalMs: 25, + totalBytes: 64 * 1_024, + } satisfies SlowSeriesDownloadOptions; + const running = await startLoopbackServer( + createSlowSeriesDownloadApp(options) + ); + let closed = false; + + try { + const response = await fetch(`${running.origin}/slow-series`); + const reader = response.body?.getReader(); + if (!reader) throw new Error('Expected a streaming response body.'); + + const firstChunk = await within(reader.read(), 1_000); + expect(firstChunk.done).toBe(false); + expect(firstChunk.value?.byteLength).toBeGreaterThan(0); + await within(reader.cancel(), 1_000); + await within(running.close(), 1_000); + closed = true; + } finally { + if (!closed) await running.close().catch(() => undefined); + } + }); +}); + describe('Xtream mock environment parsing', () => { it('uses safe defaults and enables control only for the exact flag', () => { expect(parseXtreamMockServerEnvironment({})).toEqual({ @@ -303,3 +366,28 @@ async function rawLoopbackGet( }); }); } + +function createSlowSeriesDownloadApp( + options: SlowSeriesDownloadOptions +): express.Express { + const app = express(); + app.get('/slow-series', (request, response) => { + streamSlowSeriesDownload(request, response, options); + }); + return app; +} + +async function within(promise: Promise, timeoutMs: number): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error(`Operation exceeded ${timeoutMs}ms.`)), + timeoutMs + ); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + if (timer) clearTimeout(timer); + } +} diff --git a/apps/xtream-mock-server/src/app/server.ts b/apps/xtream-mock-server/src/app/server.ts index f1d4e0f1d..0525f8a46 100644 --- a/apps/xtream-mock-server/src/app/server.ts +++ b/apps/xtream-mock-server/src/app/server.ts @@ -15,6 +15,7 @@ import { } from './performance-control.js'; import { dispatchAction } from './routes/dispatch.js'; import { getScenario } from './scenarios.js'; +import { streamSlowSeriesDownload } from './slow-series-download.js'; export { createXtreamMockServerShutdown } from './server-lifecycle.js'; @@ -47,9 +48,6 @@ const DEFAULT_NORMAL_HOST = '0.0.0.0'; const DEFAULT_CONTROL_HOST = '127.0.0.1'; const PERFORMANCE_USERNAME = 'performance'; const PERFORMANCE_PASSWORD = 'performance'; -const DOWNLOAD_STREAM_BYTES = 8 * 1024 * 1024; -const DOWNLOAD_STREAM_CHUNK_INTERVAL_MS = 100; -const DOWNLOAD_STREAM_CHUNK = Buffer.alloc(32 * 1024); const marketingRasterAssetRoot = join( process.cwd(), 'apps/xtream-mock-server/public/marketing' @@ -282,60 +280,6 @@ function isSlowSeriesDownloadRequest(request: Request): boolean { ); } -function streamSlowSeriesDownload(request: Request, response: Response): void { - let sentBytes = 0; - let stopped = false; - let timer: NodeJS.Timeout | undefined; - - const stop = () => { - stopped = true; - if (timer) { - clearTimeout(timer); - timer = undefined; - } - response.off('drain', scheduleChunk); - }; - const scheduleChunk = () => { - if (!stopped && !timer) { - timer = setTimeout(writeChunk, DOWNLOAD_STREAM_CHUNK_INTERVAL_MS); - } - }; - const writeChunk = () => { - timer = undefined; - if (stopped || response.destroyed || response.writableEnded) { - stop(); - return; - } - - const remainingBytes = DOWNLOAD_STREAM_BYTES - sentBytes; - const chunk = - remainingBytes >= DOWNLOAD_STREAM_CHUNK.length - ? DOWNLOAD_STREAM_CHUNK - : DOWNLOAD_STREAM_CHUNK.subarray(0, remainingBytes); - sentBytes += chunk.length; - const canContinue = response.write(chunk); - - if (sentBytes >= DOWNLOAD_STREAM_BYTES) { - response.end(); - } else if (canContinue) { - scheduleChunk(); - } else { - response.once('drain', scheduleChunk); - } - }; - - request.once('close', stop); - response.once('close', stop); - response.once('finish', stop); - response - .status(200) - .type('video/mp4') - .set('Content-Length', String(DOWNLOAD_STREAM_BYTES)) - .set('Cache-Control', 'no-store') - .flushHeaders(); - scheduleChunk(); -} - function isPerformanceMediaRequest( request: Request, controlEnabled: boolean diff --git a/apps/xtream-mock-server/src/app/slow-series-download.ts b/apps/xtream-mock-server/src/app/slow-series-download.ts new file mode 100644 index 000000000..2a74b0cf9 --- /dev/null +++ b/apps/xtream-mock-server/src/app/slow-series-download.ts @@ -0,0 +1,93 @@ +import type { Request, Response } from 'express'; + +export interface SlowSeriesDownloadOptions { + readonly chunkSize: number; + readonly intervalMs: number; + readonly totalBytes: number; +} + +export const DEFAULT_SLOW_SERIES_DOWNLOAD_OPTIONS: SlowSeriesDownloadOptions = { + chunkSize: 32 * 1024, + intervalMs: 100, + totalBytes: 8 * 1024 * 1024, +}; + +export function streamSlowSeriesDownload( + request: Request, + response: Response, + options: SlowSeriesDownloadOptions = DEFAULT_SLOW_SERIES_DOWNLOAD_OPTIONS +): void { + validateOptions(options); + const chunkBuffer = Buffer.alloc(options.chunkSize); + let sentBytes = 0; + let stopped = false; + let timer: NodeJS.Timeout | undefined; + + function stop(): void { + if (stopped) return; + stopped = true; + if (timer) { + clearTimeout(timer); + timer = undefined; + } + request.off('close', stop); + response.off('close', stop); + response.off('finish', stop); + response.off('drain', scheduleChunk); + } + + function scheduleChunk(): void { + if (!stopped && !timer) { + timer = setTimeout(writeChunk, options.intervalMs); + } + } + + function writeChunk(): void { + timer = undefined; + if (stopped || response.destroyed || response.writableEnded) { + stop(); + return; + } + + const remainingBytes = options.totalBytes - sentBytes; + const chunk = + remainingBytes >= chunkBuffer.length + ? chunkBuffer + : chunkBuffer.subarray(0, remainingBytes); + sentBytes += chunk.length; + const canContinue = response.write(chunk); + + if (sentBytes >= options.totalBytes) { + response.end(); + } else if (canContinue) { + scheduleChunk(); + } else { + response.once('drain', scheduleChunk); + } + } + + request.once('close', stop); + response.once('close', stop); + response.once('finish', stop); + response + .status(200) + .type('video/mp4') + .set('Content-Length', String(options.totalBytes)) + .set('Cache-Control', 'no-store') + .flushHeaders(); + scheduleChunk(); +} + +function validateOptions(options: SlowSeriesDownloadOptions): void { + if (!Number.isSafeInteger(options.totalBytes) || options.totalBytes <= 0) { + throw new Error('Slow download totalBytes must be a positive integer.'); + } + if (!Number.isSafeInteger(options.chunkSize) || options.chunkSize <= 0) { + throw new Error('Slow download chunkSize must be a positive integer.'); + } + if (!Number.isSafeInteger(options.intervalMs) || options.intervalMs < 0) { + throw new Error( + 'Slow download intervalMs must be a non-negative integer.' + ); + } +}