oxidite_queue/
job.rs

1use async_trait::async_trait;
2use serde::{Deserialize, Serialize};
3use std::time::Duration;
4use crate::Result;
5
6/// Job status
7#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
8pub enum JobStatus {
9    /// Job is waiting to be processed
10    Pending,
11    /// Job is currently running
12    Running,
13    /// Job completed successfully
14    Completed,
15    /// Job failed
16    Failed,
17    /// Job is being retried
18    Retrying,
19    /// Permanently failed after max retries
20    DeadLetter,
21}
22
23/// Job result
24pub type JobResult = Result<()>;
25
26/// Trait for background jobs
27#[async_trait]
28pub trait Job: Send + Sync + Serialize + for<'de> Deserialize<'de> {
29    /// Perform the job
30    async fn perform(&self) -> JobResult;
31    
32    /// Maximum number of retries
33    fn max_retries(&self) -> u32 {
34        3
35    }
36    
37    /// Backoff duration for retries
38    fn backoff(&self, attempt: u32) -> Duration {
39        Duration::from_secs(60 * 2_u64.pow(attempt))
40    }
41    
42    /// Job priority (higher = more important)
43    fn priority(&self) -> i32 {
44        0
45    }
46    
47    /// Job name for identification
48    fn name(&self) -> &'static str {
49        std::any::type_name::<Self>()
50    }
51}
52
53/// Job wrapper for storage
54#[derive(Debug, Clone, Serialize, Deserialize)]
55pub struct JobWrapper {
56    /// Unique job identifier
57    pub id: String,
58    /// Job type name
59    pub name: String,
60    /// Serialized job payload
61    pub payload: serde_json::Value,
62    /// Current job status
63    pub status: JobStatus,
64    /// Number of execution attempts
65    pub attempts: u32,
66    /// Maximum number of retries
67    pub max_retries: u32,
68    /// Unix timestamp of creation
69    pub created_at: i64,
70    /// Optional scheduled run time
71    pub scheduled_at: Option<i64>,
72    /// Job priority (higher = more important)
73    pub priority: i32,
74    /// Cron expression for recurring jobs
75    pub cron_schedule: Option<String>,
76    /// Last execution timestamp
77    pub last_run_at: Option<i64>,
78    /// Last error message
79    pub error: Option<String>,
80}
81
82impl JobWrapper {
83    /// Create a new job wrapper from a job implementation
84    pub fn new<J: Job>(job: &J) -> Result<Self> {
85        let payload = serde_json::to_value(job)?;
86        let now = chrono::Utc::now().timestamp();
87        
88        Ok(Self {
89            id: uuid::Uuid::new_v4().to_string(),
90            name: job.name().to_string(),
91            payload,
92            status: JobStatus::Pending,
93            attempts: 0,
94            max_retries: job.max_retries(),
95            created_at: now,
96            scheduled_at: None,
97            priority: job.priority(),
98            cron_schedule: None,
99            last_run_at: None,
100            error: None,
101        })
102    }
103
104    /// Schedule the job with a delay from now
105    pub fn with_delay(mut self, delay: Duration) -> Self {
106        let scheduled_time = chrono::Utc::now().timestamp() + delay.as_secs() as i64;
107        self.scheduled_at = Some(scheduled_time);
108        self
109    }
110
111    /// Schedule job with cron expression (e.g., "0 0 * * * *" for hourly)
112    pub fn with_cron(mut self, cron_expr: String) -> Self {
113        self.scheduled_at = Self::next_cron_run(&cron_expr);
114        self.cron_schedule = Some(cron_expr);
115        self
116    }
117
118    /// Calculate next run time from cron expression
119    fn next_cron_run(cron_expr: &str) -> Option<i64> {
120        use cron::Schedule;
121        use std::str::FromStr;
122        
123        if let Ok(schedule) = Schedule::from_str(cron_expr) {
124            if let Some(next) = schedule.upcoming(chrono::Utc).next() {
125                return Some(next.timestamp());
126            }
127        }
128        None
129    }
130
131    /// Reschedule recurring job for next run
132    pub fn reschedule(&mut self) {
133        if let Some(cron_expr) = &self.cron_schedule {
134            self.scheduled_at = Self::next_cron_run(cron_expr);
135            self.status = JobStatus::Pending;
136            self.attempts = 0;
137            self.last_run_at = Some(chrono::Utc::now().timestamp());
138        }
139    }
140
141    /// Check if job should be retried
142    pub fn should_retry(&self) -> bool {
143        self.status == JobStatus::Failed && self.attempts < self.max_retries
144    }
145
146    /// Calculate backoff duration for retry
147    pub fn calculate_backoff(&self) -> Duration {
148        // Exponential backoff: 60s * 2^attempts
149        Duration::from_secs(60 * 2_u64.pow(self.attempts))
150    }
151}
oxidite_queue/job.rs

oxidite_queue/
job.rs
