feat(release): v1.14.0 discovery and conf pages

2026-08-01 01:21:27 -04:00
parent ecc9b62842
commit 011d4b2975
42 changed files with 3091 additions and 96 deletions
@@ -0,0 +1,140 @@
+const { Resource, ResourceEdge, ResourceGroup } = require('../models/resource');
+const { WebhookEmitter } = require('./webhook_emitter');
+const crypto = require('crypto');
+
+class DiscoveryReconciler {
+  static async reconcile(sourceName, payload) {
+    const { resources = [], edges = [] } = payload;
+    let newDevices = 0;
+
+    for (const res of resources) {
+      if (!res.metadata) res.metadata = {};
+      
+      let existing = null;
+      
+      // Attempt matching by MAC if available
+      if (res.metadata.interfaces && res.metadata.interfaces.length > 0) {
+        const macs = res.metadata.interfaces.map(i => i.mac).filter(m => !!m);
+        if (macs.length > 0) {
+          const allRes = await Resource.list();
+          existing = allRes.find(r => 
+            r.metadata && r.metadata.interfaces && 
+            r.metadata.interfaces.some(i => macs.includes(i.mac))
+          );
+        }
+      }
+      
+      // Fallback matching by IP if no MAC match (weaker)
+      let ipsToMatch = [];
+      if (res.metadata.interfaces) {
+        ipsToMatch = res.metadata.interfaces.map(i => i.ip).filter(i => !!i);
+      }
+      if (res.metadata.address) {
+        res.metadata.address.split(',').forEach(a => ipsToMatch.push(a.trim()));
+      }
+      
+      if (!existing && ipsToMatch.length > 0) {
+        const allRes = await Resource.list();
+        existing = allRes.find(r => {
+          if (!r.metadata) return false;
+          if (r.metadata.address) {
+            const addrs = r.metadata.address.split(',').map(a => a.trim());
+            if (addrs.some(a => ipsToMatch.includes(a))) return true;
+          }
+          if (r.metadata.interfaces && r.metadata.interfaces.some(i => ipsToMatch.includes(i.ip))) return true;
+          return false;
+        });
+      }
+      
+      // Fallback matching by Slug or Name
+      if (!existing && (res.slug || res.name)) {
+        const allRes = await Resource.list();
+        existing = allRes.find(r => 
+          (res.slug && r.slug === res.slug) || 
+          (res.name && r.name && r.name.toLowerCase() === res.name.toLowerCase())
+        );
+      }
+
+      if (existing) {
+        // Merge metadata
+        const mergedMeta = { ...existing.metadata, ...res.metadata };
+        
+        // Merge interfaces cleanly
+        if (res.metadata.interfaces) {
+          const existingIntfs = existing.metadata.interfaces || [];
+          const newIntfs = res.metadata.interfaces;
+          // Simple union based on mac or ip
+          for (const ni of newIntfs) {
+            const idx = existingIntfs.findIndex(ei => (ni.mac && ei.mac === ni.mac) || (ni.ip && ei.ip === ni.ip));
+            if (idx >= 0) existingIntfs[idx] = { ...existingIntfs[idx], ...ni };
+            else existingIntfs.push(ni);
+          }
+          mergedMeta.interfaces = existingIntfs;
+        }
+
+        // Add discovery source
+        const sources = new Set(mergedMeta.discovery_sources || []);
+        sources.add(sourceName);
+        mergedMeta.discovery_sources = [...sources];
+        
+        mergedMeta.last_seen = Date.now();
+        
+        await existing.update({
+          name: res.name || existing.name,
+          description: res.description || existing.description,
+          metadata: mergedMeta,
+          updated_on: Math.floor(Date.now() / 1000)
+        });
+      } else {
+        // Create new
+        const sources = [sourceName];
+        res.metadata.discovery_sources = sources;
+        res.metadata.last_seen = Date.now();
+        
+        const slug = res.slug || `${res.kind}-${crypto.randomBytes(4).toString('hex')}`;
+        
+        const created = await Resource.create({
+          id: crypto.randomUUID(),
+          kind: res.kind || 'unmanaged_device',
+          name: res.name || slug,
+          slug: slug,
+          metadata: res.metadata,
+          created_on: Math.floor(Date.now() / 1000)
+        });
+        
+        newDevices++;
+        WebhookEmitter.emit('discovery.new_device', created.toJSON());
+      }
+    }
+    
+    // We can handle edges similarly if needed, but for simplicity we assume edges are managed elsewhere 
+    // or we just trust the plugins to give us explicit parent-child mappings by slug.
+    
+    if (newDevices > 0) {
+      console.log(`[DiscoveryReconciler] Source ${sourceName} discovered ${newDevices} new devices.`);
+    }
+  }
+
+  static async garbageCollect(staleMs = 7 * 24 * 60 * 60 * 1000) {
+    const allRes = await Resource.list();
+    const cutoff = Date.now() - staleMs;
+    let archived = 0;
+
+    for (const res of allRes) {
+      const meta = res.metadata || {};
+      const sources = meta.discovery_sources || [];
+      // Only garbage collect things that are exclusively auto-discovered
+      if (sources.length > 0 && !sources.includes('manual')) {
+        if (meta.last_seen && meta.last_seen < cutoff && meta.lifecycle_state !== 'archived') {
+          meta.lifecycle_state = 'archived';
+          await res.update({ metadata: meta, updated_on: Math.floor(Date.now() / 1000) });
+          archived++;
+          WebhookEmitter.emit('discovery.device_archived', res.toJSON());
+        }
+      }
+    }
+    if (archived > 0) console.log(`[DiscoveryReconciler] Garbage collected ${archived} stale devices.`);
+  }
+}
+
+module.exports = { DiscoveryReconciler };
@@ -0,0 +1,87 @@
+const { Queue, Worker } = require('bullmq');
+const { DiscoveryReconciler } = require('./discovery_reconciler');
+const Redis = require('ioredis');
+
+// Ensure Redis connection works for BullMQ
+const redisOpts = { maxRetriesPerRequest: null };
+const connection = new Redis(process.env.REDIS_URL || 'redis://127.0.0.1:6379', redisOpts);
+
+const discoveryQueue = new Queue('discovery', { connection });
+
+// Load plugins
+const fs = require('fs');
+const path = require('path');
+const pluginsDir = path.join(__dirname, '../plugins/discovery');
+
+let plugins = {};
+
+if (fs.existsSync(pluginsDir)) {
+  fs.readdirSync(pluginsDir).forEach(file => {
+    if (file.endsWith('.js')) {
+      const name = path.basename(file, '.js');
+      plugins[name] = require(path.join(pluginsDir, file));
+    }
+  });
+}
+
+const worker = new Worker('discovery', async job => {
+  if (job.name === 'run_plugin') {
+    const { pluginName, config } = job.data;
+    if (plugins[pluginName]) {
+      console.log(`[Scheduler] Running plugin: ${pluginName}`);
+      try {
+        const payload = await plugins[pluginName].discover(config);
+        await DiscoveryReconciler.reconcile(pluginName, payload);
+      } catch (err) {
+        console.error(`[Scheduler] Plugin ${pluginName} failed:`, err);
+      }
+    }
+  } else if (job.name === 'garbage_collect') {
+    console.log(`[Scheduler] Running garbage collection`);
+    await DiscoveryReconciler.garbageCollect();
+  }
+}, { connection });
+
+// Function to start scheduling
+async function initScheduler(discoveryConfig) {
+  // Clear old repeatable jobs (BullMQ v6 uses JobSchedulers)
+  try {
+    const schedulers = await discoveryQueue.getJobSchedulers();
+    for (const job of schedulers) {
+      await discoveryQueue.removeJobScheduler(job.id);
+    }
+  } catch (e) {
+    console.log('[Scheduler] Could not clear old job schedulers (may not be supported or none exist)');
+  }
+
+  // Schedule Garbage Collection
+  await discoveryQueue.add('garbage_collect', {}, { repeat: { pattern: '0 0 * * *' } }); // Daily
+
+  // Load plugin overrides from Redis
+  let overrides = {};
+  try {
+    const data = await connection.hgetall('discovery_plugins');
+    for (const [k, v] of Object.entries(data)) {
+      overrides[k] = JSON.parse(v);
+    }
+  } catch (err) {
+    console.error('[Scheduler] Failed to load plugin overrides from Redis', err);
+  }
+
+  // Schedule Plugins based on config + overrides
+  if (discoveryConfig && discoveryConfig.plugins) {
+    for (const [name, config] of Object.entries(discoveryConfig.plugins)) {
+      const mergedConfig = { ...config, ...(overrides[name] || {}) };
+      if (mergedConfig.enabled && plugins[name]) {
+        const cron = mergedConfig.cron || '0 * * * *'; // Default hourly
+        await discoveryQueue.add('run_plugin', { pluginName: name, config: mergedConfig }, { repeat: { pattern: cron } });
+        console.log(`[Scheduler] Scheduled plugin ${name} with cron ${cron}`);
+        
+        // Also run once immediately
+        await discoveryQueue.add('run_plugin', { pluginName: name, config: mergedConfig });
+      }
+    }
+  }
+}
+
+module.exports = { initScheduler, discoveryQueue, connection };
@@ -0,0 +1,35 @@
+const { Webhook } = require('../models/webhook');
+const crypto = require('crypto');
+const fetch = require('node-fetch');
+
+class WebhookEmitter {
+  static async emit(event, payload) {
+    try {
+      const hooks = await Webhook.list({ where: { isActive: true } });
+      const matched = hooks.filter(h => !h.events || h.events.length === 0 || h.events.includes(event));
+      
+      for (const hook of matched) {
+        this.sendPayload(hook, event, payload).catch(err => console.error(`Webhook ${hook.name} failed:`, err.message));
+      }
+    } catch (e) {
+      console.error('Error emitting webhook:', e);
+    }
+  }
+
+  static async sendPayload(hook, event, payload) {
+    const body = JSON.stringify({ event, payload, timestamp: Date.now() });
+    const headers = { 'Content-Type': 'application/json' };
+    
+    if (hook.secret) {
+      const signature = crypto.createHmac('sha256', hook.secret).update(body).digest('hex');
+      headers['X-Theta-Signature'] = signature;
+    }
+
+    const res = await fetch(hook.url, { method: 'POST', body, headers, timeout: 5000 });
+    if (!res.ok) {
+      throw new Error(`Status ${res.status}`);
+    }
+  }
+}
+
+module.exports = { WebhookEmitter };