using w4c_workflows.Data; using w4c_workflows.Models; using w4c_workflows.Services.Execution; using w4c_workflows.Services.Messaging; namespace w4c_workflows.Services.Runs; /// /// Creates the row for an attempt and enqueues the /// corresponding task.run job on the tenant's stream. The row is /// persisted before 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). /// 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 _logger; public TaskDispatcher( WorkflowsDbContext db, IJobQueue jobs, IConfiguration config, ILogger logger, WorkflowSourceFactory? sourceFactory = null) { _db = db; _jobs = jobs; _logger = logger; _sourceFactory = sourceFactory; _defaultWorkerRoot = ResolveWorkerRoot(config); _sharedDir = config["WorkflowSource:SharedDir"] ?? "workflows"; } public async Task 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 = ResolveWorkingDir(run.TenantId, workflowPath); 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; } /// /// Dead-letters a failed task (retries exhausted, or a failed compensation) /// by reconstructing its task.run 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. /// public async Task DeadLetterTaskAsync( WorkflowRun run, WorkflowTask task, TaskRun taskRun, CancellationToken ct) { var workingDir = ResolveWorkingDir(run.TenantId, run.Workflow.Path); var fields = TaskRunMessage.ToFields( run, task, taskRun.InputJson, workingDir, taskRun.Attempt); var dlqFields = new Dictionary(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); } /// /// 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. /// /// /// The stored Workflow.Path may or may not carry the shared-dir prefix /// (e.g. workflows/ops-heartbeat/workflow.yaml vs /// ops-heartbeat/workflow.yaml) depending on how it was compiled. To be /// robust against both generations we prefer the direct /// {root}/{dir} resolution and fall back to /// {root}/{sharedDir}/{dir} 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. /// /// 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); } /// /// Legacy overload for backward compatibility. Resolves against the default worker root. /// public string ResolveWorkingDir(string workflowPath) { var dir = Path.GetDirectoryName(workflowPath) ?? string.Empty; if (string.IsNullOrWhiteSpace(dir)) return _defaultWorkerRoot; return ResolveExistingDir(_defaultWorkerRoot, dir); } /// /// Returns the first of {root}/{dir} or {root}/{sharedDir}/{dir} /// 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 /// — returns the primary candidate unchanged so the /// caller still gets a deterministic path to surface in the error. /// 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) { // In Forgejo-backed mode, resolve from the tenant's local clone. if (_sourceFactory?.IsForgejoBacked == true && _sourceFactory.Forgejo != null) { return _sourceFactory.Forgejo.TenantCloneDir(tenantId); } // Filesystem-only mode: use the global worker root. return _defaultWorkerRoot; } 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 /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); } }