vfxreview/RenderWorker/VFXReviewWorker/JobExecutor.cs

using System.Text.Json;
using Microsoft.Extensions.Logging;

namespace VFXReviewWorker;

/// <summary>
/// Executes one claimed job end-to-end: writes crash-recovery state, runs
/// aerender, uploads the full log, and enqueues the durable complete/fail
/// report. One job at a time per machine (§7.2).
/// </summary>
public sealed class JobExecutor
{
    private readonly WorkerOptions _options;
    private readonly ApiClient _api;
    private readonly AerenderRunner _runner;
    private readonly PreviewStage _preview;
    private readonly ValidationStage _validation;
    private readonly DeliveryStage _delivery;
    private readonly ReportSpooler _spooler;
    private readonly ILogger<JobExecutor> _log;

    public JobExecutor(WorkerOptions options, ApiClient api, AerenderRunner runner, PreviewStage preview,
        ValidationStage validation, DeliveryStage delivery, ReportSpooler spooler, ILogger<JobExecutor> log)
    {
        _options = options;
        _api = api;
        _runner = runner;
        _preview = preview;
        _validation = validation;
        _delivery = delivery;
        _spooler = spooler;
        _log = log;
    }

    public async Task RunAsync(ClaimedJob job, string machineId, ServerConfig config, CancellationToken cancelJob, CancellationToken shutdown)
    {
        if (job.Type == "PREVIEW_ONLY")
        {
            await RunPreviewAsync(job, machineId, config, cancelJob, shutdown);
            return;
        }

        if (job.Type == "DELIVERY_BUILD")
        {
            await RunDeliveryAsync(job, machineId, cancelJob, shutdown);
            return;
        }

        var manifest = job.AsRenderManifest();
        var mapper = new PathMapper(_options.PathMappings);
        var outputDirLocal = mapper.Map(manifest.OutputDir);

        WriteState(new CurrentJobState
        {
            JobId = job.Id,
            ExportId = job.ExportId,
            MachineId = machineId,
            OutputDirLocal = outputDirLocal,
            ClaimedAt = DateTimeOffset.UtcNow,
        });

        try
        {
            var result = await _runner.RunAsync(job, manifest, machineId, outputDirLocal, config, cancelJob, shutdown);
            var logKey = await _api.UploadLogAsync(job.Id, machineId, result.FullLogPath, shutdown);

            if (result.Success)
            {
                long totalBytes = 0;
                int fileCount = 0;
                var prefix = AerenderRunner.OutputPatternPrefix(manifest.OutputPattern);
                try
                {
                    foreach (var f in Directory.EnumerateFiles(outputDirLocal))
                    {
                        if (!Path.GetFileName(f).StartsWith(prefix, StringComparison.OrdinalIgnoreCase)) continue;
                        fileCount++;
                        totalBytes += new FileInfo(f).Length;
                    }
                }
                catch { /* stats are best-effort in Phase 2; validation owns this in Phase 3 */ }

                if (fileCount == 0)
                {
                    // aerender can exit 0 having rendered nothing (bad frame
                    // range, template mismatch, silently swallowed error).
                    // Until Phase 3 validation exists, an empty output dir must
                    // never become READY_FOR_QC.
                    _log.LogError("Job {JobId}: aerender exited 0 but no \"{Prefix}*\" files exist in {Dir} — reporting failure",
                        job.Id, prefix, outputDirLocal);
                    _spooler.Enqueue("fail", job.Id, new Dictionary<string, object?>
                    {
                        ["machineId"] = machineId,
                        ["stage"] = "RENDER",
                        ["exitCode"] = 0,
                        ["errorMessage"] = "aerender exited 0 but produced no output files in " + outputDirLocal,
                        ["logTail"] = result.LogTail,
                        ["logFileKey"] = logKey,
                        ["retryable"] = false,
                    });
                    return;
                }

                _log.LogInformation("Job {JobId} complete: {Frames} frames in {Sec:F0}s ({Files} files, {Bytes} bytes)",
                    job.Id, manifest.TotalFrames, result.RenderSeconds, fileCount, totalBytes);
                _spooler.Enqueue("complete-render", job.Id, new Dictionary<string, object?>
                {
                    ["machineId"] = machineId,
                    ["renderSeconds"] = (int)result.RenderSeconds,
                    ["logFileKey"] = logKey,
                    ["logTail"] = result.LogTail,
                    ["exrFileCount"] = fileCount,
                    ["exrTotalBytes"] = totalBytes,
                });

                // Validation runs here, while we still hold the job and the
                // files are local. It is queued behind complete-render, and the
                // spooler's in-order replay is what guarantees the server has
                // moved the Export to VALIDATING before this lands (§8).
                if (config.ValidationEnabled)
                {
                    await RunValidationAsync(job, manifest, machineId, outputDirLocal, config, shutdown);
                }
            }
            else if (result.ErrorMessage == "Cancelled")
            {
                // Ack the cancel so the server clears the CANCEL_JOB command
                _log.LogInformation("Job {JobId} cancelled — acknowledging", job.Id);
                _spooler.Enqueue("fail", job.Id, new Dictionary<string, object?>
                {
                    ["machineId"] = machineId,
                    ["stage"] = "RENDER",
                    ["errorMessage"] = "Cancelled by server",
                    ["logTail"] = result.LogTail,
                    ["logFileKey"] = logKey,
                    ["retryable"] = false,
                });
            }
            else
            {
                _log.LogError("Job {JobId} failed (exit {Exit}, retryable {Retryable}): {Error}",
                    job.Id, result.ExitCode, result.Retryable, result.ErrorMessage);
                _spooler.Enqueue("fail", job.Id, new Dictionary<string, object?>
                {
                    ["machineId"] = machineId,
                    ["stage"] = "RENDER",
                    ["exitCode"] = result.ExitCode,
                    ["errorMessage"] = result.ErrorMessage,
                    ["logTail"] = result.LogTail,
                    ["logFileKey"] = logKey,
                    ["retryable"] = result.Retryable,
                });
            }
        }
        catch (Exception ex)
        {
            _log.LogError(ex, "Unexpected executor error for job {JobId}", job.Id);
            _spooler.Enqueue("fail", job.Id, new Dictionary<string, object?>
            {
                ["machineId"] = machineId,
                ["stage"] = "RENDER",
                ["errorMessage"] = $"Worker exception: {ex.Message}",
                ["retryable"] = true,
            });
        }
        finally
        {
            ClearState();
        }
    }

