Generated Code📋

// Generated: CreateZipOrchestratorKernel.g.cs
public abstract class CreateZipOrchestratorKernel
{
    // ── Step execution with retry, progress tracking, and audit ──
    protected async Task<StepResult> ExecuteStepWithPoliciesAsync<TStepData>(
        SagaContext<CreateZipRequest> context,
        string stepName,
        int stepOrder,
        Func<SagaContext<CreateZipRequest>, TStepData, CancellationToken, Task<StepResult>> execute,
        IAsyncPolicy? retryPolicy,
        CancellationToken ct)
        where TStepData : new();

    // ── Compensation: reverse-order execution of completed steps ──
    protected async Task CompensateAsync(
        SagaContext<CreateZipRequest> context,
        CancellationToken ct);

    // ── Audit trail ──
    protected async Task AuditAsync(
        SagaContext<CreateZipRequest> context,
        string stepName,
        string action,
        string? detail = null);

    // ── Per-step compensation (virtual, built-in for FileUploadStep) ──
    protected virtual Task CompensateUploadSourceFilesAsync(/*...*/);
    protected virtual Task CompensateCreateZipArchiveAsync(/*...*/);
    protected virtual Task CompensateUploadZipAsync(/*...*/);

    // ── Developer must implement ──
    protected abstract Task<StepResult> ExecuteCreateZipArchiveAsync(
        SagaContext<CreateZipRequest> context,
        CreateZipArchiveStepData stepData,
        CancellationToken ct);

    protected abstract CreateZipResponse MapResponse(
        SagaContext<CreateZipRequest> context);
}

// Generated: CreateZipApiOrchestrator.g.cs
public class CreateZipApiOrchestrator : CreateZipOrchestratorKernel
{
    private readonly IQueuePublisher _queue;

    public async Task<string> SubmitAsync(
        CreateZipRequest request,
        CancellationToken ct)
    {
        var taskId = Guid.NewGuid().ToString();
        var context = SagaContext<CreateZipRequest>.Create(taskId, request);

        await _taskRepo.CreateAsync(context.TaskInstance, ct);

        try
        {
            // Step 1: UploadSourceFiles (StepHost.Api)
            await ExecuteStepWithPoliciesAsync<UploadSourceFilesStepData>(
                context, "UploadSourceFiles", 1,
                ExecuteUploadSourceFilesAsync, retryPolicy: null, ct);

            // Dispatch to Worker
            var message = BuildWorkerMessage(context);
            await _queue.PublishAsync("file-processing", message, ct);
            await AuditAsync(context, "Dispatch", "Published to queue");

            return taskId;
        }
        catch (Exception)
        {
            await CompensateAsync(context, ct);
            throw;
        }
    }
}

// Generated: CreateZipWorkerOrchestrator.g.cs
public class CreateZipWorkerOrchestrator : CreateZipOrchestratorKernel
{
    private readonly IAsyncPolicy _createZipArchiveRetryPolicy;

    public async Task ProcessAsync(
        CreateZipFromFilesWorkerMessage message,
        CancellationToken ct)
    {
        await using var lockHandle = await _lock.AcquireAsync(
            $"task:{message.TaskId}", TimeSpan.FromSeconds(600), ct);

        var context = await RestoreContextAsync(message, ct);

        try
        {
            // Step 2: CreateZipArchive (StepHost.Worker) — with Polly retry
            await ExecuteStepWithPoliciesAsync<CreateZipArchiveStepData>(
                context, "CreateZipArchive", 2,
                ExecuteCreateZipArchiveAsync,
                _createZipArchiveRetryPolicy, ct);

            // Step 3: UploadZip (StepHost.Worker)
            await ExecuteStepWithPoliciesAsync<UploadZipStepData>(
                context, "UploadZip", 3,
                ExecuteUploadZipAsync, retryPolicy: null, ct);

            // Complete
            var response = MapResponse(context);
            await _taskRepo.CompleteAsync(context.TaskId, response, ct);
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            await CompensateAsync(context, ct);
            await _taskRepo.UpdateStatusAsync(message.TaskId,
                DistributedTaskStatus.Cancelled, ct);
        }
        catch (Exception)
        {
            await CompensateAsync(context, ct);
            await _taskRepo.UpdateStatusAsync(message.TaskId,
                DistributedTaskStatus.Failed, ct);
            throw;
        }
    }
}

