feat(statistics): record rolling account traffic

2026-08-21 08:25:32 +02:00
parent 54bcd82cb6
commit 7121a334d2
5 changed files with 415 additions and 7 deletions
@@ -4,9 +4,13 @@ import path from "node:path";
 import { getProviderUsageDayKey } from "../shared/provider-daily-limits";
 import type {
  DebridProvider,
+  StatisticsAccountMinuteUsage,
+  StatisticsAccountUsage,
  StatisticsDayBucket,
  StatisticsLedger,
-  StatisticsProviderBucket
+  StatisticsMinuteBucket,
+  StatisticsProviderBucket,
+  StatisticsRolling24Hours
 } from "../shared/types";
 export { aggregateStatisticsRange } from "../shared/statistics-aggregation";

@@ -24,6 +28,24 @@ const providers = new Set<DebridProvider>([
 ]);

 const renameRetryDelaysMs = [15, 40, 90];
+const minuteMs = 60_000;
+const rollingWindowMs = 24 * 60 * minuteMs;
+const retainedMinuteCount = 48 * 60;
+const maximumAccountIdLength = 128;
+const maximumAccountLabelLength = 96;
+
+const providerLabels: Record<DebridProvider, string> = {
+  realdebrid: "Real-Debrid",
+  megadebrid: "Mega-Debrid",
+  "megadebrid-api": "Mega-Debrid API",
+  "megadebrid-web": "Mega-Debrid Web",
+  bestdebrid: "BestDebrid",
+  alldebrid: "AllDebrid",
+  ddownload: "DDownload",
+  onefichier: "1Fichier",
+  debridlink: "Debrid-Link",
+  linksnappy: "LinkSnappy"
+};

 function finiteNonNegative(value: unknown): number {
  const number = Number(value);
@@ -36,6 +58,97 @@ function asRecord(value: unknown): Record<string, unknown> | null {
    : null;
 }

+function minuteStart(epochMs: number): number {
+  return Math.floor(finiteNonNegative(epochMs) / minuteMs) * minuteMs;
+}
+
+function validAccountId(value: unknown): string | null {
+  const id = String(value || "").trim();
+  return id.length > 0
+    && id.length <= maximumAccountIdLength
+    && /^[A-Za-z0-9][A-Za-z0-9._:-]*$/.test(id)
+    ? id
+    : null;
+}
+
+function maskEmailLikeLabel(value: string): string {
+  const match = /^([^@\s]+)@([^@\s]+)$/.exec(value);
+  if (!match) {
+    return value;
+  }
+  const local = match[1];
+  const hidden = "*".repeat(Math.max(3, Math.min(8, local.length - 1)));
+  return `${local.slice(0, 1)}${hidden}@${match[2]}`;
+}
+
+function safeAccountLabel(value: unknown, provider: DebridProvider): string {
+  const clean = String(value || "")
+    .replace(/[\u0000-\u001f\u007f]/g, " ")
+    .replace(/\s+/g, " ")
+    .trim()
+    .slice(0, maximumAccountLabelLength);
+  return maskEmailLikeLabel(clean || providerLabels[provider]);
+}
+
+function fallbackAccountId(provider: DebridProvider): string {
+  return `provider:${provider}`;
+}
+
+function normalizeMinuteAccounts(value: unknown): Record<string, StatisticsAccountMinuteUsage> {
+  const accounts: Record<string, StatisticsAccountMinuteUsage> = {};
+  for (const [rawId, rawUsage] of Object.entries(asRecord(value) ?? {})) {
+    const id = validAccountId(rawId);
+    const usage = asRecord(rawUsage);
+    const provider = String(usage?.provider || "") as DebridProvider;
+    const bytes = finiteNonNegative(usage?.bytes);
+    if (!id || !providers.has(provider) || bytes <= 0) {
+      continue;
+    }
+    accounts[id] = {
+      provider,
+      label: safeAccountLabel(usage?.label, provider),
+      bytes
+    };
+  }
+  return accounts;
+}
+
+function minimumRetainedMinute(now: number): number {
+  return minuteStart(now) - ((retainedMinuteCount - 1) * minuteMs);
+}
+
+function normalizeMinutes(value: unknown, now: number): StatisticsMinuteBucket[] {
+  const currentMinute = minuteStart(now);
+  const minimumMinute = minimumRetainedMinute(now);
+  const buckets = new Map<number, StatisticsMinuteBucket>();
+  for (const rawValue of Array.isArray(value) ? value : []) {
+    const record = asRecord(rawValue);
+    const minute = minuteStart(Number(record?.minute));
+    if (!record || minute < minimumMinute || minute > currentMinute) {
+      continue;
+    }
+    const accounts = normalizeMinuteAccounts(record.accounts);
+    if (Object.keys(accounts).length === 0) {
+      continue;
+    }
+    const target = buckets.get(minute) ?? { minute, downloadedBytes: 0, accounts: {} };
+    for (const [id, usage] of Object.entries(accounts)) {
+      const existing = target.accounts[id];
+      if (existing && existing.provider === usage.provider) {
+        existing.bytes += usage.bytes;
+        existing.label = usage.label;
+      } else {
+        target.accounts[id] = { ...usage };
+      }
+    }
+    target.downloadedBytes = Object.values(target.accounts).reduce((total, usage) => total + usage.bytes, 0);
+    buckets.set(minute, target);
+  }
+  return [...buckets.values()]
+    .sort((left, right) => left.minute - right.minute)
+    .slice(-retainedMinuteCount);
+}
+
 function emptyProviderBucket(): StatisticsProviderBucket {
  return { bytes: 0, completed: 0, failed: 0 };
 }
@@ -86,7 +199,7 @@ function normalizeDay(value: unknown): StatisticsDayBucket | null {
 }

 export function createStatisticsLedger(now = Date.now()): StatisticsLedger {
-  return { version: 1, startedAt: now, days: [] };
+  return { version: 2, startedAt: now, days: [], minutes: [] };
 }

 export function normalizeStatisticsLedger(value: unknown, now = Date.now()): StatisticsLedger {
@@ -102,12 +215,161 @@ export function normalizeStatisticsLedger(value: unknown, now = Date.now()): Sta
    }
  }
  return {
-    version: 1,
+    version: 2,
    startedAt: finiteNonNegative(record.startedAt) || now,
-    days: [...byDay.values()].sort((left, right) => left.day.localeCompare(right.day))
+    days: [...byDay.values()].sort((left, right) => left.day.localeCompare(right.day)),
+    minutes: normalizeMinutes(record.minutes, now)
  };
 }

