refactor: rewrite traffic usage logging

2026-09-20 08:03:39 +08:00 · 2026-09-03 01:54:28 +08:00
parent 89c2bb0ec0
commit b5293b3f65
21 changed files with 1519 additions and 431 deletions
--- 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: {
--- 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<IAppConfig> {
      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<IAppConfig>): Promise<void>
    setGlobalMaxLogFileSizeMB(nextConfig.maxLogFileSize)
    setCoreLogDisabled(nextConfig.disableCoreLog === true)
    setAppLogDisabled(nextConfig.disableAppLog === true)
+    notifyAppConfigListeners()
  })
 }
--- 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,
--- 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<void> => {
    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
    }
--- 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<number | null> {
  // 仅 PS 3.0+ 写入 \3\ 键（\1\ 键恒为 2.0，不可用）。
@@ -87,6 +88,7 @@ async function ensureSupportedWindowsPowerShell(): Promise<boolean> {
 }

 configureAppPaths()
+subscribeAppConfig((config) => setTrafficUsageEnabled(config.enableTrafficLogger === true))

 const mainLogger = createLogger('Main')

--- 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<unknown>[] = [stopCoreForExit()]
+      const cleanupTasks: Promise<unknown>[] = [stopCoreForExit(), closeTrafficUsage()]
      if (process.platform === 'darwin') {
        cleanupTasks.push(
          triggerSysProxy(false, { helperTimeout: 750, force: true }).then(() => {
--- a/src/main/traffic/database-worker.ts
+++ 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<TrafficUsageDimension, string> = {
+  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<string, StatementSync>()
+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<string, unknown>): 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<Record<TrafficUsageDimension, string>>,
+  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<string, unknown>[]).map(mapAggregate)
+}
+
+function queryOverview(
+  payload: Extract<TrafficDatabaseRequest, { action: 'overview' }>['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<string, unknown>
+  const trend = (
+    trendQuery.all(
+      payload.bucketSizeMs,
+      payload.bucketSizeMs,
+      resolution,
+      start,
+      end,
+      payload.bucketSizeMs
+    ) as Record<string, unknown>[]
+  ).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()
+})
--- a/src/main/traffic/database.ts
+++ 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<Request, 'id'>
+    : never
+  : never
+
+class TrafficDatabaseClient {
+  private worker: Worker | null = null
+  private nextId = 1
+  private readonly pending = new Map<number, PendingRequest>()
+  private closing = false
+
+  write(batch: TrafficUsageWriteBatch): Promise<void> {
+    return this.request({ action: 'write', payload: batch })
+  }
+
+  import(batch: TrafficUsageImportBatch): Promise<void> {
+    return this.request({ action: 'import', payload: batch })
+  }
+
+  overview(
+    type: TrafficUsageDimension,
+    startTime: number,
+    endTime: number,
+    bucketSizeMs: number
+  ): Promise<TrafficUsageOverview> {
+    return this.request(
+      {
+        action: 'overview',
+        payload: { type, startTime, endTime, bucketSizeMs }
+      },
+      30_000
+    )
+  }
+
+  breakdown(query: TrafficUsageBreakdownQuery): Promise<TrafficUsageAggregate[]> {
+    return this.request({ action: 'breakdown', payload: query }, 30_000)
+  }
+
+  clear(): Promise<void> {
+    return this.request({ action: 'clear' }, 30_000)
+  }
+
+  async close(): Promise<void> {
+    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<T>(request: TrafficDatabaseRequestWithoutId, timeoutMs?: number): Promise<T> {
+    const id = this.nextId++
+    return new Promise<T>((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<void> =>
+  database.write(batch)
+export const importTrafficUsage = (batch: TrafficUsageImportBatch): Promise<void> => {
+  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<TrafficUsageOverview> => database.overview(type, startTime, endTime, bucketSizeMs)
+export const queryTrafficUsageBreakdown = (
+  query: TrafficUsageBreakdownQuery
+): Promise<TrafficUsageAggregate[]> => database.breakdown(query)
+export const clearTrafficUsage = (): Promise<void> => database.clear()
+export const closeTrafficUsageDatabase = (): Promise<void> => database.close()
--- a/src/main/traffic/databaseMessages.ts
+++ 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
+}
--- a/src/main/traffic/recorder.ts
+++ 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<void> | 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<void> {
+  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<void> {
+  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()
+}
--- 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')
 }
--- 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<string, AsyncFn> = {
  mihomoCloseAllConnections,
  mihomoRules,
  mihomoRulesDisable,
+  queryTrafficUsageOverview,
+  queryTrafficUsageBreakdown,
+  importTrafficUsage,
+  clearTrafficUsage,
  mihomoProxies,
  mihomoGroups,
  mihomoProxyProviders,
--- 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',
--- 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<string, TrafficSnapshot>())
-  const logBufferRef = useRef<DataUsageLog[]>([])
-  const flushTimeoutRef = useRef<ReturnType<typeof setTimeout> | 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<typeof setTimeout> | null = null
+    let inFlight: Promise<void> | 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<void> => {
+      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<void> => {
-      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<string>()
-
-      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])
 }
--- 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<typeof setTimeout> | null = null
    let cancelled = false
+    let refreshing = false
+    let resetSelection = true

-    const refresh = async (resetSelection: boolean): Promise<void> => {
-      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<void> => {
+      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])

--- 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<string, AggregatedData>,
-  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<string, AggregatedData>): 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<TrafficUsageOverview> {
+  if (__LEGACY_BUILD__) {
+    return legacyTrafficUsageDatabase.overview(type, startTime, endTime, bucketSizeMs)
  }
+  return queryTrafficUsageOverview(type, startTime, endTime, bucketSizeMs)
+}
+
+async function breakdown(query: TrafficUsageBreakdownQuery): Promise<AggregatedData[]> {
+  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<string, AggregatedData>()
-  const buckets = new Map<number, { upload: number; download: number }>()
-
-  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<TrafficUsageOverview> {
+  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<DataUsageType, 'host'>,
  label: string,
  startTime: number,
  endTime: number
 ): Promise<AggregatedData[]> {
-  const map = new Map<string, AggregatedData>()
-
-  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<AggregatedData[]> {
-  const map = new Map<string, AggregatedData>()
-
-  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<AggregatedData[]> {
-  const map = new Map<string, AggregatedData>()
-
-  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<AggregatedData[]> {
-  const map = new Map<string, AggregatedData>()
-
-  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<void> {
+  await legacyTrafficUsageDatabase.clear()
+  if (!__LEGACY_BUILD__) await clearTrafficUsage()
 }
--- a/src/renderer/src/utils/db.ts
+++ b/src/renderer/src/utils/db.ts
@@ -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<IDBDatabase> {
-    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<void> {
-    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<DataUsageLog[]> {
-    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<IDBCursorWithValue>).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<void> {
-    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<IDBCursorWithValue>).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<void> {
-    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<void> {
-    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<IDBCursor>).result
-        if (cursor) {
-          store.delete(cursor.primaryKey)
-          cursor.continue()
-        } else {
-          resolve()
-        }
-      }
-
-      request.onerror = () => reject(request.error)
-    })
-  }
-}
-
-export const db = new DataUsageDB()
--- 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<T>(response: unknown): T {
  if (response && typeof response === 'object' && 'invokeError' in response) {
@@ -36,6 +43,17 @@ interface IpcApi {
  patchMihomoConfig: (patch: Partial<IMihomoConfig>) => Promise<void>
  mihomoSmartGroupWeights: (groupName: string) => Promise<Record<string, number>>
  mihomoSmartFlushCache: (configName?: string) => Promise<void>
+  queryTrafficUsageOverview: (
+    type: TrafficUsageDimension,
+    startTime: number,
+    endTime: number,
+    bucketSizeMs: number
+  ) => Promise<TrafficUsageOverview>
+  queryTrafficUsageBreakdown: (
+    query: TrafficUsageBreakdownQuery
+  ) => Promise<TrafficUsageAggregate[]>
+  importTrafficUsage: (batch: TrafficUsageImportBatch) => Promise<void>
+  clearTrafficUsage: () => Promise<void>
  getSmartOverrideContent: () => Promise<string | null>
  // AutoRun
  checkAutoRun: () => Promise<boolean>
@@ -208,6 +226,10 @@ export const {
  patchMihomoConfig,
  mihomoSmartGroupWeights,
  mihomoSmartFlushCache,
+  queryTrafficUsageOverview,
+  queryTrafficUsageBreakdown,
+  importTrafficUsage,
+  clearTrafficUsage,
  getSmartOverrideContent,
  // AutoRun
  checkAutoRun,
--- a/src/renderer/src/utils/legacy-traffic-db.ts
+++ 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<T>(request: IDBRequest<T>): Promise<T> {
+  return new Promise((resolve, reject) => {
+    request.onsuccess = () => resolve(request.result)
+    request.onerror = () => reject(request.error)
+  })
+}
+
+function transactionComplete(transaction: IDBTransaction): Promise<void> {
+  return new Promise((resolve, reject) => {
+    transaction.oncomplete = () => resolve()
+    transaction.onerror = () => reject(transaction.error)
+    transaction.onabort = () => reject(transaction.error)
+  })
+}
+
+function mergeRecord(map: Map<string, TrafficUsageRecord>, 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<void> | null = null
+  private backendMigrationPromise: Promise<void> | null = null
+  private lastCleanup = 0
+
+  async upsert(samples: TrafficUsageSample[]): Promise<void> {
+    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<TrafficUsageOverview> {
+    const rankings = new Map<string, TrafficUsageAggregate>()
+    const trend = new Map<number, { upload: number; download: number }>()
+    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<TrafficUsageAggregate[]> {
+    const aggregates = new Map<string, TrafficUsageAggregate>()
+    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<void> {
+    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<void> {
+    if (!this.migrationPromise) {
+      this.migrationPromise = this.runMigration().catch((error) => {
+        this.migrationPromise = null
+        throw error
+      })
+    }
+    return this.migrationPromise
+  }
+
+  migrateToBackend(importBatch: (batch: TrafficUsageImportBatch) => Promise<void>): Promise<void> {
+    if (!this.backendMigrationPromise) {
+      this.backendMigrationPromise = this.runBackendMigration(importBatch).catch((error) => {
+        this.backendMigrationPromise = null
+        throw error
+      })
+    }
+    return this.backendMigrationPromise
+  }
+
+  private async open(): Promise<IDBDatabase> {
+    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<void> {
+    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<string, TrafficUsageAggregate>,
+    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<string, TrafficUsageAggregate>): TrafficUsageAggregate[] {
+    return Array.from(aggregates.values())
+      .sort((a, b) => b.total - a.total)
+      .slice(0, TRAFFIC_USAGE_RESULT_LIMIT)
+  }
+
+  private async cleanup(): Promise<void> {
+    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<void> {
+    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<boolean> {
+    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<string, TrafficUsageRecord>()
+    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<void>
+  ): Promise<void> {
+    await this.runMigration()
+    while (!(await this.migrateBackendChunk(importBatch))) {
+      await new Promise((resolve) => setTimeout(resolve, 0))
+    }
+  }
+
+  private async migrateBackendChunk(
+    importBatch: (batch: TrafficUsageImportBatch) => Promise<void>
+  ): Promise<boolean> {
+    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<BackendMigrationState> {
+    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()
--- a/src/shared/build.d.ts
+++ b/src/shared/build.d.ts
@@ -0,0 +1 @@
+declare const __LEGACY_BUILD__: boolean
--- a/src/shared/trafficUsage.ts
+++ 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<TrafficUsageRecord, 'resolution'>
+
+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<TrafficUsageAggregate, 'label'>
+}
+
+export interface TrafficUsageBreakdownQuery {
+  groupBy: TrafficUsageDimension
+  filters: Partial<Record<TrafficUsageDimension, string>>
+  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<Record<number, number>> = {
+  [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<string, TrafficSnapshot>()
+  private pending = new Map<string, TrafficUsageSample>()
+  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 })
+  }
+}