// Generated: UploadSourceFilesStepData.g.cs
public sealed class UploadSourceFilesStepData
{
    public List<string> UploadedKeys { get; set; } = new();
    public string Bucket { get; set; } = "incoming-files";
}

// Generated: CreateZipArchiveStepData.g.cs
public sealed class CreateZipArchiveStepData
{
    public string ZipPath { get; set; } = "";
    public long ZipSizeBytes { get; set; }
}

// Generated: UploadZipStepData.g.cs
public sealed class UploadZipStepData
{
    public string ZipS3Key { get; set; } = "";
    public string ZipS3Bucket { get; set; } = "processed-files";
}

// Generated: CreateZipWorkerMessage.g.cs
public sealed class CreateZipWorkerMessage
{
    public string TaskId { get; set; } = "";

    // IFormFile → S3 keys (from UploadSourceFiles step)
    public List<string> FileS3Keys { get; set; } = new();
    public string FileS3Bucket { get; set; } = "";

    // Non-file properties carried through
    public string? OutputFileName { get; set; }

    // Completed API step data
    public UploadSourceFilesStepData UploadSourceFilesData { get; set; } = new();
}

// Generated: CreateZipConsumer.g.cs
public class CreateZipConsumer : IConsumer<CreateZipWorkerMessage>
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly IDistributedLock _lock;
    private readonly ILogger<CreateZipConsumer> _logger;

    public async Task Consume(ConsumeContext<CreateZipWorkerMessage> context)
    {
        var message = context.Message;
        _logger.LogInformation("Consuming task {TaskId}", message.TaskId);

        await using var lockHandle = await _lock.TryAcquireAsync(
            $"consumer:task:{message.TaskId}",
            TimeSpan.FromSeconds(600));

        if (lockHandle is null)
        {
            _logger.LogWarning(
                "Task {TaskId} already being processed, skipping", message.TaskId);
            return;
        }

        using var scope = _scopeFactory.CreateScope();
        var orchestrator = scope.ServiceProvider
            .GetRequiredService<CreateZipWorkerOrchestrator>();

        await orchestrator.ProcessAsync(message, context.CancellationToken);
    }
}

// Generated: CreateZipTaskController.g.cs
[ApiController]
[Route("api/tasks/create-zip")]
public class CreateZipTaskController : ControllerBase
{
    private readonly CreateZipApiOrchestrator _orchestrator;
    private readonly IDistributedTaskInstanceRepository _taskRepo;
    private readonly ITaskProgressTracker _progress;

    // ── POST: Submit a new task ──
    [HttpPost]
    public async Task<ActionResult<TaskSubmittedResponse>> Submit(
        [FromForm] CreateZipRequest request,
        CancellationToken ct)
    {
        var taskId = await _orchestrator.SubmitAsync(request, ct);
        return Accepted(new TaskSubmittedResponse { TaskId = taskId });
    }

    // ── GET: Poll for status ──
    [HttpGet("{taskId}/status")]
    public async Task<ActionResult<TaskStatusResponse>> GetStatus(
        string taskId, CancellationToken ct)
    {
        var instance = await _taskRepo.FindByIdAsync(taskId, ct);
        if (instance is null) return NotFound();
        return Ok(TaskStatusResponse.From(instance));
    }

    // ── GET: SSE progress stream ──
    [HttpGet("{taskId}/stream")]
    public async Task GetStream(string taskId, CancellationToken ct)
    {
        Response.ContentType = "text/event-stream";
        await foreach (var update in _progress.StreamAsync(taskId, ct))
        {
            await Response.WriteAsync($"data: {update.ToJson()}\n\n", ct);
            await Response.Body.FlushAsync(ct);
        }
    }

