Part XI: Composition📋

public sealed class Subscription : IHasDomainEvents
{
    public Guid Id { get; private set; }
    public Guid CustomerId { get; private set; }
    public Plan Plan { get; private set; }
    public DateTimeOffset ExpiresAt { get; private set; }
    public SubscriptionStatus Status { get; private set; }

    private readonly List<object> _domainEvents = new();
    public IReadOnlyList<object> DomainEvents => _domainEvents;
    public void ClearDomainEvents() => _domainEvents.Clear();

    public void Renew(Plan newPlan, DateTimeOffset newExpiration)
    {
        var previousPlan = Plan;
        Plan = newPlan;
        ExpiresAt = newExpiration;
        Status = SubscriptionStatus.Active;

        _domainEvents.Add(new SubscriptionRenewed(
            Id, CustomerId, previousPlan.Id, newPlan.Id,
            newExpiration, DateTimeOffset.UtcNow));
    }
}

public sealed record Plan(
    Guid Id,
    string Name,
    PlanTier Tier,
    decimal MonthlyPrice,
    int TermMonths);

public enum PlanTier
{
    Free = 0,
    Starter = 1,
    Professional = 2,
    Enterprise = 3
}

public sealed record StandardRenewal(Plan Plan, int TermMonths);
public sealed record UpgradeRenewal(Plan NewPlan, Plan OldPlan, int TermMonths);
public sealed record DowngradeRenewal(Plan NewPlan, Plan OldPlan, int TermMonths);

public abstract record SubscriptionEvent(
    Guid SubscriptionId,
    Guid CustomerId,
    DateTimeOffset OccurredAt);

public sealed record SubscriptionRenewed(
    Guid SubscriptionId,
    Guid CustomerId,
    Guid PreviousPlanId,
    Guid NewPlanId,
    DateTimeOffset NewExpiration,
    DateTimeOffset OccurredAt) : SubscriptionEvent(SubscriptionId, CustomerId, OccurredAt);

public sealed record PaymentProcessed(
    Guid SubscriptionId,
    Guid CustomerId,
    Guid PaymentId,
    decimal Amount,
    DateTimeOffset OccurredAt) : SubscriptionEvent(SubscriptionId, CustomerId, OccurredAt);

public sealed record RenewalConfirmationSent(
    Guid SubscriptionId,
    Guid CustomerId,
    string EmailAddress,
    DateTimeOffset OccurredAt) : SubscriptionEvent(SubscriptionId, CustomerId, OccurredAt);

public sealed class SubscriptionEntity
{
    public Guid Id { get; set; }
    public Guid CustomerId { get; set; }
    public Guid PlanId { get; set; }
    public string PlanName { get; set; } = string.Empty;
    public int PlanTier { get; set; }
    public decimal PlanMonthlyPrice { get; set; }
    public int PlanTermMonths { get; set; }
    public DateTimeOffset ExpiresAt { get; set; }
    public int Status { get; set; }
}

public sealed record RenewRequest(
    string SubscriptionId,
    string RequestedPlanId);

public sealed record RenewalReceipt(
    Guid SubscriptionId,
    string PlanName,
    DateTimeOffset NewExpiration,
    string RenewalType,
    decimal AmountCharged);

[ApiController]
[Route("api/subscriptions")]
public sealed class SubscriptionsController : ControllerBase
{
    private readonly IMediator _mediator;

    public SubscriptionsController(IMediator mediator)
    {
        _mediator = mediator;
    }

    [HttpPost("renew")]
    public async Task<IActionResult> Renew(
        [FromBody] RenewRequest request,
        CancellationToken ct)
    {
        // Guard validates at the API boundary.
        // If either value is null or empty, Guard.Against throws
        // an ArgumentException with the parameter name captured
        // via [CallerArgumentExpression].
        var subscriptionId = Guard.Against.NullOrEmpty(request.SubscriptionId);
        var requestedPlanId = Guard.Against.NullOrEmpty(request.RequestedPlanId);

        // Parse the string IDs into Guids.
        // Guard.Against.InvalidFormat would work here too, but
        // Guid.TryParse + Guard.Against.Null is more explicit.
        if (!Guid.TryParse(subscriptionId, out var subGuid))
            return BadRequest("Invalid subscription ID format.");
        if (!Guid.TryParse(requestedPlanId, out var planGuid))
            return BadRequest("Invalid plan ID format.");

        var command = new RenewSubscriptionCommand(subGuid, planGuid);

        // Dispatch through the mediator pipeline.
        var result = await _mediator.SendAsync(command, ct);

        // Result.Match converts success/failure to HTTP responses.
        return result.Match<IActionResult>(
            success: receipt => Ok(receipt),
            failure: error => BadRequest(error));
    }
}

public sealed record RenewSubscriptionCommand(
    Guid SubscriptionId,
    Guid RequestedPlanId) : ICommand<Result<RenewalReceipt>>;

[Injectable(Lifetime.Scoped)]
public sealed class SubscriptionRepository : ISubscriptionRepository
{
    private readonly AppDbContext _db;

    public SubscriptionRepository(AppDbContext db)
    {
        _db = db;
    }

    public async Task<Option<SubscriptionEntity>> FindByIdAsync(
        Guid id, CancellationToken ct)
    {
        var entity = await _db.Subscriptions.FindAsync(
            new object[] { id }, ct);

        return entity is not null
            ? Option<SubscriptionEntity>.Some(entity)
            : Option<SubscriptionEntity>.None;
    }

