using System.Globalization;
using System.Text;
using System.Text.Json;
using Microsoft.AspNetCore.Http.Features;
using Microsoft.Extensions.Options;
using OpenNest.Data;
namespace OpenNest.Server;
///
/// /api/nest-jobs: the server nesting queue. Desktops submit a package, poll its state,
/// read the result and cancel; the worker (OpenNest.Server.Worker) claims a job, reads
/// its package, heartbeats and reports. The server never opens a package or result: the worker
/// solves the job and the desktop validates the result against its own drawings. Progress
/// and plans the worker stores reach desktops through each job's event stream.
/// See docs/server-nesting.md.
///
internal static class NestJobRoutes
{
private static readonly JsonSerializerOptions Options = new(JsonSerializerDefaults.Web);
/// How often an idle event stream sends a keep-alive comment; tests shorten it.
internal static TimeSpan KeepAlive { get; set; } = NestJobRules.StreamKeepAlive;
public static void MapNestJobs(this IEndpointRouteBuilder app)
{
app.MapPost("/api/nest-jobs", async (HttpContext http, NestDatabase db, TimeProvider clock) =>
{
if (http.Features.Get() is { IsReadOnly: false } limit)
limit.MaxRequestBodySize = NestJobRules.MaxPackageBytes + 1024 * 1024;
if (!http.Request.HasFormContentType)
return Results.BadRequest("Expected multipart/form-data with metadata and file parts.");
IFormCollection form;
try
{
form = await http.Request.ReadFormAsync(new FormOptions { MultipartBodyLengthLimit = NestJobRules.MaxPackageBytes + 1024 * 1024 });
}
catch (Exception ex) when (ex is InvalidDataException || (ex is IOException && ex is not BadHttpRequestException))
{
return Results.BadRequest("Malformed multipart body.");
}
NestJobSubmission? submission;
try { submission = JsonSerializer.Deserialize(form["metadata"].FirstOrDefault() ?? "", Options); }
catch (JsonException) { return Results.BadRequest("Invalid metadata JSON."); }
if (submission is null || string.IsNullOrWhiteSpace(submission.EngineName))
return Results.BadRequest("The job must name an engine.");
if (new[] { submission.Title, submission.EngineName, submission.SubmittedBy }.Any(t => (t?.Length ?? 0) > NestJobRules.MaxTextLength))
return Results.BadRequest("A job title, engine or submitter is too long.");
submission.Title ??= "";
submission.SubmittedBy ??= "";
var file = form.Files["file"];
if (file is null || file.Length == 0)
return Results.BadRequest("Missing or empty 'file' part.");
if (file.Length > NestJobRules.MaxPackageBytes)
return Results.BadRequest("The nest job package is too large.");
using var stream = file.OpenReadStream();
using var buffer = new MemoryStream((int)file.Length);
await stream.CopyToAsync(buffer);
return Results.Ok(db.InsertJob(Guid.NewGuid(), submission, buffer.ToArray(), clock.GetUtcNow()));
});
// Job lists (My Jobs). The ETag lets a polling desktop skip an unchanged list.
app.MapGet("/api/nest-jobs", (HttpContext http, NestDatabase db, TimeProvider clock, IOptions json) =>
{
var query = http.Request.Query;
var filter = NestJobListFilter.All;
if (query["filter"].FirstOrDefault() is { Length: > 0 } text
&& (!Enum.TryParse(text, ignoreCase: true, out filter) || !Enum.IsDefined(filter)))
return Results.BadRequest("filter must be all, active, finished or unreviewed.");
var limit = 200;
if (query["limit"].FirstOrDefault() is { Length: > 0 } limitText
&& (!int.TryParse(limitText, NumberStyles.None, CultureInfo.InvariantCulture, out limit) || limit < 1))
return Results.BadRequest("limit must be a positive integer.");
var submittedBy = query["submittedBy"].FirstOrDefault();
if (submittedBy?.Length > NestJobRules.MaxTextLength)
return Results.BadRequest("submittedBy is too long.");
var body = JsonSerializer.SerializeToUtf8Bytes(db.ListJobs(filter, submittedBy, limit, clock.GetUtcNow()), json.Value.SerializerOptions);
var etag = $"\"{Convert.ToHexString(System.Security.Cryptography.SHA256.HashData(body), 0, 16)}\"";
http.Response.Headers.ETag = etag;
if (http.Request.Headers.IfNoneMatch.Contains(etag))
return Results.StatusCode(StatusCodes.Status304NotModified);
return Results.Bytes(body, "application/json");
});
app.MapGet("/api/nest-jobs/{id:guid}", (Guid id, NestDatabase db) =>
db.GetJob(id) is { } job ? Results.Ok(job) : Results.NotFound());
app.MapGet("/api/nest-jobs/{id:guid}/result", (Guid id, NestDatabase db) =>
{
if (db.GetJobResult(id) is { } result)
return Results.Text(result, "application/json", Encoding.UTF8);
return db.GetJob(id) is { } job
? Results.Conflict(job.Review == NestJobReview.None
? $"The job has no result: it is {job.State}."
: $"The result was {job.Review.ToString().ToLowerInvariant()} already.")
: Results.NotFound();
});
// Keep and Discard each happen once per result: If-Match names the job revision the
// desktop reviewed, and a job kept, discarded or changed meanwhile answers 409 with the job.
app.MapPost("/api/nest-jobs/{id:guid}/keep", (Guid id, HttpRequest request, NestDatabase db, TimeProvider clock) =>
Review(id, NestJobReview.Kept, request, db, clock));
app.MapPost("/api/nest-jobs/{id:guid}/discard", (Guid id, HttpRequest request, NestDatabase db, TimeProvider clock) =>
Review(id, NestJobReview.Discarded, request, db, clock));
// Removes a finished job, package and result included; an active job must be cancelled first.
app.MapDelete("/api/nest-jobs/{id:guid}", (Guid id, HttpRequest request, NestDatabase db) =>
{
var precondition = IfMatch.Read(request, out var expected);
if (precondition is not null) return precondition;
return db.DeleteJob(id, expected) switch
{
null => Results.NotFound(),
{ Removed: true } => Results.NoContent(),
{ Job: var job } => Results.Conflict(job),
};
});
app.MapGet("/api/nest-jobs/{id:guid}/plan", (Guid id, HttpResponse response, NestDatabase db) =>
{
if (db.GetJobPlan(id) is { } plan)
{
response.Headers[RemoteNestJobs.PlanRevisionHeader] = plan.Revision.ToString(CultureInfo.InvariantCulture);
return Results.Text(plan.Plan, "application/json", Encoding.UTF8);
}
return db.GetJob(id) is null ? Results.NotFound() : Results.NoContent();
});
app.MapGet("/api/nest-jobs/{id:guid}/events", StreamEvents);
app.MapPost("/api/nest-jobs/{id:guid}/cancel", (Guid id, NestDatabase db, TimeProvider clock) =>
db.CancelJob(id, clock.GetUtcNow()) is { } job ? Results.Ok(job) : Results.NotFound());
// Worker side.
app.MapPost("/api/nest-jobs/claim", async (HttpRequest request, NestDatabase db, TimeProvider clock) =>
{
var (message, error) = await ReadWorkerMessage(request);
if (error is not null) return error;
return db.ClaimJob(message!.WorkerId, clock.GetUtcNow()) is { } job ? Results.Ok(job) : Results.NoContent();
});
app.MapGet("/api/nest-jobs/{id:guid}/package", (Guid id, NestDatabase db) =>
db.GetJobPackage(id) is { } package
? Results.File(package, "application/zip", $"{id}.zip")
: Results.NotFound());
app.MapPost("/api/nest-jobs/{id:guid}/heartbeat", async (Guid id, HttpRequest request, NestDatabase db, TimeProvider clock) =>
{
var (message, error) = await ReadWorkerMessage(request);
if (error is not null) return error;
return db.HeartbeatJob(id, message!.WorkerId, clock.GetUtcNow()) is { } beat
? Results.Ok(beat)
: Results.Conflict("This worker no longer holds the job.");
});
app.MapPut("/api/nest-jobs/{id:guid}/result", async (Guid id, string? workerId, HttpContext http, NestDatabase db, TimeProvider clock) =>
{
if (string.IsNullOrWhiteSpace(workerId))
return Results.BadRequest("workerId is required.");
if (!TryReadSummary(http.Request.Query, out var summary))
return Results.BadRequest("parts, unplaced and sheets must be non-negative integers and utilization a number from 0 to 1.");
var (result, error) = await ReadJsonBody(http, NestJobRules.MaxResultBytes, "nest result");
if (error is not null) return error;
return db.CompleteJob(id, workerId, result!, clock.GetUtcNow(), summary)
? Results.Ok(db.GetJob(id))
: Results.Conflict("This worker no longer holds the job.");
});
// Progress and plans renew the lease like a heartbeat and answer as one.
app.MapPut("/api/nest-jobs/{id:guid}/progress", async (Guid id, string? workerId, HttpContext http, NestDatabase db, TimeProvider clock) =>
{
if (string.IsNullOrWhiteSpace(workerId))
return Results.BadRequest("workerId is required.");
var (progress, error) = await ReadJsonBody(http, NestJobRules.MaxProgressBytes, "progress report");
if (error is not null) return error;
return db.ReportJobProgress(id, workerId, progress!, clock.GetUtcNow()) is { } beat
? Results.Ok(beat)
: Results.Conflict("This worker no longer holds the job.");
});
app.MapPut("/api/nest-jobs/{id:guid}/plan", async (Guid id, string? workerId, HttpContext http, NestDatabase db, TimeProvider clock) =>
{
if (string.IsNullOrWhiteSpace(workerId))
return Results.BadRequest("workerId is required.");
var (plan, error) = await ReadJsonBody(http, NestJobRules.MaxPlanBytes, "nest plan");
if (error is not null) return error;
return db.ReportJobPlan(id, workerId, plan!, clock.GetUtcNow()) is { } beat
? Results.Ok(beat)
: Results.Conflict("This worker no longer holds the job.");
});
app.MapPost("/api/nest-jobs/{id:guid}/fail", async (Guid id, HttpRequest request, NestDatabase db, TimeProvider clock) =>
{
var (message, error) = await ReadWorkerMessage(request);
if (error is not null) return error;
return db.FailJob(id, message!.WorkerId, message.Error, clock.GetUtcNow())
? Results.Ok(db.GetJob(id))
: Results.Conflict("This worker no longer holds the job.");
});
app.MapPost("/api/nest-jobs/{id:guid}/release", async (Guid id, HttpRequest request, NestDatabase db, TimeProvider clock) =>
{
var (message, error) = await ReadWorkerMessage(request);
if (error is not null) return error;
return db.ReleaseJob(id, message!.WorkerId, clock.GetUtcNow())
? Results.Ok(db.GetJob(id))
: Results.Conflict("This worker no longer holds the job.");
});
app.MapPost("/api/nest-jobs/{id:guid}/cancelled", async (Guid id, HttpRequest request, NestDatabase db, TimeProvider clock) =>
{
var (message, error) = await ReadWorkerMessage(request);
if (error is not null) return error;
return db.ConfirmJobCancelled(id, message!.WorkerId, clock.GetUtcNow())
? Results.Ok(db.GetJob(id))
: Results.Conflict("This worker no longer holds the job.");
});
}
///
/// text/event-stream of one job until it is final. Each change sends event: job
/// (the record, id its revision), preceded by event: plan (id the plan's
/// revision) when a newer best plan was stored. A reconnect's Last-Event-ID skips what
/// it already has; a final job is always sent, then the stream ends. event: gone ends
/// it when the job is removed. Idle streams send a comment every .
///
private static async Task StreamEvents(Guid id, HttpContext http, NestDatabase db, IOptions json)
{
var response = http.Response;
if (db.GetJob(id) is null)
{
response.StatusCode = StatusCodes.Status404NotFound;
return;
}
var seen = long.TryParse(http.Request.Headers["Last-Event-ID"].FirstOrDefault(), NumberStyles.None,
CultureInfo.InvariantCulture, out var last) ? last : 0;
var sentPlan = seen;
var sentQueue = -1;
http.Features.Get()?.DisableBuffering();
response.ContentType = "text/event-stream";
response.Headers.CacheControl = "no-cache";
response.Headers["X-Accel-Buffering"] = "no";
var token = http.RequestAborted;
try
{
await response.Body.FlushAsync(token);
while (true)
{
var changed = db.JobsChanged;
var job = db.GetJob(id);
if (job is null)
{
await WriteEvent(response, "gone", null, "{}", token);
return;
}
if (job.PlanRevision > sentPlan && db.GetJobPlan(id) is { } plan && plan.Revision > sentPlan)
{
await WriteEvent(response, "plan", plan.Revision, plan.Plan, token);
sentPlan = plan.Revision;
}
if (job.Revision > seen || job.QueuePosition != sentQueue && sentQueue >= 0 || job.IsFinished)
{
await WriteEvent(response, "job", job.Revision, JsonSerializer.Serialize(job, json.Value.SerializerOptions), token);
seen = job.Revision;
}
sentQueue = job.QueuePosition;
if (job.IsFinished)
return;
// Changes in this process wake the stream at once; the keep-alive also re-reads.
if (await Task.WhenAny(changed, Task.Delay(KeepAlive, token)) != changed)
{
await response.WriteAsync(": keep-alive\n\n", token);
await response.Body.FlushAsync(token);
}
}
}
catch (OperationCanceledException) when (token.IsCancellationRequested)
{
// The desktop went away.
}
}
private static async Task Review(Guid id, NestJobReview review, HttpRequest request, NestDatabase db, TimeProvider clock)
{
var precondition = IfMatch.Read(request, out var expected);
if (precondition is not null) return precondition;
var reviewedBy = "";
if (request.HasJsonContentType())
{
var (json, error) = await DrawingSetRoutes.ReadJson(request, "review");
if (error is not null) return error;
try { reviewedBy = JsonSerializer.Deserialize(json!, Options)?.ReviewedBy ?? ""; }
catch (JsonException) { return Results.BadRequest("Invalid review JSON."); }
if (reviewedBy.Length > NestJobRules.MaxTextLength)
return Results.BadRequest("reviewedBy is too long.");
}
return db.ReviewJob(id, review, reviewedBy, expected, clock.GetUtcNow()) switch
{
null => Results.NotFound(),
{ Changed: true, Job: var job } => Results.Ok(job),
{ Job: var job } => Results.Conflict(job),
};
}
// The worker's summary of a result travels as query values beside it; none is fine (older workers).
private static bool TryReadSummary(IQueryCollection query, out NestJobSummary? summary)
{
summary = null;
var parts = query["parts"].FirstOrDefault();
if (string.IsNullOrEmpty(parts))
return true;
if (!int.TryParse(parts, NumberStyles.None, CultureInfo.InvariantCulture, out var placed)
|| !int.TryParse(query["unplaced"].FirstOrDefault() ?? "0", NumberStyles.None, CultureInfo.InvariantCulture, out var unplaced)
|| !int.TryParse(query["sheets"].FirstOrDefault() ?? "0", NumberStyles.None, CultureInfo.InvariantCulture, out var sheets)
|| !double.TryParse(query["utilization"].FirstOrDefault() ?? "0", NumberStyles.Float, CultureInfo.InvariantCulture, out var utilization)
|| !(utilization >= 0 && utilization <= 1))
return false;
summary = new NestJobSummary { PartsPlaced = placed, PartsUnplaced = unplaced, Sheets = sheets, Utilization = utilization };
return true;
}
private static async Task WriteEvent(HttpResponse response, string type, long? id, string data, CancellationToken token)
{
var text = new StringBuilder();
text.Append("event: ").Append(type).Append('\n');
if (id is { } value)
text.Append("id: ").Append(value.ToString(CultureInfo.InvariantCulture)).Append('\n');
foreach (var line in data.Split('\n'))
text.Append("data: ").Append(line.TrimEnd('\r')).Append('\n');
text.Append('\n');
await response.WriteAsync(text.ToString(), token);
await response.Body.FlushAsync(token);
}
private static async Task<(string? Json, IResult? Error)> ReadJsonBody(HttpContext http, long maxBytes, string what)
{
if (http.Features.Get() is { IsReadOnly: false } limit)
limit.MaxRequestBodySize = maxBytes + 1;
if (!http.Request.HasJsonContentType())
return (null, Results.BadRequest($"Expected an application/json {what}."));
using var body = new MemoryStream();
var chunk = new byte[81920];
int read;
while ((read = await http.Request.Body.ReadAsync(chunk, http.RequestAborted)) > 0)
{
if (body.Length + read > maxBytes)
return (null, Results.BadRequest($"The {what} is too large."));
body.Write(chunk, 0, read);
}
return (Encoding.UTF8.GetString(body.GetBuffer(), 0, (int)body.Length), null);
}
private static async Task<(NestJobWorkerMessage? Message, IResult? Error)> ReadWorkerMessage(HttpRequest request)
{
var (json, error) = await DrawingSetRoutes.ReadJson(request, "worker message");
if (error is not null) return (null, error);
NestJobWorkerMessage? message;
try { message = JsonSerializer.Deserialize(json!, Options); }
catch (JsonException) { return (null, Results.BadRequest("Invalid worker message JSON.")); }
if (message is null || string.IsNullOrWhiteSpace(message.WorkerId) || message.WorkerId.Length > NestJobRules.MaxTextLength)
return (null, Results.BadRequest("workerId is required."));
return (message, null);
}
}