apify_client/clients/

actor.rs

//! Client for a single Actor (`/v2/actors/{actorId}`).

use serde::Serialize;
use serde_json::Value;

use crate::client::ApifyClient;
use crate::clients::actor_version::ActorVersionClient;
use crate::clients::actor_version_collection::ActorVersionCollectionClient;
use crate::clients::base::{
    delete_resource, get_resource, post_with_body, update_resource, ResourceContext,
};
use crate::clients::build::BuildClient;
use crate::clients::build_collection::BuildCollectionClient;
use crate::clients::run::{LastRunOptions, RunClient};
use crate::clients::run_collection::RunCollectionClient;
use crate::clients::webhook_collection::WebhookCollectionClient;
use crate::common::{parse_data_envelope, QueryParams};
use crate::error::ApifyClientResult;
use crate::http_client::HttpClient;
use crate::models::{Actor, ActorRun, Build};

/// How long to keep retrying a run start that the API rejects for lack of resources (see
/// [`ActorStartOptions::wait_for_resources`]).
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum WaitForResources {
    /// Retry until the run starts, how ever long that takes.
    Forever,
    /// Retry for up to this many seconds (clamped to `0` if negative), then return the last
    /// error. `0` makes a single attempt and returns its error without retrying.
    ForSecs(f64),
}

/// Options shared by [`ActorClient::start`] and [`ActorClient::call`] (and the task
/// equivalents).
#[derive(Debug, Default, Clone)]
pub struct ActorStartOptions {
    /// Tag or number of the build to run (e.g. `latest`, `0.1.2`).
    pub build: Option<String>,
    /// Memory in megabytes allocated for the run.
    pub memory_mbytes: Option<i64>,
    /// Timeout for the run in seconds (`0` means no timeout).
    pub timeout_secs: Option<i64>,
    /// Maximum seconds to wait server-side for the run to finish (max 60).
    pub wait_for_finish: Option<i64>,
    /// Maximum number of dataset items to charge (pay-per-result Actors).
    pub max_items: Option<i64>,
    /// Maximum total charge in USD (pay-per-event Actors).
    pub max_total_charge_usd: Option<f64>,
    /// Content type of the input body. Defaults to `application/json`.
    pub content_type: Option<String>,
    /// Whether to restart the run if it fails.
    pub restart_on_error: Option<bool>,
    /// Override the Actor's permission level for this run.
    pub force_permission_level: Option<String>,
    /// Ad-hoc webhooks to attach to this run. Serialized to base64-encoded JSON as the
    /// `webhooks` query parameter, matching the reference clients.
    pub webhooks: Option<Vec<serde_json::Value>>,
    /// Retry the start while the account lacks the resources for the run, i.e. while the API
    /// rejects it with an [`ApiError`](crate::ApiError) whose `error_type` is
    /// `actor-memory-limit-exceeded` or `concurrent-runs-limit-exceeded`. Both clear as other
    /// runs or builds finish. The start is retried every 10 seconds; any other error is returned
    /// right away.
    ///
    /// [`WaitForResources::Forever`] retries until the run starts.
    /// [`WaitForResources::ForSecs`] stops retrying after that many seconds and returns the last
    /// error. `None` (the default) makes a single attempt.
    ///
    /// A run that requests more memory than the account's whole memory limit is rejected with
    /// `actor-memory-limit-exceeded` as well and never starts, so `Forever` retries it forever.
    pub wait_for_resources: Option<WaitForResources>,
}

impl ActorStartOptions {
    /// Serializes these options into run-start query parameters. Shared by the Actor and
    /// task start methods (DRY).
    pub(crate) fn apply(&self, params: &mut QueryParams) {
        params
            .add_str("build", self.build.clone())
            .add_int("memory", self.memory_mbytes)
            .add_int("timeout", self.timeout_secs)
            .add_int("waitForFinish", self.wait_for_finish)
            .add_int("maxItems", self.max_items)
            .add_float("maxTotalChargeUsd", self.max_total_charge_usd)
            .add_bool("restartOnError", self.restart_on_error)
            .add_str("forcePermissionLevel", self.force_permission_level.clone())
            .add_str("webhooks", self.encoded_webhooks());
    }

