Building Adapters

use ntl::adapter::{Adapter, AdapterHealth, ExternalPayload, Protocol};
use ntl::Signal;

trait Adapter: Send + Sync {
    fn ingest(&self, external: ExternalPayload) -> Result<Signal>;
    fn emit(&self, signal: Signal) -> Result<ExternalPayload>;
    fn protocol(&self) -> Protocol;
    fn health(&self) -> AdapterHealth;
}

use ntl::{Node, Signal, SignalType};
use ntl::adapter::{Adapter, AdapterHealth, ExternalPayload, Protocol};
use axum::{Router, Json, extract::State};
use std::sync::Arc;

struct HttpAdapter {
    node: Arc<Node>,
}

impl HttpAdapter {
    pub fn new(node: Arc<Node>) -> Self {
        Self { node }
    }

    pub fn router(self: Arc<Self>) -> Router {
        Router::new()
            .route("/signal", axum::routing::post(Self::handle_request))
            .with_state(self)
    }

    async fn handle_request(
        State(adapter): State<Arc<HttpAdapter>>,
        Json(body): Json<SignalRequest>,
    ) -> Json<SignalResponse> {
        // Translate HTTP request to NTL signal
        let signal = Signal::new(body.signal_type)
            .with_payload(body.payload)
            .with_weight(body.weight.unwrap_or(0.5))
            .with_tags(body.tags);

        // Emit into NTL network
        let signal_id = signal.emit(&adapter.node).await.unwrap();

        // Wait for correlated response
        match adapter.node
            .wait_correlation(signal_id, std::time::Duration::from_secs(30))
            .await
        {
            Ok(response) => Json(SignalResponse {
                status: "ok".into(),
                signal_id: signal_id.to_string(),
                data: Some(response.payload),
            }),
            Err(_) => Json(SignalResponse {
                status: "timeout".into(),
                signal_id: signal_id.to_string(),
                data: None,
            }),
        }
    }
}

impl Adapter for HttpAdapter {
    fn ingest(&self, external: ExternalPayload) -> Result<Signal> {
        let request: SignalRequest = serde_json::from_slice(&external.data)?;
        Ok(Signal::new(request.signal_type)
            .with_payload(request.payload)
            .with_weight(request.weight.unwrap_or(0.5)))
    }

    fn emit(&self, signal: Signal) -> Result<ExternalPayload> {
        let body = serde_json::to_vec(&signal.payload)?;
        Ok(ExternalPayload {
            data: body,
            content_type: "application/json".into(),
        })
    }

    fn protocol(&self) -> Protocol {
        Protocol::Http
    }

    fn health(&self) -> AdapterHealth {
        AdapterHealth::Healthy
    }
}

#[derive(serde::Deserialize)]
struct SignalRequest {
    signal_type: SignalType,
    payload: serde_json::Value,
    weight: Option<f32>,
    tags: Vec<String>,
}

#[derive(serde::Serialize)]
struct SignalResponse {
    status: String,
    signal_id: String,
    data: Option<serde_json::Value>,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let node = Arc::new(
        Node::builder()
            .with_config_file("~/.ntl/config.toml")
            .build()
            .await?
    );

    let adapter = Arc::new(HttpAdapter::new(node.clone()));
    let router = adapter.router();

    // Run NTL node and HTTP server concurrently
    tokio::select! {
        _ = node.run_until_shutdown() => {},
        _ = axum::Server::bind(&"0.0.0.0:3000".parse()?)
            .serve(router.into_make_service()) => {},
    }

    Ok(())
}

curl -X POST http://localhost:3000/signal \
  -H "Content-Type: application/json" \
  -d '{
    "signal_type": "query",
    "payload": {"key": "greeting"},
    "tags": ["kv", "get"]
  }'

node.register_adapter(adapter.clone()).await?;

// Now visible in:
// ntl status
// ntl adapters list

#[cfg(test)]
mod tests {
    use ntl::testing::MockNode;

    #[tokio::test]
    async fn test_http_adapter_ingest() {
        let node = MockNode::new();
        let adapter = HttpAdapter::new(Arc::new(node));

        let payload = ExternalPayload {
            data: serde_json::to_vec(&serde_json::json!({
                "signal_type": "query",
                "payload": {"key": "test"},
                "tags": ["kv"]
            })).unwrap(),
            content_type: "application/json".into(),
        };

        let signal = adapter.ingest(payload).unwrap();
        assert_eq!(signal.signal_type, SignalType::Query);
    }
}