300 lines
14 KiB
C#
300 lines
14 KiB
C#
using System.Text.Json;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Logging;
|
|
using Tiku.Application.Content;
|
|
using Tiku.Application.Learning;
|
|
using Tiku.Domain.Content;
|
|
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 const string ContentReleasePublished = "content_release_published";
|
|
internal const string LocalContentReleaseProjection = "local_content_release_v1";
|
|
}
|
|
|
|
internal sealed class LearningOutboxProcessor(
|
|
ILearningPersistence learningPersistence,
|
|
IQuestionBankPersistence contentPersistence,
|
|
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 projectionName = ResolveProjectionName(message);
|
|
var alreadyProjected = await learningPersistence.LearningProjectionReceipts.AnyAsync(
|
|
item => item.TenantId == message.TenantId &&
|
|
item.OutboxMessageId == message.Id &&
|
|
item.ProjectionName == projectionName,
|
|
cancellationToken);
|
|
if (!alreadyProjected)
|
|
{
|
|
await DispatchAsync(message, cancellationToken);
|
|
learningPersistence.LearningProjectionReceipts.Add(new LearningProjectionReceipt
|
|
{
|
|
TenantId = message.TenantId,
|
|
OutboxMessageId = message.Id,
|
|
ProjectionName = projectionName,
|
|
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 (message.SchemaVersion != 1)
|
|
throw UnsupportedEvent(message);
|
|
|
|
if (string.Equals(message.EventType, LearningOutboxEventTypes.PracticeSessionSubmitted,
|
|
StringComparison.Ordinal))
|
|
{
|
|
await ProjectWrongQuestionsAsync(message, cancellationToken);
|
|
return;
|
|
}
|
|
|
|
if (string.Equals(message.EventType, LearningOutboxEventTypes.ContentReleasePublished,
|
|
StringComparison.Ordinal))
|
|
{
|
|
await VerifyLocalContentReleaseAsync(message, cancellationToken);
|
|
return;
|
|
}
|
|
|
|
throw UnsupportedEvent(message);
|
|
}
|
|
|
|
private async Task ProjectWrongQuestionsAsync(
|
|
LearningOutboxMessage message,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
|
|
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.Id))
|
|
.Select(item => new
|
|
{
|
|
item.QuestionOwnerTenantId,
|
|
item.QuestionAssetId,
|
|
item.QuestionRevisionId,
|
|
item.QuestionPlacementId
|
|
})
|
|
.ToArrayAsync(cancellationToken);
|
|
if (questions.Length != wrongQuestionIds.Distinct().Count())
|
|
throw new InvalidOperationException("One or more V2 wrong-question deliveries could not be projected.");
|
|
|
|
var v2 = questions
|
|
.DistinctBy(item => new { item.QuestionOwnerTenantId, item.QuestionAssetId }).ToArray();
|
|
if (v2.Length > 0)
|
|
{
|
|
var ownerTenantIds = v2.Select(item => item.QuestionOwnerTenantId).ToArray();
|
|
var assetIds = v2.Select(item => item.QuestionAssetId).ToArray();
|
|
var revisionIds = v2.Select(item => item.QuestionRevisionId).ToArray();
|
|
var placementIds = v2.Select(item => item.QuestionPlacementId).ToArray();
|
|
await learningPersistence.Database.ExecuteSqlInterpolatedAsync($"""
|
|
INSERT INTO wrong_questions
|
|
(tenant_id, user_id, question_asset_owner_tenant_id, question_asset_id,
|
|
last_question_revision_id, last_question_placement_id, wrong_count, last_wrong_at, resolved_at)
|
|
SELECT {message.TenantId}, {payload.UserId}, source.owner_tenant_id, source.asset_id,
|
|
source.revision_id, source.placement_id, 1, {report.SubmittedAt}, NULL
|
|
FROM unnest({ownerTenantIds}, {assetIds}, {revisionIds}, {placementIds})
|
|
AS source(owner_tenant_id, asset_id, revision_id, placement_id)
|
|
ON CONFLICT (tenant_id, user_id, question_asset_owner_tenant_id, question_asset_id) DO UPDATE SET
|
|
last_question_revision_id = EXCLUDED.last_question_revision_id,
|
|
last_question_placement_id = EXCLUDED.last_question_placement_id,
|
|
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 VerifyLocalContentReleaseAsync(
|
|
LearningOutboxMessage message,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var payload = message.Payload.Deserialize<ContentReleasePublishedPayload>(JsonSerializerOptions.Web)
|
|
?? throw new InvalidOperationException("The content release outbox payload is invalid.");
|
|
if (payload.ReleaseId != message.AggregateId)
|
|
throw new InvalidOperationException("The content release outbox aggregate does not match its payload.");
|
|
|
|
var release = await contentPersistence.ContentReleases.AsNoTracking().SingleOrDefaultAsync(
|
|
item => item.TenantId == message.TenantId &&
|
|
item.Id == payload.ReleaseId &&
|
|
item.CurriculumVersionId == payload.CurriculumVersionId &&
|
|
item.BusinessLineId == payload.BusinessLineId &&
|
|
item.ReleaseNo == payload.ReleaseNo &&
|
|
item.SourceFingerprint == payload.SourceFingerprint &&
|
|
item.Status == ContentReleaseStatus.Published,
|
|
cancellationToken) ?? throw new InvalidOperationException(
|
|
"The published content release is missing or differs from its outbox payload.");
|
|
|
|
var candidateCount = await contentPersistence.ContentReleaseQuestions.AsNoTracking().CountAsync(
|
|
item => item.TenantId == message.TenantId &&
|
|
item.ContentReleaseId == release.Id &&
|
|
item.Status == ContentReleaseQuestionStatus.Active,
|
|
cancellationToken);
|
|
if (candidateCount == 0)
|
|
throw new InvalidOperationException("The published content release contains no active candidates.");
|
|
}
|
|
|
|
private static string ResolveProjectionName(LearningOutboxMessage message)
|
|
{
|
|
if (message.SchemaVersion != 1)
|
|
throw UnsupportedEvent(message);
|
|
|
|
return message.EventType switch
|
|
{
|
|
LearningOutboxEventTypes.PracticeSessionSubmitted =>
|
|
LearningOutboxEventTypes.WrongQuestionProjection,
|
|
LearningOutboxEventTypes.ContentReleasePublished =>
|
|
LearningOutboxEventTypes.LocalContentReleaseProjection,
|
|
_ => throw UnsupportedEvent(message)
|
|
};
|
|
}
|
|
|
|
private static InvalidOperationException UnsupportedEvent(LearningOutboxMessage message)
|
|
{
|
|
return new InvalidOperationException(
|
|
$"Unsupported learning outbox event '{message.EventType}' schema {message.SchemaVersion}.");
|
|
}
|
|
|
|
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);
|
|
|
|
private sealed record ContentReleasePublishedPayload(
|
|
Guid ReleaseId,
|
|
Guid CurriculumVersionId,
|
|
Guid BusinessLineId,
|
|
int ReleaseNo,
|
|
string SourceFingerprint);
|
|
}
|