    /// <summary>
    /// DELIVERY_BUILD job: build the dated package folder from already-QC'd
    /// exports. No AE involved — folders, links and verification only.
    /// </summary>
    private async Task RunDeliveryAsync(ClaimedJob job, string machineId, CancellationToken cancelJob, CancellationToken shutdown)
    {
        var manifest = job.AsDeliveryManifest();

        WriteState(new CurrentJobState
        {
            JobId = job.Id,
            ExportId = null,
            MachineId = machineId,
            OutputDirLocal = new PathMapper(_options.PathMappings).Map(manifest.PackageDir),
            ClaimedAt = DateTimeOffset.UtcNow,
        });

        try
        {
            var result = await _delivery.RunAsync(job, manifest, machineId, cancelJob, shutdown);

            _log.LogInformation("Delivery {Name}: {Ok} ({Verified}/{Total} shots verified)",
                manifest.PackageName, result.Ok ? "READY" : "FAILED",
                result.Items.Count(i => i.Status == "VERIFIED"), manifest.Items.Count);

            string? reportKey = null;
            if (result.ReportPath is not null)
            {
                reportKey = await _api.UploadArtifactAsync(job.Id, machineId, "report", result.ReportPath, "text/csv", shutdown);
            }

            _spooler.Enqueue("delivery", job.Id, new Dictionary<string, object?>
            {
                ["machineId"] = machineId,
                ["ok"] = result.Ok,
                ["rootPath"] = result.PackageDir,
                ["reportKey"] = reportKey,
                ["items"] = result.Items.Select(i => new Dictionary<string, object?>
                {
                    ["itemId"] = i.ItemId,
                    ["status"] = i.Status,
                    ["exrPath"] = i.ExrPath,
                    ["movPath"] = i.MovPath,
                    ["fileCount"] = i.FileCount,
                    ["totalBytes"] = i.TotalBytes,
                    ["checksum"] = i.Checksum,
                    ["linkMode"] = i.LinkMode,
                    ["errorMessage"] = i.ErrorMessage,
                }).ToList(),
                ["manifest"] = new Dictionary<string, object?>
                {
                    ["packageDir"] = result.PackageDir,
                    ["reportPath"] = result.ReportPath,
                    ["files"] = result.Files,
                },
                ["errorMessage"] = result.Ok
                    ? null
                    : string.Join("; ", result.Items.Where(i => i.Status == "FAILED").Select(i => $"{i.ShotCode}: {i.ErrorMessage}").Take(5)),
            });
        }
        catch (Exception ex)
        {
            _log.LogError(ex, "Unexpected delivery error for job {JobId}", job.Id);
            _spooler.Enqueue("delivery", job.Id, new Dictionary<string, object?>
            {
                ["machineId"] = machineId,
                ["ok"] = false,
                ["items"] = new List<object>(),
                ["errorMessage"] = $"Worker exception: {ex.Message}",
            });
        }
        finally
        {
            ClearState();
        }
    }

