perf: CPU optimization for log ingestion pipeline

CHANGELOG.md

@@ -0,0 +1,11 @@
+# Changelog
+
+## 2026-04-16 — CPU Optimization for Log Ingestion
+
+**SSE Batching** — SSE stream now buffers logs and flushes in batches (2s / 50 logs) instead of processing one-by-one. Profile state updates throttled to 10s intervals.
+
+**Hash & Dedup Cache** — Event hashing uses faster field concatenation. New in-memory dedup cache (10K entries/profile) skips DB queries for known hashes.
+
+**Batch Device Upserts** — Device updates consolidated into a single `INSERT ... ON CONFLICT UPDATE` per batch instead of individual queries per device.
+
+**Poller Interval** — Default poller interval increased from 30s to 300s. SSE is the primary source; poller is now a fallback safety-net. Override with `POLL_INTERVAL_SECONDS`.

src/lib/ingestion/dedup-cache.ts

@@ -0,0 +1,77 @@
+import { createLogger } from "@/lib/logger";
+
+const log = createLogger("dedup-cache");
+
+const DEFAULT_MAX_ENTRIES = 10_000;
+
+export class DedupCache {
+  private profileCaches = new Map<string, Set<string>>();
+  private maxEntries: number;
+
+  constructor(maxEntries = DEFAULT_MAX_ENTRIES) {
+    this.maxEntries = maxEntries;
+  }
+
+  /** Check which hashes are NOT in cache (i.e., need DB lookup) */
+  filterMisses(profileId: string, hashes: string[]): string[] {
+    let cache = this.profileCaches.get(profileId);
+    if (!cache) {
+      cache = new Set();
+      this.profileCaches.set(profileId, cache);
+    }
+
+    const misses: string[] = [];
+    for (const hash of hashes) {
+      if (!cache.has(hash)) {
+        misses.push(hash);
+      }
+    }
+    return misses;
+  }
+
+  /** Add hashes to the cache for a profile */
+  addHashes(profileId: string, hashes: string[]) {
+    let cache = this.profileCaches.get(profileId);
+    if (!cache) {
+      cache = new Set();
+      this.profileCaches.set(profileId, cache);
+    }
+
+    for (const hash of hashes) {
+      cache.add(hash);
+    }
+
+    // Evict oldest entries if over capacity
+    if (cache.size > this.maxEntries) {
+      const excess = cache.size - this.maxEntries;
+      const iterator = cache.values();
+      for (let i = 0; i < excess; i++) {
+        const entry = iterator.next();
+        if (!entry.done) {
+          cache.delete(entry.value);
+        }
+      }
+    }
+  }
+
+  /** Remove a profile's cache when profile is stopped */
+  evictProfile(profileId: string) {
+    this.profileCaches.delete(profileId);
+  }
+
+  getStats() {
+    return {
+      profiles: this.profileCaches.size,
+      totalEntries: [...this.profileCaches.values()].reduce((sum, s) => sum + s.size, 0),
+    };
+  }
+}
+
+let _cache: DedupCache | null = null;
+
+export function getDedupCache(): DedupCache {
+  if (!_cache) {
+    _cache = new DedupCache();
+  }
+  return _cache;
+}

src/lib/ingestion/ingestion-manager.ts

@@ -34,7 +34,7 @@ export class IngestionManager {
 
     const db = getDb();
     const allProfiles = await db.select().from(profiles);
-    const pollInterval = parseInt(process.env.POLL_INTERVAL_SECONDS || "30") * 1000;
+    const pollInterval = parseInt(process.env.POLL_INTERVAL_SECONDS || "300") * 1000;
 
     log.info({ profileCount: allProfiles.length }, "Found profiles");
 
@@ -69,7 +69,7 @@ export class IngestionManager {
       return;
     }
 
-    const pollInterval = parseInt(process.env.POLL_INTERVAL_SECONDS || "30") * 1000;
+    const pollInterval = parseInt(process.env.POLL_INTERVAL_SECONDS || "300") * 1000;
     await this.startProfile(profileId, pollInterval);
   }
 

src/lib/ingestion/log-buffer.ts

@@ -0,0 +1,79 @@
+import { processLogBatch } from "./log-processor";
+import type { NextDNSLog } from "@/types/nextdns";
+import { createLogger } from "@/lib/logger";
+
+const log = createLogger("log-buffer");
+
+const DEFAULT_FLUSH_INTERVAL_MS = 2000;
+const DEFAULT_MAX_BUFFER_SIZE = 50;
+const OVERFLOW_LIMIT = 500;
+
+export class LogBuffer {
+  private profileId: string;
+  private buffer: NextDNSLog[] = [];
+  private flushTimer: ReturnType<typeof setTimeout> | null = null;
+  private flushing = false;
+  private flushIntervalMs: number;
+  private maxBufferSize: number;
+
+  constructor(
+    profileId: string,
+    flushIntervalMs = DEFAULT_FLUSH_INTERVAL_MS,
+    maxBufferSize = DEFAULT_MAX_BUFFER_SIZE
+  ) {
+    this.profileId = profileId;
+    this.flushIntervalMs = flushIntervalMs;
+    this.maxBufferSize = maxBufferSize;
+  }
+
+  add(entry: NextDNSLog) {
+    this.buffer.push(entry);
+
+    if (this.buffer.length >= this.maxBufferSize || this.buffer.length >= OVERFLOW_LIMIT) {
+      this.flush();
+      return;
+    }
+
+    if (!this.flushTimer) {
+      this.flushTimer = setTimeout(() => this.flush(), this.flushIntervalMs);
+    }
+  }
+
+  flush() {
+    if (this.flushTimer) {
+      clearTimeout(this.flushTimer);
+      this.flushTimer = null;
+    }
+
+    if (this.buffer.length === 0 || this.flushing) {
+      return;
+    }
+
+    const batch = this.buffer.splice(0);
+    this.flushing = true;
+
+    processLogBatch(this.profileId, batch)
+      .then((result) => {
+        if (result.inserted > 0) {
+          log.debug(
+            { profileId: this.profileId, attempted: result.attempted, inserted: result.inserted },
+            "Buffer flush complete"
+          );
+        }
+      })
+      .catch((error) => {
+        log.error({ err: error, profileId: this.profileId, batchSize: batch.length }, "Buffer flush error");
+      })
+      .finally(() => {
+        this.flushing = false;
+      });
+  }
+
+  destroy() {
+    if (this.flushTimer) {
+      clearTimeout(this.flushTimer);
+      this.flushTimer = null;
+    }
+    this.flush();
+  }
+}