Skip to content
Ordermatic Docs

Integration Example: Adding Silent Failure Prevention to ERP Sync

This example shows how to integrate the new silent failure prevention metrics and enhanced logging into your existing ERP sync tasks.

1. Updated ERP Sync Task with Metrics Integration

Section titled “1. Updated ERP Sync Task with Metrics Integration”

Here’s how to modify your src/trigger/erp-sync.ts to include the new observability features:

import { schedules } from '@trigger.dev/sdk/v3';
import { z } from 'zod';
import { schemaTask } from '@trigger.dev/sdk/v3';
import { logger, triggerLogger } from './logger'; // Updated import
import { db } from '@/db';
import { erpConnections } from '@/db/schema/erp-connections';
import { eq } from 'drizzle-orm';
import { ERPSyncFactory } from '@/services/erp-sync/erp-sync-factory';
import {
  validateSyncHealth,
  detectApiSuccessDbFailure,
  type EnhancedSyncResult,
} from '@/services/erp-sync/sync-validation';


// Import the new metrics functions
import {
  recordSyncResult,
  recordSyncHealthFailure,
  recordDatabaseConstraintViolation,
} from '@/lib/observability/node-metrics';


export const syncERPEntity = schemaTask({
  id: 'sync-erp-entity',
  schema: syncTaskSchema,
  retry: {
    /* ... existing retry config ... */
  },
  queue: {
    /* ... existing queue config ... */
  },
  machine: {
    /* ... existing machine config ... */
  },


  run: async ({ connectionId, entityType, options }) => {
    const taskStartTime = Date.now();


    // Enhanced task start logging
    triggerLogger.taskStarted('sync-erp-entity', { connectionId, entityType, options });
    triggerLogger.syncStarted(entityType, connectionId, options);


    let apiCallSuccessful = false;
    let dbOperationSuccessful = false;
    let rawSyncResult: any;
    let enhancedResult: EnhancedSyncResult | null = null;


    try {
      // Get the ERP connection from the database
      const connection = await db.query.erpConnections.findFirst({
        where: eq(erpConnections.id, connectionId),
      });


      if (!connection) {
        throw new Error(`ERP connection ${connectionId} not found`);
      }


      // Create sync service
      const erpSyncFactory = new ERPSyncFactory(db);
      const syncService = erpSyncFactory.createSyncService(connection);


      // Execute sync with enhanced error handling
      try {
        switch (entityType) {
          case 'products':
            rawSyncResult = await syncService.syncProducts(options);
            break;
          case 'customers':
            rawSyncResult = await syncService.syncCustomers(options);
            break;
          case 'inventory':
            rawSyncResult = await syncService.syncInventory(options);
            break;
        }


        apiCallSuccessful = true;
        dbOperationSuccessful = rawSyncResult.success;


        // Check for API success but database failure
        if (apiCallSuccessful && !dbOperationSuccessful) {
          recordSyncHealthFailure(entityType, connectionId, 'api_success_db_failure', {
            error_type: 'database_operation_failed',
            records_attempted: rawSyncResult.recordsProcessed?.total || 0,
          });


          triggerLogger.silentFailureDetected(entityType, connectionId, {
            type: 'api_success_db_failure',
            recordsAttempted: rawSyncResult.recordsProcessed?.total || 0,
          });
        }


        // Perform enhanced business health validation
        enhancedResult = await validateSyncHealth(rawSyncResult, entityType, connectionId);


        // Record comprehensive metrics
        const recordsProcessed = rawSyncResult.recordsProcessed?.total || 0;
        const healthScore = enhancedResult.businessValidation.healthScore || 0;
        const qualityScore = enhancedResult.qualityMetrics?.successRate * 100 || 0;
        const varianceFromBaseline = enhancedResult.businessValidation.variancePercentage || 0;


        recordSyncResult(
          entityType,
          connectionId,
          recordsProcessed,
          healthScore,
          qualityScore,
          varianceFromBaseline
        );


        // Check for silent failures
        if (recordsProcessed === 0) {
          recordSyncHealthFailure(entityType, connectionId, 'zero_records', {
            sync_type: options?.fullSync ? 'full' : 'incremental',
            expected_minimum: enhancedResult.businessValidation.expectedMinimum?.toString() || '0',
          });


          triggerLogger.silentFailureDetected(entityType, connectionId, {
            type: 'zero_records',
            syncType: options?.fullSync ? 'full' : 'incremental',
            expectedMinimum: enhancedResult.businessValidation.expectedMinimum,
          });
        }


        if (!enhancedResult.businessValidation.isHealthy) {
          recordSyncHealthFailure(entityType, connectionId, 'business_validation', {
            health_score: healthScore.toString(),
            failure_reasons: enhancedResult.businessValidation.healthCheckReasons?.join(',') || '',
          });
        }


        // Enhanced completion logging
        const duration = Date.now() - taskStartTime;
        triggerLogger.syncCompleted(entityType, connectionId, {
          ...rawSyncResult,
          healthScore,
          qualityScore,
          varianceFromBaseline,
        });


        triggerLogger.taskCompleted('sync-erp-entity', enhancedResult, duration);


        triggerLogger.businessHealthCheck(entityType, connectionId, healthScore, {
          recordsProcessed,
          qualityScore,
          varianceFromBaseline,
          isHealthy: enhancedResult.businessValidation.isHealthy,
        });


        return enhancedResult;
      } catch (syncError) {
        dbOperationSuccessful = false;


        // Handle database constraint violations
        if (syncError instanceof Error && syncError.message.includes('constraint')) {
          const constraintMatch = syncError.message.match(/constraint "([^"]+)"/);
          const constraintName = constraintMatch ? constraintMatch[1] : 'unknown';


          recordDatabaseConstraintViolation(constraintName, entityType, 'sync');
        }


        // Record API success but DB failure if applicable
        if (apiCallSuccessful && !dbOperationSuccessful) {
          recordSyncHealthFailure(entityType, connectionId, 'api_success_db_failure', {
            error_type: 'sync_operation_failed',
            error_message: syncError.message.substring(0, 100),
          });
        }


        throw syncError;
      }
    } catch (error) {
      const duration = Date.now() - taskStartTime;


      // Enhanced error logging with metrics context
      triggerLogger.taskFailed('sync-erp-entity', error as Error, {
        connectionId,
        entityType,
        options,
        apiCallSuccessful,
        dbOperationSuccessful,
        duration,
      });


      // Log comprehensive error context
      logger.error(
        `Error syncing ${entityType} for connection ${connectionId}`,
        error instanceof Error ? error : new Error('Unknown error'),
        {
          connectionId,
          entityType,
          options,
          errorDetails: error instanceof Error ? error.message : String(error),
          apiCallSuccessful,
          dbOperationSuccessful,
          possibleSilentFailure: apiCallSuccessful && !dbOperationSuccessful,
          duration,
          // Add metrics context for debugging
          metricsRecorded: {
            syncResult: !!enhancedResult,
            healthScore: enhancedResult?.businessValidation.healthScore,
            recordsProcessed: enhancedResult?.recordsProcessed?.total,
          },
        }
      );


      throw error;
    }
  },
});

