pulse-client 2.7.0

//! Resource accessors — one per OpenAPI tag.

use reqwest::Method;
use serde_json::{json, Value};

use crate::client::PulseClient;
use crate::error::PulseError;

// ---------------------------------------------------------------------------
// AuthResource — client.auth()
// ---------------------------------------------------------------------------

pub struct AuthResource<'c> {
    pub(crate) client: &'c PulseClient,
}

impl AuthResource<'_> {
    /// `POST /api/auth/login` — exchanges username + password for a JWT.
    ///
    /// On success, the returned token is cached on the parent client so
    /// subsequent calls authenticate automatically.
    pub async fn login(&self, username: &str, password: &str) -> Result<Value, PulseError> {
        let body = json!({ "username": username, "password": password });
        let response = self
            .client
            .request(Method::POST, "/api/auth/login", Some(&body), false)
            .await?;
        cache_token(self.client, &response);
        Ok(response)
    }

    /// `POST /api/auth/refresh` — exchanges a refresh token for a fresh JWT.
    pub async fn refresh(&self, refresh_token: &str) -> Result<Value, PulseError> {
        let body = json!({ "refreshToken": refresh_token });
        let response = self
            .client
            .request(Method::POST, "/api/auth/refresh", Some(&body), false)
            .await?;
        cache_token(self.client, &response);
        Ok(response)
    }

    /// `GET /api/auth/organizations` — orgs the current user is a member of.
    pub async fn organizations(&self) -> Result<Vec<Value>, PulseError> {
        let result = self
            .client
            .request(Method::GET, "/api/auth/organizations", None::<&()>, true)
            .await?;
        Ok(unwrap_list(&result, "organizations"))
    }

    /// `POST /api/auth/switch-org` — switches the active organisation.
    /// The new JWT (with updated orgId claim) is cached on the parent client.
    pub async fn switch_org(&self, org_id: &str) -> Result<Value, PulseError> {
        let body = json!({ "orgId": org_id });
        let response = self
            .client
            .request(Method::POST, "/api/auth/switch-org", Some(&body), true)
            .await?;
        cache_token(self.client, &response);
        Ok(response)
    }
}

fn cache_token(client: &PulseClient, response: &Value) {
    if let Some(token) = response.get("token").and_then(Value::as_str) {
        if !token.is_empty() {
            client.set_token(token);
        }
    }
}

// ---------------------------------------------------------------------------
// PipelinesResource — client.pipelines()
// ---------------------------------------------------------------------------

pub struct PipelinesResource<'c> {
    pub(crate) client: &'c PulseClient,
}

impl PipelinesResource<'_> {
    /// `GET /api/pulse/pipelines` — every pipeline in the current org.
    pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
        let result = self
            .client
            .request(Method::GET, "/api/pulse/pipelines", None::<&()>, true)
            .await?;
        Ok(unwrap_list(&result, "pipelines"))
    }

    /// `GET /api/pulse/pipelines/{id}` — one pipeline by id.
    pub async fn get(&self, pipeline_id: &str) -> Result<Value, PulseError> {
        let path = format!("/api/pulse/pipelines/{}", encode_path(pipeline_id));
        self.client
            .request(Method::GET, &path, None::<&()>, true)
            .await
    }

    /// `POST /api/pulse/pipelines` — creates + deploys a new pipeline.
    ///
    /// The definition must follow the `CreatePipelineRequest` schema (see
    /// openapi.yaml). At minimum: `name` + `nodes`.
    pub async fn create(&self, definition: &Value) -> Result<Value, PulseError> {
        self.client
            .request(Method::POST, "/api/pulse/pipelines", Some(definition), true)
            .await
    }

    /// `DELETE /api/pulse/pipelines/{id}` — tears down the pipeline.
    pub async fn delete(&self, pipeline_id: &str) -> Result<(), PulseError> {
        let path = format!("/api/pulse/pipelines/{}", encode_path(pipeline_id));
        self.client
            .request(Method::DELETE, &path, None::<&()>, true)
            .await?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// AgentsResource — client.agents()
// ---------------------------------------------------------------------------

pub struct AgentsResource<'c> {
    pub(crate) client: &'c PulseClient,
}

