smom-dbis-138/services/relay/src/MessageQueue.js

/**
 * Message Queue for managing pending relay messages
 */

export class MessageQueue {
  constructor(logger) {
    this.logger = logger;
    this.queue = [];
    this.processed = new Set();
    this.failed = new Set();
    this.retryCounts = new Map();
  }
  
  async add(messageData) {
    const messageId = messageData.messageId;
    
    // Skip if already processed or failed
    if (this.processed.has(messageId) || this.failed.has(messageId)) {
      this.logger.debug(`Message ${messageId} already processed or failed, skipping`);
      return;
    }
    
    // Check if already in queue
    const existingIndex = this.queue.findIndex(m => m.messageId === messageId);
    if (existingIndex >= 0) {
      this.logger.debug(`Message ${messageId} already in queue`);
      return;
    }
    
    // Add to queue
    this.queue.push(messageData);
    this.logger.info(`Added message ${messageId} to queue. Queue size: ${this.queue.length}`);
  }
  
  async getNext() {
    if (this.queue.length === 0) {
      return null;
    }
    
    return this.queue.shift();
  }
  
  async markProcessed(messageId) {
    this.processed.add(messageId);
    this.retryCounts.delete(messageId);
    this.logger.info(`Message ${messageId} marked as processed`);
  }
  
  async markFailed(messageId) {
    this.failed.add(messageId);
    this.logger.error(`Message ${messageId} marked as failed`);
  }
  
  async getRetryCount(messageId) {
    return this.retryCounts.get(messageId) || 0;
  }
  
  async retry(messageId) {
    const count = this.retryCounts.get(messageId) || 0;
    this.retryCounts.set(messageId, count + 1);
    
    // Find message in queue or re-add it
    // In a production system, you'd store the original message data
    this.logger.info(`Message ${messageId} retry count: ${count + 1}`);
  }
  
  getStats() {
    return {
      queueSize: this.queue.length,
      processed: this.processed.size,
      failed: this.failed.size
    };
  }
}