hatchery_server/
node_sink.rs

1//! Production node-call sink on core (A2): applies node calls to the [`Hub`].
2//!
3//! Driven by the [`crate::drain_consumer`]: the node id comes from the drained stream,
4//! the op and payload from the decrypted inner message. The hub re-checks the ledger.
5
6use crate::drain_consumer::CoreHandler;
7use crate::hub::Hub;
8use hatchery_c2_protocol::ChannelKind;
9use hatchery_enroll::call::ForwardedCall;
10use serde_json::{json, Value};
11use std::sync::Arc;
12
13pub struct HubNodeSink {
14    hub: Arc<Hub>,
15}
16
17impl HubNodeSink {
18    pub fn new(hub: Arc<Hub>) -> Self {
19        Self { hub }
20    }
21
22    /// Apply one call (blocking). Returns (status, JSON body).
23    pub fn apply(&self, call: &ForwardedCall) -> (u16, Value) {
24        match self.hub.node_call(&call.node_id, &call.op, &call.payload) {
25            Ok(value) => (200, value),
26            Err(err) => (err.status, json!({ "error": err.message })),
27        }
28    }
29
30    /// Drain handler for `m4a` / g4a node-call channels: `{"op","payload"}` in,
31    /// `{"status","body"}` out.
32    pub fn handler(self: Arc<Self>) -> CoreHandler {
33        Arc::new(move |node_id: &str, _kind: ChannelKind, plain: &[u8]| {
34            let reply = match serde_json::from_slice::<Value>(plain) {
35                Ok(req) => {
36                    let op = req.get("op").and_then(Value::as_str).unwrap_or_default().to_owned();
37                    let payload = req.get("payload").cloned().unwrap_or(Value::Null);
38                    let (status, body) = self.apply(&ForwardedCall { node_id: node_id.to_owned(), op, payload });
39                    json!({ "status": status, "body": body })
40                }
41                Err(_) => json!({ "status": 400, "body": { "error": "bad request" } }),
42            };
43            serde_json::to_vec(&reply).ok()
44        })
45    }
46}
hatchery_server/node_sink.rs

hatchery_server/
node_sink.rs