    public async Task<Option<PlanEntity>> FindPlanByIdAsync(
        Guid planId, CancellationToken ct)
    {
        var entity = await _db.Plans.FindAsync(
            new object[] { planId }, ct);

        return entity is not null
            ? Option<PlanEntity>.Some(entity)
            : Option<PlanEntity>.None;
    }
}

[MapFrom(typeof(SubscriptionEntity))]
[MapTo(typeof(SubscriptionEntity))]
public sealed class SubscriptionMapper
    : IMapper<SubscriptionEntity, Subscription>
{
    [MapProperty(nameof(SubscriptionEntity.PlanId), "Plan.Id")]
    [MapProperty(nameof(SubscriptionEntity.PlanName), "Plan.Name")]
    [MapProperty(nameof(SubscriptionEntity.PlanTier), "Plan.Tier")]
    [MapProperty(nameof(SubscriptionEntity.PlanMonthlyPrice), "Plan.MonthlyPrice")]
    [MapProperty(nameof(SubscriptionEntity.PlanTermMonths), "Plan.TermMonths")]
    [MapProperty(nameof(SubscriptionEntity.Status), "Status")]
    public Subscription Map(SubscriptionEntity source)
    {
        // Source-generated: the body of this method is replaced at compile time.
        // The generator reads the [MapProperty] attributes and emits:
        //   return new Subscription
        //   {
        //       Id = source.Id,
        //       CustomerId = source.CustomerId,
        //       Plan = new Plan(source.PlanId, source.PlanName,
        //           (PlanTier)source.PlanTier, source.PlanMonthlyPrice,
        //           source.PlanTermMonths),
        //       ExpiresAt = source.ExpiresAt,
        //       Status = (SubscriptionStatus)source.Status
        //   };
        throw new NotImplementedException("Source-generated at compile time.");
    }
}

// Inside RenewSubscriptionHandler.HandleAsync

// Step 1: Look up the subscription. Returns Option<SubscriptionEntity>.
var subscriptionOption = await _repository.FindByIdAsync(
    command.SubscriptionId, ct);

// Step 2: Map the entity to the domain model.
// Option.Map transforms the inner value if it exists.
// If subscriptionOption is None, the Map is skipped entirely.
var domainOption = subscriptionOption.Map(
    entity => _mapper.Map(entity));

// Step 3: Convert Option to Result.
// Some(subscription) becomes Result.Success(subscription).
// None becomes Result.Failure("Subscription not found").
var subscriptionResult = domainOption.ToResult("Subscription not found");

// If the subscription was not found, return the failure immediately.
if (!subscriptionResult.IsSuccess)
    return Result<RenewalReceipt>.Failure(subscriptionResult.Error!);

var subscription = subscriptionResult.Value!;

// Look up the requested plan
var planOption = await _repository.FindPlanByIdAsync(
    command.RequestedPlanId, ct);

var planResult = planOption.ToResult("Requested plan not found");
if (!planResult.IsSuccess)
    return Result<RenewalReceipt>.Failure(planResult.Error!);

var requestedPlan = planResult.Value!;

private static OneOf<StandardRenewal, UpgradeRenewal, DowngradeRenewal>
    DetermineRenewalType(Plan current, Plan requested)
{
    if (current.Tier == requested.Tier)
    {
        return new StandardRenewal(requested, requested.TermMonths);
    }

    if (requested.Tier > current.Tier)
    {
        return new UpgradeRenewal(
            NewPlan: requested,
            OldPlan: current,
            TermMonths: requested.TermMonths);
    }

    return new DowngradeRenewal(
        NewPlan: requested,
        OldPlan: current,
        TermMonths: current.TermMonths);
}

// Inside RenewSubscriptionHandler.HandleAsync

OneOf<StandardRenewal, UpgradeRenewal, DowngradeRenewal> renewalType =
    DetermineRenewalType(subscription.Plan, requestedPlan);

var newExpiration = renewalType.Match(
    standard => _clock.UtcNow.AddMonths(standard.TermMonths),
    upgrade  => _clock.UtcNow.AddMonths(upgrade.TermMonths),
    downgrade => subscription.ExpiresAt
);

var chargeAmount = renewalType.Match(
    standard => standard.Plan.MonthlyPrice * standard.TermMonths,
    upgrade  => {
        // Upgrade: charge the difference pro-rated for the remaining term
        var remainingMonths = (int)Math.Ceiling(
            (subscription.ExpiresAt - _clock.UtcNow).TotalDays / 30);
        var priceDifference = upgrade.NewPlan.MonthlyPrice
            - upgrade.OldPlan.MonthlyPrice;
        var proRatedUpgrade = priceDifference * remainingMonths;
        var newTermCharge = upgrade.NewPlan.MonthlyPrice * upgrade.TermMonths;
        return proRatedUpgrade + newTermCharge;
    },
    downgrade => 0m // No charge for downgrades
);

var renewalTypeLabel = renewalType.Match(
    standard  => "Standard",
    upgrade   => "Upgrade",
    downgrade => "Downgrade"
);

