snerd_rust/
task.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3
4#[derive(Debug, Serialize, Deserialize, Clone)]
5pub struct JobErrorReturn {
6    #[serde(rename = "error")]
7    pub error_string: String,
8    pub retry_worthy: bool,
9}
10
11#[derive(Debug, Serialize, Deserialize, Clone)]
12pub struct ProgressMessage {
13    #[serde(rename = "task_id")]
14    pub task_id: String,
15    
16    #[serde(rename = "data")]
17    pub data: String,
18}
19
20#[derive(Debug, Serialize, Deserialize, Clone)]
21pub struct RetryableTask {
22    #[serde(rename = "taskId")]
23    pub task_id: String,
24
25    #[serde(rename = "retryCount")]
26    pub retry_count: i32,
27
28    #[serde(rename = "maxRetries")]
29    pub max_retries: i32,
30
31    #[serde(rename = "retryAfterHours")]
32    pub retry_after_hours: f64,
33
34    #[serde(rename = "retryAfterTime")]
35    pub retry_after_time: DateTime<Utc>,
36
37    #[serde(rename = "taskData")]
38    pub task_data: String,
39
40    #[serde(rename = "taskType")]
41    pub task_type: String,
42
43    #[serde(rename = "LastErrorObj", skip_serializing_if = "Option::is_none")]
44    pub last_error_obj: Option<String>,
45
46    #[serde(rename = "LastJobError", skip_serializing_if = "Option::is_none")]
47    pub last_job_error: Option<JobErrorReturn>,
48
49    #[serde(rename = "rateLimitGroup", skip_serializing_if = "Option::is_none")]
50    pub rate_limit_group: Option<String>,
51
52    #[serde(rename = "maxPerMinute", skip_serializing_if = "Option::is_none")]
53    pub max_per_minute: Option<i32>,
54
55    #[serde(rename = "autoDedupe", skip_serializing_if = "Option::is_none")]
56    pub auto_dedupe: Option<bool>,
57
58    #[serde(rename = "urgencyScore", skip_serializing_if = "Option::is_none")]
59    pub urgency_score: Option<f64>,
60
61    #[serde(rename = "payloadHash", skip_serializing_if = "Option::is_none")]
62    pub payload_hash: Option<String>,
63
64    #[serde(rename = "deletedAt", skip_serializing_if = "Option::is_none")]
65    pub deleted_at: Option<DateTime<Utc>>,
66
67    #[serde(rename = "executeAt")]
68    pub execute_at: DateTime<Utc>,
69
70    #[serde(rename = "cronExpression", skip_serializing_if = "Option::is_none")]
71    pub cron_expression: Option<String>,
72
73    #[serde(rename = "webhookUrl", skip_serializing_if = "Option::is_none")]
74    pub webhook_url: Option<String>,
75
76    #[serde(rename = "maxExecutionSeconds", skip_serializing_if = "Option::is_none")]
77    pub max_execution_seconds: Option<u64>,
78
79    #[serde(skip, default = "Utc::now")]
80    pub created_at: DateTime<Utc>,
81
82    #[serde(skip, default = "Utc::now")]
83    pub updated_at: DateTime<Utc>,
84}
85
86impl RetryableTask {
87    pub fn new(
88        task_id: String,
89        task_type: String,
90        task_data: String,
91        max_retries: i32,
92        retry_after_hours: f64,
93        rate_limit_group: Option<String>,
94        max_per_minute: Option<i32>,
95        auto_dedupe: Option<bool>,
96        urgency_score: Option<f64>,
97        execute_at_opt: Option<String>,
98        cron_opt: Option<String>,
99        webhook_url: Option<String>,
100        max_execution_seconds: Option<u64>,
101    ) -> Self {
102        let now = Utc::now();
103
104        // Parse execute_at if provided, else use now
105        let mut execute_at = now;
106        let parsed_cron = cron_opt.map(|c| parse_cron_syntax(&c));
107        
108        if let Some(ref exec_str) = execute_at_opt {
109            if let Ok(parsed_time) = chrono::DateTime::parse_from_rfc3339(exec_str) {
110                execute_at = parsed_time.with_timezone(&Utc);
111            }
112        } else if let Some(ref cron_expr) = parsed_cron {
113            // If no explicit execute_at is provided, but a cron is, default to the FIRST future cron tick!
114            use cron::Schedule;
115            use std::str::FromStr;
116            if let Ok(schedule) = Schedule::from_str(cron_expr) {
117                if let Some(next) = schedule.upcoming(Utc).next() {
118                    execute_at = next;
119                }
120            }
121        }
122
123        let payload_hash = if auto_dedupe.unwrap_or(false) {
124            use xxhash_rust::xxh64::xxh64;
125            let combined = format!("{}{}", task_type, task_data);
126            Some(format!("{:x}", xxh64(combined.as_bytes(), 0)))
127        } else {
128            None
129        };
130        
131        Self {
132            task_id,
133            task_type,
134            task_data,
135            max_retries,
136            retry_after_hours,
137            retry_count: 0,
138            retry_after_time: now,
139            last_error_obj: None,
140            last_job_error: None,
141            rate_limit_group,
142            max_per_minute,
143            auto_dedupe,
144            urgency_score,
145            payload_hash,
146            deleted_at: None,
147            execute_at,
148            cron_expression: parsed_cron,
149            webhook_url,
150            max_execution_seconds,
151            created_at: now,
152            updated_at: now,
153        }
154    }
155
156    pub fn mark_deleted(&mut self) {
157        self.deleted_at = Some(Utc::now());
158        self.updated_at = Utc::now();
159    }
160
161    pub fn update_retry_config(&mut self, error_msg: Option<String>) {
162        self.retry_count += 1;
163
164        // Calculate next retry time
165        let seconds = (self.retry_after_hours * 3600.0) as i64;
166        self.retry_after_time = Utc::now() + chrono::Duration::seconds(seconds);
167
168        self.last_error_obj = error_msg.clone();
169
170        if let Some(msg) = error_msg {
171            self.last_job_error = Some(JobErrorReturn {
172                error_string: msg,
173                retry_worthy: true,
174            });
175        } else {
176            self.last_job_error = None;
177        }
178
179        self.updated_at = Utc::now();
180    }
181}
182
183pub fn parse_cron_syntax(input: &str) -> String {
184    let input = input.trim();
185    // Shorthands
186    if let Some(val) = input.strip_suffix("s") {
187        if let Ok(num) = val.parse::<u32>() {
188            return format!("*/{} * * * * *", num);
189        }
190    }
191    if let Some(val) = input.strip_suffix("m") {
192        if let Ok(num) = val.parse::<u32>() {
193            return format!("0 */{} * * * *", num);
194        }
195    }
196    if let Some(val) = input.strip_suffix("h") {
197        if let Ok(num) = val.parse::<u32>() {
198            return format!("0 0 */{} * * *", num);
199        }
200    }
201    if let Some(val) = input.strip_suffix("d") {
202        if let Ok(num) = val.parse::<u32>() {
203            return format!("0 0 0 */{} * *", num);
204        }
205    }
206    
207    // Check if it's 5 fields (standard cron)
208    let parts: Vec<&str> = input.split_whitespace().collect();
209    if parts.len() == 5 {
210        return format!("0 {} *", input);
211    }
212    if parts.len() == 6 {
213        return format!("{} *", input);
214    }
215    
216    input.to_string()
217}
218
219use std::cmp::Ordering;
220
221#[derive(Clone)]
222pub struct PriorityTask(pub RetryableTask);
223
224impl PartialEq for PriorityTask {
225    fn eq(&self, other: &Self) -> bool {
226        self.0.task_id == other.0.task_id
227    }
228}
229
230impl Eq for PriorityTask {}
231
232impl PartialOrd for PriorityTask {
233    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
234        Some(self.cmp(other))
235    }
236}
237
238impl Ord for PriorityTask {
239    fn cmp(&self, other: &Self) -> Ordering {
240        let score_a = self.0.urgency_score.unwrap_or(0.0);
241        let score_b = other.0.urgency_score.unwrap_or(0.0);
242        
243        // Reverse order so the max score is popped first
244        score_a.partial_cmp(&score_b).unwrap_or(Ordering::Equal)
245    }
246}
snerd_rust/task.rs

snerd_rust/
task.rs
