// r8r: rust port of n8n.io
// Initial author: Michael Cochran <mcochran@cochranblock.org>
// Built for: Tanner Woodrum, intended cofounder
// License: Unlicense (public domain dedication, see UNLICENSE)
//
// `rabbitmq` — publish a message to a RabbitMQ queue or exchange via
// lapin (pure-rust AMQP). Param names lifted from RabbitMQ.node.ts:
// operation, mode, queue, exchange, exchangeType, routingKey,
// sendInputData, message, options.
// Credential `rabbitmq`: { hostname, port, username, password, vhost, ssl }.

use async_trait::async_trait;
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use lapin::{Connection, ConnectionProperties, BasicProperties, options::*};
use serde_json::Value;
use std::sync::Arc;

pub struct RabbitMqNode;

static META: NodeMetadata = NodeMetadata {
    name: "rabbitmq",
    display_name: "RabbitMQ",
    version: &[1],
    group: &["transform"],
    description: "Send messages to RabbitMQ",
    inputs: 1,
    outputs: 1,
    properties: &[
        PropertyDef { name: "operation", display_name: "Operation", kind: PropertyKind::Options, required: true, default: Some("sendMessage") },
        PropertyDef { name: "mode", display_name: "Mode", kind: PropertyKind::Options, required: true, default: Some("queue") },
        PropertyDef { name: "queue", display_name: "Queue / Topic", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "exchange", display_name: "Exchange", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "exchangeType", display_name: "Type", kind: PropertyKind::Options, required: false, default: Some("direct") },
        PropertyDef { name: "routingKey", display_name: "Routing Key", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "sendInputData", display_name: "Send Input Data", kind: PropertyKind::Boolean, required: false, default: Some("true") },
        PropertyDef { name: "message", display_name: "Message", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "options", display_name: "Options", kind: PropertyKind::Json, required: false, default: None },
    ],
    credentials: &[CredentialRef { name: "rabbitmq", required: true }],
};

#[async_trait]
impl Node for RabbitMqNode {
    fn metadata(&self) -> &'static NodeMetadata { &META }

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let hostname = ctx.credential_field("hostname")?
            .as_str().ok_or_else(|| NodeError::MissingCredential("rabbitmq.hostname".into()))?
            .to_string();
        let port = ctx.credential_field("port").ok()
            .and_then(|v| v.as_u64()).unwrap_or(5672);
        let username = ctx.credential_field("username").ok()
            .and_then(|v| v.as_str().map(|s| s.to_string()))
            .unwrap_or_else(|| "guest".into());
        let password = ctx.credential_field("password").ok()
            .and_then(|v| v.as_str().map(|s| s.to_string()))
            .unwrap_or_else(|| "guest".into());
        let vhost = ctx.credential_field("vhost").ok()
            .and_then(|v| v.as_str().map(|s| s.to_string()))
            .unwrap_or_else(|| "/".into());
        let ssl = ctx.credential_field("ssl").ok()
            .and_then(|v| v.as_bool()).unwrap_or(false);
        let scheme = if ssl { "amqps" } else { "amqp" };
        let vhost_enc = if vhost == "/" { "%2f".to_string() } else { vhost.clone() };
        let uri = format!("{}://{}:{}@{}:{}/{}", scheme, username, password, hostname, port, vhost_enc);

        let mode = ctx.param_opt_str("mode").unwrap_or_else(|| "queue".into());
        let send_input = ctx.node_def.parameters.get("sendInputData")
            .and_then(|v| v.as_bool()).unwrap_or(true);
        let payload = if send_input {
            let item = ctx.input.first().cloned().unwrap_or_default();
            serde_json::to_vec(&item.json).unwrap_or_default()
        } else {
            ctx.param_opt_str("message").unwrap_or_default().into_bytes()
        };

        let conn = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            Connection::connect(&uri, ConnectionProperties::default()),
        ).await
            .map_err(|_| NodeError::Other("rabbitmq connect: timed out".into()))?
            .map_err(|e| NodeError::Other(format!("rabbitmq connect: {}", e)))?;
        let channel = conn.create_channel().await
            .map_err(|e| NodeError::Other(format!("rabbitmq channel: {}", e)))?;

        let (routing, exchange) = match mode.as_str() {
            "queue" => {
                let queue = ctx.param_str("queue")?;
                let opts: Value = ctx.node_def.parameters.get("options").cloned().unwrap_or(Value::Null);
                let durable = opts.get("durable").and_then(|v| v.as_bool()).unwrap_or(true);
                let exclusive = opts.get("exclusive").and_then(|v| v.as_bool()).unwrap_or(false);
                let auto_delete = opts.get("autoDelete").and_then(|v| v.as_bool()).unwrap_or(false);
                channel.queue_declare(
                    &queue,
                    QueueDeclareOptions { durable, exclusive, auto_delete, ..Default::default() },
                    Default::default(),
                ).await.map_err(|e| NodeError::Other(format!("rabbitmq declare: {}", e)))?;
                (queue, "".to_string())
            }
            "exchange" => {
                let exchange = ctx.param_str("exchange")?;
                let routing = ctx.param_opt_str("routingKey").unwrap_or_default();
                (routing, exchange)
            }
            other => return Err(NodeError::Other(format!("rabbitmq: unknown mode `{}`", other))),
        };

        channel.basic_publish(
            &exchange, &routing,
            BasicPublishOptions::default(),
            &payload,
            BasicProperties::default(),
        ).await
            .map_err(|e| NodeError::Other(format!("rabbitmq publish: {}", e)))?
            .await
            .map_err(|e| NodeError::Other(format!("rabbitmq confirm: {}", e)))?;

        let _ = conn.close(0, "ok").await;
        Ok(vec![vec![NodeItem {
            json: serde_json::json!({ "ok": true, "exchange": exchange, "routingKey": routing }),
            ..Default::default()
        }]])
    }
}

inventory::submit! { NodeFactory::new("rabbitmq", || Arc::new(RabbitMqNode)) }