[Injectable(Lifetime.Transient)]
public sealed class LoggingBehavior<TRequest, TResult>
    : IBehavior<TRequest, TResult>
    where TRequest : IRequest<TResult>
{
    private readonly ILogger<LoggingBehavior<TRequest, TResult>> _logger;

    public LoggingBehavior(
        ILogger<LoggingBehavior<TRequest, TResult>> logger)
    {
        _logger = logger;
    }

    public async Task<TResult> HandleAsync(
        TRequest request,
        Func<Task<TResult>> next,
        CancellationToken ct)
    {
        var requestName = typeof(TRequest).Name;
        _logger.LogInformation(
            "Handling {RequestName}: {@Request}",
            requestName, request);

        var stopwatch = System.Diagnostics.Stopwatch.StartNew();
        var result = await next();
        stopwatch.Stop();

        _logger.LogInformation(
            "Handled {RequestName} in {ElapsedMs}ms",
            requestName, stopwatch.ElapsedMilliseconds);

        return result;
    }
}

[Injectable(Lifetime.Transient)]
public sealed class ValidationBehavior<TRequest, TResult>
    : IBehavior<TRequest, TResult>
    where TRequest : IRequest<TResult>
{
    private readonly IEnumerable<IValidator<TRequest>> _validators;

    public ValidationBehavior(
        IEnumerable<IValidator<TRequest>> validators)
    {
        _validators = validators;
    }

    public async Task<TResult> HandleAsync(
        TRequest request,
        Func<Task<TResult>> next,
        CancellationToken ct)
    {
        var failures = _validators
            .Select(v => v.Validate(request))
            .Where(r => !r.IsValid)
            .SelectMany(r => r.Errors)
            .ToList();

        if (failures.Count > 0)
        {
            // TResult is expected to be Result<T> for commands.
            // Return a failure result without calling the handler.
            var errorMessage = string.Join("; ",
                failures.Select(f => f.ErrorMessage));

            // Use reflection-free result construction via the
            // IResultFactory<TResult> pattern from the Result package.
            return ResultFactory<TResult>.Failure(errorMessage);
        }

        return await next();
    }
}

[Injectable(Lifetime.Scoped)]
public sealed class RenewSubscriptionHandler
    : IRequestHandler<RenewSubscriptionCommand, Result<RenewalReceipt>>
{
    private readonly ISubscriptionRepository _repository;
    private readonly IMapper<SubscriptionEntity, Subscription> _mapper;
    private readonly IClock _clock;
    private readonly SagaOrchestrator<RenewalSagaContext> _saga;

    public RenewSubscriptionHandler(
        ISubscriptionRepository repository,
        IMapper<SubscriptionEntity, Subscription> mapper,
        IClock clock,
        SagaOrchestrator<RenewalSagaContext> saga)
    {
        _repository = repository;
        _mapper = mapper;
        _clock = clock;
        _saga = saga;
    }

    public async Task<Result<RenewalReceipt>> HandleAsync(
        RenewSubscriptionCommand command,
        CancellationToken ct)
    {
        // Layer 2: Data Retrieval (Option + Mapper)
        var subscriptionOption = await _repository.FindByIdAsync(
            command.SubscriptionId, ct);

        var subscriptionResult = subscriptionOption
            .Map(entity => _mapper.Map(entity))
            .ToResult("Subscription not found");

        if (!subscriptionResult.IsSuccess)
            return Result<RenewalReceipt>.Failure(
                subscriptionResult.Error!);

        var subscription = subscriptionResult.Value!;

        // Look up the requested plan
        var planOption = await _repository.FindPlanByIdAsync(
            command.RequestedPlanId, ct);

        var planResult = planOption.ToResult("Requested plan not found");
        if (!planResult.IsSuccess)
            return Result<RenewalReceipt>.Failure(planResult.Error!);

        var requestedPlan = planResult.Value!;

        // Layer 3: Business Logic (Union + Clock)
        var renewalType = DetermineRenewalType(
            subscription.Plan, requestedPlan);

        var newExpiration = renewalType.Match(
            standard  => _clock.UtcNow.AddMonths(standard.TermMonths),
            upgrade   => _clock.UtcNow.AddMonths(upgrade.TermMonths),
            downgrade => subscription.ExpiresAt
        );

        var chargeAmount = renewalType.Match(
            standard => standard.Plan.MonthlyPrice * standard.TermMonths,
            upgrade  => {
                var remainingMonths = (int)Math.Ceiling(
                    (subscription.ExpiresAt - _clock.UtcNow).TotalDays / 30);
                var diff = upgrade.NewPlan.MonthlyPrice
                    - upgrade.OldPlan.MonthlyPrice;
                return (diff * remainingMonths)
                    + (upgrade.NewPlan.MonthlyPrice * upgrade.TermMonths);
            },
            downgrade => 0m
        );

        var renewalTypeLabel = renewalType.Match(
            standard  => "Standard",
            upgrade   => "Upgrade",
            downgrade => "Downgrade"
        );

        // Layer 5: Orchestration (Saga)
        var sagaContext = new RenewalSagaContext
        {
            SubscriptionId = subscription.Id,
            CustomerId = subscription.CustomerId,
            NewPlan = requestedPlan,
            NewExpiration = newExpiration,
            ChargeAmount = chargeAmount,
            RenewalType = renewalTypeLabel
        };

        var sagaResult = await _saga.ExecuteAsync(sagaContext, ct);

        if (!sagaResult.IsSuccess)
            return Result<RenewalReceipt>.Failure(sagaResult.Error!);

        // Build the receipt
        var receipt = new RenewalReceipt(
            subscription.Id,
            requestedPlan.Name,
            newExpiration,
            renewalTypeLabel,
            chargeAmount);

        return Result<RenewalReceipt>.Success(receipt);
    }

    private static OneOf<StandardRenewal, UpgradeRenewal, DowngradeRenewal>
        DetermineRenewalType(Plan current, Plan requested)
    {
        if (current.Tier == requested.Tier)
            return new StandardRenewal(requested, requested.TermMonths);

        if (requested.Tier > current.Tier)
            return new UpgradeRenewal(requested, current,
                requested.TermMonths);

        return new DowngradeRenewal(requested, current,
            current.TermMonths);
    }
}