impl AgentsResource<'_> {
    /// `GET /api/pulse/agents` — every deployed agent in the current org.
    pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
        let result = self
            .client
            .request(Method::GET, "/api/pulse/agents", None::<&()>, true)
            .await?;
        Ok(unwrap_list(&result, "agents"))
    }

    /// `GET /api/pulse/agents/{id}` — one agent by id.
    pub async fn get(&self, agent_id: &str) -> Result<Value, PulseError> {
        let path = format!("/api/pulse/agents/{}", encode_path(agent_id));
        self.client
            .request(Method::GET, &path, None::<&()>, true)
            .await
    }

    /// B-115 Phase 1 — `PUT /api/pulse/agents/{id}`: replace the agent's config.
    ///
    /// `config` is the FULL agent config (not a partial merge) — at minimum
    /// `name`. Optional fields (`engineType`, `inputTopic`, `outputTopic`,
    /// `description`, `instances`, `monthlyBudget`, `config`) fall back to safe
    /// defaults when omitted. See the `UpdateAgentRequest` schema in
    /// `openapi.yaml`.
    ///
    /// Today this triggers a full stop + persist + start cycle on the engine
    /// side — the agent is briefly unavailable while the swap happens.
    /// Existing state in the agent's keyed store is preserved. Phase 2
    /// (B-115-engine) will add atomic event-boundary swap so hot-reloadable
    /// changes apply with no downtime.
    ///
    /// Returns the post-update agent snapshot (same shape as [`get`](Self::get)).
    ///
    /// # Errors
    ///
    /// - [`PulseError::Validation`] on a bad config (self-loop, invalid
    ///   streaming operators)
    /// - [`PulseError::NotFound`] if the agent doesn't exist
    pub async fn update(&self, agent_id: &str, config: &Value) -> Result<Value, PulseError> {
        let path = format!("/api/pulse/agents/{}", encode_path(agent_id));
        self.client
            .request(Method::PUT, &path, Some(config), true)
            .await
    }

    /// `DELETE /api/pulse/agents/{id}` — stop the agent + remove its config row.
    ///
    /// The agent's keyed state store is also dropped. Requires the
    /// `AGENT_DELETE` permission.
    pub async fn delete(&self, agent_id: &str) -> Result<(), PulseError> {
        let path = format!("/api/pulse/agents/{}", encode_path(agent_id));
        self.client
            .request::<()>(Method::DELETE, &path, None, true)
            .await?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// TemplatesResource — client.templates()
// ---------------------------------------------------------------------------

pub struct TemplatesResource<'c> {
    pub(crate) client: &'c PulseClient,
}

impl TemplatesResource<'_> {
    /// `GET /api/pulse/templates` — the 223+ first-party templates.
    pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
        let result = self
            .client
            .request(Method::GET, "/api/pulse/templates", None::<&()>, true)
            .await?;
        Ok(unwrap_list(&result, "templates"))
    }
}

// ---------------------------------------------------------------------------
// ModelsResource — client.models()
// ---------------------------------------------------------------------------

/// `client.models()` — B-112 embedded ML model registry.
///
/// Upload ONNX models that the streaming `ml_predict` operator scores events
/// against, in-process on the Pulse engine (no model-server hop). Models are
/// org-scoped; upload / delete require the ADMIN role.
///
/// # Example
///
/// ```no_run
/// use pulse_client::{PulseClient, ModelUpload};
/// use std::collections::BTreeMap;
///
/// # async fn run(client: &PulseClient) -> Result<(), pulse_client::PulseError> {
/// let mut input = BTreeMap::new();
/// input.insert("amount".to_string(), "float".to_string());
/// input.insert("country".to_string(), "string".to_string());
///
/// client
///     .models()
///     .upload(
///         ModelUpload::from_path("fraud-classifier", "./model.onnx")
///             .input_schema(input),
///     )
///     .await?;
/// # Ok(())
/// # }
/// ```
pub struct ModelsResource<'c> {
    pub(crate) client: &'c PulseClient,
}

/// B-112 — describes a model upload to [`ModelsResource::upload`].
///
/// Supply the model bytes either by file `path` (read at upload time) or as
/// raw `data`. Exactly one of the two must be set — [`ModelsResource::upload`]
/// returns a [`PulseError::InvalidConfig`] otherwise.
#[derive(Debug, Clone, Default)]
pub struct ModelUpload {
    /// Model name referenced by `ml_predict(model = ...)`.
    pub name: String,
    /// Filesystem path to the `.onnx` file. Mutually exclusive with `data`.
    pub path: Option<String>,
    /// Raw model bytes. Mutually exclusive with `path`.
    pub data: Option<Vec<u8>>,
    /// Model runtime — only `"onnx"` is supported today. Defaults to `"onnx"`.
    pub runtime: Option<String>,
    /// Ordered feature-name → type map, used to pack features into the input
    /// tensor (in the model's input order).
    pub input_schema: Option<std::collections::BTreeMap<String, String>>,
    /// Output-name → type map (informational).
    pub output_schema: Option<std::collections::BTreeMap<String, String>>,
}