    /// Encodes the `webhooks` array as base64-encoded JSON, as required by the API.
    fn encoded_webhooks(&self) -> Option<String> {
        use base64::Engine;
        let webhooks = self.webhooks.as_ref()?;
        let json = serde_json::to_vec(webhooks).ok()?;
        Some(base64::engine::general_purpose::STANDARD.encode(json))
    }
}

/// Options for building an Actor.
#[derive(Debug, Default, Clone)]
pub struct ActorBuildOptions {
    /// If `true`, use beta versions of Apify packages.
    pub beta_packages: Option<bool>,
    /// Tag to apply to the build (e.g. `latest`).
    pub tag: Option<String>,
    /// Whether to use the Docker build cache (default `true`).
    pub use_cache: Option<bool>,
    /// Maximum seconds to wait server-side for the build to finish (max 60).
    pub wait_for_finish: Option<i64>,
}

/// Client for a specific Actor.
///
/// Provides CRUD methods plus convenience helpers to start/call the Actor, build it,
/// and access its runs, builds, versions and webhooks.
#[derive(Debug, Clone)]
pub struct ActorClient {
    root: ApifyClient,
    ctx: ResourceContext,
    base_url: String,
    id: String,
}

impl ActorClient {
    pub(crate) fn new(root: ApifyClient, http: HttpClient, base_url: &str, id: &str) -> Self {
        Self {
            root,
            ctx: ResourceContext::single(http, base_url, "actors", id),
            base_url: base_url.to_string(),
            id: id.to_string(),
        }
    }

    /// Fetches the Actor object, or `None` if it does not exist.
    pub async fn get(&self) -> ApifyClientResult<Option<Actor>> {
        get_resource(&self.ctx, None, &QueryParams::new()).await
    }

    /// Updates the Actor with the given fields and returns the updated object.
    pub async fn update<T: Serialize>(&self, new_fields: &T) -> ApifyClientResult<Actor> {
        update_resource(&self.ctx, None, new_fields).await
    }

    /// Deletes the Actor.
    pub async fn delete(&self) -> ApifyClientResult<()> {
        delete_resource(&self.ctx, None).await
    }

    /// Starts the Actor and returns immediately with the created run.
    ///
    /// `input` is any JSON-serializable value (or `None` for no input). To send a non-JSON
    /// input (e.g. a ZIP archive) as raw bytes instead, use [`start_raw`](Self::start_raw).
    /// Set [`ActorStartOptions::wait_for_resources`] to retry the start while the account lacks
    /// the memory or a concurrent-run slot for it.
    pub async fn start<T: Serialize>(
        &self,
        input: Option<&T>,
        options: ActorStartOptions,
    ) -> ApifyClientResult<ActorRun> {
        let body = match input {
            Some(value) => Some(serde_json::to_vec(value)?),
            None => None,
        };
        self.start_with_body(body, "application/json", options)
            .await
    }

    /// Starts the Actor with a raw request body instead of a JSON-serializable value.
    ///
    /// Use this for a non-JSON input, e.g. a ZIP archive paired with
    /// `options.content_type = Some("application/zip".into())`; the bytes are sent exactly as
    /// given, with no JSON serialization. For an object or array input, use
    /// [`start`](Self::start) instead, which handles the serialization. Mirrors the reference
    /// client, whose `ActorInput` accepts a plain object, an array of them, or raw bytes.
    pub async fn start_raw(
        &self,
        input: &[u8],
        options: ActorStartOptions,
    ) -> ApifyClientResult<ActorRun> {
        self.start_with_body(Some(input.to_vec()), "application/octet-stream", options)
            .await
    }

