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.Cycle, 1) .SetOnInsert(value => value.CreatedUtc, attempt.CreatedUtc) .Set(value => value.SourceAttemptId, attempt.Id ?? "") .Set(value => value.State, ReviewQueueStates.Pending) .Set(value => value.MasteredUtc, null) .Set(value => value.UpdatedUtc, now); var filter = Builders.Filter.Eq(value => value.UserId, attempt.UserId) & Builders.Filter.Eq(value => value.QuestionExternalId, attempt.QuestionExternalId) & (Builders.Filter.Ne(value => value.State, ReviewQueueStates.Mastered) | Builders.Filter.Ne(value => value.SourceAttemptId, attempt.Id ?? "")); try { await _items.UpdateOneAsync( filter, 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 scanThroughUtc = DateTime.UtcNow; var filter = Builders.Filter.Where(value => value.UserId == userId && !value.IsCorrect && !value.IsSkip); filter &= Builders.Filter.Lte(value => value.CreatedUtc, scanThroughUtc); var allMisses = await _firstAttempts.Find(filter).ToListAsync(); var existingItems = (await _items.Find(value => value.UserId == userId).ToListAsync()) .ToDictionary(value => value.QuestionExternalId, StringComparer.Ordinal); var misses = allMisses.GroupBy(value => value.QuestionExternalId, StringComparer.Ordinal) .Select(group => group.OrderByDescending(value => value.CreatedUtc).First()) .Where(attempt => !existingItems.TryGetValue(attempt.QuestionExternalId, out var item) || (item.State == ReviewQueueStates.Mastered && item.SourceAttemptId != (attempt.Id ?? "") && (!item.MasteredUtc.HasValue || attempt.CreatedUtc > item.MasteredUtc.Value))) .ToList(); if (misses.Count > 0) { var now = DateTime.UtcNow; var writes = misses.Select(attempt => new UpdateOneModel( Builders.Filter.Eq(value => value.UserId, userId) & Builders.Filter.Eq(value => value.QuestionExternalId, attempt.QuestionExternalId) & (Builders.Filter.Ne(value => value.State, ReviewQueueStates.Mastered) | Builders.Filter.Ne(value => value.SourceAttemptId, attempt.Id ?? "")), Builders.Update .SetOnInsert(value => value.UserId, userId) .SetOnInsert(value => value.QuestionExternalId, attempt.QuestionExternalId) .SetOnInsert(value => value.Cycle, 1) .SetOnInsert(value => value.CreatedUtc, attempt.CreatedUtc) .Set(value => value.SourceAttemptId, attempt.Id ?? "") .Set(value => value.State, ReviewQueueStates.Pending) .Set(value => value.MasteredUtc, null) .Set(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 CountPendingAsync(string userId, string? step3System) { if (string.IsNullOrWhiteSpace(step3System)) return await CountPendingAsync(userId); var requested = Step3SystemCatalog.Normalize(step3System) ?? throw new ArgumentException("step3System must be an official Step 3 system"); var pending = await _items.Find(value => value.UserId == userId && value.State == ReviewQueueStates.Pending).ToListAsync(); var count = 0L; foreach (var item in pending) { var question = await _questions.GetByExternalIdAsync(item.QuestionExternalId); if (question is not null && Step3SystemCatalog.Normalize(question) == requested) count++; } return count; } public Task> GetUserItemsAsync(string userId) => _items.Find(value => value.UserId == userId).ToListAsync(); public Task> GetUserReviewsAsync(string userId) => _reviews.Find(value => value.UserId == userId).ToListAsync(); public async Task GetRandomPendingAsync( string userId, IEnumerable excludedIds, string? step3System = null) { var excluded = excludedIds.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToList(); var requestedSystem = string.IsNullOrWhiteSpace(step3System) ? null : Step3SystemCatalog.Normalize(step3System); if (!string.IsNullOrWhiteSpace(step3System) && requestedSystem is null) throw new ArgumentException("step3System must be an official Step 3 system"); 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.Find(filter).ToListAsync(); foreach (var candidate in candidates.OrderBy(_ => Random.Shared.Next())) { var question = await _questions.GetByExternalIdAsync(candidate.QuestionExternalId); if (question is { IsPublished: true, FlaggedForReviewUtc: null }) { if (requestedSystem is null || Step3SystemCatalog.Normalize(question) == requestedSystem) return candidate; continue; } 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(); }