61 lines
2.3 KiB
C#
61 lines
2.3 KiB
C#
using System.Text.Json;
|
|
|
|
namespace w4c_workflows.Services.Execution;
|
|
|
|
/// <summary>
|
|
/// Wire contract for a <c>task.run</c> job pulled off the tenant's jobs stream.
|
|
/// The control plane (step 9) enqueues these; the worker reads, executes and
|
|
/// acks them. Field names are flat Redis Stream entries so the contract is
|
|
/// transport-agnostic (Redis Streams now, RabbitMQ later).
|
|
/// </summary>
|
|
public sealed record TaskInvocation(
|
|
string Type,
|
|
string RunId,
|
|
string TaskId,
|
|
string TaskKey,
|
|
string Language,
|
|
string EntryFile,
|
|
string? EntryFunction,
|
|
IReadOnlyDictionary<string, string> Env,
|
|
string? Input,
|
|
string? WorkingDir,
|
|
string TenantId,
|
|
int Attempt,
|
|
string? Server = null)
|
|
{
|
|
public const string TypeValue = "task.run";
|
|
|
|
private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web);
|
|
|
|
/// <summary>Parses a stream message into an invocation, throwing if a required field is missing.</summary>
|
|
public static TaskInvocation FromFields(IReadOnlyDictionary<string, string> fields)
|
|
{
|
|
static string Get(IReadOnlyDictionary<string, string> fields, string key) =>
|
|
fields.TryGetValue(key, out var value) && !string.IsNullOrEmpty(value)
|
|
? value
|
|
: throw new InvalidOperationException($"missing required job field '{key}'");
|
|
|
|
static int GetInt(IReadOnlyDictionary<string, string> fields, string key, int fallback) =>
|
|
fields.TryGetValue(key, out var raw) && int.TryParse(raw, out var n) ? n : fallback;
|
|
|
|
var env = fields.TryGetValue("env", out var envJson) && !string.IsNullOrWhiteSpace(envJson)
|
|
? JsonSerializer.Deserialize<Dictionary<string, string>>(envJson, Json) ?? new()
|
|
: new Dictionary<string, string>();
|
|
|
|
return new TaskInvocation(
|
|
Get(fields, "type"),
|
|
Get(fields, "run_id"),
|
|
Get(fields, "task_id"),
|
|
Get(fields, "task_key"),
|
|
Get(fields, "language"),
|
|
Get(fields, "entry_file"),
|
|
fields.GetValueOrDefault("entry_function"),
|
|
env,
|
|
fields.GetValueOrDefault("input"),
|
|
fields.GetValueOrDefault("working_dir"),
|
|
Get(fields, "tenant_id"),
|
|
GetInt(fields, "attempt", 1),
|
|
fields.GetValueOrDefault("server"));
|
|
}
|
|
}
|