w4c-workflows-api/Program.cs

using Microsoft.EntityFrameworkCore;
using Serilog;
using Serilog.Events;
using Serilog.Sinks.OpenSearch;
using StackExchange.Redis;
using Scalar.AspNetCore;
using w4c_workflows.Data;
using w4c_workflows.Middleware;
using w4c_workflows.Services;
using w4c_workflows.Services.Execution;
using w4c_workflows.Services.Messaging;
using w4c_workflows.Services.Runs;
using w4c_workflows.Services.Triggers;

// ---------------------------------------------------------------------------
// w4c-workflows-api — control plane for the custom workflows engine.
//
// Two entrypoints share this image:
//   - default        : HTTP control plane (API, trigger scheduler, Mermaid, keys)
//   - --worker       : per-tenant execution worker (pull/exec/result over Redis)
// ---------------------------------------------------------------------------

var runWorker = args.Any(a => a == "--worker");

var builder = WebApplication.CreateBuilder(args);

var openSearchUrl = builder.Configuration["OpenSearch:Url"] ?? "http://localhost:9200";
var appName = builder.Configuration["APP_NAME"] ?? (runWorker ? "workflows-worker" : "workflows-api");

Log.Logger = new LoggerConfiguration()
    .MinimumLevel.Information()
    .MinimumLevel.Override("Microsoft.AspNetCore", LogEventLevel.Warning)
    // EF Core logs the raw SQL ("Executed DbCommand ...") at Information for
    // EVERY query. The background pollers (run lifecycle, trigger scheduler,
    // handler consumer) fire those several times a second and flood the console
    // / OpenSearch. Keep our own Information logs, but drop the EF command log.
    .MinimumLevel.Override("Microsoft.EntityFrameworkCore.Database.Command", LogEventLevel.Warning)
    .Enrich.FromLogContext()
    .Enrich.WithProperty("App", appName)
    .WriteTo.Console()
    .WriteTo.OpenSearch(new OpenSearchSinkOptions(new Uri(openSearchUrl))
    {
        IndexFormat = "w4c-workflows-{0:yyyy.MM.dd}",
        AutoRegisterTemplate = false,
        BatchAction = OpenOpType.Create,
        EmitEventFailure = EmitEventFailureHandling.WriteToSelfLog,
        BufferBaseFilename = Path.Combine(Path.GetTempPath(), "serilog-opensearch-buffer"),
    })
    .CreateLogger();

builder.Host.UseSerilog();

builder.Services.AddControllers();
builder.Services.AddOpenApi();
builder.Services.AddHttpClient();
builder.Services.AddHttpContextAccessor();

// 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
// StopHost and a single early exception prevents the API from starting.
builder.Services.Configure<HostOptions>(options =>
    options.BackgroundServiceExceptionBehavior = BackgroundServiceExceptionBehavior.Ignore);