    /// Shared implementation of [`start`](Self::start) and [`start_raw`](Self::start_raw):
    /// applies the run-start options as query parameters and posts the given body, falling back
    /// to `default_content_type` only when the caller did not set `options.content_type`.
    async fn start_with_body(
        &self,
        body: Option<Vec<u8>>,
        default_content_type: &str,
        options: ActorStartOptions,
    ) -> ApifyClientResult<ActorRun> {
        let mut params = QueryParams::new();
        options.apply(&mut params);
        let content_type = options
            .content_type
            .clone()
            .unwrap_or_else(|| default_content_type.to_string());
        crate::clients::base::retry_while_resource_limited(options.wait_for_resources, || {
            post_with_body(
                &self.ctx,
                Some("runs"),
                &params,
                body.clone(),
                &content_type,
            )
        })
        .await
    }

    /// Starts the Actor and waits (client-side polling) for it to finish.
    ///
    /// `wait_secs` controls the wait budget:
    /// - `None` polls indefinitely until the run reaches a terminal state.
    /// - `Some(n)` bounds the wait to roughly `n` seconds; if the run has not finished by
    ///   then, the **last fetched (still non-terminal) run is returned** rather than an
    ///   error. Check `status` / `is_terminal()` on the result when using `Some`.
    ///
    /// Time spent retrying under [`ActorStartOptions::wait_for_resources`] does not count
    /// toward `wait_secs`, since that budget only starts once the run has actually started.
    pub async fn call<T: Serialize>(
        &self,
        input: Option<&T>,
        options: ActorStartOptions,
        wait_secs: Option<i64>,
    ) -> ApifyClientResult<ActorRun> {
        let run = self.start(input, options).await?;
        // Use the root client's run client so polling targets the canonical run route.
        self.root.run(run.id).wait_for_finish(wait_secs).await
    }

    /// Starts the Actor with a raw request body ([`start_raw`](Self::start_raw)) and waits for
    /// it to finish, exactly like [`call`](Self::call) but for a non-JSON input.
    pub async fn call_raw(
        &self,
        input: &[u8],
        options: ActorStartOptions,
        wait_secs: Option<i64>,
    ) -> ApifyClientResult<ActorRun> {
        let run = self.start_raw(input, options).await?;
        self.root.run(run.id).wait_for_finish(wait_secs).await
    }

    /// Builds the given version of the Actor and returns the created build.
    pub async fn build(
        &self,
        version_number: &str,
        options: ActorBuildOptions,
    ) -> ApifyClientResult<Build> {
        let mut params = QueryParams::new();
        params
            .add_str("version", Some(version_number.to_string()))
            .add_bool("betaPackages", options.beta_packages)
            .add_str("tag", options.tag)
            .add_bool("useCache", options.use_cache)
            .add_int("waitForFinish", options.wait_for_finish);
        post_with_body(&self.ctx, Some("builds"), &params, None, "application/json").await
    }

    /// Resolves the Actor's default build and returns a client for it.
    ///
    /// `wait_for_finish` optionally bounds how long (in seconds) the API waits for the build
    /// to finish before responding, matching the reference client's `defaultBuild(options)`.
    pub async fn default_build(
        &self,
        wait_for_finish: Option<i64>,
    ) -> ApifyClientResult<BuildClient> {
        let mut params = QueryParams::new();
        params.add_int("waitForFinish", wait_for_finish);
        let url = params.apply_to_url(&self.ctx.url(Some("builds/default")));
        let response = self
            .ctx
            .http
            .call(crate::http_client::HttpRequest {
                method: crate::http_client::HttpMethod::Get,
                url,
                headers: Default::default(),
                body: None,
                timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
            })
            .await?;
        let build: Build = parse_data_envelope(&response.body)?;
        Ok(BuildClient::new(
            self.ctx.http.clone(),
            &self.base_url,
            &build.id,
        ))
    }

    /// Validates the given input against the Actor's input schema.
    ///
    /// Uses the Actor's default build for the input schema. To validate against a specific
    /// build, use [`ActorClient::validate_input_for_build`].
    pub async fn validate_input<T: Serialize>(&self, input: &T) -> ApifyClientResult<Value> {
        self.validate_input_for_build(input, None).await
    }

