spiral-sdk 0.1.11

//! Handles to the durable Tasks that resource methods submit.

use std::fmt::{Debug, Formatter};

use uuid::Uuid;

use crate::api::tasks::{
    CreateTaskResponse, SubmitTaskRequest, TaskDefinition, TaskOptions, TaskProgress, TaskResult,
};
use crate::ids::ProjectId;
use crate::resources::task::{
    HttpTaskClient, Task, TaskDetails, TaskStatus, TaskWaitError, TaskWaitOptions,
};
use crate::{Error, Project};

/// Who executes a submitted Task. It never changes what the Task does.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum Run {
    /// Any compatible WorkerPool or standalone Worker claims the Task.
    #[default]
    Remote,
}

/// A submitted Task, which resolves to the [`TaskResult`] of versions it published.
///
/// Dropping a handle or abandoning [`Self::wait`] never cancels the Task; only
/// [`Self::cancel`] does.
#[derive(Clone)]
pub struct TaskHandle {
    tasks: HttpTaskClient,
    task_id: Uuid,
}

impl Debug for TaskHandle {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("TaskHandle")
            .field("task_id", &self.task_id)
            .finish()
    }
}

impl Project {
    /// Submit any Task definition, including kinds no resource method wraps.
    ///
    /// Returns `None` when the overlap policy skipped the submission because an equivalent
    /// Task already waits to run.
    pub async fn submit_task(
        &self,
        definition: TaskDefinition,
        options: Option<TaskOptions>,
        run: Run,
    ) -> Result<Option<TaskHandle>, Error> {
        submit(
            self.client.tasks(),
            ProjectId::from(self.id.as_str()),
            definition,
            options,
            run,
        )
        .await
    }

    /// A handle to a Task this Project already holds. It makes no request.
    pub fn task(&self, task_id: Uuid) -> TaskHandle {
        TaskHandle::new(self.client.tasks().clone(), task_id)
    }
}

/// Submit `definition` to run as `run` selects.
pub(crate) async fn submit(
    tasks: &HttpTaskClient,
    project_id: ProjectId,
    definition: TaskDefinition,
    options: Option<TaskOptions>,
    run: Run,
) -> Result<Option<TaskHandle>, Error> {
    let Run::Remote = run;
    let response = tasks
        .submit_task(
            project_id.to_string(),
            SubmitTaskRequest {
                definition,
                parent_task_id: None,
                options,
                claim: None,
                idempotency_key: Uuid::now_v7().to_string(),
            },
        )
        .await?;
    match response {
        CreateTaskResponse::Accepted(accepted) => {
            Ok(Some(TaskHandle::new(tasks.clone(), accepted.task.task_id)))
        }
        CreateTaskResponse::Skipped(_) => Ok(None),
        CreateTaskResponse::Unknown { tag, .. } => Err(Error::Protocol(format!(
            "unknown Task submission outcome `{tag}`"
        ))),
    }
}

impl TaskHandle {
    pub(crate) fn new(tasks: HttpTaskClient, task_id: Uuid) -> Self {
        Self { tasks, task_id }
    }

    /// The Task's stable identifier.
    pub fn id(&self) -> Uuid {
        self.task_id
    }

    /// Read the Task's current status, progress and Attempts.
    pub async fn details(&self) -> Result<TaskDetails, Error> {
        Ok(self.tasks.read_task(self.task_id).await?)
    }

    /// The Task's current status.
    pub async fn status(&self) -> Result<TaskStatus, Error> {
        Ok(self.details().await?.task.status)
    }

    /// The progress its current or latest Attempt last reported, if any.
    pub async fn progress(&self) -> Result<Option<TaskProgress>, Error> {
        let details = self.details().await?;
        let current = details.task.current_attempt_id;
        let attempt = details
            .attempts
            .into_iter()
            .filter(|attempt| current.is_none_or(|current| attempt.attempt_id == current))
            .max_by_key(|attempt| attempt.ordinal);
        Ok(attempt.and_then(|attempt| attempt.progress))
    }

    /// Request cancellation; the running Attempt stops at its supervisor's next check.
    pub async fn cancel(&self) -> Result<Task, Error> {
        Ok(self.tasks.cancel_task(self.task_id).await?)
    }

    /// Wait for the Task to finish and return the versions it published.
    pub async fn wait(&self) -> Result<TaskResult, Error> {
        self.wait_with(TaskWaitOptions::default()).await
    }

    /// Wait with an explicit polling interval and client deadline.
    pub async fn wait_with(&self, options: TaskWaitOptions) -> Result<TaskResult, Error> {
        let task = self
            .tasks
            .wait(self.task_id, options)
            .await
            .map_err(|error| match error {
                TaskWaitError::Timeout { task_id } => Error::TaskWaitTimeout { task_id },
                TaskWaitError::Service(error) => error.into(),
            })?;
        match task.status {
            TaskStatus::Succeeded => task.published.ok_or_else(|| {
                Error::Protocol("the succeeded Task returned no published versions".to_owned())
            }),
            TaskStatus::Cancelled => Err(Error::TaskCancelled {
                task_id: self.task_id,
            }),
            _ => {
                let failure = task.failure;
                Err(Error::TaskFailed {
                    task_id: self.task_id,
                    code: failure.as_ref().map(|failure| failure.code.clone()),
                    message: failure.map(|failure| failure.message).unwrap_or_default(),
                })
            }
        }
    }
}