feat(security): add queued IP intelligence observation and filtered retrieval for network and security views

ts/monitoring/classes.metricsmanager.ts

@@ -725,7 +725,10 @@ export class MetricsManager {
         .slice(0, 10)
         .map(([ip, data]) => ({ ip, count: data.count, bwIn: data.bwIn, bwOut: data.bwOut }));
 
-      void this.dcRouter.securityPolicyManager?.observeIps([...allIPData.keys()]);
+      this.dcRouter.securityPolicyManager?.queueObservedIps([
+        ...topIPs.map((item) => item.ip),
+        ...topIPsByBandwidth.map((item) => item.ip),
+      ]);
 
       // Build domain activity using per-IP domain request counts from Rust engine
       const connectionsByRoute = proxyMetrics.connections.byRoute();

ts/opsserver/handlers/security.handler.ts

@@ -180,7 +180,14 @@ export class SecurityHandler {
         async (dataArg) => {
           await requireOpsAuth(this.opsServerRef, dataArg, { scope: 'security:read' });
           const manager = this.opsServerRef.dcRouterRef.securityPolicyManager;
-          return { records: manager ? await manager.listIpIntelligence() : [] };
+          return {
+            records: manager
+              ? await manager.listIpIntelligence({
+                  ipAddresses: dataArg.ipAddresses,
+                  limit: dataArg.limit,
+                })
+              : [],
+          };
         },
       ),
     );

ts/security/classes.security-policy-manager.ts

@@ -19,12 +19,24 @@ export interface IRemoteIngressFirewallSnapshot {
   blockedIps: string[];
 }
 
