stabilization

This commit is contained in:
Vitali sharp8n 2026-09-02 01:12:21 +03:00
parent 6a3f702e09
commit 3e31e4a8da
7 changed files with 930 additions and 14 deletions

View file

@ -0,0 +1,278 @@
using System.Diagnostics;
using Microsoft.AspNetCore.Mvc;
using w4c_workflows.Filters;
using w4c_workflows.Services;
namespace w4c_workflows.Controllers;
/// <summary>
/// Workflow file management endpoints. Provides CRUD operations for the
/// workflow YAML definitions and their sibling code files (shell scripts,
/// JS, Python, etc.) stored in the tenant's private Forgejo repo.
///
/// Requires the Forgejo-backed mode to be active (<c>Forgejo:AdminToken</c>
/// configured). In filesystem-only mode these endpoints return 503.
/// </summary>
[ApiController]
[Route("api/workflow-files")]
public class WorkflowFilesController : ControllerBase
{
private readonly WorkflowSourceFactory _sourceFactory;
private readonly ILogger<WorkflowFilesController> _logger;
public WorkflowFilesController(
WorkflowSourceFactory sourceFactory,
ILogger<WorkflowFilesController> logger)
{
_sourceFactory = sourceFactory;
_logger = logger;
}
private string TenantId => (string?)HttpContext.Items["TenantId"]
?? throw new InvalidOperationException("TenantId not resolved by auth middleware");
/// <summary>
/// The user's Forgejo login, resolved from the tenant ID by the
/// WorkflowSourceMiddleware. Used for repo path lookups since repos
/// are per-user: <c>{login}/workflows-{login}</c>.
/// </summary>
private string ForgejoLogin => (string?)HttpContext.Items["ForgejoLogin"]
?? throw new InvalidOperationException("ForgejoLogin not resolved by WorkflowSourceMiddleware");
/// <summary>Returns the Forgejo repo info for this tenant's workflow files.</summary>
[HttpGet("repo")]
[RequireScope("read")]
public IActionResult GetRepoInfo()
{
if (_sourceFactory.Forgejo == null)
return StatusCode(503, new { error = "Forgejo-backed workflow mode is not configured." });
var login = ForgejoLogin;
var cloneDir = _sourceFactory.Forgejo.TenantCloneDir(login);
var fullName = ForgejoWorkflowRepoService.RepoFullNameForLogin(login);
var exists = Directory.Exists(Path.Combine(cloneDir, ".git"));
return Ok(new
{
repoFullName = fullName,
cloneDir,
cloned = exists,
forgejoUrl = $"{_sourceFactory.Forgejo.GetType().GetProperty("ForgejoBase")?.GetValue(_sourceFactory.Forgejo) ?? "https://forgejo.wiz4chat.com"}/{fullName}",
});
}
/// <summary>
/// Lists all files in the tenant's workflow repo (recursive, excluding .git).
/// Returns relative paths and basic file info.
/// </summary>
[HttpGet]
[RequireScope("read")]
public IActionResult ListFiles([FromQuery] string? path = null)
{
if (_sourceFactory.Forgejo == null)
return StatusCode(503, new { error = "Forgejo-backed workflow mode is not configured." });
var cloneDir = _sourceFactory.Forgejo.TenantCloneDir(ForgejoLogin);
if (!Directory.Exists(cloneDir))
return Ok(new { files = Array.Empty<object>() });
var dir = string.IsNullOrWhiteSpace(path)
? cloneDir
: Path.Combine(cloneDir, path.Replace('/', Path.DirectorySeparatorChar));
if (!Directory.Exists(dir))
return NotFound(new { error = $"Directory not found: {path}" });
var files = new List<object>();
foreach (var full in Directory.EnumerateFiles(dir))
{
var name = Path.GetFileName(full);
var rel = Path.GetRelativePath(cloneDir, full).Replace(Path.DirectorySeparatorChar, '/');
var info = new FileInfo(full);
files.Add(new { name, path = rel, size = info.Length, modified = info.LastWriteTimeUtc });
}
foreach (var full in Directory.EnumerateDirectories(dir))
{
var name = Path.GetFileName(full);
if (name == ".git") continue;
var rel = Path.GetRelativePath(cloneDir, full).Replace(Path.DirectorySeparatorChar, '/');
files.Add(new { name, path = rel, isDir = true });
}
return Ok(new { files });
}
/// <summary>Returns the content of a specific file.</summary>
[HttpGet("content")]
[RequireScope("read")]
public IActionResult GetContent([FromQuery] string path)
{
if (_sourceFactory.Forgejo == null)
return StatusCode(503, new { error = "Forgejo-backed workflow mode is not configured." });
if (string.IsNullOrWhiteSpace(path))
return BadRequest(new { error = "path is required" });
var cloneDir = _sourceFactory.Forgejo.TenantCloneDir(ForgejoLogin);
var full = Path.Combine(cloneDir, path.Replace('/', Path.DirectorySeparatorChar));
// Reject path traversal.
var fullCloneDir = Path.GetFullPath(cloneDir);
var fullFile = Path.GetFullPath(full);
if (!fullFile.StartsWith(fullCloneDir, StringComparison.Ordinal))
return BadRequest(new { error = "Invalid path" });
if (!System.IO.File.Exists(full))
return NotFound(new { error = $"File not found: {path}" });
var content = System.IO.File.ReadAllText(full);
var info = new FileInfo(full);
return Ok(new { path, content, size = info.Length, modified = info.LastWriteTimeUtc });
}
/// <summary>
/// Saves content to a file in the tenant's workflow repo.
/// Writes to the local clone (does not commit/push — use POST /commit for that).
/// </summary>
[HttpPost("save")]
[RequireScope("manage")]
public IActionResult SaveFile([FromBody] SaveWorkflowFileRequest request)
{
if (_sourceFactory.Forgejo == null)
return StatusCode(503, new { error = "Forgejo-backed workflow mode is not configured." });
if (string.IsNullOrWhiteSpace(request.Path))
return BadRequest(new { error = "path is required" });
var cloneDir = _sourceFactory.Forgejo.TenantCloneDir(ForgejoLogin);
var full = Path.Combine(cloneDir, request.Path.Replace('/', Path.DirectorySeparatorChar));
// Reject path traversal.
var fullCloneDir = Path.GetFullPath(cloneDir);
var fullFile = Path.GetFullPath(full);
if (!fullFile.StartsWith(fullCloneDir, StringComparison.Ordinal))
return BadRequest(new { error = "Invalid path" });
try
{
var dir = Path.GetDirectoryName(full);
if (dir != null && !Directory.Exists(dir))
Directory.CreateDirectory(dir);
System.IO.File.WriteAllText(full, request.Content ?? string.Empty, new System.Text.UTF8Encoding(false));
_logger.LogInformation("Saved workflow file {Path} for user {Login}", request.Path, ForgejoLogin);
return Ok(new { success = true, path = request.Path });
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to save workflow file {Path}", request.Path);
return BadRequest(new { error = ex.Message });
}
}
/// <summary>Returns the git status of the tenant's workflow repo.</summary>
[HttpGet("status")]
[RequireScope("read")]
public IActionResult GetStatus()
{
if (_sourceFactory.Forgejo == null)
return StatusCode(503, new { error = "Forgejo-backed workflow mode is not configured." });
var cloneDir = _sourceFactory.Forgejo.TenantCloneDir(ForgejoLogin);
if (!Directory.Exists(Path.Combine(cloneDir, ".git")))
return Ok(new { branch = "", changed = Array.Empty<object>(), untracked = Array.Empty<object>() });
var branch = RunGit(cloneDir, "rev-parse", "--abbrev-ref", "HEAD")?.Trim() ?? "";
var statusOutput = RunGit(cloneDir, "status", "--porcelain") ?? "";
var changed = new List<object>();
var untracked = new List<object>();
foreach (var line in statusOutput.Split('\n', StringSplitOptions.RemoveEmptyEntries))
{
if (line.Length < 4) continue;
var statusCode = line[..2].Trim();
var filePath = line[3..].Trim();
if (statusCode == "??")
untracked.Add(new { path = filePath, status = "untracked" });
else
changed.Add(new { path = filePath, status = statusCode });
}
var head = RunGit(cloneDir, "rev-parse", "--short", "HEAD")?.Trim() ?? "";
return Ok(new { branch, head, changed, untracked });
}
/// <summary>
/// Commits all changes and pushes to Forgejo. Optionally provides a commit message.
/// After committing, call POST /api/workflows/sync to recompile.
/// </summary>
[HttpPost("commit")]
[RequireScope("manage")]
public async Task<IActionResult> CommitAndPush([FromBody] CommitWorkflowRequest? request = null)
{
if (_sourceFactory.Forgejo == null)
return StatusCode(503, new { error = "Forgejo-backed workflow mode is not configured." });
var message = string.IsNullOrWhiteSpace(request?.Message)
? $"Update workflow files ({DateTime.UtcNow:yyyy-MM-dd HH:mm:ss} UTC)"
: request!.Message;
var (success, output) = await _sourceFactory.Forgejo.CommitAndPushAsync(
ForgejoLogin, message, HttpContext.RequestAborted);
_logger.LogInformation("Commit/push for login {Login}: {Success} ({Output})",
ForgejoLogin, success, output?.Trim() ?? "");
return Ok(new { success, output });
}
/// <summary>Pulls the latest changes from Forgejo.</summary>
[HttpPost("pull")]
[RequireScope("manage")]
public async Task<IActionResult> Pull()
{
if (_sourceFactory.Forgejo == null)
return StatusCode(503, new { error = "Forgejo-backed workflow mode is not configured." });
var ok = await _sourceFactory.Forgejo.PullAsync(ForgejoLogin, HttpContext.RequestAborted);
return Ok(new { success = ok });
}
private string? RunGit(string workDir, params string[] args)
{
try
{
var psi = new ProcessStartInfo("git")
{
WorkingDirectory = workDir,
RedirectStandardOutput = true,
RedirectStandardError = true,
UseShellExecute = false,
};
foreach (var arg in args) psi.ArgumentList.Add(arg);
using var process = Process.Start(psi);
if (process == null) return null;
var output = process.StandardOutput.ReadToEnd();
if (!process.WaitForExit(5000))
{
process.Kill(entireProcessTree: true);
return null;
}
return process.ExitCode == 0 ? output.Trim() : null;
}
catch (Exception ex)
{
_logger.LogDebug(ex, "git invocation failed in {Dir}", workDir);
return null;
}
}
}
public sealed record SaveWorkflowFileRequest(string Path, string? Content);
public sealed record CommitWorkflowRequest(string? Message);