2. Enhanced Sync Validation Service

Section titled “2. Enhanced Sync Validation Service”

Create or update src/services/erp-sync/sync-validation.ts:

import { recordSyncResult, recordSyncHealthFailure } from '@/lib/observability/node-metrics';


export async function validateSyncHealth(
  result: SyncResult,
  entityType: string,
  connectionId: string
): Promise<EnhancedSyncResult> {
  // Get historical baseline (implement based on your data)
  const baseline = await getHistoricalBaseline(entityType, connectionId, 30);


  // Calculate health metrics
  const recordsProcessed = result.recordsProcessed?.total || 0;
  const successRate =
    recordsProcessed > 0 ? (result.recordsProcessed?.success || 0) / recordsProcessed : 0;


  const minimumThreshold = Math.max(
    baseline.average * 0.1, // At least 10% of historical average
    entityType === 'products' ? 5 : 1 // Absolute minimums
  );


  const variancePercentage =
    baseline.average > 0 ? ((recordsProcessed - baseline.average) / baseline.average) * 100 : 0;


  const healthScore = calculateHealthScore(recordsProcessed, successRate, baseline);


  const isHealthy = recordsProcessed >= minimumThreshold && healthScore >= 80;


  const businessValidation = {
    isHealthy,
    expectedMinimum: minimumThreshold,
    actualProcessed: recordsProcessed,
    healthScore,
    variancePercentage,
    healthCheckReasons: !isHealthy
      ? [
          recordsProcessed === 0 ? 'zero_records_processed' : null,
          recordsProcessed < minimumThreshold ? 'below_minimum_threshold' : null,
          healthScore < 80 ? 'poor_health_score' : null,
        ].filter(Boolean)
      : [],
    historicalAverage: baseline.average,
  };


  const qualityMetrics = {
    successRate,
    dataQualityScore: calculateDataQuality(result),
    unexpectedNulls: result.recordsProcessed?.failed || 0,
  };


  return {
    ...result,
    businessValidation,
    qualityMetrics,
  };
}


