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 tenant's code root (either the monorepo checkout
/// or a per-tenant Forgejo clone, depending on the configured mode).
/// </summary>
public class TaskDispatcher
{
    private readonly WorkflowsDbContext _db;
    private readonly IJobQueue _jobs;
    private readonly string _defaultWorkerRoot;
    private readonly string _sharedDir;
    private readonly WorkflowSourceFactory? _sourceFactory;
    private readonly ILogger<TaskDispatcher> _logger;

    public TaskDispatcher(
        WorkflowsDbContext db,
        IJobQueue jobs,
        IConfiguration config,
        ILogger<TaskDispatcher> logger,
        WorkflowSourceFactory? sourceFactory = null)
    {
        _db = db;
        _jobs = jobs;
        _logger = logger;
        _sourceFactory = sourceFactory;
        _defaultWorkerRoot = ResolveWorkerRoot(config);
        _sharedDir = config["WorkflowSource:SharedDir"] ?? "workflows";
    }

    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 workingDir = await ResolveWorkingDirAsync(run.TenantId, workflowPath, ct);
        var fields = TaskRunMessage.ToFields(run, task, input, workingDir, 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 workingDir = await ResolveWorkingDirAsync(run.TenantId, run.Workflow.Path, ct);
        var fields = TaskRunMessage.ToFields(
            run, task, taskRun.InputJson, workingDir, 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);
    }

    /// <summary>
    /// Resolves the absolute working directory for a workflow's code files.
    /// In Forgejo-backed mode, resolves relative to the tenant's local clone.
    /// In filesystem-only mode, resolves relative to the global worker root.
    ///
    /// <para>
    /// The stored <c>Workflow.Path</c> may or may not carry the shared-dir prefix
    /// (e.g. <c>workflows/ops-heartbeat/workflow.yaml</c> vs
    /// <c>ops-heartbeat/workflow.yaml</c>) depending on how it was compiled. To be
    /// robust against both generations we prefer the direct
    /// <c>{root}/{dir}</c> resolution and fall back to
    /// <c>{root}/{sharedDir}/{dir}</c> only when the direct path does not exist but
    /// the shared-dir layout does. This prevents dispatching a job whose working
    /// directory does not exist — which is exactly what made every such task
    /// spawn-fail and cascade into the timeout/retry overload.
    /// </para>
    /// </summary>
    public string ResolveWorkingDir(string tenantId, string workflowPath)
    {
        var workerRoot = ResolveWorkerRootForTenant(tenantId);
        var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty;
        if (string.IsNullOrWhiteSpace(dir))
            return workerRoot;
        return ResolveExistingDir(workerRoot, dir);
    }

    /// <summary>
    /// Async variant used by the live dispatch path. In Forgejo-backed mode the
    /// working root is the tenant's shared local clone, whose path depends on the
    /// user's Forgejo login — resolved here (async, DB-backed) so the worker runs
    /// the exact files being edited in the source-code explorer.
    /// </summary>
    public async Task<string> ResolveWorkingDirAsync(string tenantId, string workflowPath, CancellationToken ct = default)
    {
        var workerRoot = await ResolveWorkerRootForTenantAsync(tenantId, ct);
        var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty;
        if (string.IsNullOrWhiteSpace(dir))
            return workerRoot;
        return ResolveExistingDir(workerRoot, dir);
    }

    /// <summary>
    /// Legacy overload for backward compatibility. Resolves against the default worker root.
    /// </summary>
    public string ResolveWorkingDir(string workflowPath)
    {
        var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty;
        if (string.IsNullOrWhiteSpace(dir))
            return _defaultWorkerRoot;
        return ResolveExistingDir(_defaultWorkerRoot, dir);
    }

    /// <summary>
    /// Returns the first of <c>{root}/{dir}</c> or <c>{root}/{sharedDir}/{dir}</c>
    /// that exists on disk (in that order, so the newer, already-prefixed path
    /// wins). If neither exists — or the shared dir is already part of
    /// <paramref name="dir"/> — returns the primary candidate unchanged so the
    /// caller still gets a deterministic path to surface in the error.
    /// </summary>
    private string ResolveExistingDir(string root, string dir)
    {
        var primary = Path.Combine(root, dir.Replace('/', Path.DirectorySeparatorChar));

        if (string.IsNullOrWhiteSpace(_sharedDir))
            return primary;

        // If dir is already under the shared dir (e.g. "workflows/..."), a
        // fallback would double the prefix — don't.
        var sharedPrefix = _sharedDir.Replace('/', Path.DirectorySeparatorChar);
        var dirPrefix = dir.Replace('/', Path.DirectorySeparatorChar);
        if (dirPrefix == sharedPrefix
            || dirPrefix.StartsWith(sharedPrefix + Path.DirectorySeparatorChar, StringComparison.Ordinal))
            return primary;

        var fallback = Path.Combine(root, sharedPrefix, dirPrefix.Replace('/', Path.DirectorySeparatorChar));

        // Prefer the layout that actually contains the code files.
        if (Directory.Exists(fallback) && !Directory.Exists(primary))
            return fallback;

        return primary;
    }

    private string ResolveWorkerRootForTenant(string tenantId)
    {
        // Resolve the tenant's OWN source directory — the exact directory the
        // workflow-file writer (WorkflowFilesController ->
        // WorkflowSourceFactory.ResolveTenantSourceDir) stores files in. Falling
        // back to the global/git root never matched {CopiesRoot}/{tenantId}, so
        // every task ran against a working dir that did not contain its entry
        // file (surfacing as "entry file not found: Main.cs" in the C# executor).
        if (_sourceFactory != null)
            return _sourceFactory.ResolveTenantSourceDir(tenantId);
        return _defaultWorkerRoot;
    }

    private async Task<string> ResolveWorkerRootForTenantAsync(string tenantId, CancellationToken ct)
    {
        // Same contract as the sync variant: the run's code root must be the one
        // the writer used. In Forgejo-backed mode this is the tenant's shared
        // local clone (which depends on the user's Forgejo login); otherwise it
        // is the per-tenant filesystem dir. Both are exactly where the writer
        // lands, so the entry file is always found relative to this root.
        if (_sourceFactory == null)
            return _defaultWorkerRoot;

        var login = _sourceFactory.IsForgejoBacked
            ? await _sourceFactory.ResolveForgejoLoginAsync(tenantId, ct)
            : null;
        return _sourceFactory.ResolveTenantSourceDir(tenantId, login);
    }

    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);
    }
}