public sealed class RenewalSagaContext : SagaContext
{
    // Input
    public Guid SubscriptionId { get; set; }
    public Guid CustomerId { get; set; }
    public Plan NewPlan { get; set; } = null!;
    public DateTimeOffset NewExpiration { get; set; }
    public decimal ChargeAmount { get; set; }
    public string RenewalType { get; set; } = string.Empty;

    // State captured by steps (for compensation)
    public Guid? PaymentId { get; set; }
    public Plan? PreviousPlan { get; set; }
    public DateTimeOffset? PreviousExpiration { get; set; }
    public string? CustomerEmail { get; set; }
}

[Injectable(Lifetime.Scoped)]
public sealed class ProcessPaymentStep : ISagaStep<RenewalSagaContext>
{
    private readonly IPaymentGateway _gateway;

    public ProcessPaymentStep(IPaymentGateway gateway)
    {
        _gateway = gateway;
    }

    public async Task<Result> ExecuteAsync(
        RenewalSagaContext context, CancellationToken ct)
    {
        if (context.ChargeAmount <= 0)
        {
            // No charge for downgrades -- skip payment.
            return Result.Success();
        }

        var paymentResult = await _gateway.ChargeAsync(
            context.CustomerId,
            context.ChargeAmount,
            description: $"{context.RenewalType} renewal for subscription "
                + $"{context.SubscriptionId}",
            ct);

        if (!paymentResult.IsSuccess)
            return Result.Failure(
                $"Payment failed: {paymentResult.Error}");

        // Capture the payment ID for compensation.
        context.PaymentId = paymentResult.Value;

        return Result.Success();
    }

    public async Task<Result> CompensateAsync(
        RenewalSagaContext context, CancellationToken ct)
    {
        if (!context.PaymentId.HasValue)
            return Result.Success(); // Nothing to compensate.

        // Idempotent: check if already refunded.
        var status = await _gateway.GetPaymentStatusAsync(
            context.PaymentId.Value, ct);

        if (status == PaymentStatus.Refunded)
            return Result.Success();

        var refundResult = await _gateway.RefundAsync(
            context.PaymentId.Value, ct);

        return refundResult.IsSuccess
            ? Result.Success()
            : Result.Failure(
                $"Refund failed for payment {context.PaymentId.Value}");
    }
}

[Injectable(Lifetime.Scoped)]
public sealed class UpdateSubscriptionStep : ISagaStep<RenewalSagaContext>
{
    private readonly ISubscriptionRepository _repository;
    private readonly IMapper<SubscriptionEntity, Subscription> _mapper;
    private readonly AppDbContext _db;

    public UpdateSubscriptionStep(
        ISubscriptionRepository repository,
        IMapper<SubscriptionEntity, Subscription> mapper,
        AppDbContext db)
    {
        _repository = repository;
        _mapper = mapper;
        _db = db;
    }

    public async Task<Result> ExecuteAsync(
        RenewalSagaContext context, CancellationToken ct)
    {
        var subscriptionOption = await _repository.FindByIdAsync(
            context.SubscriptionId, ct);

        var subscription = subscriptionOption
            .Map(entity => _mapper.Map(entity))
            .OrDefault(null!);

        if (subscription is null)
            return Result.Failure("Subscription vanished during saga.");

        // Capture previous state for compensation.
        context.PreviousPlan = subscription.Plan;
        context.PreviousExpiration = subscription.ExpiresAt;

        // Apply the renewal.
        // This raises a SubscriptionRenewed domain event
        // that the OutboxInterceptor will capture.
        subscription.Renew(context.NewPlan, context.NewExpiration);

        // SaveChanges triggers the OutboxInterceptor:
        // 1. Scans ChangeTracker for IHasDomainEvents entities.
        // 2. Serializes SubscriptionRenewed to JSON.
        // 3. Creates an OutboxMessage in the same transaction.
        // 4. Clears the entity's domain events list.
        await _db.SaveChangesAsync(ct);

        return Result.Success();
    }

    public async Task<Result> CompensateAsync(
        RenewalSagaContext context, CancellationToken ct)
    {
        if (context.PreviousPlan is null
            || context.PreviousExpiration is null)
            return Result.Success(); // Execute never ran or captured nothing.

        var subscriptionOption = await _repository.FindByIdAsync(
            context.SubscriptionId, ct);

        var subscription = subscriptionOption
            .Map(entity => _mapper.Map(entity))
            .OrDefault(null!);

        if (subscription is null)
            return Result.Failure(
                "Subscription vanished during compensation.");

        // Revert to previous state.
        subscription.Renew(
            context.PreviousPlan,
            context.PreviousExpiration.Value);

        await _db.SaveChangesAsync(ct);

        return Result.Success();
    }
}

[Injectable(Lifetime.Scoped)]
public sealed class SendConfirmationStep : ISagaStep<RenewalSagaContext>
{
    private readonly ICustomerRepository _customers;
    private readonly IEmailService _email;

    public SendConfirmationStep(
        ICustomerRepository customers,
        IEmailService email)
    {
        _customers = customers;
        _email = email;
    }

