fabric-platform 0.7.0

//! Rust client SDK for the Fabric platform.
//!
//! This is a thin wrapper around the Fabric REST API. Every method
//! corresponds to an endpoint under `/v1/...` on a running Fabric API.
//! For the higher-level resource-oriented design, see the TypeScript SDK
//! at `sdks/typescript/` — this crate is intentionally minimal and
//! returns `serde_json::Value` rather than typed models. A typed rewrite
//! is tracked in `specs/plans/031-rust-sdk-rewrite.md` and the Rust SDK
//! backlog in `specs/SPRINT.md`.

use reqwest::header::{HeaderMap, HeaderValue, AUTHORIZATION, CONTENT_TYPE};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::time::Duration;

#[cfg(not(target_arch = "wasm32"))]
use std::time::Instant;
#[cfg(target_arch = "wasm32")]
use web_time::Instant;

pub mod sse;
pub use sse::SseEvent;

/// Portable async sleep that works on both native (via tokio) and wasm32
/// (via gloo-timers / setTimeout).
async fn sleep(d: Duration) {
    #[cfg(not(target_arch = "wasm32"))]
    {
        tokio::time::sleep(d).await;
    }
    #[cfg(target_arch = "wasm32")]
    {
        gloo_timers::future::TimeoutFuture::new(d.as_millis() as u32).await;
    }
}

#[derive(Debug, thiserror::Error)]
pub enum FabricError {
    #[error("HTTP error: {0}")]
    Http(#[from] reqwest::Error),
    #[error("API error ({code}): {message}")]
    Api { code: String, message: String },
    #[error("{0}")]
    Other(String),
}

pub type Result<T> = std::result::Result<T, FabricError>;

// ── Workflow run output models ───────────────────────────────────────
//
// The SDK is otherwise serde_json::Value-typed, but variants is a core
// feature of the platform — consumers iterate per-variant outputs and
// artifacts on every run, so the run-output endpoint exposes typed
// structs for ergonomics. Other endpoints stay untyped pending the
// broader SDK rewrite tracked in plan 031.

/// Artifact produced by a workflow run.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunArtifact {
    pub id: String,
    pub run_id: String,
    pub asset_id: Option<String>,
    pub task_id: String,
    pub filename: String,
    pub content_type: Option<String>,
    pub produced_by: Option<String>,
    pub consumed_from: Option<Vec<String>>,
    pub metadata: Option<serde_json::Value>,
    pub artifact_type: String,
    /// Variant (0-indexed) that produced this artifact.
    #[serde(default)]
    pub variant_index: i16,
    pub created_at: Option<String>,
    pub download_url: Option<String>,
    pub download_url_expires_at: Option<String>,
}

/// Card layout shape for a variant — declared by the workflow's
/// `WorkflowOutput.kind`. Consumer UIs branch on this to pick the
/// right rendering. `None` means the workflow didn't declare a kind
/// and consumers should fall back to deriving from artifact MIME types.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum VariantKind {
    Video,
    Carousel,
    Image,
    Text,
}

/// A single variant's output and artifacts within a run.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VariantOutput {
    pub variant_index: i16,
    /// Sub-workflow that produced this entry. For bundle runs each
    /// entry has its own; for non-bundle runs every entry carries
    /// the run's single workflow_name. `None` on legacy data.
    #[serde(default)]
    pub workflow_name: Option<String>,
    /// Card shape for this variant, lifted from the workflow's
    /// `WorkflowOutput.kind`. `None` when the workflow didn't declare
    /// a kind.
    #[serde(default)]
    pub kind: Option<VariantKind>,
    pub output: Option<serde_json::Value>,
    #[serde(default)]
    pub artifacts: Vec<RunArtifact>,
}

/// One slot in a heterogeneous bundle submission.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BundleEntry {
    /// Sub-workflow name (must resolve via the registry).
    pub workflow: String,
    /// Per-sub input. Platform-level `_fabric_*` keys are injected by
    /// the engine; only domain inputs go here.
    pub input: serde_json::Value,
}

/// Result of `GET /v1/workflows/runs/{id}/output`.
///
/// Always exposes a uniform `outputs` array with one entry per variant
/// (default 1). Iterate `outputs` rather than branching on `variants`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunOutput {
    pub run_id: String,
    pub status: String,
    pub error: Option<String>,
    #[serde(default)]
    pub nodes: Vec<serde_json::Value>,
    #[serde(default = "default_variants")]
    pub variants: i16,
    #[serde(default)]
    pub outputs: Vec<VariantOutput>,
}

fn default_variants() -> i16 {
    1
}

