vfxreview/RenderWorker/VFXReviewWorker/ApiClient.cs

using System.Net;
using System.Net.Http.Headers;
using System.Net.Http.Json;
using System.Text.Json;
using Microsoft.Extensions.Logging;

namespace VFXReviewWorker;

/// <summary>
/// HTTP client for the VFXReview /api/ext pipeline endpoints. Workers talk
/// only HTTP — never a database (§2.1 principle 4).
/// </summary>
public sealed class ApiClient
{
    private readonly HttpClient _http;
    private readonly ILogger<ApiClient> _log;

    public ApiClient(WorkerOptions options, ILogger<ApiClient> log)
    {
        _log = log;
        _http = new HttpClient
        {
            BaseAddress = new Uri(options.ServerUrl.TrimEnd('/') + "/"),
            Timeout = TimeSpan.FromSeconds(30),
        };
        _http.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", options.ApiKey);
    }

    private static readonly JsonSerializerOptions Json = WorkerOptions.JsonOpts;

    // E6
    public async Task<RegisterResponse> RegisterAsync(WorkerOptions o, CancellationToken ct)
    {
        var res = await _http.PostAsJsonAsync("api/ext/workers/register", new
        {
            name = o.MachineName,
            hostname = Dns.GetHostName(),
            workerVersion = o.WorkerVersion,
            aeVersion = o.AeVersion,
            // The server hands out BLENDER_RENDER work on the strength of the
            // blender entry, so it reports the resolved path rather than a bare
            // flag: a machine claiming Blender jobs it cannot run is worse than
            // one that never claims them.
            capabilities = new
            {
                maxConcurrentJobs = 1,
                tools = new
                {
                    ffmpeg = o.FfmpegPath != null,
                    oiiotool = o.OiiotoolPath != null,
                    blender = !string.IsNullOrWhiteSpace(o.BlenderPath) && File.Exists(o.BlenderPath) ? o.BlenderPath : null,
                },
            },
        }, Json, ct);
        res.EnsureSuccessStatusCode();
        return (await res.Content.ReadFromJsonAsync<RegisterResponse>(Json, ct))!;
    }

    // E7
    public async Task<HeartbeatResponse> HeartbeatAsync(string machineId, string? currentJobId, CancellationToken ct)
    {
        var res = await _http.PostAsJsonAsync($"api/ext/workers/{machineId}/heartbeat", new
        {
            cpuPercent = (double?)null,
            memPercent = (double?)null,
            diskFreeGb = GetDiskFreeGb(),
            currentJobId,
        }, Json, ct);
        res.EnsureSuccessStatusCode();
        return (await res.Content.ReadFromJsonAsync<HeartbeatResponse>(Json, ct))!;
    }

    // E8 — null when the queue is empty / machine outside its render window (204)
    public async Task<ClaimedJob?> ClaimAsync(string machineId, string[] types, CancellationToken ct)
    {
        var res = await _http.PostAsJsonAsync("api/ext/render/jobs/claim", new
        {
            machineId,
            types,
        }, Json, ct);
        if (res.StatusCode == HttpStatusCode.NoContent) return null;
        res.EnsureSuccessStatusCode();
        var body = await res.Content.ReadFromJsonAsync<ClaimResponse>(Json, ct);
        return body?.Job;
    }

    // E9 — best-effort by contract: a progress report is a nicety, and losing
    // one must never abort work in flight. Transport failures return null here
    // rather than throwing at the call site; the lease and the reaper are what
    // actually protect the job.
    public async Task<ProgressResponse?> ProgressAsync(string jobId, string machineId, double progress,
        int? currentFrame, int? totalFrames, int? etaSeconds, string? logTail, CancellationToken ct)
    {
        try
        {
            using var req = new HttpRequestMessage(HttpMethod.Patch, $"api/ext/render/jobs/{jobId}/progress")
            {
                Content = JsonContent.Create(new { machineId, progress, currentFrame, totalFrames, etaSeconds, logTail }, options: Json),
            };
            var res = await _http.SendAsync(req, ct);
            if (!res.IsSuccessStatusCode) return null;
            return await res.Content.ReadFromJsonAsync<ProgressResponse>(Json, ct);
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            throw; // a real shutdown still stops us
        }
        catch (Exception ex)
        {
            _log.LogDebug(ex, "Progress report failed for job {JobId} (non-fatal)", jobId);
            return null;
        }
    }

