feat(notifications): alert on remaining volume

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

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

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

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> = {}): RunRemainingSnapshot {
+  return {
+    remainingBytes: 51 * GIB,
+    openItems: 2,
+    openPackages: 1,
+    unknownCount: 0,
+    speedBps: 1024 ** 2,
+    etaSeconds: 51 * 1024,
+    ...overrides
+  };
+}
+
+function setupManager(settings: Partial<AppSettings> = {}, session = emptySession()): {
+  manager: DownloadManager;
+  session: ReturnType<typeof emptySession>;
+  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<typeof emptySession>,
+  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<void> {
+  await new Promise<void>((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);
+  });
+});