View file

@ -1,4 +1,5 @@
using Microsoft.EntityFrameworkCore;
using Npgsql;
using Serilog;
using Serilog.Events;
using Serilog.Sinks.OpenSearch;
@ -82,6 +83,10 @@ else
builder.Services.AddDbContext<WorkflowsDbContext>(options =>
options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection")));
// NpgsqlDataSource for lightweight queries outside EF (e.g. login resolution).
builder.Services.AddSingleton<NpgsqlDataSource>(_ =>
NpgsqlDataSource.Create(builder.Configuration.GetConnectionString("DefaultConnection")!));
builder.Services.AddSingleton<IConnectionMultiplexer>(_ =>
ConnectionMultiplexer.Connect(
builder.Configuration["Redis:ConnectionString"] ?? "localhost:6379,abortConnect=false"));
@ -118,11 +123,16 @@ builder.Services.AddSingleton<WorkflowCompiler>();
// Workflow source: factory-based — each request creates a tenant-scoped source.
// WorkflowSource:CopiesRoot is REQUIRED; the factory throws if it is empty,
// preventing any request from serving cross-tenant workflows.
// When Forgejo:AdminToken is configured, the factory creates Forgejo-backed
// sources with per-tenant repos; otherwise it falls back to filesystem-only.
builder.Services.AddSingleton<WorkflowSourceFactory>();
builder.Services.AddScoped<IWorkflowSource>(sp =>
{
var http = sp.GetRequiredService<IHttpContextAccessor>();
var factory = sp.GetRequiredService<WorkflowSourceFactory>();
// Check if a Forgejo-resolved source was already prepared by the middleware.
if (http.HttpContext?.Items["WorkflowSource"] is IWorkflowSource forgejoSrc)
return forgejoSrc;
var tenantId = http.HttpContext?.Items["TenantId"] as string
?? throw new InvalidOperationException("TenantId not resolved by auth middleware");
return factory.Create(tenantId);
@ -215,6 +225,39 @@ var app = builder.Build();
app.UseCors();
app.UseMiddleware<AuthMiddleware>();
// Resolve the Forgejo-backed workflow source asynchronously after auth.
// The auth middleware sets TenantId; this middleware resolves the user's Forgejo
// login, clones/pulls the repo, and stores the IWorkflowSource in HttpContext.Items
// so the DI registration can pick it up synchronously.
app.Use(async (context, next) =>
{
var tenantId = context.Items["TenantId"] as string;
if (!string.IsNullOrEmpty(tenantId))
{
var factory = context.RequestServices.GetRequiredService<WorkflowSourceFactory>();
if (factory.IsForgejoBacked)
{
try
{
var source = await factory.CreateAsync(tenantId, context.RequestAborted);
context.Items["WorkflowSource"] = source;
// Store the resolved Forgejo login for controllers that need
// to resolve repo paths (WorkflowFilesController, etc.).
var login = await factory.ResolveForgejoLoginAsync(tenantId, context.RequestAborted);
if (!string.IsNullOrEmpty(login))
context.Items["ForgejoLogin"] = login;
}
catch (Exception ex)
{
var logger = context.RequestServices.GetRequiredService<ILoggerFactory>()
.CreateLogger("WorkflowSourceMiddleware");
logger.LogWarning(ex, "Forgejo-backed source failed for tenant {TenantId}, falling back to filesystem", tenantId);
}
}
}
await next();
});
// Ensure the database schema exists before serving traffic.
try
{

View file

@ -0,0 +1,442 @@
using System.Diagnostics;
namespace w4c_workflows.Services;
/// <summary>
/// Manages per-tenant private Forgejo repositories for workflow source files.
/// Each tenant gets a repo named <c>workflows-{tenantId}</c> under the configured
/// Forgejo owner. The repo contains the tenant's workflow YAML definitions and
/// their sibling code files (shell scripts, JS, Python, etc.).
///
/// On first access the repo is created via the Forgejo admin API, cloned locally,
/// and seeded with templates from <c>WorkflowSource:SharedDir</c>. Subsequent calls
/// pull remote changes before the sync service reads the files.
///
/// The local clone lives at <c>{WorkflowSource:CopiesRoot}/{tenantId}/</c> — the
/// same directory layout the legacy <see cref="PerTenantWorkflowSource"/> used, so
/// the existing <see cref="WorkflowSyncService"/> and worker runtime need no path
/// changes when the backing store switches from plain filesystem to git.
/// </summary>
public sealed class ForgejoWorkflowRepoService
{
private readonly string _forgejoBase;
private readonly string? _forgejoToken;
private readonly string? _forgejoAdminToken;
private readonly string _forgejoOwner;
private readonly string _copiesRoot;
private readonly string _sharedDir;
private readonly ILogger<ForgejoWorkflowRepoService> _logger;
public ForgejoWorkflowRepoService(IConfiguration config, ILogger<ForgejoWorkflowRepoService> logger)
{
_logger = logger;
_forgejoBase = (config["Forgejo:BaseUrl"] ?? "https://forgejo.wiz4chat.com").TrimEnd('/');
_forgejoToken = config["Forgejo:AccessToken"]?.Trim();
_forgejoAdminToken = config["Forgejo:AdminToken"]?.Trim();
_forgejoOwner = config["Forgejo:WorkflowRepoOwner"]
?? config["Forgejo:Owner"]
?? string.Empty;
_copiesRoot = Path.GetFullPath(
config["WorkflowSource:CopiesRoot"]
?? throw new InvalidOperationException(
"WorkflowSource:CopiesRoot is not configured. " +
"Set it to a writable directory path (e.g. \"/data/workflow-tenants\")."));
_sharedDir = config["WorkflowSource:SharedDir"] ?? "workflows";
}
/// <summary>True when Forgejo admin provisioning is configured.</summary>
public bool IsConfigured =>
!string.IsNullOrWhiteSpace(_forgejoAdminToken) &&
!string.IsNullOrWhiteSpace(_forgejoOwner);
/// <summary>The repo full name on Forgejo (owner/repo-name).</summary>
public string RepoFullName(string tenantId) =>
$"{_forgejoOwner.ToLowerInvariant()}/{RepoName(tenantId)}";
/// <summary>The sanitized repo name for a tenant.</summary>
public static string RepoName(string tenantId) =>
$"workflows-{Sanitize(tenantId)}";
/// <summary>
/// Resolves the Forgejo repo full name for a user. The repo lives under
/// the user's own Forgejo account: <c>{login}/workflows-{login}</c>.
/// </summary>
public static string RepoFullNameForLogin(string login) =>
$"{login.ToLowerInvariant()}/workflows-{Sanitize(login.ToLowerInvariant())}";
/// <summary>
/// Ensures the user's Forgejo repo exists. Creates it via the admin API
/// if missing (idempotent). Returns the repo full name (owner/repo).
/// Throws on hard failures.
/// </summary>
public async Task<string> EnsureRepoAsync(string login, CancellationToken ct = default)
{
if (!IsConfigured)
throw new InvalidOperationException(
"Forgejo workflow repo provisioning is not configured. " +
"Set Forgejo:AdminToken and Forgejo:Owner (or Forgejo:WorkflowRepoOwner).");
var owner = login.ToLowerInvariant();
var name = $"workflows-{Sanitize(owner)}";
var fullName = $"{owner}/{name}";
try
{
// Check if repo already exists.
using var exists = await SendAdminAsync(HttpMethod.Get,
$"/repos/{Uri.EscapeDataString(owner)}/{Uri.EscapeDataString(name)}", null, ct);
if (exists.IsSuccessStatusCode)
{
_logger.LogDebug("Forgejo workflow repo {FullName} already exists", fullName);
return fullName;
}
// Create the repo.
using var create = await SendAdminAsync(HttpMethod.Post,
$"/admin/users/{Uri.EscapeDataString(owner)}/repos",
new
{
name,
@private = true,
auto_init = true,
description = $"Workflow definitions for user {login}",
}, ct);
if (!create.IsSuccessStatusCode)
{
var body = await create.Content.ReadAsStringAsync(ct);
_logger.LogWarning("Forgejo workflow repo create {FullName} -> {Status}: {Body}",
fullName, (int)create.StatusCode, body);
throw new InvalidOperationException(
$"Could not create Forgejo workflow repo {fullName}: {(int)create.StatusCode}");
}
_logger.LogInformation("Created Forgejo workflow repo {FullName}", fullName);
return fullName;
}
catch (InvalidOperationException)
{
throw;
}
catch (Exception ex)
{
_logger.LogError(ex, "Forgejo workflow repo ensure failed for {FullName}", fullName);
throw new InvalidOperationException(
$"Could not create Forgejo workflow repo {fullName}: {ex.Message}");
}
}
/// <summary>
/// Ensures a local clone of the user's workflow repo exists. If the repo
/// doesn't exist on Forgejo yet, it is created. If the clone directory is
/// missing, the repo is cloned. Returns the absolute path to the local clone.
/// </summary>
public async Task<string> EnsureCloneAsync(string login, CancellationToken ct = default)
{
var tenantDir = TenantCloneDir(login);
if (Directory.Exists(Path.Combine(tenantDir, ".git")))
{
_logger.LogDebug("Local clone for login {Login} already exists at {Dir}", login, tenantDir);
return tenantDir;
}
// Ensure the Forgejo repo exists.
var fullName = await EnsureRepoAsync(login, ct);
var cloneUrl = $"{_forgejoBase}/{fullName}.git";
Directory.CreateDirectory(tenantDir);
// Clone the repo.
var psi = new ProcessStartInfo("git")
{
WorkingDirectory = tenantDir,
RedirectStandardOutput = true,
RedirectStandardError = true,
UseShellExecute = false,
CreateNoWindow = true,
};
psi.ArgumentList.Add("clone");
if (!string.IsNullOrWhiteSpace(_forgejoToken))
{
psi.ArgumentList.Add("-c");
psi.ArgumentList.Add($"http.extraHeader=Authorization: token {_forgejoToken}");
}
psi.ArgumentList.Add("--");
psi.ArgumentList.Add(cloneUrl);
psi.ArgumentList.Add(".");
using var process = Process.Start(psi);
if (process == null)
throw new InvalidOperationException("git clone failed to start");
var stdout = await process.StandardOutput.ReadToEndAsync(ct);
var stderr = await process.StandardError.ReadToEndAsync(ct);
await process.WaitForExitAsync(ct);
if (process.ExitCode != 0)
{
_logger.LogWarning("Clone {FullName} failed ({Code}): {Err}",
fullName, process.ExitCode, stderr.Trim());
throw new InvalidOperationException(
string.IsNullOrWhiteSpace(stderr) ? "git clone failed" : stderr.Trim());
}
_logger.LogInformation("Cloned workflow repo {FullName} into {Dir}", fullName, tenantDir);
// Clean up any baked-in extra header from clone config.
UnsetLocalConfig(tenantDir, "http.extraheader");
// If the repo is empty (freshly created with auto_init), seed with templates.
await SeedTemplatesIfEmptyAsync(tenantDir, ct);
return tenantDir;
}
/// <summary>
/// Pulls the latest changes from the remote. Returns true on success.
/// </summary>
public async Task<bool> PullAsync(string login, CancellationToken ct = default)
{
var tenantDir = TenantCloneDir(login);
if (!Directory.Exists(Path.Combine(tenantDir, ".git")))
{
_logger.LogWarning("No local clone for login {Login}, cannot pull", login);
return false;
}
var token = !string.IsNullOrWhiteSpace(_forgejoToken) ? _forgejoToken : _forgejoAdminToken;
var output = !string.IsNullOrWhiteSpace(token)
? await RunGitAsync(tenantDir, ct, "-c", $"http.extraHeader=Authorization: token {token}", "pull", "--ff-only")
: await RunGitAsync(tenantDir, ct, "pull", "--ff-only");
if (output == null)
{
_logger.LogWarning("Pull failed for login {Login}", login);
return false;
}
var ok = !output.Contains("fatal:", StringComparison.OrdinalIgnoreCase) &&
!output.Contains("Could not resolve host", StringComparison.OrdinalIgnoreCase) &&
!output.Contains("Permission denied", StringComparison.OrdinalIgnoreCase);
_logger.LogInformation("Pull for login {Login}: {Ok} ({Out})",
login, ok, output.Trim());
return ok;
}
/// <summary>
/// Commits and pushes all changes in the tenant's clone. Used after file
/// edits via the source code API to persist changes to Forgejo.
/// </summary>
public async Task<(bool Success, string Output)> CommitAndPushAsync(
string tenantId, string message, CancellationToken ct = default)
{
var tenantDir = TenantCloneDir(tenantId);
if (!Directory.Exists(Path.Combine(tenantDir, ".git")))
return (false, "No local clone");
await RunGitAsync(tenantDir, ct, "add", "-A");
var commitOutput = await RunGitAsync(tenantDir, ct, "commit", "-m", message);
if (commitOutput == null || commitOutput.Contains("nothing to commit", StringComparison.OrdinalIgnoreCase))
return (true, "Nothing to commit");
var token = !string.IsNullOrWhiteSpace(_forgejoToken) ? _forgejoToken : _forgejoAdminToken;
var pushOutput = !string.IsNullOrWhiteSpace(token)
? await RunGitAsync(tenantDir, ct, "-c", $"http.extraHeader=Authorization: token {token}", "push")
: await RunGitAsync(tenantDir, ct, "push");
var ok = pushOutput != null &&
!pushOutput.Contains("fatal:", StringComparison.OrdinalIgnoreCase) &&
!pushOutput.Contains("Permission denied", StringComparison.OrdinalIgnoreCase);
return (ok, ok ? pushOutput?.Trim() ?? "Pushed" : pushOutput?.Trim() ?? "Push failed");
}
/// <summary>The absolute path to a tenant's local clone directory.</summary>
public string TenantCloneDir(string tenantId)
=> Path.GetFullPath(Path.Combine(_copiesRoot, Sanitize(tenantId)));
/// <summary>
/// Seeds the local clone with template files from the shared directory.
/// Only runs when the repo is empty (no workflow YAML files yet).
/// </summary>
private async Task SeedTemplatesIfEmptyAsync(string cloneDir, CancellationToken ct)
{
// Check if the repo already has workflow YAML files.
var yamlCount = Directory.EnumerateFiles(cloneDir, "*.yaml", SearchOption.AllDirectories)
.Concat(Directory.EnumerateFiles(cloneDir, "*.yml", SearchOption.AllDirectories))
.Count(f => !f.Contains(".git", StringComparison.OrdinalIgnoreCase));
if (yamlCount > 0)
{
_logger.LogDebug("Clone already has {Count} YAML files, skipping seed", yamlCount);
return;
}
var sharedDir = ResolveSharedDir();
if (!Directory.Exists(sharedDir))
{
_logger.LogWarning("Shared workflow directory {Shared} does not exist; clone starts empty", sharedDir);
return;
}
_logger.LogInformation("Seeding empty clone from {Shared}", sharedDir);
// Copy ALL files (YAML + code) from the shared directory.
CopyAllFiles(sharedDir, cloneDir);
// Commit the seeded templates.
await RunGitAsync(cloneDir, ct, "add", "-A");
var commitOutput = await RunGitAsync(cloneDir, ct, "commit", "-m", "Initial workflow templates");
if (commitOutput != null && !commitOutput.Contains("nothing to commit", StringComparison.OrdinalIgnoreCase))
{
var token = !string.IsNullOrWhiteSpace(_forgejoToken) ? _forgejoToken : _forgejoAdminToken;
if (!string.IsNullOrWhiteSpace(token))
{
await RunGitAsync(cloneDir, ct, "-c", $"http.extraHeader=Authorization: token {token}", "push");
}
else
{
await RunGitAsync(cloneDir, ct, "push");
}
_logger.LogInformation("Pushed initial templates to Forgejo");
}
}
private string ResolveSharedDir()
{
var sharedRel = _sharedDir;
if (Path.IsPathRooted(sharedRel) && Directory.Exists(sharedRel))
return Path.GetFullPath(sharedRel);
// Try resolving relative to the repo root (walk up from the service dir).
var start = AppContext.BaseDirectory;
var walk = Path.GetFullPath(start);
while (true)
{
var candidate = Path.Combine(walk, sharedRel);
if (Directory.Exists(candidate))
return Path.GetFullPath(candidate);
var parent = Directory.GetParent(walk)?.FullName;
if (string.IsNullOrEmpty(parent) || parent == walk)
break;
walk = parent;
}
return Path.GetFullPath(Path.Combine(Directory.GetCurrentDirectory(), sharedRel));
}
private static void CopyAllFiles(string src, string dst)
{
foreach (var file in Directory.EnumerateFiles(src, "*", SearchOption.AllDirectories))
{
var rel = Path.GetRelativePath(src, file);
if (rel.StartsWith(".git", StringComparison.OrdinalIgnoreCase))
continue;
var destPath = Path.Combine(dst, rel);
var destDir = Path.GetDirectoryName(destPath);
if (destDir != null && !Directory.Exists(destDir))
Directory.CreateDirectory(destDir);
try
{
File.Copy(file, destPath, overwrite: true);
}
catch (Exception)
{
// Best-effort — template files that fail to copy are non-fatal.
}
}
}
private async Task<string?> RunGitAsync(string workDir, CancellationToken ct, params string[] args)
{
try
{
var psi = new ProcessStartInfo("git")
{
WorkingDirectory = workDir,
RedirectStandardOutput = true,
RedirectStandardError = true,
UseShellExecute = false,
CreateNoWindow = true,
};
foreach (var a in args) psi.ArgumentList.Add(a);
using var process = Process.Start(psi);
if (process == null) return null;
var stdout = await process.StandardOutput.ReadToEndAsync(ct);
var stderr = await process.StandardError.ReadToEndAsync(ct);
await process.WaitForExitAsync(ct);
return process.ExitCode == 0 ? stdout : stderr;
}
catch (Exception ex)
{
_logger.LogDebug(ex, "git invocation failed in {Dir}", workDir);
return null;
}
}
private async Task<HttpResponseMessage> SendAdminAsync(
HttpMethod method, string path, object? body, CancellationToken ct)
{
using var request = new HttpRequestMessage(method, $"{_forgejoBase}/api/v1{path}");
request.Headers.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue(
"Bearer", _forgejoAdminToken);
if (body != null)
{
request.Content = new StringContent(
System.Text.Json.JsonSerializer.Serialize(body, new System.Text.Json.JsonSerializerOptions
{
PropertyNamingPolicy = System.Text.Json.JsonNamingPolicy.SnakeCaseLower,
DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull,
}),
System.Text.Encoding.UTF8,
"application/json");
}
using var client = new HttpClient { Timeout = TimeSpan.FromSeconds(30) };
return await client.SendAsync(request, ct);
}
private static void UnsetLocalConfig(string repoRoot, string key)
{
try
{
var psi = new ProcessStartInfo("git")
{
WorkingDirectory = repoRoot,
RedirectStandardOutput = true,
RedirectStandardError = true,
UseShellExecute = false,
CreateNoWindow = true,
};
psi.ArgumentList.Add("config");
psi.ArgumentList.Add("--local");
psi.ArgumentList.Add("--unset");
psi.ArgumentList.Add(key);
using var p = Process.Start(psi);
p?.WaitForExit(3000);
}
catch
{
// Best-effort cleanup.
}
}
private static string Sanitize(string id)
{
if (string.IsNullOrWhiteSpace(id))
return "_";
var sb = new System.Text.StringBuilder(id.Length);
foreach (var ch in id)
sb.Append(char.IsLetterOrDigit(ch) || ch == '-' || ch == '_' ? ch : '_');
var result = sb.ToString();
return result.Length > 120 ? result[..120] : result;
}
}

View file

@ -1,4 +1,6 @@
using System.Diagnostics;
using Npgsql;
using NpgsqlTypes;
namespace w4c_workflows.Services;
@ -8,6 +10,19 @@ namespace w4c_workflows.Services;
/// middleware); the factory resolves directories and returns a source bound
/// to that tenant's workflow directory.
///
/// <b>Two modes</b> (automatic, config-driven):
/// <list type="number">
/// <item><b>Forgejo-backed</b> (preferred): when <c>Forgejo:AdminToken</c> is
/// configured, each tenant's workflow files live in a private Forgejo repo
/// (<c>workflows-{tenantId}</c>). The factory clones/pulls via
/// <see cref="ForgejoWorkflowRepoService"/> and returns a
/// <see cref="PerTenantWorkflowSource"/> backed by the local clone.
/// This gives tenants real git history, branching, and remote sync.</item>
/// <item><b>Filesystem-only</b> (fallback): when Forgejo is not configured,
/// uses plain directory copies seeded from the shared template dir.
/// Preserves the original behavior for dev/self-hosted setups.</item>
/// </list>
///
/// Tenant isolation is <b>mandatory</b>: <c>WorkflowSource:CopiesRoot</c> must
/// be set in configuration. The factory throws at construction time if it is
/// missing, preventing any request from serving cross-tenant workflows.
@ -23,12 +38,16 @@ public sealed class WorkflowSourceFactory
private readonly string _copiesRoot;
private readonly string _sharedDir;
private readonly string _repoRoot;
private readonly ForgejoWorkflowRepoService? _forgejo;
private readonly NpgsqlDataSource? _ds;
public WorkflowSourceFactory(IConfiguration config, IWebHostEnvironment env, ILoggerFactory loggers)
public WorkflowSourceFactory(IConfiguration config, IWebHostEnvironment env, ILoggerFactory loggers,
NpgsqlDataSource? ds = null)
{
_config = config;
_env = env;
_loggers = loggers;
_ds = ds;
_copiesRoot = config["WorkflowSource:CopiesRoot"]
?? throw new InvalidOperationException(
@ -37,11 +56,37 @@ public sealed class WorkflowSourceFactory
"directory path (e.g. \"/data/workflow-tenants\" or \".data/workflow-tenants\").");
_sharedDir = config["WorkflowSource:SharedDir"] ?? "workflows";
_repoRoot = ResolveRepoRoot(config, env);
// Create the Forgejo service when admin provisioning is configured.
var forgejoLogger = loggers.CreateLogger<ForgejoWorkflowRepoService>();
var forgejoSvc = new ForgejoWorkflowRepoService(config, forgejoLogger);
_forgejo = forgejoSvc.IsConfigured ? forgejoSvc : null;
if (_forgejo != null)
_loggers.CreateLogger<WorkflowSourceFactory>()
.LogInformation("Workflow source: Forgejo-backed mode (owner={Owner})",
config["Forgejo:Owner"] ?? config["Forgejo:WorkflowRepoOwner"]);
else
_loggers.CreateLogger<WorkflowSourceFactory>()
.LogInformation("Workflow source: filesystem-only mode (no Forgejo admin token)");
}
/// <summary>Creates a per-tenant workflow source. The tenant directory
/// <summary>True when Forgejo-backed mode is active.</summary>
public bool IsForgejoBacked => _forgejo != null;
/// <summary>The Forgejo repo service (null when in filesystem-only mode).</summary>
public ForgejoWorkflowRepoService? Forgejo => _forgejo;
/// <summary>
/// Creates a per-tenant workflow source.
///
/// In Forgejo-backed mode: ensures the local clone exists (creating the
/// Forgejo repo if needed), then returns a source backed by the clone.
///
/// In filesystem-only mode: the tenant directory
/// <c>{CopiesRoot}/{sanitizedTenantId}/</c> is created (and seeded from
/// the shared template dir) on first access.</summary>
/// the shared template dir) on first access.
/// </summary>
public IWorkflowSource Create(string tenantId)
{
var safeId = Sanitize(tenantId);
@ -58,6 +103,63 @@ public sealed class WorkflowSourceFactory
return new PerTenantWorkflowSource(tenantId, tenantDir, sharedDir, logger);
}
/// <summary>
/// Creates a per-tenant workflow source asynchronously. In Forgejo-backed
/// mode this ensures the local clone exists (cloning from Forgejo if needed).
/// In filesystem-only mode, delegates to <see cref="Create"/>.
/// </summary>
public async Task<IWorkflowSource> CreateAsync(string tenantId, CancellationToken ct = default)
{
if (_forgejo == null)
return Create(tenantId);
// Resolve the user's Forgejo login from the tenant ID (forgejo_id).
// The repo lives under the user's own account: {login}/workflows-{login}.
var login = await ResolveForgejoLoginAsync(tenantId, ct);
if (string.IsNullOrEmpty(login))
{
_loggers.CreateLogger<WorkflowSourceFactory>()
.LogWarning("Could not resolve Forgejo login for tenant {TenantId}, falling back to filesystem",
tenantId);
return Create(tenantId);
}
// Ensure the Forgejo repo exists and is cloned locally.
var cloneDir = await _forgejo.EnsureCloneAsync(login, ct);
// Pull latest changes before reading.
await _forgejo.PullAsync(login, ct);
var logger = _loggers.CreateLogger<PerTenantWorkflowSource>();
return new PerTenantWorkflowSource(tenantId, cloneDir, _sharedDir, logger);
}
/// <summary>
/// Resolves the Forgejo login for a tenant (forgejo_id) from the auth_users table.
/// </summary>
public async Task<string?> ResolveForgejoLoginAsync(string tenantId, CancellationToken ct)
{
if (_ds == null || !long.TryParse(tenantId, out var forgejoId))
return null;
try
{
await using var conn = await _ds.OpenConnectionAsync(ct);
await using var cmd = conn.CreateCommand();
cmd.CommandTimeout = 3;
cmd.CommandText = "SELECT login FROM auth_users WHERE forgejo_id = @fid LIMIT 1";
cmd.Parameters.Add(new NpgsqlParameter("@fid", NpgsqlDbType.Bigint) { Value = forgejoId });
var result = await cmd.ExecuteScalarAsync(ct);
return result as string;
}
catch (Exception ex)
{
_loggers.CreateLogger<WorkflowSourceFactory>()
.LogWarning(ex, "Could not resolve Forgejo login for tenant {TenantId}", tenantId);
return null;
}
}
private static string Sanitize(string id)
{
if (string.IsNullOrWhiteSpace(id))

View file

@ -10,26 +10,29 @@ namespace w4c_workflows.Services.Runs;
/// 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).
/// 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 _workerRoot;
private readonly string _defaultWorkerRoot;
private readonly WorkflowSourceFactory? _sourceFactory;
private readonly ILogger<TaskDispatcher> _logger;
public TaskDispatcher(
WorkflowsDbContext db,
IJobQueue jobs,
IConfiguration config,
ILogger<TaskDispatcher> logger)
ILogger<TaskDispatcher> logger,
WorkflowSourceFactory? sourceFactory = null)
{
_db = db;
_jobs = jobs;
_logger = logger;
_workerRoot = ResolveWorkerRoot(config);
_sourceFactory = sourceFactory;
_defaultWorkerRoot = ResolveWorkerRoot(config);
}
public async Task<Guid> DispatchAsync(
@ -57,7 +60,8 @@ public class TaskDispatcher
_db.TaskRuns.Add(taskRun);
await _db.SaveChangesAsync(ct);
var fields = TaskRunMessage.ToFields(run, task, input, ResolveWorkingDir(workflowPath), attempt);
var workingDir = ResolveWorkingDir(run.TenantId, workflowPath);
var fields = TaskRunMessage.ToFields(run, task, input, workingDir, attempt);
await _jobs.EnqueueAsync(run.TenantId, fields, ct);
_logger.LogDebug(
@ -76,8 +80,9 @@ public class TaskDispatcher
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, ResolveWorkingDir(run.Workflow.Path), taskRun.Attempt);
run, task, taskRun.InputJson, workingDir, taskRun.Attempt);
var dlqFields = new Dictionary<string, string>(fields)
{
@ -91,12 +96,41 @@ public class TaskDispatcher
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.
/// </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 Path.Combine(workerRoot, dir.Replace('/', Path.DirectorySeparatorChar));
}
/// <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 _workerRoot;
return Path.Combine(_workerRoot, dir.Replace('/', Path.DirectorySeparatorChar));
return _defaultWorkerRoot;
return Path.Combine(_defaultWorkerRoot, dir.Replace('/', Path.DirectorySeparatorChar));
}
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)

View file

@ -20,28 +20,42 @@ public sealed class SyncResult
/// tenant's DB state, and removes workflows whose files were deleted. Ids are
/// deterministic, so re-syncing upserts instead of duplicating. Files that fail
/// to compile are reported in <see cref="SyncResult.Errors"/> and not persisted.
///
/// When Forgejo-backed mode is active, the service pulls the latest changes from
/// the remote before reading files, ensuring the compiled state reflects the
/// latest committed YAML.
/// </summary>
public class WorkflowSyncService
{
private readonly WorkflowsDbContext _db;
private readonly WorkflowCompiler _compiler;
private readonly IWorkflowSource _source;
private readonly WorkflowSourceFactory? _sourceFactory;
private readonly ILogger<WorkflowSyncService> _logger;
public WorkflowSyncService(
WorkflowsDbContext db,
WorkflowCompiler compiler,
IWorkflowSource source,
ILogger<WorkflowSyncService> logger)
ILogger<WorkflowSyncService> logger,
WorkflowSourceFactory? sourceFactory = null)
{
_db = db;
_compiler = compiler;
_source = source;
_logger = logger;
_sourceFactory = sourceFactory;
}
public async Task<SyncResult> SyncAsync(string tenantId, CancellationToken ct)
{
// In Forgejo-backed mode, pull the latest changes before reading files.
if (_sourceFactory?.IsForgejoBacked == true && _sourceFactory.Forgejo != null)
{
_logger.LogDebug("Pulling latest workflow files from Forgejo for tenant {TenantId}", tenantId);
await _sourceFactory.Forgejo.PullAsync(tenantId, ct);
}
var state = _source.GetState();
var files = await _source.ListAsync(ct);

View file

@ -59,7 +59,10 @@
},
"Forgejo": {
"BaseUrl": "https://forgejo.wiz4chat.com",
"AccessToken": ""
"AccessToken": "043a13f8ee662d1ad96ca0d3018ab17907ec3470",
"AdminToken": "8d811a81579747f0d9b2e54c9068edd49618d15c",
"Owner": "test",
"WorkflowRepoOwner": ""
},
"SourceCode": {
"RepoRoot": ""