1use anyhow::Result;
2use clap::Subcommand;
3use tracing::info;
4
5use crate::config::Config;
6use crate::utils::database::DatabasePool;
7
8#[derive(Subcommand)]
9pub enum ArchiveCommand {
10 #[command(about = "Archive jobs based on policy")]
11 Run {
12 #[arg(short = 'u', long, help = "Database connection URL")]
13 database_url: Option<String>,
14 #[arg(
15 short = 'q',
16 long,
17 help = "Queue name to archive (all queues if not specified)"
18 )]
19 queue_name: Option<String>,
20 #[arg(
21 long,
22 default_value = "7",
23 help = "Days to keep completed jobs before archiving"
24 )]
25 completed_after_days: u32,
26 #[arg(
27 long,
28 default_value = "30",
29 help = "Days to keep failed jobs before archiving"
30 )]
31 failed_after_days: u32,
32 #[arg(
33 long,
34 default_value = "30",
35 help = "Days to keep dead jobs before archiving"
36 )]
37 dead_after_days: u32,
38 #[arg(
39 long,
40 default_value = "30",
41 help = "Days to keep timed out jobs before archiving"
42 )]
43 timed_out_after_days: u32,
44 #[arg(
45 long,
46 default_value = "1000",
47 help = "Maximum number of jobs to archive in one batch"
48 )]
49 batch_size: usize,
50 #[arg(
51 long,
52 default_value = "true",
53 help = "Whether to compress archived payloads"
54 )]
55 compress: bool,
56 #[arg(long, default_value = "6", help = "Compression level (0-9)")]
57 compression_level: u32,
58 #[arg(long, help = "Dry run - show what would be archived without archiving")]
59 dry_run: bool,
60 #[arg(long, help = "Reason for archival")]
61 reason: Option<String>,
62 #[arg(long, help = "Who initiated the archival")]
63 archived_by: Option<String>,
64 },
65 #[command(about = "Restore an archived job back to the active queue")]
66 Restore {
67 #[arg(short = 'u', long, help = "Database connection URL")]
68 database_url: Option<String>,
69 #[arg(help = "Job ID to restore")]
70 job_id: String,
71 },
72 #[command(about = "List archived jobs")]
73 List {
74 #[arg(short = 'u', long, help = "Database connection URL")]
75 database_url: Option<String>,
76 #[arg(short = 'q', long, help = "Filter by queue name")]
77 queue_name: Option<String>,
78 #[arg(
79 short = 'l',
80 long,
81 default_value = "100",
82 help = "Maximum number of jobs to list"
83 )]
84 limit: u32,
85 #[arg(
86 short = 'o',
87 long,
88 default_value = "0",
89 help = "Number of jobs to skip"
90 )]
91 offset: u32,
92 #[arg(long, help = "Output format (table, json, csv)")]
93 format: Option<String>,
94 },
95 #[command(about = "Get archival statistics")]
96 Stats {
97 #[arg(short = 'u', long, help = "Database connection URL")]
98 database_url: Option<String>,
99 #[arg(short = 'q', long, help = "Filter by queue name")]
100 queue_name: Option<String>,
101 #[arg(long, help = "Output format (table, json)")]
102 format: Option<String>,
103 },
104 #[command(about = "Permanently delete archived jobs")]
105 Purge {
106 #[arg(short = 'u', long, help = "Database connection URL")]
107 database_url: Option<String>,
108 #[arg(long, help = "Delete archived jobs older than this many days")]
109 older_than_days: u32,
110 #[arg(long, help = "Confirm the purge operation")]
111 confirm: bool,
112 #[arg(long, help = "Dry run - show what would be deleted")]
113 dry_run: bool,
114 },
115 #[command(about = "Set archival policy for a queue")]
116 SetPolicy {
117 #[arg(short = 'u', long, help = "Database connection URL")]
118 database_url: Option<String>,
119 #[arg(short = 'q', long, help = "Queue name")]
120 queue_name: String,
121 #[arg(long, help = "Days to keep completed jobs before archiving")]
122 completed_after_days: Option<u32>,
123 #[arg(long, help = "Days to keep failed jobs before archiving")]
124 failed_after_days: Option<u32>,
125 #[arg(long, help = "Days to keep dead jobs before archiving")]
126 dead_after_days: Option<u32>,
127 #[arg(long, help = "Days to keep timed out jobs before archiving")]
128 timed_out_after_days: Option<u32>,
129 #[arg(long, help = "Maximum number of jobs to archive in one batch")]
130 batch_size: Option<usize>,
131 #[arg(long, help = "Whether to compress archived payloads")]
132 compress: Option<bool>,
133 #[arg(long, help = "Enable or disable the policy")]
134 enabled: Option<bool>,
135 },
136 #[command(about = "Get archival policy for a queue")]
137 GetPolicy {
138 #[arg(short = 'u', long, help = "Database connection URL")]
139 database_url: Option<String>,
140 #[arg(short = 'q', long, help = "Queue name")]
141 queue_name: String,
142 #[arg(long, help = "Output format (table, json)")]
143 format: Option<String>,
144 },
145 #[command(about = "Remove archival policy for a queue")]
146 RemovePolicy {
147 #[arg(short = 'u', long, help = "Database connection URL")]
148 database_url: Option<String>,
149 #[arg(short = 'q', long, help = "Queue name")]
150 queue_name: String,
151 #[arg(long, help = "Confirm the removal")]
152 confirm: bool,
153 },
154}
155
156impl ArchiveCommand {
157 pub async fn execute(&self, config: &Config) -> Result<()> {
158 let db_url = self.get_database_url(config)?;
159 let pool = DatabasePool::connect(&db_url, config.get_connection_pool_size()).await?;
160
161 match self {
162 ArchiveCommand::Run {
163 queue_name,
164 completed_after_days,
165 failed_after_days,
166 dead_after_days,
167 timed_out_after_days,
168 batch_size,
169 compress,
170 compression_level,
171 dry_run,
172 reason,
173 archived_by,
174 ..
175 } => {
176 archive_jobs(
177 pool,
178 queue_name.as_deref(),
179 *completed_after_days,
180 *failed_after_days,
181 *dead_after_days,
182 *timed_out_after_days,
183 *batch_size,
184 *compress,
185 *compression_level,
186 *dry_run,
187 reason.as_deref(),
188 archived_by.as_deref(),
189 )
190 .await?;
191 }
192 ArchiveCommand::Restore { job_id, .. } => {
193 restore_archived_job(pool, job_id).await?;
194 }
195 ArchiveCommand::List {
196 queue_name,
197 limit,
198 offset,
199 format,
200 ..
201 } => {
202 list_archived_jobs(
203 pool,
204 queue_name.as_deref(),
205 *limit,
206 *offset,
207 format.as_deref().unwrap_or("table"),
208 )
209 .await?;
210 }
211 ArchiveCommand::Stats {
212 queue_name, format, ..
213 } => {
214 show_archival_stats(
215 pool,
216 queue_name.as_deref(),
217 format.as_deref().unwrap_or("table"),
218 )
219 .await?;
220 }
221 ArchiveCommand::Purge {
222 older_than_days,
223 confirm,
224 dry_run,
225 ..
226 } => {
227 purge_archived_jobs(pool, *older_than_days, *confirm, *dry_run).await?;
228 }
229 ArchiveCommand::SetPolicy {
230 queue_name,
231 completed_after_days,
232 failed_after_days,
233 dead_after_days,
234 timed_out_after_days,
235 batch_size,
236 compress,
237 enabled,
238 ..
239 } => {
240 set_archival_policy(
241 pool,
242 queue_name,
243 *completed_after_days,
244 *failed_after_days,
245 *dead_after_days,
246 *timed_out_after_days,
247 *batch_size,
248 *compress,
249 *enabled,
250 )
251 .await?;
252 }
253 ArchiveCommand::GetPolicy {
254 queue_name, format, ..
255 } => {
256 get_archival_policy(pool, queue_name, format.as_deref().unwrap_or("table")).await?;
257 }
258 ArchiveCommand::RemovePolicy {
259 queue_name,
260 confirm,
261 ..
262 } => {
263 remove_archival_policy(pool, queue_name, *confirm).await?;
264 }
265 }
266
267 Ok(())
268 }
269
270 fn get_database_url(&self, config: &Config) -> Result<String> {
271 match self {
272 ArchiveCommand::Run { database_url, .. } => database_url.as_ref(),
273 ArchiveCommand::Restore { database_url, .. } => database_url.as_ref(),
274 ArchiveCommand::List { database_url, .. } => database_url.as_ref(),
275 ArchiveCommand::Stats { database_url, .. } => database_url.as_ref(),
276 ArchiveCommand::Purge { database_url, .. } => database_url.as_ref(),
277 ArchiveCommand::SetPolicy { database_url, .. } => database_url.as_ref(),
278 ArchiveCommand::GetPolicy { database_url, .. } => database_url.as_ref(),
279 ArchiveCommand::RemovePolicy { database_url, .. } => database_url.as_ref(),
280 }
281 .cloned()
282 .or_else(|| config.get_database_url().map(|s| s.to_string()))
283 .ok_or_else(|| {
284 anyhow::anyhow!("Database URL not provided. Use --database-url or set in config file")
285 })
286 }
287}
288
289async fn archive_jobs(
291 pool: DatabasePool,
292 queue_name: Option<&str>,
293 completed_after_days: u32,
294 failed_after_days: u32,
295 dead_after_days: u32,
296 timed_out_after_days: u32,
297 batch_size: usize,
298 compress: bool,
299 compression_level: u32,
300 dry_run: bool,
301 reason: Option<&str>,
302 archived_by: Option<&str>,
303) -> Result<()> {
304 use chrono::Duration;
305 use hammerwork::archive::{ArchivalConfig, ArchivalPolicy, ArchivalReason};
306 use hammerwork::queue::DatabaseQueue;
307
308 info!("Running job archival...");
309
310 let policy = ArchivalPolicy::new()
311 .archive_completed_after(Duration::days(completed_after_days as i64))
312 .archive_failed_after(Duration::days(failed_after_days as i64))
313 .archive_dead_after(Duration::days(dead_after_days as i64))
314 .archive_timed_out_after(Duration::days(timed_out_after_days as i64))
315 .with_batch_size(batch_size)
316 .compress_archived_payloads(compress)
317 .enabled(!dry_run);
318
319 let config = ArchivalConfig::new().with_compression_level(compression_level);
320
321 let archival_reason = reason
322 .map(|r| match r.to_lowercase().as_str() {
323 "manual" => ArchivalReason::Manual,
324 "compliance" => ArchivalReason::Compliance,
325 "maintenance" => ArchivalReason::Maintenance,
326 _ => ArchivalReason::Automatic,
327 })
328 .unwrap_or(ArchivalReason::Automatic);
329
330 if dry_run {
331 println!("DRY RUN: Would archive jobs with the following policy:");
332 println!(" Queue: {}", queue_name.unwrap_or("ALL"));
333 println!(" Completed after: {} days", completed_after_days);
334 println!(" Failed after: {} days", failed_after_days);
335 println!(" Dead after: {} days", dead_after_days);
336 println!(" Timed out after: {} days", timed_out_after_days);
337 println!(" Batch size: {}", batch_size);
338 println!(" Compress: {}", compress);
339 println!(" Compression level: {}", compression_level);
340 println!(" Reason: {:?}", archival_reason);
341 println!(" Archived by: {}", archived_by.unwrap_or("CLI"));
342 return Ok(());
343 }
344
345 let stats = match pool {
346 DatabasePool::Postgres(pool) => {
347 let queue = hammerwork::JobQueue::new(pool);
348 queue
349 .archive_jobs(queue_name, &policy, &config, archival_reason, archived_by)
350 .await?
351 }
352 DatabasePool::MySQL(pool) => {
353 let queue = hammerwork::JobQueue::new(pool);
354 queue
355 .archive_jobs(queue_name, &policy, &config, archival_reason, archived_by)
356 .await?
357 }
358 };
359
360 println!("Archival completed successfully!");
361 println!(" Jobs archived: {}", stats.jobs_archived);
362 println!(" Bytes archived: {}", stats.bytes_archived);
363 println!(" Compression ratio: {:.2}", stats.compression_ratio);
364 println!(" Duration: {:?}", stats.operation_duration);
365
366 Ok(())
367}
368
369async fn restore_archived_job(pool: DatabasePool, job_id: &str) -> Result<()> {
370 use hammerwork::queue::DatabaseQueue;
371 use uuid::Uuid;
372
373 info!("Restoring archived job: {}", job_id);
374
375 let job_uuid = Uuid::parse_str(job_id)?;
376
377 let job = match pool {
378 DatabasePool::Postgres(pool) => {
379 let queue = hammerwork::JobQueue::new(pool);
380 queue.restore_archived_job(job_uuid).await?
381 }
382 DatabasePool::MySQL(pool) => {
383 let queue = hammerwork::JobQueue::new(pool);
384 queue.restore_archived_job(job_uuid).await?
385 }
386 };
387
388 println!("Job restored successfully!");
389 println!(" Job ID: {}", job.id);
390 println!(" Queue: {}", job.queue_name);
391 println!(" Status: {:?}", job.status);
392 println!(" Scheduled at: {}", job.scheduled_at);
393
394 Ok(())
395}
396
397async fn list_archived_jobs(
398 pool: DatabasePool,
399 queue_name: Option<&str>,
400 limit: u32,
401 offset: u32,
402 format: &str,
403) -> Result<()> {
404 use hammerwork::queue::DatabaseQueue;
405
406 info!("Listing archived jobs...");
407
408 let archived_jobs = match pool {
409 DatabasePool::Postgres(pool) => {
410 let queue = hammerwork::JobQueue::new(pool);
411 queue
412 .list_archived_jobs(queue_name, Some(limit), Some(offset))
413 .await?
414 }
415 DatabasePool::MySQL(pool) => {
416 let queue = hammerwork::JobQueue::new(pool);
417 queue
418 .list_archived_jobs(queue_name, Some(limit), Some(offset))
419 .await?
420 }
421 };
422
423 match format {
424 "json" => {
425 println!("{}", serde_json::to_string_pretty(&archived_jobs)?);
426 }
427 "csv" => {
428 println!(
429 "id,queue_name,status,created_at,archived_at,reason,payload_compressed,archived_by"
430 );
431 for job in archived_jobs {
432 println!(
433 "{},{},{:?},{},{},{:?},{},{}",
434 job.id,
435 job.queue_name,
436 job.status,
437 job.created_at,
438 job.archived_at,
439 job.archival_reason,
440 job.payload_compressed,
441 job.archived_by.unwrap_or_default()
442 );
443 }
444 }
445 _ => {
446 use comfy_table::{Attribute, Cell, ContentArrangement, Table, presets::UTF8_FULL};
448
449 let mut table = Table::new();
450 table
451 .load_preset(UTF8_FULL)
452 .set_content_arrangement(ContentArrangement::Dynamic)
453 .set_header(vec![
454 Cell::new("Job ID").add_attribute(Attribute::Bold),
455 Cell::new("Queue").add_attribute(Attribute::Bold),
456 Cell::new("Status").add_attribute(Attribute::Bold),
457 Cell::new("Created At").add_attribute(Attribute::Bold),
458 Cell::new("Archived At").add_attribute(Attribute::Bold),
459 Cell::new("Reason").add_attribute(Attribute::Bold),
460 Cell::new("Compressed").add_attribute(Attribute::Bold),
461 Cell::new("Archived By").add_attribute(Attribute::Bold),
462 ]);
463
464 for job in archived_jobs {
465 table.add_row(vec![
466 Cell::new(job.id.to_string()),
467 Cell::new(&job.queue_name),
468 Cell::new(format!("{:?}", job.status)),
469 Cell::new(job.created_at.format("%Y-%m-%d %H:%M:%S").to_string()),
470 Cell::new(job.archived_at.format("%Y-%m-%d %H:%M:%S").to_string()),
471 Cell::new(format!("{:?}", job.archival_reason)),
472 Cell::new(if job.payload_compressed { "Yes" } else { "No" }),
473 Cell::new(job.archived_by.unwrap_or_default()),
474 ]);
475 }
476
477 println!("{table}");
478 }
479 }
480
481 Ok(())
482}
483
484async fn show_archival_stats(
485 pool: DatabasePool,
486 queue_name: Option<&str>,
487 format: &str,
488) -> Result<()> {
489 use hammerwork::queue::DatabaseQueue;
490
491 info!("Getting archival statistics...");
492
493 let stats = match pool {
494 DatabasePool::Postgres(pool) => {
495 let queue = hammerwork::JobQueue::new(pool);
496 queue.get_archival_stats(queue_name).await?
497 }
498 DatabasePool::MySQL(pool) => {
499 let queue = hammerwork::JobQueue::new(pool);
500 queue.get_archival_stats(queue_name).await?
501 }
502 };
503
504 match format {
505 "json" => {
506 println!("{}", serde_json::to_string_pretty(&stats)?);
507 }
508 _ => {
509 println!("Archival Statistics");
510 println!("==================");
511 if let Some(queue) = queue_name {
512 println!("Queue: {}", queue);
513 } else {
514 println!("Queue: ALL");
515 }
516 println!("Jobs archived: {}", stats.jobs_archived);
517 println!("Jobs purged: {}", stats.jobs_purged);
518 println!("Bytes archived: {}", stats.bytes_archived);
519 println!("Bytes purged: {}", stats.bytes_purged);
520 println!("Compression ratio: {:.2}", stats.compression_ratio);
521 println!("Last run at: {}", stats.last_run_at);
522 }
523 }
524
525 Ok(())
526}
527
528async fn purge_archived_jobs(
529 pool: DatabasePool,
530 older_than_days: u32,
531 confirm: bool,
532 dry_run: bool,
533) -> Result<()> {
534 use chrono::{Duration, Utc};
535 use hammerwork::queue::DatabaseQueue;
536
537 info!(
538 "Purging archived jobs older than {} days...",
539 older_than_days
540 );
541
542 let cutoff_date = Utc::now() - Duration::days(older_than_days as i64);
543
544 if !confirm && !dry_run {
545 return Err(anyhow::anyhow!(
546 "Purge operation requires --confirm flag or --dry-run"
547 ));
548 }
549
550 if dry_run {
551 println!(
552 "DRY RUN: Would permanently delete archived jobs older than {}",
553 cutoff_date
554 );
555 return Ok(());
556 }
557
558 let deleted_count = match pool {
559 DatabasePool::Postgres(pool) => {
560 let queue = hammerwork::JobQueue::new(pool);
561 queue.purge_archived_jobs(cutoff_date).await?
562 }
563 DatabasePool::MySQL(pool) => {
564 let queue = hammerwork::JobQueue::new(pool);
565 queue.purge_archived_jobs(cutoff_date).await?
566 }
567 };
568
569 println!("Purged {} archived jobs", deleted_count);
570
571 Ok(())
572}
573
574async fn set_archival_policy(
576 _pool: DatabasePool,
577 queue_name: &str,
578 _completed_after_days: Option<u32>,
579 _failed_after_days: Option<u32>,
580 _dead_after_days: Option<u32>,
581 _timed_out_after_days: Option<u32>,
582 _batch_size: Option<usize>,
583 _compress: Option<bool>,
584 _enabled: Option<bool>,
585) -> Result<()> {
586 println!("Setting archival policy for queue: {}", queue_name);
589 println!("Note: Policy management requires additional configuration storage");
590 println!("Use the 'run' command with specific parameters for one-time archival operations");
591 Ok(())
592}
593
594async fn get_archival_policy(_pool: DatabasePool, queue_name: &str, _format: &str) -> Result<()> {
595 println!("Getting archival policy for queue: {}", queue_name);
596 println!("Note: Policy management requires additional configuration storage");
597 println!("No persistent policies are currently configured");
598 Ok(())
599}
600
601async fn remove_archival_policy(
602 _pool: DatabasePool,
603 queue_name: &str,
604 _confirm: bool,
605) -> Result<()> {
606 println!("Removing archival policy for queue: {}", queue_name);
607 println!("Note: Policy management requires additional configuration storage");
608 println!("No persistent policies are currently configured");
609 Ok(())
610}
611
612#[cfg(test)]
613mod tests {
614 use super::*;
615
616 #[test]
617 fn test_archive_command_parsing() {
618 use clap::Parser;
619
620 #[derive(Parser)]
621 struct TestApp {
622 #[command(subcommand)]
623 command: ArchiveCommand,
624 }
625
626 let app = TestApp::try_parse_from(&[
627 "test",
628 "run",
629 "--completed-after-days",
630 "7",
631 "--failed-after-days",
632 "30",
633 "--batch-size",
634 "500",
635 ]);
636
637 assert!(app.is_ok());
638 }
639
640 #[test]
641 fn test_restore_command_parsing() {
642 use clap::Parser;
643
644 #[derive(Parser)]
645 struct TestApp {
646 #[command(subcommand)]
647 command: ArchiveCommand,
648 }
649
650 let app =
651 TestApp::try_parse_from(&["test", "restore", "550e8400-e29b-41d4-a716-446655440000"]);
652
653 assert!(app.is_ok());
654 }
655}