using MongoDB.Driver; using QuestionsApi.Models; namespace QuestionsApi.Services; public sealed class ReviewQueueStore { private readonly IMongoCollection _items; private readonly IMongoCollection _reviews; private readonly IMongoCollection _firstAttempts; private readonly IMongoCollection _backfills; private readonly QuestionStore _questions; public ReviewQueueStore(MongoDatabaseProvider mongo, QuestionStore questions) { _items = mongo.Database.GetCollection("ReviewQueueItems"); _reviews = mongo.Database.GetCollection("QuestionReviewAttempts"); _firstAttempts = mongo.Database.GetCollection("QuestionAttempts"); _backfills = mongo.Database.GetCollection("ReviewQueueBackfillStates"); _questions = questions; _items.Indexes.CreateMany([ new CreateIndexModel( Builders.IndexKeys.Ascending(value => value.UserId).Ascending(value => value.QuestionExternalId), new CreateIndexOptions { Unique = true, Name = "review_queue_unique_user_question" }), new CreateIndexModel( Builders.IndexKeys.Ascending(value => value.UserId).Ascending(value => value.State), new CreateIndexOptions { Name = "review_queue_user_state" }) ]); _reviews.Indexes.CreateMany([ new CreateIndexModel( Builders.IndexKeys.Ascending(value => value.UserId).Ascending(value => value.QuestionExternalId).Ascending(value => value.Cycle), new CreateIndexOptions { Unique = true, Name = "review_attempt_unique_cycle" }), new CreateIndexModel( Builders.IndexKeys.Ascending(value => value.UserId).Descending(value => value.CreatedUtc), new CreateIndexOptions { Name = "review_attempt_user_created" }) ]); } public async Task AddFromFirstAttemptAsync(QuestionAttempt attempt) { if (!ShouldQueue(attempt)) return; var now = DateTime.UtcNow; var update = Builders.Update .SetOnInsert(value => value.UserId, attempt.UserId) .SetOnInsert(value => value.QuestionExternalId, attempt.QuestionExternalId) .SetOnInsert(value => value.SourceAttemptId, attempt.Id ?? "") .SetOnInsert(value => value.State, ReviewQueueStates.Pending) .SetOnInsert(value => value.Cycle, 1) .SetOnInsert(value => value.CreatedUtc, attempt.CreatedUtc) .SetOnInsert(value => value.UpdatedUtc, now); try { await _items.UpdateOneAsync( value => value.UserId == attempt.UserId && value.QuestionExternalId == attempt.QuestionExternalId, update, new UpdateOptions { IsUpsert = true }); } catch (MongoWriteException exception) when (exception.WriteError.Category == ServerErrorCategory.DuplicateKey) { // Another API replica created the same queue item concurrently. } } public async Task EnsureUserQueueAsync(string userId) { var checkpoint = await _backfills.Find(value => value.UserId == userId).FirstOrDefaultAsync(); var scanThroughUtc = DateTime.UtcNow; var filter = Builders.Filter.Where(value => value.UserId == userId && !value.IsCorrect && !value.IsSkip); if (checkpoint is not null) filter &= Builders.Filter.Gte(value => value.CreatedUtc, checkpoint.CompletedUtc); filter &= Builders.Filter.Lte(value => value.CreatedUtc, scanThroughUtc); var misses = await _firstAttempts.Find(filter).ToListAsync(); if (misses.Count > 0) { var now = DateTime.UtcNow; var writes = misses.Select(attempt => new UpdateOneModel( Builders.Filter.Where(value => value.UserId == userId && value.QuestionExternalId == attempt.QuestionExternalId), Builders.Update .SetOnInsert(value => value.UserId, userId) .SetOnInsert(value => value.QuestionExternalId, attempt.QuestionExternalId) .SetOnInsert(value => value.SourceAttemptId, attempt.Id ?? "") .SetOnInsert(value => value.State, ReviewQueueStates.Pending) .SetOnInsert(value => value.Cycle, 1) .SetOnInsert(value => value.CreatedUtc, attempt.CreatedUtc) .SetOnInsert(value => value.UpdatedUtc, now)) { IsUpsert = true }).ToList(); try { await _items.BulkWriteAsync(writes, new BulkWriteOptions { IsOrdered = false }); } catch (MongoBulkWriteException exception) when (exception.WriteErrors.All(error => error.Category == ServerErrorCategory.DuplicateKey)) { // Concurrent lazy backfills are equivalent and protected by the unique index. } } try { await _backfills.UpdateOneAsync( value => value.UserId == userId, Builders.Update .SetOnInsert(value => value.UserId, userId) .Max(value => value.CompletedUtc, scanThroughUtc), new UpdateOptions { IsUpsert = true }); } catch (MongoWriteException exception) when (exception.WriteError.Category == ServerErrorCategory.DuplicateKey) { // Another replica completed the same per-user migration. } } public Task CountPendingAsync(string userId) => _items.CountDocumentsAsync(value => value.UserId == userId && value.State == ReviewQueueStates.Pending); public async Task GetRandomPendingAsync(string userId, IEnumerable excludedIds) { var excluded = excludedIds.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToList(); for (var pass = 0; pass < 5; pass++) { var filter = Builders.Filter.Eq(value => value.UserId, userId) & Builders.Filter.Eq(value => value.State, ReviewQueueStates.Pending); if (excluded.Count > 0) filter &= Builders.Filter.Nin(value => value.QuestionExternalId, excluded); var candidates = await _items.Aggregate().Match(filter).Sample(20).ToListAsync(); if (candidates.Count == 0) return null; foreach (var candidate in candidates) { var question = await _questions.GetByExternalIdAsync(candidate.QuestionExternalId); if (question is { IsPublished: true, FlaggedForReviewUtc: null }) return candidate; await _items.UpdateOneAsync( value => value.Id == candidate.Id && value.State == ReviewQueueStates.Pending, Builders.Update .Set(value => value.State, ReviewQueueStates.Unavailable) .Set(value => value.UpdatedUtc, DateTime.UtcNow)); } } return null; } public async Task RecordReviewAsync( string userId, string clientAttemptId, string questionExternalId, int cycle, string choiceId, IReadOnlyCollection presentedChoiceIds, bool isCorrect, int durationSeconds) { var item = await _items.Find(value => value.UserId == userId && value.QuestionExternalId == questionExternalId && value.State == ReviewQueueStates.Pending && value.Cycle == cycle).FirstOrDefaultAsync(); if (item is null) { var prior = await _reviews.Find(value => value.UserId == userId && value.QuestionExternalId == questionExternalId && value.Cycle == cycle).FirstOrDefaultAsync(); if (prior is null) return null; return new ReviewQueueRecordResult { Attempt = prior, Item = await FindItemAsync(userId, questionExternalId) }; } var attempt = new QuestionReviewAttempt { Id = clientAttemptId, UserId = userId, QuestionExternalId = questionExternalId, Cycle = cycle, ChoiceId = choiceId, PresentedChoiceIds = presentedChoiceIds.ToList(), IsCorrect = isCorrect, DurationSeconds = Math.Clamp(durationSeconds, 0, 86_400), CreatedUtc = DateTime.UtcNow }; try { await _reviews.InsertOneAsync(attempt); } catch (MongoWriteException exception) when (exception.WriteError.Category == ServerErrorCategory.DuplicateKey) { var prior = await _reviews.Find(value => value.UserId == userId && value.QuestionExternalId == questionExternalId && value.Cycle == cycle).FirstOrDefaultAsync(); if (prior is null) return null; attempt = prior; } var now = DateTime.UtcNow; var update = Builders.Update .Inc(value => value.Cycle, 1) .Inc(value => value.ReviewCount, 1) .Set(value => value.LastReviewedUtc, attempt.CreatedUtc) .Set(value => value.UpdatedUtc, now); if (attempt.IsCorrect) { update = update .Set(value => value.State, ReviewQueueStates.Mastered) .Set(value => value.MasteredUtc, attempt.CreatedUtc); } else { update = update.Inc(value => value.IncorrectReviewCount, 1); } await _items.UpdateOneAsync( value => value.UserId == userId && value.QuestionExternalId == questionExternalId && value.State == ReviewQueueStates.Pending && value.Cycle == cycle, update); return new ReviewQueueRecordResult { Attempt = attempt, Item = await FindItemAsync(userId, questionExternalId) }; } public Task> GetReviewsSinceAsync(string userId, DateTime sinceUtc) => _reviews.Find(value => value.UserId == userId && value.CreatedUtc >= sinceUtc).ToListAsync(); public static bool ShouldQueue(QuestionAttempt attempt) => !attempt.IsCorrect && !attempt.IsSkip && !string.IsNullOrWhiteSpace(attempt.UserId) && !string.IsNullOrWhiteSpace(attempt.QuestionExternalId); private async Task FindItemAsync(string userId, string questionExternalId) => await _items.Find(value => value.UserId == userId && value.QuestionExternalId == questionExternalId).FirstOrDefaultAsync(); }