pax8SyncWorker.js

Go to the documentation of this file.

/**
 * Pax8 Product Sync Worker
 * Syncs Office 365 products from Pax8 API to local database
 * - Runs daily at 3am to sync product catalog changes
 * - Runs on startup to ensure fresh product data
 * - Can be triggered via Redis pub/sub for manual sync
 */
const Redis = require("ioredis");
const pool = require('./services/db');
const Pax8Service = require('./services/pax8Service');
 
const redisConfig = require('./config/redis');
const redis = new Redis(redisConfig);
 
const SYNC_INTERVAL = 24 * 60 * 60 * 1000; // 24 hours
const SYNC_HOUR = 3; // 3am
 
console.log('[Pax8 Sync Worker] Starting...');
 
/**
 * Sync products for a single tenant
 * @param tenantId
 * @param vendor
 * @param force
 */
async function syncTenantProducts(tenantId, vendor = 'Microsoft', force = false) {
  console.log(`[Pax8 Sync] Starting sync for tenant ${tenantId}, vendor: ${vendor}`);
 
  try {
    const pax8 = new Pax8Service(tenantId);
    
    // Fetch products from Pax8 API
    const products = await pax8.listProducts({ vendor });
 
    if (!products || products.length === 0) {
      console.log(`[Pax8 Sync] No products found for vendor ${vendor}`);
      return { total: 0, synced: 0, updated: 0, skipped: 0 };
    }
 
    console.log(`[Pax8 Sync] Found ${products.length} products, syncing to database...`);
 
    let synced = 0;
    let updated = 0;
    let skipped = 0;
 
    for (const product of products) {
      try {
        // Check if product already exists by pax8_product_id
        const existingResult = await pool.query(
          'SELECT product_id, last_synced_at FROM products WHERE pax8_product_id = $1',
          [product.id]
        );
 
        const now = new Date();
        const oneHourAgo = new Date(now.getTime() - 60 * 60 * 1000);
 
        // Skip if recently synced and not forcing
        if (!force && existingResult.rows.length > 0 && existingResult.rows[0].last_synced_at) {
          const lastSync = new Date(existingResult.rows[0].last_synced_at);
          if (lastSync > oneHourAgo) {
            skipped++;
            continue;
          }
        }
 
        // Extract product details
        const name = product.name || product.productName || 'Unknown Product';
        const description = product.description || product.productDescription || '';
        const priceRetail = product.pricing?.retail || product.price || 0;
        const priceExTax = product.pricing?.cost || priceRetail;
        const pax8Vendor = product.vendor || vendor;
        const pax8Category = product.category || null;
        const pax8Subcategory = product.subcategory || product.subCategory || null;
        const pax8Sku = product.sku || product.productSku || null;
        const billingTerm = product.billingTerm || product.billing?.term || 'monthly';
        const isAutoRenew = product.autoRenew !== false;
        const pax8Metadata = JSON.stringify(product);
 
        if (existingResult.rows.length > 0) {
          // Update existing product
          await pool.query(
            `UPDATE products SET
              name = $1, description = $2, price_retail = $3, price_ex_tax = $4,
              pax8_vendor = $5, pax8_category = $6, pax8_subcategory = $7, pax8_sku = $8,
              billing_term = $9, is_auto_renew = $10, pax8_metadata = $11,
              last_synced_at = NOW(), updated_at = NOW()
            WHERE pax8_product_id = $12`,
            [name, description, priceRetail, priceExTax, pax8Vendor, pax8Category, 
             pax8Subcategory, pax8Sku, billingTerm, isAutoRenew, pax8Metadata, product.id]
          );
          updated++;
        } else {
          // Insert new product
          await pool.query(
            `INSERT INTO products (
              tenant_id, name, description, supplier, price_retail, price_ex_tax,
              is_service, divisible, is_stock,
              pax8_product_id, pax8_vendor, pax8_category, pax8_subcategory,
              pax8_sku, billing_term, is_auto_renew, pax8_metadata,
              is_active, last_synced_at, created_at, updated_at
            ) VALUES (
              $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, NOW(), NOW(), NOW()
            )`,
            [tenantId, name, description, 'Pax8', priceRetail, priceExTax,
             true, true, false, product.id, pax8Vendor, pax8Category, pax8Subcategory,
             pax8Sku, billingTerm, isAutoRenew, pax8Metadata, true]
          );
          synced++;
        }
 
      } catch (productErr) {
        console.error(`[Pax8 Sync] Error processing product ${product.id}:`, productErr.message);
      }
    }
 
    const stats = { total: products.length, synced, updated, skipped };
    console.log(`[Pax8 Sync] Completed sync for tenant ${tenantId}:`, stats);
    
    return stats;
 
  } catch (err) {
    console.error(`[Pax8 Sync] Error syncing tenant ${tenantId}:`, err.message);
    throw err;
  }
}
 
