steda/

types.rs

//! Public queue, task, retry, and worker types.

use std::{fmt, str::FromStr, time::Duration};

use serde::{Deserialize, Serialize};
use serde_json::{Map, Value as JsonValue};
use uuid::Uuid;

/// JSON value type for task parameters, headers, checkpoints, and results.
pub type Json = JsonValue;

/// JSON object type for headers and option payloads.
pub type JsonObject = Map<String, Json>;

/// Steda logical-task identifier.
///
/// Generated by `PostgreSQL` as `UUIDv7`. Task IDs and run IDs are intentionally distinct Rust
/// types so they cannot be mixed at compile time.
#[derive(
    Debug, Clone, Copy, Serialize, Deserialize, sqlx::Type, PartialEq, Eq, Hash, PartialOrd, Ord,
)]
#[serde(transparent)]
#[sqlx(transparent)]
pub struct TaskId(Uuid);

impl TaskId {
    /// Wrap a UUID as a logical-task identifier.
    pub const fn from_uuid(value: Uuid) -> Self {
        Self(value)
    }

    /// Return the underlying UUID.
    pub const fn into_uuid(self) -> Uuid {
        self.0
    }
}

impl From<Uuid> for TaskId {
    fn from(value: Uuid) -> Self {
        Self::from_uuid(value)
    }
}

impl From<TaskId> for Uuid {
    fn from(value: TaskId) -> Self {
        value.into_uuid()
    }
}

impl fmt::Display for TaskId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

impl FromStr for TaskId {
    type Err = uuid::Error;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        value.parse().map(Self::from_uuid)
    }
}

/// Steda execution-run identifier.
///
/// Generated by `PostgreSQL` as `UUIDv7`.
#[derive(
    Debug, Clone, Copy, Serialize, Deserialize, sqlx::Type, PartialEq, Eq, Hash, PartialOrd, Ord,
)]
#[serde(transparent)]
#[sqlx(transparent)]
pub struct RunId(Uuid);

impl RunId {
    /// Wrap a UUID as an execution-run identifier.
    pub const fn from_uuid(value: Uuid) -> Self {
        Self(value)
    }

    /// Return the underlying UUID.
    pub const fn into_uuid(self) -> Uuid {
        self.0
    }
}

impl From<Uuid> for RunId {
    fn from(value: Uuid) -> Self {
        Self::from_uuid(value)
    }
}

impl From<RunId> for Uuid {
    fn from(value: RunId) -> Self {
        value.into_uuid()
    }
}

impl fmt::Display for RunId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

impl FromStr for RunId {
    type Err = uuid::Error;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        value.parse().map(Self::from_uuid)
    }
}

/// Retry strategy for failed tasks.
///
/// Retry delays use [`Duration`] at the Rust boundary. Steda converts them to the canonical
/// numeric-second representation only when crossing into `PostgreSQL`.
///
/// ```
/// use std::time::Duration;
///
/// use steda::RetryStrategy;
///
/// let strategy =
///     RetryStrategy::exponential(Duration::from_secs(5), 2.0, Some(Duration::from_secs(300)));
/// assert!(matches!(strategy, RetryStrategy::Exponential { .. }));
/// ```
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum RetryStrategy {
    /// Retry after a fixed delay.
    Fixed {
        /// Delay before retrying.
        delay: Duration,
    },

    /// Retry with exponential backoff.
    Exponential {
        /// Initial delay before retrying.
        initial_delay: Duration,
        /// Exponential multiplier.
        factor: f64,
        /// Optional upper bound for computed delay.
        max_delay: Option<Duration>,
    },

    /// Do not retry failed runs.
    None,
}

impl RetryStrategy {
    /// Create a fixed retry strategy.
    pub const fn fixed(delay: Duration) -> Self {
        Self::Fixed { delay }
    }

    /// Create an exponential retry strategy.
    pub const fn exponential(
        initial_delay: Duration,
        factor: f64,
        max_delay: Option<Duration>,
    ) -> Self {
        Self::Exponential { initial_delay, factor, max_delay }
    }

    /// Disable retries.
    pub const fn none() -> Self {
        Self::None
    }
}

/// Durable cancellation deadlines for one logical task.
///
/// `max_delay` is measured from enqueue until the first start and no longer applies once the task
/// has begun. `max_duration` is measured from the first start across the remainder of the logical
/// task, including retries and durable sleeps. `PostgreSQL` evaluates the resulting whole-second
/// deadlines against its authoritative clock.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct CancellationPolicy {
    /// Maximum duration from first start.
    pub(crate) max_duration: Option<Duration>,

    /// Maximum delay before first start.
    pub(crate) max_delay: Option<Duration>,
}

