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;
using w4c_workflows.Services.Nodes;

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 WorkflowRepoStore? _repoStore;
    private readonly ILogger<TaskDispatcher> _logger;

    public TaskDispatcher(
        WorkflowsDbContext db,
        IJobQueue jobs,
        IConfiguration config,
        ILogger<TaskDispatcher> logger,
        WorkflowSourceFactory? sourceFactory = null,
        WorkflowRepoStore? repoStore = null)
    {
        _db = db;
        _jobs = jobs;
        _logger = logger;
        _sourceFactory = sourceFactory;
        _repoStore = repoStore;
        _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>
    /// Enqueues a <c>graph.run</c> job for a node-mode run. Unlike
    /// <see cref="DispatchAsync"/> this writes no <see cref="TaskRun"/> row: the
    /// node kernel records one history row per executed node when the worker runs
    /// the graph. One job covers the whole graph, so the run is executed on the
    /// worker/queue path with uniform leases, timeouts and scale-out (S2).
    /// </summary>
    public async Task DispatchGraphAsync(
        WorkflowRun run,
        string? workingDir,
        CancellationToken ct,
        int attempt = 1)
    {
        var fields = GraphRunMessage.ToFields(run, workingDir, attempt);
        await _jobs.EnqueueAsync(run.TenantId, fields, ct);

        _logger.LogDebug(
            "Dispatched node-graph job (run {RunId}, attempt {Attempt}) for tenant {TenantId}",
            run.Id, attempt, run.TenantId);
    }

    /// <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 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>
    /// 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 async Task<string> ResolveWorkerRootForTenantAsync(string tenantId, CancellationToken ct)
    {
        // The run's code root must be the one the workflow-file 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;

        // Use the tenant's SELECTED workflow repo (the clone the file-writer and
        // the source explorer land on), falling back to the configured default
        // (repoName=null) when the store is unavailable. Before this, the worker
        // always resolved the DEFAULT repo, so a tenant that had switched to a
        // different repo got a working directory that did not contain its entry
        // file (surfacing as "entry file not found: Main.cs" in the C# executor).
        var repoName = await ResolveRepoNameAsync(tenantId, ct);

        if (_sourceFactory.IsForgejoBacked)
        {
            var login = await _sourceFactory.ResolveForgejoLoginAsync(tenantId, ct);
            if (!string.IsNullOrWhiteSpace(login))
            {
                // Materialize the tenant's repo clone if it is not already present on
                // THIS instance before resolving the working directory. The writer may
                // have cloned the repo on a different host/instance; without this the
                // worker resolves a working directory whose code files do not exist
                // ("entry file not found: Main.cs"). EnsureClone is a no-op once the
                // clone exists, so repeated dispatches stay cheap.
                await EnsureCloneAsync(tenantId, login, repoName, ct);
                return _sourceFactory.ResolveTenantSourceDir(tenantId, login, repoName);
            }
        }

        return _sourceFactory.ResolveTenantSourceDir(tenantId, null, repoName);
    }

    /// <summary>
    /// Best-effort clone/pull of the tenant's workflow repo so the working directory
    /// the worker will execute in actually contains the code files. Failures are
    /// logged and swallowed: the resolved path is still returned, so a genuinely
    /// missing repo surfaces its normal "entry file not found" at execution time
    /// rather than silently dispatching to a directory we cannot reason about.
    /// </summary>
    private async Task EnsureCloneAsync(string tenantId, string login, string? repoName, CancellationToken ct)
    {
        var forgejo = _sourceFactory?.Forgejo;
        if (forgejo == null)
            return;

        try
        {
            await forgejo.EnsureCloneAsync(tenantId, login, repoName, ct);
            // No per-dispatch PullAsync: a full git pull is too slow to run inside
            // the control-plane dispatch loop on every tick. EnsureClone guarantees
            // the working directory exists; freshness is handled by the compile/edit
            // paths that already pull the repo on write/sync.
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "Could not ensure workflow repo clone for tenant {TenantId} (login {Login}); dispatching with the resolved path as-is",
                tenantId, login);
        }
    }

    /// <summary>
    /// Resolves the tenant's currently-selected workflow repo name (as persisted in
    /// <see cref="WorkflowRepoStore"/>). Returns <c>null</c> when the store is not
    /// configured or the lookup fails, letting the caller fall back to the factory's
    /// configured default repo.
    /// </summary>
    private async Task<string?> ResolveRepoNameAsync(string tenantId, CancellationToken ct)
    {
        if (_repoStore == null)
            return null;

        try
        {
            return await _repoStore.GetNameAsync(tenantId, ct);
        }
        catch (Exception ex)
        {
            _logger.LogDebug(ex,
                "Could not resolve selected workflow repo for tenant {TenantId}; using default", tenantId);
            return null;
        }
    }

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