pulse_client/resources.rs
1//! Resource accessors — one per OpenAPI tag.
2
3use reqwest::Method;
4use serde_json::{json, Value};
5
6use crate::client::PulseClient;
7use crate::error::PulseError;
8
9// ---------------------------------------------------------------------------
10// AuthResource — client.auth()
11// ---------------------------------------------------------------------------
12
13pub struct AuthResource<'c> {
14 pub(crate) client: &'c PulseClient,
15}
16
17impl AuthResource<'_> {
18 /// `POST /api/auth/login` — exchanges username + password for a JWT.
19 ///
20 /// On success, the returned token is cached on the parent client so
21 /// subsequent calls authenticate automatically.
22 pub async fn login(&self, username: &str, password: &str) -> Result<Value, PulseError> {
23 let body = json!({ "username": username, "password": password });
24 let response = self
25 .client
26 .request(Method::POST, "/api/auth/login", Some(&body), false)
27 .await?;
28 cache_token(self.client, &response);
29 Ok(response)
30 }
31
32 /// `POST /api/auth/refresh` — exchanges a refresh token for a fresh JWT.
33 pub async fn refresh(&self, refresh_token: &str) -> Result<Value, PulseError> {
34 let body = json!({ "refreshToken": refresh_token });
35 let response = self
36 .client
37 .request(Method::POST, "/api/auth/refresh", Some(&body), false)
38 .await?;
39 cache_token(self.client, &response);
40 Ok(response)
41 }
42
43 /// `GET /api/auth/organizations` — orgs the current user is a member of.
44 pub async fn organizations(&self) -> Result<Vec<Value>, PulseError> {
45 let result = self
46 .client
47 .request(Method::GET, "/api/auth/organizations", None::<&()>, true)
48 .await?;
49 Ok(unwrap_list(&result, "organizations"))
50 }
51
52 /// `POST /api/auth/switch-org` — switches the active organisation.
53 /// The new JWT (with updated orgId claim) is cached on the parent client.
54 pub async fn switch_org(&self, org_id: &str) -> Result<Value, PulseError> {
55 let body = json!({ "orgId": org_id });
56 let response = self
57 .client
58 .request(Method::POST, "/api/auth/switch-org", Some(&body), true)
59 .await?;
60 cache_token(self.client, &response);
61 Ok(response)
62 }
63}
64
65fn cache_token(client: &PulseClient, response: &Value) {
66 if let Some(token) = response
67 .get("accessToken")
68 .or_else(|| response.get("token"))
69 .and_then(Value::as_str)
70 {
71 if !token.is_empty() {
72 client.set_token(token);
73 }
74 }
75}
76
77// ---------------------------------------------------------------------------
78// PipelinesResource — client.pipelines()
79// ---------------------------------------------------------------------------
80
81pub struct PipelinesResource<'c> {
82 pub(crate) client: &'c PulseClient,
83}
84
85impl PipelinesResource<'_> {
86 /// `GET /api/pulse/pipelines` — every pipeline in the current org.
87 pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
88 let result = self
89 .client
90 .request(Method::GET, "/api/pulse/pipelines", None::<&()>, true)
91 .await?;
92 Ok(unwrap_list(&result, "pipelines"))
93 }
94
95 /// `GET /api/pulse/pipelines/{id}` — one pipeline by id.
96 pub async fn get(&self, pipeline_id: &str) -> Result<Value, PulseError> {
97 let path = format!("/api/pulse/pipelines/{}", encode_path(pipeline_id));
98 self.client
99 .request(Method::GET, &path, None::<&()>, true)
100 .await
101 }
102
103 /// `POST /api/pulse/pipelines` — creates + deploys a new pipeline.
104 ///
105 /// The definition must follow the `CreatePipelineRequest` schema (see
106 /// openapi.yaml). At minimum: `name` + `nodes`.
107 pub async fn create(&self, definition: &Value) -> Result<Value, PulseError> {
108 self.client
109 .request(Method::POST, "/api/pulse/pipelines", Some(definition), true)
110 .await
111 }
112
113 /// `DELETE /api/pulse/pipelines/{id}` — tears down the pipeline.
114 pub async fn delete(&self, pipeline_id: &str) -> Result<(), PulseError> {
115 let path = format!("/api/pulse/pipelines/{}", encode_path(pipeline_id));
116 self.client
117 .request(Method::DELETE, &path, None::<&()>, true)
118 .await?;
119 Ok(())
120 }
121}
122
123// ---------------------------------------------------------------------------
124// AgentsResource — client.agents()
125// ---------------------------------------------------------------------------
126
127pub struct AgentsResource<'c> {
128 pub(crate) client: &'c PulseClient,
129}
130
131impl AgentsResource<'_> {
132 /// `GET /api/pulse/agents` — every deployed agent in the current org.
133 pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
134 let result = self
135 .client
136 .request(Method::GET, "/api/pulse/agents", None::<&()>, true)
137 .await?;
138 Ok(unwrap_list(&result, "agents"))
139 }
140
141 /// `GET /api/pulse/agents/{id}` — one agent by id.
142 pub async fn get(&self, agent_id: &str) -> Result<Value, PulseError> {
143 let path = format!("/api/pulse/agents/{}", encode_path(agent_id));
144 self.client
145 .request(Method::GET, &path, None::<&()>, true)
146 .await
147 }
148
149 /// B-115 Phase 1 — `PUT /api/pulse/agents/{id}`: replace the agent's config.
150 ///
151 /// `config` is the FULL agent config (not a partial merge) — at minimum
152 /// `name`. Optional fields (`engineType`, `inputTopic`, `outputTopic`,
153 /// `description`, `instances`, `monthlyBudget`, `config`) fall back to safe
154 /// defaults when omitted. See the `UpdateAgentRequest` schema in
155 /// `openapi.yaml`.
156 ///
157 /// Today this triggers a full stop + persist + start cycle on the engine
158 /// side — the agent is briefly unavailable while the swap happens.
159 /// Existing state in the agent's keyed store is preserved. Phase 2
160 /// (B-115-engine) will add atomic event-boundary swap so hot-reloadable
161 /// changes apply with no downtime.
162 ///
163 /// Returns the post-update agent snapshot (same shape as [`get`](Self::get)).
164 ///
165 /// # Errors
166 ///
167 /// - [`PulseError::Validation`] on a bad config (self-loop, invalid
168 /// streaming operators)
169 /// - [`PulseError::NotFound`] if the agent doesn't exist
170 pub async fn update(&self, agent_id: &str, config: &Value) -> Result<Value, PulseError> {
171 let path = format!("/api/pulse/agents/{}", encode_path(agent_id));
172 self.client
173 .request(Method::PUT, &path, Some(config), true)
174 .await
175 }
176
177 /// `DELETE /api/pulse/agents/{id}` — stop the agent + remove its config row.
178 ///
179 /// The agent's keyed state store is also dropped. Requires the
180 /// `AGENT_DELETE` permission.
181 pub async fn delete(&self, agent_id: &str) -> Result<(), PulseError> {
182 let path = format!("/api/pulse/agents/{}", encode_path(agent_id));
183 self.client
184 .request::<()>(Method::DELETE, &path, None, true)
185 .await?;
186 Ok(())
187 }
188}
189
190// ---------------------------------------------------------------------------
191// TemplatesResource — client.templates()
192// ---------------------------------------------------------------------------
193
194pub struct TemplatesResource<'c> {
195 pub(crate) client: &'c PulseClient,
196}
197
198impl TemplatesResource<'_> {
199 /// `GET /api/pulse/templates` — the 223+ first-party templates.
200 pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
201 let result = self
202 .client
203 .request(Method::GET, "/api/pulse/templates", None::<&()>, true)
204 .await?;
205 Ok(unwrap_list(&result, "templates"))
206 }
207}
208
209// ---------------------------------------------------------------------------
210// ModelsResource — client.models()
211// ---------------------------------------------------------------------------
212
213/// `client.models()` — B-112 embedded ML model registry.
214///
215/// Upload ONNX models that the streaming `ml_predict` operator scores events
216/// against, in-process on the Pulse engine (no model-server hop). Models are
217/// org-scoped; upload / delete require the ADMIN role.
218///
219/// # Example
220///
221/// ```no_run
222/// use pulse_client::{PulseClient, ModelUpload};
223/// use std::collections::BTreeMap;
224///
225/// # async fn run(client: &PulseClient) -> Result<(), pulse_client::PulseError> {
226/// let mut input = BTreeMap::new();
227/// input.insert("amount".to_string(), "float".to_string());
228/// input.insert("country".to_string(), "string".to_string());
229///
230/// client
231/// .models()
232/// .upload(
233/// ModelUpload::from_path("fraud-classifier", "./model.onnx")
234/// .input_schema(input),
235/// )
236/// .await?;
237/// # Ok(())
238/// # }
239/// ```
240pub struct ModelsResource<'c> {
241 pub(crate) client: &'c PulseClient,
242}
243
244/// B-112 — describes a model upload to [`ModelsResource::upload`].
245///
246/// Supply the model bytes either by file `path` (read at upload time) or as
247/// raw `data`. Exactly one of the two must be set — [`ModelsResource::upload`]
248/// returns a [`PulseError::InvalidConfig`] otherwise.
249#[derive(Debug, Clone, Default)]
250pub struct ModelUpload {
251 /// Model name referenced by `ml_predict(model = ...)`.
252 pub name: String,
253 /// Filesystem path to the `.onnx` file. Mutually exclusive with `data`.
254 pub path: Option<String>,
255 /// Raw model bytes. Mutually exclusive with `path`.
256 pub data: Option<Vec<u8>>,
257 /// Model runtime — only `"onnx"` is supported today. Defaults to `"onnx"`.
258 pub runtime: Option<String>,
259 /// Ordered feature-name → type map, used to pack features into the input
260 /// tensor (in the model's input order).
261 pub input_schema: Option<std::collections::BTreeMap<String, String>>,
262 /// Output-name → type map (informational).
263 pub output_schema: Option<std::collections::BTreeMap<String, String>>,
264}
265
266impl ModelUpload {
267 /// Upload from a filesystem path to the `.onnx` file.
268 pub fn from_path(name: impl Into<String>, path: impl Into<String>) -> Self {
269 Self {
270 name: name.into(),
271 path: Some(path.into()),
272 ..Self::default()
273 }
274 }
275
276 /// Upload from raw model bytes.
277 pub fn from_bytes(name: impl Into<String>, data: Vec<u8>) -> Self {
278 Self {
279 name: name.into(),
280 data: Some(data),
281 ..Self::default()
282 }
283 }
284
285 /// Override the runtime (default `"onnx"`).
286 pub fn runtime(mut self, runtime: impl Into<String>) -> Self {
287 self.runtime = Some(runtime.into());
288 self
289 }
290
291 /// Set the ordered input feature schema.
292 pub fn input_schema(mut self, schema: std::collections::BTreeMap<String, String>) -> Self {
293 self.input_schema = Some(schema);
294 self
295 }
296
297 /// Set the (informational) output schema.
298 pub fn output_schema(mut self, schema: std::collections::BTreeMap<String, String>) -> Self {
299 self.output_schema = Some(schema);
300 self
301 }
302}
303
304impl ModelsResource<'_> {
305 /// `POST /api/pulse/ml-models` — upload (or replace) a model.
306 ///
307 /// Sent as `multipart/form-data`: a file part named `model` carrying the
308 /// bytes, plus text parts `name`, `runtime`, and (when set) `inputSchema` /
309 /// `outputSchema` as JSON strings. Replacing an existing name hot-swaps the
310 /// model with no agent restart.
311 ///
312 /// Returns the persisted model metadata (name, runtime, sha256, version, …).
313 ///
314 /// # Errors
315 ///
316 /// - [`PulseError::InvalidConfig`] if `name` is blank, if neither or both
317 /// of `path`/`data` are set, or if the model bytes are empty.
318 /// - [`PulseError::Transport`] if reading the file at `path` fails.
319 pub async fn upload(&self, upload: ModelUpload) -> Result<Value, PulseError> {
320 if upload.name.trim().is_empty() {
321 return Err(PulseError::InvalidConfig(
322 "model name must be a non-empty string".to_string(),
323 ));
324 }
325 if upload.path.is_some() == upload.data.is_some() {
326 return Err(PulseError::InvalidConfig(
327 "provide exactly one of 'path' or 'data'".to_string(),
328 ));
329 }
330
331 let (blob, filename) = match (&upload.path, upload.data) {
332 (Some(path), None) => {
333 let bytes = std::fs::read(path)
334 .map_err(|e| PulseError::InvalidConfig(format!("read {path}: {e}")))?;
335 let filename = path
336 .rsplit(['/', '\\'])
337 .next()
338 .filter(|s| !s.is_empty())
339 .unwrap_or("model.onnx")
340 .to_string();
341 (bytes, filename)
342 }
343 (None, Some(data)) => (data, format!("{}.onnx", upload.name)),
344 // Unreachable — guarded by the XOR check above.
345 _ => unreachable!("exactly one of path/data enforced above"),
346 };
347 if blob.is_empty() {
348 return Err(PulseError::InvalidConfig(
349 "model bytes are empty".to_string(),
350 ));
351 }
352
353 let runtime = upload.runtime.unwrap_or_else(|| "onnx".to_string());
354 let model_part = reqwest::multipart::Part::bytes(blob)
355 .file_name(filename)
356 .mime_str("application/octet-stream")
357 .map_err(PulseError::Transport)?;
358 let mut form = reqwest::multipart::Form::new()
359 .text("name", upload.name)
360 .text("runtime", runtime)
361 .part("model", model_part);
362 if let Some(schema) = upload.input_schema {
363 form = form.text("inputSchema", serde_json::to_string(&schema)?);
364 }
365 if let Some(schema) = upload.output_schema {
366 form = form.text("outputSchema", serde_json::to_string(&schema)?);
367 }
368
369 self.client
370 .request_multipart("/api/pulse/ml-models", form)
371 .await
372 }
373
374 /// `GET /api/pulse/ml-models` — models registered for the caller's org.
375 pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
376 let result = self
377 .client
378 .request(Method::GET, "/api/pulse/ml-models", None::<&()>, true)
379 .await?;
380 Ok(unwrap_list(&result, "models"))
381 }
382
383 /// `GET /api/pulse/ml-models/{name}` — metadata for one model.
384 pub async fn get(&self, name: &str) -> Result<Value, PulseError> {
385 let path = format!("/api/pulse/ml-models/{}", encode_path(name));
386 self.client
387 .request(Method::GET, &path, None::<&()>, true)
388 .await
389 }
390
391 /// `DELETE /api/pulse/ml-models/{name}` — remove a model (ADMIN).
392 pub async fn delete(&self, name: &str) -> Result<(), PulseError> {
393 let path = format!("/api/pulse/ml-models/{}", encode_path(name));
394 self.client
395 .request::<()>(Method::DELETE, &path, None, true)
396 .await?;
397 Ok(())
398 }
399}
400
401// ---------------------------------------------------------------------------
402// WasmResource — client.wasm()
403// ---------------------------------------------------------------------------
404
405/// `client.wasm()` — B-110 sandboxed WASM module registry.
406///
407/// Upload WebAssembly modules that the streaming `wasm` operator runs over
408/// events, sandboxed in pure-Java Chicory on the engine (no host syscalls,
409/// bounded linear memory). Modules are org-scoped; upload / delete require the
410/// ADMIN role.
411///
412/// # Example
413///
414/// ```no_run
415/// use pulse_client::{PulseClient, WasmUpload};
416///
417/// # async fn run(client: &PulseClient) -> Result<(), pulse_client::PulseError> {
418/// client
419/// .wasm()
420/// .upload(WasmUpload::from_path("pii-redactor", "./redactor.wasm"))
421/// .await?;
422/// # Ok(())
423/// # }
424/// ```
425pub struct WasmResource<'c> {
426 pub(crate) client: &'c PulseClient,
427}
428
429/// B-110 — describes a WASM module upload to [`WasmResource::upload`].
430///
431/// Supply the module bytes either by file `path` (read at upload time) or as
432/// raw `data`. Exactly one of the two must be set — [`WasmResource::upload`]
433/// returns a [`PulseError::InvalidConfig`] otherwise.
434#[derive(Debug, Clone, Default)]
435pub struct WasmUpload {
436 /// Module name referenced by `wasm(module = ...)`.
437 pub name: String,
438 /// Filesystem path to the `.wasm` file. Mutually exclusive with `data`.
439 pub path: Option<String>,
440 /// Raw module bytes. Mutually exclusive with `path`.
441 pub data: Option<Vec<u8>>,
442 /// Optional human-readable description stored alongside the module.
443 pub description: Option<String>,
444}
445
446impl WasmUpload {
447 /// Upload from a filesystem path to the `.wasm` file.
448 pub fn from_path(name: impl Into<String>, path: impl Into<String>) -> Self {
449 Self {
450 name: name.into(),
451 path: Some(path.into()),
452 ..Self::default()
453 }
454 }
455
456 /// Upload from raw module bytes.
457 pub fn from_bytes(name: impl Into<String>, data: Vec<u8>) -> Self {
458 Self {
459 name: name.into(),
460 data: Some(data),
461 ..Self::default()
462 }
463 }
464
465 /// Attach a human-readable description.
466 pub fn description(mut self, description: impl Into<String>) -> Self {
467 self.description = Some(description.into());
468 self
469 }
470}
471
472impl WasmResource<'_> {
473 /// `POST /api/pulse/wasm-modules` — upload (or replace) a module.
474 ///
475 /// Sent as `multipart/form-data`: a file part named `module` carrying the
476 /// bytes, plus a text part `name` and (when set) `description`. The module
477 /// is validated server-side (must parse, import no host functions, export
478 /// alloc/process/memory) before persisting. Replacing an existing name
479 /// hot-swaps the module with no agent restart.
480 ///
481 /// Returns the persisted module metadata (name, sha256, version, …).
482 ///
483 /// # Errors
484 ///
485 /// - [`PulseError::InvalidConfig`] if `name` is blank, if neither or both
486 /// of `path`/`data` are set, or if the module bytes are empty.
487 /// - [`PulseError::Transport`] if reading the file at `path` fails.
488 pub async fn upload(&self, upload: WasmUpload) -> Result<Value, PulseError> {
489 if upload.name.trim().is_empty() {
490 return Err(PulseError::InvalidConfig(
491 "module name must be a non-empty string".to_string(),
492 ));
493 }
494 if upload.path.is_some() == upload.data.is_some() {
495 return Err(PulseError::InvalidConfig(
496 "provide exactly one of 'path' or 'data'".to_string(),
497 ));
498 }
499
500 let (blob, filename) = match (&upload.path, upload.data) {
501 (Some(path), None) => {
502 let bytes = std::fs::read(path)
503 .map_err(|e| PulseError::InvalidConfig(format!("read {path}: {e}")))?;
504 let filename = path
505 .rsplit(['/', '\\'])
506 .next()
507 .filter(|s| !s.is_empty())
508 .unwrap_or("module.wasm")
509 .to_string();
510 (bytes, filename)
511 }
512 (None, Some(data)) => (data, format!("{}.wasm", upload.name)),
513 // Unreachable — guarded by the XOR check above.
514 _ => unreachable!("exactly one of path/data enforced above"),
515 };
516 if blob.is_empty() {
517 return Err(PulseError::InvalidConfig(
518 "module bytes are empty".to_string(),
519 ));
520 }
521 // Client-side pre-flight validation — mirrors the server's
522 // `ChicoryWasmRunner.validateModule`. Rejecting a non-conforming module
523 // locally avoids a cryptic server 400 / runtime trap.
524 validate_wasm_module(&blob)?;
525
526 let module_part = reqwest::multipart::Part::bytes(blob)
527 .file_name(filename)
528 .mime_str("application/wasm")
529 .map_err(PulseError::Transport)?;
530 let mut form = reqwest::multipart::Form::new()
531 .text("name", upload.name)
532 .part("module", module_part);
533 if let Some(description) = upload.description {
534 form = form.text("description", description);
535 }
536
537 self.client
538 .request_multipart("/api/pulse/wasm-modules", form)
539 .await
540 }
541
542 /// `GET /api/pulse/wasm-modules` — modules registered for the caller's org.
543 pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
544 let result = self
545 .client
546 .request(Method::GET, "/api/pulse/wasm-modules", None::<&()>, true)
547 .await?;
548 Ok(unwrap_list(&result, "modules"))
549 }
550
551 /// `GET /api/pulse/wasm-modules/{name}` — metadata for one module.
552 pub async fn get(&self, name: &str) -> Result<Value, PulseError> {
553 let path = format!("/api/pulse/wasm-modules/{}", encode_path(name));
554 self.client
555 .request(Method::GET, &path, None::<&()>, true)
556 .await
557 }
558
559 /// `DELETE /api/pulse/wasm-modules/{name}` — remove a module (ADMIN).
560 pub async fn delete(&self, name: &str) -> Result<(), PulseError> {
561 let path = format!("/api/pulse/wasm-modules/{}", encode_path(name));
562 self.client
563 .request::<()>(Method::DELETE, &path, None, true)
564 .await?;
565 Ok(())
566 }
567}
568
569// ---------------------------------------------------------------------------
570// UsersResource — client.users()
571// ---------------------------------------------------------------------------
572
573pub struct UsersResource<'c> {
574 pub(crate) client: &'c PulseClient,
575}
576
577impl UsersResource<'_> {
578 /// `GET /api/pulse/users` — every user in the current org.
579 ///
580 /// Requires the caller to have the `USERS_LIST` permission atom (Owner /
581 /// Platform Admin personas by default — see B-105).
582 pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
583 let result = self
584 .client
585 .request(Method::GET, "/api/pulse/users", None::<&()>, true)
586 .await?;
587 Ok(unwrap_list(&result, "users"))
588 }
589}
590
591// ---------------------------------------------------------------------------
592// Helpers
593// ---------------------------------------------------------------------------
594
595/// Extracts a `Vec<Value>` from `result[key]`. Returns an empty Vec for
596/// missing / malformed envelopes — never panics — so callers can iterate
597/// safely.
598fn unwrap_list(result: &Value, key: &str) -> Vec<Value> {
599 result
600 .get(key)
601 .and_then(Value::as_array)
602 .cloned()
603 .unwrap_or_default()
604}
605
606/// URL-encodes a path-param segment so ids containing `/`, spaces, etc.
607/// round-trip safely. Uses the same character set as the `pulse-go`
608/// `url.PathEscape` and `pulse-java` `URLEncoder` — `+` is encoded as `%20`.
609fn encode_path(segment: &str) -> String {
610 let mut out = String::with_capacity(segment.len());
611 for b in segment.bytes() {
612 match b {
613 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
614 out.push(b as char)
615 }
616 _ => out.push_str(&format!("%{b:02X}")),
617 }
618 }
619 out
620}
621
622/// Reads an unsigned LEB128 integer from `bytes` starting at `*pos`, advancing
623/// `*pos` past the consumed bytes. Returns `None` on overrun (the caller maps
624/// that to a [`PulseError::Validation`] "malformed WASM module").
625fn read_uleb128(bytes: &[u8], pos: &mut usize) -> Option<u64> {
626 let mut result: u64 = 0;
627 let mut shift: u32 = 0;
628 loop {
629 let byte = *bytes.get(*pos)?;
630 *pos += 1;
631 result |= u64::from(byte & 0x7f) << shift;
632 if byte & 0x80 == 0 {
633 return Some(result);
634 }
635 shift += 7;
636 // A uleb128 wider than 64 bits is malformed for our purposes.
637 if shift >= 64 {
638 return None;
639 }
640 }
641}
642
643/// Builds a client-side [`PulseError::Validation`] for the WASM upload path so
644/// the caller sees the SDK's validation error (not a cryptic server 400) before
645/// any network call.
646fn wasm_validation_err(message: &str) -> PulseError {
647 PulseError::Validation {
648 path: "/api/pulse/wasm-modules".to_string(),
649 body: Some(json!({ "error": message })),
650 }
651}
652
653/// Client-side pre-upload validation of a WASM module's raw bytes — inspects the
654/// binary (does NOT execute it) and mirrors the server's
655/// `ChicoryWasmRunner.validateModule`.
656///
657/// A conforming sandbox module must: have the WASM magic + version 1 header,
658/// import zero host functions, and export `alloc`, `process`, and `memory`.
659/// Anything else is rejected with [`PulseError::Validation`] so the failure is
660/// surfaced locally instead of as a server 400 / runtime trap.
661pub fn validate_wasm_module(bytes: &[u8]) -> Result<(), PulseError> {
662 if bytes.len() < 8 {
663 return Err(wasm_validation_err("not a WASM module: too short"));
664 }
665 if bytes[0..4] != [0x00, b'a', b's', b'm'] || bytes[4..8] != [0x01, 0x00, 0x00, 0x00] {
666 return Err(wasm_validation_err("not a WASM module (bad magic/version)"));
667 }
668
669 let malformed = || wasm_validation_err("malformed WASM module");
670
671 let mut has_alloc = false;
672 let mut has_process = false;
673 let mut has_memory = false;
674
675 let mut pos = 8usize;
676 while pos < bytes.len() {
677 // Section id (1 byte) + uleb128 payload size.
678 let id = bytes[pos];
679 pos += 1;
680 let size = read_uleb128(bytes, &mut pos).ok_or_else(malformed)? as usize;
681 let payload_start = pos;
682 let payload_end = payload_start.checked_add(size).ok_or_else(malformed)?;
683 if payload_end > bytes.len() {
684 return Err(malformed());
685 }
686
687 match id {
688 // Import section — any host import disqualifies the module.
689 2 => {
690 let mut p = payload_start;
691 let count = read_uleb128(bytes, &mut p).ok_or_else(malformed)?;
692 if count > 0 {
693 return Err(wasm_validation_err(
694 "WASM module imports host functions; it must be a pure sandbox \
695 (build with no WASI/host imports)",
696 ));
697 }
698 }
699 // Export section — collect exported names.
700 7 => {
701 let mut p = payload_start;
702 let count = read_uleb128(bytes, &mut p).ok_or_else(malformed)?;
703 for _ in 0..count {
704 let name_len = read_uleb128(bytes, &mut p).ok_or_else(malformed)? as usize;
705 let name_end = p.checked_add(name_len).ok_or_else(malformed)?;
706 if name_end > payload_end {
707 return Err(malformed());
708 }
709 let name = &bytes[p..name_end];
710 p = name_end;
711 // 1 kind byte + uleb128 index.
712 if p >= payload_end {
713 return Err(malformed());
714 }
715 p += 1; // kind
716 read_uleb128(bytes, &mut p).ok_or_else(malformed)?; // index
717 match name {
718 b"alloc" => has_alloc = true,
719 b"process" => has_process = true,
720 b"memory" => has_memory = true,
721 _ => {}
722 }
723 }
724 }
725 _ => {}
726 }
727
728 // Advance past the section payload regardless of id.
729 pos = payload_end;
730 }
731
732 if !(has_alloc && has_process && has_memory) {
733 return Err(wasm_validation_err(
734 "WASM module must export alloc, process and memory",
735 ));
736 }
737 Ok(())
738}
739
740// ---------------------------------------------------------------------------
741// ConnectorsResource — client.connectors() (B-093 follow-up: catalogue parity)
742// ---------------------------------------------------------------------------
743
744/// `client.connectors()` — the connector catalogue, the same list the Pipeline
745/// Studio palette and `pulse connectors list` show. Each entry is
746/// `{subType, displayName, configFields}`; use the `subType` as a sink/source
747/// node `type` in a pipeline definition deployed via `client.pipelines()`.
748/// Bridged connectors appear only when the enterprise bridge JAR is on the
749/// server's classpath.
750pub struct ConnectorsResource<'c> {
751 pub(crate) client: &'c PulseClient,
752}
753
754impl ConnectorsResource<'_> {
755 /// `GET /api/pulse/connectors` — `{"sources": [...], "sinks": [...]}`.
756 pub async fn list(&self) -> Result<Value, PulseError> {
757 self.client
758 .request(Method::GET, "/api/pulse/connectors", None::<&()>, true)
759 .await
760 }
761
762 /// Just the sink connectors.
763 pub async fn sinks(&self) -> Result<Vec<Value>, PulseError> {
764 Ok(unwrap_list(&self.list().await?, "sinks"))
765 }
766
767 /// Just the source connectors.
768 pub async fn sources(&self) -> Result<Vec<Value>, PulseError> {
769 Ok(unwrap_list(&self.list().await?, "sources"))
770 }
771}