// Producer: creates work items
class OrderProducer {
  constructor(queue) {
    this.queue = queue;
  }

  async submitOrder(order) {
    // Validate and save order first (synchronous, user-facing)
    const savedOrder = await db.orders.create(order);

    // Then publish for async processing
    await this.queue.publish('orders.new', {
      orderId: savedOrder.id,
      items: savedOrder.items,
      userId: savedOrder.userId,
      timestamp: Date.now()
    });

    return savedOrder; // User gets immediate response
  }
}

// Consumer: processes work items
class OrderConsumer {
  async start() {
    await queue.subscribe('orders.new', async (message) => {
      try {
        const { orderId, items } = message.data;
        await this.reserveInventory(orderId, items);
        await this.calculateShipping(orderId);
        await this.sendConfirmationEmail(orderId);
        message.ack(); // Tell queue: processing succeeded
      } catch (err) {
        console.error(`Failed to process order ${message.data.orderId}:`, err);
        message.nack(); // Tell queue: processing failed, redeliver
      }
    });
  }
}

// Three competing consumers on the same queue
// Each message is delivered to exactly ONE consumer

// Consumer 1
queue.subscribe('emails.send', async (msg) => {
  await sendEmail(msg.data);
  msg.ack();
});

// Consumer 2 (same queue, same logic, different process)
queue.subscribe('emails.send', async (msg) => {
  await sendEmail(msg.data);
  msg.ack();
});

// Consumer 3
queue.subscribe('emails.send', async (msg) => {
  await sendEmail(msg.data);
  msg.ack();
});

// Queue distributes messages round-robin:
// Message 1 -> Consumer 1
// Message 2 -> Consumer 2
// Message 3 -> Consumer 3
// Message 4 -> Consumer 1
// ...

// Fan-out: every subscriber gets every message
// RabbitMQ: use a fanout exchange
// Kafka: use different consumer groups
// SQS: use SNS topic -> multiple SQS queues

// Publisher
await exchange.publish('order.placed', orderData);

// Subscriber 1: Inventory service (gets ALL order.placed messages)
exchange.subscribe('order.placed', 'inventory-queue', async (msg) => {
  await reserveStock(msg.data.items);
  msg.ack();
});

// Subscriber 2: Analytics service (also gets ALL order.placed messages)
exchange.subscribe('order.placed', 'analytics-queue', async (msg) => {
  await trackOrderMetrics(msg.data);
  msg.ack();
});

// Subscriber 3: Notification service (also gets ALL messages)
exchange.subscribe('order.placed', 'notification-queue', async (msg) => {
  await sendOrderNotification(msg.data.userId);
  msg.ack();
});

// Kafka consumer groups: best of both worlds

// Group "inventory-service" (3 instances for load balancing)
// Each order goes to exactly ONE instance in this group
const inventoryConsumer = new KafkaConsumer({
  groupId: 'inventory-service',
  topics: ['orders']
});

// Group "analytics-service" (2 instances)
// Each order ALSO goes to exactly ONE instance in this group
const analyticsConsumer = new KafkaConsumer({
  groupId: 'analytics-service',
  topics: ['orders']
});

// Result:
// Order #1 -> inventory-service (instance 2) AND analytics-service (instance 1)
// Order #2 -> inventory-service (instance 1) AND analytics-service (instance 2)
// Order #3 -> inventory-service (instance 3) AND analytics-service (instance 1)

// Manual acknowledgment (most control)
queue.subscribe('tasks', async (message) => {
  try {
    const result = await processTask(message.data);
    await saveResult(result);
    message.ack(); // "I'm done, remove this message from the queue"
  } catch (err) {
    if (isRetryable(err)) {
      message.nack(); // "I failed, put it back for another attempt"
    } else {
      message.reject(); // "I failed and retrying won't help, send to DLQ"
    }
  }
});

// SQS-style visibility timeout
const consumer = new SQSConsumer({
  queueUrl: 'https://sqs.us-east-1.amazonaws.com/123/orders',
  visibilityTimeout: 30, // seconds
  handleMessage: async (message) => {
    // You have 30 seconds to process and ack

    // For long-running tasks, extend the timeout
    if (willTakeLong(message)) {
      await sqs.changeMessageVisibility({
        ReceiptHandle: message.ReceiptHandle,
        VisibilityTimeout: 120 // extend to 2 minutes
      });
    }

    await processOrder(message.Body);
    // Auto-ack on success (SQS consumer library handles this)
  }
});

// Monitoring and scaling decisions
async function evaluateScaling(queueMetrics) {
  const {
    messageRate,       // Messages published per second
    processingRate,    // Messages consumed per second
    queueDepth,        // Current messages waiting
    avgProcessingTime, // Seconds per message
    consumerCount      // Current active consumers
  } = queueMetrics;

  // Queue growing? Need more consumers
  if (messageRate > processingRate) {
    const deficit = messageRate - processingRate;
    const consumersNeeded = Math.ceil(deficit * avgProcessingTime);
    console.log(`Need ${consumersNeeded} more consumers to keep up`);
  }

  // Queue too deep? Might need to scale producers down or consumers up
  if (queueDepth > 10000) {
    console.warn(`Queue depth at ${queueDepth}, consider scaling consumers`);
  }

  // Over-provisioned? Scale down to save costs
  if (queueDepth === 0 && processingRate > messageRate * 2) {
    console.log('Consumers idle, consider scaling down');
  }
}