digitalOceanSyncWorker.js

Go to the documentation of this file.

/**
 * Digital Ocean Sync Worker
 * Syncs DO resources (apps, droplets, databases) to local database every 5 minutes
 * This provides faster page loads and reduces API calls to DigitalOcean
 */
 
const pool = require('./db');
const DigitalOceanService = require('./services/digitalOceanService');
 
const SYNC_INTERVAL = 5 * 60 * 1000; // 5 minutes
 
/**
 * Sync DO resources for a single tenant
 * @param tenantId
 */
async function syncTenantResources(tenantId) {
  try {
    console.log(`[DO Sync] Starting sync for tenant ${tenantId}`);
    
    const doService = new DigitalOceanService(tenantId);
    const resources = await doService.getAllResources();
 
    // Sync Apps
    const activeAppIds = [];
    if (resources.apps && resources.apps.length > 0) {
      for (const app of resources.apps) {
        activeAppIds.push(app.id);
        await pool.query(
          `INSERT INTO hosting_apps (
            tenant_id, do_app_id, app_name, app_type, status, region,
            live_url, metadata, updated_at
          ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW())
          ON CONFLICT (do_app_id) DO UPDATE SET
            app_name = EXCLUDED.app_name,
            app_type = EXCLUDED.app_type,
            status = EXCLUDED.status,
            region = EXCLUDED.region,
            live_url = EXCLUDED.live_url,
            metadata = EXCLUDED.metadata,
            updated_at = NOW()`,
          [
            tenantId,
            app.id,
            app.spec?.name || 'Unnamed App',
            app.spec?.services?.[0]?.build_command ? 'nodejs' : 'static',
            app.phase || 'unknown',
            app.region?.slug || null,
            app.live_url || null,
            JSON.stringify(app)
          ]
        );
      }
      console.log(`[DO Sync] Synced ${resources.apps.length} apps for tenant ${tenantId}`);
    }
 
    // Mark apps as deleted if they no longer exist in DO
    if (activeAppIds.length > 0) {
      const deleteResult = await pool.query(
        `UPDATE hosting_apps 
         SET status = 'deleted', updated_at = NOW()
         WHERE tenant_id = $1 
         AND do_app_id NOT IN (${activeAppIds.map((_, i) => `$${i + 2}`).join(',')})
         AND status != 'deleted'`,
        [tenantId, ...activeAppIds]
      );
      if (deleteResult.rowCount > 0) {
        console.log(`[DO Sync] Marked ${deleteResult.rowCount} apps as deleted for tenant ${tenantId}`);
      }
    } else {
      // If no apps in DO, mark all as deleted
      const deleteResult = await pool.query(
        `UPDATE hosting_apps 
         SET status = 'deleted', updated_at = NOW()
         WHERE tenant_id = $1 AND status != 'deleted'`,
        [tenantId]
      );
      if (deleteResult.rowCount > 0) {
        console.log(`[DO Sync] Marked ${deleteResult.rowCount} apps as deleted (no active apps in DO)`);
      }
    }
 
    // Sync Droplets
    const activeDropletIds = [];
    if (resources.droplets && resources.droplets.length > 0) {
      for (const droplet of resources.droplets) {
        activeDropletIds.push(droplet.id);
        const publicIp = droplet.networks?.v4?.find(n => n.type === 'public')?.ip_address;
        const privateIp = droplet.networks?.v4?.find(n => n.type === 'private')?.ip_address;
 
        await pool.query(
          `INSERT INTO hosting_droplets (
            tenant_id, do_droplet_id, droplet_name, status, ip_address,
            private_ip_address, region, size, image, vcpus, memory, disk,
            tags, metadata, updated_at
          ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, NOW())
          ON CONFLICT (do_droplet_id) DO UPDATE SET
            droplet_name = EXCLUDED.droplet_name,
            status = EXCLUDED.status,
            ip_address = EXCLUDED.ip_address,
            private_ip_address = EXCLUDED.private_ip_address,
            region = EXCLUDED.region,
            size = EXCLUDED.size,
            image = EXCLUDED.image,
            vcpus = EXCLUDED.vcpus,
            memory = EXCLUDED.memory,
            disk = EXCLUDED.disk,
            tags = EXCLUDED.tags,
            metadata = EXCLUDED.metadata,
            updated_at = NOW()`,
          [
            tenantId,
            droplet.id,
            droplet.name,
            droplet.status,
            publicIp || null,
            privateIp || null,
            droplet.region?.slug || null,
            droplet.size_slug || null,
            droplet.image?.slug || droplet.image?.name || null,
            droplet.vcpus || null,
            droplet.memory || null,
            droplet.disk || null,
            JSON.stringify(droplet.tags || []),
            JSON.stringify(droplet)
          ]
        );
      }
      console.log(`[DO Sync] Synced ${resources.droplets.length} droplets for tenant ${tenantId}`);
    }
 
    // Mark droplets as deleted if they no longer exist in DO
    if (activeDropletIds.length > 0) {
      const deleteResult = await pool.query(
        `UPDATE hosting_droplets 
         SET status = 'deleted', updated_at = NOW()
         WHERE tenant_id = $1 
         AND do_droplet_id NOT IN (${activeDropletIds.map((_, i) => `$${i + 2}`).join(',')})
         AND status != 'deleted'`,
        [tenantId, ...activeDropletIds]
      );
      if (deleteResult.rowCount > 0) {
        console.log(`[DO Sync] Marked ${deleteResult.rowCount} droplets as deleted for tenant ${tenantId}`);
      }
    } else {
      const deleteResult = await pool.query(
        `UPDATE hosting_droplets 
         SET status = 'deleted', updated_at = NOW()
         WHERE tenant_id = $1 AND status != 'deleted'`,
        [tenantId]
      );
      if (deleteResult.rowCount > 0) {
        console.log(`[DO Sync] Marked ${deleteResult.rowCount} droplets as deleted (no active droplets in DO)`);
      }
    }
 
    // Sync Databases
    const activeDatabaseIds = [];
    if (resources.databases && resources.databases.length > 0) {
      for (const db of resources.databases) {
        activeDatabaseIds.push(db.id);
        await pool.query(
          `INSERT INTO hosting_databases (
            tenant_id, do_database_id, database_name, engine, version, status,
            region, size, num_nodes, connection_host, connection_port,
            tags, metadata, updated_at
          ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, NOW())
          ON CONFLICT (do_database_id) DO UPDATE SET
            database_name = EXCLUDED.database_name,
            engine = EXCLUDED.engine,
            version = EXCLUDED.version,
            status = EXCLUDED.status,
            region = EXCLUDED.region,
            size = EXCLUDED.size,
            num_nodes = EXCLUDED.num_nodes,
            connection_host = EXCLUDED.connection_host,
            connection_port = EXCLUDED.connection_port,
            tags = EXCLUDED.tags,
            metadata = EXCLUDED.metadata,
            updated_at = NOW()`,
          [
            tenantId,
            db.id,
            db.name,
            db.engine,
            db.version || null,
            db.status,
            db.region || null,
            db.size || null,
            db.num_nodes || 1,
            db.connection?.host || null,
            db.connection?.port || null,
            JSON.stringify(db.tags || []),
            JSON.stringify(db)
          ]
        );
      }
      console.log(`[DO Sync] Synced ${resources.databases.length} databases for tenant ${tenantId}`);
    }
 
    // Mark databases as deleted if they no longer exist in DO
    if (activeDatabaseIds.length > 0) {
      const deleteResult = await pool.query(
        `UPDATE hosting_databases 
         SET status = 'deleted', updated_at = NOW()
         WHERE tenant_id = $1 
         AND do_database_id NOT IN (${activeDatabaseIds.map((_, i) => `$${i + 2}`).join(',')})
         AND status != 'deleted'`,
        [tenantId, ...activeDatabaseIds]
      );
      if (deleteResult.rowCount > 0) {
        console.log(`[DO Sync] Marked ${deleteResult.rowCount} databases as deleted for tenant ${tenantId}`);
      }
    } else {
      const deleteResult = await pool.query(
        `UPDATE hosting_databases 
         SET status = 'deleted', updated_at = NOW()
         WHERE tenant_id = $1 AND status != 'deleted'`,
        [tenantId]
      );
      if (deleteResult.rowCount > 0) {
        console.log(`[DO Sync] Marked ${deleteResult.rowCount} databases as deleted (no active databases in DO)`);
      }
    }
 
    // Update last sync time
    await pool.query(
      `UPDATE digitalocean_config SET last_sync_at = NOW() WHERE tenant_id = $1`,
      [tenantId]
    );
 
    console.log(`[DO Sync] Completed sync for tenant ${tenantId}`);
  } catch (error) {
    console.error(`[DO Sync] Error syncing tenant ${tenantId}:`, error.message);
  }
}
 
/**
 * Sync all active tenants
 */
async function syncAllTenants() {
  try {
    const result = await pool.query(
      `SELECT tenant_id FROM digitalocean_config WHERE is_active = true`
    );
 
    if (result.rows.length === 0) {
      console.log('[DO Sync] No active DO configurations found');
      return;
    }
 
    console.log(`[DO Sync] Starting sync for ${result.rows.length} tenant(s)`);
 
    for (const row of result.rows) {
      await syncTenantResources(row.tenant_id);
    }
 
    console.log('[DO Sync] All tenants synced successfully');
  } catch (error) {
    console.error('[DO Sync] Error in sync cycle:', error);
  }
}
 
/**
 * Start the sync worker
 */
function startSyncWorker() {
  console.log(`[DO Sync] Worker started - syncing every ${SYNC_INTERVAL / 1000 / 60} minutes`);
  
  // Run immediately on start
  syncAllTenants();
  
  // Then run every 5 minutes
  setInterval(syncAllTenants, SYNC_INTERVAL);
}
 
// Start the worker if this file is run directly
if (require.main === module) {
  startSyncWorker();
}
 
module.exports = { startSyncWorker, syncAllTenants, syncTenantResources };