impl RunOutput {
    /// Convenience for single-variant runs: ``outputs[0].output``.
    pub fn first_output(&self) -> Option<&serde_json::Value> {
        self.outputs.first().and_then(|v| v.output.as_ref())
    }

    /// Flat iterator over artifacts across every variant.
    pub fn all_artifacts(&self) -> impl Iterator<Item = &RunArtifact> {
        self.outputs.iter().flat_map(|v| v.artifacts.iter())
    }
}

/// Synchronous-style Fabric API client.
///
/// Configure scoping defaults (`organization_id`, `team_id`, `user_id`)
/// via [`set_organization_id`](Self::set_organization_id),
/// [`set_team_id`](Self::set_team_id), and
/// [`set_user_id`](Self::set_user_id). Methods like
/// [`list_runs`](Self::list_runs) auto-inject these as query params, and
/// per-call arguments override them.
pub struct FabricClient {
    client: reqwest::Client,
    base_url: String,
    organization_id: String,
    team_id: Option<String>,
    user_id: Option<String>,
}

impl FabricClient {
    /// Create a new client authenticated with an API key.
    pub fn new(base_url: &str, api_key: &str) -> Result<Self> {
        let mut headers = HeaderMap::new();
        headers.insert(
            AUTHORIZATION,
            HeaderValue::from_str(&format!("Bearer {api_key}"))
                .map_err(|e| FabricError::Other(e.to_string()))?,
        );
        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));

        let client = reqwest::Client::builder()
            .default_headers(headers)
            .build()?;

        Ok(Self {
            client,
            base_url: base_url.trim_end_matches('/').to_string(),
            organization_id: String::new(),
            team_id: None,
            user_id: None,
        })
    }

    /// Create a new client authenticated with a principal ID header.
    pub fn with_principal(base_url: &str, principal_id: &str) -> Result<Self> {
        let mut headers = HeaderMap::new();
        headers.insert(
            "X-Principal-Id",
            HeaderValue::from_str(principal_id).map_err(|e| FabricError::Other(e.to_string()))?,
        );
        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));

        let client = reqwest::Client::builder()
            .default_headers(headers)
            .build()?;

        Ok(Self {
            client,
            base_url: base_url.trim_end_matches('/').to_string(),
            organization_id: String::new(),
            team_id: None,
            user_id: None,
        })
    }

    /// Set the organization ID used for scoped requests (e.g. workflow runs).
    pub fn set_organization_id(&mut self, org_id: &str) {
        self.organization_id = org_id.to_string();
    }

    /// Set the default team ID used for scoped run queries.
    pub fn set_team_id(&mut self, team_id: Option<&str>) {
        self.team_id = team_id.map(str::to_string);
    }

    /// Set the default user ID (Fabric principal UUID matching
    /// `fabric_workflow_runs.created_by`) used for scoped run queries.
    pub fn set_user_id(&mut self, user_id: Option<&str>) {
        self.user_id = user_id.map(str::to_string);
    }

    /// Resolve an org ID from an explicit value or the client default.
    fn resolve_org_id(&self, org_id: Option<&str>) -> Result<String> {
        org_id
            .map(str::to_string)
            .or_else(|| {
                if self.organization_id.is_empty() {
                    None
                } else {
                    Some(self.organization_id.clone())
                }
            })
            .ok_or_else(|| {
                FabricError::Other(
                    "Organization ID required — pass it or set it on the client".to_string(),
                )
            })
    }

    // ── Private helpers ──────────────────────────────────────────────

    async fn request<T: serde::de::DeserializeOwned>(
        &self,
        method: reqwest::Method,
        path: &str,
        body: Option<serde_json::Value>,
    ) -> Result<T> {
        let url = format!("{}{path}", self.base_url);
        let mut req = self.client.request(method, &url);
        if let Some(b) = body {
            req = req.json(&b);
        }
        let resp = req.send().await?;
        let status = resp.status();
        let json: serde_json::Value = resp.json().await?;

        if let Some(err) = json.get("error") {
            return Err(FabricError::Api {
                code: status.as_u16().to_string(),
                message: err
                    .as_str()
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| err.to_string()),
            });
        }

        if let Some(data) = json.get("data") {
            serde_json::from_value(data.clone())
                .map_err(|e| FabricError::Other(format!("Failed to deserialize data: {e}")))
        } else {
            serde_json::from_value(json)
                .map_err(|e| FabricError::Other(format!("Failed to deserialize response: {e}")))
        }
    }

    async fn get<T: serde::de::DeserializeOwned>(&self, path: &str) -> Result<T> {
        self.request(reqwest::Method::GET, path, None).await
    }

    async fn post<T: serde::de::DeserializeOwned>(
        &self,
        path: &str,
        body: serde_json::Value,
    ) -> Result<T> {
        self.request(reqwest::Method::POST, path, Some(body)).await
    }

    async fn post_empty(&self, path: &str) -> Result<()> {
        let url = format!("{}{path}", self.base_url);
        let resp = self.client.post(&url).send().await?;
        let status = resp.status();
        let json: serde_json::Value = resp.json().await?;

        if let Some(err) = json.get("error") {
            return Err(FabricError::Api {
                code: status.as_u16().to_string(),
                message: err
                    .as_str()
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| err.to_string()),
            });
        }
        Ok(())
    }

    async fn delete_req(&self, path: &str) -> Result<()> {
        let url = format!("{}{path}", self.base_url);
        let resp = self.client.delete(&url).send().await?;
        let status = resp.status();
        let json: serde_json::Value = resp.json().await?;

        if let Some(err) = json.get("error") {
            return Err(FabricError::Api {
                code: status.as_u16().to_string(),
                message: err
                    .as_str()
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| err.to_string()),
            });
        }
        Ok(())
    }

    // ── System ───────────────────────────────────────────────────────

    pub async fn health_check(&self) -> Result<serde_json::Value> {
        self.get("/health").await
    }

    pub async fn system_status(&self) -> Result<serde_json::Value> {
        self.get("/v1/system/status").await
    }

    // ── Identity ─────────────────────────────────────────────────────

    pub async fn get_me(&self) -> Result<serde_json::Value> {
        self.get("/v1/me").await
    }

    pub async fn get_my_organizations(&self) -> Result<Vec<serde_json::Value>> {
        self.get("/v1/me/organizations").await
    }

    pub async fn get_my_teams(&self) -> Result<Vec<serde_json::Value>> {
        self.get("/v1/me/teams").await
    }

    pub async fn get_my_permissions(&self) -> Result<Vec<serde_json::Value>> {
        self.get("/v1/me/permissions").await
    }

    // ── Organizations ────────────────────────────────────────────────

    pub async fn create_organization(&self, slug: &str, name: &str) -> Result<serde_json::Value> {
        self.post("/v1/organizations", json!({ "slug": slug, "name": name }))
            .await
    }

    pub async fn list_organizations(&self) -> Result<Vec<serde_json::Value>> {
        self.get("/v1/organizations").await
    }

    /// Get an organization by ID. Falls back to the client's organization ID when `None`.
    pub async fn get_organization(&self, org_id: Option<&str>) -> Result<serde_json::Value> {
        let id = self.resolve_org_id(org_id)?;
        self.get(&format!("/v1/organizations/{id}")).await
    }

    /// List teams in an organization. Falls back to the client's organization ID when `None`.
    pub async fn list_org_teams(&self, org_id: Option<&str>) -> Result<Vec<serde_json::Value>> {
        let id = self.resolve_org_id(org_id)?;
        self.get(&format!("/v1/organizations/{id}/teams")).await
    }

    /// List members in an organization. Falls back to the client's organization ID when `None`.
    pub async fn list_org_members(&self, org_id: Option<&str>) -> Result<Vec<serde_json::Value>> {
        let id = self.resolve_org_id(org_id)?;
        self.get(&format!("/v1/organizations/{id}/members")).await
    }

    // ── Teams ────────────────────────────────────────────────────────

    /// Create a team. Falls back to the client's organization ID when `None`.
    pub async fn create_team(
        &self,
        org_id: Option<&str>,
        slug: &str,
        name: &str,
    ) -> Result<serde_json::Value> {
        let id = self.resolve_org_id(org_id)?;
        self.post(
            &format!("/v1/organizations/{id}/teams"),
            json!({ "slug": slug, "name": name }),
        )
        .await
    }

    pub async fn get_team(&self, team_id: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/teams/{team_id}")).await
    }

    // ── Invitations ──────────────────────────────────────────────────

    /// Create an invitation. Falls back to the client's organization ID when `None`.
    pub async fn create_invitation(
        &self,
        org_id: Option<&str>,
        email: &str,
        role: &str,
    ) -> Result<serde_json::Value> {
        let id = self.resolve_org_id(org_id)?;
        self.post(
            &format!("/v1/organizations/{id}/invitations"),
            json!({ "email": email, "role": role }),
        )
        .await
    }

    pub async fn accept_invitation(&self, invitation_id: &str) -> Result<()> {
        self.post_empty(&format!("/v1/invitations/{invitation_id}/accept"))
            .await
    }

    /// Revoke an outstanding invitation. (POST to `/revoke` — not DELETE.)
    pub async fn revoke_invitation(&self, invitation_id: &str) -> Result<()> {
        self.post_empty(&format!("/v1/invitations/{invitation_id}/revoke"))
            .await
    }

    // ── Authorization ────────────────────────────────────────────────

    pub async fn check_permission(&self, action: &str, resource: Option<&str>) -> Result<bool> {
        let mut body = json!({ "action": action });
        if let Some(r) = resource {
            body["resource"] = serde_json::Value::String(r.to_string());
        }
        let resp: serde_json::Value = self.post("/v1/authz/check", body).await?;
        Ok(resp
            .get("allowed")
            .and_then(|v| v.as_bool())
            .unwrap_or(false))
    }

    pub async fn check_permissions(
        &self,
        checks: Vec<serde_json::Value>,
    ) -> Result<Vec<serde_json::Value>> {
        self.post("/v1/authz/check-batch", json!({ "checks": checks }))
            .await
    }

    // ── API Keys ─────────────────────────────────────────────────────

    /// Create an API key. Falls back to the client's organization ID when `None`.
    pub async fn create_api_key(
        &self,
        name: &str,
        org_id: Option<&str>,
        scopes: Option<Vec<&str>>,
    ) -> Result<serde_json::Value> {
        let id = self.resolve_org_id(org_id)?;
        let mut body = json!({ "name": name, "organization_id": id });
        if let Some(s) = scopes {
            body["scopes"] = serde_json::Value::from(s);
        }
        self.post("/v1/api-keys", body).await
    }

    pub async fn list_api_keys(&self) -> Result<Vec<serde_json::Value>> {
        self.get("/v1/api-keys").await
    }

    pub async fn get_api_key(&self, key_id: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/api-keys/{key_id}")).await
    }

    pub async fn delete_api_key(&self, key_id: &str) -> Result<()> {
        self.delete_req(&format!("/v1/api-keys/{key_id}")).await
    }

    pub async fn disable_api_key(&self, key_id: &str) -> Result<()> {
        self.post_empty(&format!("/v1/api-keys/{key_id}/disable"))
            .await
    }

    pub async fn rotate_api_key(&self, key_id: &str) -> Result<serde_json::Value> {
        self.post(&format!("/v1/api-keys/{key_id}/rotate"), json!({}))
            .await
    }

    // ── Workflows (registry + runs) ──────────────────────────────────

    /// Create or update a workflow in the registry.
    ///
    /// Posts to `/v1/workflow-registry`. The `body` should match
    /// `CreateRegistryRequest` on the API side (name, language, source,
    /// entry_point, etc.).
    pub async fn upsert_workflow(&self, name: &str, body: serde_json::Value) -> Result<String> {
        let mut payload = body;
        payload["name"] = serde_json::Value::String(name.to_string());
        let resp: serde_json::Value = self.post("/v1/workflow-registry", payload).await?;
        resp.get("id")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string())
            .ok_or_else(|| FabricError::Other("Missing workflow id in response".to_string()))
    }

    /// List workflows in the registry (hierarchical: global > org > team).
    ///
    /// Always passes `limit=500` because the API's default of 50
    /// silently truncates the dropdown for any installation with more
    /// than ~50 registered workflows. The API caps the value at 200
    /// internally as of writing — passing 500 just means "give me as
    /// many as you can". A pagination-aware variant can be added if
    /// anyone needs to scroll past page one, but no current consumer
    /// does.
    pub async fn list_workflows(&self) -> Result<Vec<serde_json::Value>> {
        let mut params = vec![("limit", "500".to_string())];
        if !self.organization_id.is_empty() {
            params.push(("organization_id", self.organization_id.clone()));
            if let Some(team) = &self.team_id {
                params.push(("team_id", team.clone()));
            }
        }
        let qs = params
            .iter()
            .map(|(k, v)| format!("{k}={v}"))
            .collect::<Vec<_>>()
            .join("&");
        self.get(&format!("/v1/workflow-registry?{qs}")).await
    }

    /// Submit a workflow run by name.
    ///
    /// Posts to `POST /v1/workflows/run?name=<workflow_name>` and
    /// returns the resulting run ID immediately.
    pub async fn run_workflow(
        &self,
        workflow_name: &str,
        input: serde_json::Value,
    ) -> Result<String> {
        self.run_workflow_opts(workflow_name, input, false).await
    }

    /// Like [`run_workflow`](Self::run_workflow) with server-side input
    /// validation against the workflow's registered `input_schema`
    /// (plan 038 §5). Returns 400 with per-field errors if the payload
    /// doesn't conform. Workflows without a schema are unaffected.
    pub async fn run_workflow_validated(
        &self,
        workflow_name: &str,
        input: serde_json::Value,
    ) -> Result<String> {
        self.run_workflow_opts(workflow_name, input, true).await
    }

    /// Submit a workflow with `variants` parallel executions (1–10).
    ///
    /// `variants` is part of the workflow input contract — it's set as
    /// `input.variants` on the wire. This helper inserts the value
    /// into a clone of `input` for callers that prefer the named arg.
    /// The engine fans out N parallel runs and the run-output API
    /// always returns an `outputs: [{variant_index, output, artifacts}]`
    /// array, with one entry per variant.
    pub async fn run_workflow_with_variants(
        &self,
        workflow_name: &str,
        input: serde_json::Value,
        variants: u16,
    ) -> Result<String> {
        let mut input = input;
        if let Some(obj) = input.as_object_mut() {
            obj.entry("variants").or_insert_with(|| json!(variants));
        }
        self.run_workflow_opts(workflow_name, input, false).await
    }

    /// Submit a heterogeneous **bundle** — N different workflows
    /// running in parallel — and return the run id.
    ///
    /// Each [`BundleEntry`] runs as its own subprocess; the engine
    /// aggregates per-entry outputs into the run's `outputs` array.
    /// Use [`get_run_output`](Self::get_run_output) to fetch the typed
    /// [`RunOutput`] when the run completes; each `VariantOutput`
    /// carries the producing sub-`workflow_name` and (when declared)
    /// `kind`. For one creative brief that should produce multiple
    /// kinds of artifact (video + carousel + thread), this is the
    /// primitive — `run_workflow_with_variants` is N copies of one
    /// workflow, this is N different workflows.
    pub async fn run_bundle(&self, bundle: Vec<BundleEntry>) -> Result<String> {
        let input = json!({ "bundle": bundle });
        self.run_workflow_opts("bundle/ad-hoc", input, false).await
    }

    async fn run_workflow_opts(
        &self,
        workflow_name: &str,
        input: serde_json::Value,
        validate: bool,
    ) -> Result<String> {
        let mut path = format!("/v1/workflows/run?name={}", urlencode(workflow_name));
        if !self.organization_id.is_empty() {
            path.push_str(&format!("&organization_id={}", self.organization_id));
        }
        if let Some(team) = &self.team_id {
            path.push_str(&format!("&team_id={team}"));
        }
        if validate {
            path.push_str("&validate=true");
        }
        let resp: serde_json::Value = self.post(&path, json!({ "input": input })).await?;
        resp.get("id")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string())
            .ok_or_else(|| FabricError::Other("Missing run id in response".to_string()))
    }

    pub async fn get_run(&self, run_id: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/workflows/runs/{run_id}")).await
    }

    /// Fetch a run's final output, per-task timeline, and artifacts.
    ///
    /// Hits `GET /v1/workflows/runs/{id}/output`. Unlike `get_run`,
    /// which returns the bare `fabric_workflow_runs` row, this endpoint
    /// returns the workflow's terminal output (sourced from the canonical
    /// store, falling back to Sayiir's snapshot when the eager finalizer
    /// hasn't run yet) along with any artifacts the workflow registered.
    ///
    /// Returns a typed [`RunOutput`] whose `outputs` array always has
    /// `variants` entries (default 1). Each [`VariantOutput`] carries
    /// its `output` and the [`RunArtifact`]s produced by that variant,
    /// each with a signed `download_url`.
    ///
    /// `expires_in_secs` controls the signed URL TTL (server clamps to
    /// `[1, 86_400]`, defaults to 3600). `u32` is used so the generated
    /// wasm-bindgen type comes out as `number | null` in TypeScript
    /// rather than `bigint | null`.
    pub async fn get_run_output(
        &self,
        run_id: &str,
        expires_in_secs: Option<u32>,
    ) -> Result<RunOutput> {
        let path = match expires_in_secs {
            Some(ttl) => format!("/v1/workflows/runs/{run_id}/output?expires_in={ttl}"),
            None => format!("/v1/workflows/runs/{run_id}/output"),
        };
        self.get(&path).await
    }

    /// Fetch the input/output/task schemas registered for a workflow.
    ///
    /// Hits the dedicated `GET /v1/workflow-schemas/{name}` endpoint
    /// (plan 038 §3e) rather than pulling the whole registry entry.
    /// Returns `{ name, input_schema, output_schema, task_schemas,
    /// warnings }` — the same shape the TypeScript SDK's
    /// `workflows.registry.getSchemas()` returns.
    ///
    /// Workflows that don't declare Pydantic types come back with all
    /// three schema fields set to `null`. The call still succeeds, the
    /// consumer just learns there's no machine-readable contract.
    pub async fn get_workflow_schemas(&self, name: &str) -> Result<serde_json::Value> {
        let mut path = format!("/v1/workflow-schemas/{}", urlencode(name));
        if !self.organization_id.is_empty() {
            path.push_str(&format!("?organization_id={}", self.organization_id));
        }
        self.get(&path).await
    }

    pub async fn cancel_run(&self, run_id: &str) -> Result<()> {
        self.post_empty(&format!("/v1/workflows/runs/{run_id}/cancel"))
            .await
    }

    pub async fn pause_run(&self, run_id: &str) -> Result<()> {
        self.post_empty(&format!("/v1/workflows/runs/{run_id}/pause"))
            .await
    }

    pub async fn resume_run(&self, run_id: &str) -> Result<()> {
        self.post_empty(&format!("/v1/workflows/runs/{run_id}/resume"))
            .await
    }

    pub async fn wait_for_run(&self, run_id: &str) -> Result<serde_json::Value> {
        let timeout = Duration::from_secs(300);
        let poll_interval = Duration::from_secs(2);
        let start = Instant::now();

        loop {
            let run = self.get_run(run_id).await?;
            if let Some("completed" | "failed" | "cancelled") =
                run.get("status").and_then(|v| v.as_str())
            {
                return Ok(run);
            }
            if start.elapsed() >= timeout {
                return Err(FabricError::Other(format!(
                    "Timed out waiting for run {run_id} after {timeout:?}"
                )));
            }
            sleep(poll_interval).await;
        }
    }

    /// Submit a workflow, wait for it to finish, and return the full
    /// output with artifacts that have signed download URLs.
    ///
    /// Combines `run_workflow` + `wait_for_run` + `get_run_output` so
    /// callers get downloadable artifact URLs in a single call.
    ///
    /// `expires_in_secs` controls the signed URL TTL (default 3600,
    /// max 86400).
    pub async fn run_workflow_and_get_output(
        &self,
        workflow_name: &str,
        input: serde_json::Value,
        expires_in_secs: Option<u32>,
    ) -> Result<RunOutput> {
        let run_id = self.run_workflow(workflow_name, input).await?;
        self.wait_for_run(&run_id).await?;
        self.get_run_output(&run_id, expires_in_secs).await
    }

    /// Submit a bundle, wait for every sub-workflow to finish, and
    /// return the typed [`RunOutput`] with one entry per bundle entry.
    pub async fn run_bundle_and_get_output(
        &self,
        bundle: Vec<BundleEntry>,
        expires_in_secs: Option<u32>,
    ) -> Result<RunOutput> {
        let run_id = self.run_bundle(bundle).await?;
        self.wait_for_run(&run_id).await?;
        self.get_run_output(&run_id, expires_in_secs).await
    }

    /// Download an artifact's binary content using its signed download URL.
    ///
    /// Pass a `download_url` from the artifacts returned by `get_run_output`.
    /// Returns `None` if the URL is empty. The signed URL requires no auth
    /// headers — it is self-contained.
    pub async fn download_artifact(&self, download_url: &str) -> Result<Vec<u8>> {
        let resp = self
            .client
            .get(download_url)
            .send()
            .await
            .map_err(|e| FabricError::Other(format!("download request failed: {e}")))?;
        if !resp.status().is_success() {
            return Err(FabricError::Api {
                code: resp.status().as_u16().to_string(),
                message: format!("Failed to download artifact: {}", resp.status()),
            });
        }
        resp.bytes()
            .await
            .map(|b| b.to_vec())
            .map_err(|e| FabricError::Other(format!("download body read failed: {e}")))
    }

    /// List workflow runs scoped by org / team / creator.
    ///
    /// Each of `organization_id`, `team_id`, and `created_by` falls back
    /// to the client's configured defaults when `None`. To list across
    /// all users of a team without the client's default user filter,
    /// clear the default via [`set_user_id(None)`](Self::set_user_id)
    /// first.
    ///
    /// `created_by` must be the Fabric principal UUID stored in
    /// `fabric_workflow_runs.created_by` — the same UUID that the
    /// submitter's principal had at run submit time.
    pub async fn list_runs(
        &self,
        organization_id: Option<&str>,
        team_id: Option<&str>,
        created_by: Option<&str>,
        limit: Option<i64>,
        offset: Option<i64>,
    ) -> Result<Vec<serde_json::Value>> {
        let org = self.resolve_org_id(organization_id)?;

        let team = team_id.map(str::to_string).or_else(|| self.team_id.clone());
        let user = created_by
            .map(str::to_string)
            .or_else(|| self.user_id.clone());

        let mut path = format!("/v1/workflows/runs?organization_id={}", urlencode(&org));
        if let Some(t) = team.as_deref() {
            path.push_str(&format!("&team_id={}", urlencode(t)));
        }
        if let Some(u) = user.as_deref() {
            path.push_str(&format!("&created_by={}", urlencode(u)));
        }
        if let Some(l) = limit {
            path.push_str(&format!("&limit={l}"));
        }
        if let Some(o) = offset {
            path.push_str(&format!("&offset={o}"));
        }
        self.get(&path).await
    }

    /// List workflow runs waiting for a signal (pending approvals).
    pub async fn list_waiting_runs(
        &self,
        organization_id: Option<&str>,
        team_id: Option<&str>,
        created_by: Option<&str>,
    ) -> Result<Vec<serde_json::Value>> {
        let org = self.resolve_org_id(organization_id)?;

        let team = team_id.map(str::to_string).or_else(|| self.team_id.clone());
        let user = created_by
            .map(str::to_string)
            .or_else(|| self.user_id.clone());

        let mut path = format!(
            "/v1/workflows/runs/waiting?organization_id={}",
            urlencode(&org)
        );
        if let Some(t) = team.as_deref() {
            path.push_str(&format!("&team_id={}", urlencode(t)));
        }
        if let Some(u) = user.as_deref() {
            path.push_str(&format!("&created_by={}", urlencode(u)));
        }
        self.get(&path).await
    }

    /// Cancel a workflow run with an optional reason.
    pub async fn cancel_run_with_reason(&self, run_id: &str, reason: Option<&str>) -> Result<()> {
        let body = match reason {
            Some(r) => json!({ "reason": r }),
            None => json!({}),
        };
        let _: serde_json::Value = self
            .post(&format!("/v1/workflows/runs/{run_id}/cancel"), body)
            .await?;
        Ok(())
    }

    // ── Face Swap & Motion Transfer ─────────────────────────────────

    /// Run the `video/face-swap` workflow.
    ///
    /// Swaps a persona face onto a source image/video. Provide either
    /// `target_url` (direct face URL) or `persona_gallery_id` (to pull
    /// from an org gallery).
    pub async fn face_swap(
        &self,
        source_url: &str,
        target_url: Option<&str>,
        persona_gallery_id: Option<&str>,
    ) -> Result<serde_json::Value> {
        let mut input = json!({ "source_url": source_url });
        if let Some(url) = target_url {
            input["target_url"] = serde_json::Value::String(url.to_string());
        }
        if let Some(gid) = persona_gallery_id {
            input["persona_gallery_id"] = serde_json::Value::String(gid.to_string());
        }
        let run_id = self.run_workflow("video/face-swap", input).await?;
        self.wait_for_run(&run_id).await
    }

    /// Run the `video/motion-transfer` workflow.
    ///
    /// Animates a persona image using a reference video's motion (dance,
    /// gestures, expressions).
    pub async fn motion_transfer(
        &self,
        driving_video_url: &str,
        source_image_url: Option<&str>,
        persona_gallery_id: Option<&str>,
        motion_model: Option<&str>,
    ) -> Result<serde_json::Value> {
        let mut input = json!({ "driving_video_url": driving_video_url });
        if let Some(url) = source_image_url {
            input["source_image_url"] = serde_json::Value::String(url.to_string());
        }
        if let Some(gid) = persona_gallery_id {
            input["persona_gallery_id"] = serde_json::Value::String(gid.to_string());
        }
        if let Some(model) = motion_model {
            input["motion_model"] = serde_json::Value::String(model.to_string());
        }
        let run_id = self.run_workflow("video/motion-transfer", input).await?;
        self.wait_for_run(&run_id).await
    }

    // ── Providers ────────────────────────────────────────────────────

    pub async fn list_providers(&self) -> Result<Vec<serde_json::Value>> {
        self.get("/v1/providers").await
    }

    pub async fn execute_provider(&self, body: serde_json::Value) -> Result<serde_json::Value> {
        self.post("/v1/providers/execute", body).await
    }

    pub async fn estimate_cost(&self, body: serde_json::Value) -> Result<serde_json::Value> {
        self.post("/v1/providers/estimate", body).await
    }

    // ── Usage & Audit ────────────────────────────────────────────────

    /// Get usage summary. Falls back to the client's organization ID when `None`.
    pub async fn get_org_usage(&self, org_id: Option<&str>) -> Result<serde_json::Value> {
        let id = self.resolve_org_id(org_id)?;
        self.get(&format!("/v1/organizations/{id}/usage")).await
    }

    /// Get daily usage rollup. Falls back to the client's organization ID when `None`.
    pub async fn get_org_usage_daily(
        &self,
        org_id: Option<&str>,
    ) -> Result<Vec<serde_json::Value>> {
        let id = self.resolve_org_id(org_id)?;
        self.get(&format!("/v1/organizations/{id}/usage/daily"))
            .await
    }

    /// Get audit logs. Falls back to the client's organization ID when `None`.
    pub async fn get_org_audit_logs(&self, org_id: Option<&str>) -> Result<Vec<serde_json::Value>> {
        let id = self.resolve_org_id(org_id)?;
        self.get(&format!("/v1/organizations/{id}/audit-logs"))
            .await
    }

    pub async fn get_audit_logs(&self) -> Result<Vec<serde_json::Value>> {
        self.get("/v1/audit-logs").await
    }

    // ── Webhooks ─────────────────────────────────────────────────────

    /// Create a webhook subscription. Falls back to the client's organization ID when `None`.
    ///
    /// The signing secret is generated server-side and returned once in the
    /// response as `data.secret`. Store it securely — it cannot be retrieved again.
    pub async fn create_webhook(
        &self,
        org_id: Option<&str>,
        url: &str,
        events: Vec<&str>,
    ) -> Result<serde_json::Value> {
        let id = self.resolve_org_id(org_id)?;
        let body = json!({
            "url": url,
            "event_filter": events,
        });
        self.post(&format!("/v1/organizations/{id}/webhooks"), body)
            .await
    }

    /// List webhooks. Falls back to the client's organization ID when `None`.
    pub async fn list_webhooks(&self, org_id: Option<&str>) -> Result<Vec<serde_json::Value>> {
        let id = self.resolve_org_id(org_id)?;
        self.get(&format!("/v1/organizations/{id}/webhooks")).await
    }

    pub async fn get_webhook(&self, webhook_id: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/webhooks/{webhook_id}")).await
    }

    pub async fn delete_webhook(&self, webhook_id: &str) -> Result<()> {
        self.delete_req(&format!("/v1/webhooks/{webhook_id}")).await
    }

    // ── Secrets (org-scoped) ─────────────────────────────────────────

    /// Set a secret. Falls back to the client's organization ID when `None`.
    pub async fn set_secret(&self, org_id: Option<&str>, name: &str, value: &str) -> Result<()> {
        let id = self.resolve_org_id(org_id)?;
        let _: serde_json::Value = self
            .post(
                &format!("/v1/organizations/{id}/secrets"),
                json!({ "name": name, "value": value }),
            )
            .await?;
        Ok(())
    }

    /// List secrets. Falls back to the client's organization ID when `None`.
    pub async fn list_secrets(&self, org_id: Option<&str>) -> Result<Vec<String>> {
        let id = self.resolve_org_id(org_id)?;
        self.get(&format!("/v1/organizations/{id}/secrets")).await
    }

    /// Delete a secret. Falls back to the client's organization ID when `None`.
    pub async fn delete_secret(&self, org_id: Option<&str>, name: &str) -> Result<()> {
        let id = self.resolve_org_id(org_id)?;
        self.delete_req(&format!("/v1/organizations/{id}/secrets/{name}"))
            .await
    }

    // ── Schedules ────────────────────────────────────────────────────

    /// Create a schedule for a workflow definition.
    ///
    /// Note: schedules are still mounted at `/v1/workflow-definitions/{id}/schedules`
    /// in the current API, not `/v1/workflows/{id}/schedules`.
    pub async fn create_schedule(
        &self,
        workflow_definition_id: &str,
        cron: &str,
        input_context: Option<serde_json::Value>,
    ) -> Result<serde_json::Value> {
        let mut body = json!({ "cron_expression": cron });
        if let Some(ctx) = input_context {
            body["input_context"] = ctx;
        }
        self.post(
            &format!("/v1/workflow-definitions/{workflow_definition_id}/schedules"),
            body,
        )
        .await
    }

    pub async fn list_schedules(
        &self,
        workflow_definition_id: &str,
    ) -> Result<Vec<serde_json::Value>> {
        self.get(&format!(
            "/v1/workflow-definitions/{workflow_definition_id}/schedules"
        ))
        .await
    }

    pub async fn delete_schedule(&self, schedule_id: &str) -> Result<()> {
        self.delete_req(&format!("/v1/schedules/{schedule_id}"))
            .await
    }
}

/// Minimal URL-component encoder for query values.
///
/// Handles the characters that actually appear in Fabric IDs and names:
/// spaces, `&`, `=`, `?`, `#`, `+`, `/`, and a few others. This is not
/// a full percent-encoder — a proper implementation is tracked in the
/// Rust SDK backlog.
fn urlencode(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for b in s.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('%');
                out.push_str(&format!("{b:02X}"));
            }
        }
    }
    out
}