/**
 * Sync all active tenants with Pax8 configuration
 */
async function syncAllTenants() {
  try {
    const result = await pool.query(
      `SELECT tenant_id FROM pax8_config WHERE is_active = true`
    );
 
    if (result.rows.length === 0) {
      console.log('[Pax8 Sync] No active Pax8 configurations found');
      return;
    }
 
    console.log(`[Pax8 Sync] Starting sync for ${result.rows.length} tenant(s)`);
 
    for (const row of result.rows) {
      try {
        await syncTenantProducts(row.tenant_id, 'Microsoft', false);
      } catch (error) {
        console.error(`[Pax8 Sync] Failed to sync tenant ${row.tenant_id}:`, error.message);
        // Continue with next tenant
      }
    }
 
    console.log('[Pax8 Sync] All tenants synced successfully');
  } catch (error) {
    console.error('[Pax8 Sync] Error in sync cycle:', error);
  }
}
 
/**
 * Calculate milliseconds until next scheduled sync time (3am)
 */
function msUntilNextSync() {
  const now = new Date();
  const nextSync = new Date();
  nextSync.setHours(SYNC_HOUR, 0, 0, 0);
  
  // If 3am has already passed today, schedule for tomorrow
  if (now.getHours() >= SYNC_HOUR) {
    nextSync.setDate(nextSync.getDate() + 1);
  }
  
  return nextSync.getTime() - now.getTime();
}
 
/**
 * Start the sync worker
 */
function startSyncWorker() {
  console.log(`[Pax8 Sync] Worker started - syncing daily at ${SYNC_HOUR}am`);
  
  // Run immediately on startup
  syncAllTenants().catch(err => {
    console.error('[Pax8 Sync] Initial sync failed:', err);
  });
  
  // Schedule next sync at 3am
  const msUntilSync = msUntilNextSync();
  console.log(`[Pax8 Sync] Next sync in ${Math.round(msUntilSync / 1000 / 60 / 60)} hours`);
  
  setTimeout(() => {
    syncAllTenants();
    // Then repeat every 24 hours
    setInterval(syncAllTenants, SYNC_INTERVAL);
  }, msUntilSync);
}
 
// Subscribe to manual sync requests via Redis
redis.subscribe('pax8:sync:request', (err) => {
  if (err) {
    console.error('[Pax8 Sync] Failed to subscribe:', err);
  } else {
    console.log('[Pax8 Sync] Subscribed to pax8:sync:request for manual triggers');
  }
});
 
redis.on('message', async (channel, message) => {
  if (channel !== 'pax8:sync:request') return;
 
  let jobData;
  try {
    jobData = JSON.parse(message);
  } catch (err) {
    console.error('[Pax8 Sync] Invalid job data:', message);
    return;
  }
 
  const { tenantId, vendor = 'Microsoft', force = false, jobId } = jobData;
 
  console.log(`[Pax8 Sync] Manual sync triggered - Job ${jobId} for tenant ${tenantId}`);
 
  try {
    const stats = await syncTenantProducts(tenantId, vendor, force);
 
    // Publish success result
    await redis.publish('pax8:sync:result', JSON.stringify({
      jobId,
      tenantId,
      success: true,
      stats,
      message: `Synced ${stats.synced + stats.updated} products (${stats.synced} new, ${stats.updated} updated, ${stats.skipped} skipped)`
    }));
 
  } catch (err) {
    console.error(`[Pax8 Sync] Job ${jobId} failed:`, err.message);
    
    // Publish error result
    await redis.publish('pax8:sync:result', JSON.stringify({
      jobId,
      tenantId,
      success: false,
      error: err.message,
      message: 'Product sync failed'
    }));
  }
});
 
// Handle graceful shutdown
process.on('SIGTERM', () => {
  console.log('[Pax8 Sync] SIGTERM received, shutting down gracefully...');
  redis.disconnect();
  process.exit(0);
});
 
process.on('SIGINT', () => {
  console.log('[Pax8 Sync] SIGINT received, shutting down gracefully...');
  redis.disconnect();
  process.exit(0);
});
 
// Start the worker if this file is run directly
if (require.main === module) {
  startSyncWorker();
}
 
module.exports = { startSyncWorker, syncAllTenants, syncTenantProducts };