meshcentralSync.js

Go to the documentation of this file.

/**
 * MeshCentral Auto-Sync Worker (Backup/Safety Net)
 * 
 * SYNC HIERARCHY:
 * 1. PRIMARY: meshcentral-device-sync.js (Direct DB queries, runs every 5 minutes)
 * 2. BACKUP: This worker (API-based, runs daily at 3 AM as safety net)
 * 3. DEPRECATED: POST /api/meshcentral/sync (Manual API endpoint, use /api/sync/meshcentral instead)
 * 
 * This worker runs daily as a backup to catch devices missed by the primary sync.
 * The primary sync method (meshcentral-device-sync.js) queries the MeshCentral
 * PostgreSQL database directly every 5 minutes, which is much faster and more reliable.
 * 
 * This worker serves as a safety net to:
 * - Catch devices missed by primary sync due to transient errors
 * - Verify sync accuracy once daily
 * - Update connection status for all devices comprehensive check
 * 
 * Uses hardware identifiers (hostname, MAC, motherboard UUID) for matching
 * 
 * TODO: When primary sync is proven stable for 30+ days, this worker can be
 * deprecated or converted to a weekly health check instead of daily.
 */
 
const { Worker, Queue } = require('bullmq');
const Redis = require('ioredis');
const db = require('../db');
const MeshCentralAPI = require('../lib/meshcentral-api');
 
const redisConfig = require('../config/redis');
const connection = new Redis({
  ...redisConfig,
  maxRetriesPerRequest: null // BullMQ requires this
});
 
// Create queue for MeshCentral sync jobs
const meshSyncQueue = new Queue('meshcentral-sync', { connection });
 
/**
 * Get MeshCentral API instance.
 * Creates authenticated MeshCentral API client with WebSocket connection.
 * @function getMeshAPI
 * @returns {Promise<MeshCentralAPI>} Authenticated API instance
 */
async function getMeshAPI() {
  const api = new MeshCentralAPI({
    url: process.env.MESHCENTRAL_URL || 'https://rmm-psa-meshcentral-aq48h.ondigitalocean.app',
    username: process.env.MESHCENTRAL_ADMIN_USER || 'admin',
    password: process.env.MESHCENTRAL_ADMIN_PASS || 'admin'
  });
  
  await api.login();
  await api.connectWebSocket();
  return api;
}
 
/**
 * Perform MeshCentral device sync.
 * Fetches devices from MeshCentral and matches with database agents.
 * @function syncMeshCentralDevices
 * @returns {Promise<void>}
 */
async function syncMeshCentralDevices() {
  console.log('\n🔄 [MeshSync] Starting automatic sync...');
  
  try {
    const api = await getMeshAPI();
    const nodes = await api.getNodes();
    
    console.log(`📊 [MeshSync] Found ${nodes.length} MeshCentral devices`);
    
    let linkedCount = 0;
    let updatedCount = 0;
    let newCount = 0;
    let skippedCount = 0;
    const matchDetails = [];
 
    for (const node of nodes) {
      try {
        const parsed = MeshCentralAPI.parseNodeData(node);
        
        if (!parsed.hostname) {
          console.log(`⚠️  [MeshSync] Skipping device without hostname: ${parsed.nodeId}`);
          skippedCount++;
          continue;
        }
 
        // Check if device already linked
        const existingLink = await db.query(
          'SELECT agent_id, hostname, hardware_data FROM agents WHERE meshcentral_nodeid = $1',
          [parsed.nodeId]
        );
 
        if (existingLink.rows.length > 0) {
          // Update connection status
          await db.query(`
            UPDATE agents 
            SET meshcentral_connected = $1,
                meshcentral_last_seen = $2,
                last_seen = GREATEST(last_seen, $2)
            WHERE meshcentral_nodeid = $3
          `, [
            parsed.connected,
            parsed.lastSeen,
            parsed.nodeId
          ]);
          updatedCount++;
          continue;
        }
 
        // Try to find matching agent by hostname
        const hostnameMatch = await db.query(
          'SELECT agent_id, agent_uuid, hostname FROM agents WHERE LOWER(hostname) = LOWER($1) AND meshcentral_nodeid IS NULL LIMIT 1',
          [parsed.hostname]
        );
 
        if (hostnameMatch.rows.length > 0) {
          const agent = hostnameMatch.rows[0];
          
          // Link agent to MeshCentral device
          await db.query(
            `UPDATE agents 
             SET meshcentral_nodeid = $1, 
                 meshcentral_connected = $2, 
                 meshcentral_last_seen = $3
             WHERE agent_id = $4`,
            [parsed.nodeId, parsed.connected, parsed.lastSeen, agent.agent_id]
          );
          
          linkedCount++;
          matchDetails.push({
            agentId: agent.agent_id,
            agentUuid: agent.agent_uuid,
            hostname: agent.hostname,
            nodeId: parsed.nodeId,
            method: 'hostname'
          });
          
          console.log(`✅ [MeshSync] Linked agent ${agent.agent_id} (${agent.hostname}) to MeshCentral node ${parsed.nodeId.substring(0, 20)}...`);
        } else {
          // Try MAC address matching if available
          let macMatched = false;
          
          if (parsed.hardware && parsed.hardware.macAddresses && parsed.hardware.macAddresses.length > 0) {
            for (const mac of parsed.hardware.macAddresses) {
              const macMatch = await db.query(`
                SELECT agent_id, agent_uuid, hostname, hardware_data
                FROM agents 
                WHERE meshcentral_nodeid IS NULL 
                AND hardware_data IS NOT NULL
                AND hardware_data::text ILIKE $1
                LIMIT 1
              `, [`%${mac}%`]);
 
              if (macMatch.rows.length > 0) {
                const agent = macMatch.rows[0];
                
                await db.query(
                  `UPDATE agents 
                   SET meshcentral_nodeid = $1, 
                       meshcentral_connected = $2, 
                       meshcentral_last_seen = $3
                   WHERE agent_id = $4`,
                  [parsed.nodeId, parsed.connected, parsed.lastSeen, agent.agent_id]
                );
                
                linkedCount++;
                macMatched = true;
                matchDetails.push({
                  agentId: agent.agent_id,
                  agentUuid: agent.agent_uuid,
                  hostname: agent.hostname,
                  nodeId: parsed.nodeId,
                  method: 'mac_address',
                  mac: mac
                });
                
                console.log(`✅ [MeshSync] Linked agent ${agent.agent_id} (${agent.hostname}) to MeshCentral node via MAC ${mac}`);
                break;
              }
            }
          }
          
          if (!macMatched) {
            // No match found - could create new agent or skip
            console.log(`ℹ️  [MeshSync] No matching agent for device: ${parsed.hostname} (${parsed.nodeId.substring(0, 20)}...)`);
            skippedCount++;
          }
        }
      } catch (err) {
        console.error(`❌ [MeshSync] Error processing node:`, err.message);
      }
    }
 
    const summary = {
      timestamp: new Date().toISOString(),
      total: nodes.length,
      linked: linkedCount,
      updated: updatedCount,
      new: newCount,
      skipped: skippedCount,
      matchDetails: matchDetails
    };
 
    console.log(`\n📈 [MeshSync] Sync complete:`);
    console.log(`   Total devices: ${summary.total}`);
    console.log(`   Newly linked: ${summary.linked}`);
    console.log(`   Updated: ${summary.updated}`);
    console.log(`   Skipped: ${summary.skipped}`);
    console.log('');
 
    return summary;
 
  } catch (error) {
    console.error('❌ [MeshSync] Sync failed:', error.message);
    throw error;
  }
}
 
