Best Practices

class MessageSender {
  constructor(apiKey, options = {}) {
    this.client = new Amanah Agent AI({ apiKey });
    this.maxRetries = options.maxRetries || 3;
    this.retryDelay = options.retryDelay || 1000;
    this.circuitBreaker = new CircuitBreaker();
  }
  
  async sendMessage(messageData, attempt = 1) {
    try {
      // Check circuit breaker state
      if (this.circuitBreaker.isOpen()) {
        throw new Error('Service temporarily unavailable');
      }
      
      // Validate message data
      this.validateMessage(messageData);
      
      // Send message with timeout
      const response = await Promise.race([
        this.client.messages.send(messageData),
        this.timeoutPromise(30000) // 30 second timeout
      ]);
      
      // Success - reset circuit breaker
      this.circuitBreaker.onSuccess();
      
      return {
        success: true,
        messageId: response.message_id,
        status: response.status,
        attempt: attempt
      };
      
    } catch (error) {
      console.error(\`Message send attempt \${attempt} failed:\`, error.message);
      
      // Handle different error types
      const errorType = this.classifyError(error);
      
      switch (errorType) {
        case 'RATE_LIMIT':
          return this.handleRateLimit(messageData, attempt, error);
        
        case 'NETWORK':
        case 'TIMEOUT':
        case 'SERVER_ERROR':
          return this.handleRetryableError(messageData, attempt, error);
        
        case 'VALIDATION':
        case 'AUTH':
        case 'FORBIDDEN':
          return this.handleNonRetryableError(messageData, error);
        
        default:
          return this.handleUnknownError(messageData, attempt, error);
      }
    }
  }
  
  validateMessage(messageData) {
    if (!messageData.to || !messageData.to.match(/^+[1-9]d{1,14}$/)) {
      throw new ValidationError('Invalid phone number format');
    }
    
    if (!messageData.message && !messageData.media_id && !messageData.template_id) {
      throw new ValidationError('Message content is required');
    }
    
    if (messageData.message && messageData.message.length > 4096) {
      throw new ValidationError('Message too long');
    }
  }
  
  classifyError(error) {
    if (error.message.includes('rate limit')) return 'RATE_LIMIT';
    if (error.message.includes('timeout')) return 'TIMEOUT';
    if (error.message.includes('network')) return 'NETWORK';
    if (error.status >= 500) return 'SERVER_ERROR';
    if (error.status === 401) return 'AUTH';
    if (error.status === 403) return 'FORBIDDEN';
    if (error.status === 400) return 'VALIDATION';
    return 'UNKNOWN';
  }
  
  async handleRateLimit(messageData, attempt, error) {
    const retryAfter = error.retryAfter || this.retryDelay * Math.pow(2, attempt);
    
    if (attempt < this.maxRetries) {
      console.log(\`Rate limited, retrying in \${retryAfter}ms\`);
      await this.delay(retryAfter);
      return this.sendMessage(messageData, attempt + 1);
    }
    
    return {
      success: false,
      error: 'Rate limit exceeded',
      errorType: 'RATE_LIMIT',
      retryable: false
    };
  }
  
  async handleRetryableError(messageData, attempt, error) {
    this.circuitBreaker.onFailure();
    
    if (attempt < this.maxRetries) {
      const delay = this.retryDelay * Math.pow(2, attempt - 1);
      console.log(\`Retrying in \${delay}ms (attempt \${attempt + 1}/\${this.maxRetries})\`);
      
      await this.delay(delay);
      return this.sendMessage(messageData, attempt + 1);
    }
    
    return {
      success: false,
      error: error.message,
      errorType: this.classifyError(error),
      retryable: false,
      finalAttempt: true
    };
  }
  
  async handleNonRetryableError(messageData, error) {
    return {
      success: false,
      error: error.message,
      errorType: this.classifyError(error),
      retryable: false
    };
  }
  
  delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
  
  timeoutPromise(ms) {
    return new Promise((_, reject) => 
      setTimeout(() => reject(new Error('Timeout')), ms)
    );
  }
}

// Circuit Breaker implementation
class CircuitBreaker {
  constructor(threshold = 5, timeout = 60000) {
    this.failureThreshold = threshold;
    this.timeout = timeout;
    this.failureCount = 0;
    this.lastFailureTime = null;
    this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
  }
  
  isOpen() {
    if (this.state === 'OPEN') {
      if (Date.now() - this.lastFailureTime > this.timeout) {
        this.state = 'HALF_OPEN';
        return false;
      }
      return true;
    }
    return false;
  }
  
  onSuccess() {
    this.failureCount = 0;
    this.state = 'CLOSED';
  }
  
  onFailure() {
    this.failureCount++;
    this.lastFailureTime = Date.now();
    
    if (this.failureCount >= this.failureThreshold) {
      this.state = 'OPEN';
    }
  }
}

// Error monitoring and alerting system
class ErrorMonitor {
  constructor(alertingService) {
    this.alerts = alertingService;
    this.errorCounts = new Map();
    this.thresholds = {
      error_rate: 0.05, // 5% error rate
      consecutive_failures: 10,
      response_time: 5000 // 5 seconds
    };
  }
  
  async logError(error, context) {
    const errorKey = \`\${error.type}_\${context.endpoint}\`;
    const count = this.errorCounts.get(errorKey) || 0;
    this.errorCounts.set(errorKey, count + 1);
    
    // Log structured error data
    console.error('API Error:', {
      timestamp: new Date().toISOString(),
      error_type: error.type,
      message: error.message,
      status_code: error.status,
      endpoint: context.endpoint,
      user_id: context.userId,
      request_id: context.requestId,
      stack: error.stack
    });
    
    // Check alert thresholds
    await this.checkAlertThresholds(errorKey, error, context);
  }
  
  async checkAlertThresholds(errorKey, error, context) {
    const errorCount = this.errorCounts.get(errorKey);
    
    // Alert on consecutive failures
    if (errorCount >= this.thresholds.consecutive_failures) {
      await this.alerts.send({
        severity: 'HIGH',
        title: 'Multiple API Failures Detected',
        description: \`\${errorCount} consecutive failures for \${errorKey}\`,
        context: context
      });
    }
    
    // Alert on high error rates
    const errorRate = await this.calculateErrorRate(context.endpoint);
    if (errorRate > this.thresholds.error_rate) {
      await this.alerts.send({
        severity: 'MEDIUM',
        title: 'High Error Rate Detected',
        description: \`Error rate: \${(errorRate * 100).toFixed(2)}% for \${context.endpoint}\`,
        metrics: { error_rate: errorRate }
      });
    }
  }
  
  async calculateErrorRate(endpoint) {
    // Implementation depends on your metrics storage
    // This is a simplified example
    const totalRequests = await this.getTotalRequests(endpoint, '1h');
    const errorRequests = await this.getErrorRequests(endpoint, '1h');
    
    return totalRequests > 0 ? errorRequests / totalRequests : 0;
  }
  
  // Reset error counts periodically
  resetCounts() {
    this.errorCounts.clear();
  }
}

// Usage with message sender
const errorMonitor = new ErrorMonitor(alertingService);
const messageSender = new MessageSender(apiKey);

try {
  const result = await messageSender.sendMessage(messageData);
} catch (error) {
  await errorMonitor.logError(error, {
    endpoint: '/messages',
    userId: messageData.userId,
    requestId: generateRequestId()
  });
  throw error; // Re-throw after logging
}

// Efficient bulk message processing
class BulkMessageProcessor {
  constructor(client, options = {}) {
    this.client = client;
    this.batchSize = options.batchSize || 100;
    this.concurrency = options.concurrency || 5;
    this.memoryThreshold = options.memoryThreshold || 100 * 1024 * 1024; // 100MB
  }
  
  async processBulkMessages(messages) {
    const results = [];
    
    // Process in chunks to manage memory
    for (let i = 0; i < messages.length; i += this.batchSize) {
      const batch = messages.slice(i, i + this.batchSize);
      
      // Check memory usage
      if (process.memoryUsage().heapUsed > this.memoryThreshold) {
        await this.gc(); // Force garbage collection if needed
        await this.delay(100); // Brief pause
      }
      
      // Process batch with controlled concurrency
      const batchResults = await this.processBatch(batch);
      results.push(...batchResults);
      
      // Progress logging
      const processed = Math.min(i + this.batchSize, messages.length);
      console.log(\`Processed \${processed}/\${messages.length} messages\`);
    }
    
    return results;
  }
  
  async processBatch(batch) {
    // Use semaphore to limit concurrency
    const semaphore = new Semaphore(this.concurrency);
    
    return Promise.all(batch.map(async (message) => {
      await semaphore.acquire();
      
      try {
        return await this.client.messages.send(message);
      } finally {
        semaphore.release();
      }
    }));
  }
  
  async gc() {
    if (global.gc) {
      global.gc();
    }
  }
  
  delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

// Semaphore for concurrency control
class Semaphore {
  constructor(permits) {
    this.permits = permits;
    this.waiting = [];
  }
  
  async acquire() {
    if (this.permits > 0) {
      this.permits--;
      return;
    }
    
    return new Promise(resolve => {
      this.waiting.push(resolve);
    });
  }
  
  release() {
    this.permits++;
    
    if (this.waiting.length > 0) {
      const resolve = this.waiting.shift();
      this.permits--;
      resolve();
    }
  }
}

// Distributed message processing system
const cluster = require('cluster');
const numCPUs = require('os').cpus().length;
const Redis = require('redis');
const Queue = require('bull');

if (cluster.isMaster) {
  console.log(\`Master \${process.pid} is running\`);
  
  // Fork workers
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }
  
  cluster.on('exit', (worker, code, signal) => {
    console.log(\`Worker \${worker.process.pid} died\`);
    cluster.fork(); // Restart worker
  });
  
} else {
  // Worker process
  const messageQueue = new Queue('message processing', {
    redis: {
      host: process.env.REDIS_HOST,
      port: process.env.REDIS_PORT
    }
  });
  
  // Process messages
  messageQueue.process('send-message', 10, async (job) => {
    const { messageData, userId, priority } = job.data;
    
    try {
      const client = new Amanah Agent AI({
        apiKey: process.env.AMANAHAGENT_API_KEY
      });
      
      const result = await client.messages.send(messageData);
      
      // Log success
      console.log(\`Worker \${process.pid} sent message \${result.message_id}\`);
      
      return result;
    } catch (error) {
      console.error(\`Worker \${process.pid} failed:\`, error);
      throw error;
    }
  });
  
  console.log(\`Worker \${process.pid} started\`);
}

// Queue manager for adding jobs
class MessageQueueManager {
  constructor(redisConfig) {
    this.queue = new Queue('message processing', {
      redis: redisConfig
    });
    
    // Configure queue settings
    this.queue.settings = {
      stalledInterval: 30 * 1000,
      retryDelayOnFailed: true
    };
  }
  
  async addMessage(messageData, options = {}) {
    const priority = options.priority || 'normal';
    const delay = options.delay || 0;
    const attempts = options.attempts || 3;
    
    return this.queue.add('send-message', {
      messageData,
      userId: options.userId,
      priority
    }, {
      priority: this.getPriorityScore(priority),
      delay,
      attempts,
      backoff: {
        type: 'exponential',
        delay: 2000
      }
    });
  }
  
  getPriorityScore(priority) {
    const scores = {
      high: 10,
      normal: 5,
      low: 1
    };
    return scores[priority] || 5;
  }
  
  // Bulk add with rate limiting
  async addBulkMessages(messages, batchSize = 100) {
    const results = [];
    
    for (let i = 0; i < messages.length; i += batchSize) {
      const batch = messages.slice(i, i + batchSize);
      const batchPromises = batch.map(msg => this.addMessage(msg));
      
      const batchResults = await Promise.all(batchPromises);
      results.push(...batchResults);
      
      // Brief pause between batches to avoid overwhelming the queue
      if (i + batchSize < messages.length) {
        await new Promise(resolve => setTimeout(resolve, 100));
      }
    }
    
    return results;
  }
}

// Analytics collection service
class AnalyticsCollector {
  constructor(config) {
    this.influxDB = new InfluxDB(config.influxDB);
    this.redis = new Redis(config.redis);
  }
  
  async recordMessageSent(messageData) {
    const metrics = {
      measurement: 'messages',
      tags: {
        type: messageData.type,
        user_id: messageData.userId,
        campaign_id: messageData.campaignId
      },
      fields: {
        sent: 1,
        cost: this.calculateCost(messageData.type),
        length: messageData.message?.length || 0
      },
      timestamp: Date.now()
    };
    
    await this.influxDB.writePoints([metrics]);
    
    // Update real-time counters
    await this.redis.incr('messages:sent:today');
    await this.redis.incr(\`messages:sent:user:\${messageData.userId}\`);
  }
  
  async recordDeliveryStatus(messageId, status, timestamp) {
    const metrics = {
      measurement: 'delivery_status',
      tags: {
        status: status,
        message_id: messageId
      },
      fields: {
        count: 1,
        delivery_time: status === 'delivered' ? 
          timestamp - this.getMessageSentTime(messageId) : 0
      },
      timestamp: Date.now()
    };
    
    await this.influxDB.writePoints([metrics]);
    
    // Update delivery rate cache
    const today = new Date().toISOString().split('T')[0];
    await this.redis.incr(\`delivery:\${status}:\${today}\`);
  }
  
  async getDeliveryRate(period = '24h') {
    const query = \`
      SELECT 
        SUM("count") as total_delivered
      FROM "delivery_status" 
      WHERE "status" = 'delivered' 
        AND time > now() - \${period}
    \`;
    
    const totalSentQuery = \`
      SELECT SUM("sent") as total_sent 
      FROM "messages" 
      WHERE time > now() - \${period}
    \`;
    
    const [delivered, sent] = await Promise.all([
      this.influxDB.query(query),
      this.influxDB.query(totalSentQuery)
    ]);
    
    const deliveryRate = (delivered[0]?.total_delivered || 0) / (sent[0]?.total_sent || 1);
    return Math.round(deliveryRate * 100);
  }
  
  async generateDashboardData(timeRange = '24h') {
    const [deliveryRate, avgResponseTime, errorRate, topErrors] = await Promise.all([
      this.getDeliveryRate(timeRange),
      this.getAverageResponseTime(timeRange),
      this.getErrorRate(timeRange),
      this.getTopErrors(timeRange)
    ]);
    
    return {
      deliveryRate,
      avgResponseTime,
      errorRate,
      topErrors,
      generatedAt: new Date().toISOString()
    };
  }
}