awa_model/
job.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use sqlx::prelude::FromRow;
4use std::fmt;
5
6/// Job states in the lifecycle state machine.
7#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, sqlx::Type)]
8#[sqlx(type_name = "awa.job_state", rename_all = "snake_case")]
9#[serde(rename_all = "snake_case")]
10pub enum JobState {
11    Scheduled,
12    Available,
13    Running,
14    Completed,
15    Retryable,
16    Failed,
17    Cancelled,
18    WaitingExternal,
19}
20
21impl fmt::Display for JobState {
22    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
23        f.write_str(self.as_str())
24    }
25}
26
27impl std::str::FromStr for JobState {
28    type Err = String;
29
30    fn from_str(s: &str) -> Result<Self, Self::Err> {
31        match s {
32            "scheduled" => Ok(JobState::Scheduled),
33            "available" => Ok(JobState::Available),
34            "running" => Ok(JobState::Running),
35            "completed" => Ok(JobState::Completed),
36            "retryable" => Ok(JobState::Retryable),
37            "failed" => Ok(JobState::Failed),
38            "cancelled" => Ok(JobState::Cancelled),
39            "waiting_external" => Ok(JobState::WaitingExternal),
40            other => Err(format!("unknown job state: {other}")),
41        }
42    }
43}
44
45impl JobState {
46    /// Canonical snake_case representation used by Postgres and JSON.
47    pub const fn as_str(&self) -> &'static str {
48        match self {
49            JobState::Scheduled => "scheduled",
50            JobState::Available => "available",
51            JobState::Running => "running",
52            JobState::Completed => "completed",
53            JobState::Retryable => "retryable",
54            JobState::Failed => "failed",
55            JobState::Cancelled => "cancelled",
56            JobState::WaitingExternal => "waiting_external",
57        }
58    }
59
60    /// Bit position for unique_states bitmask.
61    pub fn bit_position(&self) -> u8 {
62        match self {
63            JobState::Scheduled => 0,
64            JobState::Available => 1,
65            JobState::Running => 2,
66            JobState::Completed => 3,
67            JobState::Retryable => 4,
68            JobState::Failed => 5,
69            JobState::Cancelled => 6,
70            JobState::WaitingExternal => 7,
71        }
72    }
73
74    /// Check if this state is terminal (no further transitions possible).
75    pub fn is_terminal(&self) -> bool {
76        matches!(
77            self,
78            JobState::Completed | JobState::Failed | JobState::Cancelled
79        )
80    }
81}
82
83/// A row from the `awa.jobs` table.
84#[derive(Debug, Clone, FromRow, Serialize, Deserialize)]
85pub struct JobRow {
86    pub id: i64,
87    pub kind: String,
88    pub queue: String,
89    pub args: serde_json::Value,
90    pub state: JobState,
91    pub priority: i16,
92    pub attempt: i16,
93    pub run_lease: i64,
94    pub max_attempts: i16,
95    pub run_at: DateTime<Utc>,
96    pub heartbeat_at: Option<DateTime<Utc>>,
97    pub deadline_at: Option<DateTime<Utc>>,
98    pub attempted_at: Option<DateTime<Utc>>,
99    pub finalized_at: Option<DateTime<Utc>>,
100    pub created_at: DateTime<Utc>,
101    pub errors: Option<Vec<serde_json::Value>>,
102    pub metadata: serde_json::Value,
103    pub tags: Vec<String>,
104    pub unique_key: Option<Vec<u8>>,
105    /// Unique states bitmask — stored as BIT(8) in Postgres.
106    /// Skipped in FromRow since it's only used by the DB-side unique index.
107    #[sqlx(skip)]
108    pub unique_states: Option<u8>,
109    /// Callback ID for external webhook completion.
110    pub callback_id: Option<uuid::Uuid>,
111    /// Deadline for callback timeout.
112    pub callback_timeout_at: Option<DateTime<Utc>>,
113    /// CEL filter expression for callback resolution.
114    pub callback_filter: Option<String>,
115    /// CEL expression: does the payload indicate completion?
116    pub callback_on_complete: Option<String>,
117    /// CEL expression: does the payload indicate failure?
118    pub callback_on_fail: Option<String>,
119    /// CEL expression to transform the payload before returning.
120    pub callback_transform: Option<String>,
121    /// Structured progress reported by the handler during execution.
122    pub progress: Option<serde_json::Value>,
123}
124
125/// Options for inserting a job.
126#[derive(Debug, Clone)]
127pub struct InsertOpts {
128    pub queue: String,
129    pub priority: i16,
130    pub max_attempts: i16,
131    pub run_at: Option<DateTime<Utc>>,
132    pub deadline_duration: Option<chrono::Duration>,
133    pub metadata: serde_json::Value,
134    pub tags: Vec<String>,
135    pub unique: Option<UniqueOpts>,
136    /// Routes the job to a specific enqueue shard when the destination
137    /// queue has `enqueue_shards > 1`. Jobs sharing an `ordering_key`
138    /// land on the same shard, which preserves FIFO within that key.
139    /// `None` falls back to the per-store rotor that spreads batches
140    /// across shards. Ignored at `enqueue_shards = 1`. See ADR-025.
141    pub ordering_key: Option<Vec<u8>>,
142}
143
144impl Default for InsertOpts {
145    fn default() -> Self {
146        Self {
147            queue: "default".to_string(),
148            priority: 2,
149            max_attempts: 25,
150            run_at: None,
151            deadline_duration: None,
152            metadata: serde_json::json!({}),
153            tags: Vec::new(),
154            unique: None,
155            ordering_key: None,
156        }
157    }
158}
159
160/// Uniqueness constraint options.
161#[derive(Debug, Clone)]
162pub struct UniqueOpts {
163    /// Include queue in uniqueness calculation.
164    pub by_queue: bool,
165    /// Include args in uniqueness calculation.
166    pub by_args: bool,
167    /// Period bucket for time-based uniqueness (epoch seconds / period).
168    pub by_period: Option<i64>,
169    /// States in which uniqueness is enforced.
170    /// Default: scheduled, available, running, completed, retryable (bits 0-4).
171    pub states: u8,
172}
173
174impl Default for UniqueOpts {
175    fn default() -> Self {
176        Self {
177            by_queue: false,
178            by_args: true,
179            by_period: None,
180            // Default: bits 0-4 set (scheduled, available, running, completed, retryable)
181            states: 0b0001_1111,
182        }
183    }
184}
185
186impl UniqueOpts {
187    /// Convert the states bitmask to a BIT(8) representation for Postgres.
188    pub fn states_bits(&self) -> Vec<u8> {
189        vec![self.states]
190    }
191}
192
193/// Parameters for bulk insert.
194#[derive(Debug, Clone)]
195pub struct InsertParams {
196    pub kind: String,
197    pub args: serde_json::Value,
198    pub opts: InsertOpts,
199}
awa_model/job.rs

awa_model/
job.rs
