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 WorkflowRepoStore? _repoStore;
private readonly ILogger _logger;
public TaskDispatcher(
WorkflowsDbContext db,
IJobQueue jobs,
IConfiguration config,
ILogger 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 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;
}
///
/// 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 = await ResolveWorkingDirAsync(run.TenantId, run.Workflow.Path, ct);
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);
}
///
/// 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.
///
public async Task 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);
}
///
/// 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)
{
// 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 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;
// 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);
}
///
/// 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.
///
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);
}
}
///
/// Resolves the tenant's currently-selected workflow repo name (as persisted in
/// ). Returns null when the store is not
/// configured or the lookup fails, letting the caller fall back to the factory's
/// configured default repo.
///
private async Task 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 /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);
}
}