CQRS Pipeline

/// <summary>
/// Discriminated union for operation results.
/// Eliminates exceptions for expected failure cases.
/// </summary>
public class Result<T>
{
    public bool IsSuccess { get; }
    public bool IsFailure => !IsSuccess;
    public T Value { get; }           // Only valid when IsSuccess
    public AppError Error { get; }     // Only valid when IsFailure

    // Factory methods
    public static Result<T> Success(T value) => new(value);
    public static Result<T> Failure(AppError error) => new(error);

    // Implicit conversions
    public static implicit operator Result<T>(T value) => Success(value);
    public static implicit operator Result<T>(AppError error) => Failure(error);
}

/// <summary>
/// Structured error with code, message, and optional details.
/// </summary>
public record AppError(
    string Code,           // e.g., "Admin.NotFound"
    string Message,        // e.g., "Admin with ID {id} was not found"
    object? Details = null  // Optional additional context
);

[HttpGet("{id}")]
public async Task<IActionResult> GetAdmin(string id)
{
    var result = await _sender.Send(new GetAdminByIdQuery(id));

    return result.IsSuccess
        ? Ok(result.Value)         // 200 with data
        : NotFound(new {            // 404 with error
            error = result.Error.Message
        });
}

/// <summary>
/// AstraFlow mediator pipeline behavior that runs FluentValidation validators
/// BEFORE the request handler executes.
/// If validation fails, returns Result.Failure without hitting the handler.
/// </summary>
public class ValidationBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : AstraFlow.Mediator.IRequest<TResponse>
{
    private readonly IEnumerable<IValidator<TRequest>> _validators;

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

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        if (!_validators.Any())
            return await next(); // No validators registered

        var context = new ValidationContext<TRequest>(request);

        var failures = _validators
            .Select(v => v.Validate(context))
            .SelectMany(result => result.Errors)
            .Where(failure => failure != null)
            .ToList();

        if (failures.Any())
        {
            // Return validation error without executing handler
            var errors = failures.Select(f => new {
                Field = f.PropertyName,
                Message = f.ErrorMessage
            });
            throw new ValidationException(failures);
        }

        return await next(); // All valid — proceed to handler
    }
}

/// <summary>
/// Pipeline behavior that logs every AstraFlow mediator request with timing.
/// Logs: request type, user ID, tenant ID, execution time, and outcome.
/// Warns if execution exceeds 500ms threshold.
/// </summary>
public class LoggingBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : AstraFlow.Mediator.IRequest<TResponse>
{
    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        var requestName = typeof(TRequest).Name;
        var userId = _currentUser.GetUserId();
        var tenantId = _currentUser.GetTenantId();

        _logger.LogInformation(
            "Handling {RequestName} | User: {UserId} | Tenant: {TenantId}",
            requestName, userId, tenantId);

        var stopwatch = Stopwatch.StartNew();
        var response = await next();
        stopwatch.Stop();

        var elapsed = stopwatch.ElapsedMilliseconds;

        if (elapsed > 500) // Slow request warning
        {
            _logger.LogWarning(
                "SLOW: {RequestName} took {ElapsedMs}ms | User: {UserId}",
                requestName, elapsed, userId);
        }
        else
        {
            _logger.LogInformation(
                "Completed {RequestName} in {ElapsedMs}ms",
                requestName, elapsed);
        }

        return response;
    }
}

