1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
use crate::error::{Error, Result};
use crate::options::SqsQueueConfig; // Use the struct from options.rs
use crate::queue::{ArcJobProcessorFn, QueueInterface};
use crate::webhook::sender::JobProcessorFnAsync;
use async_trait::async_trait;
use aws_sdk_sqs as sqs;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::interval;
use tracing::{error, info};
/// SQS-based implementation of the QueueInterface
pub struct SqsQueueManager {
/// SQS client
client: sqs::Client,
/// Configuration
config: SqsQueueConfig, // Using the struct from options.rs
// todo -> change Mutex<Hashmap> to Dashmap
/// Cache of queue URLs
queue_urls: Arc<Mutex<HashMap<String, String>>>,
/// Active workers
worker_handles: Arc<Mutex<HashMap<String, Vec<tokio::task::JoinHandle<()>>>>>,
/// Flag to control worker shutdown
shutdown: Arc<Mutex<bool>>,
}
impl SqsQueueManager {
/// Create a new SQS queue manager
pub async fn new(config: SqsQueueConfig) -> Result<Self> {
// Build AWS config
let mut aws_config_builder = aws_config::from_env();
// Set region
aws_config_builder =
aws_config_builder.region(aws_sdk_sqs::config::Region::new(config.region.clone()));
// Set custom endpoint if provided (for local testing)
if let Some(endpoint) = &config.endpoint_url {
aws_config_builder = aws_config_builder.endpoint_url(endpoint);
}
// Build the AWS config
let aws_config = aws_config_builder.load().await;
// Create SQS client
let client = sqs::Client::new(&aws_config);
Ok(Self {
client,
config,
queue_urls: Arc::new(Mutex::new(HashMap::new())),
worker_handles: Arc::new(Mutex::new(HashMap::new())),
shutdown: Arc::new(Mutex::new(false)),
})
}
/// Get or create the URL for a queue
async fn get_queue_url(&self, queue_name: &str) -> Result<String> {
// Check cache first
let cached_url = {
let queue_urls = self.queue_urls.lock().await;
queue_urls.get(queue_name).cloned()
};
if let Some(url) = cached_url {
return Ok(url);
}
// If custom prefix is provided, use that to construct queue URL
if let Some(prefix) = &self.config.queue_url_prefix {
let queue_url = if self.config.fifo {
format!("{}/{}.fifo", prefix, queue_name)
} else {
format!("{}/{}", prefix, queue_name)
};
// Cache the URL
let mut queue_urls = self.queue_urls.lock().await;
queue_urls.insert(queue_name.to_string(), queue_url.clone());
return Ok(queue_url);
}
// Try to get the queue URL from AWS
let actual_queue_name = if self.config.fifo && !queue_name.ends_with(".fifo") {
format!("{}.fifo", queue_name)
} else {
queue_name.to_string()
};
match self
.client
.get_queue_url()
.queue_name(&actual_queue_name)
.send()
.await
{
Ok(output) => {
if let Some(url) = output.queue_url() {
// Cache the URL for future use
let mut queue_urls = self.queue_urls.lock().await;
queue_urls.insert(queue_name.to_string(), url.to_string());
Ok(url.to_string())
} else {
Err(Error::Queue(format!(
"No URL returned for queue: {}",
queue_name
)))
}
}
Err(e) => {
// If queue doesn't exist, try to create it
if e.to_string().contains("QueueDoesNotExist")
|| e.to_string().contains("NonExistentQueue")
{
self.create_queue(queue_name).await
} else {
Err(Error::Queue(format!("Failed to get queue URL: {}", e)))
}
}
}
}
/// Create a queue if it doesn't exist
async fn create_queue(&self, queue_name: &str) -> Result<String> {
info!("{}", format!("Creating SQS queue: {}", queue_name));
let actual_queue_name = if self.config.fifo && !queue_name.ends_with(".fifo") {
format!("{}.fifo", queue_name)
} else {
queue_name.to_string()
};
// Set up attributes for the queue using proper QueueAttributeName enum
let mut attributes = HashMap::new();
// Use the enum type for attribute names
use aws_sdk_sqs::types::QueueAttributeName;
attributes.insert(
QueueAttributeName::VisibilityTimeout,
self.config.visibility_timeout.to_string(),
);
if self.config.fifo {
attributes.insert(QueueAttributeName::FifoQueue, "true".to_string());
// Content-based deduplication for FIFO queues
attributes.insert(
QueueAttributeName::ContentBasedDeduplication,
"true".to_string(),
);
}
// Create the queue
let result = self
.client
.create_queue()
.queue_name(&actual_queue_name)
.set_attributes(Some(attributes))
.send()
.await
.map_err(|e| {
Error::Queue(format!("Failed to create SQS queue {}: {}", queue_name, e))
})?;
if let Some(url) = result.queue_url() {
// Cache the URL
let mut queue_urls = self.queue_urls.lock().await;
queue_urls.insert(queue_name.to_string(), url.to_string());
Ok(url.to_string())
} else {
Err(Error::Queue(format!(
"No URL returned after creating queue: {}",
queue_name
)))
}
}
/// Start a worker for processing messages from the queue
async fn start_worker(
&self,
queue_name: &str,
queue_url: String,
processor: ArcJobProcessorFn,
worker_id: usize,
) -> tokio::task::JoinHandle<()> {
// Clone values needed for the worker
let client = self.client.clone();
let config = self.config.clone();
let shutdown = self.shutdown.clone();
let queue_name = queue_name.to_string();
// Spawn the worker task
tokio::spawn(async move {
info!(
"{}",
format!(
"Starting SQS worker #{} for queue: {}",
worker_id, queue_name
)
);
let mut interval = interval(Duration::from_secs(1));
loop {
interval.tick().await;
// Check if we should shutdown
if *shutdown.lock().await {
info!(
"{}",
format!(
"SQS worker #{} for queue {} shutting down",
worker_id, queue_name
)
);
break;
}
// Receive messages from the queue
let result = client
.receive_message()
.queue_url(&queue_url)
.max_number_of_messages(config.max_messages)
.visibility_timeout(config.visibility_timeout)
.wait_time_seconds(config.wait_time_seconds)
.send()
.await;
match result {
Ok(response) => {
// Get messages from the response - messages() returns Option<&[Message]>
let messages = response.messages();
// Check if we received any messages
if !messages.is_empty() {
info!(
"{}",
format!(
"SQS worker #{} received {} messages from queue {}",
worker_id,
messages.len(),
queue_name
)
);
// Iterate through the messages in the slice
for message in messages {
if let Some(body) = message.body() {
// Process the message
match serde_json::from_str::<crate::webhook::types::JobData>(
body,
) {
Ok(job_data) => {
// Call the processor
match processor(job_data).await {
Ok(_) => {
// Processing succeeded, delete the message
if let Some(receipt_handle) =
message.receipt_handle()
{
if let Err(e) = client
.delete_message()
.queue_url(&queue_url)
.receipt_handle(receipt_handle)
.send()
.await
{
error!(
"{}",
format!(
"Failed to delete message from SQS queue {}: {}",
queue_name, e
)
);
}
}
}
Err(e) => {
// Processing failed, log the error
error!(
"{}",
format!(
"Error processing message from SQS queue {}: {}",
queue_name, e
)
);
}
}
}
Err(e) => {
error!(
"{}",
format!(
"Failed to deserialize message from SQS queue {}: {}",
queue_name, e
)
);
// Delete malformed messages
if let Some(receipt_handle) = message.receipt_handle() {
if let Err(e) = client
.delete_message()
.queue_url(&queue_url)
.receipt_handle(receipt_handle)
.send()
.await
{
error!(
"{}",
format!(
"Failed to delete malformed message from SQS queue {}: {}",
queue_name, e
)
);
}
}
}
}
}
}
}
}
Err(e) => {
error!(
"{}",
format!(
"Failed to receive messages from SQS queue {}: {}",
queue_name, e
)
);
// Add a small delay before retrying on error
tokio::time::sleep(Duration::from_secs(1)).await;
}
}
}
})
}
}
#[async_trait]
impl QueueInterface for SqsQueueManager {
/// Add a job to a queue
async fn add_to_queue(
&self,
queue_name: &str,
data: crate::webhook::types::JobData,
) -> Result<()> {
// Get the queue URL
let queue_url = self.get_queue_url(queue_name).await?;
// Serialize the job data
let data_json = serde_json::to_string(&data)
.map_err(|e| Error::Queue(format!("Failed to serialize job data: {}", e)))?;
// Build send message request
let mut send_message_request = self
.client
.send_message()
.queue_url(queue_url)
.message_body(data_json);
// Add FIFO-specific attributes if needed
if self.config.fifo {
// For FIFO queues, we need a message group ID and deduplication ID
if let Some(group_id) = &self.config.message_group_id {
send_message_request = send_message_request.message_group_id(group_id);
}
// Use a unique ID based on the request ID for deduplication
// Only needed if ContentBasedDeduplication is not enabled
// send_message_request = send_message_request.message_deduplication_id(&data.request_id);
}
// Send the message to SQS
let result = send_message_request.send().await.map_err(|e| {
Error::Queue(format!(
"Failed to send message to SQS queue {}: {}",
queue_name, e
))
})?;
info!(
"{}",
format!(
"Added job to SQS queue {} with ID: {}",
queue_name,
result.message_id().unwrap_or("unknown")
)
);
Ok(())
}
/// Process jobs from a queue
async fn process_queue(&self, queue_name: &str, callback: JobProcessorFnAsync) -> Result<()> {
// Get the queue URL
let queue_url = self.get_queue_url(queue_name).await?;
// Wrap the callback in an Arc for thread-safe sharing
let processor: ArcJobProcessorFn = Arc::from(callback);
// Start workers
let mut worker_handles = Vec::new();
let concurrency = self.config.concurrency as usize;
for worker_id in 0..concurrency {
let handle = self
.start_worker(queue_name, queue_url.clone(), processor.clone(), worker_id)
.await;
worker_handles.push(handle);
}
// Store the worker handles
let mut handles = self.worker_handles.lock().await;
handles.insert(queue_name.to_string(), worker_handles);
info!(
"{}",
format!(
"Started {} workers for SQS queue: {}",
concurrency, queue_name
)
);
Ok(())
}
/// Disconnect and clean up
async fn disconnect(&self) -> Result<()> {
// Signal workers to shutdown
{
let mut shutdown = self.shutdown.lock().await;
*shutdown = true;
}
// Wait for workers to finish
{
let mut handles = self.worker_handles.lock().await;
for (queue_name, workers) in handles.drain() {
info!(
"{}",
format!("Waiting for SQS queue {} workers to shutdown", queue_name)
);
for handle in workers {
// We don't want to await here as it could block indefinitely
// Just detach the tasks and let them complete on their own
handle.abort();
}
}
}
Ok(())
}
}
impl Drop for SqsQueueManager {
fn drop(&mut self) {
// Signal workers to shutdown when the manager is dropped
// We can't use async functions in drop, so we spawn a task
let shutdown = self.shutdown.clone();
tokio::spawn(async move {
let mut lock = shutdown.lock().await;
*lock = true;
});
}
}