neiam /dms

use lapin::{
    options::{BasicPublishOptions, ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions},
    types::{AMQPValue, FieldTable},
    BasicProperties, Channel, Connection, ExchangeKind,
};

use super::{dlq_name, messages::JobMessage, queue_name, routing_key, DLX_EXCHANGE, EXCHANGE};

#[derive(Clone)]
pub struct Publisher {
    channel: Channel,
}

impl Publisher {
    pub async fn new(conn: &Connection) -> Result<Self, lapin::Error> {
        let channel = conn.create_channel().await?;
        declare_topology(&channel).await?;
        Ok(Self { channel })
    }

    pub async fn publish(&self, msg: &JobMessage) -> Result<(), PublishError> {
        let payload = serde_json::to_vec(msg)?;
        let props = BasicProperties::default()
            .with_content_type("application/json".into())
            .with_delivery_mode(2) // persistent
            .with_message_id(msg.job_id.clone().into());

        self.channel
            .basic_publish(
                EXCHANGE.into(),
                routing_key(msg.kind).into(),
                BasicPublishOptions::default(),
                &payload,
                props,
            )
            .await?
            .await?;
        Ok(())
    }
}

#[derive(Debug, thiserror::Error)]
pub enum PublishError {
    #[error("amqp: {0}")]
    Amqp(#[from] lapin::Error),
    #[error("serialize: {0}")]
    Serde(#[from] serde_json::Error),
}

pub async fn publish_job(conn: &Connection, msg: &JobMessage) -> Result<(), PublishError> {
    let publisher = Publisher::new(conn).await?;
    publisher.publish(msg).await
}

/// Declares the main jobs exchange + per-kind queues bound to it, plus the
/// dead-letter exchange and per-kind DLQs. Idempotent; safe to call from any
/// process at startup.
pub async fn declare_topology(channel: &Channel) -> Result<(), lapin::Error> {
    let durable = ExchangeDeclareOptions {
        durable: true,
        ..Default::default()
    };
    channel
        .exchange_declare(EXCHANGE.into(), ExchangeKind::Topic, durable, FieldTable::default())
        .await?;
    channel
        .exchange_declare(DLX_EXCHANGE.into(), ExchangeKind::Topic, durable, FieldTable::default())
        .await?;

    for kind in crate::queue::ALL_KINDS {
        let qname = queue_name(kind);
        let dlq = dlq_name(kind);
        let rk = routing_key(kind);

        // Main queue with DLX wired up
        let mut args = FieldTable::default();
        args.insert("x-dead-letter-exchange".into(), AMQPValue::LongString(DLX_EXCHANGE.into()));
        args.insert("x-dead-letter-routing-key".into(), AMQPValue::LongString(rk.into()));
        channel
            .queue_declare(
                qname.as_str().into(),
                QueueDeclareOptions { durable: true, ..Default::default() },
                args,
            )
            .await?;
        channel
            .queue_bind(qname.as_str().into(), EXCHANGE.into(), rk.into(), QueueBindOptions::default(), FieldTable::default())
            .await?;

        // Dead-letter queue
        channel
            .queue_declare(
                dlq.as_str().into(),
                QueueDeclareOptions { durable: true, ..Default::default() },
                FieldTable::default(),
            )
            .await?;
        channel
            .queue_bind(dlq.as_str().into(), DLX_EXCHANGE.into(), rk.into(), QueueBindOptions::default(), FieldTable::default())
            .await?;
    }
    Ok(())
}