    public async Task<Result> ExecuteAsync(
        RenewalSagaContext context, CancellationToken ct)
    {
        var customerOption = await _customers.FindByIdAsync(
            context.CustomerId, ct);

        var emailResult = customerOption
            .Map(c => c.EmailAddress)
            .ToResult("Customer not found for notification.");

        if (!emailResult.IsSuccess)
            return Result.Failure(emailResult.Error!);

        var email = emailResult.Value!;
        context.CustomerEmail = email;

        await _email.SendRenewalConfirmationAsync(
            email,
            context.SubscriptionId,
            context.NewPlan.Name,
            context.NewExpiration,
            context.RenewalType,
            ct);

        return Result.Success();
    }

    public Task<Result> CompensateAsync(
        RenewalSagaContext context, CancellationToken ct)
    {
        // Emails cannot be unsent.
        // This step is placed last in the saga specifically
        // because it is irreversible.
        // If this step fails, the preceding steps are compensated.
        // If this step succeeds, nothing after it can fail.
        return Task.FromResult(Result.Success());
    }
}

var orchestrator = new SagaOrchestrator<RenewalSagaContext>(
    new ISagaStep<RenewalSagaContext>[]
    {
        processPaymentStep,    // Step 1: reversible
        updateSubscriptionStep, // Step 2: reversible
        sendConfirmationStep   // Step 3: irreversible (last)
    });

[Injectable(Lifetime.Singleton)]
public sealed class SubscriptionOutboxProcessor : BackgroundService
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly IEventStream<SubscriptionEvent> _stream;
    private readonly OutboxProcessorOptions _options;
    private readonly ILogger<SubscriptionOutboxProcessor> _logger;

    // _stream is the EventStream<SubscriptionEvent> singleton.
    // It is the "producer" side -- Publish method available.
    // Subscribers hold IEventStream<SubscriptionEvent> references
    // (the "consumer" side -- Subscribe only).
    private readonly EventStream<SubscriptionEvent> _publisher;

    public SubscriptionOutboxProcessor(
        IServiceScopeFactory scopeFactory,
        EventStream<SubscriptionEvent> publisher,
        OutboxProcessorOptions options,
        ILogger<SubscriptionOutboxProcessor> logger)
    {
        _scopeFactory = scopeFactory;
        _publisher = publisher;
        _options = options;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            try
            {
                await ProcessPendingAsync(ct);
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                _logger.LogError(ex,
                    "Error processing outbox messages");
            }

            await Task.Delay(_options.PollingInterval, ct);
        }
    }

    private async Task ProcessPendingAsync(CancellationToken ct)
    {
        using var scope = _scopeFactory.CreateScope();
        var outbox = scope.ServiceProvider
            .GetRequiredService<IOutbox>();

        var pending = await outbox.GetPendingAsync(
            _options.BatchSize, ct);

        foreach (var message in pending)
        {
            try
            {
                var eventType = Type.GetType(message.Type);
                if (eventType is null)
                {
                    _logger.LogWarning(
                        "Unknown event type {Type} in outbox message {Id}",
                        message.Type, message.Id);
                    continue;
                }

                var domainEvent = System.Text.Json.JsonSerializer
                    .Deserialize(message.Payload, eventType);

                if (domainEvent is SubscriptionEvent subscriptionEvent)
                {
                    // Publish to the reactive event stream.
                    // All subscribers receive the event.
                    _publisher.Publish(subscriptionEvent);
                }

                message.ProcessedAt = DateTimeOffset.UtcNow;
                await outbox.MarkProcessedAsync(message.Id, ct);

                _logger.LogInformation(
                    "Published outbox message {Id} of type {Type}",
                    message.Id, message.Type);
            }
            catch (Exception ex)
            {
                message.Attempts++;
                message.LastError = ex.Message;
                await outbox.UpdateAsync(message, ct);

                _logger.LogError(ex,
                    "Failed to process outbox message {Id}, " +
                    "attempt {Attempts}",
                    message.Id, message.Attempts);
            }
        }
    }
}

[Injectable(Lifetime.Singleton)]
public sealed class AnalyticsSubscriber : IHostedService, IDisposable
{
    private readonly IEventStream<SubscriptionEvent> _stream;
    private readonly IAnalyticsService _analytics;
    private readonly ILogger<AnalyticsSubscriber> _logger;
    private IDisposable? _subscription;

    public AnalyticsSubscriber(
        IEventStream<SubscriptionEvent> stream,
        IAnalyticsService analytics,
        ILogger<AnalyticsSubscriber> logger)
    {
        _stream = stream;
        _analytics = analytics;
        _logger = logger;
    }

    public Task StartAsync(CancellationToken ct)
    {
        _subscription = _stream
            .OfType<SubscriptionRenewed>()
            .Subscribe(
                onNext: async evt =>
                {
                    await _analytics.TrackRenewalAsync(
                        evt.SubscriptionId,
                        evt.PreviousPlanId,
                        evt.NewPlanId,
                        evt.NewExpiration);
                },
                onError: ex =>
                {
                    _logger.LogError(ex,
                        "Analytics subscriber error");
                });

        _logger.LogInformation("Analytics subscriber started");
        return Task.CompletedTask;
    }

    public Task StopAsync(CancellationToken ct)
    {
        _subscription?.Dispose();
        _logger.LogInformation("Analytics subscriber stopped");
        return Task.CompletedTask;
    }

    public void Dispose() => _subscription?.Dispose();
}