impl CancellationPolicy {
    /// Create an empty cancellation policy.
    pub const fn new() -> Self {
        Self { max_duration: None, max_delay: None }
    }

    /// Set the maximum duration from the first start across the logical task.
    #[must_use]
    pub const fn max_duration(mut self, duration: Duration) -> Self {
        self.max_duration = Some(duration);
        self
    }

    /// Set the maximum delay between enqueue and the first start.
    #[must_use]
    pub const fn max_delay(mut self, delay: Duration) -> Self {
        self.max_delay = Some(delay);
        self
    }
}

/// Options for spawning a task.
#[derive(Debug, Clone, Default)]
pub(crate) struct SpawnConfig {
    /// Maximum number of attempts, including the first execution.
    pub max_attempts: Option<u32>,

    /// Retry strategy.
    pub retry_strategy: Option<RetryStrategy>,

    /// Custom headers for the task.
    pub headers: Option<JsonObject>,

    /// Cancellation policy.
    pub cancellation: Option<CancellationPolicy>,

    /// Idempotency key for deduplicating task creation.
    pub idempotency_key: Option<String>,
}

/// Internal result of spawning a logical task.
#[derive(Debug, Clone, Copy)]
pub(crate) struct SpawnResult {
    /// Unique logical task identifier.
    pub task_id: TaskId,

    /// Whether this was a new task, false if deduplicated.
    pub created: bool,
}

/// Optional persisted queue-maintenance policy overrides.
///
/// Unset values leave the corresponding field at its current/default value. New queues default to
/// a 30-day terminal task TTL and a maximum of 1,000 logical-task deletions per cleanup pass.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct QueuePolicyOptions {
    /// Terminal task cleanup TTL.
    pub(crate) cleanup_ttl: Option<Duration>,

    /// Maximum rows cleaned per cleanup pass.
    pub(crate) cleanup_limit: Option<u32>,
}

impl QueuePolicyOptions {
    /// Create an empty set of queue-policy overrides.
    pub const fn new() -> Self {
        Self { cleanup_ttl: None, cleanup_limit: None }
    }

    /// Set the terminal-task cleanup TTL.
    #[must_use]
    pub const fn cleanup_ttl(mut self, cleanup_ttl: Duration) -> Self {
        self.cleanup_ttl = Some(cleanup_ttl);
        self
    }

    /// Set the maximum number of logical tasks deleted per cleanup pass.
    #[must_use]
    pub const fn cleanup_limit(mut self, cleanup_limit: u32) -> Self {
        self.cleanup_limit = Some(cleanup_limit);
        self
    }
}

/// Queue maintenance policy snapshot.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueuePolicy {
    /// Queue name.
    pub queue_name: String,

    /// Terminal task cleanup TTL.
    pub cleanup_ttl: Duration,

    /// Maximum cleanup rows per pass.
    pub cleanup_limit: u32,
}

/// Result of one retention cleanup pass for a queue.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueueCleanup {
    /// Queue name.
    pub queue_name: String,

    /// Number of terminal tasks deleted.
    pub tasks_deleted: u32,
}

/// Durable task state.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum TaskState {
    /// Task is waiting to run.
    Pending,

    /// Task is currently running.
    Running,

    /// Task is sleeping until a later time.
    Sleeping,

    /// Task completed successfully.
    Completed,

    /// Task failed terminally.
    Failed,

    /// Task was cancelled.
    Cancelled,
}

/// Internal JSON-erased task result snapshot returned by `PostgreSQL`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum TaskResultSnapshot {
    /// Task is waiting to run.
    Pending,

    /// Task is currently running.
    Running,

    /// Task is sleeping until a later time.
    Sleeping,

    /// Task completed successfully.
    Completed {
        /// Completion payload.
        result: Json,
    },

    /// Task failed terminally.
    Failed {
        /// Failure payload.
        failure: Json,
    },

    /// Task was cancelled.
    Cancelled,
}

impl TaskResultSnapshot {
    /// Whether this raw database snapshot is terminal.
    pub(crate) const fn is_terminal(&self) -> bool {
        matches!(self, Self::Completed { .. } | Self::Failed { .. } | Self::Cancelled)
    }
}

/// A claimed task ready for execution.
#[derive(Debug, Clone)]
pub(crate) struct ClaimedTask {
    /// Claimed run identifier.
    pub(crate) run_id: RunId,

    /// Logical task identifier.
    pub(crate) task_id: TaskId,

    /// Registered task name.
    pub(crate) task_name: String,

    /// Attempt number for this run.
    pub(crate) attempt: u32,

    /// Task parameters.
    pub(crate) params: Json,

    /// Task headers.
    pub(crate) headers: Option<JsonObject>,
}