videosdk-server-sdk 0.1.0

//! Dispatches deployed agents into rooms, and manages cloud deployments.

use std::collections::HashMap;
use std::sync::Arc;

use futures_util::Stream;
use reqwest::Method;
use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};

use crate::client::{CallOptions, Client};
use crate::common::{string_enum, MessageResponse};
use crate::error::{Error, Result};
use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
use crate::query::QueryBuilder;
use crate::resources::rooms::RoomWebhook;
use crate::resources::{escape, random_uuid, slugify_name};

const DEPLOYMENTS_PATH: &str = "/ai/v1/cloud/deployments";
const VERSIONS_PATH: &str = "/ai/v1/cloud/versions";
const SECRETS_PATH: &str = "/ai/v1/cloud/secrets";

/* -------------------------------- dispatch ------------------------------------ */

/// Per-dispatch room behavior.
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AgentRoomOptions {
    /// Whether to end the session when the agent leaves.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub auto_end_session: Option<bool>,
    /// How long the session may run, in seconds.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_timeout_seconds: Option<u32>,
    /// Whether the agent runs in playground mode.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub playground: Option<bool>,
    /// Whether the agent receives video.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub vision: Option<bool>,
    /// Whether the agent joins the meeting.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub join_meeting: Option<bool>,
}

/// The parameters for [`AgentsResource::dispatch`] and
/// [`connect`](AgentsResource::connect).
#[derive(Debug, Clone, Default)]
pub struct DispatchAgentParams {
    /// The deployed agent to dispatch. Required.
    pub agent_id: Option<String>,
    /// The room to join. One of `meeting_id` or `room_id` is required.
    pub meeting_id: Option<String>,
    /// An alias of `meeting_id`.
    pub room_id: Option<String>,
    /// Per-dispatch room behavior.
    pub room_options: Option<AgentRoomOptions>,
    /// When present, also places an outbound SIP call as part of the dispatch.
    pub sip_options: Option<Map<String, Value>>,
    /// Free-form metadata handed to the agent.
    pub metadata: Option<Map<String, Value>>,
    /// Pins a specific deployment version.
    pub version_id: Option<String>,
    /// Resolves a no-code published version by tag.
    pub version_tag: Option<String>,
    /// Forwarded to the registry when set.
    pub wait: Option<Value>,
}

/// The parameters for [`AgentsResource::general_dispatch`], a managed or
/// low-code dispatch. Unlike the others, this ignores `version_tag` and `wait`.
#[derive(Debug, Clone, Default)]
pub struct GeneralDispatchAgentParams {
    /// The deployed agent to dispatch.
    pub agent_id: Option<String>,
    /// The room to join. One of `meeting_id` or `room_id` is required.
    pub meeting_id: Option<String>,
    /// An alias of `meeting_id`.
    pub room_id: Option<String>,
    /// Per-dispatch room behavior.
    pub room_options: Option<AgentRoomOptions>,
    /// When present, also places an outbound SIP call.
    pub sip_options: Option<Map<String, Value>>,
    /// Free-form metadata handed to the agent.
    pub metadata: Option<Map<String, Value>>,
    /// The deployment version. Required.
    pub version_id: Option<String>,
}

impl From<GeneralDispatchAgentParams> for DispatchAgentParams {
    fn from(params: GeneralDispatchAgentParams) -> Self {
        Self {
            agent_id: params.agent_id,
            meeting_id: params.meeting_id,
            room_id: params.room_id,
            room_options: params.room_options,
            sip_options: params.sip_options,
            metadata: params.metadata,
            version_id: params.version_id,
            version_tag: None,
            wait: None,
        }
    }
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct DispatchWire {
    #[serde(skip_serializing_if = "Option::is_none")]
    agent_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    room_options: Option<AgentRoomOptions>,
    #[serde(skip_serializing_if = "Option::is_none")]
    sip_options: Option<Map<String, Value>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    metadata: Option<Map<String, Value>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    version_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    version_tag: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    wait: Option<Value>,
    /// Always sent, even though the caller may have supplied only `room_id`.
    meeting_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    room_id: Option<String>,
}

/// The result of dispatching an agent.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DispatchResult {
    /// Whether the dispatch was accepted.
    pub success: Option<bool>,
    /// The dispatch job id.
    pub job_id: Option<String>,
    /// The dispatch status.
    pub status: Option<String>,
    /// The worker that picked up the job.
    pub worker_id: Option<String>,
    /// The room the agent joined.
    pub room_id: Option<String>,
    /// The agent that was dispatched.
    pub agent_id: Option<String>,
    /// Cloud-deployment details.
    pub cloud: Option<Value>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// The registry configuration a worker uses to initialize.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AgentInitConfig {
    /// The agent registry URL.
    pub registry_url: String,
}