/// <summary>
/// Pipeline behavior that handles query caching (ICacheable) and mutation invalidation (IInvalidatesCache).
/// Integrates with the cache stampede prevention, eviction controls, and tenant feature token caching.
/// </summary>
public class CachingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
    where TRequest : AstraFlow.Mediator.IRequest<TResponse>
{
    private readonly ICacheService _cache;
    private readonly ICurrentUser _currentUser;
    private readonly ILogger<CachingBehavior<TRequest, TResponse>> _logger;

    public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next, CancellationToken ct)
    {
        // 1. Invalidation Check (IInvalidatesCache) for Mutations
        if (request is IInvalidatesCache invalidator)
        {
            var response = await next();
            
            // Evict exact keys and prefix-based namespaces (e.g., Module regions)
            foreach (var key in invalidator.CacheKeysToInvalidate)
                await _cache.RemoveAsync(key, ct);
                
            foreach (var prefix in invalidator.CachePrefixesToInvalidate)
                await _cache.RemoveByPrefixAsync(prefix, ct);
                
            return response;
        }

        // 2. Cache Read Check (ICacheable) for Queries
        if (request is not ICacheable cacheable)
            return await next();

        // Build scoped cache key to prevent cross-tenant/cross-user data leaks
        var scopedKey = BuildScopedCacheKey(cacheable.CacheKey);
        
        // Retrieve or generate with stampede prevention locks and double-checked locking
        return await _cache.GetOrCreateAsync(
            scopedKey,
            async () => await next(),
            cacheable.Expiration ?? TimeSpan.FromMinutes(30),
            ct);
    }

    private string BuildScopedCacheKey(string baseKey)
    {
        var tenantId = _currentUser.EffectiveTenantId ?? "platform";
        var userId = _currentUser.Id ?? "anonymous";
        var role = _currentUser.ActorRole ?? "user";
        var hierarchy = _currentUser.IsHierarchyScope ? "hierarchy" : "own";
        
        return $"{baseKey}::scope:t={tenantId};u={userId};a={role};h={hierarchy}";
    }
}

/// <summary>
/// Implement this interface on queries to enable caching.
/// </summary>
public interface ICacheable
{
    string CacheKey { get; }
    TimeSpan? Expiration { get; }
}

/// <summary>
/// Implement this interface on commands to invalidate caching regions.
/// </summary>
public interface IInvalidatesCache
{
    string[] CacheKeysToInvalidate { get; }
    string[] CachePrefixesToInvalidate { get; }
}

// MemoryCacheService implementation details for Stampede Prevention and Key Eviction
public class MemoryCacheService : ICacheService
{
    private readonly IMemoryCache _memoryCache;
    private readonly ConcurrentDictionary<string, SemaphoreSlim> _locks = new();
    private readonly ConcurrentDictionary<string, byte> _keys = new();
    private const int MaxTrackedKeys = 10000;

    public async Task<T> GetOrCreateAsync<T>(string key, Func<Task<T>> factory, TimeSpan expiration, CancellationToken ct)
    {
        // Check cache first (fast path)
        if (_memoryCache.TryGetValue(key, out T? value))
            return value!;

        // Acquire key-specific lock to prevent Cache Stampede
        var semaphore = _locks.GetOrAdd(key, _ => new SemaphoreSlim(1, 1));
        await semaphore.WaitAsync(ct);
        try
        {
            // Double-checked lock: re-query cache
            if (_memoryCache.TryGetValue(key, out value))
                return value!;

            // Execute factory (DB/API call)
            value = await factory();
            
            // Unbounded Growth Prevention: evict oldest keys if over limit
            if (_keys.Count >= MaxTrackedKeys)
                EvictOldestKeys();

            // Set in cache with PostEvictionCallback to clean up semaphores
            var cacheEntryOptions = new MemoryCacheEntryOptions()
                .SetAbsoluteExpiration(expiration)
                .RegisterPostEvictionCallback((k, v, reason, state) =>
                {
                    _keys.TryRemove((string)k, out _);
                    if (_locks.TryRemove((string)k, out var sem))
                        sem.Dispose();
                });

            _memoryCache.Set(key, value, cacheEntryOptions);
            _keys.TryAdd(key, 0);

            return value;
        }
        finally
        {
            semaphore.Release();
        }
    }
}

// Entitlements Module Feature Cache Global Eviction
public class FeatureCache : IFeatureCache
{
    private readonly IMemoryCache _memoryCache;
    private CancellationTokenSource _globalEvictionSource = new();

    public void InvalidateAll()
    {
        // Cancel the global token source to immediately invalidate all cached feature sets
        var oldSource = Interlocked.Exchange(ref _globalEvictionSource, new CancellationTokenSource());
        oldSource.Cancel();
        oldSource.Dispose();
    }

    public void CacheTenantFeatures(Guid tenantId, Dictionary<string, string> features)
    {
        var cacheKey = $"tenant:{tenantId}:features";
        var options = new MemoryCacheEntryOptions()
            .SetSlidingExpiration(TimeSpan.FromMinutes(10))
            .SetAbsoluteExpiration(TimeSpan.FromMinutes(30))
            // Bind to the global eviction token
            .AddExpirationToken(new CancellationChangeToken(_globalEvictionSource.Token));

        _memoryCache.Set(cacheKey, features, options);
    }
}

