azums_core/
model.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4use uuid::Uuid;
5
6/// Per-queue job execution ordering policy.
7#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)]
8pub enum QueueOrdering {
9    /// Process jobs in exact First-In, First-Out order by creation time (`created_at ASC`).
10    #[default]
11    Fifo,
12    /// Process jobs as fast as possible without strict creation order guarantees.
13    Fastest,
14}
15
16/// Configuration options for a job queue.
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18pub struct QueueConfig {
19    pub ordering: QueueOrdering,
20}
21
22impl Default for QueueConfig {
23    fn default() -> Self {
24        Self {
25            ordering: QueueOrdering::Fifo,
26        }
27    }
28}
29
30impl QueueConfig {
31    pub fn new(ordering: QueueOrdering) -> Self {
32        Self { ordering }
33    }
34}
35
36/// Lightweight job summary model returned when listing jobs in Admin UI or APIs.
37#[derive(Debug, Clone, Serialize, Deserialize)]
38#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
39pub struct JobListItem {
40    pub id: Uuid,
41    pub queue: String,
42    pub job_type: String,
43    pub status: String,
44
45    pub run_at: DateTime<Utc>,
46    pub priority: i32,
47    pub max_attempts: i32,
48
49    pub last_error_code: Option<String>,
50    pub last_error_message: Option<String>,
51
52    pub dlq_reason_code: Option<String>,
53
54    pub created_at: DateTime<Utc>,
55    pub updated_at: DateTime<Utc>,
56}
57
58/// Primary job entity representing a unit of work stored in a storage backend.
59///
60/// # Examples
61///
62/// ```rust
63/// use azums_core::Job;
64///
65/// let job = Job::new("email_send", serde_json::json!({"to": "user@example.com"}))
66///     .queue("emails")
67///     .priority(10)
68///     .max_attempts(5);
69///
70/// assert_eq!(job.queue, "emails");
71/// assert_eq!(job.priority, 10);
72/// assert_eq!(job.max_attempts, 5);
73/// assert_eq!(job.payload["to"], "user@example.com");
74/// ```
75#[derive(Debug, Clone, Serialize, Deserialize)]
76#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
77pub struct Job {
78    pub dataset_id: String,
79    pub replay_of_job_id: Option<Uuid>,
80
81    pub id: Uuid,
82    pub queue: String,
83    pub job_type: String,
84    #[cfg_attr(feature = "sqlx", sqlx(rename = "payload_json"))]
85    pub payload: Value,
86    pub run_at: DateTime<Utc>,
87    pub status: String,
88    pub priority: i32,
89    pub max_attempts: i32,
90
91    pub locked_at: Option<DateTime<Utc>>,
92    pub locked_by: Option<String>,
93    pub lock_expires_at: Option<DateTime<Utc>>,
94
95    pub dlq_reason_code: Option<String>,
96    pub dlq_at: Option<DateTime<Utc>>,
97
98    pub created_at: DateTime<Utc>,
99    pub updated_at: DateTime<Utc>,
100}
101
102impl Job {
103    /// Creates a new `Job` with default queue `"default"`, priority `0`, and max attempts `25`.
104    ///
105    /// # Examples
106    ///
107    /// ```rust
108    /// use azums_core::Job;
109    ///
110    /// let job = Job::new("greet", serde_json::json!({"name": "World"}));
111    /// assert_eq!(job.job_type, "greet");
112    /// assert_eq!(job.payload["name"], "World");
113    /// ```
114    pub fn new(job_type: impl Into<String>, payload: Value) -> Self {
115        let now = Utc::now();
116        Self {
117            dataset_id: "default".to_string(),
118            replay_of_job_id: None,
119            id: Uuid::new_v4(),
120            queue: "default".to_string(),
121            job_type: job_type.into(),
122            payload,
123            run_at: now,
124            status: JobStatus::Queued.as_str().to_string(),
125            priority: 0,
126            max_attempts: 25,
127            locked_at: None,
128            locked_by: None,
129            lock_expires_at: None,
130            dlq_reason_code: None,
131            dlq_at: None,
132            created_at: now,
133            updated_at: now,
134        }
135    }
136
137    /// Sets target queue name for this job.
138    pub fn queue(mut self, queue: impl Into<String>) -> Self {
139        self.queue = queue.into();
140        self
141    }
142
143    /// Sets job execution priority (higher numbers are leased first).
144    pub fn priority(mut self, priority: i32) -> Self {
145        self.priority = priority;
146        self
147    }
148
149    /// Sets maximum retry attempts before moving job to Dead-Letter Queue (DLQ).
150    pub fn max_attempts(mut self, max_attempts: i32) -> Self {
151        self.max_attempts = max_attempts;
152        self
153    }
154
155    /// Sets scheduled execution timestamp (`run_at`).
156    pub fn run_at(mut self, run_at: DateTime<Utc>) -> Self {
157        self.run_at = run_at;
158        self
159    }
160
161    /// Returns reference to job JSON payload.
162    pub fn payload_json(&self) -> &Value {
163        &self.payload
164    }
165
166    /// Deserializes the JSON payload into a concrete type `T`.
167    ///
168    /// # Examples
169    ///
170    /// ```rust
171    /// use azums_core::{Job, Error};
172    /// use serde::Deserialize;
173    ///
174    /// #[derive(Deserialize, Debug, PartialEq)]
175    /// struct EmailPayload {
176    ///     to: String,
177    /// }
178    ///
179    /// let job = Job::new("email", serde_json::json!({"to": "a@b.com"}));
180    /// let payload: EmailPayload = job.payload_typed().unwrap();
181    /// assert_eq!(payload.to, "a@b.com");
182    /// ```
183    pub fn payload_typed<T: serde::de::DeserializeOwned>(&self) -> Result<T, crate::error::Error> {
184        serde_json::from_value(self.payload.clone())
185            .map_err(crate::error::Error::PayloadDeserialization)
186    }
187}
188
189/// Trait-based job processor interface for structured background workers.
190#[async_trait::async_trait]
191pub trait JobProcessor: Send + Sync {
192    /// Processes a single background job execution attempt.
193    async fn process(&self, job: Job) -> anyhow::Result<()>;
194}
195
196/// Specification for enqueueing a new job into a storage backend.
197#[derive(Debug, Clone, Serialize, Deserialize)]
198pub struct NewJob {
199    pub queue: String,
200    pub job_type: String,
201    pub payload_json: Value,
202    pub run_at: DateTime<Utc>,
203    pub priority: i32,
204    pub max_attempts: i32,
205}
206
207impl From<Job> for NewJob {
208    fn from(job: Job) -> Self {
209        NewJob {
210            queue: job.queue,
211            job_type: job.job_type,
212            payload_json: job.payload,
213            run_at: job.run_at,
214            priority: job.priority,
215            max_attempts: job.max_attempts,
216        }
217    }
218}
219
220/// Enumeration of possible job lifecycle states.
221#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
222pub enum JobStatus {
223    Queued,
224    Running,
225    Succeeded,
226    Failed,
227    Dlq,
228    Canceled,
229}
230
231impl JobStatus {
232    /// Returns static string representation of job status.
233    ///
234    /// # Examples
235    ///
236    /// ```rust
237    /// use azums_core::JobStatus;
238    /// assert_eq!(JobStatus::Queued.as_str(), "queued");
239    /// assert_eq!(JobStatus::Dlq.as_str(), "dlq");
240    /// ```
241    pub fn as_str(&self) -> &'static str {
242        match self {
243            JobStatus::Queued => "queued",
244            JobStatus::Running => "running",
245            JobStatus::Succeeded => "succeeded",
246            JobStatus::Failed => "failed",
247            JobStatus::Dlq => "dlq",
248            JobStatus::Canceled => "canceled",
249        }
250    }
251}
252
253/// Asynchronous job handler closure type alias.
254pub type JobHandler = std::sync::Arc<
255    dyn Fn(Job) -> std::pin::Pin<Box<dyn std::future::Future<Output = anyhow::Result<()>> + Send>>
256        + Send
257        + Sync,
258>;
259
260/// Represents an immutable event stored within a durable stream log.
261#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
262#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
263pub struct Event {
264    /// Monotonically increasing 1-based sequence number within the stream.
265    pub sequence_no: i64,
266    /// Name of the target stream log (e.g., "orders", "audit_logs").
267    pub stream_name: String,
268    /// Domain-specific identifier for the event type (e.g., "order_created").
269    pub event_type: String,
270    /// JSON payload content of the event.
271    pub payload_json: serde_json::Value,
272    /// Timestamp when the event was appended to the stream log.
273    pub created_at: DateTime<Utc>,
274}
275
276/// Input model for publishing a new event into a stream log.
277#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
278pub struct NewEvent {
279    /// Domain-specific identifier for the event type (e.g., "order_created").
280    pub event_type: String,
281    /// JSON payload content of the event.
282    pub payload_json: serde_json::Value,
283}
284
285impl NewEvent {
286    /// Creates a new `NewEvent` with the specified event type and JSON payload.
287    pub fn new(event_type: impl Into<String>, payload_json: serde_json::Value) -> Self {
288        Self {
289            event_type: event_type.into(),
290            payload_json,
291        }
292    }
293}
294
295/// Status and offset information for a consumer group registered on a stream log.
296#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
297#[cfg_attr(feature = "sqlx", derive(sqlx::FromRow))]
298pub struct ConsumerGroupStatus {
299    /// Identifier of the consumer group (e.g., "analytics_processor").
300    pub consumer_group: String,
301    /// Name of the stream log.
302    pub stream_name: String,
303    /// Highest sequence number successfully acknowledged by this consumer group.
304    pub last_acked_seq: i64,
305    /// Timestamp when the offset was last updated.
306    pub updated_at: DateTime<Utc>,
307}
azums_core/model.rs

azums_core/
model.rs
