diff --git a/src/worker/job_worker.rs b/src/worker/job_worker.rs index c3b3606..9ba9ff6 100644 --- a/src/worker/job_worker.rs +++ b/src/worker/job_worker.rs @@ -1243,20 +1243,42 @@ impl JobWorker { true }; - let all_ok = asr_done && trace_done; + // Check TKG nodes exist + let has_asr_or_asrx_for_tkg = + job_processors.is_empty() || job_processors.iter().any(|p| p == "asrx" || p == "asr"); + let has_face_for_tkg = job_processors.is_empty() || job_processors.iter().any(|p| p == "face"); + + let tkg_done: bool = if has_asr_or_asrx_for_tkg && has_face_for_tkg { + let tkg_nodes_table = schema::table_name("tkg_nodes"); + sqlx::query_scalar::<_, i32>(&format!( + "SELECT 1 FROM {tkg_nodes_table} WHERE file_uuid = $1 LIMIT 1" + )) + .bind(uuid) + .fetch_optional(pool) + .await + .unwrap_or(None) + .unwrap_or(0) > 0 + } else { + tracing::info!("[Ingestion] No TKG needed for {}", uuid); + true + }; + + let all_ok = asr_done && trace_done && tkg_done; tracing::info!( - "[Ingestion] all_ok={} (asr_done={}, trace_done={}) for uuid={}", + "[Ingestion] all_ok={} (asr_done={}, trace_done={}, tkg_done={}) for uuid={}", all_ok, asr_done, trace_done, + tkg_done, uuid ); if !all_ok { tracing::info!( - "[Ingestion] waiting (uuid={}): asr_done={} trace_done={}", + "[Ingestion] waiting (uuid={}): asr_done={} trace_done={} tkg_done={}", uuid, asr_done, - trace_done + trace_done, + tkg_done ); } all_ok