Implement persistent review queue
This commit is contained in:
@@ -0,0 +1,231 @@
|
||||
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.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<QuestionAttempt>.Filter.Where(value => value.UserId == userId && !value.IsCorrect && !value.IsSkip);
|
||||
if (checkpoint is not null)
|
||||
filter &= Builders<QuestionAttempt>.Filter.Gte(value => value.CreatedUtc, checkpoint.CompletedUtc);
|
||||
filter &= Builders<QuestionAttempt>.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<ReviewQueueItem>(
|
||||
Builders<ReviewQueueItem>.Filter.Where(value => value.UserId == userId && value.QuestionExternalId == attempt.QuestionExternalId),
|
||||
Builders<ReviewQueueItem>.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<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<ReviewQueueItem?> GetRandomPendingAsync(string userId, IEnumerable<string> excludedIds)
|
||||
{
|
||||
var excluded = excludedIds.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToList();
|
||||
for (var pass = 0; pass < 5; pass++)
|
||||
{
|
||||
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.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 }) return candidate;
|
||||
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();
|
||||
}
|
||||
Reference in New Issue
Block a user