1use anyhow::Result;
2use clap::Subcommand;
3use serde_json::Value;
4use std::collections::{HashMap, HashSet, VecDeque};
5use tracing::warn;
6use uuid::Uuid;
7
8use crate::config::Config;
9use crate::utils::database::DatabasePool;
10
11#[derive(Debug, Clone)]
12pub struct SpawnNode {
13 pub id: String,
14 pub queue_name: String,
15 pub status: String,
16 pub depends_on: Vec<String>,
17 pub spawn_config: Option<Value>,
18 pub created_at: String,
19 pub workflow_id: Option<String>,
20 pub workflow_name: Option<String>,
21}
22
23#[derive(Debug, Clone)]
24pub struct SpawnOperation {
25 pub parent_job_id: String,
26 pub spawned_jobs: Vec<String>,
27 pub spawned_at: String,
28 pub operation_id: Option<String>,
29 pub config: Option<Value>,
30}
31
32#[derive(Subcommand)]
33pub enum SpawnCommand {
34 #[command(about = "List active spawn operations")]
35 List {
36 #[arg(short = 'u', long, help = "Database connection URL")]
37 database_url: Option<String>,
38 #[arg(short, long, help = "Maximum number of operations to display")]
39 limit: Option<u32>,
40 #[arg(long, help = "Show only recent spawn operations")]
41 recent: bool,
42 #[arg(long, help = "Show spawn operations for specific queue")]
43 queue: Option<String>,
44 },
45 #[command(about = "Show spawn tree hierarchy for a job")]
46 Tree {
47 #[arg(short = 'u', long, help = "Database connection URL")]
48 database_url: Option<String>,
49 #[arg(help = "Job ID (parent or child)")]
50 job_id: String,
51 #[arg(long, help = "Show full spawn tree (both up and down)")]
52 full: bool,
53 #[arg(long, help = "Show only children of this job")]
54 children_only: bool,
55 #[arg(long, help = "Output format (text, json, mermaid)")]
56 format: Option<String>,
57 },
58 #[command(about = "Show spawn statistics")]
59 Stats {
60 #[arg(short = 'u', long, help = "Database connection URL")]
61 database_url: Option<String>,
62 #[arg(short = 'q', long, help = "Filter by queue name")]
63 queue: Option<String>,
64 #[arg(long, help = "Time period in hours (default: 24)")]
65 hours: Option<u32>,
66 #[arg(long, help = "Show detailed breakdown")]
67 detailed: bool,
68 },
69 #[command(about = "Track spawn lineage for a job")]
70 Lineage {
71 #[arg(short = 'u', long, help = "Database connection URL")]
72 database_url: Option<String>,
73 #[arg(help = "Job ID")]
74 job_id: String,
75 #[arg(long, help = "Show ancestor chain")]
76 ancestors: bool,
77 #[arg(long, help = "Show descendant chain")]
78 descendants: bool,
79 #[arg(long, help = "Maximum depth to traverse")]
80 depth: Option<u32>,
81 },
82 #[command(about = "Show jobs waiting for spawn completion")]
83 Pending {
84 #[arg(short = 'u', long, help = "Database connection URL")]
85 database_url: Option<String>,
86 #[arg(short = 'q', long, help = "Filter by queue name")]
87 queue: Option<String>,
88 #[arg(long, help = "Show spawn configuration details")]
89 show_config: bool,
90 },
91 #[command(about = "Monitor spawn operations in real-time")]
92 Monitor {
93 #[arg(short = 'u', long, help = "Database connection URL")]
94 database_url: Option<String>,
95 #[arg(long, help = "Refresh interval in seconds (default: 5)")]
96 interval: Option<u32>,
97 #[arg(short = 'q', long, help = "Filter by queue name")]
98 queue: Option<String>,
99 },
100}
101
102impl SpawnCommand {
103 pub async fn execute(&self, config: Config) -> Result<()> {
104 let db_url = self.get_database_url(&config)?;
105 let pool = DatabasePool::connect(&db_url, config.get_connection_pool_size()).await?;
106
107 match self {
108 SpawnCommand::List {
109 limit,
110 recent,
111 queue,
112 ..
113 } => {
114 self.list_spawn_operations(pool, *limit, *recent, queue.as_deref())
115 .await
116 }
117 SpawnCommand::Tree {
118 job_id,
119 full,
120 children_only,
121 format,
122 ..
123 } => {
124 self.show_spawn_tree(pool, job_id, *full, *children_only, format.as_deref())
125 .await
126 }
127 SpawnCommand::Stats {
128 queue,
129 hours,
130 detailed,
131 ..
132 } => {
133 self.show_spawn_stats(pool, queue.as_deref(), *hours, *detailed)
134 .await
135 }
136 SpawnCommand::Lineage {
137 job_id,
138 ancestors,
139 descendants,
140 depth,
141 ..
142 } => {
143 self.show_spawn_lineage(pool, job_id, *ancestors, *descendants, *depth)
144 .await
145 }
146 SpawnCommand::Pending {
147 queue, show_config, ..
148 } => {
149 self.show_pending_spawns(pool, queue.as_deref(), *show_config)
150 .await
151 }
152 SpawnCommand::Monitor {
153 interval, queue, ..
154 } => {
155 self.monitor_spawn_operations(pool, *interval, queue.as_deref())
156 .await
157 }
158 }
159 }
160
161 pub fn get_database_url(&self, config: &Config) -> Result<String> {
162 let url_option = match self {
163 SpawnCommand::List { database_url, .. } => database_url,
164 SpawnCommand::Tree { database_url, .. } => database_url,
165 SpawnCommand::Stats { database_url, .. } => database_url,
166 SpawnCommand::Lineage { database_url, .. } => database_url,
167 SpawnCommand::Pending { database_url, .. } => database_url,
168 SpawnCommand::Monitor { database_url, .. } => database_url,
169 };
170
171 url_option
172 .as_ref()
173 .map(|s| s.as_str())
174 .or(config.get_database_url())
175 .map(|s| s.to_string())
176 .ok_or_else(|| anyhow::anyhow!("Database URL is required"))
177 }
178
179 async fn list_spawn_operations(
180 &self,
181 pool: DatabasePool,
182 limit: Option<u32>,
183 recent: bool,
184 queue_filter: Option<&str>,
185 ) -> Result<()> {
186 let limit = limit.unwrap_or(20);
187 let time_filter = if recent {
188 "AND created_at > NOW() - INTERVAL '1 hour'"
189 } else {
190 ""
191 };
192
193 let queue_clause = if let Some(queue) = queue_filter {
194 format!("AND queue_name = '{}'", queue)
195 } else {
196 String::new()
197 };
198
199 let query = format!(
200 r#"
201 SELECT parent.id as parent_id, parent.queue_name, parent.created_at,
202 parent.payload->'_spawn_config' as spawn_config,
203 COUNT(child.id) as spawned_count,
204 parent.workflow_id, parent.workflow_name
205 FROM hammerwork_jobs parent
206 LEFT JOIN hammerwork_jobs child ON child.depends_on @> CONCAT('[\"', parent.id, '\"]')::jsonb
207 WHERE parent.payload ? '_spawn_config'
208 AND parent.status IN ('Completed', 'Running')
209 {} {}
210 GROUP BY parent.id, parent.queue_name, parent.created_at, parent.payload, parent.workflow_id, parent.workflow_name
211 ORDER BY parent.created_at DESC
212 LIMIT {}
213 "#,
214 time_filter, queue_clause, limit
215 );
216
217 match pool {
218 DatabasePool::Postgres(pg_pool) => {
219 let rows = sqlx::query(&query).fetch_all(&pg_pool).await?;
220
221 println!("📊 Spawn Operations");
222 println!("{}", "=".repeat(80));
223
224 if rows.is_empty() {
225 println!("No spawn operations found.");
226 return Ok(());
227 }
228
229 println!(
230 "{:<8} {:<15} {:<12} {:<20} {:<10} {:<15}",
231 "Parent", "Queue", "Children", "Spawned At", "Operation", "Workflow"
232 );
233 println!("{}", "-".repeat(80));
234
235 for row in rows {
236 use sqlx::Row;
237 let parent_id: Uuid = row.get("parent_id");
238 let queue_name: String = row.get("queue_name");
239 let spawned_count: i64 = row.get("spawned_count");
240 let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
241 let spawn_config: Option<Value> = row.try_get("spawn_config").ok().flatten();
242
243 let operation_id = spawn_config
244 .as_ref()
245 .and_then(|c| c.get("operation_id"))
246 .and_then(|id| id.as_str())
247 .unwrap_or("none");
248
249 let workflow = row
250 .try_get::<Option<String>, _>("workflow_name")
251 .unwrap_or(None)
252 .unwrap_or_else(|| "none".to_string());
253
254 println!(
255 "{:<8} {:<15} {:<12} {:<20} {:<10} {:<15}",
256 &parent_id.to_string()[..8],
257 &queue_name[..std::cmp::min(15, queue_name.len())],
258 spawned_count,
259 created_at.format("%m-%d %H:%M:%S"),
260 &operation_id[..std::cmp::min(10, operation_id.len())],
261 &workflow[..std::cmp::min(15, workflow.len())]
262 );
263 }
264 }
265 DatabasePool::MySQL(mysql_pool) => {
266 let mysql_query = format!(
268 r#"
269 SELECT parent.id as parent_id, parent.queue_name, parent.created_at,
270 JSON_EXTRACT(parent.payload, '$._spawn_config') as spawn_config,
271 COUNT(child.id) as spawned_count,
272 parent.workflow_id, parent.workflow_name
273 FROM hammerwork_jobs parent
274 LEFT JOIN hammerwork_jobs child ON JSON_CONTAINS(child.depends_on, CONCAT('"', parent.id, '"'))
275 WHERE JSON_EXTRACT(parent.payload, '$._spawn_config') IS NOT NULL
276 AND parent.status IN ('Completed', 'Running')
277 {}
278 {}
279 GROUP BY parent.id, parent.queue_name, parent.created_at, parent.payload, parent.workflow_id, parent.workflow_name
280 ORDER BY parent.created_at DESC
281 LIMIT {}
282 "#,
283 if recent {
284 "AND parent.created_at > DATE_SUB(NOW(), INTERVAL 1 HOUR)"
285 } else {
286 ""
287 },
288 queue_clause,
289 limit
290 );
291
292 let rows = sqlx::query(&mysql_query).fetch_all(&mysql_pool).await?;
293
294 println!("📊 Spawn Operations");
295 println!("{}", "=".repeat(80));
296
297 if rows.is_empty() {
298 println!("No spawn operations found.");
299 return Ok(());
300 }
301
302 println!(
303 "{:<8} {:<15} {:<12} {:<20} {:<10} {:<15}",
304 "Parent", "Queue", "Children", "Spawned At", "Operation", "Workflow"
305 );
306 println!("{}", "-".repeat(80));
307
308 for row in rows {
309 use sqlx::Row;
310 let parent_id: String = row.get("parent_id");
311 let queue_name: String = row.get("queue_name");
312 let spawned_count: i64 = row.get("spawned_count");
313 let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
314 let spawn_config_str: Option<String> =
315 row.try_get("spawn_config").ok().flatten();
316
317 let operation_id = spawn_config_str
318 .as_ref()
319 .and_then(|config_str| serde_json::from_str::<Value>(config_str).ok())
320 .and_then(|config| {
321 config
322 .get("operation_id")
323 .and_then(|id| id.as_str().map(|s| s.to_string()))
324 })
325 .unwrap_or_else(|| "none".to_string());
326
327 let workflow = row
328 .try_get::<Option<String>, _>("workflow_name")
329 .unwrap_or(None)
330 .unwrap_or_else(|| "none".to_string());
331
332 println!(
333 "{:<8} {:<15} {:<12} {:<20} {:<10} {:<15}",
334 &parent_id[..std::cmp::min(8, parent_id.len())],
335 &queue_name[..std::cmp::min(15, queue_name.len())],
336 spawned_count,
337 created_at.format("%m-%d %H:%M:%S"),
338 &operation_id[..std::cmp::min(10, operation_id.len())],
339 &workflow[..std::cmp::min(15, workflow.len())]
340 );
341 }
342 }
343 }
344
345 Ok(())
346 }
347
348 async fn show_spawn_tree(
349 &self,
350 pool: DatabasePool,
351 job_id: &str,
352 show_full: bool,
353 children_only: bool,
354 format: Option<&str>,
355 ) -> Result<()> {
356 let format = format.unwrap_or("text");
357 let job_uuid = Uuid::parse_str(job_id)?;
358
359 let target_job = self.get_spawn_node(&pool, &job_uuid).await?;
361 let target_job = match target_job {
362 Some(job) => job,
363 None => {
364 println!("Job not found: {}", job_id);
365 return Ok(());
366 }
367 };
368
369 let spawn_tree = if show_full || !children_only {
371 self.collect_full_spawn_tree(&pool, &target_job).await?
372 } else {
373 self.collect_spawn_children(&pool, &target_job).await?
374 };
375
376 match format {
377 "text" => self.print_spawn_tree_text(&spawn_tree, &target_job.id),
378 "json" => self.print_spawn_tree_json(&spawn_tree)?,
379 "mermaid" => self.print_spawn_tree_mermaid(&spawn_tree, &target_job.id),
380 _ => {
381 println!("Unsupported format: {}. Use: text, json, mermaid", format);
382 return Ok(());
383 }
384 }
385
386 Ok(())
387 }
388
389 async fn show_spawn_stats(
390 &self,
391 pool: DatabasePool,
392 queue_filter: Option<&str>,
393 hours: Option<u32>,
394 detailed: bool,
395 ) -> Result<()> {
396 let hours = hours.unwrap_or(24);
397
398 let queue_clause = if let Some(queue) = queue_filter {
399 format!("AND queue_name = '{}'", queue)
400 } else {
401 String::new()
402 };
403
404 println!("📈 Spawn Statistics (Last {} hours)", hours);
405 println!("{}", "=".repeat(60));
406
407 match pool {
408 DatabasePool::Postgres(pg_pool) => {
409 let total_query = format!(
411 r#"
412 SELECT COUNT(*) as total_spawn_ops,
413 AVG(spawned_count) as avg_children,
414 MAX(spawned_count) as max_children
415 FROM (
416 SELECT parent.id, COUNT(child.id) as spawned_count
417 FROM hammerwork_jobs parent
418 LEFT JOIN hammerwork_jobs child ON child.depends_on @> CONCAT('[\"', parent.id, '\"]')::jsonb
419 WHERE parent.payload ? '_spawn_config'
420 AND parent.created_at > NOW() - INTERVAL '{} hours'
421 {}
422 GROUP BY parent.id
423 ) spawn_stats
424 "#,
425 hours, queue_clause
426 );
427
428 if let Ok(row) = sqlx::query(&total_query).fetch_one(&pg_pool).await {
429 use sqlx::Row;
430 let total: i64 = row.get("total_spawn_ops");
431 let avg: Option<f64> = row.try_get("avg_children").ok().flatten();
432 let max: Option<i64> = row.try_get("max_children").ok().flatten();
433
434 println!("Total Spawn Operations: {}", total);
435 if let Some(avg) = avg {
436 println!("Average Children per Spawn: {:.1}", avg);
437 }
438 if let Some(max) = max {
439 println!("Maximum Children in Single Spawn: {}", max);
440 }
441 }
442
443 if detailed {
444 println!("\n📋 Breakdown by Queue:");
445 let breakdown_query = format!(
446 r#"
447 SELECT queue_name,
448 COUNT(*) as spawn_count,
449 AVG(spawned_count) as avg_children
450 FROM (
451 SELECT parent.queue_name, COUNT(child.id) as spawned_count
452 FROM hammerwork_jobs parent
453 LEFT JOIN hammerwork_jobs child ON child.depends_on @> CONCAT('[\"', parent.id, '\"]')::jsonb
454 WHERE parent.payload ? '_spawn_config'
455 AND parent.created_at > NOW() - INTERVAL '{} hours'
456 {}
457 GROUP BY parent.id, parent.queue_name
458 ) spawn_breakdown
459 GROUP BY queue_name
460 ORDER BY spawn_count DESC
461 "#,
462 hours, queue_clause
463 );
464
465 let rows = sqlx::query(&breakdown_query).fetch_all(&pg_pool).await?;
466
467 println!(
468 "{:<20} {:<12} {:<15}",
469 "Queue", "Operations", "Avg Children"
470 );
471 println!("{}", "-".repeat(47));
472
473 for row in rows {
474 use sqlx::Row;
475 let queue: String = row.get("queue_name");
476 let count: i64 = row.get("spawn_count");
477 let avg: Option<f64> = row.try_get("avg_children").ok().flatten();
478
479 println!(
480 "{:<20} {:<12} {:<15.1}",
481 &queue[..std::cmp::min(20, queue.len())],
482 count,
483 avg.unwrap_or(0.0)
484 );
485 }
486 }
487 }
488 DatabasePool::MySQL(mysql_pool) => {
489 let total_query = format!(
491 r#"
492 SELECT COUNT(*) as total_spawn_ops,
493 AVG(spawned_count) as avg_children,
494 MAX(spawned_count) as max_children
495 FROM (
496 SELECT parent.id, COUNT(child.id) as spawned_count
497 FROM hammerwork_jobs parent
498 LEFT JOIN hammerwork_jobs child ON JSON_CONTAINS(child.depends_on, CONCAT('"', parent.id, '"'))
499 WHERE JSON_EXTRACT(parent.payload, '$._spawn_config') IS NOT NULL
500 AND parent.created_at > DATE_SUB(NOW(), INTERVAL {} HOUR)
501 {}
502 GROUP BY parent.id
503 ) spawn_stats
504 "#,
505 hours, queue_clause
506 );
507
508 if let Ok(row) = sqlx::query(&total_query).fetch_one(&mysql_pool).await {
509 use sqlx::Row;
510 let total: i64 = row.get("total_spawn_ops");
511 let avg: Option<f64> = row.try_get("avg_children").ok();
512 let max: Option<i64> = row.try_get("max_children").ok();
513
514 println!("Total Spawn Operations: {}", total);
515 if let Some(avg) = avg {
516 println!("Average Children per Spawn: {:.1}", avg);
517 }
518 if let Some(max) = max {
519 println!("Maximum Children in Single Spawn: {}", max);
520 }
521 }
522
523 if detailed {
524 println!("\n📋 Breakdown by Queue:");
525 let breakdown_query = format!(
526 r#"
527 SELECT queue_name,
528 COUNT(*) as spawn_count,
529 AVG(spawned_count) as avg_children
530 FROM (
531 SELECT parent.queue_name, COUNT(child.id) as spawned_count
532 FROM hammerwork_jobs parent
533 LEFT JOIN hammerwork_jobs child ON JSON_CONTAINS(child.depends_on, CONCAT('"', parent.id, '"'))
534 WHERE JSON_EXTRACT(parent.payload, '$._spawn_config') IS NOT NULL
535 AND parent.created_at > DATE_SUB(NOW(), INTERVAL {} HOUR)
536 {}
537 GROUP BY parent.id, parent.queue_name
538 ) spawn_breakdown
539 GROUP BY queue_name
540 ORDER BY spawn_count DESC
541 "#,
542 hours, queue_clause
543 );
544
545 let rows = sqlx::query(&breakdown_query).fetch_all(&mysql_pool).await?;
546
547 println!(
548 "{:<20} {:<12} {:<15}",
549 "Queue", "Operations", "Avg Children"
550 );
551 println!("{}", "-".repeat(47));
552
553 for row in rows {
554 use sqlx::Row;
555 let queue: String = row.get("queue_name");
556 let count: i64 = row.get("spawn_count");
557 let avg: Option<f64> = row.try_get("avg_children").ok();
558
559 println!(
560 "{:<20} {:<12} {:<15.1}",
561 &queue[..std::cmp::min(20, queue.len())],
562 count,
563 avg.unwrap_or(0.0)
564 );
565 }
566 }
567 }
568 }
569
570 Ok(())
571 }
572
573 async fn show_spawn_lineage(
574 &self,
575 pool: DatabasePool,
576 job_id: &str,
577 show_ancestors: bool,
578 show_descendants: bool,
579 max_depth: Option<u32>,
580 ) -> Result<()> {
581 let job_uuid = Uuid::parse_str(job_id)?;
582 let max_depth = max_depth.unwrap_or(10);
583
584 let target_job = self.get_spawn_node(&pool, &job_uuid).await?;
585 let target_job = match target_job {
586 Some(job) => job,
587 None => {
588 println!("Job not found: {}", job_id);
589 return Ok(());
590 }
591 };
592
593 println!("🔗 Spawn Lineage for {}", job_id);
594 println!(
595 "Queue: {} | Status: {}",
596 target_job.queue_name, target_job.status
597 );
598 println!("{}", "=".repeat(60));
599
600 if show_ancestors || (!show_descendants) {
602 println!("\n⬆️ Ancestor Chain:");
603 let ancestors = self
604 .collect_ancestors(&pool, &target_job, max_depth)
605 .await?;
606 if ancestors.is_empty() {
607 println!(" No spawn ancestors found (this is a root job)");
608 } else {
609 for (depth, ancestor) in ancestors.iter().enumerate() {
610 let indent = " ".repeat(depth + 1);
611 println!(
612 "{}└─ [{}] {} ({})",
613 indent,
614 &ancestor.id[..8],
615 ancestor.queue_name,
616 ancestor.status
617 );
618 }
619 }
620 }
621
622 if show_descendants || (!show_ancestors) {
624 println!("\n⬇️ Descendant Chain:");
625 let descendants = self
626 .collect_descendants(&pool, &target_job, max_depth)
627 .await?;
628 if descendants.is_empty() {
629 println!(" No spawn descendants found (this job hasn't spawned children)");
630 } else {
631 self.print_descendant_tree(&descendants, &target_job.id, 0);
632 }
633 }
634
635 Ok(())
636 }
637
638 async fn show_pending_spawns(
639 &self,
640 pool: DatabasePool,
641 queue_filter: Option<&str>,
642 show_config: bool,
643 ) -> Result<()> {
644 let queue_clause = if let Some(queue) = queue_filter {
645 format!("AND queue_name = '{}'", queue)
646 } else {
647 String::new()
648 };
649
650 println!("⏳ Jobs with Pending Spawn Operations");
651 println!("{}", "=".repeat(70));
652
653 match pool {
654 DatabasePool::Postgres(pg_pool) => {
655 let query = format!(
656 r#"
657 SELECT id, queue_name, status, created_at,
658 payload->'_spawn_config' as spawn_config
659 FROM hammerwork_jobs
660 WHERE payload ? '_spawn_config'
661 AND status IN ('Running', 'Pending')
662 {}
663 ORDER BY created_at DESC
664 LIMIT 50
665 "#,
666 queue_clause
667 );
668
669 let rows = sqlx::query(&query).fetch_all(&pg_pool).await?;
670
671 if rows.is_empty() {
672 println!("No jobs with pending spawn operations found.");
673 return Ok(());
674 }
675
676 for row in rows {
677 use sqlx::Row;
678 let id: Uuid = row.get("id");
679 let queue_name: String = row.get("queue_name");
680 let status: String = row.get("status");
681 let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
682 let spawn_config: Option<Value> = row.try_get("spawn_config").ok().flatten();
683
684 println!(
685 "📋 Job: {} | Queue: {} | Status: {} | Created: {}",
686 &id.to_string()[..8],
687 queue_name,
688 status,
689 created_at.format("%m-%d %H:%M:%S")
690 );
691
692 if show_config {
693 if let Some(config) = spawn_config {
694 println!(
695 " Spawn Config: {}",
696 serde_json::to_string_pretty(&config)
697 .unwrap_or_else(|_| "Invalid JSON".to_string())
698 );
699 }
700 println!();
701 }
702 }
703 }
704 DatabasePool::MySQL(mysql_pool) => {
705 let query = format!(
706 r#"
707 SELECT id, queue_name, status, created_at,
708 JSON_EXTRACT(payload, '$._spawn_config') as spawn_config
709 FROM hammerwork_jobs
710 WHERE JSON_EXTRACT(payload, '$._spawn_config') IS NOT NULL
711 AND status IN ('Running', 'Pending')
712 {}
713 ORDER BY created_at DESC
714 LIMIT 50
715 "#,
716 queue_clause
717 );
718
719 let rows = sqlx::query(&query).fetch_all(&mysql_pool).await?;
720
721 if rows.is_empty() {
722 println!("No jobs with pending spawn operations found.");
723 return Ok(());
724 }
725
726 for row in rows {
727 use sqlx::Row;
728 let id: String = row.get("id");
729 let queue_name: String = row.get("queue_name");
730 let status: String = row.get("status");
731 let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
732 let spawn_config_str: Option<String> =
733 row.try_get("spawn_config").ok().flatten();
734
735 println!(
736 "📋 Job: {} | Queue: {} | Status: {} | Created: {}",
737 &id[..std::cmp::min(8, id.len())],
738 queue_name,
739 status,
740 created_at.format("%m-%d %H:%M:%S")
741 );
742
743 if show_config {
744 if let Some(ref config_str) = spawn_config_str {
745 match serde_json::from_str::<Value>(config_str) {
746 Ok(config) => {
747 println!(
748 " Spawn Config: {}",
749 serde_json::to_string_pretty(&config)
750 .unwrap_or_else(|_| "Invalid JSON".to_string())
751 );
752 }
753 Err(_) => {
754 println!(" Spawn Config: {}", config_str);
755 }
756 }
757 }
758 println!();
759 }
760 }
761 }
762 }
763
764 Ok(())
765 }
766
767 async fn monitor_spawn_operations(
768 &self,
769 pool: DatabasePool,
770 interval: Option<u32>,
771 queue_filter: Option<&str>,
772 ) -> Result<()> {
773 let interval = std::time::Duration::from_secs(interval.unwrap_or(5) as u64);
774
775 println!("🔄 Monitoring Spawn Operations (Press Ctrl+C to stop)");
776 println!("Refresh interval: {:?}", interval);
777 if let Some(queue) = queue_filter {
778 println!("Queue filter: {}", queue);
779 }
780 println!("{}", "=".repeat(80));
781
782 match &pool {
785 DatabasePool::Postgres(_pg_pool) => {
786 warn!("Monitor mode requires reconnection for each refresh");
789 Err(anyhow::anyhow!(
790 "Monitor mode needs database URL to reconnect. Use list/stats commands instead for one-time queries."
791 ))
792 }
793 DatabasePool::MySQL(_) => {
794 warn!("Monitor mode requires reconnection for each refresh");
795 Err(anyhow::anyhow!(
796 "Monitor mode needs database URL to reconnect. Use list/stats commands instead for one-time queries."
797 ))
798 }
799 }
800
801 }
804
805 async fn get_spawn_node(
808 &self,
809 pool: &DatabasePool,
810 job_id: &Uuid,
811 ) -> Result<Option<SpawnNode>> {
812 let query = r#"
813 SELECT id, queue_name, status, depends_on,
814 payload->'_spawn_config' as spawn_config,
815 created_at, workflow_id, workflow_name
816 FROM hammerwork_jobs
817 WHERE id = $1
818 "#;
819
820 match pool {
821 DatabasePool::Postgres(pg_pool) => {
822 if let Some(row) = sqlx::query(query)
823 .bind(job_id)
824 .fetch_optional(pg_pool)
825 .await?
826 {
827 Ok(Some(self.postgres_row_to_spawn_node(&row)?))
828 } else {
829 Ok(None)
830 }
831 }
832 DatabasePool::MySQL(mysql_pool) => {
833 let mysql_query = query.replace("$1", "?");
834 if let Some(row) = sqlx::query(&mysql_query)
835 .bind(job_id.to_string())
836 .fetch_optional(mysql_pool)
837 .await?
838 {
839 Ok(Some(self.mysql_row_to_spawn_node(&row)?))
840 } else {
841 Ok(None)
842 }
843 }
844 }
845 }
846
847 async fn collect_spawn_children(
848 &self,
849 pool: &DatabasePool,
850 parent: &SpawnNode,
851 ) -> Result<Vec<SpawnNode>> {
852 let query = r#"
853 SELECT id, queue_name, status, depends_on,
854 payload->'_spawn_config' as spawn_config,
855 created_at, workflow_id, workflow_name
856 FROM hammerwork_jobs
857 WHERE depends_on @> $1
858 ORDER BY created_at
859 "#;
860
861 let mut children = Vec::new();
862
863 match pool {
864 DatabasePool::Postgres(pg_pool) => {
865 let parent_json = format!("[\"{}\"]", parent.id);
866 let rows = sqlx::query(query)
867 .bind(&parent_json)
868 .fetch_all(pg_pool)
869 .await?;
870
871 for row in rows {
872 children.push(self.postgres_row_to_spawn_node(&row)?);
873 }
874 }
875 DatabasePool::MySQL(mysql_pool) => {
876 let mysql_query = r#"
877 SELECT id, queue_name, status, depends_on,
878 JSON_EXTRACT(payload, '$._spawn_config') as spawn_config,
879 created_at, workflow_id, workflow_name
880 FROM hammerwork_jobs
881 WHERE JSON_CONTAINS(depends_on, ?)
882 ORDER BY created_at
883 "#;
884
885 let parent_json = format!("\"{}\"", parent.id);
886 let rows = sqlx::query(mysql_query)
887 .bind(&parent_json)
888 .fetch_all(mysql_pool)
889 .await?;
890
891 for row in rows {
892 children.push(self.mysql_row_to_spawn_node(&row)?);
893 }
894 }
895 }
896
897 Ok(children)
898 }
899
900 async fn collect_full_spawn_tree(
901 &self,
902 pool: &DatabasePool,
903 target: &SpawnNode,
904 ) -> Result<Vec<SpawnNode>> {
905 let mut all_nodes = HashMap::new();
906 let mut to_visit = VecDeque::new();
907 let mut visited = HashSet::new();
908
909 all_nodes.insert(target.id.clone(), target.clone());
911 to_visit.push_back(target.id.clone());
912
913 while let Some(job_id) = to_visit.pop_front() {
915 if visited.contains(&job_id) {
916 continue;
917 }
918 visited.insert(job_id.clone());
919
920 if let Some(job) = all_nodes.get(&job_id).cloned() {
921 let children = self.collect_spawn_children(pool, &job).await?;
923 for child in children {
924 if !all_nodes.contains_key(&child.id) {
925 all_nodes.insert(child.id.clone(), child.clone());
926 to_visit.push_back(child.id.clone());
927 }
928 }
929
930 for parent_id in &job.depends_on {
932 if let Ok(parent_uuid) = Uuid::parse_str(parent_id) {
933 if let Ok(Some(parent)) = self.get_spawn_node(pool, &parent_uuid).await {
934 if !all_nodes.contains_key(&parent.id) {
935 all_nodes.insert(parent.id.clone(), parent.clone());
936 to_visit.push_back(parent.id.clone());
937 }
938 }
939 }
940 }
941 }
942 }
943
944 Ok(all_nodes.into_values().collect())
945 }
946
947 async fn collect_ancestors(
948 &self,
949 pool: &DatabasePool,
950 job: &SpawnNode,
951 max_depth: u32,
952 ) -> Result<Vec<SpawnNode>> {
953 let mut ancestors = Vec::new();
954 let mut current = job.clone();
955 let mut depth = 0;
956
957 while depth < max_depth && !current.depends_on.is_empty() {
958 let mut parent_found = false;
960 for parent_id in ¤t.depends_on {
961 if let Ok(parent_uuid) = Uuid::parse_str(parent_id) {
962 if let Ok(Some(parent)) = self.get_spawn_node(pool, &parent_uuid).await {
963 if parent.spawn_config.is_some() {
964 ancestors.push(parent.clone());
965 current = parent;
966 parent_found = true;
967 break;
968 }
969 }
970 }
971 }
972
973 if !parent_found {
974 break;
975 }
976
977 depth += 1;
978 }
979
980 ancestors.reverse(); Ok(ancestors)
982 }
983
984 async fn collect_descendants(
985 &self,
986 pool: &DatabasePool,
987 job: &SpawnNode,
988 max_depth: u32,
989 ) -> Result<Vec<SpawnNode>> {
990 let mut descendants = Vec::new();
991 let mut to_visit = VecDeque::new();
992 let mut visited = HashSet::new();
993
994 to_visit.push_back((job.clone(), 0));
995
996 while let Some((current_job, depth)) = to_visit.pop_front() {
997 if depth >= max_depth || visited.contains(¤t_job.id) {
998 continue;
999 }
1000 visited.insert(current_job.id.clone());
1001
1002 let children = self.collect_spawn_children(pool, ¤t_job).await?;
1003 for child in children {
1004 descendants.push(child.clone());
1005 to_visit.push_back((child, depth + 1));
1006 }
1007 }
1008
1009 Ok(descendants)
1010 }
1011
1012 fn postgres_row_to_spawn_node(&self, row: &sqlx::postgres::PgRow) -> Result<SpawnNode> {
1013 use sqlx::Row;
1014
1015 let id: Uuid = row.get("id");
1016 let depends_on = self.parse_json_array(row.try_get("depends_on").ok())?;
1017 let spawn_config: Option<Value> = row.try_get("spawn_config").ok().flatten();
1018 let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
1019
1020 let workflow_id: Option<String> = match row.try_get::<Option<Uuid>, _>("workflow_id") {
1021 Ok(Some(uuid)) => Some(uuid.to_string()),
1022 Ok(None) => None,
1023 Err(_) => None,
1024 };
1025
1026 Ok(SpawnNode {
1027 id: id.to_string(),
1028 queue_name: row.get("queue_name"),
1029 status: row.get("status"),
1030 depends_on,
1031 spawn_config,
1032 created_at: created_at.to_string(),
1033 workflow_id,
1034 workflow_name: row.try_get("workflow_name").ok(),
1035 })
1036 }
1037
1038 fn mysql_row_to_spawn_node(&self, row: &sqlx::mysql::MySqlRow) -> Result<SpawnNode> {
1039 use sqlx::Row;
1040
1041 let id: String = row.get("id");
1042 let depends_on = self.parse_json_array(row.try_get("depends_on").ok())?;
1043
1044 let spawn_config: Option<Value> = match row.try_get::<Option<String>, _>("spawn_config") {
1046 Ok(Some(config_str)) => serde_json::from_str(&config_str).ok(),
1047 Ok(None) => None,
1048 Err(_) => {
1049 row.try_get("spawn_config").ok().flatten()
1051 }
1052 };
1053
1054 let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
1055
1056 Ok(SpawnNode {
1057 id,
1058 queue_name: row.get("queue_name"),
1059 status: row.get("status"),
1060 depends_on,
1061 spawn_config,
1062 created_at: created_at.to_string(),
1063 workflow_id: row.try_get("workflow_id").ok(),
1064 workflow_name: row.try_get("workflow_name").ok(),
1065 })
1066 }
1067
1068 fn parse_json_array(&self, json_value: Option<Value>) -> Result<Vec<String>> {
1069 match json_value {
1070 Some(Value::Array(arr)) => Ok(arr
1071 .into_iter()
1072 .filter_map(|v| v.as_str().map(|s| s.to_string()))
1073 .collect()),
1074 _ => Ok(Vec::new()),
1075 }
1076 }
1077
1078 fn print_spawn_tree_text(&self, nodes: &[SpawnNode], target_id: &str) {
1079 println!("\n🌳 Spawn Tree");
1080 println!("{}", "=".repeat(60));
1081
1082 let node_map: HashMap<String, &SpawnNode> =
1083 nodes.iter().map(|n| (n.id.clone(), n)).collect();
1084
1085 let mut roots: Vec<&SpawnNode> = nodes
1087 .iter()
1088 .filter(|node| {
1089 !node.depends_on.iter().any(|dep_id| {
1090 node_map
1091 .get(dep_id)
1092 .is_some_and(|dep| dep.spawn_config.is_some())
1093 })
1094 })
1095 .collect();
1096
1097 if roots.is_empty() {
1098 roots = nodes.iter().collect();
1099 }
1100
1101 roots.sort_by(|a, b| a.created_at.cmp(&b.created_at));
1102
1103 let mut visited = HashSet::new();
1104 for root in roots {
1105 if !visited.contains(&root.id) {
1106 self.print_spawn_node_tree(root, &node_map, &mut visited, 0, target_id);
1107 }
1108 }
1109 }
1110
1111 #[allow(clippy::only_used_in_recursion)]
1112 fn print_spawn_node_tree(
1113 &self,
1114 node: &SpawnNode,
1115 node_map: &HashMap<String, &SpawnNode>,
1116 visited: &mut HashSet<String>,
1117 depth: usize,
1118 target_id: &str,
1119 ) {
1120 if visited.contains(&node.id) {
1121 return;
1122 }
1123 visited.insert(node.id.clone());
1124
1125 let indent = " ".repeat(depth);
1126 let marker = if depth == 0 { "┌─" } else { "├─" };
1127 let highlight = if node.id == target_id { " ⭐" } else { "" };
1128 let spawn_indicator = if node.spawn_config.is_some() {
1129 "🚀"
1130 } else {
1131 "📝"
1132 };
1133
1134 println!(
1135 "{}{}{} [{}] {} ({}){}",
1136 indent,
1137 marker,
1138 spawn_indicator,
1139 &node.id[..8],
1140 node.queue_name,
1141 node.status,
1142 highlight
1143 );
1144
1145 let children: Vec<&SpawnNode> = node_map
1147 .values()
1148 .filter(|child| child.depends_on.contains(&node.id) && child.id != node.id)
1149 .copied()
1150 .collect();
1151
1152 for child in children {
1153 self.print_spawn_node_tree(child, node_map, visited, depth + 1, target_id);
1154 }
1155 }
1156
1157 fn print_spawn_tree_json(&self, nodes: &[SpawnNode]) -> Result<()> {
1158 println!("\n📄 Spawn Tree (JSON)");
1159 println!("{}", "=".repeat(60));
1160
1161 let tree_data = serde_json::json!({
1162 "spawn_tree": {
1163 "nodes": nodes.iter().map(|node| {
1164 serde_json::json!({
1165 "id": node.id,
1166 "queue": node.queue_name,
1167 "status": node.status,
1168 "depends_on": node.depends_on,
1169 "has_spawn_config": node.spawn_config.is_some(),
1170 "spawn_config": node.spawn_config,
1171 "created_at": node.created_at,
1172 "workflow_id": node.workflow_id,
1173 "workflow_name": node.workflow_name
1174 })
1175 }).collect::<Vec<_>>(),
1176 "edges": self.build_spawn_edges(nodes)
1177 }
1178 });
1179
1180 println!("{}", serde_json::to_string_pretty(&tree_data)?);
1181 Ok(())
1182 }
1183
1184 fn print_spawn_tree_mermaid(&self, nodes: &[SpawnNode], target_id: &str) {
1185 println!("\n🧜♀️ Spawn Tree (Mermaid)");
1186 println!("{}", "=".repeat(60));
1187 println!("graph TD");
1188 println!(" subgraph \"🚀 Spawn Tree\"");
1189
1190 for node in nodes {
1192 let short_id = &node.id[..8];
1193 let status_class = match node.status.as_str() {
1194 "Completed" => ":::completed",
1195 "Failed" => ":::failed",
1196 "Running" => ":::running",
1197 "Pending" => ":::pending",
1198 _ => ":::default",
1199 };
1200
1201 let spawn_indicator = if node.spawn_config.is_some() {
1202 "🚀"
1203 } else {
1204 "📝"
1205 };
1206 let target_indicator = if node.id == target_id { "⭐" } else { "" };
1207
1208 println!(
1209 " {}[\"{}{}<br/>{}<br/>{}\"]{}",
1210 short_id, spawn_indicator, target_indicator, short_id, node.status, status_class
1211 );
1212 }
1213
1214 let node_map: HashMap<String, &SpawnNode> =
1216 nodes.iter().map(|n| (n.id.clone(), n)).collect();
1217
1218 for node in nodes {
1219 for dep_id in &node.depends_on {
1220 if let Some(parent) = node_map.get(dep_id) {
1221 if parent.spawn_config.is_some() {
1222 println!(" {} -->|spawns| {}", &dep_id[..8], &node.id[..8]);
1223 }
1224 }
1225 }
1226 }
1227
1228 println!(" end");
1229 println!();
1230
1231 println!(" classDef completed fill:#d4edda,stroke:#155724");
1233 println!(" classDef failed fill:#f8d7da,stroke:#721c24");
1234 println!(" classDef running fill:#cce7ff,stroke:#004085");
1235 println!(" classDef pending fill:#fff3cd,stroke:#856404");
1236 println!(" classDef default fill:#e2e3e5,stroke:#383d41");
1237 }
1238
1239 fn build_spawn_edges(&self, nodes: &[SpawnNode]) -> Vec<Value> {
1240 let node_map: HashMap<String, &SpawnNode> =
1241 nodes.iter().map(|n| (n.id.clone(), n)).collect();
1242
1243 let mut edges = Vec::new();
1244
1245 for node in nodes {
1246 for dep_id in &node.depends_on {
1247 if let Some(parent) = node_map.get(dep_id) {
1248 if parent.spawn_config.is_some() {
1249 edges.push(serde_json::json!({
1250 "from": dep_id,
1251 "to": node.id,
1252 "type": "spawn",
1253 "relationship": "parent_spawned_child"
1254 }));
1255 }
1256 }
1257 }
1258 }
1259
1260 edges
1261 }
1262
1263 fn print_descendant_tree(&self, descendants: &[SpawnNode], _target_id: &str, depth: usize) {
1264 if descendants.is_empty() {
1265 return;
1266 }
1267
1268 for (i, descendant) in descendants.iter().enumerate() {
1269 let indent = " ".repeat(depth + 1);
1270 let marker = if i == descendants.len() - 1 {
1271 "└─"
1272 } else {
1273 "├─"
1274 };
1275
1276 println!(
1277 "{}{}📝 [{}] {} ({})",
1278 indent,
1279 marker,
1280 &descendant.id[..8],
1281 descendant.queue_name,
1282 descendant.status
1283 );
1284 }
1285 }
1286}
1287
1288#[cfg(test)]
1289mod tests {
1290 use super::*;
1291 use clap::Parser;
1292
1293 #[test]
1294 fn test_spawn_command_variants() {
1295 let commands = vec![
1297 SpawnCommand::List {
1298 database_url: None,
1299 limit: Some(10),
1300 recent: false,
1301 queue: None,
1302 },
1303 SpawnCommand::Tree {
1304 database_url: None,
1305 job_id: "test-job".to_string(),
1306 full: false,
1307 children_only: false,
1308 format: None,
1309 },
1310 SpawnCommand::Stats {
1311 database_url: None,
1312 queue: None,
1313 hours: Some(24),
1314 detailed: false,
1315 },
1316 SpawnCommand::Lineage {
1317 database_url: None,
1318 job_id: "test-job".to_string(),
1319 ancestors: false,
1320 descendants: false,
1321 depth: None,
1322 },
1323 SpawnCommand::Pending {
1324 database_url: None,
1325 queue: None,
1326 show_config: false,
1327 },
1328 SpawnCommand::Monitor {
1329 database_url: None,
1330 interval: Some(5),
1331 queue: None,
1332 },
1333 ];
1334
1335 assert_eq!(commands.len(), 6);
1337 }
1338
1339 #[test]
1340 fn test_spawn_node_creation() {
1341 let spawn_node = SpawnNode {
1342 id: "test-spawn-123".to_string(),
1343 queue_name: "spawning-queue".to_string(),
1344 status: "Completed".to_string(),
1345 depends_on: vec!["parent-job".to_string()],
1346 spawn_config: Some(serde_json::json!({"max_spawn_count": 5})),
1347 created_at: "2024-01-01T00:00:00Z".to_string(),
1348 workflow_id: Some("workflow-123".to_string()),
1349 workflow_name: Some("test-workflow".to_string()),
1350 };
1351
1352 assert_eq!(spawn_node.id, "test-spawn-123");
1353 assert!(spawn_node.spawn_config.is_some());
1354 assert_eq!(spawn_node.depends_on.len(), 1);
1355 }
1356
1357 #[test]
1358 fn test_spawn_operation_creation() {
1359 let spawn_op = SpawnOperation {
1360 parent_job_id: "parent-123".to_string(),
1361 spawned_jobs: vec!["child-1".to_string(), "child-2".to_string()],
1362 spawned_at: "2024-01-01T00:00:00Z".to_string(),
1363 operation_id: Some("op-123".to_string()),
1364 config: Some(serde_json::json!({"batch_size": 10})),
1365 };
1366
1367 assert_eq!(spawn_op.parent_job_id, "parent-123");
1368 assert_eq!(spawn_op.spawned_jobs.len(), 2);
1369 assert!(spawn_op.operation_id.is_some());
1370 assert!(spawn_op.config.is_some());
1371 }
1372
1373 #[test]
1374 fn test_json_array_parsing() {
1375 let spawn_cmd = SpawnCommand::List {
1376 database_url: None,
1377 limit: None,
1378 recent: false,
1379 queue: None,
1380 };
1381
1382 let json_array = serde_json::json!(["job1", "job2", "job3"]);
1384 let result = spawn_cmd.parse_json_array(Some(json_array)).unwrap();
1385 assert_eq!(result, vec!["job1", "job2", "job3"]);
1386
1387 let result = spawn_cmd.parse_json_array(None).unwrap();
1389 assert!(result.is_empty());
1390
1391 let json_object = serde_json::json!({"key": "value"});
1393 let result = spawn_cmd.parse_json_array(Some(json_object)).unwrap();
1394 assert!(result.is_empty());
1395
1396 let mixed_array = serde_json::json!(["string1", 123, true, "string2", null]);
1398 let result = spawn_cmd.parse_json_array(Some(mixed_array)).unwrap();
1399 assert_eq!(result, vec!["string1", "string2"]);
1400 }
1401
1402 #[test]
1403 fn test_get_database_url() {
1404 let config = Config::default();
1405
1406 let cmd = SpawnCommand::List {
1408 database_url: Some("postgres://test".to_string()),
1409 limit: None,
1410 recent: false,
1411 queue: None,
1412 };
1413 let result = cmd.get_database_url(&config);
1414 assert!(result.is_ok());
1415 assert_eq!(result.unwrap(), "postgres://test");
1416
1417 let cmd = SpawnCommand::List {
1419 database_url: None,
1420 limit: None,
1421 recent: false,
1422 queue: None,
1423 };
1424 let result = cmd.get_database_url(&config);
1425 assert!(result.is_err());
1426 }
1427
1428 #[test]
1429 fn test_spawn_command_parsing() {
1430 #[derive(Parser)]
1431 struct TestApp {
1432 #[command(subcommand)]
1433 command: SpawnCommand,
1434 }
1435
1436 let app = TestApp::try_parse_from(&[
1438 "test",
1439 "list",
1440 "--limit",
1441 "50",
1442 "--recent",
1443 "--queue",
1444 "test-queue",
1445 ]);
1446 assert!(app.is_ok());
1447 match app.unwrap().command {
1448 SpawnCommand::List {
1449 limit,
1450 recent,
1451 queue,
1452 ..
1453 } => {
1454 assert_eq!(limit, Some(50));
1455 assert!(recent);
1456 assert_eq!(queue, Some("test-queue".to_string()));
1457 }
1458 _ => panic!("Wrong command variant"),
1459 }
1460
1461 let app = TestApp::try_parse_from(&[
1463 "test",
1464 "tree",
1465 "550e8400-e29b-41d4-a716-446655440000",
1466 "--full",
1467 "--format",
1468 "json",
1469 ]);
1470 assert!(app.is_ok());
1471 match app.unwrap().command {
1472 SpawnCommand::Tree {
1473 job_id,
1474 full,
1475 format,
1476 ..
1477 } => {
1478 assert_eq!(job_id, "550e8400-e29b-41d4-a716-446655440000");
1479 assert!(full);
1480 assert_eq!(format, Some("json".to_string()));
1481 }
1482 _ => panic!("Wrong command variant"),
1483 }
1484
1485 let app = TestApp::try_parse_from(&[
1487 "test",
1488 "stats",
1489 "--hours",
1490 "48",
1491 "--detailed",
1492 "--queue",
1493 "stats-queue",
1494 ]);
1495 assert!(app.is_ok());
1496 match app.unwrap().command {
1497 SpawnCommand::Stats {
1498 hours,
1499 detailed,
1500 queue,
1501 ..
1502 } => {
1503 assert_eq!(hours, Some(48));
1504 assert!(detailed);
1505 assert_eq!(queue, Some("stats-queue".to_string()));
1506 }
1507 _ => panic!("Wrong command variant"),
1508 }
1509
1510 let app = TestApp::try_parse_from(&[
1512 "test",
1513 "lineage",
1514 "test-job-id",
1515 "--ancestors",
1516 "--descendants",
1517 "--depth",
1518 "5",
1519 ]);
1520 assert!(app.is_ok());
1521 match app.unwrap().command {
1522 SpawnCommand::Lineage {
1523 job_id,
1524 ancestors,
1525 descendants,
1526 depth,
1527 ..
1528 } => {
1529 assert_eq!(job_id, "test-job-id");
1530 assert!(ancestors);
1531 assert!(descendants);
1532 assert_eq!(depth, Some(5));
1533 }
1534 _ => panic!("Wrong command variant"),
1535 }
1536
1537 let app = TestApp::try_parse_from(&[
1539 "test",
1540 "pending",
1541 "--queue",
1542 "pending-queue",
1543 "--show-config",
1544 ]);
1545 assert!(app.is_ok());
1546 match app.unwrap().command {
1547 SpawnCommand::Pending {
1548 queue, show_config, ..
1549 } => {
1550 assert_eq!(queue, Some("pending-queue".to_string()));
1551 assert!(show_config);
1552 }
1553 _ => panic!("Wrong command variant"),
1554 }
1555
1556 let app = TestApp::try_parse_from(&[
1558 "test",
1559 "monitor",
1560 "--interval",
1561 "10",
1562 "--queue",
1563 "monitor-queue",
1564 ]);
1565 assert!(app.is_ok());
1566 match app.unwrap().command {
1567 SpawnCommand::Monitor {
1568 interval, queue, ..
1569 } => {
1570 assert_eq!(interval, Some(10));
1571 assert_eq!(queue, Some("monitor-queue".to_string()));
1572 }
1573 _ => panic!("Wrong command variant"),
1574 }
1575 }
1576
1577 #[test]
1578 fn test_build_spawn_edges() {
1579 let cmd = SpawnCommand::List {
1580 database_url: None,
1581 limit: None,
1582 recent: false,
1583 queue: None,
1584 };
1585
1586 let nodes = vec![
1587 SpawnNode {
1588 id: "parent-1".to_string(),
1589 queue_name: "queue1".to_string(),
1590 status: "Completed".to_string(),
1591 depends_on: vec![],
1592 spawn_config: Some(serde_json::json!({"spawn": true})),
1593 created_at: "2024-01-01T00:00:00Z".to_string(),
1594 workflow_id: None,
1595 workflow_name: None,
1596 },
1597 SpawnNode {
1598 id: "child-1".to_string(),
1599 queue_name: "queue2".to_string(),
1600 status: "Running".to_string(),
1601 depends_on: vec!["parent-1".to_string()],
1602 spawn_config: None,
1603 created_at: "2024-01-01T00:01:00Z".to_string(),
1604 workflow_id: None,
1605 workflow_name: None,
1606 },
1607 SpawnNode {
1608 id: "child-2".to_string(),
1609 queue_name: "queue2".to_string(),
1610 status: "Pending".to_string(),
1611 depends_on: vec!["parent-1".to_string()],
1612 spawn_config: None,
1613 created_at: "2024-01-01T00:02:00Z".to_string(),
1614 workflow_id: None,
1615 workflow_name: None,
1616 },
1617 ];
1618
1619 let edges = cmd.build_spawn_edges(&nodes);
1620 assert_eq!(edges.len(), 2);
1621
1622 assert_eq!(edges[0]["from"], "parent-1");
1624 assert_eq!(edges[0]["to"], "child-1");
1625 assert_eq!(edges[0]["type"], "spawn");
1626
1627 assert_eq!(edges[1]["from"], "parent-1");
1629 assert_eq!(edges[1]["to"], "child-2");
1630 assert_eq!(edges[1]["type"], "spawn");
1631 }
1632
1633 #[test]
1634 fn test_spawn_node_with_no_dependencies() {
1635 let node = SpawnNode {
1636 id: "standalone-job".to_string(),
1637 queue_name: "solo-queue".to_string(),
1638 status: "Completed".to_string(),
1639 depends_on: vec![],
1640 spawn_config: None,
1641 created_at: "2024-01-01T00:00:00Z".to_string(),
1642 workflow_id: None,
1643 workflow_name: None,
1644 };
1645
1646 assert!(node.depends_on.is_empty());
1647 assert!(node.spawn_config.is_none());
1648 assert!(node.workflow_id.is_none());
1649 }
1650
1651 #[test]
1652 fn test_spawn_node_with_workflow() {
1653 let node = SpawnNode {
1654 id: "workflow-job".to_string(),
1655 queue_name: "workflow-queue".to_string(),
1656 status: "Running".to_string(),
1657 depends_on: vec!["parent-1".to_string(), "parent-2".to_string()],
1658 spawn_config: Some(serde_json::json!({
1659 "operation_id": "op-123",
1660 "batch_size": 100
1661 })),
1662 created_at: "2024-01-01T00:00:00Z".to_string(),
1663 workflow_id: Some("workflow-123".to_string()),
1664 workflow_name: Some("data-processing-workflow".to_string()),
1665 };
1666
1667 assert_eq!(node.depends_on.len(), 2);
1668 assert!(node.spawn_config.is_some());
1669 assert_eq!(node.workflow_id, Some("workflow-123".to_string()));
1670 assert_eq!(
1671 node.workflow_name,
1672 Some("data-processing-workflow".to_string())
1673 );
1674 }
1675
1676 #[test]
1677 fn test_spawn_operation_without_config() {
1678 let op = SpawnOperation {
1679 parent_job_id: "parent-456".to_string(),
1680 spawned_jobs: vec![],
1681 spawned_at: "2024-01-01T12:00:00Z".to_string(),
1682 operation_id: None,
1683 config: None,
1684 };
1685
1686 assert!(op.spawned_jobs.is_empty());
1687 assert!(op.operation_id.is_none());
1688 assert!(op.config.is_none());
1689 }
1690
1691 #[test]
1692 fn test_format_options() {
1693 let formats = vec!["text", "json", "mermaid"];
1695 for format in formats {
1696 let cmd = SpawnCommand::Tree {
1697 database_url: None,
1698 job_id: "test".to_string(),
1699 full: false,
1700 children_only: false,
1701 format: Some(format.to_string()),
1702 };
1703
1704 match cmd {
1705 SpawnCommand::Tree {
1706 format: Some(f), ..
1707 } => {
1708 assert!(["text", "json", "mermaid"].contains(&f.as_str()));
1709 }
1710 _ => panic!("Wrong command variant"),
1711 }
1712 }
1713 }
1714
1715 #[test]
1716 fn test_spawn_config_parsing() {
1717 let configs = vec![
1719 serde_json::json!({"operation_id": "op-1"}),
1720 serde_json::json!({"max_spawn_count": 100, "batch_size": 10}),
1721 serde_json::json!({"spawn_after_seconds": 60}),
1722 serde_json::json!({}), ];
1724
1725 for config in configs {
1726 let node = SpawnNode {
1727 id: "test".to_string(),
1728 queue_name: "test".to_string(),
1729 status: "Pending".to_string(),
1730 depends_on: vec![],
1731 spawn_config: Some(config.clone()),
1732 created_at: "2024-01-01T00:00:00Z".to_string(),
1733 workflow_id: None,
1734 workflow_name: None,
1735 };
1736
1737 assert!(node.spawn_config.is_some());
1738 assert_eq!(node.spawn_config.unwrap(), config);
1739 }
1740 }
1741
1742 #[test]
1743 fn test_job_status_variants() {
1744 let statuses = vec![
1745 "Pending",
1746 "Running",
1747 "Completed",
1748 "Failed",
1749 "Retrying",
1750 "Dead",
1751 "TimedOut",
1752 ];
1753
1754 for status in statuses {
1755 let node = SpawnNode {
1756 id: "test".to_string(),
1757 queue_name: "test".to_string(),
1758 status: status.to_string(),
1759 depends_on: vec![],
1760 spawn_config: None,
1761 created_at: "2024-01-01T00:00:00Z".to_string(),
1762 workflow_id: None,
1763 workflow_name: None,
1764 };
1765
1766 assert!(!node.status.is_empty());
1767 }
1768 }
1769
1770 #[test]
1771 fn test_command_defaults() {
1772 let list_cmd = SpawnCommand::List {
1774 database_url: None,
1775 limit: None,
1776 recent: false,
1777 queue: None,
1778 };
1779
1780 match list_cmd {
1781 SpawnCommand::List {
1782 limit,
1783 recent,
1784 queue,
1785 ..
1786 } => {
1787 assert!(limit.is_none());
1788 assert!(!recent);
1789 assert!(queue.is_none());
1790 }
1791 _ => panic!("Wrong variant"),
1792 }
1793
1794 let monitor_cmd = SpawnCommand::Monitor {
1795 database_url: None,
1796 interval: None,
1797 queue: None,
1798 };
1799
1800 match monitor_cmd {
1801 SpawnCommand::Monitor { interval, .. } => {
1802 assert!(interval.is_none());
1803 }
1804 _ => panic!("Wrong variant"),
1805 }
1806 }
1807
1808 #[test]
1809 fn test_complex_spawn_tree() {
1810 let nodes = vec![
1812 SpawnNode {
1813 id: "root".to_string(),
1814 queue_name: "root-queue".to_string(),
1815 status: "Completed".to_string(),
1816 depends_on: vec![],
1817 spawn_config: Some(serde_json::json!({"spawn": true})),
1818 created_at: "2024-01-01T00:00:00Z".to_string(),
1819 workflow_id: Some("wf-1".to_string()),
1820 workflow_name: Some("Main Workflow".to_string()),
1821 },
1822 SpawnNode {
1823 id: "level1-a".to_string(),
1824 queue_name: "processing".to_string(),
1825 status: "Completed".to_string(),
1826 depends_on: vec!["root".to_string()],
1827 spawn_config: Some(serde_json::json!({"spawn": true})),
1828 created_at: "2024-01-01T00:01:00Z".to_string(),
1829 workflow_id: Some("wf-1".to_string()),
1830 workflow_name: Some("Main Workflow".to_string()),
1831 },
1832 SpawnNode {
1833 id: "level1-b".to_string(),
1834 queue_name: "processing".to_string(),
1835 status: "Running".to_string(),
1836 depends_on: vec!["root".to_string()],
1837 spawn_config: None,
1838 created_at: "2024-01-01T00:02:00Z".to_string(),
1839 workflow_id: Some("wf-1".to_string()),
1840 workflow_name: Some("Main Workflow".to_string()),
1841 },
1842 SpawnNode {
1843 id: "level2-a".to_string(),
1844 queue_name: "finalize".to_string(),
1845 status: "Pending".to_string(),
1846 depends_on: vec!["level1-a".to_string()],
1847 spawn_config: None,
1848 created_at: "2024-01-01T00:03:00Z".to_string(),
1849 workflow_id: Some("wf-1".to_string()),
1850 workflow_name: Some("Main Workflow".to_string()),
1851 },
1852 ];
1853
1854 let cmd = SpawnCommand::List {
1855 database_url: None,
1856 limit: None,
1857 recent: false,
1858 queue: None,
1859 };
1860
1861 let edges = cmd.build_spawn_edges(&nodes);
1862 assert_eq!(edges.len(), 3); }
1864}