w4c-workflows-api/Services/Runs/TaskDispatcher.cs

using w4c_workflows.Data;
using w4c_workflows.Models;
using w4c_workflows.Services.Execution;
using w4c_workflows.Services.Messaging;

namespace w4c_workflows.Services.Runs;

/// <summary>
/// Creates the <see cref="TaskRun"/> row for an attempt and enqueues the
/// corresponding <c>task.run</c> job on the tenant's stream. The row is
/// persisted <em>before</em> the job is enqueued so a fast worker result can
/// never race a missing TaskRun. The working directory is derived from the
/// workflow's path under the worker's code root (a shared checkout volume in
/// deployment).
/// </summary>
public class TaskDispatcher
{
    private readonly WorkflowsDbContext _db;
    private readonly IJobQueue _jobs;
    private readonly string _workerRoot;
    private readonly ILogger<TaskDispatcher> _logger;

    public TaskDispatcher(
        WorkflowsDbContext db,
        IJobQueue jobs,
        IConfiguration config,
        ILogger<TaskDispatcher> logger)
    {
        _db = db;
        _jobs = jobs;
        _logger = logger;
        _workerRoot = ResolveWorkerRoot(config);
    }

    public async Task<Guid> DispatchAsync(
        WorkflowRun run,
        WorkflowTask task,
        string workflowPath,
        string? input,
        int attempt,
        CancellationToken ct,
        bool isCompensation = false)
    {
        var taskRun = new TaskRun
        {
            Id = Guid.NewGuid(),
            RunId = run.Id,
            TaskId = task.Id,
            Attempt = attempt,
            Status = TaskRunStatus.Running,
            InputJson = input,
            RetryCount = attempt - 1,
            IsCompensation = isCompensation,
            StartedAt = DateTime.UtcNow,
        };

        _db.TaskRuns.Add(taskRun);
        await _db.SaveChangesAsync(ct);

        var fields = TaskRunMessage.ToFields(run, task, input, ResolveWorkingDir(workflowPath), attempt);
        await _jobs.EnqueueAsync(run.TenantId, fields, ct);

        _logger.LogDebug(
            "Dispatched task {TaskKey} (run {RunId}, attempt {Attempt}, compensation {IsCompensation}) for tenant {TenantId}",
            task.Key, run.Id, attempt, isCompensation, run.TenantId);

        return taskRun.Id;
    }

    /// <summary>
    /// Dead-letters a failed task (retries exhausted, or a failed compensation)
    /// by reconstructing its <c>task.run</c> fields and publishing them to the
    /// tenant DLQ. The control-plane failure path has no live jobs-stream entry
    /// left to acknowledge, so this uses a publish (not copy+ack) write.
    /// </summary>
    public async Task DeadLetterTaskAsync(
        WorkflowRun run, WorkflowTask task, TaskRun taskRun, CancellationToken ct)
    {
        var fields = TaskRunMessage.ToFields(
            run, task, taskRun.InputJson, ResolveWorkingDir(run.Workflow.Path), taskRun.Attempt);

        var dlqFields = new Dictionary<string, string>(fields)
        {
            ["dlq.error"] = taskRun.Error ?? string.Empty,
        };

        await _jobs.PublishDeadLetterAsync(run.TenantId, dlqFields, "max_retries_exceeded", ct);

        _logger.LogWarning(
            "Dead-lettered task {TaskKey} (run {RunId}, attempt {Attempt})",
            task.Key, run.Id, taskRun.Attempt);
    }

    public string ResolveWorkingDir(string workflowPath)
    {
        var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty;
        if (string.IsNullOrWhiteSpace(dir))
            return _workerRoot;
        return Path.Combine(_workerRoot, dir.Replace('/', Path.DirectorySeparatorChar));
    }

    private static string ResolveWorkerRoot(IConfiguration config)
    {
        // Explicit override wins: a dedicated worker mount (e.g. a shared checkout
        // volume in deployment) or an explicit repo root.
        var workerRoot = config["Workflows:WorkerRoot"];
        if (!string.IsNullOrWhiteSpace(workerRoot))
            return Path.GetFullPath(workerRoot);

        var repoRoot = config["SourceCode:RepoRoot"];
        if (!string.IsNullOrWhiteSpace(repoRoot))
            return Path.GetFullPath(repoRoot);

        // Dev fallback: resolve the monorepo/git root the same way the control
        // plane's WorkflowSource does, so the worker and the source reader agree
        // on where workflow code files live. Without this, a worker started from
        // the service directory would resolve files under <cwd>/workflows/...,
        // which is wrong (the code files sit at the repo root).
        var start = Directory.GetCurrentDirectory();
        var walk = Path.GetFullPath(start);
        while (true)
        {
            if (Directory.Exists(Path.Combine(walk, ".git")))
                return Path.GetFullPath(walk);
            var parent = Directory.GetParent(walk)?.FullName;
            if (string.IsNullOrEmpty(parent) || parent == walk)
                break;
            walk = parent;
        }

        return Path.GetFullPath(start);
    }
}