function calculateHealthScore(
  recordsProcessed: number,
  successRate: number,
  baseline: any
): number {
  const factors = {
    recordCount: Math.min(1, recordsProcessed / (baseline.average || 1)),
    successRate: successRate,
    errorRate: 1 - (1 - successRate), // Inverted error rate
  };


  return Math.min(
    100,
    Object.values(factors).reduce((sum, factor) => sum + factor * 33.33, 0)
  );
}

3. Usage in Existing Tasks

Section titled “3. Usage in Existing Tasks”

Update other Trigger tasks to use the enhanced logging:

// In src/trigger/customer-pricing.ts
import { triggerLogger } from './logger';
import { recordSyncResult, recordSyncHealthFailure } from '@/lib/observability/node-metrics';


export const syncCustomerPricingTask = schemaTask({
  id: "sync-customer-pricing",
  schema: /* ... existing schema ... */,
  run: async ({ connectionId, fullSync, batchSize }) => {
    triggerLogger.taskStarted('sync-customer-pricing', { connectionId, fullSync, batchSize });


    try {
      // ... existing logic ...


      const result = await syncService.syncCustomers({ fullSync, batchSize });


      // Record metrics
      recordSyncResult('customers', connectionId, result.recordsProcessed.total, 100, 95, 0);


      triggerLogger.taskCompleted('sync-customer-pricing', result);


      return result;
    } catch (error) {
      triggerLogger.taskFailed('sync-customer-pricing', error as Error, { connectionId });
      throw error;
    }
  },
});

4. Environment Variables for Production

Section titled “4. Environment Variables for Production”

Add to your .env.production:

Terminal window
# Enhanced observability
APP_VERSION=1.0.0
ENABLE_SIGNOZ_LOGS=true
DEBUG_LOGGING=false


# Service identification for Trigger tasks
TRIGGER_SERVICE_NAME=erp-unlocked-trigger-tasks


# SigNoz configuration (if not already set)
OTEL_EXPORTER_OTLP_ENDPOINT=https://ingest.us.signoz.cloud:443
OTEL_EXPORTER_OTLP_HEADERS=signoz-ingestion-key=your-key
OTEL_EXPORTER_OTLP_LOGS_ENDPOINT=https://ingest.us.signoz.cloud:443/v1/logs

5. Testing the Integration

Section titled “5. Testing the Integration”

Test the enhanced observability:

// Trigger a test sync
await syncERPEntity.trigger({
  connectionId: 'your-connection-id',
  entityType: 'products',
  options: { fullSync: false, batchSize: 10, maxItems: 5 },
});

Then check SigNoz for:

  1. Logs under service erp-unlocked-trigger-tasks
  2. Metrics showing up in your dashboards
  3. Alerts triggering for silent failures

This integration provides complete observability while maintaining compatibility with your existing infrastructure.