From 6a3f702e09832048109ef906969d97fb03956620 Mon Sep 17 00:00:00 2001 From: Vitali sharp8n Date: Tue, 1 Sep 2026 22:31:27 +0300 Subject: [PATCH] worfklow integration per tenant support --- .gitignore | 3 + Program.cs | 88 ++++++-- Services/InMemoryLeaseService.cs | 38 ++++ Services/Messaging/InMemoryTransport.cs | 129 +++++++++++ Services/PerTenantWorkflowSource.cs | 248 ++++++++++++++++++++++ Services/Triggers/InMemoryTriggerState.cs | 26 +++ appsettings.json | 4 + w4c-workflows-api.csproj | 2 + 8 files changed, 515 insertions(+), 23 deletions(-) create mode 100644 Services/InMemoryLeaseService.cs create mode 100644 Services/Messaging/InMemoryTransport.cs create mode 100644 Services/PerTenantWorkflowSource.cs create mode 100644 Services/Triggers/InMemoryTriggerState.cs diff --git a/.gitignore b/.gitignore index c200be7..fe6bcd7 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,6 @@ x86/ # Docker .env + +# Per-tenant workflow copies (WorkflowSource:CopiesRoot) +.data/ \ No newline at end of file diff --git a/Program.cs b/Program.cs index ad6d171..f44e9d8 100644 --- a/Program.cs +++ b/Program.cs @@ -55,6 +55,11 @@ builder.Services.AddOpenApi(); builder.Services.AddHttpClient(); builder.Services.AddHttpContextAccessor(); +// Lite mode flag: when true, the engine uses SQLite + in-process channels +// instead of PostgreSQL + Redis. Intended for self-hosted single-container +// deployments where external infrastructure is not available. +var liteMode = builder.Configuration.GetValue("UseLiteMode"); + // A transient failure in ONE background service (worker loop, trigger scheduler, // run lifecycle, handler consumer) must never take the whole control plane down. // Without this, HostOptions.BackgroundServiceExceptionBehavior defaults to @@ -62,28 +67,66 @@ builder.Services.AddHttpContextAccessor(); builder.Services.Configure(options => options.BackgroundServiceExceptionBehavior = BackgroundServiceExceptionBehavior.Ignore); -builder.Services.AddDbContext(options => - options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); +if (liteMode) +{ + // Lite mode: SQLite. The connection string should be something like + // "Data Source=/app/data/workflows.db". + var sqliteCs = builder.Configuration.GetConnectionString("DefaultConnection") + ?? "Data Source=workflows.db"; + builder.Services.AddDbContext(options => + options.UseSqlite(sqliteCs)); + Log.Logger.Information("Lite mode: using SQLite ({ConnectionString})", sqliteCs); +} +else +{ + builder.Services.AddDbContext(options => + options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); -builder.Services.AddSingleton(_ => - ConnectionMultiplexer.Connect( - builder.Configuration["Redis:ConnectionString"] ?? "localhost:6379,abortConnect=false")); + builder.Services.AddSingleton(_ => + ConnectionMultiplexer.Connect( + builder.Configuration["Redis:ConnectionString"] ?? "localhost:6379,abortConnect=false")); +} builder.Services.AddScoped(); -// Transport + lease: Redis Streams now, RabbitMQ later behind the same interfaces. -builder.Services.AddSingleton(); -builder.Services.AddSingleton(sp => sp.GetRequiredService()); -builder.Services.AddSingleton(sp => sp.GetRequiredService()); -builder.Services.AddSingleton(); +// Transport + lease. +if (liteMode) +{ + // Lite mode: in-process channels replace Redis Streams. + builder.Services.AddSingleton(); + builder.Services.AddSingleton(sp => sp.GetRequiredService()); + builder.Services.AddSingleton(sp => sp.GetRequiredService()); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + Log.Logger.Information("Lite mode: using in-memory transport + trigger state (no Redis)"); +} +else +{ + // Full mode: Redis Streams. + builder.Services.AddSingleton(); + builder.Services.AddSingleton(sp => sp.GetRequiredService()); + builder.Services.AddSingleton(sp => sp.GetRequiredService()); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); +} // YAML compile pipeline (step 4). builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); -// Git sync (step 5): read workflows/*.yaml from the local monorepo checkout. -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. +builder.Services.AddSingleton(); +builder.Services.AddScoped(sp => +{ + var http = sp.GetRequiredService(); + var factory = sp.GetRequiredService(); + var tenantId = http.HttpContext?.Items["TenantId"] as string + ?? throw new InvalidOperationException("TenantId not resolved by auth middleware"); + return factory.Create(tenantId); +}); builder.Services.AddScoped(); // Mermaid diagrams (step 6). @@ -118,7 +161,6 @@ builder.Services.AddSingleton(); // control-plane entrypoint, never in the worker. They share the run launcher — // the seam that creates pending runs for the run lifecycle engine (step 9). builder.Services.AddSingleton(TimeProvider.System); -builder.Services.AddSingleton(); builder.Services.AddScoped(); // Run lifecycle engine (step 9): dispatch pending runs + consume task.results + @@ -173,9 +215,7 @@ var app = builder.Build(); app.UseCors(); app.UseMiddleware(); -// Ensure the `workflows` schema exists before serving traffic (mirrors the -// resilient "ensure schema" pattern from w4c-auth). A migration failure must -// not crash the host — health and logs still report the degraded state. +// Ensure the database schema exists before serving traffic. try { using var scope = app.Services.CreateScope(); @@ -183,13 +223,15 @@ try await db.Database.MigrateAsync(); // Additive schema, applied idempotently (raw SQL, no new EF migration) so it is - // safe on every startup against both freshly-created and long-lived databases: - // • "ArchivedAt" — W7 soft-delete so sync can stop wiping historical TaskRuns. - // • "Server" — S8 target managed-server for remote (SSH) task execution. - await db.Database.ExecuteSqlRawAsync( - "ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"ArchivedAt\" timestamptz NULL;"); - await db.Database.ExecuteSqlRawAsync( - "ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"Server\" text NULL;"); + // safe on every startup against both freshly-created and long-lived databases. + // Only applies to PostgreSQL (SQLite handles columns automatically via EF). + if (!liteMode) + { + await db.Database.ExecuteSqlRawAsync( + "ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"ArchivedAt\" timestamptz NULL;"); + await db.Database.ExecuteSqlRawAsync( + "ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"Server\" text NULL;"); + } } catch (Exception ex) { diff --git a/Services/InMemoryLeaseService.cs b/Services/InMemoryLeaseService.cs new file mode 100644 index 0000000..4690d41 --- /dev/null +++ b/Services/InMemoryLeaseService.cs @@ -0,0 +1,38 @@ +using System.Collections.Concurrent; + +namespace w4c_workflows.Services; + +/// +/// In-memory lease service for lite/self-hosted mode where Redis is not available. +/// Uses a ConcurrentDictionary with TTL tracking. Not distributed — only works +/// for a single-process deployment (which is the lite mode use case). +/// +public sealed class InMemoryLeaseService : ILeaseService +{ + private record Lease(string Owner, DateTime ExpiresAt); + + private readonly ConcurrentDictionary _leases = new(); + + public Task AcquireAsync(string tenantId, string runId, string owner, TimeSpan ttl) + { + var key = $"{tenantId}:{runId}"; + var lease = new Lease(owner, DateTime.UtcNow + ttl); + return Task.FromResult(_leases.TryAdd(key, lease)); + } + + public Task RenewAsync(string tenantId, string runId, string owner, TimeSpan ttl) + { + var key = $"{tenantId}:{runId}"; + if (!_leases.TryGetValue(key, out var current) || current.Owner != owner) + return Task.FromResult(false); + _leases[key] = new Lease(owner, DateTime.UtcNow + ttl); + return Task.FromResult(true); + } + + public Task ReleaseAsync(string tenantId, string runId, string owner) + { + var key = $"{tenantId}:{runId}"; + // Remove only if the owner still matches (compare-and-delete). + return Task.FromResult(_leases.TryRemove(key, out var lease) && lease.Owner == owner); + } +} \ No newline at end of file diff --git a/Services/Messaging/InMemoryTransport.cs b/Services/Messaging/InMemoryTransport.cs new file mode 100644 index 0000000..5dd7cf6 --- /dev/null +++ b/Services/Messaging/InMemoryTransport.cs @@ -0,0 +1,129 @@ +using System.Collections.Concurrent; +using System.Threading.Channels; + +namespace w4c_workflows.Services.Messaging; + +/// +/// In-process channel-based transport for IJobQueue and IEventBus. +/// Used in lite/self-hosted mode where Redis is not available. Store-backed +/// (ConcurrentDictionary of channels per stream), survives thread-boundary +/// handoffs but NOT process restarts. For self-hosted single-process deployment +/// this is acceptable (restart = redeliver pending jobs from DB state). +/// +public sealed class InMemoryTransport : IJobQueue, IEventBus +{ + private readonly ConcurrentDictionary> _channels = new(); + private long _idSeq; + + // ---- IJobQueue ---- + + public Task EnsureGroupAsync(string tenantId, CancellationToken ct) + => Task.CompletedTask; // No-op: channel consumers are implicit. + + public Task EnqueueAsync(string tenantId, IReadOnlyDictionary fields, CancellationToken ct) + { + var stream = $"wf:{tenantId}:jobs"; + var id = $"0-{Interlocked.Increment(ref _idSeq)}"; + var msg = new StreamMessage(id, fields); + GetOrAddChannel(stream).Writer.TryWrite(msg); + return Task.FromResult(id); + } + + public Task> ReadGroupAsync(string tenantId, string consumer, int count, CancellationToken ct) + => DrainAsync($"wf:{tenantId}:jobs", count, ct); + + public Task> ClaimPendingAsync(string tenantId, string consumer, TimeSpan minIdle, int count, CancellationToken ct) + => Task.FromResult>(Array.Empty()); // No pending in memory. + + public Task AckAsync(string tenantId, string messageId, CancellationToken ct) + => Task.FromResult(1L); // Ack is a no-op in memory. + + public Task DeadLetterAsync(string tenantId, string messageId, IReadOnlyDictionary fields, string reason, CancellationToken ct) + { + var dlq = $"wf:{tenantId}:dlq"; + var id = $"0-{Interlocked.Increment(ref _idSeq)}"; + var dlqFields = new Dictionary(fields) { ["deadLetterReason"] = reason }; + GetOrAddChannel(dlq).Writer.TryWrite(new StreamMessage(id, dlqFields)); + return Task.CompletedTask; + } + + public Task PublishDeadLetterAsync(string tenantId, IReadOnlyDictionary fields, string reason, CancellationToken ct) + { + var dlq = $"wf:{tenantId}:dlq"; + var id = $"0-{Interlocked.Increment(ref _idSeq)}"; + var dlqFields = new Dictionary(fields) { ["deadLetterReason"] = reason }; + GetOrAddChannel(dlq).Writer.TryWrite(new StreamMessage(id, dlqFields)); + return Task.CompletedTask; + } + + // ---- IEventBus ---- + + public Task PublishAsync(string tenantId, string stream, IReadOnlyDictionary fields, CancellationToken ct) + { + var id = $"0-{Interlocked.Increment(ref _idSeq)}"; + var msg = new StreamMessage(id, fields); + // Expand {tenant} placeholder. + var resolved = stream.Replace("{tenant}", tenantId); + GetOrAddChannel(resolved).Writer.TryWrite(msg); + return Task.FromResult(id); + } + + Task IEventBus.EnsureGroupAsync(string tenantId, string stream, CancellationToken ct) + => Task.CompletedTask; + + Task> IEventBus.ReadGroupAsync(string tenantId, string stream, string consumer, int count, CancellationToken ct) + { + var resolved = stream.Replace("{tenant}", tenantId); + return DrainAsync(resolved, count, ct); + } + + Task> IEventBus.ClaimPendingAsync(string tenantId, string stream, string consumer, TimeSpan minIdle, int count, CancellationToken ct) + => Task.FromResult>(Array.Empty()); + + Task IEventBus.AckAsync(string tenantId, string stream, string messageId, CancellationToken ct) + => Task.FromResult(1L); + + // ---- Helpers ---- + + private Channel GetOrAddChannel(string key) + => _channels.GetOrAdd(key, _ => Channel.CreateUnbounded( + new UnboundedChannelOptions { SingleReader = false, SingleWriter = false })); + + private async Task> DrainAsync(string stream, int count, CancellationToken ct) + { + if (!_channels.TryGetValue(stream, out var ch)) + return Array.Empty(); + + var result = new List(count); + while (result.Count < count) + { + try + { + if (ch.Reader.TryRead(out var msg)) + { + result.Add(msg); + } + else + { + // Wait briefly for a message. + using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + cts.CancelAfter(TimeSpan.FromMilliseconds(50)); + try + { + var read = await ch.Reader.ReadAsync(cts.Token); + result.Add(read); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + break; // No more messages available right now. + } + } + } + catch (OperationCanceledException) + { + break; + } + } + return result; + } +} \ No newline at end of file diff --git a/Services/PerTenantWorkflowSource.cs b/Services/PerTenantWorkflowSource.cs new file mode 100644 index 0000000..4673fd3 --- /dev/null +++ b/Services/PerTenantWorkflowSource.cs @@ -0,0 +1,248 @@ +using System.Diagnostics; + +namespace w4c_workflows.Services; + +/// +/// Factory for creating instances scoped to a +/// specific tenant. The caller provides the tenant ID explicitly (from auth +/// middleware); the factory resolves directories and returns a source bound +/// to that tenant's workflow directory. +/// +/// 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. +/// +/// Registered as a Singleton (shared config, no per-request state); +/// the created sources are lightweight and backed by filesystem operations. +/// +public sealed class WorkflowSourceFactory +{ + private readonly IConfiguration _config; + private readonly IWebHostEnvironment _env; + private readonly ILoggerFactory _loggers; + private readonly string _copiesRoot; + private readonly string _sharedDir; + private readonly string _repoRoot; + + public WorkflowSourceFactory(IConfiguration config, IWebHostEnvironment env, ILoggerFactory loggers) + { + _config = config; + _env = env; + _loggers = loggers; + + _copiesRoot = config["WorkflowSource:CopiesRoot"] + ?? throw new InvalidOperationException( + "WorkflowSource:CopiesRoot is not configured. " + + "Per-tenant filesystem isolation is required — set it to a writable " + + "directory path (e.g. \"/data/workflow-tenants\" or \".data/workflow-tenants\")."); + _sharedDir = config["WorkflowSource:SharedDir"] ?? "workflows"; + _repoRoot = ResolveRepoRoot(config, env); + } + + /// Creates a per-tenant workflow source. The tenant directory + /// {CopiesRoot}/{sanitizedTenantId}/ is created (and seeded from + /// the shared template dir) on first access. + public IWorkflowSource Create(string tenantId) + { + var safeId = Sanitize(tenantId); + var tenantDir = Path.GetFullPath(Path.Combine(_copiesRoot, safeId)); + + var sharedRel = _sharedDir; + string sharedDir; + if (Path.IsPathRooted(sharedRel) && Directory.Exists(sharedRel)) + sharedDir = Path.GetFullPath(sharedRel); + else + sharedDir = Path.GetFullPath(Path.Combine(_repoRoot, sharedRel)); + + var logger = _loggers.CreateLogger(); + return new PerTenantWorkflowSource(tenantId, tenantDir, sharedDir, logger); + } + + 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; + } + + private static string ResolveRepoRoot(IConfiguration config, IWebHostEnvironment env) + { + var configured = config["SourceCode:RepoRoot"]; + if (!string.IsNullOrWhiteSpace(configured) && Directory.Exists(configured)) + return Path.GetFullPath(configured); + + var start = AppContext.BaseDirectory; + if (!string.IsNullOrEmpty(env.ContentRootPath) && Directory.Exists(env.ContentRootPath)) + start = env.ContentRootPath; + + 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(Directory.GetCurrentDirectory()); + } +} + +/// +/// Per-tenant workflow YAML source with directory isolation. Each tenant gets +/// its own directory at {CopiesRoot}/{tenantId}/, initialized from the +/// shared template directory on first access. +/// +public sealed class PerTenantWorkflowSource : IWorkflowSource +{ + private readonly string _tenantDir; + private readonly string _sharedDir; + private readonly ILogger _logger; + + public PerTenantWorkflowSource(string tenantId, string tenantDir, string sharedDir, ILogger logger) + { + _tenantDir = tenantDir; + _sharedDir = sharedDir; + _logger = logger; + _logger.LogInformation( + "Per-tenant workflow source: tenant={TenantId} dir={Dir} shared={Shared}", + tenantId, _tenantDir, _sharedDir); + } + + public Task> ListAsync(CancellationToken ct) + { + ct.ThrowIfCancellationRequested(); + EnsureInitialized(); + return Task.FromResult(ListYamlFiles(_tenantDir)); + } + + public async Task ReadAsync(string path, CancellationToken ct) + { + EnsureInitialized(); + return await File.ReadAllTextAsync( + Path.Combine(_tenantDir, path.Replace('/', Path.DirectorySeparatorChar)), ct); + } + + public WorkflowSourceState GetState() + { + EnsureInitialized(); + var head = RunGit(_tenantDir, "rev-parse", "HEAD"); + var status = RunGit(_tenantDir, "status", "--porcelain"); + return new WorkflowSourceState( + string.IsNullOrWhiteSpace(head) ? null : head, + !string.IsNullOrEmpty(status)); + } + + /// + /// Ensures the tenant directory exists and has been seeded from the shared + /// template directory (first-access initialization). + /// + private void EnsureInitialized() + { + if (Directory.Exists(_tenantDir)) + return; + + _logger.LogInformation("Initializing tenant workflow directory {Dir} from {Shared}", _tenantDir, _sharedDir); + Directory.CreateDirectory(_tenantDir); + + if (!Directory.Exists(_sharedDir)) + { + _logger.LogWarning("Shared workflow directory {Shared} does not exist; tenant starts empty", _sharedDir); + return; + } + + CopyYamlFiles(_sharedDir, _tenantDir, _logger); + } + + private static IReadOnlyList ListYamlFiles(string dir) + { + var files = new List(); + if (!Directory.Exists(dir)) + return files; + + foreach (var full in Directory.EnumerateFiles(dir, "*", SearchOption.AllDirectories)) + { + var ext = Path.GetExtension(full); + if (ext is not (".yaml" or ".yml")) + continue; + // Skip .git + var rel = Path.GetRelativePath(dir, full); + if (rel.StartsWith(".git", StringComparison.OrdinalIgnoreCase)) + continue; + files.Add(new WorkflowFile(rel.Replace(Path.DirectorySeparatorChar, '/'))); + } + + return files.OrderBy(f => f.Path, StringComparer.Ordinal).ToList(); + } + + /// + /// Copies YAML files from src to dst, skipping .git and non-yaml files. + /// + private static void CopyYamlFiles(string src, string dst, ILogger logger) + { + foreach (var file in Directory.EnumerateFiles(src, "*", SearchOption.AllDirectories)) + { + var ext = Path.GetExtension(file); + if (ext is not (".yaml" or ".yml")) + continue; + + 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 ex) + { + logger.LogWarning(ex, "Failed to copy template file {File}", rel); + } + } + } + + 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; + } + } +} diff --git a/Services/Triggers/InMemoryTriggerState.cs b/Services/Triggers/InMemoryTriggerState.cs new file mode 100644 index 0000000..50fe130 --- /dev/null +++ b/Services/Triggers/InMemoryTriggerState.cs @@ -0,0 +1,26 @@ +using System.Collections.Concurrent; + +namespace w4c_workflows.Services.Triggers; + +/// +/// In-memory trigger state for lite/self-hosted mode where Redis is not available. +/// Survives only for the process lifetime (restart = re-fire once, which is safe +/// because delivery is at-least-once and the run lifecycle dedups via correlation id). +/// +public sealed class InMemoryTriggerState : ITriggerState +{ + private readonly ConcurrentDictionary _state = new(); + + public Task GetLastFireAsync(string tenantId, Guid workflowId, CancellationToken ct) + { + var key = $"{tenantId}:{workflowId}"; + return Task.FromResult(_state.TryGetValue(key, out var ts) ? (DateTimeOffset?)ts : null); + } + + public Task SetLastFireAsync(string tenantId, Guid workflowId, DateTimeOffset at, CancellationToken ct) + { + var key = $"{tenantId}:{workflowId}"; + _state[key] = at; + return Task.CompletedTask; + } +} \ No newline at end of file diff --git a/appsettings.json b/appsettings.json index 0c6e436..39b5877 100644 --- a/appsettings.json +++ b/appsettings.json @@ -63,5 +63,9 @@ }, "SourceCode": { "RepoRoot": "" + }, + "WorkflowSource": { + "CopiesRoot": ".data/workflow-tenants", + "SharedDir": "workflows" } } diff --git a/w4c-workflows-api.csproj b/w4c-workflows-api.csproj index 833ec9b..addfd2e 100644 --- a/w4c-workflows-api.csproj +++ b/w4c-workflows-api.csproj @@ -26,6 +26,8 @@ all + +