+const OBSERVED_IP_QUEUE_LIMIT = 512;
+const OBSERVED_IP_BATCH_LIMIT = 20;
+const OBSERVED_IP_QUEUE_CONCURRENCY = 2;
+const OBSERVED_IP_REQUEUE_THROTTLE_MS = 60_000;
+
 export class SecurityPolicyManager {
   private readonly smartNetwork = new plugins.smartnetwork.SmartNetwork({
     cacheTtl: 24 * 60 * 60 * 1000,
+    ipIntelligenceTimeout: 5_000,
   });
   private readonly intelligenceRefreshMs: number;
-  private readonly inFlightObservations = new Set<string>();
+  private readonly inFlightObservations = new Map<string, Promise<void>>();
+  private readonly queuedObservations = new Set<string>();
+  private readonly observationQueue: string[] = [];
+  private readonly lastQueuedAt = new Map<string, number>();
+  private activeQueuedObservations = 0;
+  private queueDrainScheduled = false;
+  private isStopping = false;
   private readonly onPolicyChanged?: () => void | Promise<void>;
 
   constructor(options: ISecurityPolicyManagerOptions = {}) {
@@ -37,6 +49,9 @@ export class SecurityPolicyManager {
   }
 
   public async stop(): Promise<void> {
+    this.isStopping = true;
+    this.observationQueue.length = 0;
+    this.queuedObservations.clear();
     await this.smartNetwork.stop();
   }
 
@@ -45,13 +60,55 @@ export class SecurityPolicyManager {
     await Promise.allSettled(uniqueIps.map((ip) => this.observeIp(ip)));
   }
 
+  public queueObservedIps(ips: string[]): void {
+    if (this.isStopping) return;
+
+    const now = Date.now();
+    const uniqueIps = [...new Set(ips.map((ip) => this.normalizeIp(ip)).filter(Boolean) as string[])];
+
+    for (const ip of uniqueIps.slice(0, OBSERVED_IP_BATCH_LIMIT)) {
+      if (!this.isPublicIp(ip)) continue;
+      if (this.inFlightObservations.has(ip) || this.queuedObservations.has(ip)) continue;
+
+      const lastQueuedAt = this.lastQueuedAt.get(ip);
+      if (lastQueuedAt && now - lastQueuedAt < OBSERVED_IP_REQUEUE_THROTTLE_MS) continue;
+
+      if (this.observationQueue.length >= OBSERVED_IP_QUEUE_LIMIT) {
+        const droppedIp = this.observationQueue.shift();
+        if (droppedIp) this.queuedObservations.delete(droppedIp);
+      }
+
+      this.observationQueue.push(ip);
+      this.queuedObservations.add(ip);
+      this.lastQueuedAt.set(ip, now);
+    }
+
+    this.pruneQueuedIpMemory(now);
+    this.scheduleQueueDrain();
+  }
+
   public async observeIp(ipAddress: string, options: { force?: boolean } = {}): Promise<void> {
     const ip = this.normalizeIp(ipAddress);
-    if (!ip || !this.isPublicIp(ip) || this.inFlightObservations.has(ip)) {
+    if (!ip || !this.isPublicIp(ip)) {
       return;
     }
 
-    this.inFlightObservations.add(ip);
+    const existingObservation = this.inFlightObservations.get(ip);
+    if (existingObservation) {
+      await existingObservation;
+      if (!options.force) return;
+    }
+
+    const observationPromise = this.performObserveIp(ip, options).finally(() => {
+      if (this.inFlightObservations.get(ip) === observationPromise) {
+        this.inFlightObservations.delete(ip);
+      }
+    });
+    this.inFlightObservations.set(ip, observationPromise);
+    await observationPromise;
+  }
+
+  private async performObserveIp(ip: string, options: { force?: boolean } = {}): Promise<void> {
     try {
       const now = Date.now();
       let doc = await IpIntelligenceDoc.findByIp(ip);
@@ -81,8 +138,6 @@ export class SecurityPolicyManager {
       }
     } catch (err) {
       logger.log('warn', `Failed to enrich IP ${ip}: ${(err as Error).message}`);
-    } finally {
-      this.inFlightObservations.delete(ip);
     }
   }
 
@@ -90,8 +145,22 @@ export class SecurityPolicyManager {
     return (await SecurityBlockRuleDoc.findAll()).map((doc) => this.ruleFromDoc(doc));
   }
 
-  public async listIpIntelligence(): Promise<IIpIntelligenceRecord[]> {
-    return (await IpIntelligenceDoc.findAll()).map((doc) => this.intelligenceFromDoc(doc));
+  public async listIpIntelligence(options: { ipAddresses?: string[]; limit?: number } = {}): Promise<IIpIntelligenceRecord[]> {
+    const limit = Number.isInteger(options.limit) && options.limit! > 0
+      ? Math.min(options.limit!, 500)
+      : undefined;
+
+    let docs: IpIntelligenceDoc[];
+    if (options.ipAddresses?.length) {
+      const ips = [...new Set(options.ipAddresses.map((ip) => this.normalizeIp(ip)).filter(Boolean) as string[])];
+      const results = await Promise.all(ips.map((ip) => IpIntelligenceDoc.findByIp(ip)));
+      docs = results.filter(Boolean) as IpIntelligenceDoc[];
+    } else {
+      docs = await IpIntelligenceDoc.findAll();
+    }
+
+    const sortedDocs = docs.sort((a, b) => (b.lastSeenAt || 0) - (a.lastSeenAt || 0));
+    return (limit ? sortedDocs.slice(0, limit) : sortedDocs).map((doc) => this.intelligenceFromDoc(doc));
   }
 
   public async refreshIpIntelligence(ipAddress: string): Promise<IIpIntelligenceRecord | null> {
@@ -104,6 +173,45 @@ export class SecurityPolicyManager {
     return doc ? this.intelligenceFromDoc(doc) : null;
   }
 
+  private scheduleQueueDrain(): void {
+    if (this.queueDrainScheduled || this.isStopping) return;
+    this.queueDrainScheduled = true;
+    setTimeout(() => {
+      this.queueDrainScheduled = false;
+      this.drainObservationQueue();
+    }, 0);
+  }
+
+  private drainObservationQueue(): void {
+    if (this.isStopping) return;
+
+    while (
+      this.activeQueuedObservations < OBSERVED_IP_QUEUE_CONCURRENCY &&
+      this.observationQueue.length > 0
+    ) {
+      const ip = this.observationQueue.shift()!;
+      this.queuedObservations.delete(ip);
+      this.activeQueuedObservations++;
+      void this.observeIp(ip)
+        .catch(() => undefined)
+        .finally(() => {
+          this.activeQueuedObservations--;
+          if (this.observationQueue.length > 0) {
+            this.scheduleQueueDrain();
+          }
+        });
+    }
+  }
+
+  private pruneQueuedIpMemory(now: number): void {
+    if (this.lastQueuedAt.size <= OBSERVED_IP_QUEUE_LIMIT * 2) return;
+    for (const [ip, lastQueuedAt] of this.lastQueuedAt) {
+      if (now - lastQueuedAt > OBSERVED_IP_REQUEUE_THROTTLE_MS * 2) {
+        this.lastQueuedAt.delete(ip);
+      }
+    }
+  }
+
   public async listAuditEvents(limit = 100): Promise<ISecurityPolicyAuditEvent[]> {
     return (await SecurityPolicyAuditDoc.findRecent(limit)).map((doc) => ({
       id: doc.id,