import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Enable publisher confirms on this channel
# After this, every publish gets a confirmation (ack or nack) from the broker
channel.confirm_delivery()

channel.queue_declare(queue='orders', durable=True)

def publish_with_confirmation(order):
    """Publish a message and handle confirmation or failure."""
    try:
        channel.basic_publish(
            exchange='',
            routing_key='orders',
            body=json.dumps(order),
            properties=pika.BasicProperties(
                delivery_mode=2,  # Persistent -- broker writes to disk before confirming
            ),
            mandatory=True,  # Return the message if it cannot be routed to any queue
        )
        # If we get here without an exception, the broker confirmed receipt
        print(f"Confirmed: order {order['id']}")
    except pika.exceptions.UnroutableError:
        # Message could not be routed to any queue (no matching binding)
        print(f"UNROUTABLE: order {order['id']} -- check exchange/queue bindings")
    except pika.exceptions.NackError:
        # Broker explicitly rejected the message (e.g., disk full, queue limit reached)
        print(f"REJECTED: order {order['id']} -- broker nacked, retry or alert")

# Publish orders with confirmation
for i in range(5):
    publish_with_confirmation({"id": f"ORD-{i}", "amount": 99.99})

connection.close()

# Step 1: Declare the queue as durable (queue metadata survives restart)
channel.queue_declare(queue='payments', durable=True)

# Step 2: Publish messages as persistent (message body written to disk)
channel.basic_publish(
    exchange='',
    routing_key='payments',
    body=json.dumps({"payment_id": "pay-123", "amount": 250.00}),
    properties=pika.BasicProperties(
        delivery_mode=2,  # 1 = transient (in-memory only), 2 = persistent (written to disk)
    )
)

# auto_ack=True -- the broker deletes the message the instant it is delivered
# If your consumer crashes AFTER receiving but BEFORE processing, the message is gone
channel.basic_consume(
    queue='payments',
    on_message_callback=callback,
    auto_ack=True  # DO NOT use this for anything you cannot afford to lose
)

def process_payment(ch, method, properties, body):
    payment = json.loads(body)
    try:
        # Process the payment -- this is where the real work happens
        charge_customer(payment)
        update_database(payment)

        # Only acknowledge AFTER all processing is complete
        # If the worker crashes before this line, the message goes back to the queue
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(f"Processed and acked: {payment['payment_id']}")

    except Exception as e:
        print(f"Failed: {e}")
        # Reject the message -- requeue=False sends it to the dead letter exchange
        # requeue=True puts it back in the queue (careful: can cause infinite retry loops)
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

channel.basic_qos(prefetch_count=10)  # Allow up to 10 unacked messages per consumer
channel.basic_consume(queue='payments', on_message_callback=process_payment)

# Without prefetch: RabbitMQ sends messages as fast as possible
# Problem 1: If consumer is slow, thousands of messages pile up in its memory
# Problem 2: Round-robin distribution means slow consumers get the same share as fast ones

# With prefetch: RabbitMQ only sends the next message when the consumer has capacity
channel.basic_qos(prefetch_count=1)   # Conservative: one at a time, perfect fairness
channel.basic_qos(prefetch_count=10)  # Balanced: allows batching for throughput
channel.basic_qos(prefetch_count=100) # Aggressive: high throughput, less fairness

# Quorum queues are declared by setting the x-queue-type argument
channel.queue_declare(
    queue='critical-orders',
    durable=True,  # Quorum queues are always durable (this is enforced)
    arguments={
        'x-queue-type': 'quorum',           # Use Raft-based replication
        'x-quorum-initial-group-size': 3,    # Replicate across 3 nodes (must be odd for majority)
    }
)

Write path (Raft consensus):
  Producer ──▶ Leader node
                  │
                  ├──▶ Follower 1 (replicate)
                  ├──▶ Follower 2 (replicate)
                  │
                  │  Wait for MAJORITY to confirm (2 out of 3)
                  │
                  └──▶ Confirm to producer

If leader fails:
  Followers detect failure ──▶ Election ──▶ New leader chosen
  Queue continues serving with no data loss (majority had the data)

# On node 2, join node 1's cluster
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# Verify cluster health
rabbitmqctl cluster_status

# Set the partition handling strategy (do this before partitions happen)
# pause_minority: the minority partition stops accepting connections (safe choice)
# In rabbitmq.conf:
# cluster_partition_handling = pause_minority

# Client-side connection with multiple hosts (failover)
import pika

# Try each host in order until one succeeds
parameters = [
    pika.ConnectionParameters(host='rabbit-node1'),
    pika.ConnectionParameters(host='rabbit-node2'),
    pika.ConnectionParameters(host='rabbit-node3'),
]

# Pika tries each parameter set in order
connection = pika.BlockingConnection(parameters)

# Set up a retry mechanism using DLX and message TTL
# Messages that fail go to a "wait" queue with a TTL, then route back to the main queue

# Main queue: failed messages route to the retry exchange
channel.queue_declare(
    queue='orders',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'retry-exchange',
        'x-dead-letter-routing-key': 'orders-retry',
    }
)

# Retry queue: messages wait here for 30 seconds, then route back to main queue
channel.exchange_declare(exchange='retry-exchange', exchange_type='direct')
channel.queue_declare(
    queue='orders-retry',
    durable=True,
    arguments={
        'x-dead-letter-exchange': '',          # Default exchange routes to named queue
        'x-dead-letter-routing-key': 'orders', # Route back to the main 'orders' queue
        'x-message-ttl': 30000,                # Wait 30 seconds before retrying
    }
)
channel.queue_bind(exchange='retry-exchange', queue='orders-retry', routing_key='orders-retry')

# In the consumer: reject failures to trigger the retry cycle
def on_message(ch, method, properties, body):
    retry_count = (properties.headers or {}).get('x-retry-count', 0)

    if retry_count >= 3:
        # Max retries exceeded -- send to dead letter queue for manual investigation
        print(f"Max retries exceeded, sending to DLQ")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        return

    try:
        process_order(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"Retry {retry_count + 1}/3: {e}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

# Use pika's built-in connection recovery for automatic reconnection
import pika

parameters = pika.ConnectionParameters(
    host='localhost',
    connection_attempts=5,        # Try 5 times before giving up
    retry_delay=2,                # Wait 2 seconds between attempts
    heartbeat=30,                 # Detect dead connections within 30 seconds
    blocked_connection_timeout=60, # Timeout if broker is blocking due to resource alarm
)

# For production, wrap your consumer in a reconnection loop
while True:
    try:
        connection = pika.BlockingConnection(parameters)
        channel = connection.channel()
        channel.basic_consume(queue='orders', on_message_callback=process_order)
        channel.start_consuming()
    except pika.exceptions.AMQPConnectionError as e:
        print(f"Connection lost: {e}, reconnecting in 5 seconds...")
        time.sleep(5)
    except KeyboardInterrupt:
        break