use anyhow::Result;
use clap::Subcommand;
use serde_json::Value;
use std::collections::{HashMap, HashSet, VecDeque};
use tracing::warn;
use uuid::Uuid;
use crate::config::Config;
use crate::utils::database::DatabasePool;
#[derive(Debug, Clone)]
pub struct SpawnNode {
pub id: String,
pub queue_name: String,
pub status: String,
pub depends_on: Vec<String>,
pub spawn_config: Option<Value>,
pub created_at: String,
pub workflow_id: Option<String>,
pub workflow_name: Option<String>,
}
#[derive(Debug, Clone)]
pub struct SpawnOperation {
pub parent_job_id: String,
pub spawned_jobs: Vec<String>,
pub spawned_at: String,
pub operation_id: Option<String>,
pub config: Option<Value>,
}
#[derive(Subcommand)]
pub enum SpawnCommand {
#[command(about = "List active spawn operations")]
List {
#[arg(short = 'u', long, help = "Database connection URL")]
database_url: Option<String>,
#[arg(short, long, help = "Maximum number of operations to display")]
limit: Option<u32>,
#[arg(long, help = "Show only recent spawn operations")]
recent: bool,
#[arg(long, help = "Show spawn operations for specific queue")]
queue: Option<String>,
},
#[command(about = "Show spawn tree hierarchy for a job")]
Tree {
#[arg(short = 'u', long, help = "Database connection URL")]
database_url: Option<String>,
#[arg(help = "Job ID (parent or child)")]
job_id: String,
#[arg(long, help = "Show full spawn tree (both up and down)")]
full: bool,
#[arg(long, help = "Show only children of this job")]
children_only: bool,
#[arg(long, help = "Output format (text, json, mermaid)")]
format: Option<String>,
},
#[command(about = "Show spawn statistics")]
Stats {
#[arg(short = 'u', long, help = "Database connection URL")]
database_url: Option<String>,
#[arg(short = 'q', long, help = "Filter by queue name")]
queue: Option<String>,
#[arg(long, help = "Time period in hours (default: 24)")]
hours: Option<u32>,
#[arg(long, help = "Show detailed breakdown")]
detailed: bool,
},
#[command(about = "Track spawn lineage for a job")]
Lineage {
#[arg(short = 'u', long, help = "Database connection URL")]
database_url: Option<String>,
#[arg(help = "Job ID")]
job_id: String,
#[arg(long, help = "Show ancestor chain")]
ancestors: bool,
#[arg(long, help = "Show descendant chain")]
descendants: bool,
#[arg(long, help = "Maximum depth to traverse")]
depth: Option<u32>,
},
#[command(about = "Show jobs waiting for spawn completion")]
Pending {
#[arg(short = 'u', long, help = "Database connection URL")]
database_url: Option<String>,
#[arg(short = 'q', long, help = "Filter by queue name")]
queue: Option<String>,
#[arg(long, help = "Show spawn configuration details")]
show_config: bool,
},
#[command(about = "Monitor spawn operations in real-time")]
Monitor {
#[arg(short = 'u', long, help = "Database connection URL")]
database_url: Option<String>,
#[arg(long, help = "Refresh interval in seconds (default: 5)")]
interval: Option<u32>,
#[arg(short = 'q', long, help = "Filter by queue name")]
queue: Option<String>,
},
}
impl SpawnCommand {
pub async fn execute(&self, config: Config) -> Result<()> {
let db_url = self.get_database_url(&config)?;
let pool = DatabasePool::connect(&db_url, config.get_connection_pool_size()).await?;
match self {
SpawnCommand::List {
limit,
recent,
queue,
..
} => {
self.list_spawn_operations(pool, *limit, *recent, queue.as_deref())
.await
}
SpawnCommand::Tree {
job_id,
full,
children_only,
format,
..
} => {
self.show_spawn_tree(pool, job_id, *full, *children_only, format.as_deref())
.await
}
SpawnCommand::Stats {
queue,
hours,
detailed,
..
} => {
self.show_spawn_stats(pool, queue.as_deref(), *hours, *detailed)
.await
}
SpawnCommand::Lineage {
job_id,
ancestors,
descendants,
depth,
..
} => {
self.show_spawn_lineage(pool, job_id, *ancestors, *descendants, *depth)
.await
}
SpawnCommand::Pending {
queue, show_config, ..
} => {
self.show_pending_spawns(pool, queue.as_deref(), *show_config)
.await
}
SpawnCommand::Monitor {
interval, queue, ..
} => {
self.monitor_spawn_operations(pool, *interval, queue.as_deref())
.await
}
}
}
pub fn get_database_url(&self, config: &Config) -> Result<String> {
let url_option = match self {
SpawnCommand::List { database_url, .. } => database_url,
SpawnCommand::Tree { database_url, .. } => database_url,
SpawnCommand::Stats { database_url, .. } => database_url,
SpawnCommand::Lineage { database_url, .. } => database_url,
SpawnCommand::Pending { database_url, .. } => database_url,
SpawnCommand::Monitor { database_url, .. } => database_url,
};
url_option
.as_ref()
.map(|s| s.as_str())
.or(config.get_database_url())
.map(|s| s.to_string())
.ok_or_else(|| anyhow::anyhow!("Database URL is required"))
}
async fn list_spawn_operations(
&self,
pool: DatabasePool,
limit: Option<u32>,
recent: bool,
queue_filter: Option<&str>,
) -> Result<()> {
let limit = limit.unwrap_or(20);
let time_filter = if recent {
"AND created_at > NOW() - INTERVAL '1 hour'"
} else {
""
};
let queue_clause = if let Some(queue) = queue_filter {
format!("AND queue_name = '{}'", queue)
} else {
String::new()
};
let query = format!(
r#"
SELECT parent.id as parent_id, parent.queue_name, parent.created_at,
parent.payload->'_spawn_config' as spawn_config,
COUNT(child.id) as spawned_count,
parent.workflow_id, parent.workflow_name
FROM hammerwork_jobs parent
LEFT JOIN hammerwork_jobs child ON child.depends_on @> CONCAT('[\"', parent.id, '\"]')::jsonb
WHERE parent.payload ? '_spawn_config'
AND parent.status IN ('Completed', 'Running')
{} {}
GROUP BY parent.id, parent.queue_name, parent.created_at, parent.payload, parent.workflow_id, parent.workflow_name
ORDER BY parent.created_at DESC
LIMIT {}
"#,
time_filter, queue_clause, limit
);
match pool {
DatabasePool::Postgres(pg_pool) => {
let rows = sqlx::query(&query).fetch_all(&pg_pool).await?;
println!("📊 Spawn Operations");
println!("{}", "=".repeat(80));
if rows.is_empty() {
println!("No spawn operations found.");
return Ok(());
}
println!(
"{:<8} {:<15} {:<12} {:<20} {:<10} {:<15}",
"Parent", "Queue", "Children", "Spawned At", "Operation", "Workflow"
);
println!("{}", "-".repeat(80));
for row in rows {
use sqlx::Row;
let parent_id: Uuid = row.get("parent_id");
let queue_name: String = row.get("queue_name");
let spawned_count: i64 = row.get("spawned_count");
let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
let spawn_config: Option<Value> = row.try_get("spawn_config").ok().flatten();
let operation_id = spawn_config
.as_ref()
.and_then(|c| c.get("operation_id"))
.and_then(|id| id.as_str())
.unwrap_or("none");
let workflow = row
.try_get::<Option<String>, _>("workflow_name")
.unwrap_or(None)
.unwrap_or_else(|| "none".to_string());
println!(
"{:<8} {:<15} {:<12} {:<20} {:<10} {:<15}",
&parent_id.to_string()[..8],
&queue_name[..std::cmp::min(15, queue_name.len())],
spawned_count,
created_at.format("%m-%d %H:%M:%S"),
&operation_id[..std::cmp::min(10, operation_id.len())],
&workflow[..std::cmp::min(15, workflow.len())]
);
}
}
DatabasePool::MySQL(mysql_pool) => {
let mysql_query = format!(
r#"
SELECT parent.id as parent_id, parent.queue_name, parent.created_at,
JSON_EXTRACT(parent.payload, '$._spawn_config') as spawn_config,
COUNT(child.id) as spawned_count,
parent.workflow_id, parent.workflow_name
FROM hammerwork_jobs parent
LEFT JOIN hammerwork_jobs child ON JSON_CONTAINS(child.depends_on, CONCAT('"', parent.id, '"'))
WHERE JSON_EXTRACT(parent.payload, '$._spawn_config') IS NOT NULL
AND parent.status IN ('Completed', 'Running')
{}
{}
GROUP BY parent.id, parent.queue_name, parent.created_at, parent.payload, parent.workflow_id, parent.workflow_name
ORDER BY parent.created_at DESC
LIMIT {}
"#,
if recent {
"AND parent.created_at > DATE_SUB(NOW(), INTERVAL 1 HOUR)"
} else {
""
},
queue_clause,
limit
);
let rows = sqlx::query(&mysql_query).fetch_all(&mysql_pool).await?;
println!("📊 Spawn Operations");
println!("{}", "=".repeat(80));
if rows.is_empty() {
println!("No spawn operations found.");
return Ok(());
}
println!(
"{:<8} {:<15} {:<12} {:<20} {:<10} {:<15}",
"Parent", "Queue", "Children", "Spawned At", "Operation", "Workflow"
);
println!("{}", "-".repeat(80));
for row in rows {
use sqlx::Row;
let parent_id: String = row.get("parent_id");
let queue_name: String = row.get("queue_name");
let spawned_count: i64 = row.get("spawned_count");
let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
let spawn_config_str: Option<String> =
row.try_get("spawn_config").ok().flatten();
let operation_id = spawn_config_str
.as_ref()
.and_then(|config_str| serde_json::from_str::<Value>(config_str).ok())
.and_then(|config| {
config
.get("operation_id")
.and_then(|id| id.as_str().map(|s| s.to_string()))
})
.unwrap_or_else(|| "none".to_string());
let workflow = row
.try_get::<Option<String>, _>("workflow_name")
.unwrap_or(None)
.unwrap_or_else(|| "none".to_string());
println!(
"{:<8} {:<15} {:<12} {:<20} {:<10} {:<15}",
&parent_id[..std::cmp::min(8, parent_id.len())],
&queue_name[..std::cmp::min(15, queue_name.len())],
spawned_count,
created_at.format("%m-%d %H:%M:%S"),
&operation_id[..std::cmp::min(10, operation_id.len())],
&workflow[..std::cmp::min(15, workflow.len())]
);
}
}
}
Ok(())
}
async fn show_spawn_tree(
&self,
pool: DatabasePool,
job_id: &str,
show_full: bool,
children_only: bool,
format: Option<&str>,
) -> Result<()> {
let format = format.unwrap_or("text");
let job_uuid = Uuid::parse_str(job_id)?;
let target_job = self.get_spawn_node(&pool, &job_uuid).await?;
let target_job = match target_job {
Some(job) => job,
None => {
println!("Job not found: {}", job_id);
return Ok(());
}
};
let spawn_tree = if show_full || !children_only {
self.collect_full_spawn_tree(&pool, &target_job).await?
} else {
self.collect_spawn_children(&pool, &target_job).await?
};
match format {
"text" => self.print_spawn_tree_text(&spawn_tree, &target_job.id),
"json" => self.print_spawn_tree_json(&spawn_tree)?,
"mermaid" => self.print_spawn_tree_mermaid(&spawn_tree, &target_job.id),
_ => {
println!("Unsupported format: {}. Use: text, json, mermaid", format);
return Ok(());
}
}
Ok(())
}
async fn show_spawn_stats(
&self,
pool: DatabasePool,
queue_filter: Option<&str>,
hours: Option<u32>,
detailed: bool,
) -> Result<()> {
let hours = hours.unwrap_or(24);
let queue_clause = if let Some(queue) = queue_filter {
format!("AND queue_name = '{}'", queue)
} else {
String::new()
};
println!("📈 Spawn Statistics (Last {} hours)", hours);
println!("{}", "=".repeat(60));
match pool {
DatabasePool::Postgres(pg_pool) => {
let total_query = format!(
r#"
SELECT COUNT(*) as total_spawn_ops,
AVG(spawned_count) as avg_children,
MAX(spawned_count) as max_children
FROM (
SELECT parent.id, COUNT(child.id) as spawned_count
FROM hammerwork_jobs parent
LEFT JOIN hammerwork_jobs child ON child.depends_on @> CONCAT('[\"', parent.id, '\"]')::jsonb
WHERE parent.payload ? '_spawn_config'
AND parent.created_at > NOW() - INTERVAL '{} hours'
{}
GROUP BY parent.id
) spawn_stats
"#,
hours, queue_clause
);
if let Ok(row) = sqlx::query(&total_query).fetch_one(&pg_pool).await {
use sqlx::Row;
let total: i64 = row.get("total_spawn_ops");
let avg: Option<f64> = row.try_get("avg_children").ok().flatten();
let max: Option<i64> = row.try_get("max_children").ok().flatten();
println!("Total Spawn Operations: {}", total);
if let Some(avg) = avg {
println!("Average Children per Spawn: {:.1}", avg);
}
if let Some(max) = max {
println!("Maximum Children in Single Spawn: {}", max);
}
}
if detailed {
println!("\n📋 Breakdown by Queue:");
let breakdown_query = format!(
r#"
SELECT queue_name,
COUNT(*) as spawn_count,
AVG(spawned_count) as avg_children
FROM (
SELECT parent.queue_name, COUNT(child.id) as spawned_count
FROM hammerwork_jobs parent
LEFT JOIN hammerwork_jobs child ON child.depends_on @> CONCAT('[\"', parent.id, '\"]')::jsonb
WHERE parent.payload ? '_spawn_config'
AND parent.created_at > NOW() - INTERVAL '{} hours'
{}
GROUP BY parent.id, parent.queue_name
) spawn_breakdown
GROUP BY queue_name
ORDER BY spawn_count DESC
"#,
hours, queue_clause
);
let rows = sqlx::query(&breakdown_query).fetch_all(&pg_pool).await?;
println!(
"{:<20} {:<12} {:<15}",
"Queue", "Operations", "Avg Children"
);
println!("{}", "-".repeat(47));
for row in rows {
use sqlx::Row;
let queue: String = row.get("queue_name");
let count: i64 = row.get("spawn_count");
let avg: Option<f64> = row.try_get("avg_children").ok().flatten();
println!(
"{:<20} {:<12} {:<15.1}",
&queue[..std::cmp::min(20, queue.len())],
count,
avg.unwrap_or(0.0)
);
}
}
}
DatabasePool::MySQL(mysql_pool) => {
let total_query = format!(
r#"
SELECT COUNT(*) as total_spawn_ops,
AVG(spawned_count) as avg_children,
MAX(spawned_count) as max_children
FROM (
SELECT parent.id, COUNT(child.id) as spawned_count
FROM hammerwork_jobs parent
LEFT JOIN hammerwork_jobs child ON JSON_CONTAINS(child.depends_on, CONCAT('"', parent.id, '"'))
WHERE JSON_EXTRACT(parent.payload, '$._spawn_config') IS NOT NULL
AND parent.created_at > DATE_SUB(NOW(), INTERVAL {} HOUR)
{}
GROUP BY parent.id
) spawn_stats
"#,
hours, queue_clause
);
if let Ok(row) = sqlx::query(&total_query).fetch_one(&mysql_pool).await {
use sqlx::Row;
let total: i64 = row.get("total_spawn_ops");
let avg: Option<f64> = row.try_get("avg_children").ok();
let max: Option<i64> = row.try_get("max_children").ok();
println!("Total Spawn Operations: {}", total);
if let Some(avg) = avg {
println!("Average Children per Spawn: {:.1}", avg);
}
if let Some(max) = max {
println!("Maximum Children in Single Spawn: {}", max);
}
}
if detailed {
println!("\n📋 Breakdown by Queue:");
let breakdown_query = format!(
r#"
SELECT queue_name,
COUNT(*) as spawn_count,
AVG(spawned_count) as avg_children
FROM (
SELECT parent.queue_name, COUNT(child.id) as spawned_count
FROM hammerwork_jobs parent
LEFT JOIN hammerwork_jobs child ON JSON_CONTAINS(child.depends_on, CONCAT('"', parent.id, '"'))
WHERE JSON_EXTRACT(parent.payload, '$._spawn_config') IS NOT NULL
AND parent.created_at > DATE_SUB(NOW(), INTERVAL {} HOUR)
{}
GROUP BY parent.id, parent.queue_name
) spawn_breakdown
GROUP BY queue_name
ORDER BY spawn_count DESC
"#,
hours, queue_clause
);
let rows = sqlx::query(&breakdown_query).fetch_all(&mysql_pool).await?;
println!(
"{:<20} {:<12} {:<15}",
"Queue", "Operations", "Avg Children"
);
println!("{}", "-".repeat(47));
for row in rows {
use sqlx::Row;
let queue: String = row.get("queue_name");
let count: i64 = row.get("spawn_count");
let avg: Option<f64> = row.try_get("avg_children").ok();
println!(
"{:<20} {:<12} {:<15.1}",
&queue[..std::cmp::min(20, queue.len())],
count,
avg.unwrap_or(0.0)
);
}
}
}
}
Ok(())
}
async fn show_spawn_lineage(
&self,
pool: DatabasePool,
job_id: &str,
show_ancestors: bool,
show_descendants: bool,
max_depth: Option<u32>,
) -> Result<()> {
let job_uuid = Uuid::parse_str(job_id)?;
let max_depth = max_depth.unwrap_or(10);
let target_job = self.get_spawn_node(&pool, &job_uuid).await?;
let target_job = match target_job {
Some(job) => job,
None => {
println!("Job not found: {}", job_id);
return Ok(());
}
};
println!("🔗 Spawn Lineage for {}", job_id);
println!(
"Queue: {} | Status: {}",
target_job.queue_name, target_job.status
);
println!("{}", "=".repeat(60));
if show_ancestors || (!show_descendants) {
println!("\n⬆️ Ancestor Chain:");
let ancestors = self
.collect_ancestors(&pool, &target_job, max_depth)
.await?;
if ancestors.is_empty() {
println!(" No spawn ancestors found (this is a root job)");
} else {
for (depth, ancestor) in ancestors.iter().enumerate() {
let indent = " ".repeat(depth + 1);
println!(
"{}└─ [{}] {} ({})",
indent,
&ancestor.id[..8],
ancestor.queue_name,
ancestor.status
);
}
}
}
if show_descendants || (!show_ancestors) {
println!("\n⬇️ Descendant Chain:");
let descendants = self
.collect_descendants(&pool, &target_job, max_depth)
.await?;
if descendants.is_empty() {
println!(" No spawn descendants found (this job hasn't spawned children)");
} else {
self.print_descendant_tree(&descendants, &target_job.id, 0);
}
}
Ok(())
}
async fn show_pending_spawns(
&self,
pool: DatabasePool,
queue_filter: Option<&str>,
show_config: bool,
) -> Result<()> {
let queue_clause = if let Some(queue) = queue_filter {
format!("AND queue_name = '{}'", queue)
} else {
String::new()
};
println!("⏳ Jobs with Pending Spawn Operations");
println!("{}", "=".repeat(70));
match pool {
DatabasePool::Postgres(pg_pool) => {
let query = format!(
r#"
SELECT id, queue_name, status, created_at,
payload->'_spawn_config' as spawn_config
FROM hammerwork_jobs
WHERE payload ? '_spawn_config'
AND status IN ('Running', 'Pending')
{}
ORDER BY created_at DESC
LIMIT 50
"#,
queue_clause
);
let rows = sqlx::query(&query).fetch_all(&pg_pool).await?;
if rows.is_empty() {
println!("No jobs with pending spawn operations found.");
return Ok(());
}
for row in rows {
use sqlx::Row;
let id: Uuid = row.get("id");
let queue_name: String = row.get("queue_name");
let status: String = row.get("status");
let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
let spawn_config: Option<Value> = row.try_get("spawn_config").ok().flatten();
println!(
"📋 Job: {} | Queue: {} | Status: {} | Created: {}",
&id.to_string()[..8],
queue_name,
status,
created_at.format("%m-%d %H:%M:%S")
);
if show_config {
if let Some(config) = spawn_config {
println!(
" Spawn Config: {}",
serde_json::to_string_pretty(&config)
.unwrap_or_else(|_| "Invalid JSON".to_string())
);
}
println!();
}
}
}
DatabasePool::MySQL(mysql_pool) => {
let query = format!(
r#"
SELECT id, queue_name, status, created_at,
JSON_EXTRACT(payload, '$._spawn_config') as spawn_config
FROM hammerwork_jobs
WHERE JSON_EXTRACT(payload, '$._spawn_config') IS NOT NULL
AND status IN ('Running', 'Pending')
{}
ORDER BY created_at DESC
LIMIT 50
"#,
queue_clause
);
let rows = sqlx::query(&query).fetch_all(&mysql_pool).await?;
if rows.is_empty() {
println!("No jobs with pending spawn operations found.");
return Ok(());
}
for row in rows {
use sqlx::Row;
let id: String = row.get("id");
let queue_name: String = row.get("queue_name");
let status: String = row.get("status");
let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
let spawn_config_str: Option<String> =
row.try_get("spawn_config").ok().flatten();
println!(
"📋 Job: {} | Queue: {} | Status: {} | Created: {}",
&id[..std::cmp::min(8, id.len())],
queue_name,
status,
created_at.format("%m-%d %H:%M:%S")
);
if show_config {
if let Some(ref config_str) = spawn_config_str {
match serde_json::from_str::<Value>(config_str) {
Ok(config) => {
println!(
" Spawn Config: {}",
serde_json::to_string_pretty(&config)
.unwrap_or_else(|_| "Invalid JSON".to_string())
);
}
Err(_) => {
println!(" Spawn Config: {}", config_str);
}
}
}
println!();
}
}
}
}
Ok(())
}
async fn monitor_spawn_operations(
&self,
pool: DatabasePool,
interval: Option<u32>,
queue_filter: Option<&str>,
) -> Result<()> {
let interval = std::time::Duration::from_secs(interval.unwrap_or(5) as u64);
println!("🔄 Monitoring Spawn Operations (Press Ctrl+C to stop)");
println!("Refresh interval: {:?}", interval);
if let Some(queue) = queue_filter {
println!("Queue filter: {}", queue);
}
println!("{}", "=".repeat(80));
match &pool {
DatabasePool::Postgres(_pg_pool) => {
warn!("Monitor mode requires reconnection for each refresh");
Err(anyhow::anyhow!(
"Monitor mode needs database URL to reconnect. Use list/stats commands instead for one-time queries."
))
}
DatabasePool::MySQL(_) => {
warn!("Monitor mode requires reconnection for each refresh");
Err(anyhow::anyhow!(
"Monitor mode needs database URL to reconnect. Use list/stats commands instead for one-time queries."
))
}
}
}
async fn get_spawn_node(
&self,
pool: &DatabasePool,
job_id: &Uuid,
) -> Result<Option<SpawnNode>> {
let query = r#"
SELECT id, queue_name, status, depends_on,
payload->'_spawn_config' as spawn_config,
created_at, workflow_id, workflow_name
FROM hammerwork_jobs
WHERE id = $1
"#;
match pool {
DatabasePool::Postgres(pg_pool) => {
if let Some(row) = sqlx::query(query)
.bind(job_id)
.fetch_optional(pg_pool)
.await?
{
Ok(Some(self.postgres_row_to_spawn_node(&row)?))
} else {
Ok(None)
}
}
DatabasePool::MySQL(mysql_pool) => {
let mysql_query = query.replace("$1", "?");
if let Some(row) = sqlx::query(&mysql_query)
.bind(job_id.to_string())
.fetch_optional(mysql_pool)
.await?
{
Ok(Some(self.mysql_row_to_spawn_node(&row)?))
} else {
Ok(None)
}
}
}
}
async fn collect_spawn_children(
&self,
pool: &DatabasePool,
parent: &SpawnNode,
) -> Result<Vec<SpawnNode>> {
let query = r#"
SELECT id, queue_name, status, depends_on,
payload->'_spawn_config' as spawn_config,
created_at, workflow_id, workflow_name
FROM hammerwork_jobs
WHERE depends_on @> $1
ORDER BY created_at
"#;
let mut children = Vec::new();
match pool {
DatabasePool::Postgres(pg_pool) => {
let parent_json = format!("[\"{}\"]", parent.id);
let rows = sqlx::query(query)
.bind(&parent_json)
.fetch_all(pg_pool)
.await?;
for row in rows {
children.push(self.postgres_row_to_spawn_node(&row)?);
}
}
DatabasePool::MySQL(mysql_pool) => {
let mysql_query = r#"
SELECT id, queue_name, status, depends_on,
JSON_EXTRACT(payload, '$._spawn_config') as spawn_config,
created_at, workflow_id, workflow_name
FROM hammerwork_jobs
WHERE JSON_CONTAINS(depends_on, ?)
ORDER BY created_at
"#;
let parent_json = format!("\"{}\"", parent.id);
let rows = sqlx::query(mysql_query)
.bind(&parent_json)
.fetch_all(mysql_pool)
.await?;
for row in rows {
children.push(self.mysql_row_to_spawn_node(&row)?);
}
}
}
Ok(children)
}
async fn collect_full_spawn_tree(
&self,
pool: &DatabasePool,
target: &SpawnNode,
) -> Result<Vec<SpawnNode>> {
let mut all_nodes = HashMap::new();
let mut to_visit = VecDeque::new();
let mut visited = HashSet::new();
all_nodes.insert(target.id.clone(), target.clone());
to_visit.push_back(target.id.clone());
while let Some(job_id) = to_visit.pop_front() {
if visited.contains(&job_id) {
continue;
}
visited.insert(job_id.clone());
if let Some(job) = all_nodes.get(&job_id).cloned() {
let children = self.collect_spawn_children(pool, &job).await?;
for child in children {
if !all_nodes.contains_key(&child.id) {
all_nodes.insert(child.id.clone(), child.clone());
to_visit.push_back(child.id.clone());
}
}
for parent_id in &job.depends_on {
if let Ok(parent_uuid) = Uuid::parse_str(parent_id) {
if let Ok(Some(parent)) = self.get_spawn_node(pool, &parent_uuid).await {
if !all_nodes.contains_key(&parent.id) {
all_nodes.insert(parent.id.clone(), parent.clone());
to_visit.push_back(parent.id.clone());
}
}
}
}
}
}
Ok(all_nodes.into_values().collect())
}
async fn collect_ancestors(
&self,
pool: &DatabasePool,
job: &SpawnNode,
max_depth: u32,
) -> Result<Vec<SpawnNode>> {
let mut ancestors = Vec::new();
let mut current = job.clone();
let mut depth = 0;
while depth < max_depth && !current.depends_on.is_empty() {
let mut parent_found = false;
for parent_id in ¤t.depends_on {
if let Ok(parent_uuid) = Uuid::parse_str(parent_id) {
if let Ok(Some(parent)) = self.get_spawn_node(pool, &parent_uuid).await {
if parent.spawn_config.is_some() {
ancestors.push(parent.clone());
current = parent;
parent_found = true;
break;
}
}
}
}
if !parent_found {
break;
}
depth += 1;
}
ancestors.reverse(); Ok(ancestors)
}
async fn collect_descendants(
&self,
pool: &DatabasePool,
job: &SpawnNode,
max_depth: u32,
) -> Result<Vec<SpawnNode>> {
let mut descendants = Vec::new();
let mut to_visit = VecDeque::new();
let mut visited = HashSet::new();
to_visit.push_back((job.clone(), 0));
while let Some((current_job, depth)) = to_visit.pop_front() {
if depth >= max_depth || visited.contains(¤t_job.id) {
continue;
}
visited.insert(current_job.id.clone());
let children = self.collect_spawn_children(pool, ¤t_job).await?;
for child in children {
descendants.push(child.clone());
to_visit.push_back((child, depth + 1));
}
}
Ok(descendants)
}
fn postgres_row_to_spawn_node(&self, row: &sqlx::postgres::PgRow) -> Result<SpawnNode> {
use sqlx::Row;
let id: Uuid = row.get("id");
let depends_on = self.parse_json_array(row.try_get("depends_on").ok())?;
let spawn_config: Option<Value> = row.try_get("spawn_config").ok().flatten();
let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
let workflow_id: Option<String> = match row.try_get::<Option<Uuid>, _>("workflow_id") {
Ok(Some(uuid)) => Some(uuid.to_string()),
Ok(None) => None,
Err(_) => None,
};
Ok(SpawnNode {
id: id.to_string(),
queue_name: row.get("queue_name"),
status: row.get("status"),
depends_on,
spawn_config,
created_at: created_at.to_string(),
workflow_id,
workflow_name: row.try_get("workflow_name").ok(),
})
}
fn mysql_row_to_spawn_node(&self, row: &sqlx::mysql::MySqlRow) -> Result<SpawnNode> {
use sqlx::Row;
let id: String = row.get("id");
let depends_on = self.parse_json_array(row.try_get("depends_on").ok())?;
let spawn_config: Option<Value> = match row.try_get::<Option<String>, _>("spawn_config") {
Ok(Some(config_str)) => serde_json::from_str(&config_str).ok(),
Ok(None) => None,
Err(_) => {
row.try_get("spawn_config").ok().flatten()
}
};
let created_at: chrono::DateTime<chrono::Utc> = row.get("created_at");
Ok(SpawnNode {
id,
queue_name: row.get("queue_name"),
status: row.get("status"),
depends_on,
spawn_config,
created_at: created_at.to_string(),
workflow_id: row.try_get("workflow_id").ok(),
workflow_name: row.try_get("workflow_name").ok(),
})
}
fn parse_json_array(&self, json_value: Option<Value>) -> Result<Vec<String>> {
match json_value {
Some(Value::Array(arr)) => Ok(arr
.into_iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect()),
_ => Ok(Vec::new()),
}
}
fn print_spawn_tree_text(&self, nodes: &[SpawnNode], target_id: &str) {
println!("\n🌳 Spawn Tree");
println!("{}", "=".repeat(60));
let node_map: HashMap<String, &SpawnNode> =
nodes.iter().map(|n| (n.id.clone(), n)).collect();
let mut roots: Vec<&SpawnNode> = nodes
.iter()
.filter(|node| {
!node.depends_on.iter().any(|dep_id| {
node_map
.get(dep_id)
.is_some_and(|dep| dep.spawn_config.is_some())
})
})
.collect();
if roots.is_empty() {
roots = nodes.iter().collect();
}
roots.sort_by(|a, b| a.created_at.cmp(&b.created_at));
let mut visited = HashSet::new();
for root in roots {
if !visited.contains(&root.id) {
self.print_spawn_node_tree(root, &node_map, &mut visited, 0, target_id);
}
}
}
#[allow(clippy::only_used_in_recursion)]
fn print_spawn_node_tree(
&self,
node: &SpawnNode,
node_map: &HashMap<String, &SpawnNode>,
visited: &mut HashSet<String>,
depth: usize,
target_id: &str,
) {
if visited.contains(&node.id) {
return;
}
visited.insert(node.id.clone());
let indent = " ".repeat(depth);
let marker = if depth == 0 { "┌─" } else { "├─" };
let highlight = if node.id == target_id { " ⭐" } else { "" };
let spawn_indicator = if node.spawn_config.is_some() {
"🚀"
} else {
"📝"
};
println!(
"{}{}{} [{}] {} ({}){}",
indent,
marker,
spawn_indicator,
&node.id[..8],
node.queue_name,
node.status,
highlight
);
let children: Vec<&SpawnNode> = node_map
.values()
.filter(|child| child.depends_on.contains(&node.id) && child.id != node.id)
.copied()
.collect();
for child in children {
self.print_spawn_node_tree(child, node_map, visited, depth + 1, target_id);
}
}
fn print_spawn_tree_json(&self, nodes: &[SpawnNode]) -> Result<()> {
println!("\n📄 Spawn Tree (JSON)");
println!("{}", "=".repeat(60));
let tree_data = serde_json::json!({
"spawn_tree": {
"nodes": nodes.iter().map(|node| {
serde_json::json!({
"id": node.id,
"queue": node.queue_name,
"status": node.status,
"depends_on": node.depends_on,
"has_spawn_config": node.spawn_config.is_some(),
"spawn_config": node.spawn_config,
"created_at": node.created_at,
"workflow_id": node.workflow_id,
"workflow_name": node.workflow_name
})
}).collect::<Vec<_>>(),
"edges": self.build_spawn_edges(nodes)
}
});
println!("{}", serde_json::to_string_pretty(&tree_data)?);
Ok(())
}
fn print_spawn_tree_mermaid(&self, nodes: &[SpawnNode], target_id: &str) {
println!("\n🧜♀️ Spawn Tree (Mermaid)");
println!("{}", "=".repeat(60));
println!("graph TD");
println!(" subgraph \"🚀 Spawn Tree\"");
for node in nodes {
let short_id = &node.id[..8];
let status_class = match node.status.as_str() {
"Completed" => ":::completed",
"Failed" => ":::failed",
"Running" => ":::running",
"Pending" => ":::pending",
_ => ":::default",
};
let spawn_indicator = if node.spawn_config.is_some() {
"🚀"
} else {
"📝"
};
let target_indicator = if node.id == target_id { "⭐" } else { "" };
println!(
" {}[\"{}{}<br/>{}<br/>{}\"]{}",
short_id, spawn_indicator, target_indicator, short_id, node.status, status_class
);
}
let node_map: HashMap<String, &SpawnNode> =
nodes.iter().map(|n| (n.id.clone(), n)).collect();
for node in nodes {
for dep_id in &node.depends_on {
if let Some(parent) = node_map.get(dep_id) {
if parent.spawn_config.is_some() {
println!(" {} -->|spawns| {}", &dep_id[..8], &node.id[..8]);
}
}
}
}
println!(" end");
println!();
println!(" classDef completed fill:#d4edda,stroke:#155724");
println!(" classDef failed fill:#f8d7da,stroke:#721c24");
println!(" classDef running fill:#cce7ff,stroke:#004085");
println!(" classDef pending fill:#fff3cd,stroke:#856404");
println!(" classDef default fill:#e2e3e5,stroke:#383d41");
}
fn build_spawn_edges(&self, nodes: &[SpawnNode]) -> Vec<Value> {
let node_map: HashMap<String, &SpawnNode> =
nodes.iter().map(|n| (n.id.clone(), n)).collect();
let mut edges = Vec::new();
for node in nodes {
for dep_id in &node.depends_on {
if let Some(parent) = node_map.get(dep_id) {
if parent.spawn_config.is_some() {
edges.push(serde_json::json!({
"from": dep_id,
"to": node.id,
"type": "spawn",
"relationship": "parent_spawned_child"
}));
}
}
}
}
edges
}
fn print_descendant_tree(&self, descendants: &[SpawnNode], _target_id: &str, depth: usize) {
if descendants.is_empty() {
return;
}
for (i, descendant) in descendants.iter().enumerate() {
let indent = " ".repeat(depth + 1);
let marker = if i == descendants.len() - 1 {
"└─"
} else {
"├─"
};
println!(
"{}{}📝 [{}] {} ({})",
indent,
marker,
&descendant.id[..8],
descendant.queue_name,
descendant.status
);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use clap::Parser;
#[test]
fn test_spawn_command_variants() {
let commands = vec![
SpawnCommand::List {
database_url: None,
limit: Some(10),
recent: false,
queue: None,
},
SpawnCommand::Tree {
database_url: None,
job_id: "test-job".to_string(),
full: false,
children_only: false,
format: None,
},
SpawnCommand::Stats {
database_url: None,
queue: None,
hours: Some(24),
detailed: false,
},
SpawnCommand::Lineage {
database_url: None,
job_id: "test-job".to_string(),
ancestors: false,
descendants: false,
depth: None,
},
SpawnCommand::Pending {
database_url: None,
queue: None,
show_config: false,
},
SpawnCommand::Monitor {
database_url: None,
interval: Some(5),
queue: None,
},
];
assert_eq!(commands.len(), 6);
}
#[test]
fn test_spawn_node_creation() {
let spawn_node = SpawnNode {
id: "test-spawn-123".to_string(),
queue_name: "spawning-queue".to_string(),
status: "Completed".to_string(),
depends_on: vec!["parent-job".to_string()],
spawn_config: Some(serde_json::json!({"max_spawn_count": 5})),
created_at: "2024-01-01T00:00:00Z".to_string(),
workflow_id: Some("workflow-123".to_string()),
workflow_name: Some("test-workflow".to_string()),
};
assert_eq!(spawn_node.id, "test-spawn-123");
assert!(spawn_node.spawn_config.is_some());
assert_eq!(spawn_node.depends_on.len(), 1);
}
#[test]
fn test_spawn_operation_creation() {
let spawn_op = SpawnOperation {
parent_job_id: "parent-123".to_string(),
spawned_jobs: vec!["child-1".to_string(), "child-2".to_string()],
spawned_at: "2024-01-01T00:00:00Z".to_string(),
operation_id: Some("op-123".to_string()),
config: Some(serde_json::json!({"batch_size": 10})),
};
assert_eq!(spawn_op.parent_job_id, "parent-123");
assert_eq!(spawn_op.spawned_jobs.len(), 2);
assert!(spawn_op.operation_id.is_some());
assert!(spawn_op.config.is_some());
}
#[test]
fn test_json_array_parsing() {
let spawn_cmd = SpawnCommand::List {
database_url: None,
limit: None,
recent: false,
queue: None,
};
let json_array = serde_json::json!(["job1", "job2", "job3"]);
let result = spawn_cmd.parse_json_array(Some(json_array)).unwrap();
assert_eq!(result, vec!["job1", "job2", "job3"]);
let result = spawn_cmd.parse_json_array(None).unwrap();
assert!(result.is_empty());
let json_object = serde_json::json!({"key": "value"});
let result = spawn_cmd.parse_json_array(Some(json_object)).unwrap();
assert!(result.is_empty());
let mixed_array = serde_json::json!(["string1", 123, true, "string2", null]);
let result = spawn_cmd.parse_json_array(Some(mixed_array)).unwrap();
assert_eq!(result, vec!["string1", "string2"]);
}
#[test]
fn test_get_database_url() {
let config = Config::default();
let cmd = SpawnCommand::List {
database_url: Some("postgres://test".to_string()),
limit: None,
recent: false,
queue: None,
};
let result = cmd.get_database_url(&config);
assert!(result.is_ok());
assert_eq!(result.unwrap(), "postgres://test");
let cmd = SpawnCommand::List {
database_url: None,
limit: None,
recent: false,
queue: None,
};
let result = cmd.get_database_url(&config);
assert!(result.is_err());
}
#[test]
fn test_spawn_command_parsing() {
#[derive(Parser)]
struct TestApp {
#[command(subcommand)]
command: SpawnCommand,
}
let app = TestApp::try_parse_from(&[
"test",
"list",
"--limit",
"50",
"--recent",
"--queue",
"test-queue",
]);
assert!(app.is_ok());
match app.unwrap().command {
SpawnCommand::List {
limit,
recent,
queue,
..
} => {
assert_eq!(limit, Some(50));
assert!(recent);
assert_eq!(queue, Some("test-queue".to_string()));
}
_ => panic!("Wrong command variant"),
}
let app = TestApp::try_parse_from(&[
"test",
"tree",
"550e8400-e29b-41d4-a716-446655440000",
"--full",
"--format",
"json",
]);
assert!(app.is_ok());
match app.unwrap().command {
SpawnCommand::Tree {
job_id,
full,
format,
..
} => {
assert_eq!(job_id, "550e8400-e29b-41d4-a716-446655440000");
assert!(full);
assert_eq!(format, Some("json".to_string()));
}
_ => panic!("Wrong command variant"),
}
let app = TestApp::try_parse_from(&[
"test",
"stats",
"--hours",
"48",
"--detailed",
"--queue",
"stats-queue",
]);
assert!(app.is_ok());
match app.unwrap().command {
SpawnCommand::Stats {
hours,
detailed,
queue,
..
} => {
assert_eq!(hours, Some(48));
assert!(detailed);
assert_eq!(queue, Some("stats-queue".to_string()));
}
_ => panic!("Wrong command variant"),
}
let app = TestApp::try_parse_from(&[
"test",
"lineage",
"test-job-id",
"--ancestors",
"--descendants",
"--depth",
"5",
]);
assert!(app.is_ok());
match app.unwrap().command {
SpawnCommand::Lineage {
job_id,
ancestors,
descendants,
depth,
..
} => {
assert_eq!(job_id, "test-job-id");
assert!(ancestors);
assert!(descendants);
assert_eq!(depth, Some(5));
}
_ => panic!("Wrong command variant"),
}
let app = TestApp::try_parse_from(&[
"test",
"pending",
"--queue",
"pending-queue",
"--show-config",
]);
assert!(app.is_ok());
match app.unwrap().command {
SpawnCommand::Pending {
queue, show_config, ..
} => {
assert_eq!(queue, Some("pending-queue".to_string()));
assert!(show_config);
}
_ => panic!("Wrong command variant"),
}
let app = TestApp::try_parse_from(&[
"test",
"monitor",
"--interval",
"10",
"--queue",
"monitor-queue",
]);
assert!(app.is_ok());
match app.unwrap().command {
SpawnCommand::Monitor {
interval, queue, ..
} => {
assert_eq!(interval, Some(10));
assert_eq!(queue, Some("monitor-queue".to_string()));
}
_ => panic!("Wrong command variant"),
}
}
#[test]
fn test_build_spawn_edges() {
let cmd = SpawnCommand::List {
database_url: None,
limit: None,
recent: false,
queue: None,
};
let nodes = vec![
SpawnNode {
id: "parent-1".to_string(),
queue_name: "queue1".to_string(),
status: "Completed".to_string(),
depends_on: vec![],
spawn_config: Some(serde_json::json!({"spawn": true})),
created_at: "2024-01-01T00:00:00Z".to_string(),
workflow_id: None,
workflow_name: None,
},
SpawnNode {
id: "child-1".to_string(),
queue_name: "queue2".to_string(),
status: "Running".to_string(),
depends_on: vec!["parent-1".to_string()],
spawn_config: None,
created_at: "2024-01-01T00:01:00Z".to_string(),
workflow_id: None,
workflow_name: None,
},
SpawnNode {
id: "child-2".to_string(),
queue_name: "queue2".to_string(),
status: "Pending".to_string(),
depends_on: vec!["parent-1".to_string()],
spawn_config: None,
created_at: "2024-01-01T00:02:00Z".to_string(),
workflow_id: None,
workflow_name: None,
},
];
let edges = cmd.build_spawn_edges(&nodes);
assert_eq!(edges.len(), 2);
assert_eq!(edges[0]["from"], "parent-1");
assert_eq!(edges[0]["to"], "child-1");
assert_eq!(edges[0]["type"], "spawn");
assert_eq!(edges[1]["from"], "parent-1");
assert_eq!(edges[1]["to"], "child-2");
assert_eq!(edges[1]["type"], "spawn");
}
#[test]
fn test_spawn_node_with_no_dependencies() {
let node = SpawnNode {
id: "standalone-job".to_string(),
queue_name: "solo-queue".to_string(),
status: "Completed".to_string(),
depends_on: vec![],
spawn_config: None,
created_at: "2024-01-01T00:00:00Z".to_string(),
workflow_id: None,
workflow_name: None,
};
assert!(node.depends_on.is_empty());
assert!(node.spawn_config.is_none());
assert!(node.workflow_id.is_none());
}
#[test]
fn test_spawn_node_with_workflow() {
let node = SpawnNode {
id: "workflow-job".to_string(),
queue_name: "workflow-queue".to_string(),
status: "Running".to_string(),
depends_on: vec!["parent-1".to_string(), "parent-2".to_string()],
spawn_config: Some(serde_json::json!({
"operation_id": "op-123",
"batch_size": 100
})),
created_at: "2024-01-01T00:00:00Z".to_string(),
workflow_id: Some("workflow-123".to_string()),
workflow_name: Some("data-processing-workflow".to_string()),
};
assert_eq!(node.depends_on.len(), 2);
assert!(node.spawn_config.is_some());
assert_eq!(node.workflow_id, Some("workflow-123".to_string()));
assert_eq!(
node.workflow_name,
Some("data-processing-workflow".to_string())
);
}
#[test]
fn test_spawn_operation_without_config() {
let op = SpawnOperation {
parent_job_id: "parent-456".to_string(),
spawned_jobs: vec![],
spawned_at: "2024-01-01T12:00:00Z".to_string(),
operation_id: None,
config: None,
};
assert!(op.spawned_jobs.is_empty());
assert!(op.operation_id.is_none());
assert!(op.config.is_none());
}
#[test]
fn test_format_options() {
let formats = vec!["text", "json", "mermaid"];
for format in formats {
let cmd = SpawnCommand::Tree {
database_url: None,
job_id: "test".to_string(),
full: false,
children_only: false,
format: Some(format.to_string()),
};
match cmd {
SpawnCommand::Tree {
format: Some(f), ..
} => {
assert!(["text", "json", "mermaid"].contains(&f.as_str()));
}
_ => panic!("Wrong command variant"),
}
}
}
#[test]
fn test_spawn_config_parsing() {
let configs = vec![
serde_json::json!({"operation_id": "op-1"}),
serde_json::json!({"max_spawn_count": 100, "batch_size": 10}),
serde_json::json!({"spawn_after_seconds": 60}),
serde_json::json!({}), ];
for config in configs {
let node = SpawnNode {
id: "test".to_string(),
queue_name: "test".to_string(),
status: "Pending".to_string(),
depends_on: vec![],
spawn_config: Some(config.clone()),
created_at: "2024-01-01T00:00:00Z".to_string(),
workflow_id: None,
workflow_name: None,
};
assert!(node.spawn_config.is_some());
assert_eq!(node.spawn_config.unwrap(), config);
}
}
#[test]
fn test_job_status_variants() {
let statuses = vec![
"Pending",
"Running",
"Completed",
"Failed",
"Retrying",
"Dead",
"TimedOut",
];
for status in statuses {
let node = SpawnNode {
id: "test".to_string(),
queue_name: "test".to_string(),
status: status.to_string(),
depends_on: vec![],
spawn_config: None,
created_at: "2024-01-01T00:00:00Z".to_string(),
workflow_id: None,
workflow_name: None,
};
assert!(!node.status.is_empty());
}
}
#[test]
fn test_command_defaults() {
let list_cmd = SpawnCommand::List {
database_url: None,
limit: None,
recent: false,
queue: None,
};
match list_cmd {
SpawnCommand::List {
limit,
recent,
queue,
..
} => {
assert!(limit.is_none());
assert!(!recent);
assert!(queue.is_none());
}
_ => panic!("Wrong variant"),
}
let monitor_cmd = SpawnCommand::Monitor {
database_url: None,
interval: None,
queue: None,
};
match monitor_cmd {
SpawnCommand::Monitor { interval, .. } => {
assert!(interval.is_none());
}
_ => panic!("Wrong variant"),
}
}
#[test]
fn test_complex_spawn_tree() {
let nodes = vec![
SpawnNode {
id: "root".to_string(),
queue_name: "root-queue".to_string(),
status: "Completed".to_string(),
depends_on: vec![],
spawn_config: Some(serde_json::json!({"spawn": true})),
created_at: "2024-01-01T00:00:00Z".to_string(),
workflow_id: Some("wf-1".to_string()),
workflow_name: Some("Main Workflow".to_string()),
},
SpawnNode {
id: "level1-a".to_string(),
queue_name: "processing".to_string(),
status: "Completed".to_string(),
depends_on: vec!["root".to_string()],
spawn_config: Some(serde_json::json!({"spawn": true})),
created_at: "2024-01-01T00:01:00Z".to_string(),
workflow_id: Some("wf-1".to_string()),
workflow_name: Some("Main Workflow".to_string()),
},
SpawnNode {
id: "level1-b".to_string(),
queue_name: "processing".to_string(),
status: "Running".to_string(),
depends_on: vec!["root".to_string()],
spawn_config: None,
created_at: "2024-01-01T00:02:00Z".to_string(),
workflow_id: Some("wf-1".to_string()),
workflow_name: Some("Main Workflow".to_string()),
},
SpawnNode {
id: "level2-a".to_string(),
queue_name: "finalize".to_string(),
status: "Pending".to_string(),
depends_on: vec!["level1-a".to_string()],
spawn_config: None,
created_at: "2024-01-01T00:03:00Z".to_string(),
workflow_id: Some("wf-1".to_string()),
workflow_name: Some("Main Workflow".to_string()),
},
];
let cmd = SpawnCommand::List {
database_url: None,
limit: None,
recent: false,
queue: None,
};
let edges = cmd.build_spawn_edges(&nodes);
assert_eq!(edges.len(), 3); }
}