Ingest live 3proxy traffic from access logs

server/lib/traffic.ts

@@ -0,0 +1,184 @@
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import type { DailyTrafficBucket, ProxyUserRecord } from '../../src/shared/contracts';
+import type { RuntimePaths } from './config';
+
+const LOG_FILE_PREFIX = '3proxy.log';
+const RECENT_WINDOW_MS = 15 * 60 * 1000;
+const CONNECTION_WINDOW_MS = 5 * 60 * 1000;
+const HISTORY_DAYS = 5;
+
+export interface ObservedTraffic {
+  totalBytes: number;
+  liveConnections: number;
+  activeUsers: number;
+  daily: DailyTrafficBucket[];
+  userBytesByName: Map<string, number>;
+  recentUsers: Set<string>;
+}
+
+interface ParsedLogRecord {
+  timestamp: Date;
+  username: string;
+  bytes: number;
+}
+
+export async function readObservedTraffic(
+  runtimePaths: RuntimePaths,
+  users: ProxyUserRecord[],
+  now = new Date(),
+): Promise<ObservedTraffic> {
+  const records = await readParsedRecords(runtimePaths.logPath);
+  const currentDayKey = toDayKey(now);
+  const historyStart = startOfDay(now);
+  const history = buildHistoryDays(historyStart, HISTORY_DAYS);
+  const dailyTotals = new Map(history.map((entry) => [entry.key, 0]));
+  const userBytesByName = new Map(users.map((user) => [user.username, 0]));
+  const recentUsers = new Set<string>();
+  let liveConnections = 0;
+
+  records.forEach((record) => {
+    const dayKey = toDayKey(record.timestamp);
+    if (dailyTotals.has(dayKey)) {
+      dailyTotals.set(dayKey, (dailyTotals.get(dayKey) ?? 0) + record.bytes);
+    }
+
+    if (dayKey === currentDayKey && userBytesByName.has(record.username)) {
+      userBytesByName.set(record.username, (userBytesByName.get(record.username) ?? 0) + record.bytes);
+    }
+
+    const ageMs = now.getTime() - record.timestamp.getTime();
+    if (ageMs >= 0 && ageMs <= RECENT_WINDOW_MS) {
+      recentUsers.add(record.username);
+    }
+
+    if (ageMs >= 0 && ageMs <= CONNECTION_WINDOW_MS) {
+      liveConnections += 1;
+    }
+  });
+
+  const activeUsers = Array.from(userBytesByName.values()).filter((value) => value > 0).length;
+  const totalBytes = Array.from(userBytesByName.values()).reduce((sum, value) => sum + value, 0);
+  const daily = history.map((entry) => ({
+    day: entry.label,
+    bytes: dailyTotals.get(entry.key) ?? 0,
+    share: 0,
+  }));
+  const peak = Math.max(...daily.map((entry) => entry.bytes), 0);
+
+  daily.forEach((entry) => {
+    entry.share = peak > 0 ? entry.bytes / peak : 0;
+  });
+
+  return {
+    totalBytes,
+    liveConnections,
+    activeUsers,
+    daily,
+    userBytesByName,
+    recentUsers,
+  };
+}
+
+async function readParsedRecords(logPath: string): Promise<ParsedLogRecord[]> {
+  const logDir = path.dirname(logPath);
+
+  try {
+    const files = await fs.readdir(logDir);
+    const logFiles = files
+      .filter((file) => file === LOG_FILE_PREFIX || file.startsWith(`${LOG_FILE_PREFIX}.`))
+      .sort()
+      .slice(-HISTORY_DAYS - 2);
+
+    const parsed = await Promise.all(logFiles.map(async (file) => parseLogFile(path.join(logDir, file))));
+    return parsed.flat();
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
+      return [];
+    }
+
+    throw error;
+  }
+}
+
+async function parseLogFile(filePath: string): Promise<ParsedLogRecord[]> {
+  try {
+    const content = await fs.readFile(filePath, 'utf8');
+    return content
+      .split(/\r?\n/)
+      .map(parseLogLine)
+      .filter((record): record is ParsedLogRecord => record !== null);
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
+      return [];
+    }
+
+    throw error;
+  }
+}
+
+function parseLogLine(line: string): ParsedLogRecord | null {
+  const trimmed = line.trim();
+
+  if (!trimmed || trimmed.includes('Accepting connections') || trimmed.includes('Exiting thread')) {
+    return null;
+  }
+
+  const match = trimmed.match(
+    /^(\d{12}\.\d{3})\s+\d+\s+\d+\s+(\S+)\s+\S+\s+\S+\s+(\d+)\s+(\d+)\s+\d+\s+/,
+  );
+
+  if (!match) {
+    return null;
+  }
+
+  const [, rawTimestamp, username, bytesIn, bytesOut] = match;
+  if (username === '-') {
+    return null;
+  }
+
+  const timestamp = parse3proxyTimestamp(rawTimestamp);
+  if (!timestamp) {
+    return null;
+  }
+
+  return {
+    timestamp,
+    username,
+    bytes: Number(bytesIn) + Number(bytesOut),
+  };
+}
+
+function parse3proxyTimestamp(value: string): Date | null {
+  const match = value.match(/^(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})\.(\d{3})$/);
+  if (!match) {
+    return null;
+  }
+
+  const [, yy, mm, dd, hh, min, ss, ms] = match;
+  const year = 2000 + Number(yy);
+  const date = new Date(year, Number(mm) - 1, Number(dd), Number(hh), Number(min), Number(ss), Number(ms));
+
+  return Number.isNaN(date.getTime()) ? null : date;
+}
+
+function buildHistoryDays(today: Date, count: number) {
+  return Array.from({ length: count }, (_value, index) => {
+    const date = new Date(today);
+    date.setDate(today.getDate() - (count - index - 1));
+    return {
+      key: toDayKey(date),
+      label: date.toLocaleDateString('en-US', { weekday: 'short' }),
+    };
+  });
+}
+
+function startOfDay(date: Date): Date {
+  return new Date(date.getFullYear(), date.getMonth(), date.getDate());
+}
+
+function toDayKey(date: Date): string {
+  const month = String(date.getMonth() + 1).padStart(2, '0');
+  const day = String(date.getDate()).padStart(2, '0');
+  return `${date.getFullYear()}-${month}-${day}`;
+}