    // ── GET: WebSocket upgrade ──
    [HttpGet("{taskId}/ws")]
    public async Task GetWebSocket(string taskId)
    {
        if (!HttpContext.WebSockets.IsWebSocketRequest)
        {
            HttpContext.Response.StatusCode = 400;
            return;
        }
        var ws = await HttpContext.WebSockets.AcceptWebSocketAsync();
        await _progress.PipeToWebSocketAsync(taskId, ws, HttpContext.RequestAborted);
    }

    // ── GET: Completed result ──
    [HttpGet("{taskId}/result")]
    public async Task<ActionResult<CreateZipResponse>> GetResult(
        string taskId, CancellationToken ct)
    {
        var instance = await _taskRepo.FindByIdAsync(taskId, ct);
        if (instance is null) return NotFound();
        if (instance.Status != DistributedTaskStatus.Completed)
            return Conflict(new { message = "Task not yet completed" });
        return Ok(instance.DeserializeResponse<CreateZipResponse>());
    }

    // ── DELETE: Cancel a running task ──
    [HttpDelete("{taskId}")]
    public async Task<ActionResult> Cancel(
        string taskId, CancellationToken ct)
    {
        var instance = await _taskRepo.FindByIdAsync(taskId, ct);
        if (instance is null) return NotFound();
        if (instance.Status is DistributedTaskStatus.Completed
            or DistributedTaskStatus.Failed
            or DistributedTaskStatus.Cancelled)
            return Conflict(new { message = "Task already terminal" });

        await _taskRepo.UpdateStatusAsync(taskId,
            DistributedTaskStatus.Cancelled, ct);
        return Accepted();
    }
}

// Generated: CreateZipTaskEvents.g.cs
public sealed record TaskSubmittedEvent(
    string TaskId,
    string TaskName,
    DateTimeOffset Timestamp) : IDomainEvent;

public sealed record StepCompletedEvent(
    string TaskId,
    string StepName,
    int StepOrder,
    TimeSpan Duration,
    DateTimeOffset Timestamp) : IDomainEvent;

public sealed record StepFailedEvent(
    string TaskId,
    string StepName,
    int StepOrder,
    string ErrorMessage,
    int AttemptCount,
    DateTimeOffset Timestamp) : IDomainEvent;

public sealed record TaskCompletedEvent(
    string TaskId,
    TimeSpan TotalDuration,
    DateTimeOffset Timestamp) : IDomainEvent;

public sealed record TaskCancelledEvent(
    string TaskId,
    string? CancelledByStep,
    DateTimeOffset Timestamp) : IDomainEvent;

// Generated: CreateZipRetryPolicies.g.cs
public static class CreateZipRetryPolicies
{
    // From: [RetryPolicy("CreateZipArchive", MaxRetries = 3,
    //     BackoffType = BackoffType.Exponential, DelayMs = 500)]
    public static IAsyncPolicy CreateZipArchivePolicy { get; } =
        Policy
            .Handle<Exception>()
            .WaitAndRetryAsync(
                retryCount: 3,
                sleepDurationProvider: attempt =>
                    TimeSpan.FromMilliseconds(500 * Math.Pow(2, attempt - 1)),
                onRetry: (exception, delay, attempt, context) =>
                {
                    var logger = context.GetLogger();
                    logger?.LogWarning(exception,
                        "Retry {Attempt}/3 for CreateZipArchive after {Delay}ms",
                        attempt, delay.TotalMilliseconds);
                });

    // Steps without [RetryPolicy] get no policy (null is passed to the kernel)
}

// Generated: CreateZipTaskHub.g.cs
[Authorize]
public class CreateZipTaskHub : Hub
{
    public async Task Subscribe(string taskId)
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, $"task:{taskId}");
    }

    public async Task Unsubscribe(string taskId)
    {
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, $"task:{taskId}");
    }
}

// Used by the progress tracker:
// await _hubContext.Clients.Group($"task:{taskId}")
//     .SendAsync("StepProgress", update);

