1use anyhow::Result;
2use clap::Subcommand;
3use sqlx::Row;
4use tracing::info;
5
6use crate::config::Config;
7use crate::utils::database::DatabasePool;
8use crate::utils::display::StatsTable;
9
10#[derive(Subcommand)]
11pub enum QueueCommand {
12 #[command(about = "List all queues")]
13 List {
14 #[arg(short, long, help = "Database connection URL")]
15 database_url: Option<String>,
16 },
17 #[command(about = "Show queue statistics")]
18 Stats {
19 #[arg(short, long, help = "Database connection URL")]
20 database_url: Option<String>,
21 #[arg(short = 'n', long, help = "Specific queue name")]
22 queue: Option<String>,
23 #[arg(long, help = "Show detailed breakdown by priority")]
24 detailed: bool,
25 },
26 #[command(about = "Clear all jobs from a queue")]
27 Clear {
28 #[arg(short, long, help = "Database connection URL")]
29 database_url: Option<String>,
30 #[arg(short = 'n', long, help = "Queue name")]
31 queue: String,
32 #[arg(long, help = "Only clear pending jobs")]
33 pending_only: bool,
34 #[arg(long, help = "Confirm the operation")]
35 confirm: bool,
36 },
37 #[command(about = "Pause a queue (prevent job processing)")]
38 Pause {
39 #[arg(short, long, help = "Database connection URL")]
40 database_url: Option<String>,
41 #[arg(short = 'n', long, help = "Queue name")]
42 queue: String,
43 },
44 #[command(about = "Resume a paused queue")]
45 Resume {
46 #[arg(short, long, help = "Database connection URL")]
47 database_url: Option<String>,
48 #[arg(short = 'n', long, help = "Queue name")]
49 queue: String,
50 },
51 #[command(about = "List all paused queues")]
52 Paused {
53 #[arg(short, long, help = "Database connection URL")]
54 database_url: Option<String>,
55 },
56 #[command(about = "Get queue health status")]
57 Health {
58 #[arg(short, long, help = "Database connection URL")]
59 database_url: Option<String>,
60 #[arg(short = 'n', long, help = "Specific queue name")]
61 queue: Option<String>,
62 },
63}
64
65impl QueueCommand {
66 pub async fn execute(&self, config: &Config) -> Result<()> {
67 let db_url = self.get_database_url(config)?;
68 let pool = DatabasePool::connect(&db_url, config.get_connection_pool_size()).await?;
69
70 match self {
71 QueueCommand::List { .. } => {
72 list_queues(pool).await?;
73 }
74 QueueCommand::Stats {
75 queue, detailed, ..
76 } => {
77 show_queue_stats(pool, queue.clone(), *detailed).await?;
78 }
79 QueueCommand::Clear {
80 queue,
81 pending_only,
82 confirm,
83 ..
84 } => {
85 clear_queue(pool, queue, *pending_only, *confirm).await?;
86 }
87 QueueCommand::Pause { queue, .. } => {
88 pause_queue(pool, queue).await?;
89 }
90 QueueCommand::Resume { queue, .. } => {
91 resume_queue(pool, queue).await?;
92 }
93 QueueCommand::Paused { .. } => {
94 list_paused_queues(pool).await?;
95 }
96 QueueCommand::Health { queue, .. } => {
97 show_queue_health(pool, queue.clone()).await?;
98 }
99 }
100 Ok(())
101 }
102
103 fn get_database_url(&self, config: &Config) -> Result<String> {
104 let url_option = match self {
105 QueueCommand::List { database_url, .. } => database_url,
106 QueueCommand::Stats { database_url, .. } => database_url,
107 QueueCommand::Clear { database_url, .. } => database_url,
108 QueueCommand::Pause { database_url, .. } => database_url,
109 QueueCommand::Resume { database_url, .. } => database_url,
110 QueueCommand::Paused { database_url, .. } => database_url,
111 QueueCommand::Health { database_url, .. } => database_url,
112 };
113
114 url_option
115 .as_ref()
116 .map(|s| s.as_str())
117 .or(config.get_database_url())
118 .map(|s| s.to_string())
119 .ok_or_else(|| anyhow::anyhow!("Database URL is required"))
120 }
121}
122
123async fn list_queues(pool: DatabasePool) -> Result<()> {
124 let query = r#"
125 SELECT
126 j.queue_name,
127 COUNT(j.id) as total_jobs,
128 COUNT(CASE WHEN j.status = 'pending' THEN 1 END) as pending,
129 COUNT(CASE WHEN j.status = 'running' THEN 1 END) as running,
130 COUNT(CASE WHEN j.status = 'completed' THEN 1 END) as completed,
131 COUNT(CASE WHEN j.status = 'failed' THEN 1 END) as failed,
132 COUNT(CASE WHEN j.status = 'dead' THEN 1 END) as dead,
133 CASE WHEN p.queue_name IS NOT NULL THEN true ELSE false END as is_paused
134 FROM hammerwork_jobs j
135 LEFT JOIN hammerwork_queue_pause p ON j.queue_name = p.queue_name
136 GROUP BY j.queue_name, p.queue_name
137 ORDER BY j.queue_name
138 "#;
139
140 let mut table = comfy_table::Table::new();
141 table.set_header(vec![
142 "Queue",
143 "Status",
144 "Total",
145 "Pending",
146 "Running",
147 "Completed",
148 "Failed",
149 "Dead",
150 ]);
151
152 match pool {
153 DatabasePool::Postgres(pg_pool) => {
154 let rows = sqlx::query(query).fetch_all(&pg_pool).await?;
155 for row in rows {
156 let queue_name: String = row.try_get("queue_name")?;
157 let total: i64 = row.try_get("total_jobs")?;
158 let pending: i64 = row.try_get("pending")?;
159 let running: i64 = row.try_get("running")?;
160 let completed: i64 = row.try_get("completed")?;
161 let failed: i64 = row.try_get("failed")?;
162 let dead: i64 = row.try_get("dead")?;
163 let is_paused: bool = row.try_get("is_paused")?;
164
165 let status = if is_paused {
166 "⏸️ Paused"
167 } else {
168 "▶️ Active"
169 };
170
171 table.add_row(vec![
172 queue_name,
173 status.to_string(),
174 total.to_string(),
175 pending.to_string(),
176 running.to_string(),
177 completed.to_string(),
178 failed.to_string(),
179 dead.to_string(),
180 ]);
181 }
182 }
183 DatabasePool::MySQL(mysql_pool) => {
184 let rows = sqlx::query(query).fetch_all(&mysql_pool).await?;
185 for row in rows {
186 let queue_name: String = row.try_get("queue_name")?;
187 let total: i64 = row.try_get("total_jobs")?;
188 let pending: i64 = row.try_get("pending")?;
189 let running: i64 = row.try_get("running")?;
190 let completed: i64 = row.try_get("completed")?;
191 let failed: i64 = row.try_get("failed")?;
192 let dead: i64 = row.try_get("dead")?;
193 let is_paused: bool = row.try_get("is_paused")?;
194
195 let status = if is_paused {
196 "⏸️ Paused"
197 } else {
198 "▶️ Active"
199 };
200
201 table.add_row(vec![
202 queue_name,
203 status.to_string(),
204 total.to_string(),
205 pending.to_string(),
206 running.to_string(),
207 completed.to_string(),
208 failed.to_string(),
209 dead.to_string(),
210 ]);
211 }
212 }
213 }
214
215 println!("📋 Queue Overview");
216 println!("{}", table);
217 Ok(())
218}
219
220async fn show_queue_stats(pool: DatabasePool, queue: Option<String>, detailed: bool) -> Result<()> {
221 if detailed {
222 show_detailed_stats(pool, queue).await
223 } else {
224 show_basic_stats(pool, queue).await
225 }
226}
227
228async fn show_basic_stats(pool: DatabasePool, queue: Option<String>) -> Result<()> {
229 let mut base_query = String::from("SELECT status, COUNT(*) as count FROM hammerwork_jobs");
230
231 if let Some(queue_name) = &queue {
232 base_query.push_str(&format!(" WHERE queue_name = '{}'", queue_name));
233 }
234
235 base_query.push_str(" GROUP BY status ORDER BY status");
236
237 let mut table = comfy_table::Table::new();
238 table.set_header(vec!["Status", "Count"]);
239
240 match &pool {
241 DatabasePool::Postgres(pg_pool) => {
242 let rows = sqlx::query(&base_query).fetch_all(pg_pool).await?;
243 for row in rows {
244 let status: String = row.try_get("status")?;
245 let count: i64 = row.try_get("count")?;
246
247 let status_icon = match status.as_str() {
248 "pending" => "🟡",
249 "running" => "🔵",
250 "completed" => "🟢",
251 "failed" => "🔴",
252 "dead" => "💀",
253 "retrying" => "🟠",
254 _ => "❓",
255 };
256
257 table.add_row(vec![
258 format!("{} {}", status_icon, status),
259 count.to_string(),
260 ]);
261 }
262 }
263 DatabasePool::MySQL(mysql_pool) => {
264 let rows = sqlx::query(&base_query).fetch_all(mysql_pool).await?;
265 for row in rows {
266 let status: String = row.try_get("status")?;
267 let count: i64 = row.try_get("count")?;
268
269 let status_icon = match status.as_str() {
270 "pending" => "🟡",
271 "running" => "🔵",
272 "completed" => "🟢",
273 "failed" => "🔴",
274 "dead" => "💀",
275 "retrying" => "🟠",
276 _ => "❓",
277 };
278
279 table.add_row(vec![
280 format!("{} {}", status_icon, status),
281 count.to_string(),
282 ]);
283 }
284 }
285 }
286
287 println!("📊 Queue Statistics");
288 if let Some(q) = &queue {
289 println!("Queue: {}", q);
290 }
291 println!("{}", table);
292
293 let total_query = if let Some(queue_name) = &queue {
295 format!(
296 "SELECT COUNT(*) as total FROM hammerwork_jobs WHERE queue_name = '{}'",
297 queue_name
298 )
299 } else {
300 "SELECT COUNT(*) as total FROM hammerwork_jobs".to_string()
301 };
302
303 match &pool {
304 DatabasePool::Postgres(pg_pool) => {
305 let total_row = sqlx::query(&total_query).fetch_one(pg_pool).await?;
306 let total: i64 = total_row.try_get("total")?;
307 println!("\n📈 Total jobs: {}", total);
308 }
309 DatabasePool::MySQL(mysql_pool) => {
310 let total_row = sqlx::query(&total_query).fetch_one(mysql_pool).await?;
311 let total: i64 = total_row.try_get("total")?;
312 println!("\n📈 Total jobs: {}", total);
313 }
314 }
315
316 Ok(())
317}
318
319async fn show_detailed_stats(pool: DatabasePool, queue: Option<String>) -> Result<()> {
320 let mut base_query =
321 String::from("SELECT status, priority, COUNT(*) as count FROM hammerwork_jobs");
322
323 if let Some(queue_name) = &queue {
324 base_query.push_str(&format!(" WHERE queue_name = '{}'", queue_name));
325 }
326
327 base_query.push_str(" GROUP BY status, priority ORDER BY status, priority");
328
329 let mut stats_table = StatsTable::new();
330
331 match pool {
332 DatabasePool::Postgres(pg_pool) => {
333 let rows = sqlx::query(&base_query).fetch_all(&pg_pool).await?;
334 for row in rows {
335 let status: String = row.try_get("status")?;
336 let priority: String = row.try_get("priority")?;
337 let count: i64 = row.try_get("count")?;
338
339 stats_table.add_stats_row(&status, &priority, count);
340 }
341 }
342 DatabasePool::MySQL(mysql_pool) => {
343 let rows = sqlx::query(&base_query).fetch_all(&mysql_pool).await?;
344 for row in rows {
345 let status: String = row.try_get("status")?;
346 let priority: String = row.try_get("priority")?;
347 let count: i64 = row.try_get("count")?;
348
349 stats_table.add_stats_row(&status, &priority, count);
350 }
351 }
352 }
353
354 println!("📊 Detailed Queue Statistics");
355 if let Some(q) = &queue {
356 println!("Queue: {}", q);
357 }
358 println!("{}", stats_table);
359
360 Ok(())
361}
362
363async fn clear_queue(
364 pool: DatabasePool,
365 queue: &str,
366 pending_only: bool,
367 confirm: bool,
368) -> Result<()> {
369 if !confirm {
370 println!(
371 "⚠️ This will permanently delete jobs from queue '{}'. Use --confirm to proceed.",
372 queue
373 );
374 return Ok(());
375 }
376
377 let condition = if pending_only {
378 " AND status = 'pending'"
379 } else {
380 ""
381 };
382
383 let affected = match pool {
384 DatabasePool::Postgres(ref pg_pool) => {
385 let query = format!(
386 "DELETE FROM hammerwork_jobs WHERE queue_name = $1{}",
387 condition
388 );
389 let result = sqlx::query(&query).bind(queue).execute(pg_pool).await?;
390 result.rows_affected()
391 }
392 DatabasePool::MySQL(ref mysql_pool) => {
393 let query = format!(
394 "DELETE FROM hammerwork_jobs WHERE queue_name = ?{}",
395 condition
396 );
397 let result = sqlx::query(&query).bind(queue).execute(mysql_pool).await?;
398 result.rows_affected()
399 }
400 };
401
402 let job_type = if pending_only { "pending" } else { "all" };
403 info!(
404 "✅ Cleared {} {} jobs from queue '{}'",
405 affected, job_type, queue
406 );
407 Ok(())
408}
409
410async fn pause_queue(pool: DatabasePool, queue: &str) -> Result<()> {
411 match pool {
412 DatabasePool::Postgres(pg_pool) => {
413 let result = sqlx::query(
414 r#"
415 INSERT INTO hammerwork_queue_pause (queue_name, paused_by, paused_at, created_at, updated_at)
416 VALUES ($1, $2, NOW(), NOW(), NOW())
417 ON CONFLICT (queue_name)
418 DO UPDATE SET
419 paused_by = EXCLUDED.paused_by,
420 paused_at = NOW(),
421 updated_at = NOW()
422 "#,
423 )
424 .bind(queue)
425 .bind("cli")
426 .execute(&pg_pool)
427 .await?;
428
429 if result.rows_affected() > 0 {
430 println!("⏸️ Queue '{}' has been paused", queue);
431 info!("Queue '{}' has been paused via CLI", queue);
432 } else {
433 println!("⚠️ Failed to pause queue '{}'", queue);
434 }
435 }
436 DatabasePool::MySQL(mysql_pool) => {
437 let result = sqlx::query(
438 r#"
439 INSERT INTO hammerwork_queue_pause (queue_name, paused_by, paused_at, created_at, updated_at)
440 VALUES (?, ?, NOW(), NOW(), NOW())
441 ON DUPLICATE KEY UPDATE
442 paused_by = VALUES(paused_by),
443 paused_at = NOW(),
444 updated_at = NOW()
445 "#,
446 )
447 .bind(queue)
448 .bind("cli")
449 .execute(&mysql_pool)
450 .await?;
451
452 if result.rows_affected() > 0 {
453 println!("⏸️ Queue '{}' has been paused", queue);
454 info!("Queue '{}' has been paused via CLI", queue);
455 } else {
456 println!("⚠️ Failed to pause queue '{}'", queue);
457 }
458 }
459 }
460
461 Ok(())
462}
463
464async fn resume_queue(pool: DatabasePool, queue: &str) -> Result<()> {
465 match pool {
466 DatabasePool::Postgres(pg_pool) => {
467 let result = sqlx::query("DELETE FROM hammerwork_queue_pause WHERE queue_name = $1")
468 .bind(queue)
469 .execute(&pg_pool)
470 .await?;
471
472 if result.rows_affected() > 0 {
473 println!("▶️ Queue '{}' has been resumed", queue);
474 info!("Queue '{}' has been resumed via CLI", queue);
475 } else {
476 println!("ℹ️ Queue '{}' was not paused", queue);
477 }
478 }
479 DatabasePool::MySQL(mysql_pool) => {
480 let result = sqlx::query("DELETE FROM hammerwork_queue_pause WHERE queue_name = ?")
481 .bind(queue)
482 .execute(&mysql_pool)
483 .await?;
484
485 if result.rows_affected() > 0 {
486 println!("▶️ Queue '{}' has been resumed", queue);
487 info!("Queue '{}' has been resumed via CLI", queue);
488 } else {
489 println!("ℹ️ Queue '{}' was not paused", queue);
490 }
491 }
492 }
493
494 Ok(())
495}
496
497async fn show_queue_health(pool: DatabasePool, queue: Option<String>) -> Result<()> {
498 let mut health_table = comfy_table::Table::new();
500 health_table.set_header(vec!["Metric", "Value", "Status"]);
501
502 let queue_filter = if let Some(q) = &queue {
503 format!(" WHERE queue_name = '{}'", q)
504 } else {
505 String::new()
506 };
507
508 match pool {
509 DatabasePool::Postgres(pg_pool) => {
510 let total_query = format!(
512 "SELECT COUNT(*) as count FROM hammerwork_jobs{}",
513 queue_filter
514 );
515 let total_row = sqlx::query(&total_query).fetch_one(&pg_pool).await?;
516 let total_jobs: i64 = total_row.try_get("count")?;
517
518 let failed_query = format!(
520 "SELECT COUNT(*) as count FROM hammerwork_jobs{} {} failed_at > NOW() - INTERVAL '1 hour'",
521 queue_filter,
522 if queue_filter.is_empty() {
523 "WHERE"
524 } else {
525 "AND"
526 }
527 );
528 let failed_row = sqlx::query(&failed_query).fetch_one(&pg_pool).await?;
529 let recent_failures: i64 = failed_row.try_get("count")?;
530
531 let long_running_query = format!(
533 "SELECT COUNT(*) as count FROM hammerwork_jobs{} {} status = 'running' AND started_at < NOW() - INTERVAL '1 hour'",
534 queue_filter,
535 if queue_filter.is_empty() {
536 "WHERE"
537 } else {
538 "AND"
539 }
540 );
541 let long_running_row = sqlx::query(&long_running_query).fetch_one(&pg_pool).await?;
542 let long_running: i64 = long_running_row.try_get("count")?;
543
544 health_table.add_row(vec![
546 "Total Jobs".to_string(),
547 total_jobs.to_string(),
548 if total_jobs < 10000 {
549 "🟢 Good"
550 } else {
551 "🟡 High"
552 }
553 .to_string(),
554 ]);
555
556 health_table.add_row(vec![
557 "Recent Failures (1h)".to_string(),
558 recent_failures.to_string(),
559 if recent_failures == 0 {
560 "🟢 Good"
561 } else if recent_failures < 10 {
562 "🟡 Moderate"
563 } else {
564 "🔴 High"
565 }
566 .to_string(),
567 ]);
568
569 health_table.add_row(vec![
570 "Long-running Jobs (>1h)".to_string(),
571 long_running.to_string(),
572 if long_running == 0 {
573 "🟢 Good"
574 } else if long_running < 5 {
575 "🟡 Moderate"
576 } else {
577 "🔴 High"
578 }
579 .to_string(),
580 ]);
581 }
582 DatabasePool::MySQL(mysql_pool) => {
583 let total_query = format!(
585 "SELECT COUNT(*) as count FROM hammerwork_jobs{}",
586 queue_filter
587 );
588 let total_row = sqlx::query(&total_query).fetch_one(&mysql_pool).await?;
589 let total_jobs: i64 = total_row.try_get("count")?;
590
591 let failed_query = format!(
592 "SELECT COUNT(*) as count FROM hammerwork_jobs{} {} failed_at > DATE_SUB(NOW(), INTERVAL 1 HOUR)",
593 queue_filter,
594 if queue_filter.is_empty() {
595 "WHERE"
596 } else {
597 "AND"
598 }
599 );
600 let failed_row = sqlx::query(&failed_query).fetch_one(&mysql_pool).await?;
601 let recent_failures: i64 = failed_row.try_get("count")?;
602
603 let long_running_query = format!(
604 "SELECT COUNT(*) as count FROM hammerwork_jobs{} {} status = 'running' AND started_at < DATE_SUB(NOW(), INTERVAL 1 HOUR)",
605 queue_filter,
606 if queue_filter.is_empty() {
607 "WHERE"
608 } else {
609 "AND"
610 }
611 );
612 let long_running_row = sqlx::query(&long_running_query)
613 .fetch_one(&mysql_pool)
614 .await?;
615 let long_running: i64 = long_running_row.try_get("count")?;
616
617 health_table.add_row(vec![
618 "Total Jobs".to_string(),
619 total_jobs.to_string(),
620 if total_jobs < 10000 {
621 "🟢 Good"
622 } else {
623 "🟡 High"
624 }
625 .to_string(),
626 ]);
627
628 health_table.add_row(vec![
629 "Recent Failures (1h)".to_string(),
630 recent_failures.to_string(),
631 if recent_failures == 0 {
632 "🟢 Good"
633 } else if recent_failures < 10 {
634 "🟡 Moderate"
635 } else {
636 "🔴 High"
637 }
638 .to_string(),
639 ]);
640
641 health_table.add_row(vec![
642 "Long-running Jobs (>1h)".to_string(),
643 long_running.to_string(),
644 if long_running == 0 {
645 "🟢 Good"
646 } else if long_running < 5 {
647 "🟡 Moderate"
648 } else {
649 "🔴 High"
650 }
651 .to_string(),
652 ]);
653 }
654 }
655
656 println!("🏥 Queue Health");
657 if let Some(q) = &queue {
658 println!("Queue: {}", q);
659 }
660 println!("{}", health_table);
661
662 Ok(())
663}
664
665async fn list_paused_queues(pool: DatabasePool) -> Result<()> {
666 let query = r#"
667 SELECT
668 queue_name,
669 paused_at,
670 paused_by,
671 reason
672 FROM hammerwork_queue_pause
673 ORDER BY paused_at DESC
674 "#;
675
676 let mut table = comfy_table::Table::new();
677 table.set_header(vec!["Queue Name", "Paused At", "Paused By", "Reason"]);
678
679 match pool {
680 DatabasePool::Postgres(pg_pool) => {
681 let rows = sqlx::query(query).fetch_all(&pg_pool).await?;
682
683 if rows.is_empty() {
684 println!("✅ No paused queues found - all queues are active");
685 return Ok(());
686 }
687
688 for row in rows {
689 let queue_name: String = row.try_get("queue_name")?;
690 let paused_at: chrono::DateTime<chrono::Utc> = row.try_get("paused_at")?;
691 let paused_by: Option<String> = row.try_get("paused_by")?;
692 let reason: Option<String> = row.try_get("reason")?;
693
694 table.add_row(vec![
695 queue_name,
696 paused_at.format("%Y-%m-%d %H:%M:%S UTC").to_string(),
697 paused_by.unwrap_or_else(|| "Unknown".to_string()),
698 reason.unwrap_or_else(|| "-".to_string()),
699 ]);
700 }
701 }
702 DatabasePool::MySQL(mysql_pool) => {
703 let rows = sqlx::query(query).fetch_all(&mysql_pool).await?;
704
705 if rows.is_empty() {
706 println!("✅ No paused queues found - all queues are active");
707 return Ok(());
708 }
709
710 for row in rows {
711 let queue_name: String = row.try_get("queue_name")?;
712 let paused_at: chrono::DateTime<chrono::Utc> = row.try_get("paused_at")?;
713 let paused_by: Option<String> = row.try_get("paused_by")?;
714 let reason: Option<String> = row.try_get("reason")?;
715
716 table.add_row(vec![
717 queue_name,
718 paused_at.format("%Y-%m-%d %H:%M:%S UTC").to_string(),
719 paused_by.unwrap_or_else(|| "Unknown".to_string()),
720 reason.unwrap_or_else(|| "-".to_string()),
721 ]);
722 }
723 }
724 }
725
726 println!("⏸️ Paused Queues");
727 println!("{}", table);
728 Ok(())
729}