impl ModelUpload {
    /// Upload from a filesystem path to the `.onnx` file.
    pub fn from_path(name: impl Into<String>, path: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            path: Some(path.into()),
            ..Self::default()
        }
    }

    /// Upload from raw model bytes.
    pub fn from_bytes(name: impl Into<String>, data: Vec<u8>) -> Self {
        Self {
            name: name.into(),
            data: Some(data),
            ..Self::default()
        }
    }

    /// Override the runtime (default `"onnx"`).
    pub fn runtime(mut self, runtime: impl Into<String>) -> Self {
        self.runtime = Some(runtime.into());
        self
    }

    /// Set the ordered input feature schema.
    pub fn input_schema(mut self, schema: std::collections::BTreeMap<String, String>) -> Self {
        self.input_schema = Some(schema);
        self
    }

    /// Set the (informational) output schema.
    pub fn output_schema(mut self, schema: std::collections::BTreeMap<String, String>) -> Self {
        self.output_schema = Some(schema);
        self
    }
}

impl ModelsResource<'_> {
    /// `POST /api/pulse/ml-models` — upload (or replace) a model.
    ///
    /// Sent as `multipart/form-data`: a file part named `model` carrying the
    /// bytes, plus text parts `name`, `runtime`, and (when set) `inputSchema` /
    /// `outputSchema` as JSON strings. Replacing an existing name hot-swaps the
    /// model with no agent restart.
    ///
    /// Returns the persisted model metadata (name, runtime, sha256, version, …).
    ///
    /// # Errors
    ///
    /// - [`PulseError::InvalidConfig`] if `name` is blank, if neither or both
    ///   of `path`/`data` are set, or if the model bytes are empty.
    /// - [`PulseError::Transport`] if reading the file at `path` fails.
    pub async fn upload(&self, upload: ModelUpload) -> Result<Value, PulseError> {
        if upload.name.trim().is_empty() {
            return Err(PulseError::InvalidConfig(
                "model name must be a non-empty string".to_string(),
            ));
        }
        if upload.path.is_some() == upload.data.is_some() {
            return Err(PulseError::InvalidConfig(
                "provide exactly one of 'path' or 'data'".to_string(),
            ));
        }

        let (blob, filename) = match (&upload.path, upload.data) {
            (Some(path), None) => {
                let bytes = std::fs::read(path)
                    .map_err(|e| PulseError::InvalidConfig(format!("read {path}: {e}")))?;
                let filename = path
                    .rsplit(['/', '\\'])
                    .next()
                    .filter(|s| !s.is_empty())
                    .unwrap_or("model.onnx")
                    .to_string();
                (bytes, filename)
            }
            (None, Some(data)) => (data, format!("{}.onnx", upload.name)),
            // Unreachable — guarded by the XOR check above.
            _ => unreachable!("exactly one of path/data enforced above"),
        };
        if blob.is_empty() {
            return Err(PulseError::InvalidConfig(
                "model bytes are empty".to_string(),
            ));
        }

        let runtime = upload.runtime.unwrap_or_else(|| "onnx".to_string());
        let model_part = reqwest::multipart::Part::bytes(blob)
            .file_name(filename)
            .mime_str("application/octet-stream")
            .map_err(PulseError::Transport)?;
        let mut form = reqwest::multipart::Form::new()
            .text("name", upload.name)
            .text("runtime", runtime)
            .part("model", model_part);
        if let Some(schema) = upload.input_schema {
            form = form.text("inputSchema", serde_json::to_string(&schema)?);
        }
        if let Some(schema) = upload.output_schema {
            form = form.text("outputSchema", serde_json::to_string(&schema)?);
        }

        self.client
            .request_multipart("/api/pulse/ml-models", form)
            .await
    }

    /// `GET /api/pulse/ml-models` — models registered for the caller's org.
    pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
        let result = self
            .client
            .request(Method::GET, "/api/pulse/ml-models", None::<&()>, true)
            .await?;
        Ok(unwrap_list(&result, "models"))
    }

    /// `GET /api/pulse/ml-models/{name}` — metadata for one model.
    pub async fn get(&self, name: &str) -> Result<Value, PulseError> {
        let path = format!("/api/pulse/ml-models/{}", encode_path(name));
        self.client
            .request(Method::GET, &path, None::<&()>, true)
            .await
    }

    /// `DELETE /api/pulse/ml-models/{name}` — remove a model (ADMIN).
    pub async fn delete(&self, name: &str) -> Result<(), PulseError> {
        let path = format!("/api/pulse/ml-models/{}", encode_path(name));
        self.client
            .request::<()>(Method::DELETE, &path, None, true)
            .await?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// WasmResource — client.wasm()
// ---------------------------------------------------------------------------

/// `client.wasm()` — B-110 sandboxed WASM module registry.
///
/// Upload WebAssembly modules that the streaming `wasm` operator runs over
/// events, sandboxed in pure-Java Chicory on the engine (no host syscalls,
/// bounded linear memory). Modules are org-scoped; upload / delete require the
/// ADMIN role.
///
/// # Example
///
/// ```no_run
/// use pulse_client::{PulseClient, WasmUpload};
///
/// # async fn run(client: &PulseClient) -> Result<(), pulse_client::PulseError> {
/// client
///     .wasm()
///     .upload(WasmUpload::from_path("pii-redactor", "./redactor.wasm"))
///     .await?;
/// # Ok(())
/// # }
/// ```
pub struct WasmResource<'c> {
    pub(crate) client: &'c PulseClient,
}

