diff --git a/src/main/download-manager.ts b/src/main/download-manager.ts index 278137b..849a285 100644 --- a/src/main/download-manager.ts +++ b/src/main/download-manager.ts @@ -97,9 +97,13 @@ import { buildHistoryEntry, buildPackageDigestEvents, buildPackageNotificationEvent, + buildRemainingThresholdNotificationEvent, buildRunNotificationEvent, buildRunResult, - type PackageResultEnvelope + evaluateRemainingThreshold, + type PackageResultEnvelope, + type RemainingThresholdState, + type RunRemainingSnapshot } from "./notification-events"; type ActiveTask = { @@ -471,6 +475,7 @@ type RunLifecycleContext = { startedAt: number; packageGenerations: Map; downloadsFinished: boolean; + remainingNotification: RemainingThresholdState; }; function generateHistoryId(): string { @@ -6859,8 +6864,9 @@ export class DownloadManager extends EventEmitter { saveStatisticsLedger(this.storagePaths.statisticsFile, this.statisticsLedger); } - private emitState(force = false): void { - const now = nowMs(); + private emitState(force = false): void { + this.evaluateRemainingNotification(); + const now = nowMs(); const MIN_FORCE_GAP_MS = 120; if (force) { const sinceLastEmit = now - this.lastStateEmitAt; @@ -11757,7 +11763,13 @@ export class DownloadManager extends EventEmitter { for (const packageId of packageIds) { packageGenerations.set(packageId, this.getPackageResultGeneration(packageId)); } - const context: RunLifecycleContext = { id: uuidv4(), startedAt, packageGenerations, downloadsFinished }; + const context: RunLifecycleContext = { + id: uuidv4(), + startedAt, + packageGenerations, + downloadsFinished, + remainingNotification: { snapshot: null, crossings: 0 } + }; this.runContexts.set(context.id, context); return context; } @@ -11866,6 +11878,11 @@ export class DownloadManager extends EventEmitter { private trackPackagePostProcessResult(packageId: string): void { const generation = this.getPackageResultGeneration(packageId); const key = this.packageResultKey(packageId, generation); + for (const context of this.runContexts.values()) { + if (context.packageGenerations.get(packageId) === generation) { + return; + } + } if (this.suppressedPackageResults.has(key)) { return; } @@ -11888,6 +11905,72 @@ export class DownloadManager extends EventEmitter { }); } + private buildRunRemainingSnapshot(): RunRemainingSnapshot { + const openPackages = new Set(); + let remainingBytes = 0; + let openItems = 0; + let unknownCount = 0; + for (const itemId of this.runItemIds) { + const item = this.session.items[itemId]; + if (!item || isFinishedStatus(item.status)) { + continue; + } + const pkg = this.session.packages[item.packageId]; + if (!pkg || pkg.cancelled || !pkg.enabled) { + continue; + } + openItems += 1; + openPackages.add(pkg.id); + if (item.totalBytes === null) { + unknownCount += 1; + } else { + remainingBytes += Math.max(0, item.totalBytes - item.downloadedBytes); + } + } + const speedBps = !this.session.running || this.session.paused + ? 0 + : Math.max(0, Math.floor(this.speedBytesLastWindow / SPEED_WINDOW_SECONDS)); + const etaSeconds = unknownCount === 0 && speedBps > 0 + ? Math.ceil(remainingBytes / speedBps) + : remainingBytes === 0 && unknownCount === 0 + ? 0 + : -1; + return { + remainingBytes, + openItems, + openPackages: openPackages.size, + unknownCount, + speedBps, + etaSeconds + }; + } + + private evaluateRemainingNotification(): void { + const context = this.activeRunContextId ? this.runContexts.get(this.activeRunContextId) : undefined; + if (!context || context.downloadsFinished) { + return; + } + const current = this.buildRunRemainingSnapshot(); + const previous = context.remainingNotification.snapshot; + context.remainingNotification.snapshot = current; + if (!this.settings.notifyOnRemainingBelow) { + return; + } + const thresholdBytes = this.settings.notifyRemainingThresholdGb * 1024 ** 3; + const decision = evaluateRemainingThreshold(previous, current, thresholdBytes); + if (!decision.emit) { + return; + } + context.remainingNotification.crossings += 1; + this.queueNotificationEvent(buildRemainingThresholdNotificationEvent( + context.id, + context.remainingNotification.crossings, + current, + thresholdBytes, + nowMs() + )); + } + private queueSuccessfulPackageResult(envelope: PackageResultEnvelope): void { const key = this.packageResultKey(envelope.result.packageId, envelope.generation); this.successDigestResults.set(key, envelope); diff --git a/src/main/notification-events.ts b/src/main/notification-events.ts index 6822276..1c549a8 100644 --- a/src/main/notification-events.ts +++ b/src/main/notification-events.ts @@ -62,6 +62,25 @@ export interface HistoryEntryContext { provider: DebridProvider | null; } +export interface RunRemainingSnapshot { + remainingBytes: number; + openItems: number; + openPackages: number; + unknownCount: number; + speedBps: number; + etaSeconds: number; +} + +export interface RemainingThresholdDecision { + emit: boolean; + remainingBytes?: number; +} + +export interface RemainingThresholdState { + snapshot: RunRemainingSnapshot | null; + crossings: number; +} + const SUCCESS_TTL_MS = 6 * 60 * 60 * 1000; const IMPORTANT_TTL_MS = 24 * 60 * 60 * 1000; const DIGEST_PACKAGE_LIMIT = 20; @@ -73,6 +92,25 @@ function finiteNonNegative(value: unknown): number { return Number.isFinite(numeric) && numeric > 0 ? numeric : 0; } +export function evaluateRemainingThreshold( + previous: RunRemainingSnapshot | null, + current: RunRemainingSnapshot, + thresholdBytes: number +): RemainingThresholdDecision { + const threshold = finiteNonNegative(thresholdBytes); + if (!previous + || threshold <= 0 + || previous.openItems <= 0 + || current.openItems <= 0 + || previous.unknownCount > 0 + || current.unknownCount > 0 + || previous.remainingBytes <= threshold + || current.remainingBytes > threshold) { + return { emit: false }; + } + return { emit: true, remainingBytes: current.remainingBytes }; +} + function formatBytes(bytes: number): string { let value = finiteNonNegative(bytes); const units = ["B", "KB", "MB", "GB", "TB", "PB"]; @@ -285,6 +323,31 @@ export function buildRunNotificationEvent(result: RunResult): NotificationEvent ); } +export function buildRemainingThresholdNotificationEvent( + runId: string, + crossing: number, + snapshot: RunRemainingSnapshot, + thresholdBytes: number, + createdAt: number +): NotificationEvent { + return event( + `run:${runId}:remaining_threshold_crossed:${crossing}`, + "remaining_threshold_crossed", + "success", + createdAt, + "📉 Restmenge erreicht", + `Der aktive Durchlauf liegt bei oder unter ${formatBytes(thresholdBytes)}.`, + 0x3498db, + [ + { name: "Restmenge", value: formatBytes(snapshot.remainingBytes), inline: true }, + { name: "Offene Pakete", value: String(snapshot.openPackages), inline: true }, + { name: "Offene Dateien", value: String(snapshot.openItems), inline: true }, + { name: "Geschwindigkeit", value: snapshot.speedBps > 0 ? `${formatBytes(snapshot.speedBps)}/s` : "—", inline: true }, + { name: "ETA", value: snapshot.etaSeconds >= 0 ? formatDuration(snapshot.etaSeconds) : "—", inline: true } + ] + ); +} + export function buildHistoryEntry( result: PackageResult, context: HistoryEntryContext diff --git a/tests/notify-hooks.test.ts b/tests/notify-hooks.test.ts index e5b2e89..577fb66 100644 --- a/tests/notify-hooks.test.ts +++ b/tests/notify-hooks.test.ts @@ -650,6 +650,52 @@ describe("authoritative run completion", () => { expect(history.map((entry) => entry.name)).toEqual([packageA.name]); }); + it("keeps run A ownership when its real deferred follow-up starts during run B before run B stops", async () => { + const { manager, session, events, history } = setup({ autoExtractWhenStopped: true, maxParallelExtract: 1 }); + const packageA = addPackage(session, ["queued"], "deferred-owner-package"); + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + + await manager.start(); + const packageAItem = session.items[packageA.itemIds[0]]; + packageAItem.status = "completed"; + packageAItem.downloadedBytes = 1_000; + packageAItem.totalBytes = 1_000; + packageAItem.progressPercent = 100; + packageAItem.fullStatus = "Fertig"; + packageA.status = "completed"; + state.runOutcomes.set(packageAItem.id, "completed"); + state.packagePostProcessActive = 1; + + let releaseCollection = (): void => {}; + const collectionGate = new Promise((resolve) => { + releaseCollection = resolve; + }); + const collect = vi.spyOn(state, "collectMkvFilesToLibrary").mockImplementation(async () => collectionGate); + const packageAMainPostProcess = state.runPackagePostProcessing(packageA.id); + await vi.waitFor(() => expect(state.packagePostProcessWaiters).toHaveLength(1)); + state.finishRun(); + + const packageB = addPackage(session, ["queued"], "active-run-package"); + await manager.start(); + expect(state.runPackageIds).toEqual(new Set([packageB.id])); + + state.releasePostProcessSlot(); + await packageAMainPostProcess; + await vi.waitFor(() => expect(collect).toHaveBeenCalled()); + const deferredTasks = [...(state.packageDeferredPostProcessTasks.get(packageA.id) || [])]; + expect(deferredTasks).toHaveLength(1); + + manager.stop(); + releaseCollection(); + await Promise.allSettled(deferredTasks); + await flushNotifications(); + + expect(events.filter((event) => event.type === "package_completed")).toHaveLength(1); + expect(events.filter((event) => event.type === "run_completed")).toHaveLength(1); + expect(history.map((entry) => entry.name)).toEqual([packageA.name]); + }); + it("does not reactivate a suppressed foreign package when another start recovers it from disk", async () => { const { manager, session, events, history } = setup({ autoExtractWhenStopped: true }); const packageA = addPackage(session, ["queued"], "suppressed-recovery-package"); diff --git a/tests/remaining-notification.test.ts b/tests/remaining-notification.test.ts new file mode 100644 index 0000000..5d54b35 --- /dev/null +++ b/tests/remaining-notification.test.ts @@ -0,0 +1,345 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { DownloadManager } from "../src/main/download-manager"; +import { defaultSettings } from "../src/main/constants"; +import { + evaluateRemainingThreshold, + type RunRemainingSnapshot +} from "../src/main/notification-events"; +import type { NotificationEvent } from "../src/main/notification-outbox"; +import { createStoragePaths, emptySession } from "../src/main/storage"; +import { shutdownItemLogs } from "../src/main/item-log"; +import { shutdownPackageLogs } from "../src/main/package-log"; +import { shutdownRenameLog } from "../src/main/rename-log"; +import type { AppSettings, PackageEntry } from "../src/shared/types"; + +const GIB = 1024 ** 3; +const tempDirs: string[] = []; + +afterEach(() => { + vi.restoreAllMocks(); + shutdownItemLogs(); + shutdownPackageLogs(); + shutdownRenameLog(); + for (const dir of tempDirs.splice(0)) { + try { + fs.rmSync(dir, { recursive: true, force: true }); + } catch { + } + } +}); + +function snapshot(overrides: Partial = {}): RunRemainingSnapshot { + return { + remainingBytes: 51 * GIB, + openItems: 2, + openPackages: 1, + unknownCount: 0, + speedBps: 1024 ** 2, + etaSeconds: 51 * 1024, + ...overrides + }; +} + +function setupManager(settings: Partial = {}, session = emptySession()): { + manager: DownloadManager; + session: ReturnType; + events: NotificationEvent[]; + settings: AppSettings; +} { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-remaining-")); + tempDirs.push(root); + const events: NotificationEvent[] = []; + const resolvedSettings = { + ...defaultSettings(), + token: "rd-token", + outputDir: path.join(root, "out"), + extractDir: path.join(root, "extract"), + notifyUrl: "https://discord.com/api/webhooks/123/abc", + notifyOnPackageCompleted: false, + notifyOnPackageFailed: false, + notifyOnRunFinished: false, + notifyOnRemainingBelow: true, + notifyRemainingThresholdGb: 50, + autoExtract: false, + ...settings + }; + const manager = new DownloadManager( + resolvedSettings, + session, + createStoragePaths(path.join(root, "state")), + { + enqueueNotification: async (event: NotificationEvent) => { + events.push(event); + } + } + ); + return { manager, session, events, settings: resolvedSettings }; +} + +function addPackage( + session: ReturnType, + packageId: string, + totalBytes: number | null, + downloadedBytes = 0, + enabled = true +): PackageEntry { + const now = Date.now(); + const itemId = `${packageId}-item`; + const pkg: PackageEntry = { + id: packageId, + name: packageId, + outputDir: `C:/out/${packageId}`, + extractDir: `C:/extract/${packageId}`, + status: "queued", + itemIds: [itemId], + cancelled: false, + enabled, + priority: "normal", + createdAt: now, + updatedAt: now + }; + session.packages[packageId] = pkg; + session.packageOrder.push(packageId); + session.items[itemId] = { + id: itemId, + packageId, + url: `https://dummy/${packageId}`, + provider: null, + status: "queued", + retries: 0, + speedBps: 0, + downloadedBytes, + totalBytes, + progressPercent: totalBytes && totalBytes > 0 ? Math.floor((downloadedBytes / totalBytes) * 100) : 0, + fileName: `${packageId}.bin`, + targetPath: "", + resumable: true, + attempts: 0, + lastError: "", + fullStatus: "Wartet", + createdAt: now, + updatedAt: now + }; + return pkg; +} + +function internal(manager: DownloadManager): any { + return manager as any; +} + +async function flushNotifications(): Promise { + await new Promise((resolve) => setTimeout(resolve, 0)); +} + +describe("remaining threshold evaluation", () => { + it("emits when known remaining bytes cross from above to exactly the threshold", () => { + const previous = snapshot(); + const current = snapshot({ remainingBytes: 50 * GIB, etaSeconds: 50 * 1024 }); + + expect(evaluateRemainingThreshold(previous, current, 50 * GIB)).toEqual({ + emit: true, + remainingBytes: 50 * GIB + }); + }); + + it("blocks a crossing when either snapshot contains an unknown open size", () => { + const above = snapshot(); + const below = snapshot({ remainingBytes: 49 * GIB, etaSeconds: 49 * 1024 }); + + expect(evaluateRemainingThreshold(above, { ...below, unknownCount: 1 }, 50 * GIB)).toEqual({ emit: false }); + expect(evaluateRemainingThreshold({ ...above, unknownCount: 1 }, below, 50 * GIB)).toEqual({ emit: false }); + }); + + it("suppresses a crossing when no open run item remains", () => { + expect(evaluateRemainingThreshold( + snapshot(), + snapshot({ remainingBytes: 0, openItems: 0, openPackages: 0, etaSeconds: 0 }), + 50 * GIB + )).toEqual({ emit: false }); + }); + + it("emits once per crossing and re-arms only after remaining work rises above the threshold", () => { + const above = snapshot(); + const below = snapshot({ remainingBytes: 49 * GIB, etaSeconds: 49 * 1024 }); + + expect(evaluateRemainingThreshold(above, below, 50 * GIB).emit).toBe(true); + expect(evaluateRemainingThreshold(below, below, 50 * GIB).emit).toBe(false); + expect(evaluateRemainingThreshold(below, above, 50 * GIB).emit).toBe(false); + expect(evaluateRemainingThreshold(above, below, 50 * GIB).emit).toBe(true); + }); + + it("does not synthesize a crossing for a new or restored run that first appears below the threshold", () => { + const below = snapshot({ remainingBytes: 49 * GIB, etaSeconds: 49 * 1024 }); + + expect(evaluateRemainingThreshold(null, below, 50 * GIB)).toEqual({ emit: false }); + }); +}); + +describe("run-scoped remaining notifications", () => { + it("calculates known remainder, speed and ETA only from open enabled items in the active run", async () => { + const { manager, session } = setupManager(); + const active = addPackage(session, "active-package", 60 * GIB, 9 * GIB); + const disabled = addPackage(session, "disabled-package", null, 0, false); + const notStarted = addPackage(session, "not-started-package", 400 * GIB); + const completedItemId = `${active.id}-completed-item`; + active.itemIds.push(completedItemId); + session.items[completedItemId] = { + ...session.items[active.itemIds[0]], + id: completedItemId, + status: "completed", + downloadedBytes: 300 * GIB, + totalBytes: 300 * GIB, + progressPercent: 100, + fullStatus: "Fertig" + }; + const overrunItemId = `${active.id}-overrun-item`; + active.itemIds.push(overrunItemId); + session.items[overrunItemId] = { + ...session.items[active.itemIds[0]], + id: overrunItemId, + downloadedBytes: 2 * GIB, + totalBytes: GIB, + progressPercent: 100 + }; + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + + await manager.start({ excludePackageIds: new Set([notStarted.id]) }); + state.speedBytesLastWindow = GIB; + + expect(state.buildRunRemainingSnapshot()).toEqual({ + remainingBytes: 51 * GIB, + openItems: 2, + openPackages: 1, + unknownCount: 0, + speedBps: GIB, + etaSeconds: 51 + }); + expect(state.runPackageIds).toEqual(new Set([active.id])); + expect(state.runPackageIds.has(disabled.id)).toBe(false); + }); + + it("enqueues one complete event per crossing and re-arms after new work raises the remainder", async () => { + const { manager, session, events } = setupManager(); + const pkg = addPackage(session, "crossing-package", 51 * GIB); + const item = session.items[pkg.itemIds[0]]; + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + + await manager.start(); + state.speedBytesLastWindow = GIB; + item.downloadedBytes = GIB; + state.evaluateRemainingNotification(); + state.evaluateRemainingNotification(); + await flushNotifications(); + + expect(events.filter((event) => event.type === "remaining_threshold_crossed")).toHaveLength(1); + expect(events[0].payload.fields).toEqual([ + { name: "Restmenge", value: "50 GB", inline: true }, + { name: "Offene Pakete", value: "1", inline: true }, + { name: "Offene Dateien", value: "1", inline: true }, + { name: "Geschwindigkeit", value: "1 GB/s", inline: true }, + { name: "ETA", value: "0:50", inline: true } + ]); + + item.downloadedBytes = 0; + state.evaluateRemainingNotification(); + item.downloadedBytes = 2 * GIB; + state.evaluateRemainingNotification(); + await flushNotifications(); + + expect(events.filter((event) => event.type === "remaining_threshold_crossed")).toHaveLength(2); + expect(new Set(events.map((event) => event.id)).size).toBe(2); + }); + + it("waits for unknown sizes to become known above the threshold before allowing a crossing", async () => { + const { manager, session, events } = setupManager(); + const pkg = addPackage(session, "unknown-package", null); + const item = session.items[pkg.itemIds[0]]; + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + + await manager.start(); + item.totalBytes = 49 * GIB; + state.evaluateRemainingNotification(); + await flushNotifications(); + expect(events).toHaveLength(0); + + item.totalBytes = 60 * GIB; + state.evaluateRemainingNotification(); + item.downloadedBytes = 11 * GIB; + state.evaluateRemainingNotification(); + await flushNotifications(); + + expect(events.filter((event) => event.type === "remaining_threshold_crossed")).toHaveLength(1); + }); + + it("allows the same package to cross again in a new run", async () => { + const { manager, session, events } = setupManager(); + const pkg = addPackage(session, "new-run-package", 51 * GIB); + const item = session.items[pkg.itemIds[0]]; + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + + await manager.start(); + item.downloadedBytes = 2 * GIB; + state.evaluateRemainingNotification(); + manager.stop(); + + item.downloadedBytes = 0; + await manager.start(); + item.downloadedBytes = 2 * GIB; + state.evaluateRemainingNotification(); + await flushNotifications(); + + expect(events.filter((event) => event.type === "remaining_threshold_crossed")).toHaveLength(2); + expect(new Set(events.map((event) => event.id)).size).toBe(2); + }); + + it("does not duplicate a prior crossing when a restored queue first appears below the threshold", async () => { + const first = setupManager(); + const pkg = addPackage(first.session, "restart-package", 51 * GIB); + const item = first.session.items[pkg.itemIds[0]]; + const firstState = internal(first.manager); + vi.spyOn(firstState, "ensureScheduler").mockResolvedValue(undefined); + + await first.manager.start(); + item.downloadedBytes = 2 * GIB; + firstState.evaluateRemainingNotification(); + await flushNotifications(); + expect(first.events).toHaveLength(1); + + const restored = setupManager({}, first.session); + const restoredState = internal(restored.manager); + vi.spyOn(restoredState, "ensureScheduler").mockResolvedValue(undefined); + await restored.manager.start(); + restoredState.evaluateRemainingNotification(); + await flushNotifications(); + + expect(restored.events).toHaveLength(0); + }); + + it("suppresses a threshold event when the same evaluation completes the last open item", async () => { + const { manager, session, events } = setupManager(); + const pkg = addPackage(session, "final-package", 51 * GIB); + const item = session.items[pkg.itemIds[0]]; + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + + await manager.start(); + item.status = "completed"; + item.downloadedBytes = 51 * GIB; + item.progressPercent = 100; + item.fullStatus = "Fertig"; + pkg.status = "completed"; + state.runOutcomes.set(item.id, "completed"); + state.evaluateRemainingNotification(); + state.finishRun(); + await flushNotifications(); + + expect(events.filter((event) => event.type === "remaining_threshold_crossed")).toHaveLength(0); + }); +});