tiku-backend.net/Tiku.Infrastructure/Learning/Outbox/LearningOutboxProcessor.cs

using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Tiku.Application.Learning;
using Tiku.Domain.Learning;
using Tiku.Infrastructure.Persistence;

namespace Tiku.Infrastructure.Learning;

internal static class LearningOutboxEventTypes
{
    internal const string PracticeSessionSubmitted = "practice_session_submitted";
    internal const string WrongQuestionProjection = "wrong_questions_v1";
}

internal sealed class LearningOutboxProcessor(
    ILearningPersistence learningPersistence,
    ILogger<LearningOutboxProcessor> logger) : ILearningOutboxProcessor
{
    private const int MaxAttempts = 10;
    private static readonly TimeSpan LeaseDuration = TimeSpan.FromMinutes(2);

    public async Task<int> ProcessPendingAsync(
        string workerId,
        int batchSize = 100,
        CancellationToken cancellationToken = default)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(workerId);
        var processed = 0;
        var limit = Math.Clamp(batchSize, 1, 1000);
        for (var index = 0; index < limit; index++)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var messageId = await TryClaimNextAsync(workerId, cancellationToken);
            if (!messageId.HasValue) break;

            learningPersistence.ChangeTracker.Clear();
            if (await ProcessClaimedAsync(messageId.Value, workerId, cancellationToken)) processed++;
            learningPersistence.ChangeTracker.Clear();
        }

        return processed;
    }

    private async Task<Guid?> TryClaimNextAsync(string workerId, CancellationToken cancellationToken)
    {
        var now = DateTimeOffset.UtcNow;
        var lockExpiresAt = now.Add(LeaseDuration);
        var ids = await learningPersistence.Database.SqlQuery<Guid>($"""
            UPDATE learning_outbox_messages AS message
            SET locked_by = {workerId},
                lock_expires_at = {lockExpiresAt},
                attempt_count = message.attempt_count + 1
            WHERE message.id = (
                SELECT candidate.id
                FROM learning_outbox_messages AS candidate
                WHERE candidate.processed_at IS NULL
                  AND candidate.dead_lettered_at IS NULL
                  AND candidate.available_at <= {now}
                  AND (candidate.lock_expires_at IS NULL OR candidate.lock_expires_at <= {now})
                ORDER BY candidate.occurred_at, candidate.id
                FOR UPDATE SKIP LOCKED
                LIMIT 1
            )
            RETURNING message.id AS "Value"
            """).ToArrayAsync(cancellationToken);
        return ids.Length == 0 ? null : ids[0];
    }

    private async Task<bool> ProcessClaimedAsync(
        Guid messageId,
        string workerId,
        CancellationToken cancellationToken)
    {
        await using var transaction = await learningPersistence.Database.BeginTransactionAsync(cancellationToken);
        LearningOutboxMessage? message = null;
        try
        {
            message = await learningPersistence.LearningOutboxMessages.SingleOrDefaultAsync(
                item => item.Id == messageId && item.LockedBy == workerId && item.ProcessedAt == null,
                cancellationToken);
            if (message is null)
            {
                await transaction.RollbackAsync(cancellationToken);
                return false;
            }

            var alreadyProjected = await learningPersistence.LearningProjectionReceipts.AnyAsync(
                item => item.TenantId == message.TenantId &&
                        item.OutboxMessageId == message.Id &&
                        item.ProjectionName == LearningOutboxEventTypes.WrongQuestionProjection,
                cancellationToken);
            if (!alreadyProjected)
            {
                await DispatchAsync(message, cancellationToken);
                learningPersistence.LearningProjectionReceipts.Add(new LearningProjectionReceipt
                {
                    TenantId = message.TenantId,
                    OutboxMessageId = message.Id,
                    ProjectionName = LearningOutboxEventTypes.WrongQuestionProjection,
                    ProcessedAt = DateTimeOffset.UtcNow
                });
            }

            message.ProcessedAt = DateTimeOffset.UtcNow;
            message.LockedBy = null;
            message.LockExpiresAt = null;
            message.LastError = null;
            await learningPersistence.SaveChangesAsync(cancellationToken);
            await transaction.CommitAsync(cancellationToken);
            return true;
        }
        catch (OperationCanceledException)
        {
            await transaction.RollbackAsync(CancellationToken.None);
            throw;
        }
        catch (Exception exception)
        {
            await transaction.RollbackAsync(CancellationToken.None);
            learningPersistence.ChangeTracker.Clear();
            await ReleaseAfterFailureAsync(messageId, workerId, message?.AttemptCount ?? 1, exception,
                cancellationToken);
            return false;
        }
    }

    private async Task DispatchAsync(LearningOutboxMessage message, CancellationToken cancellationToken)
    {
        if (!string.Equals(message.EventType, LearningOutboxEventTypes.PracticeSessionSubmitted,
                StringComparison.Ordinal) || message.SchemaVersion != 1)
            throw new InvalidOperationException(
                $"Unsupported learning outbox event '{message.EventType}' schema {message.SchemaVersion}.");

        var payload = message.Payload.Deserialize<PracticeSessionSubmittedPayload>(JsonSerializerOptions.Web)
                      ?? throw new InvalidOperationException("The practice submission outbox payload is invalid.");
        if (payload.ReportId != message.AggregateId)
            throw new InvalidOperationException("The practice submission outbox aggregate does not match its payload.");

        var report = await learningPersistence.PracticeSessionReports.AsNoTracking().SingleOrDefaultAsync(
            item => item.TenantId == message.TenantId &&
                    item.Id == payload.ReportId &&
                    item.PracticeSessionId == payload.PracticeSessionId &&
                    item.UserId == payload.UserId,
            cancellationToken)
                     ?? throw new InvalidOperationException("The submitted practice report no longer exists.");
        var wrongQuestionIds = report.WrongQuestionIds.Deserialize<Guid[]>() ?? [];
        if (wrongQuestionIds.Length == 0) return;

        var questions = await learningPersistence.PracticeSessionQuestions.AsNoTracking()
            .Where(item => item.TenantId == message.TenantId &&
                           item.PracticeSessionId == payload.PracticeSessionId &&
                           wrongQuestionIds.Contains(item.QuestionReferenceId))
            .Select(item => new
            {
                item.QuestionReferenceId,
                item.QuestionOwnerTenantId,
                item.QuestionId
            })
            .ToArrayAsync(cancellationToken);
        if (questions.Length != wrongQuestionIds.Distinct().Count())
            throw new InvalidOperationException("One or more wrong-question references could not be projected.");

        var referenceIds = questions.Select(item => item.QuestionReferenceId).ToArray();
        var ownerTenantIds = questions.Select(item => item.QuestionOwnerTenantId).ToArray();
        var questionIds = questions.Select(item => item.QuestionId).ToArray();
        await learningPersistence.Database.ExecuteSqlInterpolatedAsync($"""
            INSERT INTO wrong_questions
                (tenant_id, user_id, question_reference_id, question_owner_tenant_id, question_id,
                 wrong_count, last_wrong_at, resolved_at)
            SELECT {message.TenantId}, {payload.UserId}, source.question_reference_id,
                   source.question_owner_tenant_id, source.question_id, 1, {report.SubmittedAt}, NULL
            FROM unnest({referenceIds}, {ownerTenantIds}, {questionIds})
                AS source(question_reference_id, question_owner_tenant_id, question_id)
            ON CONFLICT (tenant_id, user_id, question_reference_id) DO UPDATE SET
                wrong_count = wrong_questions.wrong_count + 1,
                last_wrong_at = GREATEST(wrong_questions.last_wrong_at, EXCLUDED.last_wrong_at),
                resolved_at = NULL
            """, cancellationToken);
    }

    private async Task ReleaseAfterFailureAsync(
        Guid messageId,
        string workerId,
        int attemptCount,
        Exception exception,
        CancellationToken cancellationToken)
    {
        var now = DateTimeOffset.UtcNow;
        var deadLetteredAt = attemptCount >= MaxAttempts ? now : (DateTimeOffset?)null;
        var availableAt = now.AddSeconds(Math.Min(300, 2 << Math.Min(attemptCount, 7)));
        var error = exception.Message.Length <= 2000 ? exception.Message : exception.Message[..2000];
        await learningPersistence.LearningOutboxMessages
            .Where(item => item.Id == messageId && item.LockedBy == workerId && item.ProcessedAt == null)
            .ExecuteUpdateAsync(setters => setters
                    .SetProperty(item => item.LockedBy, (string?)null)
                    .SetProperty(item => item.LockExpiresAt, (DateTimeOffset?)null)
                    .SetProperty(item => item.AvailableAt, availableAt)
                    .SetProperty(item => item.DeadLetteredAt, deadLetteredAt)
                    .SetProperty(item => item.LastError, error),
                cancellationToken);
        logger.LogWarning(exception,
            "Learning outbox message {MessageId} failed on attempt {AttemptCount}.", messageId, attemptCount);
    }

    private sealed record PracticeSessionSubmittedPayload(Guid ReportId, Guid PracticeSessionId, Guid UserId);
}