/**
 * BullMQ Worker - Processes sync jobs
 */
const worker = new Worker('meshcentral-sync', async (job) => {
  console.log(`\n🔄 [MeshSync Worker] Processing job ${job.id}...`);
  
  try {
    const result = await syncMeshCentralDevices();
    return result;
  } catch (error) {
    console.error(`❌ [MeshSync Worker] Job ${job.id} failed:`, error.message);
    throw error;
  }
}, {
  connection,
  concurrency: 1, // Only run one sync job at a time
  limiter: {
    max: 1,
    duration: 60000 // Max 1 job per minute
  }
});
 
worker.on('completed', (job, result) => {
  console.log(`✅ [MeshSync Worker] Job ${job.id} completed: ${result.linked} linked, ${result.updated} updated`);
});
 
worker.on('failed', (job, err) => {
  console.error(`❌ [MeshSync Worker] Job ${job?.id} failed:`, err.message);
});
 
/**
 * Schedule recurring sync job (every 10 minutes)
 */
async function scheduleRecurringSync() {
  // Remove any existing repeatable jobs
  const repeatableJobs = await meshSyncQueue.getRepeatableJobs();
  for (const job of repeatableJobs) {
    await meshSyncQueue.removeRepeatableByKey(job.key);
  }
 
  // Add new repeatable job - daily at 3 AM (backup to webhook system)
  await meshSyncQueue.add(
    'sync-devices',
    {},
    {
      repeat: {
        pattern: '0 3 * * *', // Daily at 3:00 AM
        immediately: true // Run immediately on startup
      },
      removeOnComplete: {
        count: 10 // Keep last 10 completed jobs
      },
      removeOnFail: {
        count: 20 // Keep last 20 failed jobs
      }
    }
  );
 
  console.log('🕐 [MeshSync] Scheduled to run daily at 3:00 AM (backup sync)');
  console.log('ℹ️  [MeshSync] Primary sync via webhook: /api/webhooks/meshcentral');
}
 
// Initialize on startup
scheduleRecurringSync().catch(err => {
  console.error('❌ [MeshSync] Failed to schedule sync:', err);
});
 
// Manual trigger function (can be called from API)
/**
 * Trigger manual MeshCentral device sync.
 * Adds high-priority sync job to queue.
 * @function triggerManualSync
 * @returns {Promise<string>} Job ID of queued sync task
 */
async function triggerManualSync() {
  const job = await meshSyncQueue.add('manual-sync', {}, { priority: 1 });
  console.log(`🔄 [MeshSync] Manual sync triggered - Job ID: ${job.id}`);
  return job.id;
}
 
console.log('✅ [MeshSync] Worker initialized and ready');
 
module.exports = {
  meshSyncQueue,
  triggerManualSync,
  syncMeshCentralDevices
};