[Injectable(Lifetime.Singleton)]
public sealed class DashboardSubscriber : IHostedService, IDisposable
{
    private readonly IEventStream<SubscriptionEvent> _stream;
    private readonly IHubContext<DashboardHub> _hub;
    private readonly ILogger<DashboardSubscriber> _logger;
    private IDisposable? _subscription;

    public DashboardSubscriber(
        IEventStream<SubscriptionEvent> stream,
        IHubContext<DashboardHub> hub,
        ILogger<DashboardSubscriber> logger)
    {
        _stream = stream;
        _hub = hub;
        _logger = logger;
    }

    public Task StartAsync(CancellationToken ct)
    {
        // Buffer events in 5-second windows for batch updates.
        _subscription = _stream
            .Buffer(TimeSpan.FromSeconds(5))
            .Filter(batch => batch.Count > 0)
            .Subscribe(
                onNext: async batch =>
                {
                    var summary = new DashboardUpdate
                    {
                        TotalRenewals = batch.Count,
                        Upgrades = batch.Count(
                            e => e is SubscriptionRenewed r
                                && r.NewPlanId != r.PreviousPlanId),
                        Timestamp = DateTimeOffset.UtcNow
                    };

                    await _hub.Clients.All.SendAsync(
                        "ReceiveDashboardUpdate", summary);
                },
                onError: ex =>
                {
                    _logger.LogError(ex,
                        "Dashboard subscriber error");
                });

        _logger.LogInformation("Dashboard subscriber started");
        return Task.CompletedTask;
    }

    public Task StopAsync(CancellationToken ct)
    {
        _subscription?.Dispose();
        _logger.LogInformation("Dashboard subscriber stopped");
        return Task.CompletedTask;
    }

    public void Dispose() => _subscription?.Dispose();
}

public static class ServiceCollectionExtensions
{
    public static IServiceCollection AddSubscriptionRenewal(
        this IServiceCollection services)
    {
        // ── Source-generated registrations ──────────────────────
        // Registers all [Injectable] services in this assembly:
        // - SubscriptionRepository (Scoped)
        // - RenewSubscriptionHandler (Scoped)
        // - ProcessPaymentStep (Scoped)
        // - UpdateSubscriptionStep (Scoped)
        // - SendConfirmationStep (Scoped)
        // - LoggingBehavior<,> (Transient, open generic)
        // - ValidationBehavior<,> (Transient, open generic)
        // - AnalyticsSubscriber (Singleton)
        // - DashboardSubscriber (Singleton)
        // - SubscriptionOutboxProcessor (Singleton)
        services.AddServices();

        // ── Clock ──────────────────────────────────────────────
        // SystemClock.Instance delegates to DateTimeOffset.UtcNow.
        // Tests replace this with FakeClock.
        services.AddSingleton<IClock>(SystemClock.Instance);

        // ── Outbox ─────────────────────────────────────────────
        // The OutboxInterceptor hooks into EF Core's SaveChanges
        // pipeline to capture domain events as OutboxMessages.
        services.AddDbContext<AppDbContext>(options =>
            options
                .UseSqlServer(/* connection string */)
                .AddInterceptors(new OutboxInterceptor()));

        services.AddScoped<IOutbox, EfCoreOutbox>();

        // ── Outbox processor options ───────────────────────────
        services.AddSingleton(new OutboxProcessorOptions
        {
            PollingInterval = TimeSpan.FromSeconds(5),
            BatchSize = 100,
            MaxRetryAttempts = 3,
            RetentionPeriod = TimeSpan.FromDays(7)
        });

        // ── Saga ───────────────────────────────────────────────
        // The orchestrator is assembled from its three steps,
        // resolved from the container.
        services.AddScoped<SagaOrchestrator<RenewalSagaContext>>(sp =>
            new SagaOrchestrator<RenewalSagaContext>(
                new ISagaStep<RenewalSagaContext>[]
                {
                    sp.GetRequiredService<ProcessPaymentStep>(),
                    sp.GetRequiredService<UpdateSubscriptionStep>(),
                    sp.GetRequiredService<SendConfirmationStep>()
                }));

        // ── Reactive ───────────────────────────────────────────
        // A single EventStream instance shared by the producer
        // (OutboxProcessor) and all consumers (subscribers).
        var eventStream = new EventStream<SubscriptionEvent>();
        services.AddSingleton(eventStream);
        services.AddSingleton<IEventStream<SubscriptionEvent>>(
            eventStream);

        // ── Hosted services ────────────────────────────────────
        services.AddHostedService<SubscriptionOutboxProcessor>();
        services.AddHostedService<AnalyticsSubscriber>();
        services.AddHostedService<DashboardSubscriber>();

        return services;
    }
}

public sealed class SubscriptionRenewalIntegrationTests
{
    private readonly FakeClock _clock;
    private readonly InMemoryOutbox _outbox;
    private readonly TestEventStream<SubscriptionEvent> _eventStream;
    private readonly FakePaymentGateway _paymentGateway;
    private readonly FakeEmailService _emailService;
    private readonly InMemorySubscriptionRepository _repository;

    public SubscriptionRenewalIntegrationTests()
    {
        _clock = new FakeClock(
            new DateTimeOffset(2026, 1, 15, 0, 0, 0, TimeSpan.Zero));
        _outbox = new InMemoryOutbox();
        _eventStream = new TestEventStream<SubscriptionEvent>();
        _paymentGateway = new FakePaymentGateway();
        _emailService = new FakeEmailService();
        _repository = new InMemorySubscriptionRepository();
    }