/* ------------------------------- deployments ---------------------------------- */

string_enum! {
    /// The compute profile of a deployment version.
    AgentComputeProfile {
        /// A small CPU profile. The default.
        CPU_SMALL => "cpu-small",
        /// A medium CPU profile.
        CPU_MEDIUM => "cpu-medium",
        /// A large CPU profile.
        CPU_LARGE => "cpu-large",
    }
}

/// A cloud agent deployment.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AgentDeployment {
    /// The deployment id.
    pub id: String,
    /// The agent this deployment belongs to.
    pub agent_id: Option<String>,
    /// The deployment's display name.
    pub name: Option<String>,
    /// The template the deployment was created from.
    pub template: Option<String>,
    /// The deployment's status.
    pub status: Option<String>,
    /// The attached env-secret id.
    pub env_secret: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// A deployment version.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AgentDeploymentVersion {
    /// The version id.
    pub id: String,
    /// The deployment this version belongs to.
    pub deployment_id: Option<String>,
    /// The agent this version belongs to.
    pub agent_id: Option<String>,
    /// The version's display name.
    pub name: Option<String>,
    /// The container image.
    pub image: Option<Value>,
    /// The region the version runs in.
    pub region: Option<String>,
    /// The minimum replica count.
    pub min_replica: Option<u32>,
    /// The maximum replica count.
    pub max_replica: Option<u32>,
    /// The version's status.
    pub status: Option<String>,
    /// Whether this version is active.
    pub active: Option<bool>,
    /// The version tag.
    pub version_tag: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// A single console-log entry.
#[derive(Debug, Clone, Deserialize)]
pub struct AgentLogEntry {
    /// The log line.
    pub log: String,
    /// When the line was emitted.
    pub timestamp: Option<String>,
    /// Structured metadata attached to the line.
    pub metadata: Option<Map<String, Value>>,
}

/// A container image for a deployment version.
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DeployImage {
    /// The full image reference, e.g. `registry.videosdk.live/acme/agent:1.4`.
    pub image_url: String,
    /// The registry host. Parsed out of `image_url` when absent.
    #[serde(rename = "imageCR", skip_serializing_if = "Option::is_none")]
    pub image_cr: Option<String>,
    /// One of `public` (the default), `private` — which needs an
    /// `image_pull_secret` — or `managed`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub image_type: Option<String>,
    /// The pull-secret id, for a private registry.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub image_pull_secret: Option<String>,
}

/// A container image for a deployment: either a reference string, or a full
/// [`DeployImage`].
///
/// ```
/// # use videosdk::{AgentImage, DeployImage};
/// let image: AgentImage = "registry.videosdk.live/acme/agent:1.4".into();
/// let private: AgentImage = DeployImage {
///     image_url: "registry.example.com/agent".into(),
///     image_pull_secret: Some("secret-1".into()),
///     ..Default::default()
/// }
/// .into();
/// ```
#[derive(Debug, Clone)]
pub enum AgentImage {
    /// An image reference, e.g. `registry.videosdk.live/acme/agent:1.4`.
    Reference(String),
    /// A full image object, for a private or managed registry.
    Full(DeployImage),
}

impl From<&str> for AgentImage {
    fn from(reference: &str) -> Self {
        AgentImage::Reference(reference.to_string())
    }
}

impl From<String> for AgentImage {
    fn from(reference: String) -> Self {
        AgentImage::Reference(reference)
    }
}

impl From<DeployImage> for AgentImage {
    fn from(image: DeployImage) -> Self {
        AgentImage::Full(image)
    }
}

/// The replica scaling of a deployment.
#[derive(Debug, Clone, Default)]
pub struct AgentScaling {
    /// The minimum replica count.
    pub min: Option<u32>,
    /// The maximum replica count. Must be at most 50, and at least `min`. The
    /// server rejects values outside that range rather than clamping them.
    pub max: Option<u32>,
}

