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();
    this.messageStore = new Map();
    this.inFlight = 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;
    }
    
    this.messageStore.set(messageId, messageData);
    // 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;
    }
    
    const messageData = this.queue.shift();
    if (messageData && messageData.messageId) {
      this.inFlight.set(messageData.messageId, messageData);
    }
    return messageData;
  }
  
  async markProcessed(messageId) {
    this.processed.add(messageId);
    this.retryCounts.delete(messageId);
    this.inFlight.delete(messageId);
    this.messageStore.delete(messageId);
    this.logger.info(`Message ${messageId} marked as processed`);
  }
  
  async markFailed(messageId) {
    this.failed.add(messageId);
    this.retryCounts.delete(messageId);
    this.inFlight.delete(messageId);
    this.logger.error(`Message ${messageId} marked as failed`);
  }
  
  async getRetryCount(messageId) {
    return this.retryCounts.get(messageId) || 0;
  }

  async resetRetryCount(messageId) {
    this.retryCounts.delete(messageId);
  }
  
  async retry(messageId, options = {}) {
    const { increment = true } = options;
    const count = this.retryCounts.get(messageId) || 0;
    if (increment) {
      this.retryCounts.set(messageId, count + 1);
    }

    const existingIndex = this.queue.findIndex(m => m.messageId === messageId);
    if (existingIndex >= 0) {
      const nextCount = increment ? count + 1 : count;
      this.logger.info(`Message ${messageId} retry count: ${nextCount}`);
      return;
    }

    const messageData = this.inFlight.get(messageId) || this.messageStore.get(messageId);
    if (!messageData) {
      this.logger.warn(`Cannot requeue ${messageId}; original message payload is unavailable`);
      return;
    }

    this.inFlight.delete(messageId);
    this.queue.push(messageData);
    const nextCount = increment ? count + 1 : count;
    this.logger.info(`Message ${messageId} requeued. Retry count: ${nextCount}. Queue size: ${this.queue.length}`);
  }
  
  getStats() {
    return {
      queueSize: this.queue.length,
      processed: this.processed.size,
      failed: this.failed.size
    };
  }
}