// Generated: CreateZipWebhookNotifier.g.cs
public class CreateZipWebhookNotifier : BackgroundService
{
    private readonly ITaskProgressTracker _progress;
    private readonly IHttpClientFactory _httpClientFactory;
    private readonly IDistributedTaskInstanceRepository _taskRepo;
    private readonly ILogger<CreateZipWebhookNotifier> _logger;

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        await foreach (var update in _progress.SubscribeAllAsync(
            "CreateZipFromFiles", ct))
        {
            var instance = await _taskRepo.FindByIdAsync(update.TaskId, ct);
            if (instance?.Listening != ListeningStrategy.Webhook) continue;
            if (string.IsNullOrEmpty(instance.WebhookUrl)) continue;

            await DeliverWithRetryAsync(instance.WebhookUrl, update, ct);
        }
    }

    private async Task DeliverWithRetryAsync(
        string url, TaskProgressUpdate update, CancellationToken ct)
    {
        var client = _httpClientFactory.CreateClient("webhook");
        var payload = JsonSerializer.Serialize(new
        {
            taskId = update.TaskId,
            step = update.StepName,
            status = update.Status.ToString(),
            timestamp = DateTimeOffset.UtcNow,
        });

        // 3 attempts with exponential backoff
        for (var attempt = 1; attempt <= 3; attempt++)
        {
            try
            {
                var response = await client.PostAsync(url,
                    new StringContent(payload, Encoding.UTF8,
                        "application/json"), ct);
                if (response.IsSuccessStatusCode) return;
            }
            catch (Exception ex) when (attempt < 3)
            {
                _logger.LogWarning(ex,
                    "Webhook delivery attempt {Attempt} failed for {TaskId}",
                    attempt, update.TaskId);
                await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, attempt)), ct);
            }
        }

        _logger.LogError(
            "Webhook delivery failed after 3 attempts for task {TaskId} to {Url}",
            update.TaskId, url);
    }
}

// Generated: CreateZipReconciliationServiceBase.g.cs
public abstract class CreateZipReconciliationServiceBase : BackgroundService
{
    private readonly IS3Client _s3;
    private readonly IDistributedTaskInstanceRepository _taskRepo;
    private readonly ILogger _logger;
    private readonly TimeSpan _interval = TimeSpan.FromHours(1);

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            await Task.Delay(_interval, ct);
            await ReconcileAsync(ct);
        }
    }

    private async Task ReconcileAsync(CancellationToken ct)
    {
        // List S3 objects in task buckets
        var incomingKeys = await _s3.ListAsync("incoming-files", "uploads/", ct);
        var processedKeys = await _s3.ListAsync("processed-files", "zips/", ct);

        foreach (var key in incomingKeys.Concat(processedKeys))
        {
            var taskId = ExtractTaskId(key);
            var instance = await _taskRepo.FindByIdAsync(taskId, ct);

            if (instance is null || IsOrphaned(instance))
            {
                if (await ShouldDeleteAsync(key, instance, ct))
                {
                    var bucket = key.StartsWith("uploads/")
                        ? "incoming-files" : "processed-files";
                    await _s3.DeleteAsync(bucket, key, ct);
                    _logger.LogInformation("Deleted orphaned S3 object {Key}", key);
                }
            }
        }
    }

    // Developer implements: custom logic for determining deletion eligibility
    protected abstract Task<bool> ShouldDeleteAsync(
        string s3Key,
        DistributedTaskInstance? instance,
        CancellationToken ct);

    private static bool IsOrphaned(DistributedTaskInstance instance)
        => instance.Status is DistributedTaskStatus.Failed
            or DistributedTaskStatus.Cancelled
            or DistributedTaskStatus.CompensationFailed
        && instance.CompletedAt < DateTimeOffset.UtcNow.AddHours(-2);

    private static string ExtractTaskId(string key)
        => key.Split('/')[1]; // uploads/{taskId}/file.pdf → taskId
}