/// The parameters for [`AgentDeploymentResource::create`].
#[derive(Debug, Clone, Default)]
pub struct DeployAgentParams {
    /// The deployment name. Required.
    pub name: String,
    /// The container image. Required in practice.
    pub image: Option<AgentImage>,
    /// Environment variables for the agent, stored as a secret attached to the
    /// deployment.
    pub env: HashMap<String, String>,
    /// The replica scaling.
    pub scaling: Option<AgentScaling>,
    /// The compute profile. Defaults to `cpu-small`.
    pub profile: Option<AgentComputeProfile>,
    /// The region. Defaults to `us002`.
    pub region: Option<String>,
    /// Attaches to an existing agent. A new one is generated when absent.
    pub agent_id: Option<String>,
    /// The version tag.
    pub version_tag: Option<String>,
    /// The session webhook. Its `events` list is effectively required.
    pub webhook: Option<RoomWebhook>,
    /// Names the generated env secret. Derived from `name` when absent.
    pub env_secret_name: Option<String>,
}

/// The result of [`AgentDeploymentResource::create`].
#[derive(Debug, Clone)]
pub struct DeployedAgent {
    /// The deployment that was created.
    pub deployment: AgentDeployment,
    /// The version created, and auto-activated, for this deploy.
    pub version: AgentDeploymentVersion,
    /// The env-secret id, when `env` was provided.
    pub secret_id: Option<String>,
}

/// Infers `image_cr` and `image_type` when the caller did not set them.
fn normalize_deploy_image(image: Option<AgentImage>) -> DeployImage {
    let mut image = match image {
        None => DeployImage::default(),
        Some(AgentImage::Reference(reference)) => DeployImage {
            image_url: reference,
            ..Default::default()
        },
        Some(AgentImage::Full(image)) => image,
    };

    if image.image_cr.is_none() {
        let first_segment = image.image_url.split('/').next().unwrap_or("");
        // A registry host carries a `.` or a `:`; anything else is a Docker Hub
        // short reference.
        image.image_cr = Some(
            if !first_segment.is_empty() && first_segment.contains(['.', ':']) {
                first_segment.to_string()
            } else {
                "docker.io".to_string()
            },
        );
    }

    if image.image_type.is_none() {
        image.image_type = Some(
            if image.image_pull_secret.is_some() {
                "private"
            } else {
                "public"
            }
            .to_string(),
        );
    }
    image
}

/// The query parameters for [`AgentDeploymentResource::list`].
#[derive(Debug, Clone, Default)]
pub struct ListDeploymentsParams {
    /// The 1-based page number.
    pub page: Option<u32>,
    /// Items per page.
    pub per_page: Option<u32>,
    /// An opaque cursor from a previous page.
    pub cursor: Option<String>,
    /// Filters by agent id.
    pub agent_id: Option<String>,
    /// A regex match on the deployment name.
    pub name: Option<String>,
    /// Filters by deployment status.
    pub status: Option<String>,
    /// Filters by deployment id.
    pub deployment_id: Option<String>,
}

/// The query parameters for [`AgentDeploymentResource::logs`].
#[derive(Debug, Clone, Default)]
pub struct ListDeploymentLogsParams {
    /// The 1-based page number.
    pub page: Option<u32>,
    /// Items per page.
    pub per_page: Option<u32>,
    /// An opaque cursor from a previous page.
    pub cursor: Option<String>,
    /// Filters by agent id.
    pub agent_id: Option<String>,
    /// Filters by version id.
    pub version_id: Option<String>,
    /// Filters by session id.
    pub session_id: Option<String>,
    /// Filters by user id.
    pub user_id: Option<String>,
    /// Filters by room id.
    pub room_id: Option<String>,
    /// The start of the query window, as an RFC 3339 timestamp.
    ///
    /// The server parses this with `new Date(...)`, which yields `NaN` for a
    /// stringified epoch, so a formatted timestamp is required.
    pub start: Option<String>,
    /// The end of the query window, as an RFC 3339 timestamp.
    pub end: Option<String>,
    /// The sort order: `1` ascending, `-1` descending.
    pub sort: Option<String>,
}

macro_rules! pagination {
    ($name:ident) => {
        impl $name {
            fn pagination(&self) -> ListParams {
                ListParams {
                    page: self.page,
                    per_page: self.per_page,
                    cursor: self.cursor.clone(),
                }
            }
        }
    };
}

