Producer-Consumer Pattern in C#

using System.Threading.Channels;
using Microsoft.Extensions.Logging;

// [step] Define typed work items with priority
public enum Priority { Low, Normal, High, Critical }

public record WorkItem<T>(
    string Id, T Payload, Priority Priority,
    DateTime EnqueuedAt)
{
    public static WorkItem<T> Create(T payload,
        Priority priority = Priority.Normal) =>
        new(Guid.NewGuid().ToString(), payload, priority, DateTime.UtcNow);
}

public record ProcessingResult<T>(
    string WorkItemId, bool Success, T? Result,
    TimeSpan Duration, string? Error = null);

// [step] Producer with backpressure and batching support
public sealed class Producer<T>(
    ChannelWriter<WorkItem<T>> writer,
    ILogger<Producer<T>> logger)
{
    private long _produced;

    public long ProducedCount => Interlocked.Read(ref _produced);

    public async Task ProduceAsync(
        T payload, Priority priority = Priority.Normal,
        CancellationToken ct = default)
    {
        var item = WorkItem<T>.Create(payload, priority);
        await writer.WriteAsync(item, ct);
        Interlocked.Increment(ref _produced);
        logger.LogDebug("Produced {Id} (priority: {Priority})",
            item.Id, priority);
    }

    public async Task ProduceBatchAsync(
        IEnumerable<T> payloads, Priority priority = Priority.Normal,
        CancellationToken ct = default)
    {
        foreach (var payload in payloads)
        {
            ct.ThrowIfCancellationRequested();
            await ProduceAsync(payload, priority, ct);
        }
    }

    public void Complete() => writer.TryComplete();
}

// [step] Consumer with parallel processing and error handling
public sealed class Consumer<TIn, TOut>(
    ChannelReader<WorkItem<TIn>> reader,
    Func<TIn, CancellationToken, Task<TOut>> processor,
    ILogger<Consumer<TIn, TOut>> logger)
{
    private long _processed;
    private long _failed;

    public long ProcessedCount => Interlocked.Read(ref _processed);
    public long FailedCount => Interlocked.Read(ref _failed);

    public async Task RunAsync(
        int concurrency = 1,
        CancellationToken ct = default)
    {
        var workers = Enumerable.Range(0, concurrency)
            .Select(i => ProcessLoopAsync($"Worker-{i}", ct));
        await Task.WhenAll(workers);
    }

    private async Task ProcessLoopAsync(
        string workerName, CancellationToken ct)
    {
        logger.LogInformation("{Worker} started", workerName);

        await foreach (var item in reader.ReadAllAsync(ct))
        {
            var sw = System.Diagnostics.Stopwatch.StartNew();
            try
            {
                var result = await processor(item.Payload, ct);
                Interlocked.Increment(ref _processed);
                logger.LogDebug("{Worker} processed {Id} in {Duration}ms",
                    workerName, item.Id, sw.ElapsedMilliseconds);
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                Interlocked.Increment(ref _failed);
                logger.LogError(ex, "{Worker} failed on {Id}",
                    workerName, item.Id);
            }
        }

        logger.LogInformation("{Worker} completed", workerName);
    }
}

// [step] Pipeline builder for composing producer-consumer chains
public static class PipelineBuilder
{
    public static (Producer<T>, Consumer<T, TOut>) Create<T, TOut>(
        Func<T, CancellationToken, Task<TOut>> processor,
        int capacity,
        ILoggerFactory loggerFactory)
    {
        var channel = Channel.CreateBounded<WorkItem<T>>(
            new BoundedChannelOptions(capacity)
            {
                FullMode = BoundedChannelFullMode.Wait,
                SingleReader = false,
                SingleWriter = false,
            });

        var producer = new Producer<T>(channel.Writer,
            loggerFactory.CreateLogger<Producer<T>>());
        var consumer = new Consumer<T, TOut>(channel.Reader, processor,
            loggerFactory.CreateLogger<Consumer<T, TOut>>());

        return (producer, consumer);
    }
}