+export function projectStatisticsLedger(ledger: StatisticsLedger, now = Date.now()): StatisticsLedger {
+  return normalizeStatisticsLedger({ ...ledger, minutes: [] }, now);
+}
+
+export function addStatisticsAccountBytesInPlace(
+  ledger: StatisticsLedger,
+  provider: DebridProvider,
+  byteDelta: number,
+  accountId?: string,
+  accountLabel?: string,
+  epochMs = Date.now()
+): StatisticsAccountUsage | null {
+  const bytes = finiteNonNegative(byteDelta);
+  if (bytes <= 0 || !providers.has(provider)) {
+    return null;
+  }
+  const minute = minuteStart(epochMs);
+  const id = validAccountId(accountId) ?? fallbackAccountId(provider);
+  const label = safeAccountLabel(accountLabel, provider);
+  let bucket: StatisticsMinuteBucket | undefined = ledger.minutes[ledger.minutes.length - 1];
+  if (!bucket || bucket.minute !== minute) {
+    ledger.minutes = ledger.minutes.filter((entry) => entry.minute >= minimumRetainedMinute(epochMs) && entry.minute <= minute);
+    bucket = ledger.minutes.find((entry) => entry.minute === minute);
+    if (!bucket) {
+      bucket = { minute, downloadedBytes: 0, accounts: {} };
+      ledger.minutes.push(bucket);
+      ledger.minutes.sort((left, right) => left.minute - right.minute);
+      if (ledger.minutes.length > retainedMinuteCount) {
+        ledger.minutes.splice(0, ledger.minutes.length - retainedMinuteCount);
+      }
+    }
+  }
+  if (!bucket) {
+    return null;
+  }
+  const existing = bucket.accounts[id];
+  if (existing && existing.provider === provider) {
+    existing.bytes += bytes;
+    existing.label = label;
+  } else {
+    bucket.accounts[id] = { provider, label, bytes };
+  }
+  bucket.downloadedBytes = Object.values(bucket.accounts).reduce((total, usage) => total + usage.bytes, 0);
+  return { id, provider, label, bytes };
+}
+
+function aggregateRollingAccountStatistics(ledger: StatisticsLedger, now: number): StatisticsRolling24Hours {
+  const currentMinute = minuteStart(now);
+  const from = minuteStart(now - rollingWindowMs);
+  const accounts = new Map<string, StatisticsAccountUsage>();
+  for (const bucket of ledger.minutes) {
+    if (bucket.minute < from || bucket.minute > currentMinute) {
+      continue;
+    }
+    for (const [id, usage] of Object.entries(bucket.accounts)) {
+      const existing = accounts.get(id);
+      if (existing && existing.provider === usage.provider) {
+        existing.bytes += usage.bytes;
+        existing.label = usage.label;
+      } else {
+        accounts.set(id, { id, provider: usage.provider, label: usage.label, bytes: usage.bytes });
+      }
+    }
+  }
+  const rows = [...accounts.values()].sort((left, right) =>
+    right.bytes - left.bytes
+    || left.label.localeCompare(right.label, "de-DE", { sensitivity: "base" })
+    || left.id.localeCompare(right.id)
+  );
+  return {
+    from,
+    to: now,
+    downloadedBytes: rows.reduce((total, account) => total + account.bytes, 0),
+    accounts: rows
+  };
+}
+
+export class RollingAccountStatisticsAccumulator {
+  private ledger: StatisticsLedger;
+  private minute = -1;
+  private aggregate: StatisticsRolling24Hours;
+
+  public constructor(ledger: StatisticsLedger, now = Date.now()) {
+    this.ledger = ledger;
+    this.aggregate = aggregateRollingAccountStatistics(ledger, now);
+    this.minute = minuteStart(now);
+  }
+
+  public record(
+    provider: DebridProvider,
+    byteDelta: number,
+    accountId?: string,
+    accountLabel?: string,
+    epochMs = Date.now()
+  ): void {
+    this.refresh(epochMs);
+    const recorded = addStatisticsAccountBytesInPlace(
+      this.ledger,
+      provider,
+      byteDelta,
+      accountId,
+      accountLabel,
+      epochMs
+    );
+    if (!recorded || minuteStart(epochMs) < this.aggregate.from || minuteStart(epochMs) > this.minute) {
+      return;
+    }
+    const existing = this.aggregate.accounts.find((account) => account.id === recorded.id && account.provider === recorded.provider);
+    if (existing) {
+      existing.bytes += recorded.bytes;
+      existing.label = recorded.label;
+    } else {
+      this.aggregate.accounts.push({ ...recorded });
+    }
+    this.aggregate.downloadedBytes += recorded.bytes;
+    this.aggregate.accounts.sort((left, right) =>
+      right.bytes - left.bytes
+      || left.label.localeCompare(right.label, "de-DE", { sensitivity: "base" })
+      || left.id.localeCompare(right.id)
+    );
+    this.aggregate.to = epochMs;
+  }
+
+  public snapshot(now = Date.now()): StatisticsRolling24Hours {
+    this.refresh(now);
+    return {
+      ...this.aggregate,
+      to: now,
+      accounts: this.aggregate.accounts.map((account) => ({ ...account }))
+    };
+  }
+
+  public reset(ledger: StatisticsLedger, now = Date.now()): void {
+    this.ledger = ledger;
+    this.minute = minuteStart(now);
+    this.aggregate = aggregateRollingAccountStatistics(ledger, now);
+  }
+
+  private refresh(now: number): void {
+    const currentMinute = minuteStart(now);
+    if (currentMinute === this.minute) {
+      return;
+    }
+    this.minute = currentMinute;
+    this.aggregate = aggregateRollingAccountStatistics(this.ledger, now);
+  }
+}
+
 function updateDay(
  ledger: StatisticsLedger,
  epochMs: number,
@@ -57,10 +57,37 @@ export interface StatisticsDayBucket {
  providers: Partial<Record<DebridProvider, StatisticsProviderBucket>>;
 }

+export interface StatisticsAccountMinuteUsage {
+  provider: DebridProvider;
+  label: string;
+  bytes: number;
+}
+
+export interface StatisticsMinuteBucket {
+  minute: number;
+  downloadedBytes: number;
+  accounts: Record<string, StatisticsAccountMinuteUsage>;
+}
+
+export interface StatisticsAccountUsage {
+  id: string;
+  provider: DebridProvider;
+  label: string;
+  bytes: number;
+}
+
+export interface StatisticsRolling24Hours {
+  from: number;
+  to: number;
+  downloadedBytes: number;
+  accounts: StatisticsAccountUsage[];
+}
+
 export interface StatisticsLedger {
-  version: 1;
+  version: 2;
  startedAt: number;
  days: StatisticsDayBucket[];
+  minutes: StatisticsMinuteBucket[];
 }

 export interface DownloadStats {
@@ -76,6 +103,7 @@ export interface DownloadStats {
  totalRuntimeMs: number;
  runtimeMeasuredAt: number;
  statistics?: StatisticsLedger;
+  rolling24Hours?: StatisticsRolling24Hours;
 }

 export interface DebridAccountStatus {