pagination!(ListDeploymentsParams);
pagination!(ListDeploymentLogsParams);

/// Cloud agent deployments. Reached via [`AgentsResource::deployment`].
#[derive(Debug, Clone, Copy)]
pub struct AgentDeploymentResource<'a> {
    client: &'a Client,
}

impl<'a> AgentDeploymentResource<'a> {
    /// Lists cloud agent deployments, one page at a time.
    pub async fn list(&self, params: ListDeploymentsParams) -> Result<Page<AgentDeployment>> {
        paginate(
            self.fetcher(&params),
            &params.pagination(),
            "deployments",
            None,
        )
        .await
    }

    /// Lists deployments, transparently fetching every page.
    pub fn list_stream(
        &self,
        params: ListDeploymentsParams,
    ) -> impl Stream<Item = Result<AgentDeployment>> + Send {
        auto_page(
            self.fetcher(&params),
            params.pagination(),
            "deployments",
            None,
        )
    }

    /// Fetches a single deployment by id.
    ///
    /// There is no get-by-id route, so this filters the list.
    pub async fn get(&self, deployment_id: &str) -> Result<AgentDeployment> {
        let page = self
            .list(ListDeploymentsParams {
                deployment_id: Some(deployment_id.to_string()),
                per_page: Some(1),
                ..Default::default()
            })
            .await?;

        page.data
            .iter()
            .find(|deployment| deployment.id == deployment_id)
            .or_else(|| page.data.first())
            .cloned()
            .ok_or_else(|| Error::not_found(format!("agent deployment {deployment_id} not found")))
    }