    // E10 / E11 — lifecycle reports, sent via the spooler for durability.
    // Returns true when the server accepted (2xx) OR permanently rejected (409
    // contradictory / 404 gone) — both mean "stop retrying".
    public async Task<bool> SendLifecycleAsync(string kind, string jobId, Dictionary<string, object?> payload, CancellationToken ct)
    {
        var path = kind switch
        {
            "fail" => $"api/ext/render/jobs/{jobId}/fail",
            "complete-render" => $"api/ext/render/jobs/{jobId}/complete-render",
            "validation" => $"api/ext/render/jobs/{jobId}/validation",
            "delivery" => $"api/ext/render/jobs/{jobId}/delivery",
            "finalize" => $"api/ext/render/jobs/{jobId}/finalize",
            _ => throw new ArgumentOutOfRangeException(nameof(kind), kind, "unknown lifecycle report kind"),
        };
        var res = await _http.PostAsJsonAsync(path, payload, Json, ct);
        if (res.IsSuccessStatusCode) return true;
        if (res.StatusCode is HttpStatusCode.Conflict or HttpStatusCode.NotFound)
        {
            _log.LogWarning("Server permanently rejected {Kind} for job {JobId}: {Status} {Body}",
                kind, jobId, (int)res.StatusCode, await res.Content.ReadAsStringAsync(ct));
            return true; // don't retry contradictory/vanished reports
        }
        _log.LogWarning("Lifecycle report {Kind} for {JobId} failed with {Status}; will retry", kind, jobId, (int)res.StatusCode);
        return false;
    }

    // E20 + PUT upload. Returns the storage key, or null when the upload failed.
    public async Task<string?> UploadArtifactAsync(
        string jobId, string machineId, string kind, string filePath, string contentType, CancellationToken ct)
    {
        try
        {
            if (!File.Exists(filePath))
            {
                _log.LogWarning("Artifact {Kind} not found at {Path}", kind, filePath);
                return null;
            }

            var presignRes = await _http.PostAsJsonAsync($"api/ext/render/jobs/{jobId}/artifact-presign", new
            {
                machineId,
                kind,
                fileName = Path.GetFileName(filePath),
                contentType,
            }, Json, ct);
            if (!presignRes.IsSuccessStatusCode)
            {
                _log.LogWarning("Presign for {Kind} failed: {Status}", kind, (int)presignRes.StatusCode);
                return null;
            }
            var presign = await presignRes.Content.ReadFromJsonAsync<PresignResponse>(Json, ct);
            if (presign is null) return null;

            // Media files can be large — generous timeout, separate client so the
            // presigned URL is not sent with our Authorization header.
            using var upload = new HttpClient { Timeout = TimeSpan.FromMinutes(30) };
            await using var stream = File.OpenRead(filePath);
            var content = new StreamContent(stream);
            content.Headers.ContentType = new MediaTypeHeaderValue(contentType);
            var putRes = await upload.PutAsync(presign.PresignedUrl, content, ct);
            if (!putRes.IsSuccessStatusCode)
            {
                _log.LogWarning("Upload of {Kind} failed: {Status}", kind, (int)putRes.StatusCode);
                return null;
            }
            return presign.Key;
        }
        catch (Exception ex)
        {
            _log.LogWarning(ex, "Artifact upload ({Kind}) failed for job {JobId}", kind, jobId);
            return null;
        }
    }

    public Task<string?> UploadLogAsync(string jobId, string machineId, string logFilePath, CancellationToken ct) =>
        UploadArtifactAsync(jobId, machineId, "log", logFilePath, "text/plain", ct);

    /// <summary>
    /// Reports the deployed worker version so the web UI can show what is
    /// running where. Fire-and-forget — never worth failing a startup over.
    /// </summary>
    public async Task ReportVersionAsync(WorkerOptions o, CancellationToken ct)
    {
        try
        {
            await _http.PostAsJsonAsync("api/ext/tools/ping", new
            {
                tool = "RENDER_WORKER",
                machineName = o.MachineName,
                version = o.WorkerVersion,
                hostUser = Environment.UserName,
                hostApp = $"{Environment.OSVersion.VersionString} ({Dns.GetHostName()})",
                details = new
                {
                    aeVersion = o.AeVersion,
                    aerenderPath = o.AerenderPath,
                    ffmpeg = o.FfmpegPath is not null,
                    plateSearchRoots = o.PlateSearchRoots,
                },
            }, Json, ct);
        }
        catch (Exception ex)
        {
            _log.LogDebug(ex, "Version ping failed (non-fatal)");
        }
    }

    // E3 — used by crash-recovery reconcile (§7.5)
    public async Task<ExportDetail?> GetExportAsync(string exportId, CancellationToken ct)
    {
        var res = await _http.GetAsync($"api/ext/exports/{exportId}", ct);
        if (!res.IsSuccessStatusCode) return null;
        var body = await res.Content.ReadFromJsonAsync<ExportDetailResponse>(Json, ct);
        return body?.Export;
    }

    private static double? GetDiskFreeGb()
    {
        try
        {
            var root = Path.GetPathRoot(Environment.SystemDirectory);
            if (root is null) return null;
            return Math.Round(new DriveInfo(root).AvailableFreeSpace / 1_073_741_824.0, 1);
        }
        catch
        {
            return null;
        }
    }
}