    /// <summary>
    /// Runs the validation engine and spools the E12 report. Deliberately
    /// non-fatal on its own errors: a crash in the checks must not lose a
    /// finished render, so it is reported as a failed validation run rather
    /// than being allowed to escape.
    /// </summary>
    private async Task RunValidationAsync(
        ClaimedJob job, RenderManifest manifest, string machineId, string outputDirLocal,
        ServerConfig config, CancellationToken shutdown)
    {
        var started = DateTimeOffset.UtcNow;
        try
        {
            // Keep the lease alive: the job stays open through validation.
            await _api.ProgressAsync(job.Id, machineId, 1.0, manifest.FrameEnd, manifest.TotalFrames, 0,
                "Validating rendered sequence…", shutdown);

            var outcome = await Task.Run(() => _validation.Run(manifest, outputDirLocal, config), shutdown);
            var seconds = (DateTimeOffset.UtcNow - started).TotalSeconds;

            var failed = outcome.Checks.Where(c => c.Status == "FAIL").Select(c => c.CheckName).ToList();
            if (failed.Count > 0)
            {
                _log.LogWarning("Job {JobId} validation FAILED in {Sec:F1}s: {Checks}", job.Id, seconds, string.Join(", ", failed));
            }
            else
            {
                _log.LogInformation("Job {JobId} validation passed in {Sec:F1}s ({Count} checks, {Files} files)",
                    job.Id, seconds, outcome.Checks.Count, outcome.FileCount);
            }

            _spooler.Enqueue("validation", job.Id, new Dictionary<string, object?>
            {
                ["machineId"] = machineId,
                ["checks"] = outcome.Checks,
                ["checksum"] = outcome.Checksum,
                ["checksumAlgo"] = "xxh64",
                ["exrFileCount"] = outcome.FileCount,
                ["exrTotalBytes"] = outcome.TotalBytes,
                ["validationSeconds"] = (int)seconds,
            });
        }
        catch (Exception ex)
        {
            _log.LogError(ex, "Validation engine error for job {JobId}", job.Id);
            _spooler.Enqueue("validation", job.Id, new Dictionary<string, object?>
            {
                ["machineId"] = machineId,
                ["checks"] = new[]
                {
                    ValidationCheck.Fail("engine_error", "validation completes", "worker exception",
                        $"The validation engine threw: {ex.Message}"),
                },
            });
        }
    }

    /// <summary>
    /// PREVIEW_ONLY job: rebuild the shot around the rendered EXRs with slate
    /// and burn-ins, render the delivery MOV + review MP4, upload the review
    /// media, and finalize (which registers the internal-only Version).
    /// </summary>
    private async Task RunPreviewAsync(ClaimedJob job, string machineId, ServerConfig config, CancellationToken cancelJob, CancellationToken shutdown)
    {
        var manifest = job.AsPreviewManifest();

        WriteState(new CurrentJobState
        {
            JobId = job.Id,
            ExportId = job.ExportId,
            MachineId = machineId,
            OutputDirLocal = new PathMapper(_options.PathMappings).Map(manifest.OutputDir),
            ClaimedAt = DateTimeOffset.UtcNow,
        });

        try
        {
            var result = await _preview.RunAsync(job, manifest, machineId, config, cancelJob, shutdown);

            if (!result.Success)
            {
                _log.LogError("Preview job {JobId} failed (retryable {Retryable}): {Error}",
                    job.Id, result.Retryable, result.ErrorMessage);
                _spooler.Enqueue("fail", job.Id, new Dictionary<string, object?>
                {
                    ["machineId"] = machineId,
                    ["stage"] = "PREVIEW",
                    ["errorMessage"] = result.ErrorMessage,
                    ["retryable"] = result.Retryable,
                });
                return; // temp AEP is deliberately kept for debugging
            }

            var previewKey = await _api.UploadArtifactAsync(
                job.Id, machineId, "preview", result.Mp4Path!, "video/mp4", shutdown);
            if (previewKey is null)
            {
                // The renders are on the SAN and fine — only the upload failed,
                // so this is worth retrying without rebuilding anything.
                _spooler.Enqueue("fail", job.Id, new Dictionary<string, object?>
                {
                    ["machineId"] = machineId,
                    ["stage"] = "PREVIEW",
                    ["errorMessage"] = "Preview MP4 upload failed",
                    ["retryable"] = true,
                });
                return;
            }

            string? thumbKey = null;
            if (result.ThumbnailPath is not null)
            {
                thumbKey = await _api.UploadArtifactAsync(
                    job.Id, machineId, "thumbnail", result.ThumbnailPath, "image/jpeg", shutdown);
            }

            _spooler.Enqueue("finalize", job.Id, new Dictionary<string, object?>
            {
                ["machineId"] = machineId,
                ["artifacts"] = new Dictionary<string, object?>
                {
                    ["deliveryMovPath"] = manifest.MovOutput,
                    ["previewMovKey"] = previewKey,
                    ["thumbnailKey"] = thumbKey,
                },
                ["renderStats"] = new Dictionary<string, object?>
                {
                    ["previewSeconds"] = (int)result.Seconds,
                },
                ["media"] = new Dictionary<string, object?>
                {
                    ["width"] = result.Build?.Width,
                    ["height"] = result.Build?.Height,
                    ["frameCount"] = result.Build?.FrameCount,
                    ["fps"] = manifest.Fps,
                    ["fileName"] = Path.GetFileName(result.Mp4Path!),
                },
            });

            // Only clean up once the outputs are safely reported.
            CleanScratch(job.Id, result.TempAep);
        }
        catch (Exception ex)
        {
            _log.LogError(ex, "Unexpected preview error for job {JobId}", job.Id);
            _spooler.Enqueue("fail", job.Id, new Dictionary<string, object?>
            {
                ["machineId"] = machineId,
                ["stage"] = "PREVIEW",
                ["errorMessage"] = $"Worker exception: {ex.Message}",
                ["retryable"] = true,
            });
        }
        finally
        {
            ClearState();
        }
    }