    /// Fetches the latest version of a deployment.
    pub async fn get_latest_version(&self, deployment_id: &str) -> Result<AgentDeploymentVersion> {
        let path = format!("{DEPLOYMENTS_PATH}/{}/latest", escape(deployment_id));
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

    /// Lists a deployment's console logs, one page at a time.
    ///
    /// This is a historical query — the last 24 hours by default — not a live
    /// stream.
    pub async fn logs(
        &self,
        deployment_id: &str,
        params: ListDeploymentLogsParams,
    ) -> Result<Page<AgentLogEntry>> {
        let fetcher = self.logs_fetcher(deployment_id, &params);
        paginate(fetcher, &params.pagination(), "logs", None).await
    }

    /// Lists a deployment's console logs, transparently fetching every page.
    pub fn logs_stream(
        &self,
        deployment_id: &str,
        params: ListDeploymentLogsParams,
    ) -> impl Stream<Item = Result<AgentLogEntry>> + Send {
        let fetcher = self.logs_fetcher(deployment_id, &params);
        auto_page(fetcher, params.pagination(), "logs", None)
    }

    /// Soft-deletes a deployment.
    pub async fn delete(&self, deployment_id: &str, force: bool) -> Result<MessageResponse> {
        let path = format!("{DEPLOYMENTS_PATH}/{}/delete", escape(deployment_id));
        let body = json!({ "force": force });
        self.client
            .json(Method::POST, &path, CallOptions::json(&body)?)
            .await
    }

    /// Deploys a pre-built agent image.
    ///
    /// This makes **three calls**: a secret holding `env` (only when `env` is
    /// non-empty), the deployment, then its auto-activated version.
    ///
    /// It is **not transactional**. If a later call fails, the earlier artifacts
    /// already exist — clean up with [`delete`](AgentDeploymentResource::delete).
    pub async fn create(&self, params: DeployAgentParams) -> Result<DeployedAgent> {
        let image = normalize_deploy_image(params.image);

        // 1. The env secret, when there are environment variables to store.
        let mut secret_id = None;
        if !params.env.is_empty() {
            let name = params.env_secret_name.clone().unwrap_or_else(|| {
                let suffix: String = random_uuid().chars().take(8).collect();
                format!("{}-env-{suffix}", slugify_name(&params.name))
            });
            let body = json!({"name": name, "keys": params.env, "type": "NORMAL"});

            #[derive(Deserialize)]
            struct Secret {
                id: String,
            }
            let secret: Secret = self
                .client
                .data(Method::POST, SECRETS_PATH, CallOptions::json(&body)?)
                .await?;
            secret_id = Some(secret.id);
        }

        // 2. The deployment. Send `agentId`/`envSecret` only when set: the server
        // auto-generates an agent id via `agentId ?? generateAgentId()`, which an
        // explicit empty string would defeat — and the schema then rejects it.
        let mut deploy_body = Map::new();
        deploy_body.insert("name".into(), json!(params.name));
        if let Some(agent_id) = params.agent_id.as_deref().filter(|id| !id.is_empty()) {
            deploy_body.insert("agentId".into(), json!(agent_id));
        }
        if let Some(secret_id) = &secret_id {
            deploy_body.insert("envSecret".into(), json!(secret_id));
        }
        let deployment: AgentDeployment = self
            .client
            .json(
                Method::POST,
                DEPLOYMENTS_PATH,
                CallOptions::json(&deploy_body)?,
            )
            .await?;

        // 3. The version, which the server activates for us.
        let scaling = params.scaling.unwrap_or_default();
        let version_body = VersionWire {
            agent_id: deployment.agent_id.as_deref(),
            deployment_id: Some(&deployment.id),
            image,
            profile: params.profile.unwrap_or(AgentComputeProfile::CPU_SMALL),
            min_replica: scaling.min,
            max_replica: scaling.max,
            region: params.region.as_deref(),
            version_tag: params.version_tag.as_deref(),
            webhook: params.webhook.as_ref(),
        };
        let version: AgentDeploymentVersion = self
            .client
            .json(
                Method::POST,
                VERSIONS_PATH,
                CallOptions::json(&version_body)?,
            )
            .await?;

        Ok(DeployedAgent {
            deployment,
            version,
            secret_id,
        })
    }

    /// Re-points a deployment at a version, or stops it.
    ///
    /// `activate: true` rolls onto `version_id`; `activate: false` stops the
    /// running version. Not needed after
    /// [`create`](AgentDeploymentResource::create), which already activates.
    pub async fn set_version_active(
        &self,
        version_id: &str,
        activate: bool,
        force: bool,
    ) -> Result<MessageResponse> {
        let path = format!("{VERSIONS_PATH}/{}/activate", escape(version_id));
        let body = json!({"activate": activate, "force": force});
        self.client
            .json(Method::PUT, &path, CallOptions::json(&body)?)
            .await
    }

    fn fetcher(&self, params: &ListDeploymentsParams) -> PageFetcher {
        let client = self.client.clone();
        let params = params.clone();
        Arc::new(move |page, per_page| {
            let client = client.clone();
            let params = params.clone();
            Box::pin(async move {
                let query = QueryBuilder::new()
                    .opt("page", page)
                    .opt("perPage", per_page)
                    .opt_str("agentId", params.agent_id.as_deref())
                    .opt_str("name", params.name.as_deref())
                    .opt_str("status", params.status.as_deref())
                    .opt_str("deploymentId", params.deployment_id.as_deref())
                    .into_pairs();
                client
                    .json::<Value>(
                        Method::GET,
                        DEPLOYMENTS_PATH,
                        CallOptions::new().query(query),
                    )
                    .await
            })
        })
    }

    fn logs_fetcher(&self, deployment_id: &str, params: &ListDeploymentLogsParams) -> PageFetcher {
        let client = self.client.clone();
        let params = params.clone();
        let path = format!("{DEPLOYMENTS_PATH}/{}/console-logs", escape(deployment_id));
        Arc::new(move |page, per_page| {
            let client = client.clone();
            let params = params.clone();
            let path = path.clone();
            Box::pin(async move {
                let query = QueryBuilder::new()
                    .opt("page", page)
                    .opt("perPage", per_page)
                    .opt_str("agentId", params.agent_id.as_deref())
                    .opt_str("versionId", params.version_id.as_deref())
                    .opt_str("sessionId", params.session_id.as_deref())
                    .opt_str("userId", params.user_id.as_deref())
                    .opt_str("roomId", params.room_id.as_deref())
                    .opt_str("start", params.start.as_deref())
                    .opt_str("end", params.end.as_deref())
                    .opt_str("sort", params.sort.as_deref())
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, &path, CallOptions::new().query(query))
                    .await
            })
        })
    }
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct VersionWire<'a> {
    #[serde(skip_serializing_if = "Option::is_none")]
    agent_id: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    deployment_id: Option<&'a str>,
    image: DeployImage,
    profile: AgentComputeProfile,
    #[serde(skip_serializing_if = "Option::is_none")]
    min_replica: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    max_replica: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    region: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    version_tag: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    webhook: Option<&'a RoomWebhook>,
}