    [Fact]
    public async Task Standard_renewal_extends_expiration_by_term_months()
    {
        // Arrange
        var plan = new Plan(
            Guid.NewGuid(), "Professional", PlanTier.Professional,
            99.99m, TermMonths: 12);

        var subscription = CreateSubscription(
            plan, expiresAt: _clock.UtcNow.AddMonths(1));

        _repository.Add(subscription);
        _repository.AddPlan(plan);
        _paymentGateway.NextChargeResult =
            Result<Guid>.Success(Guid.NewGuid());

        var handler = CreateHandler();
        var command = new RenewSubscriptionCommand(
            subscription.Id, plan.Id);

        // Act
        var result = await handler.HandleAsync(
            command, CancellationToken.None);

        // Assert
        Assert.True(result.IsSuccess);

        var receipt = result.Value!;
        Assert.Equal("Standard", receipt.RenewalType);
        Assert.Equal(plan.Name, receipt.PlanName);

        // Clock: expiration is exactly 12 months from the fake clock's UtcNow.
        Assert.Equal(
            _clock.UtcNow.AddMonths(12),
            receipt.NewExpiration);

        // Payment: charged the full annual price.
        Assert.Single(_paymentGateway.Charges);
        Assert.Equal(
            99.99m * 12,
            _paymentGateway.Charges[0].Amount);

        // Email: confirmation sent.
        Assert.Single(_emailService.SentEmails);
    }

    [Fact]
    public async Task Upgrade_renewal_charges_prorated_difference()
    {
        // Arrange
        var starterPlan = new Plan(
            Guid.NewGuid(), "Starter", PlanTier.Starter,
            29.99m, TermMonths: 12);

        var proPlan = new Plan(
            Guid.NewGuid(), "Professional", PlanTier.Professional,
            99.99m, TermMonths: 12);

        // Subscription expires in 6 months (6 months remaining).
        var subscription = CreateSubscription(
            starterPlan, expiresAt: _clock.UtcNow.AddMonths(6));

        _repository.Add(subscription);
        _repository.AddPlan(proPlan);
        _paymentGateway.NextChargeResult =
            Result<Guid>.Success(Guid.NewGuid());

        var handler = CreateHandler();
        var command = new RenewSubscriptionCommand(
            subscription.Id, proPlan.Id);

        // Act
        var result = await handler.HandleAsync(
            command, CancellationToken.None);

        // Assert
        Assert.True(result.IsSuccess);
        Assert.Equal("Upgrade", result.Value!.RenewalType);

        // Price difference: $99.99 - $29.99 = $70.00/month.
        // Pro-rated for 6 remaining months: $420.00.
        // Plus new 12-month term at $99.99/month: $1,199.88.
        // Total: $1,619.88.
        var expectedCharge =
            (99.99m - 29.99m) * 6  // Pro-rated upgrade
            + 99.99m * 12;         // New term
        Assert.Equal(
            expectedCharge,
            _paymentGateway.Charges[0].Amount);
    }

    [Fact]
    public async Task Downgrade_renewal_charges_nothing()
    {
        // Arrange
        var proPlan = new Plan(
            Guid.NewGuid(), "Professional", PlanTier.Professional,
            99.99m, TermMonths: 12);

        var starterPlan = new Plan(
            Guid.NewGuid(), "Starter", PlanTier.Starter,
            29.99m, TermMonths: 12);

        var originalExpiration = _clock.UtcNow.AddMonths(6);
        var subscription = CreateSubscription(
            proPlan, expiresAt: originalExpiration);

        _repository.Add(subscription);
        _repository.AddPlan(starterPlan);

        var handler = CreateHandler();
        var command = new RenewSubscriptionCommand(
            subscription.Id, starterPlan.Id);

        // Act
        var result = await handler.HandleAsync(
            command, CancellationToken.None);

        // Assert
        Assert.True(result.IsSuccess);
        Assert.Equal("Downgrade", result.Value!.RenewalType);
        Assert.Equal(0m, result.Value!.AmountCharged);

        // Expiration unchanged: downgrades keep the current expiration.
        Assert.Equal(originalExpiration, result.Value!.NewExpiration);

        // No payment was processed.
        Assert.Empty(_paymentGateway.Charges);
    }

    [Fact]
    public async Task Missing_subscription_returns_failure()
    {
        // Arrange -- no subscription in repository.
        var handler = CreateHandler();
        var command = new RenewSubscriptionCommand(
            Guid.NewGuid(), Guid.NewGuid());

        // Act
        var result = await handler.HandleAsync(
            command, CancellationToken.None);

        // Assert
        Assert.False(result.IsSuccess);
        Assert.Equal("Subscription not found", result.Error);
    }

    [Fact]
    public async Task Payment_failure_triggers_compensation()
    {
        // Arrange
        var plan = new Plan(
            Guid.NewGuid(), "Professional", PlanTier.Professional,
            99.99m, TermMonths: 12);

        var subscription = CreateSubscription(
            plan, expiresAt: _clock.UtcNow.AddMonths(1));

        _repository.Add(subscription);
        _repository.AddPlan(plan);

        // Simulate payment failure.
        _paymentGateway.NextChargeResult =
            Result<Guid>.Failure("Card declined");

        var handler = CreateHandler();
        var command = new RenewSubscriptionCommand(
            subscription.Id, plan.Id);

        // Act
        var result = await handler.HandleAsync(
            command, CancellationToken.None);

        // Assert
        Assert.False(result.IsSuccess);
        Assert.Contains("Payment failed", result.Error!);

        // No email was sent (saga compensated before step 3).
        Assert.Empty(_emailService.SentEmails);
    }

