OPC/ControlPlane.Api/Endpoints/ProvisioningEndpoints.cs

using ControlPlane.Api.Services;
using ControlPlane.Core.Messages;
using ControlPlane.Core.Models;
using ControlPlane.Core.Services;
using MassTransit;
using System.Text.Json;

namespace ControlPlane.Api.Endpoints;

public static class ProvisioningEndpoints
{
    private static readonly JsonSerializerOptions JsonOpts = new(JsonSerializerDefaults.Web);

    public static IEndpointRouteBuilder MapProvisioningEndpoints(this IEndpointRouteBuilder app)
    {
        var group = app.MapGroup("/api/provision").WithTags("Provisioning");

        group.MapPost("/", QueueProvisioningJob);
        group.MapGet("/{id:guid}", GetJobStatus);
        group.MapGet("/{id:guid}/stream", StreamJobEvents);

        app.MapGet("/api/tenants", GetTenants).WithTags("Tenants");

        return app;
    }

    private static async Task<IResult> QueueProvisioningJob(
        ProvisioningRequest request,
        Dictionary<Guid, ProvisioningJob> jobs,
        IPublishEndpoint bus)
    {
        var job = new ProvisioningJob
        {
            ClientName  = request.ClientName,
            StateCode   = request.StateCode.ToUpperInvariant(),
            Subdomain   = request.Subdomain,
            AdminEmail  = request.AdminEmail,
            SiteCode    = request.SiteCode,
            Environment = request.Environment,
            Tier        = request.Tier,
            Status      = ProvisioningStatus.Pending
        };

        jobs[job.Id] = job;

        await bus.Publish(new ProvisionClientCommand
        {
            JobId       = job.Id,
            ClientName  = job.ClientName,
            StateCode   = job.StateCode,
            Subdomain   = job.Subdomain,
            AdminEmail  = job.AdminEmail,
            SiteCode    = job.SiteCode,
            Environment = job.Environment,
            Tier        = job.Tier
        });

        return Results.Accepted($"/api/provision/{job.Id}", new { job.Id, job.Status });
    }

    private static IResult GetJobStatus(Guid id, Dictionary<Guid, ProvisioningJob> jobs) =>
        jobs.TryGetValue(id, out var job) ? Results.Ok(job) : Results.NotFound();

    private static IResult GetTenants(TenantRegistryService registry) =>
        Results.Ok(registry.GetAll());

    private static async Task StreamJobEvents(
        Guid id,
        SseEventBus bus,
        Dictionary<Guid, ProvisioningJob> jobs,
        HttpContext ctx,
        CancellationToken cancellationToken)
    {
        if (!jobs.ContainsKey(id))
        {
            ctx.Response.StatusCode = 404;
            return;
        }

        ctx.Response.Headers.ContentType  = "text/event-stream";
        ctx.Response.Headers.CacheControl = "no-cache";
        ctx.Response.Headers.Connection   = "keep-alive";

        var channel = bus.Subscribe(id);
        try
        {
            await foreach (var evt in channel.Reader.ReadAllAsync(cancellationToken))
            {
                var json = JsonSerializer.Serialize(evt, JsonOpts);
                await ctx.Response.WriteAsync($"data: {json}\n\n", cancellationToken);
                await ctx.Response.Body.FlushAsync(cancellationToken);

                if (evt.Type is "job_complete" or "job_failed") break;
            }
        }
        catch (OperationCanceledException)
        {
            // Client disconnected (e.g. browser refresh) — not an error.
        }
        finally
        {
            bus.Unsubscribe(id, channel);
        }
    }
}