/* ---------------------------------- facade ------------------------------------ */

/// Dispatches deployed agents into rooms. Reached via [`Client::agents`].
#[derive(Debug, Clone, Copy)]
pub struct AgentsResource<'a> {
    client: &'a Client,
}

impl<'a> AgentsResource<'a> {
    pub(crate) fn new(client: &'a Client) -> Self {
        Self { client }
    }

    /// Cloud agent deployments.
    pub fn deployment(&self) -> AgentDeploymentResource<'a> {
        AgentDeploymentResource {
            client: self.client,
        }
    }

    /// Dispatches an agent into a room.
    pub async fn dispatch(&self, params: DispatchAgentParams) -> Result<DispatchResult> {
        self.send("/v2/agent/dispatch", params).await
    }

    /// An alias of [`dispatch`](AgentsResource::dispatch), on its own route.
    pub async fn connect(&self, params: DispatchAgentParams) -> Result<DispatchResult> {
        self.send("/v2/agent/connect", params).await
    }

    /// Dispatches a managed or low-code agent. Requires `version_id`.
    pub async fn general_dispatch(
        &self,
        params: GeneralDispatchAgentParams,
    ) -> Result<DispatchResult> {
        self.send("/v2/agent/general/dispatch", params.into()).await
    }

    /// Fetches the agent registry URL a worker uses to initialize.
    pub async fn init_config(&self) -> Result<AgentInitConfig> {
        self.client
            .data(Method::POST, "/v2/agent/init-config", CallOptions::new())
            .await
    }

    async fn send(&self, path: &str, params: DispatchAgentParams) -> Result<DispatchResult> {
        // The two ids alias each other; the server wants both populated.
        let meeting_id = params
            .meeting_id
            .as_deref()
            .or(params.room_id.as_deref())
            .filter(|id| !id.is_empty())
            .ok_or_else(|| Error::validation("agents.dispatch() requires meeting_id (or room_id)"))?
            .to_string();
        let room_id = params
            .room_id
            .filter(|id| !id.is_empty())
            .unwrap_or_else(|| meeting_id.clone());

        let body = DispatchWire {
            agent_id: params.agent_id,
            room_options: params.room_options,
            sip_options: params.sip_options,
            metadata: params.metadata,
            version_id: params.version_id,
            version_tag: params.version_tag,
            wait: params.wait,
            meeting_id,
            room_id: Some(room_id),
        };
        self.client
            .data(Method::POST, path, CallOptions::json(&body)?)
            .await
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_registry_host_is_parsed_out_of_the_image_url() {
        let image = normalize_deploy_image(Some("registry.videosdk.live/acme/agent:1.4".into()));
        assert_eq!(image.image_cr.as_deref(), Some("registry.videosdk.live"));
        assert_eq!(image.image_type.as_deref(), Some("public"));

        // A host with a port also counts.
        let image = normalize_deploy_image(Some("localhost:5000/agent".into()));
        assert_eq!(image.image_cr.as_deref(), Some("localhost:5000"));
    }

    #[test]
    fn a_short_reference_falls_back_to_docker_hub() {
        for reference in ["acme/agent:1.4", "agent", "library/agent"] {
            let image = normalize_deploy_image(Some(reference.into()));
            assert_eq!(image.image_cr.as_deref(), Some("docker.io"), "{reference}");
        }
    }

    #[test]
    fn a_pull_secret_makes_the_image_private() {
        let image = normalize_deploy_image(Some(
            DeployImage {
                image_url: "registry.example.com/agent".into(),
                image_pull_secret: Some("secret-1".into()),
                ..Default::default()
            }
            .into(),
        ));
        assert_eq!(image.image_type.as_deref(), Some("private"));
        assert_eq!(image.image_url, "registry.example.com/agent");
    }

    #[test]
    fn an_explicit_image_object_is_never_overwritten() {
        let image = normalize_deploy_image(Some(
            DeployImage {
                image_url: "registry.example.com/agent".into(),
                image_cr: Some("custom.registry".into()),
                image_type: Some("managed".into()),
                image_pull_secret: None,
            }
            .into(),
        ));
        assert_eq!(image.image_url, "registry.example.com/agent");
        assert_eq!(image.image_cr.as_deref(), Some("custom.registry"));
        assert_eq!(image.image_type.as_deref(), Some("managed"));
    }
}