/// B-110 — describes a WASM module upload to [`WasmResource::upload`].
///
/// Supply the module bytes either by file `path` (read at upload time) or as
/// raw `data`. Exactly one of the two must be set — [`WasmResource::upload`]
/// returns a [`PulseError::InvalidConfig`] otherwise.
#[derive(Debug, Clone, Default)]
pub struct WasmUpload {
    /// Module name referenced by `wasm(module = ...)`.
    pub name: String,
    /// Filesystem path to the `.wasm` file. Mutually exclusive with `data`.
    pub path: Option<String>,
    /// Raw module bytes. Mutually exclusive with `path`.
    pub data: Option<Vec<u8>>,
    /// Optional human-readable description stored alongside the module.
    pub description: Option<String>,
}

impl WasmUpload {
    /// Upload from a filesystem path to the `.wasm` file.
    pub fn from_path(name: impl Into<String>, path: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            path: Some(path.into()),
            ..Self::default()
        }
    }

    /// Upload from raw module bytes.
    pub fn from_bytes(name: impl Into<String>, data: Vec<u8>) -> Self {
        Self {
            name: name.into(),
            data: Some(data),
            ..Self::default()
        }
    }

    /// Attach a human-readable description.
    pub fn description(mut self, description: impl Into<String>) -> Self {
        self.description = Some(description.into());
        self
    }
}