builder.Services.AddDbContext<WorkflowsDbContext>(options =>
    options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection")));

builder.Services.AddSingleton<IConnectionMultiplexer>(_ =>
    ConnectionMultiplexer.Connect(
        builder.Configuration["Redis:ConnectionString"] ?? "localhost:6379,abortConnect=false"));

builder.Services.AddScoped<ApiKeyService>();

// Transport + lease: Redis Streams now, RabbitMQ later behind the same interfaces.
builder.Services.AddSingleton<RedisStreamsTransport>();
builder.Services.AddSingleton<IJobQueue>(sp => sp.GetRequiredService<RedisStreamsTransport>());
builder.Services.AddSingleton<IEventBus>(sp => sp.GetRequiredService<RedisStreamsTransport>());
builder.Services.AddSingleton<ILeaseService, LeaseService>();

// YAML compile pipeline (step 4).
builder.Services.AddSingleton<LanguageRegistry>();
builder.Services.AddSingleton<WorkflowValidator>();
builder.Services.AddSingleton<WorkflowCompiler>();

// Git sync (step 5): read workflows/*.yaml from the local monorepo checkout.
builder.Services.AddSingleton<IWorkflowSource, LocalRepoWorkflowSource>();
builder.Services.AddScoped<WorkflowSyncService>();

// Mermaid diagrams (step 6).
builder.Services.AddSingleton<MermaidGeneratorService>();

// Rich HTML workflow preview (backend-rendered, embeddable in an iframe).
builder.Services.AddSingleton<WorkflowHtmlRenderer>();
builder.Services.AddSingleton<RenderTokenService>();

// Runtime registry + worker executors (step 7).
// The executors map each language to its execution strategy; the registry
// (which also backs GET /api/languages) reports per-runtime availability.
builder.Services.AddSingleton<IScriptExecutor>(new SubprocessScriptExecutor("shell", "sh", builder.Configuration));
builder.Services.AddSingleton<IScriptExecutor>(new SubprocessScriptExecutor("python", "python3", builder.Configuration));
builder.Services.AddSingleton<IScriptExecutor>(new SubprocessScriptExecutor("javascript", "node", builder.Configuration));
builder.Services.AddSingleton<IScriptExecutor>(new TypeScriptExecutor(builder.Configuration));
builder.Services.AddSingleton<IScriptExecutor>(new CSharpScriptExecutor());
// W9: `agent` step type — invokes an LLM agent via chatapi over HTTP (D3), keeping
// workflows-api decoupled from BotSharp.
builder.Services.AddSingleton<IScriptExecutor>(sp =>
    new AgentScriptExecutor(
        sp.GetRequiredService<IHttpClientFactory>(),
        builder.Configuration,
        sp.GetRequiredService<ILogger<AgentScriptExecutor>>()));
builder.Services.AddSingleton<RuntimeRegistry>();

// S8: remote (SSH) task execution via the w4c-webapi server-console exec endpoint.
// Used only when a task carries a `server` reference; local subprocess stays the default.
builder.Services.AddSingleton<RemoteServerExecutor>();

// Trigger engine (step 8): the scheduler + handler consumer run only in the
// 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<ITriggerState, RedisTriggerState>();
builder.Services.AddScoped<IRunLauncher, RunLauncher>();

// Run lifecycle engine (step 9): dispatch pending runs + consume task.results +
// advance the nextId chain + durable checkpoint/resume.
builder.Services.AddSingleton<RunLifecycleEngine>();
builder.Services.AddScoped<TaskDispatcher>();
builder.Services.AddScoped<DurableStateStore>();

if (runWorker)
{
    // Dedicated worker process (scaled/separate deployment).
    builder.Services.AddHostedService<WorkerHostService>();
}
else
{
    // Control-plane entrypoint. In single-node / self-hosted deployments one
    // process does both: it accounts, dispatches, AND runs the worker loop so
    // dispatched jobs actually execute. This closes the gap where the API
    // created run + task.run jobs but nothing consumed them, leaving every run
    // 'running' forever. A dedicated --worker can still be used for scale-out.
    builder.Services.AddHostedService<WorkerHostService>();
    builder.Services.AddHostedService<TriggerScheduler>();
    builder.Services.AddHostedService<HandlerStreamConsumer>();
    builder.Services.AddHostedService<RunLifecycleService>();
}

builder.Services.AddCors(options =>
{
    options.AddDefaultPolicy(policy =>
    {
        var origins = builder.Configuration.GetSection("Cors:Origins").Get<string[]>();
        if (origins is { Length: > 0 })
        {
            // Production: restrict to explicitly configured origins.
            policy.WithOrigins(origins)
                  .AllowAnyMethod()
                  .AllowAnyHeader()
                  .AllowCredentials();
        }
        else
        {
            // Dev fallback: no origins configured → allow all.
            policy.AllowAnyOrigin()
                  .AllowAnyMethod()
                  .AllowAnyHeader();
        }
    });
});

var app = builder.Build();

app.UseCors();
app.UseMiddleware<AuthMiddleware>();

// 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.
try
{
    using var scope = app.Services.CreateScope();
    var db = scope.ServiceProvider.GetRequiredService<WorkflowsDbContext>();
    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;");
}
catch (Exception ex)
{
    Log.Logger.Error(ex, "Failed to apply EF Core migrations on startup");
}

app.MapOpenApi();
app.MapScalarApiReference();
app.MapControllers();

if (runWorker)
{
    Log.Logger.Information("w4c-workflows-worker entrypoint selected; worker loop running for tenant {TenantId}",
        builder.Configuration["Workflows:TenantId"] ?? Environment.GetEnvironmentVariable("WF_TENANT_ID") ?? "default");
}

app.Run();

Log.CloseAndFlush();