Add outbound traffic breakdown to device charts

src/server/services/deviceInventoryService.ts

@@ -141,6 +141,24 @@ interface TrafficCursor {
   download: bigint;
 }
 
+interface OutboundTrafficSample {
+  observedAt: string;
+  vpnBytes: string;
+  directTrackedBytes: string;
+  directIpv4Bytes: string;
+  unknownBytes: string;
+}
+
+interface OutboundTrafficCursor {
+  signature: string;
+  routeEpoch: string;
+  directEpoch: string;
+  vpn: bigint;
+  directTracked: bigint;
+  directIpv4: bigint;
+  unknown: bigint;
+}
+
 interface DeviceObservation {
   mac: string;
   ip: string;
@@ -689,6 +707,8 @@ export function createDeviceInventoryService({
   let policyQueue: Promise<unknown> = Promise.resolve();
   const trafficHistoryByMac = new Map<string, TrafficSample[]>();
   const trafficCursorByMac = new Map<string, TrafficCursor>();
+  const outboundTrafficHistoryByDeviceId = new Map<string, OutboundTrafficSample[]>();
+  const outboundTrafficCursorByDeviceId = new Map<string, OutboundTrafficCursor>();
   let globalTrafficHistory: TrafficSample[] = [];
   let globalTrafficCursor: TrafficCursor | null = null;
   const hostnameAttempts = new Map<string, number>();
@@ -763,6 +783,79 @@ export function createDeviceInventoryService({
     }
   }
 
+  function captureOutboundTrafficHistory(state: InventoryState) {
+    const routeEpoch = typeof domainTrafficSnapshot.epoch === 'string' ? domainTrafficSnapshot.epoch : '';
+    const directEpoch = typeof directTrafficSnapshot.epoch === 'string' ? directTrafficSnapshot.epoch : '';
+    const routeObservedAt = validTimestamp(domainTrafficSnapshot.observedAt)
+      ? String(domainTrafficSnapshot.observedAt)
+      : '';
+    const directObservedAt = validTimestamp(directTrafficSnapshot.observedAt)
+      ? String(directTrafficSnapshot.observedAt)
+      : '';
+    const signature = `${routeEpoch}|${routeObservedAt}|${directEpoch}|${directObservedAt}`;
+    if (signature === '|||') return;
+
+    const totals = new Map<string, Omit<OutboundTrafficCursor, 'signature' | 'routeEpoch' | 'directEpoch'>>();
+    const totalFor = (deviceId: string) => {
+      const existing = totals.get(deviceId) || { vpn: 0n, directTracked: 0n, directIpv4: 0n, unknown: 0n };
+      totals.set(deviceId, existing);
+      return existing;
+    };
+    for (const value of Array.isArray(domainTrafficSnapshot.routes) ? domainTrafficSnapshot.routes : []) {
+      const row = record(value);
+      const deviceId = String(row.deviceId || '');
+      const outbound = String(row.outbound || '');
+      const uploadBytes = String(row.uploadBytes || '');
+      const downloadBytes = String(row.downloadBytes || '');
+      if (!DEVICE_ID_PATTERN.test(deviceId) || !['vpn', 'direct', 'unknown'].includes(outbound)
+        || !COUNTER_PATTERN.test(uploadBytes) || !COUNTER_PATTERN.test(downloadBytes)) continue;
+      const amount = BigInt(uploadBytes) + BigInt(downloadBytes);
+      const total = totalFor(deviceId);
+      if (outbound === 'vpn') total.vpn += amount;
+      else if (outbound === 'direct') total.directTracked += amount;
+      else total.unknown += amount;
+    }
+    for (const value of Array.isArray(directTrafficSnapshot.series) ? directTrafficSnapshot.series : []) {
+      const row = record(value);
+      const deviceId = String(row.deviceId || '');
+      const uploadBytes = String(row.uploadBytes || '');
+      const downloadBytes = String(row.downloadBytes || '');
+      if (!DEVICE_ID_PATTERN.test(deviceId)
+        || !COUNTER_PATTERN.test(uploadBytes) || !COUNTER_PATTERN.test(downloadBytes)) continue;
+      totalFor(deviceId).directIpv4 += BigInt(uploadBytes) + BigInt(downloadBytes);
+    }
+
+    const knownIds = new Set(state.devices.map(({ id }) => id));
+    const observedAt = [routeObservedAt, directObservedAt].filter(Boolean).sort().at(-1) || '';
+    for (const device of state.devices) {
+      const total = totalFor(device.id);
+      const current: OutboundTrafficCursor = { signature, routeEpoch, directEpoch, ...total };
+      const previous = outboundTrafficCursorByDeviceId.get(device.id);
+      outboundTrafficCursorByDeviceId.set(device.id, current);
+      if (!previous || previous.signature === signature || !observedAt) continue;
+      const routeDelta = (value: bigint, before: bigint) => (
+        routeEpoch && routeEpoch === previous.routeEpoch && value > before ? value - before : 0n
+      );
+      const directDelta = directEpoch && directEpoch === previous.directEpoch && total.directIpv4 > previous.directIpv4
+        ? total.directIpv4 - previous.directIpv4
+        : 0n;
+      const samples = outboundTrafficHistoryByDeviceId.get(device.id) || [];
+      outboundTrafficHistoryByDeviceId.set(device.id, [...samples, {
+        observedAt,
+        vpnBytes: routeDelta(total.vpn, previous.vpn).toString(),
+        directTrackedBytes: routeDelta(total.directTracked, previous.directTracked).toString(),
+        directIpv4Bytes: directDelta.toString(),
+        unknownBytes: routeDelta(total.unknown, previous.unknown).toString(),
+      }].slice(-TRAFFIC_HISTORY_LIMIT));
+    }
+    for (const deviceId of outboundTrafficCursorByDeviceId.keys()) {
+      if (!knownIds.has(deviceId)) {
+        outboundTrafficCursorByDeviceId.delete(deviceId);
+        outboundTrafficHistoryByDeviceId.delete(deviceId);
+      }
+    }
+  }
+
   function serializePolicy<T>(action: () => Promise<T> | T): Promise<T> {
     const result = policyQueue.then(() => action(), () => action());
     policyQueue = result.catch(() => {});
@@ -826,6 +919,7 @@ export function createDeviceInventoryService({
         proxyDownloadBytes: proxyTraffic?.downloadBytes || '0',
         proxyTrafficObservedAt: proxyTraffic?.observedAt || null,
         trafficHistory: trafficHistoryByMac.get(device.mac) || [],
+        outboundTrafficHistory: outboundTrafficHistoryByDeviceId.get(device.id) || [],
         desiredPolicy: policy.desired,
         appliedPolicy: policy.applied,
         policyStatus: policy.status,
@@ -1375,6 +1469,7 @@ export function createDeviceInventoryService({
         };
       });
       captureTrafficHistory(nextState);
+      captureOutboundTrafficHistory(nextState);
       if (typeof policyResult?.transportError === 'string') {
         commitPolicyFailure(new Error(policyResult.transportError));
       }