From 3e31e4a8dad2661123291234412503984ef8ce15 Mon Sep 17 00:00:00 2001 From: Vitali sharp8n Date: Wed, 2 Sep 2026 01:12:21 +0300 Subject: [PATCH] stabilization --- Controllers/WorkflowFilesController.cs | 278 ++++++++++++++++ Program.cs | 43 +++ Services/ForgejoWorkflowRepoService.cs | 442 +++++++++++++++++++++++++ Services/PerTenantWorkflowSource.cs | 108 +++++- Services/Runs/TaskDispatcher.cs | 52 ++- Services/WorkflowSyncService.cs | 16 +- appsettings.json | 5 +- 7 files changed, 930 insertions(+), 14 deletions(-) create mode 100644 Controllers/WorkflowFilesController.cs create mode 100644 Services/ForgejoWorkflowRepoService.cs diff --git a/Controllers/WorkflowFilesController.cs b/Controllers/WorkflowFilesController.cs new file mode 100644 index 0000000..ce3c0f1 --- /dev/null +++ b/Controllers/WorkflowFilesController.cs @@ -0,0 +1,278 @@ +using System.Diagnostics; +using Microsoft.AspNetCore.Mvc; +using w4c_workflows.Filters; +using w4c_workflows.Services; + +namespace w4c_workflows.Controllers; + +/// +/// 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 (Forgejo:AdminToken +/// configured). In filesystem-only mode these endpoints return 503. +/// +[ApiController] +[Route("api/workflow-files")] +public class WorkflowFilesController : ControllerBase +{ + private readonly WorkflowSourceFactory _sourceFactory; + private readonly ILogger _logger; + + public WorkflowFilesController( + WorkflowSourceFactory sourceFactory, + ILogger logger) + { + _sourceFactory = sourceFactory; + _logger = logger; + } + + private string TenantId => (string?)HttpContext.Items["TenantId"] + ?? throw new InvalidOperationException("TenantId not resolved by auth middleware"); + + /// + /// The user's Forgejo login, resolved from the tenant ID by the + /// WorkflowSourceMiddleware. Used for repo path lookups since repos + /// are per-user: {login}/workflows-{login}. + /// + private string ForgejoLogin => (string?)HttpContext.Items["ForgejoLogin"] + ?? throw new InvalidOperationException("ForgejoLogin not resolved by WorkflowSourceMiddleware"); + + /// Returns the Forgejo repo info for this tenant's workflow files. + [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}", + }); + } + + /// + /// Lists all files in the tenant's workflow repo (recursive, excluding .git). + /// Returns relative paths and basic file info. + /// + [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() }); + + 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(); + 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 }); + } + + /// Returns the content of a specific file. + [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 }); + } + + /// + /// 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). + /// + [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 }); + } + } + + /// Returns the git status of the tenant's workflow repo. + [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(), untracked = Array.Empty() }); + + var branch = RunGit(cloneDir, "rev-parse", "--abbrev-ref", "HEAD")?.Trim() ?? ""; + var statusOutput = RunGit(cloneDir, "status", "--porcelain") ?? ""; + + var changed = new List(); + var untracked = new List(); + + 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 }); + } + + /// + /// Commits all changes and pushes to Forgejo. Optionally provides a commit message. + /// After committing, call POST /api/workflows/sync to recompile. + /// + [HttpPost("commit")] + [RequireScope("manage")] + public async Task 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 }); + } + + /// Pulls the latest changes from Forgejo. + [HttpPost("pull")] + [RequireScope("manage")] + public async Task 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); diff --git a/Program.cs b/Program.cs index f44e9d8..a8c3f15 100644 --- a/Program.cs +++ b/Program.cs @@ -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(options => options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); + // NpgsqlDataSource for lightweight queries outside EF (e.g. login resolution). + builder.Services.AddSingleton(_ => + NpgsqlDataSource.Create(builder.Configuration.GetConnectionString("DefaultConnection")!)); + builder.Services.AddSingleton(_ => ConnectionMultiplexer.Connect( builder.Configuration["Redis:ConnectionString"] ?? "localhost:6379,abortConnect=false")); @@ -118,11 +123,16 @@ builder.Services.AddSingleton(); // 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(); builder.Services.AddScoped(sp => { var http = sp.GetRequiredService(); var factory = sp.GetRequiredService(); + // 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(); +// 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(); + 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() + .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 { diff --git a/Services/ForgejoWorkflowRepoService.cs b/Services/ForgejoWorkflowRepoService.cs new file mode 100644 index 0000000..002e242 --- /dev/null +++ b/Services/ForgejoWorkflowRepoService.cs @@ -0,0 +1,442 @@ +using System.Diagnostics; + +namespace w4c_workflows.Services; + +/// +/// Manages per-tenant private Forgejo repositories for workflow source files. +/// Each tenant gets a repo named workflows-{tenantId} 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 WorkflowSource:SharedDir. Subsequent calls +/// pull remote changes before the sync service reads the files. +/// +/// The local clone lives at {WorkflowSource:CopiesRoot}/{tenantId}/ — the +/// same directory layout the legacy used, so +/// the existing and worker runtime need no path +/// changes when the backing store switches from plain filesystem to git. +/// +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 _logger; + + public ForgejoWorkflowRepoService(IConfiguration config, ILogger 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"; + } + + /// True when Forgejo admin provisioning is configured. + public bool IsConfigured => + !string.IsNullOrWhiteSpace(_forgejoAdminToken) && + !string.IsNullOrWhiteSpace(_forgejoOwner); + + /// The repo full name on Forgejo (owner/repo-name). + public string RepoFullName(string tenantId) => + $"{_forgejoOwner.ToLowerInvariant()}/{RepoName(tenantId)}"; + + /// The sanitized repo name for a tenant. + public static string RepoName(string tenantId) => + $"workflows-{Sanitize(tenantId)}"; + + /// + /// Resolves the Forgejo repo full name for a user. The repo lives under + /// the user's own Forgejo account: {login}/workflows-{login}. + /// + public static string RepoFullNameForLogin(string login) => + $"{login.ToLowerInvariant()}/workflows-{Sanitize(login.ToLowerInvariant())}"; + + /// + /// 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. + /// + public async Task 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}"); + } + } + + /// + /// 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. + /// + public async Task 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; + } + + /// + /// Pulls the latest changes from the remote. Returns true on success. + /// + public async Task 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; + } + + /// + /// Commits and pushes all changes in the tenant's clone. Used after file + /// edits via the source code API to persist changes to Forgejo. + /// + 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"); + } + + /// The absolute path to a tenant's local clone directory. + public string TenantCloneDir(string tenantId) + => Path.GetFullPath(Path.Combine(_copiesRoot, Sanitize(tenantId))); + + /// + /// Seeds the local clone with template files from the shared directory. + /// Only runs when the repo is empty (no workflow YAML files yet). + /// + 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 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 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; + } +} diff --git a/Services/PerTenantWorkflowSource.cs b/Services/PerTenantWorkflowSource.cs index 4673fd3..c69d859 100644 --- a/Services/PerTenantWorkflowSource.cs +++ b/Services/PerTenantWorkflowSource.cs @@ -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. /// +/// Two modes (automatic, config-driven): +/// +/// Forgejo-backed (preferred): when Forgejo:AdminToken is +/// configured, each tenant's workflow files live in a private Forgejo repo +/// (workflows-{tenantId}). The factory clones/pulls via +/// and returns a +/// backed by the local clone. +/// This gives tenants real git history, branching, and remote sync. +/// Filesystem-only (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. +/// +/// /// Tenant isolation is mandatory: WorkflowSource:CopiesRoot 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(); + var forgejoSvc = new ForgejoWorkflowRepoService(config, forgejoLogger); + _forgejo = forgejoSvc.IsConfigured ? forgejoSvc : null; + + if (_forgejo != null) + _loggers.CreateLogger() + .LogInformation("Workflow source: Forgejo-backed mode (owner={Owner})", + config["Forgejo:Owner"] ?? config["Forgejo:WorkflowRepoOwner"]); + else + _loggers.CreateLogger() + .LogInformation("Workflow source: filesystem-only mode (no Forgejo admin token)"); } - /// Creates a per-tenant workflow source. The tenant directory + /// True when Forgejo-backed mode is active. + public bool IsForgejoBacked => _forgejo != null; + + /// The Forgejo repo service (null when in filesystem-only mode). + public ForgejoWorkflowRepoService? Forgejo => _forgejo; + + /// + /// 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 /// {CopiesRoot}/{sanitizedTenantId}/ is created (and seeded from - /// the shared template dir) on first access. + /// the shared template dir) on first access. + /// public IWorkflowSource Create(string tenantId) { var safeId = Sanitize(tenantId); @@ -58,6 +103,63 @@ public sealed class WorkflowSourceFactory return new PerTenantWorkflowSource(tenantId, tenantDir, sharedDir, logger); } + /// + /// 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 . + /// + public async Task 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() + .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(); + return new PerTenantWorkflowSource(tenantId, cloneDir, _sharedDir, logger); + } + + /// + /// Resolves the Forgejo login for a tenant (forgejo_id) from the auth_users table. + /// + public async Task 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() + .LogWarning(ex, "Could not resolve Forgejo login for tenant {TenantId}", tenantId); + return null; + } + } + private static string Sanitize(string id) { if (string.IsNullOrWhiteSpace(id)) diff --git a/Services/Runs/TaskDispatcher.cs b/Services/Runs/TaskDispatcher.cs index 1611aa7..19a8f0e 100644 --- a/Services/Runs/TaskDispatcher.cs +++ b/Services/Runs/TaskDispatcher.cs @@ -10,26 +10,29 @@ namespace w4c_workflows.Services.Runs; /// 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 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). /// 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 _logger; public TaskDispatcher( WorkflowsDbContext db, IJobQueue jobs, IConfiguration config, - ILogger logger) + ILogger logger, + WorkflowSourceFactory? sourceFactory = null) { _db = db; _jobs = jobs; _logger = logger; - _workerRoot = ResolveWorkerRoot(config); + _sourceFactory = sourceFactory; + _defaultWorkerRoot = ResolveWorkerRoot(config); } public async Task 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(fields) { @@ -91,12 +96,41 @@ public class TaskDispatcher 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. + /// + 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)); + } + + /// + /// 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 _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) diff --git a/Services/WorkflowSyncService.cs b/Services/WorkflowSyncService.cs index 4205f8b..7c0d24e 100644 --- a/Services/WorkflowSyncService.cs +++ b/Services/WorkflowSyncService.cs @@ -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 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. /// public class WorkflowSyncService { private readonly WorkflowsDbContext _db; private readonly WorkflowCompiler _compiler; private readonly IWorkflowSource _source; + private readonly WorkflowSourceFactory? _sourceFactory; private readonly ILogger _logger; public WorkflowSyncService( WorkflowsDbContext db, WorkflowCompiler compiler, IWorkflowSource source, - ILogger logger) + ILogger logger, + WorkflowSourceFactory? sourceFactory = null) { _db = db; _compiler = compiler; _source = source; _logger = logger; + _sourceFactory = sourceFactory; } public async Task 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); diff --git a/appsettings.json b/appsettings.json index 39b5877..888ae5d 100644 --- a/appsettings.json +++ b/appsettings.json @@ -59,7 +59,10 @@ }, "Forgejo": { "BaseUrl": "https://forgejo.wiz4chat.com", - "AccessToken": "" + "AccessToken": "043a13f8ee662d1ad96ca0d3018ab17907ec3470", + "AdminToken": "8d811a81579747f0d9b2e54c9068edd49618d15c", + "Owner": "test", + "WorkflowRepoOwner": "" }, "SourceCode": { "RepoRoot": ""