    [Fact]
    public async Task Clock_advance_produces_different_expiration()
    {
        // Arrange
        var plan = new Plan(
            Guid.NewGuid(), "Monthly", PlanTier.Starter,
            9.99m, TermMonths: 1);

        var subscription = CreateSubscription(
            plan, expiresAt: _clock.UtcNow.AddDays(5));

        _repository.Add(subscription);
        _repository.AddPlan(plan);
        _paymentGateway.NextChargeResult =
            Result<Guid>.Success(Guid.NewGuid());

        var handler = CreateHandler();

        // Act: renew at the current clock time.
        var result1 = await handler.HandleAsync(
            new RenewSubscriptionCommand(subscription.Id, plan.Id),
            CancellationToken.None);

        // Advance the clock by 10 days.
        _clock.Advance(TimeSpan.FromDays(10));

        // Reset the repository to the original state for a second renewal.
        _repository.Clear();
        _repository.Add(CreateSubscription(
            plan, expiresAt: _clock.UtcNow.AddDays(5)));
        _repository.AddPlan(plan);

        var result2 = await handler.HandleAsync(
            new RenewSubscriptionCommand(subscription.Id, plan.Id),
            CancellationToken.None);

        // Assert: the two expirations differ by exactly 10 days.
        Assert.True(result1.IsSuccess);
        Assert.True(result2.IsSuccess);
        Assert.Equal(
            TimeSpan.FromDays(10),
            result2.Value!.NewExpiration - result1.Value!.NewExpiration);
    }

    // ── Helper methods ──────────────────────────────────────────

    private RenewSubscriptionHandler CreateHandler()
    {
        var mapper = new SubscriptionMapper();
        var saga = new SagaOrchestrator<RenewalSagaContext>(
            new ISagaStep<RenewalSagaContext>[]
            {
                new ProcessPaymentStep(_paymentGateway),
                new UpdateSubscriptionStep(
                    _repository, mapper,
                    new InMemoryDbContext(_outbox)),
                new SendConfirmationStep(
                    new InMemoryCustomerRepository(),
                    _emailService)
            });

        return new RenewSubscriptionHandler(
            _repository, mapper, _clock, saga);
    }

    private Subscription CreateSubscription(
        Plan plan, DateTimeOffset expiresAt)
    {
        return new Subscription
        {
            Id = Guid.NewGuid(),
            CustomerId = Guid.NewGuid(),
            Plan = plan,
            ExpiresAt = expiresAt,
            Status = SubscriptionStatus.Active
        };
    }
}

[HttpPost("renew")]
public async Task<IActionResult> Renew([FromBody] RenewRequest request)
{
    if (string.IsNullOrEmpty(request.SubscriptionId))
        return BadRequest("SubscriptionId is required.");
    if (string.IsNullOrEmpty(request.RequestedPlanId))
        return BadRequest("RequestedPlanId is required.");

    if (!Guid.TryParse(request.SubscriptionId, out var subId))
        return BadRequest("Invalid subscription ID.");
    if (!Guid.TryParse(request.RequestedPlanId, out var planId))
        return BadRequest("Invalid plan ID.");

    var entity = await _db.Subscriptions.FindAsync(subId);
    if (entity == null) return NotFound("Subscription not found.");

    var plan = await _db.Plans.FindAsync(planId);
    if (plan == null) return NotFound("Plan not found.");

    // Manual mapping -- 10 property assignments
    var subscription = new Subscription
    {
        Id = entity.Id,
        CustomerId = entity.CustomerId,
        // ... 8 more properties ...
    };

    // Inline business logic
    string renewalType;
    DateTimeOffset newExpiration;
    decimal charge;

    if (subscription.Plan.Tier == plan.Tier)
    {
        renewalType = "Standard";
        newExpiration = DateTimeOffset.UtcNow.AddMonths(plan.TermMonths);
        charge = plan.MonthlyPrice * plan.TermMonths;
    }
    else if (plan.Tier > subscription.Plan.Tier)
    {
        renewalType = "Upgrade";
        newExpiration = DateTimeOffset.UtcNow.AddMonths(plan.TermMonths);
        // Pro-rated calculation inline...
        charge = /* complex formula */;
    }
    else
    {
        renewalType = "Downgrade";
        newExpiration = subscription.ExpiresAt;
        charge = 0;
    }

    // Inline orchestration with try-catch
    Guid? paymentId = null;
    try
    {
        if (charge > 0)
        {
            paymentId = await _paymentGateway.ChargeAsync(
                subscription.CustomerId, charge);
        }

        subscription.Plan = plan;
        subscription.ExpiresAt = newExpiration;
        await _db.SaveChangesAsync();

        // Publish event manually -- dual-write problem!
        await _messageBroker.PublishAsync(
            new SubscriptionRenewed(/* ... */));

        await _emailService.SendConfirmationAsync(/* ... */);
    }
    catch (Exception ex)
    {
        // Manual compensation
        if (paymentId.HasValue)
        {
            try { await _paymentGateway.RefundAsync(paymentId.Value); }
            catch { /* log and hope */ }
        }
        return StatusCode(500, ex.Message);
    }

    return Ok(new { renewalType, newExpiration, charge });
}