// Generated: CreateZipArchiveService.g.cs
public class CreateZipArchiveService : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            await Task.Delay(TimeSpan.FromHours(6), ct);

            using var scope = _scopeFactory.CreateScope();
            var context = scope.ServiceProvider
                .GetRequiredService<DistributedTaskDbContext>();

            var cutoff = DateTimeOffset.UtcNow.AddDays(-30);

            await context.Database.ExecuteSqlRawAsync(@"
                INSERT INTO saga.DistributedTasks_Archive
                SELECT * FROM saga.DistributedTasks
                WHERE (Status = {0} OR Status = {1})
                  AND CompletedAt < {2};
                -- ... cascade to steps and audit entries ...
                DELETE FROM saga.DistributedTasks
                WHERE (Status = {0} OR Status = {1})
                  AND CompletedAt < {2};",
                (int)DistributedTaskStatus.Completed,
                (int)DistributedTaskStatus.Failed,
                cutoff, ct);
        }
    }
}

// Generated: CreateZipServiceExtensions.g.cs
public static class CreateZipServiceExtensions
{
    public static IServiceCollection AddCreateZipFromFilesTask(
        this IServiceCollection services)
    {
        // Orchestrators
        services.AddScoped<CreateZipApiOrchestrator>();
        services.AddScoped<CreateZipWorkerOrchestrator>();

        // Queue consumer
        services.AddScoped<IConsumer<CreateZipWorkerMessage>, CreateZipConsumer>();

        // Retry policies
        services.AddSingleton(CreateZipRetryPolicies.CreateZipArchivePolicy);

        // SignalR hub
        services.AddSignalR();

        // Webhook notifier
        services.AddHostedService<CreateZipWebhookNotifier>();

        // Reconciliation (if developer registered the concrete implementation)
        services.AddHostedService<CreateZipReconciliationService>();

        // Archive service
        services.AddHostedService<CreateZipArchiveService>();

        // Controller is discovered by ASP.NET Core's convention-based routing
        // (it inherits from ControllerBase and has [ApiController])

        return services;
    }
}

builder.Services.AddCreateZipFromFilesTask();

public partial class CreateZipFromFilesTask
{
    protected override async Task<StepResult> ExecuteCreateZipArchiveAsync(
        SagaContext<CreateZipRequest> context,
        CreateZipArchiveStepData stepData,
        CancellationToken ct)
    {
        var uploadData = context.GetStepData<UploadSourceFilesStepData>();
        // ... download files, create ZIP, populate stepData ...
        return StepResult.Success();
    }

    protected override CreateZipResponse MapResponse(
        SagaContext<CreateZipRequest> context)
    {
        // ... assemble response from step data ...
    }
}

public class CreateZipReconciliationService
    : CreateZipReconciliationServiceBase
{
    protected override Task<bool> ShouldDeleteAsync(
        string s3Key,
        DistributedTaskInstance? instance,
        CancellationToken ct)
    {
        // Delete if no task record exists or task failed >24h ago
        if (instance is null) return Task.FromResult(true);
        if (instance.Status == DistributedTaskStatus.CompensationFailed
            && instance.CompletedAt < DateTimeOffset.UtcNow.AddHours(-24))
            return Task.FromResult(true);

        return Task.FromResult(false);
    }
}

StepData ValueObjects (4-6)        ← no dependencies
WorkerMessage (7)                  ← depends on StepData
RetryPolicies (11)                 ← no dependencies
Domain Events (10)                 ← no dependencies
OrchestratorKernel (1)             ← depends on StepData, Events
ApiOrchestrator (2)                ← depends on Kernel, WorkerMessage
WorkerOrchestrator (3)             ← depends on Kernel, RetryPolicies
Consumer (8)                       ← depends on WorkerOrchestrator
Controller (9)                     ← depends on ApiOrchestrator
SignalR Hub (12)                   ← no dependencies
Webhook Notifier (13)              ← no dependencies
ReconciliationServiceBase (14)     ← no dependencies
ArchiveService (15)                ← no dependencies
ServiceExtensions (16)             ← depends on everything above