feat(downloads): apply bandwidth and schedule policy

src/main/domain/download-policy-integration.test.ts

@@ -0,0 +1,59 @@
+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { describe, expect, it } from 'vitest';
+import { decideDownloadStart, normalizeDownloadPolicy } from './download-policy';
+
+describe('download policy integration contract', () => {
+    it('keeps a normalized persisted policy after a config-shaped restart payload', () => {
+        const persisted = JSON.parse(JSON.stringify({
+            download_policy: {
+                throttle: { maxBytesPerSecond: 1_572_864 },
+                windows: [{ start: '22:00', end: '06:00' }, { start: '22:00', end: '06:00' }, { start: 'bad', end: '12:00' }]
+            }
+        }));
+
+        expect(normalizeDownloadPolicy(persisted.download_policy)).toEqual({
+            throttle: { maxBytesPerSecond: 1_572_864 },
+            windows: [{ start: '22:00', end: '06:00' }]
+        });
+
+        const mainSource = readFileSync(join(process.cwd(), 'src', 'main.ts'), 'utf8');
+        expect(mainSource).toContain('download_policy: { throttle: null, windows: [] }');
+        expect(mainSource).toContain('download_policy: normalizeDownloadPolicy(input.download_policy)');
+    });
+
+    it('blocks automatic queue starts outside the local window but allows a manual override', () => {
+        const policy = normalizeDownloadPolicy({
+            throttle: { maxBytesPerSecond: 1_048_576 },
+            windows: [{ start: '22:00', end: '06:00' }]
+        });
+        const now = new Date(2026, 0, 13, 13, 0);
+
+        expect(decideDownloadStart(policy, now)).toMatchObject({
+            allowed: false,
+            reason: 'outside-window',
+            nextStart: new Date(2026, 0, 13, 22, 0)
+        });
+        expect(decideDownloadStart(policy, now, true)).toMatchObject({
+            allowed: true,
+            reason: 'manual-override',
+            maxBytesPerSecond: 1_048_576
+        });
+
+        const mainSource = readFileSync(join(process.cwd(), 'src', 'main.ts'), 'utf8');
+        expect(mainSource).toContain('function scheduleQueueProcessing(manualOverride = false)');
+        expect(mainSource).toContain('scheduleQueueProcessing(manualOverride === true)');
+        expect(mainSource).toContain('scheduleDownloadPolicyWake(decision.nextStart)');
+    });
+
+    it('uses the app-side stdout transform and retains the existing Streamlink argument pipeline', () => {
+        const source = readFileSync(join(process.cwd(), 'src', 'main.ts'), 'utf8');
+        const start = source.indexOf('function downloadVODPart(');
+        const end = source.indexOf('const outputFinished = output.finished', start);
+        const section = source.slice(start, end);
+
+        expect(section).toContain('createTokenBucketTransform');
+        expect(section).toContain("const args = [...streamlinkCmd.prefixArgs, url, getStreamlinkStreamArg(), '--stdout'];");
+        expect(section).not.toMatch(/args\.push\([^\n]*(?:bandwidth|rate-limit|max-rate|throttle)/i);
+    });
+});

src/main/domain/pausable-output.test.ts

@@ -1,6 +1,7 @@
 import { PassThrough, Writable } from 'stream';
 import { describe, expect, it } from 'vitest';
 import { createPausableOutput } from './pausable-output';
+import { createTokenBucketTransform } from './token-bucket-transform';
 
 function waitForTurn(): Promise<void> {
     return new Promise((resolve) => setImmediate(resolve));
@@ -86,4 +87,23 @@ describe('createPausableOutput', () => {
         expect(target.destroyed).toBe(true);
         expect(Buffer.concat(chunks).toString()).toBe('behalten');
     });
+
+    it('bricht eine wartende app-seitige Drosselung zusammen mit dem Ausgabestrom ab', async () => {
+        const source = new PassThrough();
+        const target = new Writable({
+            write(_chunk, _encoding, callback) {
+                callback();
+            }
+        });
+        const throttle = createTokenBucketTransform(1);
+        const output = createPausableOutput(source, target, throttle);
+
+        source.write('a');
+        source.write('b');
+        await output.cancel();
+
+        expect(source.destroyed).toBe(true);
+        expect(target.destroyed).toBe(true);
+        expect(throttle.destroyed).toBe(true);
+    });
 });