    private static void CleanScratch(string jobId, string? tempAep)
    {
        PreviewStage.TryDelete(Path.Combine(WorkerOptions.ScratchDir, $"job_{jobId}_context.json"));
        PreviewStage.TryDelete(Path.Combine(WorkerOptions.ScratchDir, $"job_{jobId}_result.json"));
        PreviewStage.TryDelete(Path.Combine(WorkerOptions.ScratchDir, $"job_{jobId}_run.jsx"));
        PreviewStage.TryDelete(Path.Combine(WorkerOptions.ScratchDir, $"job_{jobId}_thumb.jpg"));
        if (tempAep is not null) PreviewStage.TryDelete(tempAep);
    }

    /// <summary>
    /// Crash recovery (§7.5): on service start, if current-job.json names a
    /// job, ask the server what became of it. Still ours (CLAIMED/RUNNING) →
    /// we crashed mid-render; report a retryable fail so the server requeues
    /// (never resume a partial aerender). Reassigned/finished → discard state;
    /// the next attempt's PrepareOutputDir clears any orphaned frames.
    /// </summary>
    public async Task ReconcileAsync(string machineId, CancellationToken ct)
    {
        var state = ReadState();
        if (state is null) return;
        _log.LogWarning("Found crash-recovery state for job {JobId} — reconciling", state.JobId);
        try
        {
            var stillOurs = false;
            if (!string.IsNullOrEmpty(state.ExportId))
            {
                var export = await _api.GetExportAsync(state.ExportId!, ct);
                var job = export?.RenderJobs.FirstOrDefault(j => j.Id == state.JobId);
                stillOurs = job?.Status is "CLAIMED" or "RUNNING";
            }
            if (stillOurs)
            {
                _log.LogWarning("Server still shows job {JobId} as ours — reporting crash for requeue", state.JobId);
                _spooler.Enqueue("fail", state.JobId, new Dictionary<string, object?>
                {
                    ["machineId"] = machineId,
                    ["stage"] = "RENDER",
                    ["errorMessage"] = "Worker restarted mid-render (crash/power loss); never resuming a partial render",
                    ["retryable"] = true,
                });
            }
        }
        catch (Exception ex)
        {
            _log.LogWarning(ex, "Reconcile failed for job {JobId}; discarding local state (reaper covers the lease)", state.JobId);
        }
        ClearState();
    }

    private static void WriteState(CurrentJobState state)
    {
        Directory.CreateDirectory(WorkerOptions.DataDir);
        File.WriteAllText(WorkerOptions.StateFilePath, JsonSerializer.Serialize(state, WorkerOptions.JsonOpts));
    }

    private static CurrentJobState? ReadState()
    {
        try
        {
            if (!File.Exists(WorkerOptions.StateFilePath)) return null;
            return JsonSerializer.Deserialize<CurrentJobState>(File.ReadAllText(WorkerOptions.StateFilePath), WorkerOptions.JsonOpts);
        }
        catch
        {
            return null;
        }
    }

    private static void ClearState()
    {
        try { File.Delete(WorkerOptions.StateFilePath); } catch { }
    }
}