    /// Validates the given input against the input schema of a specific Actor build.
    ///
    /// `build` is the optional tag or number of the Actor build whose input schema is used for
    /// validation (e.g. `"latest"` or `"1.2.34"`); passing `None` uses the default build, which
    /// is equivalent to [`ActorClient::validate_input`]. This maps to the spec's optional `build`
    /// query parameter on `POST /v2/actors/{actorId}/validate-input`.
    pub async fn validate_input_for_build<T: Serialize>(
        &self,
        input: &T,
        build: Option<&str>,
    ) -> ApifyClientResult<Value> {
        let body = serde_json::to_vec(input)?;
        let mut params = QueryParams::new();
        params.add_str("build", build);
        // `validate-input` returns a bare `{ "valid": ... }` object, *not* the usual
        // `{ "data": ... }` envelope, so it must skip `parse_data_envelope`.
        crate::clients::base::post_action_raw(
            &self.ctx,
            Some("validate-input"),
            &params,
            Some(body),
            Some("application/json"),
        )
        .await
    }

    /// Returns a client for the last run of this Actor, optionally filtered by run status.
    ///
    /// `status` filters by run status (e.g. `"SUCCEEDED"`, `"FAILED"`, `"RUNNING"`); pass `None`
    /// to leave it unfiltered. This maps to the `status` query parameter on
    /// `GET /v2/actors/{actorId}/runs/last` and mirrors the reference client's `lastRun({ status })`.
    /// To also filter by `origin`, use [`ActorClient::last_run_with_options`].
    pub fn last_run(&self, status: Option<&str>) -> RunClient {
        self.last_run_with_options(LastRunOptions {
            status: status.map(str::to_owned),
            origin: None,
        })
    }

    /// Returns a client for the last run of this Actor, applying the given [`LastRunOptions`]
    /// (e.g. [`LastRunOptions::status`] and/or [`LastRunOptions::origin`]).
    ///
    /// `status` filters by run status (e.g. `"SUCCEEDED"`, `"FAILED"`, `"RUNNING"`); `origin` filters
    /// by how the run was started, with accepted values being the platform's run origins (e.g.
    /// `"DEVELOPMENT"`, `"WEB"`, `"API"`, `"SCHEDULER"`). Both are documented optional query
    /// parameters on `GET /v2/actors/{actorId}/runs/last` and match the reference client's
    /// `lastRun({ status, origin })`; leave a field as `None` to omit it.
    pub fn last_run_with_options(&self, options: LastRunOptions) -> RunClient {
        let mut client = RunClient::new(
            self.root.clone(),
            self.ctx.http.clone(),
            &self.ctx.url(None),
            "runs",
            "last",
        );
        if let Some(status) = options.status.as_deref() {
            client.set_base_param("status", status);
        }
        if let Some(origin) = options.origin.as_deref() {
            client.set_base_param("origin", origin);
        }
        client
    }

    /// Returns a client for this Actor's build collection.
    pub fn builds(&self) -> BuildCollectionClient {
        BuildCollectionClient::with_base(self.ctx.http.clone(), &self.ctx.url(None), "builds")
    }

    /// Returns a client for this Actor's run collection.
    pub fn runs(&self) -> RunCollectionClient {
        RunCollectionClient::new(self.ctx.http.clone(), &self.ctx.url(None), "runs")
    }

    /// Returns a client for a specific version of this Actor.
    pub fn version(&self, version_number: &str) -> ActorVersionClient {
        ActorVersionClient::new(self.ctx.http.clone(), &self.ctx.url(None), version_number)
    }

    /// Returns a client for this Actor's version collection.
    pub fn versions(&self) -> ActorVersionCollectionClient {
        ActorVersionCollectionClient::new(self.ctx.http.clone(), &self.ctx.url(None))
    }

    /// Returns a client for this Actor's webhook collection.
    pub fn webhooks(&self) -> WebhookCollectionClient {
        WebhookCollectionClient::with_base(self.ctx.http.clone(), &self.ctx.url(None))
    }

    /// The Actor's ID (or `username~name`) as provided.
    pub fn id(&self) -> &str {
        &self.id
    }
}