Files
step3/api/QuestionsApi/Services/ReviewQueueStore.cs
T

270 lines
14 KiB
C#

using MongoDB.Driver;
using QuestionsApi.Models;
namespace QuestionsApi.Services;
public sealed class ReviewQueueStore
{
private readonly IMongoCollection<ReviewQueueItem> _items;
private readonly IMongoCollection<QuestionReviewAttempt> _reviews;
private readonly IMongoCollection<QuestionAttempt> _firstAttempts;
private readonly IMongoCollection<ReviewQueueBackfillState> _backfills;
private readonly QuestionStore _questions;
public ReviewQueueStore(MongoDatabaseProvider mongo, QuestionStore questions)
{
_items = mongo.Database.GetCollection<ReviewQueueItem>("ReviewQueueItems");
_reviews = mongo.Database.GetCollection<QuestionReviewAttempt>("QuestionReviewAttempts");
_firstAttempts = mongo.Database.GetCollection<QuestionAttempt>("QuestionAttempts");
_backfills = mongo.Database.GetCollection<ReviewQueueBackfillState>("ReviewQueueBackfillStates");
_questions = questions;
_items.Indexes.CreateMany([
new CreateIndexModel<ReviewQueueItem>(
Builders<ReviewQueueItem>.IndexKeys.Ascending(value => value.UserId).Ascending(value => value.QuestionExternalId),
new CreateIndexOptions { Unique = true, Name = "review_queue_unique_user_question" }),
new CreateIndexModel<ReviewQueueItem>(
Builders<ReviewQueueItem>.IndexKeys.Ascending(value => value.UserId).Ascending(value => value.State),
new CreateIndexOptions { Name = "review_queue_user_state" })
]);
_reviews.Indexes.CreateMany([
new CreateIndexModel<QuestionReviewAttempt>(
Builders<QuestionReviewAttempt>.IndexKeys.Ascending(value => value.UserId).Ascending(value => value.QuestionExternalId).Ascending(value => value.Cycle),
new CreateIndexOptions { Unique = true, Name = "review_attempt_unique_cycle" }),
new CreateIndexModel<QuestionReviewAttempt>(
Builders<QuestionReviewAttempt>.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<ReviewQueueItem>.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<ReviewQueueItem>.Filter.Eq(value => value.UserId, attempt.UserId)
& Builders<ReviewQueueItem>.Filter.Eq(value => value.QuestionExternalId, attempt.QuestionExternalId)
& (Builders<ReviewQueueItem>.Filter.Ne(value => value.State, ReviewQueueStates.Mastered)
| Builders<ReviewQueueItem>.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<QuestionAttempt>.Filter.Where(value => value.UserId == userId && !value.IsCorrect && !value.IsSkip);
filter &= Builders<QuestionAttempt>.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<ReviewQueueItem>(
Builders<ReviewQueueItem>.Filter.Eq(value => value.UserId, userId)
& Builders<ReviewQueueItem>.Filter.Eq(value => value.QuestionExternalId, attempt.QuestionExternalId)
& (Builders<ReviewQueueItem>.Filter.Ne(value => value.State, ReviewQueueStates.Mastered)
| Builders<ReviewQueueItem>.Filter.Ne(value => value.SourceAttemptId, attempt.Id ?? "")),
Builders<ReviewQueueItem>.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<ReviewQueueItem> 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<ReviewQueueBackfillState>.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<long> CountPendingAsync(string userId) =>
_items.CountDocumentsAsync(value => value.UserId == userId && value.State == ReviewQueueStates.Pending);
public async Task<long> 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<List<ReviewQueueItem>> GetUserItemsAsync(string userId) =>
_items.Find(value => value.UserId == userId).ToListAsync();
public Task<List<QuestionReviewAttempt>> GetUserReviewsAsync(string userId) =>
_reviews.Find(value => value.UserId == userId).ToListAsync();
public async Task<ReviewQueueItem?> GetRandomPendingAsync(
string userId, IEnumerable<string> 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<ReviewQueueItem>.Filter.Eq(value => value.UserId, userId)
& Builders<ReviewQueueItem>.Filter.Eq(value => value.State, ReviewQueueStates.Pending);
if (excluded.Count > 0) filter &= Builders<ReviewQueueItem>.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<ReviewQueueItem>.Update.Set(value => value.State, ReviewQueueStates.Unavailable)
.Set(value => value.UpdatedUtc, DateTime.UtcNow));
}
return null;
}
public async Task<ReviewQueueRecordResult?> RecordReviewAsync(
string userId,
string clientAttemptId,
string questionExternalId,
int cycle,
string choiceId,
IReadOnlyCollection<string> 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<ReviewQueueItem>.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<List<QuestionReviewAttempt>> 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<ReviewQueueItem?> FindItemAsync(string userId, string questionExternalId) =>
await _items.Find(value => value.UserId == userId && value.QuestionExternalId == questionExternalId).FirstOrDefaultAsync();
}