// 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)
//
// `webhook` — trigger entry node. Starts an axum HTTP server, blocks until
// the first request lands on the configured path, emits the body as an item,
// then shuts the server down. v0.1: one-shot. Long-lived listener is v0.2.

use async_trait::async_trait;
use axum::{body::Bytes, routing::any, Router};
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use serde_json::Value;
use std::sync::Arc;
use tokio::sync::{oneshot, Mutex};

pub struct WebhookNode;

static META: NodeMetadata = NodeMetadata {
    name: "webhook",
    display_name: "Webhook",
    version: &[1, 2],
    group: &["trigger"],
    description: "Starts an HTTP server and emits the first incoming request body as an item",
    inputs: 0,
    outputs: 1,
    properties: &[],
    credentials: &[],
};

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

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let port = ctx.node_def.parameters.get("port")
            .and_then(|v| v.as_u64())
            .unwrap_or(8080) as u16;
        let path = ctx.node_def.parameters.get("path")
            .and_then(|v| v.as_str())
            .unwrap_or("/webhook")
            .to_string();

        let (tx, rx) = oneshot::channel::<Value>();
        let shared = Arc::new(Mutex::new(Some(tx)));

        let handler_shared = shared.clone();
        let app = Router::new().route(
            &path,
            any(move |body: Bytes| {
                let shared = handler_shared.clone();
                async move {
                    let parsed: Value = serde_json::from_slice(&body)
                        .unwrap_or_else(|_| Value::String(String::from_utf8_lossy(&body).into_owned()));
                    if let Some(tx) = shared.lock().await.take() {
                        let _ = tx.send(parsed);
                    }
                    "ok"
                }
            }),
        );

        let listener = tokio::net::TcpListener::bind(("0.0.0.0", port)).await
            .map_err(|e| NodeError::Other(format!("webhook bind {}: {}", port, e)))?;

        tracing::info!(port, path = %path, "webhook listening for first request");

        let server = tokio::spawn(async move {
            let _ = axum::serve(listener, app).await;
        });

        let body = rx.await
            .map_err(|e| NodeError::Other(format!("webhook recv: {}", e)))?;
        server.abort();

        Ok(vec![vec![NodeItem { json: body, ..Default::default() }]])
    }
}

inventory::submit! { NodeFactory::new("webhook", || Arc::new(WebhookNode)) }