impl WasmResource<'_> {
    /// `POST /api/pulse/wasm-modules` — upload (or replace) a module.
    ///
    /// Sent as `multipart/form-data`: a file part named `module` carrying the
    /// bytes, plus a text part `name` and (when set) `description`. The module
    /// is validated server-side (must parse, import no host functions, export
    /// alloc/process/memory) before persisting. Replacing an existing name
    /// hot-swaps the module with no agent restart.
    ///
    /// Returns the persisted module metadata (name, sha256, version, …).
    ///
    /// # Errors
    ///
    /// - [`PulseError::InvalidConfig`] if `name` is blank, if neither or both
    ///   of `path`/`data` are set, or if the module bytes are empty.
    /// - [`PulseError::Transport`] if reading the file at `path` fails.
    pub async fn upload(&self, upload: WasmUpload) -> Result<Value, PulseError> {
        if upload.name.trim().is_empty() {
            return Err(PulseError::InvalidConfig(
                "module name must be a non-empty string".to_string(),
            ));
        }
        if upload.path.is_some() == upload.data.is_some() {
            return Err(PulseError::InvalidConfig(
                "provide exactly one of 'path' or 'data'".to_string(),
            ));
        }

        let (blob, filename) = match (&upload.path, upload.data) {
            (Some(path), None) => {
                let bytes = std::fs::read(path)
                    .map_err(|e| PulseError::InvalidConfig(format!("read {path}: {e}")))?;
                let filename = path
                    .rsplit(['/', '\\'])
                    .next()
                    .filter(|s| !s.is_empty())
                    .unwrap_or("module.wasm")
                    .to_string();
                (bytes, filename)
            }
            (None, Some(data)) => (data, format!("{}.wasm", upload.name)),
            // Unreachable — guarded by the XOR check above.
            _ => unreachable!("exactly one of path/data enforced above"),
        };
        if blob.is_empty() {
            return Err(PulseError::InvalidConfig(
                "module bytes are empty".to_string(),
            ));
        }
        // Client-side pre-flight validation — mirrors the server's
        // `ChicoryWasmRunner.validateModule`. Rejecting a non-conforming module
        // locally avoids a cryptic server 400 / runtime trap.
        validate_wasm_module(&blob)?;

        let module_part = reqwest::multipart::Part::bytes(blob)
            .file_name(filename)
            .mime_str("application/wasm")
            .map_err(PulseError::Transport)?;
        let mut form = reqwest::multipart::Form::new()
            .text("name", upload.name)
            .part("module", module_part);
        if let Some(description) = upload.description {
            form = form.text("description", description);
        }

        self.client
            .request_multipart("/api/pulse/wasm-modules", form)
            .await
    }

    /// `GET /api/pulse/wasm-modules` — modules registered for the caller's org.
    pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
        let result = self
            .client
            .request(Method::GET, "/api/pulse/wasm-modules", None::<&()>, true)
            .await?;
        Ok(unwrap_list(&result, "modules"))
    }

    /// `GET /api/pulse/wasm-modules/{name}` — metadata for one module.
    pub async fn get(&self, name: &str) -> Result<Value, PulseError> {
        let path = format!("/api/pulse/wasm-modules/{}", encode_path(name));
        self.client
            .request(Method::GET, &path, None::<&()>, true)
            .await
    }

    /// `DELETE /api/pulse/wasm-modules/{name}` — remove a module (ADMIN).
    pub async fn delete(&self, name: &str) -> Result<(), PulseError> {
        let path = format!("/api/pulse/wasm-modules/{}", encode_path(name));
        self.client
            .request::<()>(Method::DELETE, &path, None, true)
            .await?;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// UsersResource — client.users()
// ---------------------------------------------------------------------------

pub struct UsersResource<'c> {
    pub(crate) client: &'c PulseClient,
}

