hatchery_server/
hub.rs

1//! P6: the hub behind the server control API: node calls forwarded by C2 and operator (HQ)
2//! calls for mail and tasks. All state lives under the data dir; admission is always read from
3//! the P2 ledger, never from the request.
4
5use crate::messenger::{Messenger, MessengerError, OPERATOR};
6use crate::provision::{read_ledger, LedgerEntry, NodeStatus};
7use crate::tasks::TaskStore;
8use serde_json::{json, Value};
9use std::path::{Path, PathBuf};
10use std::sync::{Condvar, Mutex};
11use std::time::{Duration, Instant};
12
13#[derive(Debug)]
14pub struct HubError {
15    pub status: u16,
16    pub message: String,
17}
18
19impl HubError {
20    fn new(status: u16, message: impl Into<String>) -> Self {
21        Self { status, message: message.into() }
22    }
23}
24
25impl From<MessengerError> for HubError {
26    fn from(e: MessengerError) -> Self {
27        match e {
28            MessengerError::Forbidden => Self::new(403, "refused"),
29            MessengerError::Bad(m) => Self::new(400, m),
30            MessengerError::Internal(m) => Self::new(500, m),
31        }
32    }
33}
34
35pub struct Hub {
36    data_dir: PathBuf,
37    messenger: Mutex<Messenger>,
38    tasks: Mutex<TaskStore>,
39    /// Bumped on every message sent; lets HQ long-poll for new mail instead of polling blindly.
40    mail_signal: (Mutex<u64>, Condvar),
41    /// hq-core task kernel (former harness), when built in and its databases open.
42    #[cfg(feature = "harness")]
43    kernel: Mutex<KernelSlot>,
44}
45
46fn str_field<'a>(v: &'a Value, name: &str) -> Result<&'a str, HubError> {
47    v.get(name).and_then(Value::as_str).ok_or_else(|| HubError::new(400, format!("missing field: {name}")))
48}
49
50#[cfg(feature = "harness")]
51enum KernelSlot {
52    None,
53    /// Store-only kernel opened in-process (no runtime).
54    Local(crate::kernel::Kernel),
55    /// The kernel runtime owns the databases; requests are forwarded to it.
56    Remote(crate::kernel_runtime::RemoteKernel),
57}
58
59impl Hub {
60    /// Hand the kernel databases to the runtime: closes the in-process store first.
61    #[cfg(feature = "harness")]
62    pub fn use_remote_kernel(&self, remote: crate::kernel_runtime::RemoteKernel) {
63        *self.kernel.lock().expect("kernel") = KernelSlot::Remote(remote);
64    }
65
66    pub fn open(data_dir: &Path) -> Result<Self, crate::error::ServerError> {
67        Ok(Self {
68            data_dir: data_dir.to_owned(),
69            messenger: Mutex::new(Messenger::open(data_dir)?),
70            tasks: Mutex::new(TaskStore::open(data_dir)?),
71            mail_signal: (Mutex::new(0), Condvar::new()),
72            #[cfg(feature = "harness")]
73            kernel: Mutex::new(match crate::kernel::Kernel::open(data_dir) {
74                Ok(k) => KernelSlot::Local(k),
75                Err(err) => {
76                    tracing::warn!(error = %err, "task kernel unavailable");
77                    KernelSlot::None
78                }
79            }),
80        })
81    }
82
83    fn bump_mail(&self) {
84        *self.mail_signal.0.lock().expect("signal") += 1;
85        self.mail_signal.1.notify_all();
86    }
87
88    fn entry(&self, node_id: &str) -> Option<LedgerEntry> {
89        read_ledger(&self.data_dir).ok()?.into_iter().find(|e| e.node_id == node_id)
90    }
91
92    /// Active (member/trusted) node holding `capability`, per the ledger.
93    fn admitted(&self, node_id: &str, capability: &str) -> bool {
94        node_id != OPERATOR
95            && self.entry(node_id).is_some_and(|e| {
96                matches!(e.status, NodeStatus::Member | NodeStatus::Trusted) && e.grants.capabilities.iter().any(|c| c == capability)
97            })
98    }
99
100    /// A call made on behalf of a node. The caller (C2) has authenticated the node and checked its
101    /// policy; the hub checks the ledger again, independently.
102    pub fn node_call(&self, node_id: &str, op: &str, payload: &Value) -> Result<Value, HubError> {
103        let capability = match op.split_once('.').map(|p| p.0) {
104            Some("mail") => "mail",
105            Some("task") => "tasks",
106            _ => return Err(HubError::new(404, "unknown operation")),
107        };
108        if !self.admitted(node_id, capability) {
109            return Err(HubError::new(403, "refused"));
110        }
111        match op {
112            "mail.send" => {
113                let key = payload.get("idempotency_key").and_then(Value::as_str).map(str::to_owned);
114                let id = self.messenger.lock().expect("messenger").send(node_id, str_field(payload, "to")?, str_field(payload, "subject")?, str_field(payload, "body")?, key)?;
115                self.bump_mail();
116                Ok(json!({ "message_id": id }))
117            }
118            "mail.inbox" => {
119                let since = payload.get("since_ms").and_then(Value::as_u64).unwrap_or(0);
120                let limit = payload.get("limit").and_then(Value::as_u64).unwrap_or(50).min(256) as u16;
121                let mail = self.messenger.lock().expect("messenger").inbox(node_id, since, limit)?;
122                Ok(json!({ "messages": mail }))
123            }
124            "mail.ack" => {
125                self.messenger.lock().expect("messenger").ack(node_id, str_field(payload, "message_id")?)?;
126                Ok(json!({ "ok": true }))
127            }
128            "task.list" => {
129                let tasks = self.tasks.lock().expect("tasks").list(Some(node_id), true).map_err(|e| HubError::new(500, e.to_string()))?;
130                Ok(json!({ "tasks": tasks }))
131            }
132            "task.update" => {
133                let id = payload.get("id").and_then(Value::as_i64).ok_or_else(|| HubError::new(400, "missing field: id"))?;
134                let status = str_field(payload, "status")?;
135                if !matches!(status, "in_progress" | "done") {
136                    return Err(HubError::new(400, "status must be in_progress or done"));
137                }
138                let changed = self.tasks.lock().expect("tasks").set_status(id, status, Some(node_id)).map_err(|e| HubError::new(500, e.to_string()))?;
139                if changed { Ok(json!({ "ok": true })) } else { Err(HubError::new(403, "refused")) }
140            }
141            _ => Err(HubError::new(404, "unknown operation")),
142        }
143    }
144
145    /// Operator (HQ) calls, authenticated by the control bearer in the HTTP layer.
146    pub fn operator_call(&self, method: &str, path: &str, body: &Value) -> Result<Value, HubError> {
147        let (path, query) = path.split_once('?').unwrap_or((path, ""));
148        let parts: Vec<&str> = path.trim_matches('/').split('/').collect();
149        #[cfg(feature = "harness")]
150        if let ["v1", "kernel", rest @ ..] = parts.as_slice() {
151            let slot = self.kernel.lock().expect("kernel");
152            let result = match (&*slot, method, rest) {
153                (KernelSlot::Remote(r), "POST", ["operator"]) => {
154                    let r = r.clone();
155                    drop(slot);
156                    r.call(body)
157                }
158                (KernelSlot::Remote(_), _, _) => {
159                    return Err(HubError::new(409, "kernel runs as a runtime; use POST /v1/kernel/operator"))
160                }
161                (KernelSlot::Local(_), "POST", ["operator"]) => {
162                    return Err(HubError::new(409, "kernel runtime not running"))
163                }
164                (KernelSlot::Local(_), _, _) => {
165                    let mut slot = slot;
166                    let KernelSlot::Local(k) = &mut *slot else { unreachable!() };
167                    k.call(method, rest, body)
168                }
169                (KernelSlot::None, _, _) => return Err(HubError::new(503, "task kernel unavailable")),
170            };
171            return result.map_err(|e| HubError::new(e.status, e.message));
172        }
173        match (method, parts.as_slice()) {
174            ("POST", ["v1", "tasks"]) => {
175                let assignee = body.get("assignee").and_then(Value::as_str);
176                if let Some(node) = assignee {
177                    if !self.admitted(node, "tasks") {
178                        return Err(HubError::new(403, "assignee does not hold the tasks capability"));
179                    }
180                }
181                let task = self.tasks.lock().expect("tasks").create(str_field(body, "title")?, body.get("body").and_then(Value::as_str).unwrap_or(""), assignee).map_err(|e| HubError::new(500, e.to_string()))?;
182                Ok(json!({ "task": task }))
183            }
184            ("GET", ["v1", "tasks"]) => {
185                let assignee = query.split('&').find_map(|kv| kv.strip_prefix("assignee="));
186                let tasks = self.tasks.lock().expect("tasks").list(assignee, false).map_err(|e| HubError::new(500, e.to_string()))?;
187                Ok(json!({ "tasks": tasks }))
188            }
189            ("POST", ["v1", "tasks", id, "assign"]) => {
190                let id: i64 = id.parse().map_err(|_| HubError::new(400, "bad id"))?;
191                let node = str_field(body, "assignee")?;
192                if !self.admitted(node, "tasks") {
193                    return Err(HubError::new(403, "assignee does not hold the tasks capability"));
194                }
195                if self.tasks.lock().expect("tasks").assign(id, node).map_err(|e| HubError::new(500, e.to_string()))? { Ok(json!({ "ok": true })) } else { Err(HubError::new(409, "task cannot be assigned")) }
196            }
197            ("POST", ["v1", "tasks", id, "cancel"]) => {
198                let id: i64 = id.parse().map_err(|_| HubError::new(400, "bad id"))?;
199                if self.tasks.lock().expect("tasks").set_status(id, "cancelled", None).map_err(|e| HubError::new(500, e.to_string()))? { Ok(json!({ "ok": true })) } else { Err(HubError::new(409, "task is final")) }
200            }
201            ("POST", ["v1", "mail", "send"]) => {
202                let id = self.messenger.lock().expect("messenger").send(OPERATOR, str_field(body, "to")?, str_field(body, "subject")?, str_field(body, "body")?, None)?;
203                self.bump_mail();
204                Ok(json!({ "message_id": id }))
205            }
206            // Long-poll for HQ: returns as soon as the operator inbox holds a message newer than
207            // `since_ms`, or after `timeout_secs` (max 30) with an empty list.
208            ("POST", ["v1", "mail", "wait"]) => {
209                let since = body.get("since_ms").and_then(Value::as_u64).unwrap_or(0);
210                let deadline = Instant::now() + Duration::from_secs(body.get("timeout_secs").and_then(Value::as_u64).unwrap_or(25).min(30));
211                loop {
212                    let seen = *self.mail_signal.0.lock().expect("signal");
213                    let mail: Vec<_> = self.messenger.lock().expect("messenger").inbox(OPERATOR, since, 100)?.into_iter().filter(|m| m.created_at_ms > since).collect();
214                    if !mail.is_empty() {
215                        return Ok(json!({ "messages": mail }));
216                    }
217                    let now = Instant::now();
218                    if now >= deadline {
219                        return Ok(json!({ "messages": [] }));
220                    }
221                    let guard = self.mail_signal.0.lock().expect("signal");
222                    if *guard == seen {
223                        let _ = self.mail_signal.1.wait_timeout(guard, deadline - now).expect("signal wait");
224                    }
225                }
226            }
227            ("POST", ["v1", "mail", "thread"]) => {
228                let mail = self.messenger.lock().expect("messenger").thread(str_field(body, "node")?, 200)?;
229                Ok(json!({ "messages": mail }))
230            }
231            ("POST", ["v1", "mail", "inbox"]) => {
232                let mail = self.messenger.lock().expect("messenger").inbox(OPERATOR, 0, 100)?;
233                Ok(json!({ "messages": mail }))
234            }
235            _ => Err(HubError::new(404, "not found")),
236        }
237    }
238}
hatchery_server/hub.rs

hatchery_server/
hub.rs
