diff --git a/electron.vite.config.ts b/electron.vite.config.ts index b23b4da4..d2f5e07d 100644 --- a/electron.vite.config.ts +++ b/electron.vite.config.ts @@ -21,6 +21,7 @@ const legacyExternal = ['sysproxy-rs', 'electron', 'utf-8-validate', 'bufferutil export default defineConfig({ main: { + define: { __LEGACY_BUILD__: JSON.stringify(isLegacyBuild) }, plugins: isLegacyBuild ? [] : [externalizeDepsPlugin()], build: isLegacyBuild ? { rollupOptions: { external: legacyExternal, output: { format: 'cjs' } } } @@ -39,6 +40,7 @@ export default defineConfig({ } }, renderer: { + define: { __LEGACY_BUILD__: JSON.stringify(isLegacyBuild) }, build: { rollupOptions: { input: { diff --git a/src/main/config/app.ts b/src/main/config/app.ts index 76d42bfd..e9ca9ef2 100644 --- a/src/main/config/app.ts +++ b/src/main/config/app.ts @@ -13,6 +13,17 @@ import { setAppLogDisabled } from '../utils/logger' let appConfig: IAppConfig // config.yaml const appConfigWriteQueue = new WriteQueue() +const appConfigListeners = new Set<(config: IAppConfig) => void>() + +function notifyAppConfigListeners(): void { + for (const listener of appConfigListeners) listener(appConfig) +} + +export function subscribeAppConfig(listener: (config: IAppConfig) => void): () => void { + appConfigListeners.add(listener) + if (appConfig) listener(appConfig) + return () => appConfigListeners.delete(listener) +} function cloneDefaultConfig(): IAppConfig { return JSON.parse(JSON.stringify(defaultConfig)) as IAppConfig @@ -32,6 +43,7 @@ export async function getAppConfig(force = false): Promise { setCoreLogDisabled(mergedConfig.disableCoreLog === true) setAppLogDisabled(mergedConfig.disableAppLog === true) appConfig = mergedConfig + notifyAppConfigListeners() }) } if (typeof appConfig !== 'object') appConfig = cloneDefaultConfig() @@ -54,5 +66,6 @@ export async function patchAppConfig(patch: Partial): Promise setGlobalMaxLogFileSizeMB(nextConfig.maxLogFileSize) setCoreLogDisabled(nextConfig.disableCoreLog === true) setAppLogDisabled(nextConfig.disableAppLog === true) + notifyAppConfigListeners() }) } diff --git a/src/main/config/index.ts b/src/main/config/index.ts index 0f869e3a..08be947c 100644 --- a/src/main/config/index.ts +++ b/src/main/config/index.ts @@ -1,4 +1,4 @@ -export { getAppConfig, patchAppConfig } from './app' +export { getAppConfig, patchAppConfig, subscribeAppConfig } from './app' export { getControledMihomoConfig, patchControledMihomoConfig } from './controledMihomo' export { getProfile, diff --git a/src/main/core/mihomoApi.ts b/src/main/core/mihomoApi.ts index e382608f..027d6751 100644 --- a/src/main/core/mihomoApi.ts +++ b/src/main/core/mihomoApi.ts @@ -7,6 +7,7 @@ import { mainWindow } from '../window' import { tray } from '../resolve/tray' import { calcTraffic } from '../utils/calc' import { floatingWindow } from '../resolve/floatingWindow' +import { recordTrafficUsage } from '../traffic/recorder' import { createLogger } from '../utils/logger' import { mihomoWorkConfigPath } from '../utils/dirs' import { generateProfile, getRuntimeConfig } from './factory' @@ -632,7 +633,11 @@ const mihomoConnections = async (): Promise => { const data = e.data as string connectionsStream.retry = MAX_RETRY try { - mainWindow?.webContents.send('mihomoConnections', JSON.parse(data) as IMihomoConnectionsInfo) + const info = JSON.parse(data) as IMihomoConnectionsInfo + recordTrafficUsage(info) + if (__LEGACY_BUILD__ || mainWindow?.isVisible()) { + mainWindow?.webContents.send('mihomoConnections', info) + } } catch { // ignore } diff --git a/src/main/index.ts b/src/main/index.ts index a208c02f..5d1fbd23 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -5,7 +5,7 @@ import { app, dialog, ipcMain } from 'electron' import i18next from 'i18next' import { initI18n } from '../shared/i18n' import { registerIpcMainHandlers } from './utils/ipc' -import { getAppConfig, patchAppConfig } from './config' +import { getAppConfig, patchAppConfig, subscribeAppConfig } from './config' import { beginCoreInitialization, completeCoreInitialization, @@ -42,6 +42,7 @@ import { getSystemLanguage } from './lifecycle' import { configureAppPaths } from './utils/dirs' +import { setTrafficUsageEnabled } from './traffic/recorder' async function getWindowsPowerShellMajorVersion(): Promise { // 仅 PS 3.0+ 写入 \3\ 键（\1\ 键恒为 2.0，不可用）。 @@ -87,6 +88,7 @@ async function ensureSupportedWindowsPowerShell(): Promise { } configureAppPaths() +subscribeAppConfig((config) => setTrafficUsageEnabled(config.enableTrafficLogger === true)) const mainLogger = createLogger('Main') diff --git a/src/main/lifecycle.ts b/src/main/lifecycle.ts index 11cae364..fcee4ba1 100644 --- a/src/main/lifecycle.ts +++ b/src/main/lifecycle.ts @@ -6,6 +6,7 @@ import { app, powerMonitor } from 'electron' import { stopCoreForExit, cleanupCoreWatcher } from './core/manager' import { primeAdminPrivilegesCache } from './core/admin' import { triggerSysProxy, disableSysProxySync } from './sys/sysproxy' +import { closeTrafficUsage } from './traffic/recorder' import { exePath } from './utils/dirs' import { saveMainWindowState } from './window' @@ -109,7 +110,7 @@ export function setupAppLifecycle(): void { sysProxyDisabled = true } - const cleanupTasks: Promise[] = [stopCoreForExit()] + const cleanupTasks: Promise[] = [stopCoreForExit(), closeTrafficUsage()] if (process.platform === 'darwin') { cleanupTasks.push( triggerSysProxy(false, { helperTimeout: 750, force: true }).then(() => { diff --git a/src/main/traffic/database-worker.ts b/src/main/traffic/database-worker.ts new file mode 100644 index 00000000..ae658ec4 --- /dev/null +++ b/src/main/traffic/database-worker.ts @@ -0,0 +1,289 @@ +import { parentPort, workerData } from 'worker_threads' +import { DatabaseSync, type StatementSync } from 'node:sqlite' +import { + TRAFFIC_USAGE_RESULT_LIMIT, + TRAFFIC_USAGE_RETENTION, + TRAFFIC_USAGE_RESOLUTIONS, + trafficUsageResolution, + type TrafficUsageAggregate, + type TrafficUsageDimension, + type TrafficUsageOverview, + type TrafficUsageSample, + type TrafficUsageTrendPoint +} from '../../shared/trafficUsage' +import type { + TrafficDatabaseRequest, + TrafficDatabaseResponse, + TrafficUsageImportBatch, + TrafficUsageWriteBatch +} from './databaseMessages' + +const port = parentPort +if (!port) throw new Error('Traffic database worker has no parent port') + +const databasePath = (workerData as { databasePath: string }).databasePath +const database = new DatabaseSync(databasePath, { timeout: 5000 }) + +const columns: Record = { + sourceIP: 'source_ip', + host: 'host', + outbound: 'outbound', + process: 'process' +} + +function migrateDatabase(): void { + const version = Number( + (database.prepare('PRAGMA user_version').get() as { user_version?: number }).user_version ?? 0 + ) + if (version > 1) throw new Error(`Unsupported traffic database version: ${version}`) + if (version === 0) { + database.exec(` + CREATE TABLE traffic_usage ( + resolution INTEGER NOT NULL, + bucket INTEGER NOT NULL, + source_ip TEXT NOT NULL, + host TEXT NOT NULL, + outbound TEXT NOT NULL, + process TEXT NOT NULL, + upload INTEGER NOT NULL, + download INTEGER NOT NULL, + samples INTEGER NOT NULL, + PRIMARY KEY (resolution, bucket, source_ip, host, outbound, process) + ) WITHOUT ROWID, STRICT; + CREATE TABLE traffic_usage_batches ( + id TEXT PRIMARY KEY, + created_at INTEGER NOT NULL + ) WITHOUT ROWID, STRICT; + PRAGMA user_version = 1; + `) + } +} + +database.exec(` + PRAGMA journal_mode = WAL; + PRAGMA synchronous = NORMAL; + PRAGMA temp_store = FILE; + PRAGMA cache_size = -8192; + PRAGMA journal_size_limit = 8388608; +`) +migrateDatabase() + +const upsert = database.prepare(` + INSERT INTO traffic_usage ( + resolution, bucket, source_ip, host, outbound, process, upload, download, samples + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (resolution, bucket, source_ip, host, outbound, process) DO UPDATE SET + upload = upload + excluded.upload, + download = download + excluded.download, + samples = samples + excluded.samples +`) +const addBatch = database.prepare( + 'INSERT OR IGNORE INTO traffic_usage_batches (id, created_at) VALUES (?, ?)' +) +const cleanupUsage = database.prepare( + 'DELETE FROM traffic_usage WHERE resolution = ? AND bucket < ?' +) +const cleanupBatches = database.prepare( + "DELETE FROM traffic_usage_batches WHERE created_at < ? AND id NOT LIKE 'indexeddb:%'" +) +const totalsQuery = database.prepare(` + SELECT SUM(upload) AS upload, SUM(download) AS download, SUM(samples) AS count + FROM traffic_usage + WHERE resolution = ? AND bucket BETWEEN ? AND ? +`) +const trendQuery = database.prepare(` + SELECT CAST(bucket / ? AS INTEGER) * ? AS timestamp, + SUM(upload) AS upload, + SUM(download) AS download + FROM traffic_usage + WHERE resolution = ? AND bucket BETWEEN ? AND ? + GROUP BY CAST(bucket / ? AS INTEGER) + ORDER BY timestamp +`) +const aggregateStatements = new Map() +let lastCleanup = 0 + +function runTransaction(action: () => void): void { + database.exec('BEGIN IMMEDIATE') + try { + action() + database.exec('COMMIT') + } catch (error) { + database.exec('ROLLBACK') + throw error + } +} + +function writeRecord(record: TrafficUsageSample, resolution: number, bucket: number): void { + upsert.run( + resolution, + bucket, + record.sourceIP, + record.host, + record.outbound, + record.process, + record.upload, + record.download, + record.count + ) +} + +function persistBatch(id: string, write: () => void): void { + runTransaction(() => { + const inserted = addBatch.run(id, Date.now()) + if (inserted.changes === 0) return + write() + }) + + const now = Date.now() + if (now - lastCleanup < 24 * 60 * 60 * 1000) return + runTransaction(() => { + for (const resolution of TRAFFIC_USAGE_RESOLUTIONS) { + cleanupUsage.run(resolution, now - TRAFFIC_USAGE_RETENTION[resolution]) + } + cleanupBatches.run(now - 2 * 24 * 60 * 60 * 1000) + }) + lastCleanup = now +} + +function writeBatch(batch: TrafficUsageWriteBatch): void { + persistBatch(batch.id, () => { + for (const sample of batch.samples) { + for (const resolution of TRAFFIC_USAGE_RESOLUTIONS) { + writeRecord(sample, resolution, Math.floor(sample.bucket / resolution) * resolution) + } + } + }) +} + +function importBatch(batch: TrafficUsageImportBatch): void { + persistBatch(batch.id, () => { + for (const record of batch.records) { + writeRecord(record, record.resolution, record.bucket) + } + }) +} + +function mapAggregate(row: Record): TrafficUsageAggregate { + const upload = Number(row.upload ?? 0) + const download = Number(row.download ?? 0) + return { + label: String(row.label ?? ''), + upload, + download, + total: upload + download, + count: Number(row.count ?? 0) + } +} + +function aggregateQuery( + groupBy: TrafficUsageDimension, + filters: Partial>, + startTime: number, + endTime: number +): TrafficUsageAggregate[] { + const resolution = trafficUsageResolution(startTime, endTime) + const values: (string | number)[] = [ + resolution, + Math.floor(startTime / resolution) * resolution, + Math.floor(endTime / resolution) * resolution + ] + const clauses = ['resolution = ?', 'bucket BETWEEN ? AND ?'] + for (const [dimension, value] of Object.entries(filters) as [TrafficUsageDimension, string][]) { + clauses.push(`${columns[dimension]} = ?`) + values.push(value) + } + const column = columns[groupBy] + const sql = ` + SELECT ${column} AS label, + SUM(upload) AS upload, + SUM(download) AS download, + SUM(samples) AS count + FROM traffic_usage + WHERE ${clauses.join(' AND ')} + GROUP BY ${column} + ORDER BY SUM(upload) + SUM(download) DESC + LIMIT ${TRAFFIC_USAGE_RESULT_LIMIT} + ` + let statement = aggregateStatements.get(sql) + if (!statement) { + statement = database.prepare(sql) + aggregateStatements.set(sql, statement) + } + return (statement.all(...values) as Record[]).map(mapAggregate) +} + +function queryOverview( + payload: Extract['payload'] +): TrafficUsageOverview { + const resolution = trafficUsageResolution(payload.startTime, payload.endTime) + const start = Math.floor(payload.startTime / resolution) * resolution + const end = Math.floor(payload.endTime / resolution) * resolution + const rankings = aggregateQuery(payload.type, {}, payload.startTime, payload.endTime) + const totalsRow = totalsQuery.get(resolution, start, end) as Record + const trend = ( + trendQuery.all( + payload.bucketSizeMs, + payload.bucketSizeMs, + resolution, + start, + end, + payload.bucketSizeMs + ) as Record[] + ).map((row): TrafficUsageTrendPoint => ({ + timestamp: Number(row.timestamp), + upload: Number(row.upload ?? 0), + download: Number(row.download ?? 0) + })) + const upload = Number(totalsRow.upload ?? 0) + const download = Number(totalsRow.download ?? 0) + return { + rankings, + trend, + totals: { + upload, + download, + total: upload + download, + count: Number(totalsRow.count ?? 0) + } + } +} + +function handleRequest(request: TrafficDatabaseRequest): TrafficDatabaseResponse['result'] { + switch (request.action) { + case 'write': + writeBatch(request.payload) + return + case 'import': + importBatch(request.payload) + return + case 'overview': + return queryOverview(request.payload) + case 'breakdown': + return aggregateQuery( + request.payload.groupBy, + request.payload.filters, + request.payload.startTime, + request.payload.endTime + ) + case 'clear': + runTransaction(() => + database.exec('DELETE FROM traffic_usage; DELETE FROM traffic_usage_batches') + ) + return + case 'close': + database.close() + return + } +} + +port.on('message', (request: TrafficDatabaseRequest) => { + let response: TrafficDatabaseResponse + try { + response = { id: request.id, result: handleRequest(request) } + } catch (error) { + response = { id: request.id, error: error instanceof Error ? error.message : String(error) } + } + port.postMessage(response) + if (request.action === 'close') port.close() +}) diff --git a/src/main/traffic/database.ts b/src/main/traffic/database.ts new file mode 100644 index 00000000..b4c4ae22 --- /dev/null +++ b/src/main/traffic/database.ts @@ -0,0 +1,161 @@ +import type { Worker } from 'worker_threads' +import type { + TrafficUsageAggregate, + TrafficUsageBreakdownQuery, + TrafficUsageDimension, + TrafficUsageImportBatch, + TrafficUsageOverview, + TrafficUsageWriteBatch +} from '../../shared/trafficUsage' +import { TRAFFIC_USAGE_MIGRATION_CHUNK_SIZE } from '../../shared/trafficUsage' +import { trafficUsageDatabasePath } from '../utils/dirs' +import createTrafficDatabaseWorker from './database-worker?nodeWorker' +import type { TrafficDatabaseRequest, TrafficDatabaseResponse } from './databaseMessages' + +type PendingRequest = { + resolve: (value: unknown) => void + reject: (reason: unknown) => void + timeout?: NodeJS.Timeout +} + +type TrafficDatabaseRequestWithoutId = TrafficDatabaseRequest extends infer Request + ? Request extends { id: number } + ? Omit + : never + : never + +class TrafficDatabaseClient { + private worker: Worker | null = null + private nextId = 1 + private readonly pending = new Map() + private closing = false + + write(batch: TrafficUsageWriteBatch): Promise { + return this.request({ action: 'write', payload: batch }) + } + + import(batch: TrafficUsageImportBatch): Promise { + return this.request({ action: 'import', payload: batch }) + } + + overview( + type: TrafficUsageDimension, + startTime: number, + endTime: number, + bucketSizeMs: number + ): Promise { + return this.request( + { + action: 'overview', + payload: { type, startTime, endTime, bucketSizeMs } + }, + 30_000 + ) + } + + breakdown(query: TrafficUsageBreakdownQuery): Promise { + return this.request({ action: 'breakdown', payload: query }, 30_000) + } + + clear(): Promise { + return this.request({ action: 'clear' }, 30_000) + } + + async close(): Promise { + if (!this.worker || this.closing) return + this.closing = true + try { + await this.request({ action: 'close' }, 5000) + } finally { + this.worker = null + this.closing = false + } + } + + private getWorker(): Worker { + if (this.worker) return this.worker + if (__LEGACY_BUILD__) throw new Error('Traffic database is unavailable in the legacy build') + + const worker = createTrafficDatabaseWorker({ + workerData: { databasePath: trafficUsageDatabasePath() } + }) + worker.on('message', (response: TrafficDatabaseResponse) => { + const pending = this.pending.get(response.id) + if (!pending) return + this.pending.delete(response.id) + if (pending.timeout) clearTimeout(pending.timeout) + if (response.error) pending.reject(new Error(response.error)) + else pending.resolve(response.result) + }) + worker.on('error', (error) => this.handleWorkerFailure(worker, error)) + worker.on('exit', (code) => { + if (this.worker !== worker) return + if (code !== 0 && !this.closing) + this.handleWorkerFailure( + worker, + new Error(`Traffic database worker exited with code ${code}`) + ) + else this.worker = null + }) + worker.unref() + this.worker = worker + return worker + } + + private request(request: TrafficDatabaseRequestWithoutId, timeoutMs?: number): Promise { + const id = this.nextId++ + return new Promise((resolve, reject) => { + const pending: PendingRequest = { + resolve: (value) => resolve(value as T), + reject + } + if (timeoutMs) { + pending.timeout = setTimeout(() => { + this.pending.delete(id) + reject(new Error(`Traffic database request timed out: ${request.action}`)) + }, timeoutMs) + } + this.pending.set(id, pending) + try { + this.getWorker().postMessage({ ...request, id } as TrafficDatabaseRequest) + } catch (error) { + this.pending.delete(id) + if (pending.timeout) clearTimeout(pending.timeout) + reject(error) + } + }) + } + + private handleWorkerFailure(worker: Worker, reason: unknown): void { + if (this.worker !== worker) return + const error = reason instanceof Error ? reason : new Error(String(reason)) + for (const pending of this.pending.values()) { + if (pending.timeout) clearTimeout(pending.timeout) + pending.reject(error) + } + this.pending.clear() + this.worker = null + } +} + +const database = new TrafficDatabaseClient() + +export const writeTrafficUsage = (batch: TrafficUsageWriteBatch): Promise => + database.write(batch) +export const importTrafficUsage = (batch: TrafficUsageImportBatch): Promise => { + if (batch.records.length > TRAFFIC_USAGE_MIGRATION_CHUNK_SIZE) { + throw new Error('Traffic usage migration batch is too large') + } + return database.import({ ...batch, id: `indexeddb:${batch.id}` }) +} +export const queryTrafficUsageOverview = ( + type: TrafficUsageDimension, + startTime: number, + endTime: number, + bucketSizeMs: number +): Promise => database.overview(type, startTime, endTime, bucketSizeMs) +export const queryTrafficUsageBreakdown = ( + query: TrafficUsageBreakdownQuery +): Promise => database.breakdown(query) +export const clearTrafficUsage = (): Promise => database.clear() +export const closeTrafficUsageDatabase = (): Promise => database.close() diff --git a/src/main/traffic/databaseMessages.ts b/src/main/traffic/databaseMessages.ts new file mode 100644 index 00000000..a24b7b08 --- /dev/null +++ b/src/main/traffic/databaseMessages.ts @@ -0,0 +1,31 @@ +import type { + TrafficUsageBreakdownQuery, + TrafficUsageImportBatch, + TrafficUsageOverview, + TrafficUsageWriteBatch +} from '../../shared/trafficUsage' + +export type { TrafficUsageImportBatch, TrafficUsageWriteBatch } from '../../shared/trafficUsage' + +export type TrafficDatabaseRequest = + | { id: number; action: 'write'; payload: TrafficUsageWriteBatch } + | { id: number; action: 'import'; payload: TrafficUsageImportBatch } + | { + id: number + action: 'overview' + payload: { + type: TrafficUsageBreakdownQuery['groupBy'] + startTime: number + endTime: number + bucketSizeMs: number + } + } + | { id: number; action: 'breakdown'; payload: TrafficUsageBreakdownQuery } + | { id: number; action: 'clear' } + | { id: number; action: 'close' } + +export interface TrafficDatabaseResponse { + id: number + result?: TrafficUsageOverview | TrafficUsageOverview['rankings'] | undefined + error?: string +} diff --git a/src/main/traffic/recorder.ts b/src/main/traffic/recorder.ts new file mode 100644 index 00000000..03b80115 --- /dev/null +++ b/src/main/traffic/recorder.ts @@ -0,0 +1,104 @@ +import { createLogger } from '../utils/logger' +import { + TRAFFIC_USAGE_FLUSH_THRESHOLD, + TrafficUsageAccumulator, + type TrafficUsageSample, + type TrafficUsageWriteBatch +} from '../../shared/trafficUsage' +import { closeTrafficUsageDatabase, writeTrafficUsage } from './database' + +const FLUSH_DELAY_MS = 5000 +const recorderLogger = createLogger('TrafficUsage') +const accumulator = new TrafficUsageAccumulator() + +let enabled = false +let flushTimer: NodeJS.Timeout | null = null +let inFlight: Promise | null = null +let retryBatch: TrafficUsageWriteBatch | null = null +let batchSequence = 0 +let lastDroppedCount = 0 +let shuttingDown = false + +function clearFlushTimer(): void { + if (!flushTimer) return + clearTimeout(flushTimer) + flushTimer = null +} + +function scheduleFlush(delay = FLUSH_DELAY_MS): void { + if (__LEGACY_BUILD__ || !enabled || shuttingDown || flushTimer) return + flushTimer = setTimeout(() => { + flushTimer = null + void flushTrafficUsage() + }, delay) + flushTimer.unref() +} + +function nextBatch(samples: TrafficUsageSample[]): TrafficUsageWriteBatch { + return { + id: `${process.pid}-${Date.now()}-${batchSequence++}`, + samples + } +} + +export function setTrafficUsageEnabled(nextEnabled: boolean): void { + if (__LEGACY_BUILD__ || enabled === nextEnabled) return + enabled = nextEnabled + clearFlushTimer() + accumulator.setEnabled(nextEnabled) + if (!nextEnabled) retryBatch = null +} + +export function recordTrafficUsage(info: IMihomoConnectionsInfo): void { + if (__LEGACY_BUILD__ || !enabled) return + const shouldFlush = accumulator.addSnapshot(info) + if (accumulator.droppedCount !== lastDroppedCount) { + lastDroppedCount = accumulator.droppedCount + recorderLogger.warn( + `Dropped ${lastDroppedCount} traffic usage records after reaching the pending limit` + ) + } + if (shouldFlush && !retryBatch) void flushTrafficUsage() + else if (retryBatch || accumulator.pendingSize > 0) scheduleFlush() +} + +export async function flushTrafficUsage(): Promise { + if (__LEGACY_BUILD__ || !enabled || inFlight) return inFlight ?? Promise.resolve() + + const batch = retryBatch ?? nextBatch(accumulator.takePending()) + if (batch.samples.length === 0) return + retryBatch = batch + inFlight = writeTrafficUsage(batch) + .then(() => { + if (retryBatch?.id === batch.id) retryBatch = null + }) + .catch((error) => { + recorderLogger.warn('Failed to persist traffic usage', error) + }) + .finally(() => { + inFlight = null + if (enabled && (retryBatch || accumulator.pendingSize > 0)) { + scheduleFlush( + accumulator.pendingSize >= TRAFFIC_USAGE_FLUSH_THRESHOLD && !retryBatch + ? 0 + : FLUSH_DELAY_MS + ) + } + }) + return inFlight +} + +export async function closeTrafficUsage(): Promise { + if (__LEGACY_BUILD__) return + shuttingDown = true + clearFlushTimer() + if (inFlight) await inFlight + clearFlushTimer() + if (enabled && (retryBatch || accumulator.pendingSize > 0)) { + await flushTrafficUsage() + if (inFlight) await inFlight + } + enabled = false + accumulator.setEnabled(false) + await closeTrafficUsageDatabase() +} diff --git a/src/main/utils/dirs.ts b/src/main/utils/dirs.ts index 0771ca0c..2af5ecb4 100644 --- a/src/main/utils/dirs.ts +++ b/src/main/utils/dirs.ts @@ -98,6 +98,10 @@ export function appConfigPath(): string { return path.join(dataDir(), 'config.yaml') } +export function trafficUsageDatabasePath(): string { + return path.join(dataDir(), 'traffic-usage.db') +} + export function controledMihomoConfigPath(): string { return path.join(dataDir(), 'mihomo.yaml') } diff --git a/src/main/utils/ipc.ts b/src/main/utils/ipc.ts index 29de2fd8..ea992b23 100644 --- a/src/main/utils/ipc.ts +++ b/src/main/utils/ipc.ts @@ -132,6 +132,12 @@ import { patchPluginItem } from '../resolve/plugin' import { getPluginConfig } from '../config/plugin' +import { + clearTrafficUsage, + importTrafficUsage, + queryTrafficUsageBreakdown, + queryTrafficUsageOverview +} from '../traffic/database' import { getImageDataURL } from './image' import { get as httpGet } from './chromeRequest' import { getIconDataURL } from './icon' @@ -236,6 +242,10 @@ const asyncHandlers: Record = { mihomoCloseAllConnections, mihomoRules, mihomoRulesDisable, + queryTrafficUsageOverview, + queryTrafficUsageBreakdown, + importTrafficUsage, + clearTrafficUsage, mihomoProxies, mihomoGroups, mihomoProxyProviders, diff --git a/src/preload/index.ts b/src/preload/index.ts index 26258d0c..d5af8124 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -24,6 +24,10 @@ const validInvokeChannels = [ 'patchMihomoConfig', 'mihomoSmartGroupWeights', 'mihomoSmartFlushCache', + 'queryTrafficUsageOverview', + 'queryTrafficUsageBreakdown', + 'importTrafficUsage', + 'clearTrafficUsage', // AutoRun 'checkAutoRun', 'enableAutoRun', diff --git a/src/renderer/src/hooks/use-traffic-logger.ts b/src/renderer/src/hooks/use-traffic-logger.ts index 85b44a7c..b8a987f7 100644 --- a/src/renderer/src/hooks/use-traffic-logger.ts +++ b/src/renderer/src/hooks/use-traffic-logger.ts @@ -1,172 +1,90 @@ import { useEffect, useRef } from 'react' -import { db, type DataUsageLog } from '@renderer/utils/db' +import { legacyTrafficUsageDatabase } from '@renderer/utils/legacy-traffic-db' +import { importTrafficUsage } from '@renderer/utils/ipc' +import { + TRAFFIC_USAGE_FLUSH_THRESHOLD, + TrafficUsageAccumulator +} from '../../../shared/trafficUsage' const FLUSH_DELAY_MS = 5000 -const RETENTION_MS = 30 * 24 * 60 * 60 * 1000 - -interface TrafficSnapshot { - upload: number - download: number -} export function useTrafficLogger(enabled = true): void { - const connectionLastDataRef = useRef(new Map()) - const logBufferRef = useRef([]) - const flushTimeoutRef = useRef | null>(null) - const lastTotalsRef = useRef({ upload: 0, download: 0 }) - const enabledRef = useRef(enabled) - const runIdRef = useRef(0) - enabledRef.current = enabled + const accumulatorRef = useRef(new TrafficUsageAccumulator()) useEffect(() => { + if (__LEGACY_BUILD__) return + void legacyTrafficUsageDatabase + .migrateToBackend(importTrafficUsage) + .catch((error) => console.error('[TrafficLogger] migration failed', error)) + }, []) + + useEffect(() => { + const accumulator = accumulatorRef.current + const active = __LEGACY_BUILD__ && enabled + accumulator.setEnabled(active) + if (!active) return + + let disposed = false + let flushTimer: ReturnType | null = null + let inFlight: Promise | null = null + const clearFlushTimer = (): void => { - if (flushTimeoutRef.current) { - clearTimeout(flushTimeoutRef.current) - flushTimeoutRef.current = null - } + if (!flushTimer) return + clearTimeout(flushTimer) + flushTimer = null } - const resetRuntimeState = (): void => { - clearFlushTimer() - connectionLastDataRef.current.clear() - logBufferRef.current = [] - lastTotalsRef.current = { upload: 0, download: 0 } + const scheduleFlush = (delay = FLUSH_DELAY_MS): void => { + if (disposed || flushTimer) return + flushTimer = setTimeout(() => { + flushTimer = null + void flush() + }, delay) } - const isCurrentRun = (runId: number): boolean => - enabledRef.current && runIdRef.current === runId + const flush = async (): Promise => { + if (disposed || inFlight) return inFlight ?? Promise.resolve() + const records = accumulator.takePending() + if (records.length === 0) return - if (!enabled) { - runIdRef.current += 1 - resetRuntimeState() - return + let failed = false + inFlight = legacyTrafficUsageDatabase + .upsert(records) + .catch((error) => { + failed = true + if (!disposed) accumulator.merge(records) + console.error('[TrafficLogger] flush failed', error) + }) + .finally(() => { + inFlight = null + if (!disposed && accumulator.pendingSize > 0) { + scheduleFlush( + !failed && accumulator.pendingSize >= TRAFFIC_USAGE_FLUSH_THRESHOLD + ? 0 + : FLUSH_DELAY_MS + ) + } + }) + return inFlight } - runIdRef.current += 1 - const runId = runIdRef.current - const enabledAt = Date.now() - resetRuntimeState() - - const flushLogs = async (): Promise => { - if (!isCurrentRun(runId)) { - logBufferRef.current = [] - return - } - - const toFlush = logBufferRef.current - if (toFlush.length === 0) return - logBufferRef.current = [] - - try { - await db.open() - if (!isCurrentRun(runId)) return - await db.addLogs(toFlush) - if (!isCurrentRun(runId)) return - await db.cleanup(Date.now() - RETENTION_MS) - } catch (e) { - console.error('[TrafficLogger] flush failed', e) - } - } - - const scheduleFlush = (): void => { - if (!isCurrentRun(runId)) return - if (flushTimeoutRef.current) return - flushTimeoutRef.current = setTimeout(async () => { - flushTimeoutRef.current = null - if (!isCurrentRun(runId)) { - logBufferRef.current = [] - return - } - await flushLogs() - }, FLUSH_DELAY_MS) - } - - const shouldLogInitialSnapshot = (conn: IMihomoConnectionDetail): boolean => { - const startAt = Date.parse(conn.start) - return Number.isFinite(startAt) && startAt >= enabledAt - } - - const handler = (_e: unknown, ...args: unknown[]): void => { - if (!isCurrentRun(runId)) return - + const handler = (_event: unknown, ...args: unknown[]): void => { const info = args[0] as IMihomoConnectionsInfo | undefined if (!info) return - - const uploadTotal = info.uploadTotal || 0 - const downloadTotal = info.downloadTotal || 0 - - // Detect service restart (totals decreased) - if ( - uploadTotal < lastTotalsRef.current.upload || - downloadTotal < lastTotalsRef.current.download - ) { - connectionLastDataRef.current.clear() - logBufferRef.current = [] - } - lastTotalsRef.current = { upload: uploadTotal, download: downloadTotal } - - const connections = info.connections ?? [] - if (connections.length === 0) { - connectionLastDataRef.current.clear() - return - } - - const now = Date.now() - let hasDeltas = false - const activeConnectionIds = new Set() - - for (const conn of connections) { - activeConnectionIds.add(conn.id) - - const currentUpload = conn.upload || 0 - const currentDownload = conn.download || 0 - const last = connectionLastDataRef.current.get(conn.id) - - connectionLastDataRef.current.set(conn.id, { - upload: currentUpload, - download: currentDownload - }) - - const uploadDelta = last - ? Math.max(0, currentUpload - last.upload) - : shouldLogInitialSnapshot(conn) - ? currentUpload - : 0 - const downloadDelta = last - ? Math.max(0, currentDownload - last.download) - : shouldLogInitialSnapshot(conn) - ? currentDownload - : 0 - - if (uploadDelta === 0 && downloadDelta === 0) continue - - hasDeltas = true - logBufferRef.current.push({ - timestamp: now, - sourceIP: conn.metadata.sourceIP || 'Inner', - host: conn.metadata.host || conn.metadata.destinationIP || 'Unknown', - process: conn.metadata.process || 'Unknown', - outbound: conn.chains?.[0] || 'DIRECT', - upload: uploadDelta, - download: downloadDelta - }) - } - - for (const id of connectionLastDataRef.current.keys()) { - if (!activeConnectionIds.has(id)) { - connectionLastDataRef.current.delete(id) - } - } - - if (hasDeltas) scheduleFlush() + if (accumulator.addSnapshot(info)) void flush() + else if (accumulator.pendingSize > 0) scheduleFlush() } + void legacyTrafficUsageDatabase + .migrateLegacyLogs() + .catch((error) => console.error('[TrafficLogger] migration failed', error)) window.electron.ipcRenderer.on('mihomoConnections', handler) return (): void => { + disposed = true + clearFlushTimer() window.electron.ipcRenderer.removeListener('mihomoConnections', handler) - runIdRef.current += 1 - resetRuntimeState() + accumulator.setEnabled(false) } }, [enabled]) } diff --git a/src/renderer/src/pages/traffic.tsx b/src/renderer/src/pages/traffic.tsx index 5be4d17a..5da7bf05 100644 --- a/src/renderer/src/pages/traffic.tsx +++ b/src/renderer/src/pages/traffic.tsx @@ -7,10 +7,10 @@ import { getSubStatsByHost, getDevicesByHost, getProxyStatsByHost, + clearTrafficUsageData, type AggregatedData, type DataUsageType } from '@renderer/utils/dataUsage' -import { db } from '@renderer/utils/db' import { Button, Tab, Tabs } from '@heroui/react' import React, { useCallback, useEffect, useRef, useState } from 'react' import { useTranslation } from 'react-i18next' @@ -65,24 +65,14 @@ const TrafficPage: React.FC = () => { isCancelled: () => boolean = () => false ) => { const { start, end, bucketSizeMs: bms } = getTimeRange(timeRange) - const { rankings: agg, trend } = await getTrafficOverview(activeView, start, end, bms) + const { rankings: agg, trend, totals } = await getTrafficOverview(activeView, start, end, bms) if (isCancelled() || generation !== loadGenerationRef.current) return setBucketSizeMs(bms) setRankings(agg) setTrendData(trend) - setTotalStats( - agg.reduce( - (acc, r) => ({ - upload: acc.upload + r.upload, - download: acc.download + r.download, - total: acc.total + r.total, - count: acc.count + r.count - }), - { upload: 0, download: 0, total: 0, count: 0 } - ) - ) + setTotalStats(totals) if (resetSelection) { setSelectedRow(null) @@ -98,22 +88,43 @@ const TrafficPage: React.FC = () => { const generation = ++loadGenerationRef.current let refreshTimer: ReturnType | null = null let cancelled = false + let refreshing = false + let resetSelection = true - const refresh = async (resetSelection: boolean): Promise => { - await load(resetSelection, generation, () => cancelled) - if (cancelled || generation !== loadGenerationRef.current) return + const clearRefreshTimer = (): void => { + if (refreshTimer === null) return + clearTimeout(refreshTimer) + refreshTimer = null + } + + const refresh = async (): Promise => { + if (cancelled || document.hidden || refreshing) return + refreshing = true + try { + await load(resetSelection, generation, () => cancelled || document.hidden) + } finally { + refreshing = false + } + if (cancelled || document.hidden || generation !== loadGenerationRef.current) return + resetSelection = false refreshTimer = setTimeout(() => { - void refresh(false) + refreshTimer = null + void refresh() }, AUTO_REFRESH_INTERVAL_MS) } - void refresh(true) + const handleVisibilityChange = (): void => { + if (document.hidden) clearRefreshTimer() + else void refresh() + } + + document.addEventListener('visibilitychange', handleVisibilityChange) + void refresh() return () => { cancelled = true - if (refreshTimer !== null) { - clearTimeout(refreshTimer) - } + clearRefreshTimer() + document.removeEventListener('visibilitychange', handleVisibilityChange) } }, [load]) @@ -175,7 +186,7 @@ const TrafficPage: React.FC = () => { ) const handleClearAll = useCallback(async () => { - await db.clearAll() + await clearTrafficUsageData() await load() }, [load]) diff --git a/src/renderer/src/utils/dataUsage.ts b/src/renderer/src/utils/dataUsage.ts index dc7f1967..08fbbd21 100644 --- a/src/renderer/src/utils/dataUsage.ts +++ b/src/renderer/src/utils/dataUsage.ts @@ -1,59 +1,51 @@ -import { db, type DataUsageLog } from '@renderer/utils/db' +import { + clearTrafficUsage, + queryTrafficUsageBreakdown, + queryTrafficUsageOverview +} from '@renderer/utils/ipc' +import { legacyTrafficUsageDatabase } from '@renderer/utils/legacy-traffic-db' +import type { + TrafficUsageAggregate, + TrafficUsageBreakdownQuery, + TrafficUsageDimension, + TrafficUsageOverview, + TrafficUsageTrendPoint +} from '../../../shared/trafficUsage' -export type DataUsageType = 'sourceIP' | 'host' | 'outbound' | 'process' +export type DataUsageType = TrafficUsageDimension +export type AggregatedData = TrafficUsageAggregate -export interface AggregatedData { - label: string - upload: number - download: number - total: number - count: number -} - -interface TrafficTrendPoint { - timestamp: number - upload: number - download: number -} - -function addAggregatedLog( - map: Map, - label: string, - log: DataUsageLog -): void { - const existing = map.get(label) - if (existing) { - existing.upload += log.upload - existing.download += log.download - existing.total += log.upload + log.download - existing.count += 1 - return +function fillTrend( + trend: TrafficUsageTrendPoint[], + startTime: number, + endTime: number, + bucketSizeMs: number +): TrafficUsageTrendPoint[] { + const values = new Map(trend.map((point) => [point.timestamp, point])) + const result: TrafficUsageTrendPoint[] = [] + const first = Math.floor(startTime / bucketSizeMs) * bucketSizeMs + const last = Math.floor(endTime / bucketSizeMs) * bucketSizeMs + for (let timestamp = first; timestamp <= last; timestamp += bucketSizeMs) { + result.push(values.get(timestamp) ?? { timestamp, upload: 0, download: 0 }) } - - map.set(label, { - label, - upload: log.upload, - download: log.download, - total: log.upload + log.download, - count: 1 - }) + return result } -function sortAggregatedData(map: Map): AggregatedData[] { - return Array.from(map.values()).sort((a, b) => b.total - a.total) -} - -function getDimensionLabel(type: DataUsageType, log: DataUsageLog): string { - switch (type) { - case 'sourceIP': - return log.sourceIP - case 'host': - return log.host - case 'outbound': - return log.outbound - case 'process': - return log.process +async function overview( + type: DataUsageType, + startTime: number, + endTime: number, + bucketSizeMs: number +): Promise { + if (__LEGACY_BUILD__) { + return legacyTrafficUsageDatabase.overview(type, startTime, endTime, bucketSizeMs) } + return queryTrafficUsageOverview(type, startTime, endTime, bucketSizeMs) +} + +async function breakdown(query: TrafficUsageBreakdownQuery): Promise { + if (__LEGACY_BUILD__) return legacyTrafficUsageDatabase.breakdown(query) + return queryTrafficUsageBreakdown(query) } export async function getTrafficOverview( @@ -61,96 +53,51 @@ export async function getTrafficOverview( startTime: number, endTime: number, bucketSizeMs: number -): Promise<{ rankings: AggregatedData[]; trend: TrafficTrendPoint[] }> { - const rankings = new Map() - const buckets = new Map() - - for (let time = startTime; time <= endTime; time += bucketSizeMs) { - buckets.set(Math.floor(time / bucketSizeMs) * bucketSizeMs, { upload: 0, download: 0 }) - } - - await db.iterate(startTime, endTime, (log) => { - addAggregatedLog(rankings, getDimensionLabel(type, log), log) - - const bucket = buckets.get(Math.floor(log.timestamp / bucketSizeMs) * bucketSizeMs) - if (bucket) { - bucket.upload += log.upload - bucket.download += log.download - } - }) - +): Promise { + const result = await overview(type, startTime, endTime, bucketSizeMs) return { - rankings: sortAggregatedData(rankings), - trend: Array.from(buckets.entries()) - .map(([timestamp, data]) => ({ timestamp, ...data })) - .sort((a, b) => a.timestamp - b.timestamp) + ...result, + trend: fillTrend(result.trend, startTime, endTime, bucketSizeMs) } } -export async function getSubStatsByHost( +export function getSubStatsByHost( dimension: Exclude, label: string, startTime: number, endTime: number ): Promise { - const map = new Map() - - await db.iterate(startTime, endTime, (log) => { - if (getDimensionLabel(dimension, log) === label) { - addAggregatedLog(map, log.host, log) - } - }) - - return sortAggregatedData(map) + return breakdown({ groupBy: 'host', filters: { [dimension]: label }, startTime, endTime }) } -export async function getDevicesByHost( +export function getDevicesByHost( host: string, startTime: number, endTime: number ): Promise { - const map = new Map() - - await db.iterate(startTime, endTime, (log) => { - if (log.host === host) { - addAggregatedLog(map, log.sourceIP, log) - } - }) - - return sortAggregatedData(map) + return breakdown({ groupBy: 'sourceIP', filters: { host }, startTime, endTime }) } -export async function getProxyStatsByHost( +export function getProxyStatsByHost( dimension: DataUsageType, parentLabel: string, host: string, startTime: number, endTime: number ): Promise { - const map = new Map() - - await db.iterate(startTime, endTime, (log) => { - if (log.host === host && getDimensionLabel(dimension, log) === parentLabel) { - addAggregatedLog(map, log.outbound, log) - } + const filters = + dimension === 'host' + ? { host: parentLabel, sourceIP: host } + : { [dimension]: parentLabel, host } + return breakdown({ + groupBy: 'outbound', + filters, + startTime, + endTime }) - - return sortAggregatedData(map) } -export async function getDevicesByProxyAndHost( - proxy: string, - host: string, - startTime: number, - endTime: number -): Promise { - const map = new Map() - - await db.iterate(startTime, endTime, (log) => { - if (log.outbound === proxy && log.host === host) { - addAggregatedLog(map, log.sourceIP, log) - } - }) - - return sortAggregatedData(map) +export async function clearTrafficUsageData(): Promise { + await legacyTrafficUsageDatabase.clear() + if (!__LEGACY_BUILD__) await clearTrafficUsage() } diff --git a/src/renderer/src/utils/db.ts b/src/renderer/src/utils/db.ts deleted file mode 100644 index 739713b9..00000000 --- a/src/renderer/src/utils/db.ts +++ /dev/null @@ -1,143 +0,0 @@ -export interface DataUsageLog { - id?: number - timestamp: number - sourceIP: string - host: string - outbound: string - process: string - upload: number - download: number -} - -const DB_NAME = 'clashparty_db' -const STORE_NAME = 'data_usage_logs' -const DB_VERSION = 1 - -export class DataUsageDB { - private db: IDBDatabase | null = null - - async open(): Promise { - if (this.db) return this.db - - return new Promise((resolve, reject) => { - const request = indexedDB.open(DB_NAME, DB_VERSION) - - request.onupgradeneeded = (event) => { - const db = (event.target as IDBOpenDBRequest).result - if (!db.objectStoreNames.contains(STORE_NAME)) { - const store = db.createObjectStore(STORE_NAME, { keyPath: 'id', autoIncrement: true }) - store.createIndex('timestamp', 'timestamp', { unique: false }) - store.createIndex('sourceIP', 'sourceIP', { unique: false }) - store.createIndex('host', 'host', { unique: false }) - store.createIndex('outbound', 'outbound', { unique: false }) - store.createIndex('process', 'process', { unique: false }) - } - } - - request.onsuccess = (event) => { - this.db = (event.target as IDBOpenDBRequest).result - resolve(this.db) - } - - request.onerror = () => reject(request.error) - }) - } - - async addLogs(logs: DataUsageLog[]): Promise { - if (logs.length === 0) return - const db = await this.open() - return new Promise((resolve, reject) => { - const tx = db.transaction([STORE_NAME], 'readwrite') - const store = tx.objectStore(STORE_NAME) - logs.forEach((log) => store.add(log)) - tx.oncomplete = () => resolve() - tx.onerror = () => reject(tx.error) - }) - } - - async query(startTime: number, endTime: number): Promise { - const db = await this.open() - return new Promise((resolve, reject) => { - const tx = db.transaction([STORE_NAME], 'readonly') - const index = tx.objectStore(STORE_NAME).index('timestamp') - const request = index.openCursor(IDBKeyRange.bound(startTime, endTime)) - const results: DataUsageLog[] = [] - - request.onsuccess = (event) => { - const cursor = (event.target as IDBRequest).result - if (cursor) { - results.push(cursor.value) - cursor.continue() - } else { - resolve(results) - } - } - - request.onerror = () => reject(request.error) - }) - } - - async iterate( - startTime: number, - endTime: number, - callback: (log: DataUsageLog) => void - ): Promise { - const db = await this.open() - return new Promise((resolve, reject) => { - const tx = db.transaction([STORE_NAME], 'readonly') - const index = tx.objectStore(STORE_NAME).index('timestamp') - const request = index.openCursor(IDBKeyRange.bound(startTime, endTime)) - - request.onsuccess = (event) => { - const cursor = (event.target as IDBRequest).result - if (!cursor) return - - try { - callback(cursor.value as DataUsageLog) - cursor.continue() - } catch (error) { - tx.abort() - reject(error) - } - } - - request.onerror = () => reject(request.error) - tx.oncomplete = () => resolve() - tx.onerror = () => reject(tx.error) - tx.onabort = () => reject(tx.error) - }) - } - - async clearAll(): Promise { - const db = await this.open() - return new Promise((resolve, reject) => { - const tx = db.transaction([STORE_NAME], 'readwrite') - const request = tx.objectStore(STORE_NAME).clear() - request.onsuccess = () => resolve() - request.onerror = () => reject(request.error) - }) - } - - async cleanup(beforeTime: number): Promise { - const db = await this.open() - return new Promise((resolve, reject) => { - const tx = db.transaction([STORE_NAME], 'readwrite') - const store = tx.objectStore(STORE_NAME) - const request = store.index('timestamp').openKeyCursor(IDBKeyRange.upperBound(beforeTime)) - - request.onsuccess = (event) => { - const cursor = (event.target as IDBRequest).result - if (cursor) { - store.delete(cursor.primaryKey) - cursor.continue() - } else { - resolve() - } - } - - request.onerror = () => reject(request.error) - }) - } -} - -export const db = new DataUsageDB() diff --git a/src/renderer/src/utils/ipc.ts b/src/renderer/src/utils/ipc.ts index ef340e33..7455301f 100644 --- a/src/renderer/src/utils/ipc.ts +++ b/src/renderer/src/utils/ipc.ts @@ -1,4 +1,11 @@ import { TitleBarOverlayOptions } from 'electron' +import type { + TrafficUsageAggregate, + TrafficUsageBreakdownQuery, + TrafficUsageDimension, + TrafficUsageImportBatch, + TrafficUsageOverview +} from '../../../shared/trafficUsage' function checkIpcError(response: unknown): T { if (response && typeof response === 'object' && 'invokeError' in response) { @@ -36,6 +43,17 @@ interface IpcApi { patchMihomoConfig: (patch: Partial) => Promise mihomoSmartGroupWeights: (groupName: string) => Promise> mihomoSmartFlushCache: (configName?: string) => Promise + queryTrafficUsageOverview: ( + type: TrafficUsageDimension, + startTime: number, + endTime: number, + bucketSizeMs: number + ) => Promise + queryTrafficUsageBreakdown: ( + query: TrafficUsageBreakdownQuery + ) => Promise + importTrafficUsage: (batch: TrafficUsageImportBatch) => Promise + clearTrafficUsage: () => Promise getSmartOverrideContent: () => Promise // AutoRun checkAutoRun: () => Promise @@ -208,6 +226,10 @@ export const { patchMihomoConfig, mihomoSmartGroupWeights, mihomoSmartFlushCache, + queryTrafficUsageOverview, + queryTrafficUsageBreakdown, + importTrafficUsage, + clearTrafficUsage, getSmartOverrideContent, // AutoRun checkAutoRun, diff --git a/src/renderer/src/utils/legacy-traffic-db.ts b/src/renderer/src/utils/legacy-traffic-db.ts new file mode 100644 index 00000000..82f53c79 --- /dev/null +++ b/src/renderer/src/utils/legacy-traffic-db.ts @@ -0,0 +1,477 @@ +import { + TRAFFIC_USAGE_AGGREGATION_LIMIT, + TRAFFIC_USAGE_MIGRATION_CHUNK_SIZE, + TRAFFIC_USAGE_RESOLUTIONS, + TRAFFIC_USAGE_RESULT_LIMIT, + TRAFFIC_USAGE_RETENTION, + trafficUsageRecordKey, + trafficUsageResolution, + type TrafficUsageAggregate, + type TrafficUsageBreakdownQuery, + type TrafficUsageDimension, + type TrafficUsageImportBatch, + type TrafficUsageOverview, + type TrafficUsageRecord, + type TrafficUsageSample +} from '../../../shared/trafficUsage' + +const DB_NAME = 'clashparty_db' +const DB_VERSION = 2 +const LEGACY_STORE = 'data_usage_logs' +const USAGE_STORE = 'traffic_usage_rollups' +const META_STORE = 'traffic_usage_meta' +const RESOLUTION_BUCKET_INDEX = 'resolution_bucket' +const MIGRATION_KEY = 'legacy_migration' +const BACKEND_MIGRATION_KEY = 'backend_migration' + +interface LegacyDataUsageLog { + id: number + timestamp: number + sourceIP: string + host: string + outbound: string + process: string + upload: number + download: number +} + +interface MigrationState { + key: typeof MIGRATION_KEY + lastId: number + complete: boolean +} + +interface BackendMigrationState { + key: typeof BACKEND_MIGRATION_KEY + migrationId: string + lastKey?: IDBValidKey + sequence: number + complete: boolean +} + +function requestResult(request: IDBRequest): Promise { + return new Promise((resolve, reject) => { + request.onsuccess = () => resolve(request.result) + request.onerror = () => reject(request.error) + }) +} + +function transactionComplete(transaction: IDBTransaction): Promise { + return new Promise((resolve, reject) => { + transaction.oncomplete = () => resolve() + transaction.onerror = () => reject(transaction.error) + transaction.onabort = () => reject(transaction.error) + }) +} + +function mergeRecord(map: Map, record: TrafficUsageRecord): void { + const key = trafficUsageRecordKey(record) + const current = map.get(key) + if (current) { + current.upload += record.upload + current.download += record.download + current.count += record.count + } else { + map.set(key, record) + } +} + +function legacyRecords(log: LegacyDataUsageLog): TrafficUsageRecord[] { + return TRAFFIC_USAGE_RESOLUTIONS.map((resolution) => ({ + resolution, + bucket: Math.floor(log.timestamp / resolution) * resolution, + sourceIP: log.sourceIP, + host: log.host, + outbound: log.outbound, + process: log.process, + upload: log.upload, + download: log.download, + count: 1 + })) +} + +function putAggregatedRecord(store: IDBObjectStore, record: TrafficUsageRecord): void { + const request = store.get([ + record.resolution, + record.bucket, + record.sourceIP, + record.host, + record.outbound, + record.process + ]) + request.onsuccess = () => { + const current = request.result as TrafficUsageRecord | undefined + store.put( + current + ? { + ...current, + upload: current.upload + record.upload, + download: current.download + record.download, + count: current.count + record.count + } + : record + ) + } +} + +function putAggregatedRecords(store: IDBObjectStore, records: TrafficUsageRecord[]): void { + for (const record of records) putAggregatedRecord(store, record) +} + +function putAggregatedSamples(store: IDBObjectStore, samples: TrafficUsageSample[]): void { + for (const sample of samples) { + for (const resolution of TRAFFIC_USAGE_RESOLUTIONS) { + putAggregatedRecord(store, { + ...sample, + resolution, + bucket: Math.floor(sample.bucket / resolution) * resolution + }) + } + } +} + +function dimensionValue(record: TrafficUsageRecord, dimension: TrafficUsageDimension): string { + return record[dimension] +} + +class LegacyTrafficUsageDatabase { + private database: IDBDatabase | null = null + private migrationPromise: Promise | null = null + private backendMigrationPromise: Promise | null = null + private lastCleanup = 0 + + async upsert(samples: TrafficUsageSample[]): Promise { + if (samples.length === 0) return + const database = await this.open() + const transaction = database.transaction([USAGE_STORE, META_STORE], 'readwrite') + putAggregatedSamples(transaction.objectStore(USAGE_STORE), samples) + transaction.objectStore(META_STORE).delete(BACKEND_MIGRATION_KEY) + await transactionComplete(transaction) + await this.cleanup() + } + + async overview( + type: TrafficUsageDimension, + startTime: number, + endTime: number, + bucketSizeMs: number + ): Promise { + const rankings = new Map() + const trend = new Map() + const totals = { upload: 0, download: 0, total: 0, count: 0 } + + await this.iterate(startTime, endTime, (record) => { + totals.upload += record.upload + totals.download += record.download + totals.total += record.upload + record.download + totals.count += record.count + this.addAggregate(rankings, dimensionValue(record, type), record) + + const timestamp = Math.floor(record.bucket / bucketSizeMs) * bucketSizeMs + const bucket = trend.get(timestamp) + if (bucket) { + bucket.upload += record.upload + bucket.download += record.download + } else { + trend.set(timestamp, { upload: record.upload, download: record.download }) + } + }) + + return { + rankings: this.sorted(rankings), + trend: Array.from(trend, ([timestamp, data]) => ({ timestamp, ...data })).sort( + (a, b) => a.timestamp - b.timestamp + ), + totals + } + } + + async breakdown(query: TrafficUsageBreakdownQuery): Promise { + const aggregates = new Map() + await this.iterate(query.startTime, query.endTime, (record) => { + for (const [dimension, value] of Object.entries(query.filters) as [ + TrafficUsageDimension, + string + ][]) { + if (dimensionValue(record, dimension) !== value) return + } + this.addAggregate(aggregates, dimensionValue(record, query.groupBy), record) + }) + return this.sorted(aggregates) + } + + async clear(): Promise { + await this.migrationPromise + await this.backendMigrationPromise + const database = await this.open() + const stores = [USAGE_STORE, META_STORE] + if (database.objectStoreNames.contains(LEGACY_STORE)) stores.push(LEGACY_STORE) + const transaction = database.transaction(stores, 'readwrite') + transaction.objectStore(USAGE_STORE).clear() + if (stores.includes(LEGACY_STORE)) transaction.objectStore(LEGACY_STORE).clear() + transaction.objectStore(META_STORE).put({ + key: MIGRATION_KEY, + lastId: 0, + complete: true + } satisfies MigrationState) + transaction.objectStore(META_STORE).delete(BACKEND_MIGRATION_KEY) + await transactionComplete(transaction) + } + + migrateLegacyLogs(): Promise { + if (!this.migrationPromise) { + this.migrationPromise = this.runMigration().catch((error) => { + this.migrationPromise = null + throw error + }) + } + return this.migrationPromise + } + + migrateToBackend(importBatch: (batch: TrafficUsageImportBatch) => Promise): Promise { + if (!this.backendMigrationPromise) { + this.backendMigrationPromise = this.runBackendMigration(importBatch).catch((error) => { + this.backendMigrationPromise = null + throw error + }) + } + return this.backendMigrationPromise + } + + private async open(): Promise { + if (this.database) return this.database + const request = indexedDB.open(DB_NAME, DB_VERSION) + request.onupgradeneeded = () => { + const database = request.result + if (!database.objectStoreNames.contains(USAGE_STORE)) { + const store = database.createObjectStore(USAGE_STORE, { + keyPath: ['resolution', 'bucket', 'sourceIP', 'host', 'outbound', 'process'] + }) + store.createIndex(RESOLUTION_BUCKET_INDEX, ['resolution', 'bucket']) + } + if (!database.objectStoreNames.contains(META_STORE)) { + database.createObjectStore(META_STORE, { keyPath: 'key' }) + } + } + const database = await requestResult(request) + database.onversionchange = () => database.close() + this.database = database + return database + } + + private async iterate( + startTime: number, + endTime: number, + callback: (record: TrafficUsageRecord) => void + ): Promise { + const resolution = trafficUsageResolution(startTime, endTime) + const database = await this.open() + const transaction = database.transaction(USAGE_STORE, 'readonly') + const index = transaction.objectStore(USAGE_STORE).index(RESOLUTION_BUCKET_INDEX) + const range = IDBKeyRange.bound( + [resolution, Math.floor(startTime / resolution) * resolution], + [resolution, Math.floor(endTime / resolution) * resolution] + ) + const request = index.openCursor(range) + request.onsuccess = () => { + const cursor = request.result + if (!cursor) return + callback(cursor.value as TrafficUsageRecord) + cursor.continue() + } + request.onerror = () => transaction.abort() + await transactionComplete(transaction) + } + + private addAggregate( + aggregates: Map, + label: string, + record: TrafficUsageRecord + ): void { + const current = aggregates.get(label) + if (current) { + current.upload += record.upload + current.download += record.download + current.total += record.upload + record.download + current.count += record.count + return + } + if (aggregates.size >= TRAFFIC_USAGE_AGGREGATION_LIMIT) return + aggregates.set(label, { + label, + upload: record.upload, + download: record.download, + total: record.upload + record.download, + count: record.count + }) + } + + private sorted(aggregates: Map): TrafficUsageAggregate[] { + return Array.from(aggregates.values()) + .sort((a, b) => b.total - a.total) + .slice(0, TRAFFIC_USAGE_RESULT_LIMIT) + } + + private async cleanup(): Promise { + const now = Date.now() + if (now - this.lastCleanup < 24 * 60 * 60 * 1000) return + const database = await this.open() + const transaction = database.transaction(USAGE_STORE, 'readwrite') + const index = transaction.objectStore(USAGE_STORE).index(RESOLUTION_BUCKET_INDEX) + for (const resolution of TRAFFIC_USAGE_RESOLUTIONS) { + const range = IDBKeyRange.bound( + [resolution, 0], + [resolution, now - TRAFFIC_USAGE_RETENTION[resolution]], + false, + true + ) + const request = index.openKeyCursor(range) + request.onsuccess = () => { + const cursor = request.result + if (!cursor) return + transaction.objectStore(USAGE_STORE).delete(cursor.primaryKey) + cursor.continue() + } + } + await transactionComplete(transaction) + this.lastCleanup = now + } + + private async runMigration(): Promise { + const database = await this.open() + if (!database.objectStoreNames.contains(LEGACY_STORE)) return + + while (!(await this.migrateChunk())) { + await new Promise((resolve) => setTimeout(resolve, 0)) + } + } + + private async migrateChunk(): Promise { + const database = await this.open() + const transaction = database.transaction([LEGACY_STORE, USAGE_STORE, META_STORE], 'readwrite') + const legacyStore = transaction.objectStore(LEGACY_STORE) + const usageStore = transaction.objectStore(USAGE_STORE) + const metaStore = transaction.objectStore(META_STORE) + const state = (await requestResult(metaStore.get(MIGRATION_KEY))) as MigrationState | undefined + if (state?.complete) { + await transactionComplete(transaction) + return true + } + + const aggregates = new Map() + let count = 0 + let lastId = state?.lastId ?? 0 + let complete = false + const range = lastId > 0 ? IDBKeyRange.lowerBound(lastId, true) : undefined + const request = legacyStore.openCursor(range) + request.onsuccess = () => { + const cursor = request.result + if (!cursor) { + complete = true + } else { + const log = cursor.value as LegacyDataUsageLog + lastId = log.id + for (const record of legacyRecords(log)) mergeRecord(aggregates, record) + count += 1 + } + + if (cursor && count < TRAFFIC_USAGE_MIGRATION_CHUNK_SIZE) { + cursor.continue() + return + } + + putAggregatedRecords(usageStore, Array.from(aggregates.values())) + if (aggregates.size > 0) metaStore.delete(BACKEND_MIGRATION_KEY) + metaStore.put({ key: MIGRATION_KEY, lastId, complete } satisfies MigrationState) + if (complete) legacyStore.clear() + } + request.onerror = () => transaction.abort() + await transactionComplete(transaction) + return complete + } + + private async runBackendMigration( + importBatch: (batch: TrafficUsageImportBatch) => Promise + ): Promise { + await this.runMigration() + while (!(await this.migrateBackendChunk(importBatch))) { + await new Promise((resolve) => setTimeout(resolve, 0)) + } + } + + private async migrateBackendChunk( + importBatch: (batch: TrafficUsageImportBatch) => Promise + ): Promise { + const state = await this.backendMigrationState() + if (state.complete) return true + + const { records, lastKey, complete } = await this.backendMigrationChunk(state.lastKey) + if (records.length > 0) { + await importBatch({ id: `${state.migrationId}-${state.sequence}`, records }) + } + + const database = await this.open() + const transaction = database.transaction([USAGE_STORE, META_STORE], 'readwrite') + if (complete) transaction.objectStore(USAGE_STORE).clear() + transaction.objectStore(META_STORE).put({ + key: BACKEND_MIGRATION_KEY, + migrationId: state.migrationId, + lastKey, + sequence: state.sequence + 1, + complete + } satisfies BackendMigrationState) + await transactionComplete(transaction) + return complete + } + + private async backendMigrationState(): Promise { + const database = await this.open() + const transaction = database.transaction(META_STORE, 'readonly') + const state = (await requestResult( + transaction.objectStore(META_STORE).get(BACKEND_MIGRATION_KEY) + )) as BackendMigrationState | undefined + await transactionComplete(transaction) + if (state) return state + + const initial: BackendMigrationState = { + key: BACKEND_MIGRATION_KEY, + migrationId: crypto.randomUUID(), + sequence: 0, + complete: false + } + const createTransaction = database.transaction(META_STORE, 'readwrite') + createTransaction.objectStore(META_STORE).put(initial) + await transactionComplete(createTransaction) + return initial + } + + private async backendMigrationChunk( + lastKey?: IDBValidKey + ): Promise<{ records: TrafficUsageRecord[]; lastKey?: IDBValidKey; complete: boolean }> { + const database = await this.open() + const transaction = database.transaction(USAGE_STORE, 'readonly') + const store = transaction.objectStore(USAGE_STORE) + const records: TrafficUsageRecord[] = [] + let nextLastKey = lastKey + let complete = false + const request = store.openCursor( + lastKey === undefined ? undefined : IDBKeyRange.lowerBound(lastKey, true) + ) + request.onsuccess = () => { + const cursor = request.result + if (!cursor) { + complete = true + return + } + if (records.length >= TRAFFIC_USAGE_MIGRATION_CHUNK_SIZE) return + records.push(cursor.value as TrafficUsageRecord) + nextLastKey = cursor.primaryKey + cursor.continue() + } + request.onerror = () => transaction.abort() + await transactionComplete(transaction) + return { records, lastKey: nextLastKey, complete } + } +} + +export const legacyTrafficUsageDatabase = new LegacyTrafficUsageDatabase() diff --git a/src/shared/build.d.ts b/src/shared/build.d.ts new file mode 100644 index 00000000..5f653a52 --- /dev/null +++ b/src/shared/build.d.ts @@ -0,0 +1 @@ +declare const __LEGACY_BUILD__: boolean diff --git a/src/shared/trafficUsage.ts b/src/shared/trafficUsage.ts new file mode 100644 index 00000000..97cddce5 --- /dev/null +++ b/src/shared/trafficUsage.ts @@ -0,0 +1,229 @@ +export type TrafficUsageDimension = 'sourceIP' | 'host' | 'outbound' | 'process' + +export interface TrafficUsageRecord { + resolution: number + bucket: number + sourceIP: string + host: string + outbound: string + process: string + upload: number + download: number + count: number +} + +export type TrafficUsageSample = Omit + +export interface TrafficUsageWriteBatch { + id: string + samples: TrafficUsageSample[] +} + +export interface TrafficUsageImportBatch { + id: string + records: TrafficUsageRecord[] +} + +export interface TrafficUsageAggregate { + label: string + upload: number + download: number + total: number + count: number +} + +export interface TrafficUsageTrendPoint { + timestamp: number + upload: number + download: number +} + +export interface TrafficUsageOverview { + rankings: TrafficUsageAggregate[] + trend: TrafficUsageTrendPoint[] + totals: Omit +} + +export interface TrafficUsageBreakdownQuery { + groupBy: TrafficUsageDimension + filters: Partial> + startTime: number + endTime: number +} + +export const TRAFFIC_USAGE_RESOLUTIONS = [ + 5 * 60 * 1000, + 60 * 60 * 1000, + 24 * 60 * 60 * 1000 +] as const + +export const TRAFFIC_USAGE_RETENTION: Readonly> = { + [TRAFFIC_USAGE_RESOLUTIONS[0]]: 25 * 60 * 60 * 1000, + [TRAFFIC_USAGE_RESOLUTIONS[1]]: 8 * 24 * 60 * 60 * 1000, + [TRAFFIC_USAGE_RESOLUTIONS[2]]: 31 * 24 * 60 * 60 * 1000 +} + +export const TRAFFIC_USAGE_RESULT_LIMIT = 500 +export const TRAFFIC_USAGE_AGGREGATION_LIMIT = 10_000 +export const TRAFFIC_USAGE_MIGRATION_CHUNK_SIZE = 500 +export const TRAFFIC_USAGE_FLUSH_THRESHOLD = Math.ceil(5_000 / TRAFFIC_USAGE_RESOLUTIONS.length) +export const TRAFFIC_USAGE_PENDING_LIMIT = Math.ceil(20_000 / TRAFFIC_USAGE_RESOLUTIONS.length) + +interface TrafficSnapshot { + upload: number + download: number + generation: number +} + +const TRAFFIC_USAGE_KEY_SEPARATOR = '\u001f' + +export function trafficUsageRecordKey(record: TrafficUsageRecord): string { + return `${record.resolution}${TRAFFIC_USAGE_KEY_SEPARATOR}${record.bucket}${TRAFFIC_USAGE_KEY_SEPARATOR}${record.sourceIP}${TRAFFIC_USAGE_KEY_SEPARATOR}${record.host}${TRAFFIC_USAGE_KEY_SEPARATOR}${record.outbound}${TRAFFIC_USAGE_KEY_SEPARATOR}${record.process}` +} + +function trafficUsageSampleKey(sample: TrafficUsageSample): string { + return `${sample.bucket}${TRAFFIC_USAGE_KEY_SEPARATOR}${sample.sourceIP}${TRAFFIC_USAGE_KEY_SEPARATOR}${sample.host}${TRAFFIC_USAGE_KEY_SEPARATOR}${sample.outbound}${TRAFFIC_USAGE_KEY_SEPARATOR}${sample.process}` +} + +export function trafficUsageResolution(startTime: number, endTime: number): number { + const range = endTime - startTime + if (range <= TRAFFIC_USAGE_RETENTION[TRAFFIC_USAGE_RESOLUTIONS[0]]) { + return TRAFFIC_USAGE_RESOLUTIONS[0] + } + if (range <= TRAFFIC_USAGE_RETENTION[TRAFFIC_USAGE_RESOLUTIONS[1]]) { + return TRAFFIC_USAGE_RESOLUTIONS[1] + } + return TRAFFIC_USAGE_RESOLUTIONS[2] +} + +export class TrafficUsageAccumulator { + private readonly lastConnections = new Map() + private pending = new Map() + private lastUploadTotal = 0 + private lastDownloadTotal = 0 + private generation = 0 + private enabledAt = 0 + private enabled = false + private droppedRecords = 0 + + setEnabled(enabled: boolean, now = Date.now()): void { + if (this.enabled === enabled) return + this.enabled = enabled + this.reset() + if (enabled) this.enabledAt = now + } + + addSnapshot(info: IMihomoConnectionsInfo, now = Date.now()): boolean { + if (!this.enabled) return false + + const uploadTotal = info.uploadTotal || 0 + const downloadTotal = info.downloadTotal || 0 + if (uploadTotal < this.lastUploadTotal || downloadTotal < this.lastDownloadTotal) { + this.lastConnections.clear() + this.pending.clear() + } + this.lastUploadTotal = uploadTotal + this.lastDownloadTotal = downloadTotal + + const connections = info.connections ?? [] + if (connections.length === 0) { + this.lastConnections.clear() + return false + } + + const generation = ++this.generation + for (const connection of connections) { + const currentUpload = connection.upload || 0 + const currentDownload = connection.download || 0 + const previous = this.lastConnections.get(connection.id) + let upload: number + let download: number + if (previous) { + upload = Math.max(0, currentUpload - previous.upload) + download = Math.max(0, currentDownload - previous.download) + previous.upload = currentUpload + previous.download = currentDownload + previous.generation = generation + } else { + const startedAt = Date.parse(connection.start) + const includeInitial = Number.isFinite(startedAt) && startedAt >= this.enabledAt + upload = includeInitial ? currentUpload : 0 + download = includeInitial ? currentDownload : 0 + this.lastConnections.set(connection.id, { + upload: currentUpload, + download: currentDownload, + generation + }) + } + if (upload === 0 && download === 0) continue + + const sourceIP = connection.metadata.sourceIP || 'Inner' + const host = connection.metadata.host || connection.metadata.destinationIP || 'Unknown' + const outbound = connection.chains?.[0] || 'DIRECT' + const process = connection.metadata.process || 'Unknown' + this.addSample({ + bucket: Math.floor(now / TRAFFIC_USAGE_RESOLUTIONS[0]) * TRAFFIC_USAGE_RESOLUTIONS[0], + sourceIP, + host, + outbound, + process, + upload, + download, + count: 1 + }) + } + + for (const [id, snapshot] of this.lastConnections) { + if (snapshot.generation !== generation) this.lastConnections.delete(id) + } + return this.pending.size >= TRAFFIC_USAGE_FLUSH_THRESHOLD + } + + takePending(): TrafficUsageSample[] { + if (this.pending.size === 0) return [] + const records = Array.from(this.pending.values()) + this.pending = new Map() + return records + } + + merge(samples: TrafficUsageSample[]): void { + for (const sample of samples) this.addSample(sample) + } + + reset(): void { + this.lastConnections.clear() + this.pending.clear() + this.lastUploadTotal = 0 + this.lastDownloadTotal = 0 + this.generation = 0 + this.enabledAt = 0 + } + + get pendingSize(): number { + return this.pending.size + } + + get activeConnectionCount(): number { + return this.lastConnections.size + } + + get droppedCount(): number { + return this.droppedRecords + } + + private addSample(sample: TrafficUsageSample): void { + const key = trafficUsageSampleKey(sample) + const current = this.pending.get(key) + if (current) { + current.upload += sample.upload + current.download += sample.download + current.count += sample.count + return + } + if (this.pending.size >= TRAFFIC_USAGE_PENDING_LIMIT) { + this.droppedRecords += 1 + return + } + this.pending.set(key, { ...sample }) + } +}