using System.Net;
using System.Net.Http.Headers;
using System.Net.Http.Json;
using System.Text.Json;
using Microsoft.Extensions.Logging;
namespace VFXReviewWorker;
///
/// HTTP client for the VFXReview /api/ext pipeline endpoints. Workers talk
/// only HTTP — never a database (§2.1 principle 4).
///
public sealed class ApiClient
{
private readonly HttpClient _http;
private readonly ILogger _log;
public ApiClient(WorkerOptions options, ILogger 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 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,
capabilities = new { maxConcurrentJobs = 1, tools = new { ffmpeg = o.FfmpegPath != null, oiiotool = o.OiiotoolPath != null } },
}, Json, ct);
res.EnsureSuccessStatusCode();
return (await res.Content.ReadFromJsonAsync(Json, ct))!;
}
// E7
public async Task 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(Json, ct))!;
}
// E8 — null when the queue is empty / machine outside its render window (204)
public async Task 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(Json, ct);
return body?.Job;
}
// E9 — best-effort; failures are swallowed by the caller (lease/reaper covers us)
public async Task ProgressAsync(string jobId, string machineId, double progress,
int? currentFrame, int? totalFrames, int? etaSeconds, string? logTail, CancellationToken ct)
{
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(Json, ct);
}
// 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 SendLifecycleAsync(string kind, string jobId, Dictionary payload, CancellationToken ct)
{
var path = kind switch
{
"fail" => $"api/ext/render/jobs/{jobId}/fail",
"complete-render" => $"api/ext/render/jobs/{jobId}/complete-render",
"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 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(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 UploadLogAsync(string jobId, string machineId, string logFilePath, CancellationToken ct) =>
UploadArtifactAsync(jobId, machineId, "log", logFilePath, "text/plain", ct);
// E3 — used by crash-recovery reconcile (§7.5)
public async Task 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(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;
}
}
}