public static IServiceCollection AddCoreApplication(
    this IServiceCollection services,
    IConfiguration? configuration = null,
    params Type[] handlerAssemblyMarkerTypes)
{
    var mediatorOptions = configuration?
        .GetSection(MediatorOptions.SectionName)
        .Get<MediatorOptions>() ?? new MediatorOptions();

    services.AddScripeMediator(
        validateRequestCoverage: mediatorOptions.ValidateRequestHandlerCoverage,
        assemblyMarkerTypes: handlerAssemblyMarkerTypes);

    RegisterConfiguredPipelineBehaviors(services, mediatorOptions);

    foreach (var markerType in handlerAssemblyMarkerTypes)
        services.AddValidatorsFromAssemblyContaining(markerType);

    return services;
}

// OutboxInterceptor intercepts SavingChangesAsync
public class OutboxInterceptor : SaveChangesInterceptor
{
    public override async ValueTask<InterceptionResult<int>> SavingChangesAsync(
        DbContextEventData eventData,
        InterceptionResult<int> result,
        CancellationToken ct = default)
    {
        var context = eventData.Context!;
        
        // 1. Capture events from aggregate roots implementing IHasDomainEvents
        var entities = context.ChangeTracker.Entries<IHasDomainEvents>()
            .Where(e => e.Entity.DomainEvents.Any())
            .ToList();
            
        var domainEvents = entities.SelectMany(e => e.Entity.DomainEvents).ToList();
        
        // 2. Clear events from entities to avoid double processing
        entities.ForEach(e => e.Entity.ClearDomainEvents());
        
        // 3. Serialize events to OutboxMessage entities
        var outboxMessages = domainEvents.Select(evt => new OutboxMessage
        {
            Id = Guid.NewGuid(),
            Type = evt.GetType().AssemblyQualifiedName!,
            Content = JsonSerializer.Serialize(evt, evt.GetType()),
            OccurredOnUtc = DateTime.UtcNow
        }).ToList();
        
        // 4. Save messages in the same database transaction
        await context.Set<OutboxMessage>().AddRangeAsync(outboxMessages, ct);
        
        return await base.SavingChangesAsync(eventData, result, ct);
    }
}

// OutboxProcessorJob polls and processes messages every minute
public class OutboxProcessorJob : IAutoRegisteredJob
{
    public string JobId => "core-outbox-processor";
    public string CronExpression => "* * * * *"; // Every minute

    public async Task ExecuteAsync(CancellationToken ct = default)
    {
        foreach (var dbContext in _registry.GetContexts())
        {
            var messages = await dbContext.OutboxMessages
                .Where(m => m.ProcessedOnUtc == null && m.RetryCount < 5)
                .OrderBy(m => m.OccurredOnUtc)
                .Take(50)
                .ToListAsync(ct);
                
            foreach (var message in messages)
            {
                try
                {
                    var type = Type.GetType(message.Type)!;
                    var domainEvent = JsonSerializer.Deserialize(message.Content, type);
                    
                    // Dispatch locally via Mediator and externally via EventBus
                    await _publisher.PublishAsync(domainEvent, ct);
                    
                    message.ProcessedOnUtc = DateTime.UtcNow;
                }
                catch (Exception ex)
                {
                    message.Error = ex.Message;
                    message.RetryCount++;
                }
            }
            await dbContext.SaveChangesAsync(ct);
        }
    }
}

// OutboxCleanupJob runs daily to purge old processed messages
public class OutboxCleanupJob : RecurringJobBase
{
    public override string JobId => "core-outbox-cleanup";
    public override string CronExpression => "0 5 * * *"; // Daily at 5 AM UTC

    protected override async Task ExecuteJobAsync(CancellationToken ct)
    {
        var cutoff = DateTime.UtcNow.AddDays(-7);
        foreach (var dbContext in _registry.GetContexts())
        {
            // Retrieve IDs to delete and use EF Core Attach/RemoveRange to avoid full load
            var oldMessageIds = await dbContext.OutboxMessages
                .Where(m => m.ProcessedOnUtc != null && m.ProcessedOnUtc < cutoff)
                .Select(m => m.Id)
                .Take(1000)
                .ToListAsync(ct);
                
            var stubs = oldMessageIds.Select(id => new OutboxMessage { Id = id });
            dbContext.OutboxMessages.RemoveRange(stubs);
            await dbContext.SaveChangesAsync(ct);
        }
    }
}