apalis_core/task/

builder.rs

//! # Task Builder
//!
//! The `TaskBuilder` module provides a flexible builder pattern for constructing [`Task`] instances
//! with customizable configuration options. It allows users to specify arguments, context, extensions,
//! task identifiers, attempt information, status, and scheduling details for tasks.
//!
//! ## Features
//! - Create tasks with required arguments and optional context.
//! - Attach custom extensions/data to tasks.
//! - Assign unique task identifiers.
//! - Configure attempt and status information.
//! - Schedule tasks to run at specific times, after delays, or at intervals (seconds, minutes, hours).
//! - Build tasks with sensible defaults for omitted fields.
//!
//! ## Usage
//! Use [`TaskBuilder`] to incrementally configure a task, then call `.build()` to obtain a [`Task`] instance.
//! Convenience methods are provided for common scheduling scenarios.
//!
//! ### Example
//! ```rust,ignore
//! let task = TaskBuilder::new(args)
//!     .max_attempts(3)
//!     .run_in_minutes(10)
//!     .build();
//! ```
//!
use crate::{
    backend::queue::Queue,
    task::{
        ExecutionContext, Task,
        attempt::Attempt,
        extensions::Extensions,
        metadata::{Metadata, MetadataStore},
        status::Status,
        task_id::TaskId,
    },
};
use std::{
    fmt::Debug,
    sync::Arc,
    time::{Duration, SystemTime, UNIX_EPOCH},
};

/// Builder for creating [`Task`] instances with optional configuration
#[derive(Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[must_use = "TaskBuilder is used to construct a Task. Ensure to call `.build()` to create the Task instance."]
pub struct TaskBuilder<Args> {
    /// The arguments for the task
    pub(super) args: Args,
    /// Execution context for the task, including metadata and extensions
    pub(super) ctx: ExecutionContext,
}

impl<Args> TaskBuilder<Args> {
    /// Create a new TaskBuilder with the required args
    pub fn new(args: Args) -> Self {
        Self {
            args,
            ctx: ExecutionContext::default(),
        }
    }

    /// Set the task's metadata
    pub fn with_metadata(mut self, metadata: MetadataStore) -> Self {
        self.ctx.metadata = metadata;
        self
    }

    /// Set the task's runtime data
    pub fn with_data(mut self, data: Extensions) -> Self {
        self.ctx.data = data;
        self
    }

    /// Insert a value into the task's data context
    pub fn data<D: Clone + Send + Sync + 'static>(mut self, value: D) -> Self {
        self.ctx.data.insert(value);
        self
    }

    /// Insert a value into the task's metadata
    pub fn metadata<M>(mut self, value: &M) -> Self
    where
        M: Metadata,
        M::Error: Debug,
    {
        value
            .inject(&mut self.ctx.metadata)
            .expect("Could not add Metadata");
        self
    }

    /// Set the task ID
    pub fn task_id(mut self, task_id: TaskId) -> Self {
        self.ctx.task_id = Some(task_id);
        self
    }

    /// Set the attempt information
    pub fn attempt(mut self, attempt: usize) -> Self {
        self.ctx.attempt = Attempt::new_with_value(attempt);
        self
    }

    /// Set the task status
    pub fn status(mut self, status: Status) -> Self {
        self.ctx.status = status.into();
        self
    }

    /// Set the maximum number of attempts allowed for the task
    pub fn max_attempts(mut self, max_attempts: usize) -> Self {
        self.ctx.max_attempts = Some(max_attempts);
        self
    }

    /// Set the priority of the task
    pub fn priority(mut self, priority: usize) -> Self {
        self.ctx.priority = Some(priority);
        self
    }

    /// Set the queue this task belongs to
    pub fn queue(mut self, queue: Queue) -> Self {
        self.ctx.queue = Some(queue);
        self
    }

    /// Schedule the task to run at a specific Unix timestamp
    pub fn run_at_timestamp(mut self, timestamp: u64) -> Self {
        self.ctx.run_at = Some(timestamp);
        self
    }

    /// Schedule the task to run at a specific SystemTime
    pub fn run_at_time(mut self, time: SystemTime) -> Self {
        let timestamp = time
            .duration_since(UNIX_EPOCH)
            .expect("Time went backwards")
            .as_secs();
        self.ctx.run_at = Some(timestamp);
        self
    }

    /// Schedule the task to run after a delay from now
    pub fn run_after(mut self, delay: Duration) -> Self {
        let now = SystemTime::now();
        let run_time = now + delay;
        let timestamp = run_time
            .duration_since(UNIX_EPOCH)
            .expect("Time went backwards")
            .as_secs();
        self.ctx.run_at = Some(timestamp);
        self
    }

    /// Schedule the task to run in the specified number of seconds
    pub fn run_in_seconds(self, seconds: u64) -> Self {
        self.run_after(Duration::from_secs(seconds))
    }

    /// Schedule the task to run in the specified number of minutes
    pub fn run_in_minutes(self, minutes: u64) -> Self {
        self.run_after(Duration::from_secs(minutes * 60))
    }

    /// Schedule the task to run in the specified number of hours
    pub fn run_in_hours(self, hours: u64) -> Self {
        self.run_after(Duration::from_secs(hours * 3600))
    }

    /// Set the idempotency key
    pub fn idempotency_key<S: AsRef<str>>(mut self, idempotency_key: S) -> Self {
        self.ctx.idempotency_key = Some(idempotency_key.as_ref().to_owned());
        self
    }

    /// Set the time the task was completed
    pub fn done_at(mut self, done_at: Option<u64>) -> Self {
        self.ctx.done_at = done_at;
        self
    }

    /// Set the time the task was locked
    pub fn lock_at(mut self, lock_at: Option<u64>) -> Self {
        self.ctx.lock_at = lock_at;
        self
    }

    /// Set the worker/process identifier holding the lock on this task
    pub fn lock_by(mut self, lock_by: Option<String>) -> Self {
        self.ctx.lock_by = lock_by;
        self
    }

    /// Build the Task with default context
    pub fn build(self) -> Task<Args> {
        Task {
            args: self.args,
            ctx: Arc::new(self.ctx),
        }
    }
}

impl<Args> TaskBuilder<Args> {
    /// Maps the `args` field using the provided function, returning a result.
    pub fn try_map_args<F, NewArgs, Err>(self, f: F) -> Result<TaskBuilder<NewArgs>, Err>
    where
        F: FnOnce(Args) -> Result<NewArgs, Err>,
    {
        Ok(TaskBuilder {
            args: f(self.args)?,
            ctx: self.ctx,
        })
    }
    /// Maps the `args` field using the provided function, returning the new args.
    pub fn map_args<F, NewArgs>(self, f: F) -> TaskBuilder<NewArgs>
    where
        F: FnOnce(Args) -> NewArgs,
    {
        TaskBuilder {
            args: f(self.args),
            ctx: self.ctx,
        }
    }

    /// Maps only the `execution_context` field.
    pub fn map_context<F>(self, f: F) -> Self
    where
        F: FnOnce(ExecutionContext) -> ExecutionContext,
    {
        Self {
            args: self.args,
            ctx: f(self.ctx),
        }
    }
}