src/main/domain/pausable-output.ts

@@ -1,4 +1,4 @@
-import { Readable, Writable } from 'stream';
+import { Readable, Transform, Writable } from 'stream';
 
 export interface PausableOutput {
     pause(): void;
@@ -8,7 +8,7 @@ export interface PausableOutput {
     finished: Promise<void>;
 }
 
-export function createPausableOutput(source: Readable, target: Writable): PausableOutput {
+export function createPausableOutput(source: Readable, target: Writable, transform?: Transform): PausableOutput {
     let paused = false;
     let settled = false;
     let resolveFinished: () => void = () => {};
@@ -21,7 +21,8 @@ export function createPausableOutput(source: Readable, target: Writable): Pausab
     const closed = new Promise<void>((resolve) => {
         resolveClosed = resolve;
     });
-    const attach = () => source.pipe(target, { end: false });
+    const outputSource = transform ? source.pipe(transform) : source;
+    const attach = () => outputSource.pipe(target, { end: false });
     const finish = () => {
         if (!settled) target.end();
     };
@@ -42,15 +43,16 @@ export function createPausableOutput(source: Readable, target: Writable): Pausab
         source.destroy(error);
         rejectFinished(error);
     });
-    source.once('end', finish);
+    outputSource.once('end', finish);
     source.once('error', (error) => target.destroy(error));
+    if (transform) transform.once('error', (error) => target.destroy(error));
     attach();
 
     return {
         pause() {
             if (paused || settled) return;
             paused = true;
-            source.unpipe(target);
+            outputSource.unpipe(target);
             source.pause();
         },
         resume() {
@@ -62,8 +64,9 @@ export function createPausableOutput(source: Readable, target: Writable): Pausab
         async cancel() {
             if (!settled) {
                 paused = false;
-                source.unpipe(target);
+                outputSource.unpipe(target);
                 source.destroy();
+                transform?.destroy();
                 target.destroy();
             }
             await closed;

src/main/domain/token-bucket-transform.test.ts

@@ -0,0 +1,78 @@
+import { PassThrough } from 'node:stream';
+import { describe, expect, it } from 'vitest';
+import { createTokenBucketTransform, type TokenBucketClock } from './token-bucket-transform';
+
+class ManualClock implements TokenBucketClock {
+    private nextTimerId = 0;
+    private readonly timers = new Map<number, { dueAt: number; callback: () => void }>();
+    nowMs = 0;
+
+    now(): number {
+        return this.nowMs;
+    }
+
+    setTimeout(callback: () => void, delayMs: number): ReturnType<typeof setTimeout> {
+        const id = ++this.nextTimerId;
+        this.timers.set(id, { dueAt: this.nowMs + delayMs, callback });
+        return id as unknown as ReturnType<typeof setTimeout>;
+    }
+
+    clearTimeout(handle: ReturnType<typeof setTimeout>): void {
+        this.timers.delete(handle as unknown as number);
+    }
+
+    advance(ms: number): void {
+        this.nowMs += ms;
+        while (true) {
+            const due = [...this.timers.entries()]
+                .filter(([, timer]) => timer.dueAt <= this.nowMs)
+                .sort(([, left], [, right]) => left.dueAt - right.dueAt)[0];
+            if (!due) return;
+            this.timers.delete(due[0]);
+            due[1].callback();
+        }
+    }
+
+    get timerCount(): number {
+        return this.timers.size;
+    }
+}
+
+describe('app-side token bucket transform', () => {
+    it('backpressures stdout after its initial bucket without changing the source bytes', () => {
+        const clock = new ManualClock();
+        const transform = createTokenBucketTransform(2, clock);
+        const output: Buffer[] = [];
+        transform.on('data', (chunk: Buffer) => output.push(Buffer.from(chunk)));
+
+        transform.write(Buffer.from('ab'));
+        transform.write(Buffer.from('cd'));
+
+        expect(Buffer.concat(output).toString()).toBe('ab');
+        expect(clock.timerCount).toBe(1);
+
+        clock.advance(999);
+        expect(Buffer.concat(output).toString()).toBe('ab');
+
+        clock.advance(1);
+        expect(Buffer.concat(output).toString()).toBe('abcd');
+    });
+
+    it('cancels a pending throttle timer when the output stream is destroyed', () => {
+        const clock = new ManualClock();
+        const source = new PassThrough();
+        const transform = createTokenBucketTransform(1, clock);
+        const output: Buffer[] = [];
+        source.pipe(transform).on('data', (chunk: Buffer) => output.push(Buffer.from(chunk)));
+
+        source.write(Buffer.from('a'));
+        source.write(Buffer.from('b'));
+        expect(clock.timerCount).toBe(1);
+
+        transform.destroy();
+        clock.advance(10_000);
+
+        expect(clock.timerCount).toBe(0);
+        expect(Buffer.concat(output).toString()).toBe('a');
+    });
+});

src/main/domain/token-bucket-transform.ts

@@ -0,0 +1,58 @@
+import { Transform } from 'node:stream';
+
+export interface TokenBucketClock {
+    now(): number;
+    setTimeout(callback: () => void, delayMs: number): ReturnType<typeof setTimeout>;
+    clearTimeout(handle: ReturnType<typeof setTimeout>): void;
+}
+
+const systemClock: TokenBucketClock = {
+    now: () => Date.now(),
+    setTimeout: (callback, delayMs) => setTimeout(callback, delayMs),
+    clearTimeout: (handle) => clearTimeout(handle),
+};
+
+class TokenBucketTransform extends Transform {
+    private availableBytes: number;
+    private lastRefillAt: number;
+    private timer: ReturnType<typeof setTimeout> | null = null;
+
+    constructor(private readonly maxBytesPerSecond: number, private readonly clock: TokenBucketClock) {
+        super();
+        this.availableBytes = maxBytesPerSecond;
+        this.lastRefillAt = clock.now();
+    }
+
+    override _transform(chunk: Buffer, _encoding: BufferEncoding, callback: (error?: Error | null) => void): void {
+        const output = Buffer.from(chunk);
+        const capacity = Math.max(this.maxBytesPerSecond, output.length);
+        const release = (): void => {
+            this.timer = null;
+            if (this.destroyed) return;
+            const now = this.clock.now();
+            const elapsed = Math.max(0, now - this.lastRefillAt);
+            this.availableBytes = Math.min(capacity, this.availableBytes + (elapsed * this.maxBytesPerSecond) / 1000);
+            this.lastRefillAt = now;
+            if (this.availableBytes >= output.length) {
+                this.availableBytes -= output.length;
+                this.push(output);
+                callback();
+                return;
+            }
+            const delayMs = Math.max(1, Math.ceil(((output.length - this.availableBytes) * 1000) / this.maxBytesPerSecond));
+            this.timer = this.clock.setTimeout(release, delayMs);
+        };
+        release();
+    }
+
+    override _destroy(error: Error | null, callback: (error: Error | null) => void): void {
+        if (this.timer) this.clock.clearTimeout(this.timer);
+        this.timer = null;
+        callback(error);
+    }
+}
+
+export function createTokenBucketTransform(maxBytesPerSecond: number, clock: TokenBucketClock = systemClock): Transform {
+    if (!Number.isSafeInteger(maxBytesPerSecond) || maxBytesPerSecond <= 0) throw new RangeError('maxBytesPerSecond must be a positive safe integer');
+    return new TokenBucketTransform(maxBytesPerSecond, clock);
+}