impl UsersResource<'_> {
    /// `GET /api/pulse/users` — every user in the current org.
    ///
    /// Requires the caller to have the `USERS_LIST` permission atom (Owner /
    /// Platform Admin personas by default — see B-105).
    pub async fn list(&self) -> Result<Vec<Value>, PulseError> {
        let result = self
            .client
            .request(Method::GET, "/api/pulse/users", None::<&()>, true)
            .await?;
        Ok(unwrap_list(&result, "users"))
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Extracts a `Vec<Value>` from `result[key]`. Returns an empty Vec for
/// missing / malformed envelopes — never panics — so callers can iterate
/// safely.
fn unwrap_list(result: &Value, key: &str) -> Vec<Value> {
    result
        .get(key)
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default()
}

/// URL-encodes a path-param segment so ids containing `/`, spaces, etc.
/// round-trip safely. Uses the same character set as the `pulse-go`
/// `url.PathEscape` and `pulse-java` `URLEncoder` — `+` is encoded as `%20`.
fn encode_path(segment: &str) -> String {
    let mut out = String::with_capacity(segment.len());
    for b in segment.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(b as char)
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

/// Reads an unsigned LEB128 integer from `bytes` starting at `*pos`, advancing
/// `*pos` past the consumed bytes. Returns `None` on overrun (the caller maps
/// that to a [`PulseError::Validation`] "malformed WASM module").
fn read_uleb128(bytes: &[u8], pos: &mut usize) -> Option<u64> {
    let mut result: u64 = 0;
    let mut shift: u32 = 0;
    loop {
        let byte = *bytes.get(*pos)?;
        *pos += 1;
        result |= u64::from(byte & 0x7f) << shift;
        if byte & 0x80 == 0 {
            return Some(result);
        }
        shift += 7;
        // A uleb128 wider than 64 bits is malformed for our purposes.
        if shift >= 64 {
            return None;
        }
    }
}

/// Builds a client-side [`PulseError::Validation`] for the WASM upload path so
/// the caller sees the SDK's validation error (not a cryptic server 400) before
/// any network call.
fn wasm_validation_err(message: &str) -> PulseError {
    PulseError::Validation {
        path: "/api/pulse/wasm-modules".to_string(),
        body: Some(json!({ "error": message })),
    }
}

/// Client-side pre-upload validation of a WASM module's raw bytes — inspects the
/// binary (does NOT execute it) and mirrors the server's
/// `ChicoryWasmRunner.validateModule`.
///
/// A conforming sandbox module must: have the WASM magic + version 1 header,
/// import zero host functions, and export `alloc`, `process`, and `memory`.
/// Anything else is rejected with [`PulseError::Validation`] so the failure is
/// surfaced locally instead of as a server 400 / runtime trap.
pub fn validate_wasm_module(bytes: &[u8]) -> Result<(), PulseError> {
    if bytes.len() < 8 {
        return Err(wasm_validation_err("not a WASM module: too short"));
    }
    if bytes[0..4] != [0x00, b'a', b's', b'm'] || bytes[4..8] != [0x01, 0x00, 0x00, 0x00] {
        return Err(wasm_validation_err("not a WASM module (bad magic/version)"));
    }

    let malformed = || wasm_validation_err("malformed WASM module");

    let mut has_alloc = false;
    let mut has_process = false;
    let mut has_memory = false;

    let mut pos = 8usize;
    while pos < bytes.len() {
        // Section id (1 byte) + uleb128 payload size.
        let id = bytes[pos];
        pos += 1;
        let size = read_uleb128(bytes, &mut pos).ok_or_else(malformed)? as usize;
        let payload_start = pos;
        let payload_end = payload_start.checked_add(size).ok_or_else(malformed)?;
        if payload_end > bytes.len() {
            return Err(malformed());
        }

        match id {
            // Import section — any host import disqualifies the module.
            2 => {
                let mut p = payload_start;
                let count = read_uleb128(bytes, &mut p).ok_or_else(malformed)?;
                if count > 0 {
                    return Err(wasm_validation_err(
                        "WASM module imports host functions; it must be a pure sandbox \
                         (build with no WASI/host imports)",
                    ));
                }
            }
            // Export section — collect exported names.
            7 => {
                let mut p = payload_start;
                let count = read_uleb128(bytes, &mut p).ok_or_else(malformed)?;
                for _ in 0..count {
                    let name_len = read_uleb128(bytes, &mut p).ok_or_else(malformed)? as usize;
                    let name_end = p.checked_add(name_len).ok_or_else(malformed)?;
                    if name_end > payload_end {
                        return Err(malformed());
                    }
                    let name = &bytes[p..name_end];
                    p = name_end;
                    // 1 kind byte + uleb128 index.
                    if p >= payload_end {
                        return Err(malformed());
                    }
                    p += 1; // kind
                    read_uleb128(bytes, &mut p).ok_or_else(malformed)?; // index
                    match name {
                        b"alloc" => has_alloc = true,
                        b"process" => has_process = true,
                        b"memory" => has_memory = true,
                        _ => {}
                    }
                }
            }
            _ => {}
        }

        // Advance past the section payload regardless of id.
        pos = payload_end;
    }

    if !(has_alloc && has_process && has_memory) {
        return Err(wasm_validation_err(
            "WASM module must export alloc, process and memory",
        ));
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// ConnectorsResource — client.connectors() (B-093 follow-up: catalogue parity)
// ---------------------------------------------------------------------------

/// `client.connectors()` — the connector catalogue, the same list the Pipeline
/// Studio palette and `pulse connectors list` show. Each entry is
/// `{subType, displayName, configFields}`; use the `subType` as a sink/source
/// node `type` in a pipeline definition deployed via `client.pipelines()`.
/// Bridged connectors appear only when the enterprise bridge JAR is on the
/// server's classpath.
pub struct ConnectorsResource<'c> {
    pub(crate) client: &'c PulseClient,
}

impl ConnectorsResource<'_> {
    /// `GET /api/pulse/connectors` — `{"sources": [...], "sinks": [...]}`.
    pub async fn list(&self) -> Result<Value, PulseError> {
        self.client
            .request(Method::GET, "/api/pulse/connectors", None::<&()>, true)
            .await
    }

    /// Just the sink connectors.
    pub async fn sinks(&self) -> Result<Vec<Value>, PulseError> {
        Ok(unwrap_list(&self.list().await?, "sinks"))
    }

    /// Just the source connectors.
    pub async fn sources(&self) -> Result<Vec<Value>, PulseError> {
        Ok(unwrap_list(&self.list().await?, "sources"))
    }
}