Fix database connection pooling and UI question filtering
Some checks failed
Admin portal tests / test (push) Has been cancelled

- Replaced single Postgres Connection with Pool across the server to prevent runTx locking
- Added _filterForEnabledSources to client's _fetchUnansweredForPanel to strictly enforce local configuration

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
nathanAndrinoid 2026-06-05 22:04:06 +00:00
parent be57c17c15
commit a7c370b743
47 changed files with 431 additions and 181 deletions

View File

@ -142,6 +142,14 @@ class QuestionSourceConfigRepository {
return false; return false;
} }
/// True when spelling is enabled and its effective mix is 100%.
static bool isSpellingFullyWeighted(List<QuestionSourcePreference> sources) {
if (!isSpellingEnabled(sources)) {
return false;
}
return (effectiveMix(sources)['spelling'] ?? 0) >= 100;
}
/// Effective share (0–100) for each enabled source; empty when none enabled. /// Effective share (0–100) for each enabled source; empty when none enabled.
static Map<String, int> effectiveMix( static Map<String, int> effectiveMix(
List<QuestionSourcePreference> sources, List<QuestionSourcePreference> sources,

View File

@ -170,9 +170,13 @@ class HomeScreen extends StatelessWidget {
); );
final bool spellingEnabled = final bool spellingEnabled =
QuestionSourceConfigRepository.isSpellingEnabled(sources); QuestionSourceConfigRepository.isSpellingEnabled(sources);
final bool spellingFullyWeighted =
QuestionSourceConfigRepository.isSpellingFullyWeighted(
sources,
);
final bool showSpellingPlaceholder = panelOpen && final bool showSpellingPlaceholder = panelOpen &&
spellingEnabled && spellingEnabled &&
!marketEnabled && (spellingFullyWeighted || !marketEnabled) &&
(question == null || !question.isSpelling); (question == null || !question.isSpelling);
final bool showMarketPlaceholder = panelOpen && final bool showMarketPlaceholder = panelOpen &&
marketEnabled && marketEnabled &&

View File

@ -105,9 +105,6 @@ class _LandingScreenState extends State<LandingScreen> {
case 'cancelled-popup-request': case 'cancelled-popup-request':
case 'auth/popup-closed-by-user': case 'auth/popup-closed-by-user':
return 'Sign-in was cancelled. Tap below to try again.'; return 'Sign-in was cancelled. Tap below to try again.';
case 'auth/unauthorized-domain':
return 'This site is not authorized for Firebase Auth. Add your '
'domain in Firebase Console → Authentication → Authorized domains.';
} }
} }

View File

@ -156,9 +156,7 @@ class QuestionsHubService {
try { try {
// Align client filters with server-side source preferences before fetch. // Align client filters with server-side source preferences before fetch.
await QuestionSourceConfigRepository.instance.loadFromRemote(); await QuestionSourceConfigRepository.instance.loadFromRemote();
final List<IncomingQuestion> fetched = await _api.fetchUnanswered(); final List<IncomingQuestion> active = await _fetchUnansweredForPanel();
// GET /questions already returns only server-enabled sources.
final List<IncomingQuestion> active = fetched;
questionPanelOpen.value = true; questionPanelOpen.value = true;
@ -198,6 +196,10 @@ class QuestionsHubService {
if (QuestionSourceConfigRepository.isMarketOnlyEnabled(sources)) { if (QuestionSourceConfigRepository.isMarketOnlyEnabled(sources)) {
return question.isMarketHistory; return question.isMarketHistory;
} }
if (QuestionSourceConfigRepository.isSpellingFullyWeighted(sources) &&
question.isMarketHistory) {
return false;
}
if (!QuestionSourceConfigRepository.isMarketHistoryEnabled(sources)) { if (!QuestionSourceConfigRepository.isMarketHistoryEnabled(sources)) {
return question.sourceId != 'market_history'; return question.sourceId != 'market_history';
} }
@ -213,6 +215,54 @@ class QuestionsHubService {
return questions.where(_acceptsQuestionSource).toList(growable: false); return questions.where(_acceptsQuestionSource).toList(growable: false);
} }
/// Polls with a spinner-covered retry loop until the server has the right
/// question type ready.
///
/// When spelling is configured at any percentage the server may still be
/// running catalog seeding or pipeline setup. We keep retrying (the spinner
/// overlay from [questionActionBusy] blocks the UI) until:
/// • a spelling question appears in the result, or
/// • spelling is not 100% and we have any non-empty result (market is valid
/// at that point in the mix), or
/// • [_maxPanelFetchAttempts] are exhausted.
static const int _maxPanelFetchAttempts = 10;
static const Duration _panelFetchPause = Duration(milliseconds: 300);
Future<List<IncomingQuestion>> _fetchUnansweredForPanel() async {
final List<QuestionSourcePreference> sources =
QuestionSourceConfigRepository.instance.current;
final Map<String, int> mix =
QuestionSourceConfigRepository.effectiveMix(sources);
final int spellingShare = mix['spelling'] ?? 0;
// No spelling configured — single fetch is sufficient.
if (spellingShare == 0) {
return _api.fetchUnanswered();
}
for (var attempt = 0; attempt < _maxPanelFetchAttempts; attempt++) {
final List<IncomingQuestion> fetched = await _api.fetchUnanswered();
final List<IncomingQuestion> filtered = _filterForEnabledSources(fetched);
// A spelling question is ready — accept immediately.
if (filtered.any((IncomingQuestion q) => q.isSpelling)) {
return filtered;
}
// Spelling is not the only source and we have something — valid mix result.
if (filtered.isNotEmpty && spellingShare < 100) {
return filtered;
}
// Queue is still empty or only has wrong-source questions — wait and retry.
if (attempt < _maxPanelFetchAttempts - 1) {
await Future.delayed(_panelFetchPause);
}
}
final List<IncomingQuestion> finalFetch = await _api.fetchUnanswered();
return _filterForEnabledSources(finalFetch);
}
void closeQuestionPanel() { void closeQuestionPanel() {
questionPanelOpen.value = false; questionPanelOpen.value = false;
} }
@ -253,7 +303,7 @@ class QuestionsHubService {
return; return;
} }
final List<IncomingQuestion> refreshed = await _api.fetchUnanswered(); final List<IncomingQuestion> refreshed = await _fetchUnansweredForPanel();
if (refreshed.isEmpty) { if (refreshed.isEmpty) {
if (result.nextQuestion != null && if (result.nextQuestion != null &&
_acceptsQuestionSource(result.nextQuestion!)) { _acceptsQuestionSource(result.nextQuestion!)) {
@ -279,7 +329,8 @@ class QuestionsHubService {
orElse: () => refreshed.first, orElse: () => refreshed.first,
); );
} else if (result.nextQuestion != null && } else if (result.nextQuestion != null &&
result.nextQuestion!.id != skippedQuestionId) { result.nextQuestion!.id != skippedQuestionId &&
_acceptsQuestionSource(result.nextQuestion!)) {
next = result.nextQuestion!; next = result.nextQuestion!;
} }
@ -370,7 +421,7 @@ class QuestionsHubService {
return; return;
} }
final List<IncomingQuestion> refreshed = await _api.fetchUnanswered(); final List<IncomingQuestion> refreshed = await _fetchUnansweredForPanel();
final List<IncomingQuestion> remaining = refreshed final List<IncomingQuestion> remaining = refreshed
.where((IncomingQuestion q) => q.id != answeredQuestionId) .where((IncomingQuestion q) => q.id != answeredQuestionId)
.toList(); .toList();
@ -428,7 +479,7 @@ class QuestionsHubService {
return false; return false;
} }
final List<IncomingQuestion> refreshed = await _api.fetchUnanswered(); final List<IncomingQuestion> refreshed = await _fetchUnansweredForPanel();
final List<IncomingQuestion> active = final List<IncomingQuestion> active =
refreshed.where(_acceptsQuestionSource).toList(growable: false); refreshed.where(_acceptsQuestionSource).toList(growable: false);
if (active.isNotEmpty) { if (active.isNotEmpty) {

View File

@ -155,10 +155,6 @@ Future<void> main() async {
final TradableAssetsDb tradableAssetsDb = final TradableAssetsDb tradableAssetsDb =
TradableAssetsDb(db.connection); TradableAssetsDb(db.connection);
final AlpacaAssetsClient assetsClient = AlpacaAssetsClient(env: env.alpaca); final AlpacaAssetsClient assetsClient = AlpacaAssetsClient(env: env.alpaca);
final MarketHistoryApiRateLimiter historyRateLimiter =
MarketHistoryApiRateLimiter(
requestsPerMinute: mh.apiRequestsPerMinute,
);
final AlpacaMarketDataClient historyMarketDataClient = final AlpacaMarketDataClient historyMarketDataClient =
AlpacaMarketDataClient(env: env.alpaca); AlpacaMarketDataClient(env: env.alpaca);
final TradableAssetsSync tradableAssetsSync = TradableAssetsSync( final TradableAssetsSync tradableAssetsSync = TradableAssetsSync(

View File

@ -3,14 +3,14 @@ import 'dart:io';
import 'package:postgres/postgres.dart'; import 'package:postgres/postgres.dart';
class ProfileDb { class ProfileDb {
ProfileDb(this._connection); ProfileDb(this._pool);
final Connection _connection; final Pool _pool;
static Future<ProfileDb> connect(String databaseUrl) async { static Future<ProfileDb> connect(String databaseUrl) async {
final Uri uri = Uri.parse(databaseUrl); final Uri uri = Uri.parse(databaseUrl);
final Connection connection = await Connection.open( final Pool pool = Pool.withEndpoints(
Endpoint( [Endpoint(
host: uri.host.isEmpty ? 'localhost' : uri.host, host: uri.host.isEmpty ? 'localhost' : uri.host,
port: uri.hasPort ? uri.port : 5432, port: uri.hasPort ? uri.port : 5432,
database: uri.pathSegments.isNotEmpty ? uri.pathSegments.last : 'postgres', database: uri.pathSegments.isNotEmpty ? uri.pathSegments.last : 'postgres',
@ -18,13 +18,13 @@ class ProfileDb {
password: uri.userInfo.contains(':') password: uri.userInfo.contains(':')
? uri.userInfo.split(':').skip(1).join(':') ? uri.userInfo.split(':').skip(1).join(':')
: null, : null,
), )],
settings: const ConnectionSettings(sslMode: SslMode.disable), settings: const PoolSettings(sslMode: SslMode.disable),
); );
return ProfileDb(connection); return ProfileDb(pool);
} }
Connection get connection => _connection; Pool get connection => _pool;
Future<void> migrate() async { Future<void> migrate() async {
final List<File> files = Directory('migrations') final List<File> files = Directory('migrations')
@ -43,13 +43,13 @@ class ProfileDb {
.toList(); .toList();
for (final String statement in statements) { for (final String statement in statements) {
await _connection.execute(statement); await _pool.execute(statement);
} }
} }
} }
Future<Map<String, dynamic>?> getProfile(String firebaseUid) async { Future<Map<String, dynamic>?> getProfile(String firebaseUid) async {
final Result result = await _connection.execute( final Result result = await _pool.execute(
Sql.named( Sql.named(
'SELECT firebase_uid, email, display_name, photo_url, locale, timezone, ' 'SELECT firebase_uid, email, display_name, photo_url, locale, timezone, '
'onboarding_done, revision, updated_at ' 'onboarding_done, revision, updated_at '
@ -69,7 +69,7 @@ class ProfileDb {
required Map<String, dynamic> body, required Map<String, dynamic> body,
required int clientRevision, required int clientRevision,
}) async { }) async {
final Result existing = await _connection.execute( final Result existing = await _pool.execute(
Sql.named('SELECT revision FROM users WHERE firebase_uid = @uid'), Sql.named('SELECT revision FROM users WHERE firebase_uid = @uid'),
parameters: <String, dynamic>{'uid': firebaseUid}, parameters: <String, dynamic>{'uid': firebaseUid},
); );
@ -82,7 +82,7 @@ class ProfileDb {
} }
} }
await _connection.execute( await _pool.execute(
Sql.named( Sql.named(
''' '''
INSERT INTO users ( INSERT INTO users (
@ -119,7 +119,7 @@ class ProfileDb {
return (await getProfile(firebaseUid))!; return (await getProfile(firebaseUid))!;
} }
Future<void> close() => _connection.close(); Future<void> close() => _pool.close();
Map<String, dynamic> _rowToJson(ResultRow row) { Map<String, dynamic> _rowToJson(ResultRow row) {
final DateTime updatedAt = row[8]! as DateTime; final DateTime updatedAt = row[8]! as DateTime;

View File

@ -41,7 +41,7 @@ class MarketHistoryAdminPortalConfig {
Handler marketHistoryAdminHandler({ Handler marketHistoryAdminHandler({
required FirebaseAuthVerifier auth, required FirebaseAuthVerifier auth,
required Connection connection, required Pool connection,
required Set<String> adminFirebaseUids, required Set<String> adminFirebaseUids,
MarketHistoryAdminActions? actions, MarketHistoryAdminActions? actions,
MarketHistoryAdminPortalConfig? portalConfig, MarketHistoryAdminPortalConfig? portalConfig,

View File

@ -32,7 +32,7 @@ Handler questionsHandler({
final Map<String, dynamic>? question = final Map<String, dynamic>? question =
await questionDispatcher.dispatchNextQuestion(firebaseUid); await questionDispatcher.dispatchNextQuestion(firebaseUid);
final int unansweredCount = final int unansweredCount =
await questionsDb.countUnansweredQuestions(firebaseUid); await questionDispatcher.countEnabledUnansweredQuestions(firebaseUid);
final Map<String, dynamic> score = final Map<String, dynamic> score =
await questionsDb.getGuessScoreSummary(firebaseUid); await questionsDb.getGuessScoreSummary(firebaseUid);
return _jsonResponse(200, <String, dynamic>{ return _jsonResponse(200, <String, dynamic>{
@ -92,7 +92,7 @@ Handler questionsHandler({
final Map<String, dynamic>? question = final Map<String, dynamic>? question =
await questionDispatcher.dispatchNextQuestion(firebaseUid); await questionDispatcher.dispatchNextQuestion(firebaseUid);
final int unansweredCount = final int unansweredCount =
await questionsDb.countUnansweredQuestions(firebaseUid); await questionDispatcher.countEnabledUnansweredQuestions(firebaseUid);
return _jsonResponse(200, <String, dynamic>{ return _jsonResponse(200, <String, dynamic>{
'score': score, 'score': score,
'unansweredCount': unansweredCount, 'unansweredCount': unansweredCount,
@ -136,7 +136,7 @@ Handler questionsHandler({
); );
} }
var unansweredCount = var unansweredCount =
await questionsDb.countUnansweredQuestions(firebaseUid); await questionDispatcher.countEnabledUnansweredQuestions(firebaseUid);
final Map<String, dynamic>? nextQuestion = final Map<String, dynamic>? nextQuestion =
await questionDispatcher.nextQuestionAfterAnswer( await questionDispatcher.nextQuestionAfterAnswer(
firebaseUid: firebaseUid, firebaseUid: firebaseUid,
@ -226,18 +226,12 @@ Handler questionsHandler({
return _jsonResponse(404, <String, dynamic>{'error': 'Not found'}); return _jsonResponse(404, <String, dynamic>{'error': 'Not found'});
} }
final int unansweredCount = final int unansweredCount =
await questionsDb.countUnansweredQuestions(firebaseUid); await questionDispatcher.countEnabledUnansweredQuestions(firebaseUid);
final Map<String, dynamic>? nextRow = final Map<String, dynamic>? nextQuestion =
await questionsDb.findNextUnansweredAfterDefer( await questionDispatcher.nextQuestionAfterDefer(
assignedUserId: firebaseUid, firebaseUid: firebaseUid,
deferredQuestionId: id, deferredQuestionId: id,
); );
final Map<String, dynamic>? nextQuestion = nextRow == null
? null
: questionsDb.toClientPayload(
nextRow,
unansweredCount: unansweredCount,
);
return _jsonResponse(200, <String, dynamic>{ return _jsonResponse(200, <String, dynamic>{
'question': updated, 'question': updated,
'unansweredCount': unansweredCount, 'unansweredCount': unansweredCount,

View File

@ -150,7 +150,7 @@ class QuestionPipeline {
} }
} }
Future<void> _deliverPipelineQuestion({ Future<Map<String, dynamic>> _deliverPipelineQuestion({
required String firebaseUid, required String firebaseUid,
required String questionText, required String questionText,
required num correctAnswer, required num correctAnswer,
@ -159,7 +159,7 @@ class QuestionPipeline {
required String pipelineStep, required String pipelineStep,
Map<String, dynamic>? metadata, Map<String, dynamic>? metadata,
}) async { }) async {
await _questionService.createAndDeliverQuestion( return _questionService.createAndDeliverQuestion(
assignedUserId: firebaseUid, assignedUserId: firebaseUid,
questionText: questionText:
testMode ? PipelineTestQuestions.text : questionText, testMode ? PipelineTestQuestions.text : questionText,
@ -452,6 +452,7 @@ class QuestionPipeline {
); );
final int unansweredCount = final int unansweredCount =
await _questionDispatcher?.countEnabledUnansweredQuestions(firebaseUid) ??
await _questionsDb.countUnansweredQuestions(firebaseUid); await _questionsDb.countUnansweredQuestions(firebaseUid);
return _questionsDb.toClientPayload( return _questionsDb.toClientPayload(
question, question,
@ -472,6 +473,7 @@ class QuestionPipeline {
final Map<String, dynamic>? question = final Map<String, dynamic>? question =
await _questionDispatcher?.dispatchNextQuestion(firebaseUid); await _questionDispatcher?.dispatchNextQuestion(firebaseUid);
final int unansweredCount = final int unansweredCount =
await _questionDispatcher?.countEnabledUnansweredQuestions(firebaseUid) ??
await _questionsDb.countUnansweredQuestions(firebaseUid); await _questionsDb.countUnansweredQuestions(firebaseUid);
return <String, dynamic>{ return <String, dynamic>{
'unansweredCount': unansweredCount, 'unansweredCount': unansweredCount,
@ -507,27 +509,6 @@ class QuestionPipeline {
); );
} }
Future<void> _startRootPipeline(String firebaseUid) async {
if (!await _canEnqueue(firebaseUid)) {
return;
}
await _questionsDb.upsertPipelineState(
assignedUserId: firebaseUid,
pipelineKey: PipelineKeys.root,
step: PipelineSteps.chooseTrack,
context: <String, dynamic>{},
);
await _deliverPipelineQuestion(
firebaseUid: firebaseUid,
questionText:
'Which topics interest you? Swipe toward +10 for weather questions, '
'or toward -10 for geography. Submit your preference.',
correctAnswer: 0,
sourceTag: 'pipeline:root:choose_track',
pipelineKey: PipelineKeys.root,
pipelineStep: PipelineSteps.chooseTrack,
);
}
Future<void> _handleRootAnswer({ Future<void> _handleRootAnswer({
required String firebaseUid, required String firebaseUid,
@ -743,7 +724,12 @@ class QuestionPipeline {
} }
} }
/// Reserves a spelling quiz slot; word selection runs at tile load via [prepareSpellingQuestion]. /// Creates a fully-prepared spelling word quiz, selecting the word immediately.
///
/// Word selection is done here rather than deferred to tile load so the
/// question is immediately dispatchable and the client never needs a separate
/// prepare call. Falls back to a [spelling_pending] placeholder only when no
/// catalog words are available.
Future<void> _startSpellingPipeline(String firebaseUid) async { Future<void> _startSpellingPipeline(String firebaseUid) async {
if (!await _canEnqueue(firebaseUid)) { if (!await _canEnqueue(firebaseUid)) {
return; return;
@ -752,22 +738,57 @@ class QuestionPipeline {
return; return;
} }
await _spellingItemsDb.ensureCatalogSeeded();
final SpellingItemRecord? item =
await _spellingSelector.pickForUser(firebaseUid);
if (item == null) {
// No words available yet — create a pending placeholder as a fallback.
await _questionsDb.upsertPipelineState(
assignedUserId: firebaseUid,
pipelineKey: PipelineKeys.spelling,
step: PipelineSteps.spellingWordQuiz,
context: <String, dynamic>{},
);
await _deliverPipelineQuestion(
firebaseUid: firebaseUid,
questionText: 'Spelling question loading…',
correctAnswer: 0,
sourceTag: 'spelling:word_quiz',
pipelineKey: PipelineKeys.spelling,
pipelineStep: PipelineSteps.spellingWordQuiz,
metadata: <String, dynamic>{'spelling_pending': true},
);
return;
}
// Word selected — store context and create a fully-prepared question.
await _questionsDb.upsertPipelineState( await _questionsDb.upsertPipelineState(
assignedUserId: firebaseUid, assignedUserId: firebaseUid,
pipelineKey: PipelineKeys.spelling, pipelineKey: PipelineKeys.spelling,
step: PipelineSteps.spellingWordQuiz, step: PipelineSteps.spellingWordQuiz,
context: <String, dynamic>{}, context: item.toPipelineContext(),
); );
await _deliverPipelineQuestion(
await _spellingProgressDb.ensureLink(
firebaseUid: firebaseUid, firebaseUid: firebaseUid,
questionText: testMode spellingItemId: item.id,
? PipelineTestQuestions.text );
: 'Spelling question loading…',
correctAnswer: 0, final Map<String, dynamic> payload = await _deliverPipelineQuestion(
firebaseUid: firebaseUid,
questionText: 'Spell the word you hear.',
correctAnswer: item.correctAnswer,
sourceTag: 'spelling:word_quiz', sourceTag: 'spelling:word_quiz',
pipelineKey: PipelineKeys.spelling, pipelineKey: PipelineKeys.spelling,
pipelineStep: PipelineSteps.spellingWordQuiz, pipelineStep: PipelineSteps.spellingWordQuiz,
metadata: <String, dynamic>{'spelling_pending': true}, metadata: item.toQuestionMetadata(),
);
await _spellingQuestionLinksDb.linkQuestion(
questionId: payload['id'] as String,
firebaseUid: firebaseUid,
spellingItemId: item.id,
); );
} }

View File

@ -24,19 +24,18 @@ class QuestionDispatcher {
final QuestionSourcePreferencesDb _preferencesDb; final QuestionSourcePreferencesDb _preferencesDb;
final int maxQueuedQuestions; final int maxQueuedQuestions;
/// Maximum maintenance passes before giving up on an empty queue.
static const int _maxQueueAttempts = 6;
/// Ensures at least one enabled-source question exists; returns mix-ordered head. /// Ensures at least one enabled-source question exists; returns mix-ordered head.
Future<Map<String, dynamic>?> dispatchNextQuestion(String firebaseUid) async { Future<Map<String, dynamic>?> dispatchNextQuestion(String firebaseUid) async {
await _questionsDb.ensureUserExists(firebaseUid); await _questionsDb.ensureUserExists(firebaseUid);
var rows = await _listEnabledUnansweredOrdered(firebaseUid); var rows = await _listEnabledUnansweredOrdered(firebaseUid);
if (rows.isEmpty) { for (var attempt = 0;
if (!await _enqueueOneFromMix(firebaseUid)) { rows.isEmpty && attempt < _maxQueueAttempts;
await _ensureSpellingQuestionWhenNeeded(firebaseUid); attempt++) {
} await _runQueueMaintenance(firebaseUid);
rows = await _listEnabledUnansweredOrdered(firebaseUid);
}
if (rows.isEmpty) {
await _ensureSpellingQuestionWhenNeeded(firebaseUid);
rows = await _listEnabledUnansweredOrdered(firebaseUid); rows = await _listEnabledUnansweredOrdered(firebaseUid);
} }
@ -46,41 +45,40 @@ class QuestionDispatcher {
return _toClientPayload(rows.first, rows.length); return _toClientPayload(rows.first, rows.length);
} }
/// Ensures an enabled-source question exists before the question panel opens.
Future<void> preparePanelQueue(String firebaseUid) async {
await _questionsDb.ensureUserExists(firebaseUid);
var rows = await _listEnabledUnansweredOrdered(firebaseUid);
if (rows.isEmpty) {
await _enqueueOneFromMix(firebaseUid);
rows = await _listEnabledUnansweredOrdered(firebaseUid);
}
if (rows.isEmpty) {
await _ensureSpellingQuestionWhenNeeded(firebaseUid);
}
}
/// Unanswered questions for enabled sources only, ordered by configured mix. /// Unanswered questions for enabled sources only, ordered by configured mix.
///
/// Runs [_runQueueMaintenance] in a loop until at least one question is ready
/// or [_maxQueueAttempts] are exhausted. This ensures spelling enqueue,
/// catalog seeding, and pipeline-state setup all complete before returning.
Future<List<Map<String, dynamic>>> listPanelQuestions( Future<List<Map<String, dynamic>>> listPanelQuestions(
String firebaseUid, String firebaseUid,
) async { ) async {
await preparePanelQueue(firebaseUid); await _questionsDb.ensureUserExists(firebaseUid);
var rows = await _listEnabledUnansweredOrdered(firebaseUid); var rows = await _listEnabledUnansweredOrdered(firebaseUid);
if (rows.isEmpty) { for (var attempt = 0;
await _enqueueOneFromMix(firebaseUid); rows.isEmpty && attempt < _maxQueueAttempts;
rows = await _listEnabledUnansweredOrdered(firebaseUid); attempt++) {
} await _runQueueMaintenance(firebaseUid);
if (rows.isEmpty) {
await _ensureSpellingQuestionWhenNeeded(firebaseUid);
rows = await _listEnabledUnansweredOrdered(firebaseUid); rows = await _listEnabledUnansweredOrdered(firebaseUid);
} }
return rows return rows
.map( .map((Map<String, dynamic> row) => _toClientPayload(row, rows.length))
(Map<String, dynamic> row) =>
_toClientPayload(row, rows.length),
)
.toList(growable: false); .toList(growable: false);
} }
/// Counts the number of unanswered questions that belong to enabled sources.
Future<int> countEnabledUnansweredQuestions(String firebaseUid) async {
final List<ResolvedQuestionSourcePreference> prefs =
await _preferencesDb.getResolved(firebaseUid);
final Map<String, int> mix = QuestionSourceMix.effectiveMixFromResolved(prefs);
final Set<String> allowed = _enabledSourceIds(prefs)
.where((String id) => (mix[id] ?? 0) > 0)
.toSet();
return _questionsDb.countUnansweredForAllowedSources(firebaseUid, allowed);
}
/// Next unanswered enabled-source question after an answer. /// Next unanswered enabled-source question after an answer.
Future<Map<String, dynamic>?> nextQuestionAfterAnswer({ Future<Map<String, dynamic>?> nextQuestionAfterAnswer({
required String firebaseUid, required String firebaseUid,
@ -96,7 +94,9 @@ class QuestionDispatcher {
final Map<String, int> mix = QuestionSourceMix.effectiveMixFromResolved(prefs); final Map<String, int> mix = QuestionSourceMix.effectiveMixFromResolved(prefs);
final Map<String, int> delivered = final Map<String, int> delivered =
await _questionsDb.countDeliveredBySourceId(firebaseUid); await _questionsDb.countDeliveredBySourceId(firebaseUid);
final Set<String> allowed = _enabledSourceIds(prefs); final Set<String> allowed = _enabledSourceIds(prefs)
.where((String id) => (mix[id] ?? 0) > 0)
.toSet();
final Map<String, dynamic>? row = await _questionsDb.findHeadUnansweredForMix( final Map<String, dynamic>? row = await _questionsDb.findHeadUnansweredForMix(
firebaseUid, firebaseUid,
effectiveMixPercent: mix, effectiveMixPercent: mix,
@ -111,6 +111,24 @@ class QuestionDispatcher {
return _toClientPayload(row, rows.length); return _toClientPayload(row, rows.length);
} }
/// Next unanswered enabled-source question after deferring a question.
Future<Map<String, dynamic>?> nextQuestionAfterDefer({
required String firebaseUid,
required String deferredQuestionId,
}) async {
final List<Map<String, dynamic>> rows =
await _listEnabledUnansweredOrdered(firebaseUid);
if (rows.isEmpty) {
return dispatchNextQuestion(firebaseUid);
}
for (final Map<String, dynamic> row in rows) {
if (row['id'] != deferredQuestionId) {
return _toClientPayload(row, rows.length);
}
}
return _toClientPayload(rows.first, rows.length);
}
/// Fills the queue up to [maxQueuedQuestions] using mix-sequential source picks. /// Fills the queue up to [maxQueuedQuestions] using mix-sequential source picks.
Future<void> topUpQueue(String firebaseUid) async { Future<void> topUpQueue(String firebaseUid) async {
await _questionsDb.ensureUserExists(firebaseUid); await _questionsDb.ensureUserExists(firebaseUid);
@ -127,6 +145,16 @@ class QuestionDispatcher {
} }
} }
/// Runs both mix-based enqueue and spelling-specific maintenance in one pass.
///
/// Both helpers are idempotent (spelling checks for an existing unanswered
/// word-quiz before creating a new one), so calling this in a loop is safe.
Future<void> _runQueueMaintenance(String firebaseUid) async {
if (!await _enqueueOneFromMix(firebaseUid)) {
await _ensureSpellingQuestionWhenNeeded(firebaseUid);
}
}
Future<bool> _enqueueOneFromMix(String firebaseUid) async { Future<bool> _enqueueOneFromMix(String firebaseUid) async {
final List<ResolvedQuestionSourcePreference> prefs = final List<ResolvedQuestionSourcePreference> prefs =
await _preferencesDb.getResolved(firebaseUid); await _preferencesDb.getResolved(firebaseUid);
@ -144,7 +172,7 @@ class QuestionDispatcher {
final List<String> order = _sourceTryOrderForEnqueue( final List<String> order = _sourceTryOrderForEnqueue(
prefs: prefs, prefs: prefs,
primarySourceId: primary, primarySourceId: primary,
); ).where((String id) => (mix[id] ?? 0) > 0).toList(growable: false);
final int? marketBatchCap = _marketBatchCapForMix(prefs); final int? marketBatchCap = _marketBatchCapForMix(prefs);
@ -167,7 +195,11 @@ class QuestionDispatcher {
final List<ResolvedQuestionSourcePreference> prefs = final List<ResolvedQuestionSourcePreference> prefs =
await _preferencesDb.getResolved(firebaseUid); await _preferencesDb.getResolved(firebaseUid);
final Map<String, int> mix = QuestionSourceMix.effectiveMixFromResolved(prefs); final Map<String, int> mix = QuestionSourceMix.effectiveMixFromResolved(prefs);
final Set<String> allowed = _enabledSourceIds(prefs); // Only include sources that are both enabled and have a non-zero mix share.
// A source with enabled=true but ratio=0 should not serve questions.
final Set<String> allowed = _enabledSourceIds(prefs)
.where((String id) => (mix[id] ?? 0) > 0)
.toSet();
if (allowed.isEmpty) { if (allowed.isEmpty) {
return <Map<String, dynamic>>[]; return <Map<String, dynamic>>[];
} }

View File

@ -200,7 +200,6 @@ class QuestionService {
return payload; return payload;
} }
/// On SignalR connect: deliver an existing unanswered question only (no create).
Future<void> deliverPendingQuestionOnConnect( Future<void> deliverPendingQuestionOnConnect(
QuestionsHubConnection connection, QuestionsHubConnection connection,
) async { ) async {

View File

@ -7,7 +7,7 @@ import 'question_source_preferences.dart';
class QuestionSourcePreferencesDb { class QuestionSourcePreferencesDb {
QuestionSourcePreferencesDb(this._connection); QuestionSourcePreferencesDb(this._connection);
final Connection _connection; final Pool _connection;
Future<List<ResolvedQuestionSourcePreference>> getResolved( Future<List<ResolvedQuestionSourcePreference>> getResolved(
String firebaseUid, String firebaseUid,

View File

@ -15,9 +15,9 @@ import 'trading/user_trading_state_db.dart';
class QuestionsDb { class QuestionsDb {
QuestionsDb(this._connection); QuestionsDb(this._connection);
final Connection _connection; final Pool _connection;
Connection get connection => _connection; Pool get connection => _connection;
static const Uuid _uuid = Uuid(); static const Uuid _uuid = Uuid();

View File

@ -7,7 +7,7 @@ import 'spelling_item_record.dart';
class SpellingItemsDb { class SpellingItemsDb {
SpellingItemsDb(this._connection); SpellingItemsDb(this._connection);
final Connection _connection; final Pool _connection;
/// Inserts curriculum from [SpellingWordList] when the catalog is empty. /// Inserts curriculum from [SpellingWordList] when the catalog is empty.
Future<void> ensureCatalogSeeded() async { Future<void> ensureCatalogSeeded() async {

View File

@ -4,7 +4,7 @@ import 'package:postgres/postgres.dart';
class SpellingQuestionLinksDb { class SpellingQuestionLinksDb {
SpellingQuestionLinksDb(this._connection); SpellingQuestionLinksDb(this._connection);
final Connection _connection; final Pool _connection;
/// Creates a link when the spelling tile loads; idempotent per [questionId]. /// Creates a link when the spelling tile loads; idempotent per [questionId].
Future<void> linkQuestion({ Future<void> linkQuestion({

View File

@ -6,7 +6,7 @@ import 'spelling_progress_policy.dart';
class UserSpellingProgressDb { class UserSpellingProgressDb {
UserSpellingProgressDb(this._connection); UserSpellingProgressDb(this._connection);
final Connection _connection; final Pool _connection;
/// Ensures a per-user link to [spellingItemId] with scores starting at zero. /// Ensures a per-user link to [spellingItemId] with scores starting at zero.
Future<void> ensureLink({ Future<void> ensureLink({

View File

@ -11,7 +11,7 @@ abstract final class GuessScoreStore {
/// Counters and total are reconciled from [market_history_prospective_answers] /// Counters and total are reconciled from [market_history_prospective_answers]
/// when stored JSON is missing or behind answer history (cross-device safe). /// when stored JSON is missing or behind answer history (cross-device safe).
static Future<Map<String, dynamic>> loadSummary( static Future<Map<String, dynamic>> loadSummary(
Connection connection, Pool connection,
String firebaseUid, String firebaseUid,
) async { ) async {
final UserTradingStateDb tradingStateDb = UserTradingStateDb(connection); final UserTradingStateDb tradingStateDb = UserTradingStateDb(connection);
@ -94,7 +94,7 @@ abstract final class GuessScoreStore {
} }
static Future<_AnswerStats> _fetchAnswerStats( static Future<_AnswerStats> _fetchAnswerStats(
Connection connection, Pool connection,
String firebaseUid, String firebaseUid,
) async { ) async {
final Result result = await connection.execute( final Result result = await connection.execute(
@ -127,7 +127,7 @@ abstract final class GuessScoreStore {
} }
static Future<num> _recomputeTotalScore( static Future<num> _recomputeTotalScore(
Connection connection, Pool connection,
String firebaseUid, String firebaseUid,
) async { ) async {
final Result rows = await connection.execute( final Result rows = await connection.execute(
@ -153,7 +153,7 @@ abstract final class GuessScoreStore {
} }
static Future<String?> _latestSlotWire( static Future<String?> _latestSlotWire(
Connection connection, Pool connection,
String firebaseUid, String firebaseUid,
) async { ) async {
final Result result = await connection.execute( final Result result = await connection.execute(

View File

@ -36,7 +36,7 @@ class MarketDataSnapshot {
class MarketDataDb { class MarketDataDb {
MarketDataDb(this._connection); MarketDataDb(this._connection);
final Connection _connection; final Pool _connection;
Future<MarketDataSnapshot> insertSnapshot({ Future<MarketDataSnapshot> insertSnapshot({
required String symbol, required String symbol,

View File

@ -122,7 +122,7 @@ class MarketDataHistorySync {
required AlpacaMarketDataClient marketDataClient, required AlpacaMarketDataClient marketDataClient,
required TradableAssetsDb tradableAssetsDb, required TradableAssetsDb tradableAssetsDb,
required MarketDataDb marketDataDb, required MarketDataDb marketDataDb,
required Connection connection, required Pool connection,
this.batchSize = MarketHistoryConfig.historySyncBatchSize, this.batchSize = MarketHistoryConfig.historySyncBatchSize,
this.maxSymbols = MarketHistoryConfig.historySyncMaxSymbols, this.maxSymbols = MarketHistoryConfig.historySyncMaxSymbols,
this.windowDays = MarketHistoryConfig.windowDays, this.windowDays = MarketHistoryConfig.windowDays,

View File

@ -29,7 +29,7 @@ class MarketDataRetentionResult {
/// [market_data_archive] inside the same transaction, then delete. /// [market_data_archive] inside the same transaction, then delete.
class MarketDataRetention { class MarketDataRetention {
MarketDataRetention({ MarketDataRetention({
required Connection connection, required Pool connection,
this.windowDays = MarketHistoryConfig.windowDays, this.windowDays = MarketHistoryConfig.windowDays,
this.batchSize = MarketHistoryConfig.retentionBatchSize, this.batchSize = MarketHistoryConfig.retentionBatchSize,
void Function(String sql)? onExecute, void Function(String sql)? onExecute,
@ -37,7 +37,7 @@ class MarketDataRetention {
_recorder = SyncRunRecorder(connection), _recorder = SyncRunRecorder(connection),
_onExecute = onExecute; _onExecute = onExecute;
final Connection _connection; final Pool _connection;
final SyncRunRecorder _recorder; final SyncRunRecorder _recorder;
final void Function(String sql)? _onExecute; final void Function(String sql)? _onExecute;

View File

@ -15,7 +15,7 @@ class AdminTriggerResult {
/// On-demand market-history operations for the admin portal. /// On-demand market-history operations for the admin portal.
class MarketHistoryAdminActions { class MarketHistoryAdminActions {
MarketHistoryAdminActions({ MarketHistoryAdminActions({
required Connection connection, required Pool connection,
required AdminRunStage runUniverse, required AdminRunStage runUniverse,
required AdminRunStage runBackfill, required AdminRunStage runBackfill,
required Future<void> Function(DateTime now, bool archive, int windowDays) required Future<void> Function(DateTime now, bool archive, int windowDays)
@ -28,7 +28,7 @@ class MarketHistoryAdminActions {
_runBackfill = runBackfill, _runBackfill = runBackfill,
_runCleanup = runCleanup; _runCleanup = runCleanup;
final Connection _connection; final Pool _connection;
final SyncRunRecorder _recorder; final SyncRunRecorder _recorder;
final AdminRunStage _runUniverse; final AdminRunStage _runUniverse;
final AdminRunStage _runBackfill; final AdminRunStage _runBackfill;

View File

@ -34,7 +34,7 @@ class ProspectiveQuestionsRefreshResult {
/// (and at refresh) when bar history is purged. /// (and at refresh) when bar history is purged.
class MarketHistoryProspectiveQuestions { class MarketHistoryProspectiveQuestions {
MarketHistoryProspectiveQuestions({ MarketHistoryProspectiveQuestions({
required Connection connection, required Pool connection,
MarketHistoryQuestionAudit? questionAudit, MarketHistoryQuestionAudit? questionAudit,
}) : _connection = connection, }) : _connection = connection,
_questionAudit = _questionAudit =
@ -46,7 +46,7 @@ class MarketHistoryProspectiveQuestions {
/// Single-flight refresh across worker ticks (`pg_advisory_xact_lock`). /// Single-flight refresh across worker ticks (`pg_advisory_xact_lock`).
static const int refreshAdvisoryLockKey = 824238151; static const int refreshAdvisoryLockKey = 824238151;
final Connection _connection; final Pool _connection;
final MarketHistoryQuestionAudit _questionAudit; final MarketHistoryQuestionAudit _questionAudit;
/// Upserts top-50% volume questions for the current default slot pair. /// Upserts top-50% volume questions for the current default slot pair.
@ -227,20 +227,6 @@ class MarketHistoryProspectiveQuestions {
); );
} }
Future<DateTime?> _latestCompareUntil() async {
final Result result = await _connection.execute(
'''
SELECT compare_until
FROM market_history_prospective_questions
ORDER BY compare_until DESC
LIMIT 1
''',
);
if (result.isEmpty) {
return null;
}
return (result.first[0]! as DateTime).toUtc();
}
static String _questionText(String symbol) => static String _questionText(String symbol) =>
'What was the percent price change for $symbol from the prior ' 'What was the percent price change for $symbol from the prior '

View File

@ -28,12 +28,12 @@ class WeeklyMover {
/// Read-side queries over 4-hour history bars for question rules. /// Read-side queries over 4-hour history bars for question rules.
class MarketHistoryQuery { class MarketHistoryQuery {
MarketHistoryQuery({ MarketHistoryQuery({
required Connection connection, required Pool connection,
TradableAssetsDb? tradableAssetsDb, TradableAssetsDb? tradableAssetsDb,
}) : _connection = connection, }) : _connection = connection,
_tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection); _tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection);
final Connection _connection; final Pool _connection;
final TradableAssetsDb _tradableAssetsDb; final TradableAssetsDb _tradableAssetsDb;
/// Symbols with ≥ [minBars] 4-hour closes in [`asOf` − window, `asOf`), whose /// Symbols with ≥ [minBars] 4-hour closes in [`asOf` − window, `asOf`), whose

View File

@ -290,12 +290,12 @@ DateTime questionAuditStepNewerCompareUntil({
/// Read-side audit of the two newest 4-hour bars per active tradable symbol. /// Read-side audit of the two newest 4-hour bars per active tradable symbol.
class MarketHistoryQuestionAudit { class MarketHistoryQuestionAudit {
MarketHistoryQuestionAudit({ MarketHistoryQuestionAudit({
required Connection connection, required Pool connection,
TradableAssetsDb? tradableAssetsDb, TradableAssetsDb? tradableAssetsDb,
}) : _connection = connection, }) : _connection = connection,
_tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection); _tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection);
final Connection _connection; final Pool _connection;
final TradableAssetsDb _tradableAssetsDb; final TradableAssetsDb _tradableAssetsDb;
Future<QuestionAuditPage> page({ Future<QuestionAuditPage> page({

View File

@ -87,7 +87,7 @@ class MarketHistoryWeekCoverageReport {
/// Validates session-half bar coverage per Eastern day for the admin week view. /// Validates session-half bar coverage per Eastern day for the admin week view.
class MarketHistoryWeekCoverage { class MarketHistoryWeekCoverage {
MarketHistoryWeekCoverage({ MarketHistoryWeekCoverage({
required Connection connection, required Pool connection,
TradableAssetsDb? tradableAssetsDb, TradableAssetsDb? tradableAssetsDb,
this.windowDays = MarketHistoryConfig.windowDays, this.windowDays = MarketHistoryConfig.windowDays,
this.timeframe = MarketHistoryConfig.barTimeframe, this.timeframe = MarketHistoryConfig.barTimeframe,
@ -95,7 +95,7 @@ class MarketHistoryWeekCoverage {
}) : _tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection), }) : _tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection),
_connection = connection; _connection = connection;
final Connection _connection; final Pool _connection;
final TradableAssetsDb _tradableAssetsDb; final TradableAssetsDb _tradableAssetsDb;
final int windowDays; final int windowDays;
final String timeframe; final String timeframe;

View File

@ -6,7 +6,7 @@ import 'market_history_session_slot.dart';
class ProspectiveGuessAssignmentsDb { class ProspectiveGuessAssignmentsDb {
ProspectiveGuessAssignmentsDb(this._connection); ProspectiveGuessAssignmentsDb(this._connection);
final Connection _connection; final Pool _connection;
static const String statusPending = 'pending'; static const String statusPending = 'pending';
static const String statusAnswered = 'answered'; static const String statusAnswered = 'answered';

View File

@ -28,7 +28,7 @@ class ProspectiveSlotBatch {
/// that batch). [slot_start] advances after every top-half symbol is answered. /// that batch). [slot_start] advances after every top-half symbol is answered.
class ProspectiveGuessSelection { class ProspectiveGuessSelection {
ProspectiveGuessSelection({ ProspectiveGuessSelection({
required Connection connection, required Pool connection,
UserTradingStateDb? tradingStateDb, UserTradingStateDb? tradingStateDb,
ProspectiveGuessAssignmentsDb? assignmentsDb, ProspectiveGuessAssignmentsDb? assignmentsDb,
MarketHistoryQuestionAudit? questionAudit, MarketHistoryQuestionAudit? questionAudit,
@ -40,7 +40,7 @@ class ProspectiveGuessSelection {
_questionAudit = _questionAudit =
questionAudit ?? MarketHistoryQuestionAudit(connection: connection); questionAudit ?? MarketHistoryQuestionAudit(connection: connection);
final Connection _connection; final Pool _connection;
final UserTradingStateDb _tradingStateDb; final UserTradingStateDb _tradingStateDb;
final ProspectiveGuessAssignmentsDb _assignmentsDb; final ProspectiveGuessAssignmentsDb _assignmentsDb;
final MarketHistoryQuestionAudit _questionAudit; final MarketHistoryQuestionAudit _questionAudit;

View File

@ -183,8 +183,6 @@ class RuleEngine {
DateTime? lastFiredAt, DateTime? lastFiredAt,
DateTime? now, DateTime? now,
}) { }) {
final DateTime evaluatedAt = (now ?? _clock()).toUtc();
if (rule.type != 'guess_weekly_move') { if (rule.type != 'guess_weekly_move') {
return RuleEvaluation( return RuleEvaluation(
rule: rule, rule: rule,

View File

@ -75,7 +75,7 @@ class SyncRunCounts {
class SyncRunRecorder { class SyncRunRecorder {
SyncRunRecorder(this._connection); SyncRunRecorder(this._connection);
final Connection _connection; final Pool _connection;
static const String abortSupersededMessage = static const String abortSupersededMessage =
'aborted: superseded by new worker sync'; 'aborted: superseded by new worker sync';

View File

@ -36,7 +36,7 @@ class TradableAssetRow {
class TradableAssetsDb { class TradableAssetsDb {
TradableAssetsDb(this._connection); TradableAssetsDb(this._connection);
final Connection _connection; final Pool _connection;
/// Upserts every entry in [assets] (PK on `symbol`) using [now] as the /// Upserts every entry in [assets] (PK on `symbol`) using [now] as the
/// shared `refreshed_at`, then marks any rows whose `refreshed_at` is /// shared `refreshed_at`, then marks any rows whose `refreshed_at` is

View File

@ -42,7 +42,7 @@ class TradableAssetsSync {
TradableAssetsSync({ TradableAssetsSync({
required AlpacaAssetsClient assetsClient, required AlpacaAssetsClient assetsClient,
required TradableAssetsDb assetsDb, required TradableAssetsDb assetsDb,
required Connection connection, required Pool connection,
}) : _assetsClient = assetsClient, }) : _assetsClient = assetsClient,
_assetsDb = assetsDb, _assetsDb = assetsDb,
_recorder = SyncRunRecorder(connection); _recorder = SyncRunRecorder(connection);

View File

@ -46,7 +46,7 @@ class TradeOrder {
class TradeOrdersDb { class TradeOrdersDb {
TradeOrdersDb(this._connection); TradeOrdersDb(this._connection);
final Connection _connection; final Pool _connection;
static const Uuid _uuid = Uuid(); static const Uuid _uuid = Uuid();
/// Counts non-terminal orders submitted at or after [since]. /// Counts non-terminal orders submitted at or after [since].

View File

@ -8,7 +8,7 @@ import 'trading_config.dart';
class TradingConfigDb { class TradingConfigDb {
TradingConfigDb(this._connection); TradingConfigDb(this._connection);
final Connection _connection; final Pool _connection;
Future<EffectiveTradingConfig?> resolveEffectiveConfig(String firebaseUid) async { Future<EffectiveTradingConfig?> resolveEffectiveConfig(String firebaseUid) async {
final Result result = await _connection.execute( final Result result = await _connection.execute(

View File

@ -8,7 +8,7 @@ import 'market_history_session_slot.dart';
class UserTradingStateDb { class UserTradingStateDb {
UserTradingStateDb(this._connection); UserTradingStateDb(this._connection);
final Connection _connection; final Pool _connection;
static const String ingestContextKey = 'ingest_last_fetch'; static const String ingestContextKey = 'ingest_last_fetch';
static const String rulesContextKey = 'rules'; static const String rulesContextKey = 'rules';

View File

@ -28,7 +28,7 @@ class MarketHistorySchedulerReport {
/// are aborted so a hung prior sync cannot block the worker. /// are aborted so a hung prior sync cannot block the worker.
class MarketHistoryScheduler { class MarketHistoryScheduler {
MarketHistoryScheduler({ MarketHistoryScheduler({
required Connection connection, required Pool connection,
this.config = const MarketHistorySchedulerConfig(), this.config = const MarketHistorySchedulerConfig(),
Future<void> Function(DateTime now)? runUniverse, Future<void> Function(DateTime now)? runUniverse,
Future<void> Function(DateTime now)? runBackfill, Future<void> Function(DateTime now)? runBackfill,
@ -43,7 +43,7 @@ class MarketHistoryScheduler {
_runProspectiveQuestions = runProspectiveQuestions, _runProspectiveQuestions = runProspectiveQuestions,
_backfillIsDue = backfillIsDue; _backfillIsDue = backfillIsDue;
final Connection _connection; final Pool _connection;
final SyncRunRecorder _recorder; final SyncRunRecorder _recorder;
final MarketHistorySchedulerConfig config; final MarketHistorySchedulerConfig config;
final Future<void> Function(DateTime now)? _runUniverse; final Future<void> Function(DateTime now)? _runUniverse;

View File

@ -20,10 +20,10 @@ class TestDb {
TestDb._(this.db, this._connection, this.databaseUrl); TestDb._(this.db, this._connection, this.databaseUrl);
final ProfileDb db; final ProfileDb db;
final Connection _connection; final Pool _connection;
final String databaseUrl; final String databaseUrl;
Connection get connection => _connection; Pool get connection => _connection;
static const String testDatabaseName = 'cyberhybridhub_test'; static const String testDatabaseName = 'cyberhybridhub_test';
@ -68,9 +68,9 @@ class TestDb {
: 'postgres'; : 'postgres';
final String adminUrl = uri.replace(path: '/$adminDb').toString(); final String adminUrl = uri.replace(path: '/$adminDb').toString();
final Connection admin = await Connection.open( final Pool admin = Pool.withEndpoints(
_endpointFromUrl(adminUrl), [_endpointFromUrl(adminUrl)],
settings: const ConnectionSettings(sslMode: SslMode.disable), settings: const PoolSettings(sslMode: SslMode.disable),
); );
try { try {
final Result exists = await admin.execute( final Result exists = await admin.execute(

View File

@ -18,7 +18,7 @@ import '../helpers/mock_http_client.dart';
import '../helpers/test_db.dart'; import '../helpers/test_db.dart';
Future<void> _seedTradables( Future<void> _seedTradables(
Connection connection, Pool connection,
List<String> symbols, List<String> symbols,
) async { ) async {
final TradableAssetsDb db = TradableAssetsDb(connection); final TradableAssetsDb db = TradableAssetsDb(connection);

View File

@ -68,7 +68,7 @@ Future<Response> _post(
} }
Future<void> _insertRun( Future<void> _insertRun(
Connection connection, { Pool connection, {
required String kind, required String kind,
required DateTime startedAt, required DateTime startedAt,
DateTime? finishedAt, DateTime? finishedAt,

View File

@ -39,7 +39,7 @@ Future<Response> _get(Handler handler, {required String bearer}) async {
} }
Future<void> _insertRecord( Future<void> _insertRecord(
Connection connection, Pool connection,
AdminSyncRunRecord run, AdminSyncRunRecord run,
) async { ) async {
await connection.execute( await connection.execute(

View File

@ -218,7 +218,7 @@ void main() {
} }
final DateTime t0 = DateTime.utc(2026, 5, 26, 12); final DateTime t0 = DateTime.utc(2026, 5, 26, 12);
final Connection connection = testDb!.connection; final Pool connection = testDb!.connection;
await scheduler( await scheduler(
runUniverse: (DateTime now) => recordStage(TradableAssetsSync.kind, now), runUniverse: (DateTime now) => recordStage(TradableAssetsSync.kind, now),

View File

@ -36,7 +36,7 @@ void main() {
return; return;
} }
final Connection connection = testDb!.connection; final Pool connection = testDb!.connection;
final DateTime asOf = DateTime.utc(2026, 5, 23, 13); final DateTime asOf = DateTime.utc(2026, 5, 23, 13);
// First insert. // First insert.
@ -97,7 +97,7 @@ void main() {
return; return;
} }
final Connection connection = testDb!.connection; final Pool connection = testDb!.connection;
final DateTime asOfBase = DateTime.utc(2026, 5, 23, 14); final DateTime asOfBase = DateTime.utc(2026, 5, 23, 14);
// Insert without naming timeframe — must default to 'tick'. // Insert without naming timeframe — must default to 'tick'.
@ -169,7 +169,7 @@ void main() {
return; return;
} }
final Connection connection = testDb!.connection; final Pool connection = testDb!.connection;
await connection.execute( await connection.execute(
''' '''
@ -199,7 +199,7 @@ void main() {
return; return;
} }
final Connection connection = testDb!.connection; final Pool connection = testDb!.connection;
// Seed enough rows that the planner has a real choice. // Seed enough rows that the planner has a real choice.
for (int i = 0; i < 60; i++) { for (int i = 0; i < 60; i++) {
@ -260,7 +260,7 @@ void main() {
return; return;
} }
final Connection connection = testDb!.connection; final Pool connection = testDb!.connection;
final DateTime started = DateTime.utc(2026, 5, 26, 10); final DateTime started = DateTime.utc(2026, 5, 26, 10);
final DateTime finished = DateTime.utc(2026, 5, 26, 10, 0, 5); final DateTime finished = DateTime.utc(2026, 5, 26, 10, 0, 5);

View File

@ -23,7 +23,7 @@ void main() {
return; return;
} }
final Connection connection = testDb!.connection; final Pool connection = testDb!.connection;
final Result tables = await connection.execute( final Result tables = await connection.execute(
''' '''
SELECT table_name SELECT table_name

View File

@ -3,6 +3,7 @@ library;
import 'package:cyberhybridhub_server/pipeline/question_pipeline.dart'; import 'package:cyberhybridhub_server/pipeline/question_pipeline.dart';
import 'package:cyberhybridhub_server/question_dispatcher.dart'; import 'package:cyberhybridhub_server/question_dispatcher.dart';
import 'package:cyberhybridhub_server/questions_db.dart';
import 'package:postgres/postgres.dart'; import 'package:postgres/postgres.dart';
import 'package:test/test.dart'; import 'package:test/test.dart';
@ -50,6 +51,49 @@ void main() {
expect(payload['presentation'], 'spelling_word'); expect(payload['presentation'], 'spelling_word');
}); });
test('spelling 100% with market enabled at 0 does not enqueue market fallback',
() async {
if (testDb == null) {
markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests');
return;
}
const String uid = 'dispatcher-spelling-100-no-market-fallback';
await testDb!.seedUser(uid);
await testDb!.questionSourcePreferencesDb.save(
uid,
<Map<String, dynamic>>[
<String, dynamic>{'id': 'market_history', 'enabled': true, 'ratio': 0},
<String, dynamic>{'id': 'spelling', 'enabled': true, 'ratio': 100},
],
);
final QuestionDispatcher dispatcher = testDb!.questionDispatcher();
await dispatcher.topUpQueue(uid);
final Result rows = await testDb!.connection.execute(
Sql.named(
'''
SELECT source_tag, pipeline_key
FROM questions
WHERE assigned_user_id = @uid AND user_response IS NULL
''',
),
parameters: <String, dynamic>{'uid': uid},
);
expect(rows, isNotEmpty);
for (final ResultRow row in rows) {
expect(
QuestionsDb.resolveSourceId(<String, dynamic>{
'sourceTag': row[0],
'pipelineKey': row[1],
}),
'spelling',
);
}
});
test('70/30 mix yields spelling and respects fall-through over cycles', () async { test('70/30 mix yields spelling and respects fall-through over cycles', () async {
if (testDb == null) { if (testDb == null) {
markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests'); markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests');

View File

@ -24,6 +24,92 @@ void main() {
await testDb?.close(); await testDb?.close();
}); });
test(
'listPanelQuestions excludes market when market_history enabled at ratio 0',
() async {
if (testDb == null) {
markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests');
return;
}
// The critical scenario: user sets spelling=100%, market stays enabled but
// ratio=0. A stale market question is already in the DB. The dispatcher
// must never serve it.
const String uid = 'spelling-100-market-0-panel-uid';
await testDb!.seedUser(uid);
await testDb!.questionSourcePreferencesDb.save(
uid,
<Map<String, dynamic>>[
<String, dynamic>{'id': 'market_history', 'enabled': true, 'ratio': 0},
<String, dynamic>{'id': 'spelling', 'enabled': true, 'ratio': 100},
],
);
await testDb!.questionsDb.createQuestion(
assignedUserId: uid,
questionText: 'Stale market question',
correctAnswer: 10,
sourceTag: 'market_history:prospective',
pipelineKey: 'trading',
);
final QuestionDispatcher dispatcher = testDb!.questionDispatcher();
final List<Map<String, dynamic>> panel =
await dispatcher.listPanelQuestions(uid);
expect(panel, isNotEmpty);
for (final Map<String, dynamic> payload in panel) {
expect(payload['sourceId'], isNot('market_history'));
}
expect(panel.first['sourceId'], 'spelling');
expect(panel.first['presentation'], 'spelling_word');
});
test('nextQuestionAfterAnswer never returns market when spelling is 100%',
() async {
if (testDb == null) {
markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests');
return;
}
const String uid = 'spelling-100-next-answer-uid';
await testDb!.seedUser(uid);
await testDb!.questionSourcePreferencesDb.save(
uid,
<Map<String, dynamic>>[
<String, dynamic>{'id': 'market_history', 'enabled': true, 'ratio': 0},
<String, dynamic>{'id': 'spelling', 'enabled': true, 'ratio': 100},
],
);
// Seed a stale market question that should never surface.
await testDb!.questionsDb.createQuestion(
assignedUserId: uid,
questionText: 'Old market question',
correctAnswer: 5,
sourceTag: 'market_history:prospective',
pipelineKey: 'trading',
);
final QuestionDispatcher dispatcher = testDb!.questionDispatcher();
// Simulate the context of "just answered a spelling question".
final Map<String, dynamic> fakeAnswered = <String, dynamic>{
'sourceTag': 'spelling:word_quiz',
'pipelineKey': 'spelling',
'pipelineStep': 'spelling_word_quiz',
};
final Map<String, dynamic>? next = await dispatcher.nextQuestionAfterAnswer(
firebaseUid: uid,
answeredQuestion: fakeAnswered,
);
expect(next, isNotNull);
expect(next!['sourceId'], isNot('market_history'));
expect(next['sourceId'], 'spelling');
});
test('listPanelQuestions excludes market when market_history disabled', () async { test('listPanelQuestions excludes market when market_history disabled', () async {
if (testDb == null) { if (testDb == null) {
markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests'); markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests');

View File

@ -63,20 +63,25 @@ void main() {
expect(rows.first[1], PipelineSteps.spellingWordQuiz); expect(rows.first[1], PipelineSteps.spellingWordQuiz);
expect(rows.first[2], 'spelling:word_quiz'); expect(rows.first[2], 'spelling:word_quiz');
// Word is selected eagerly at pipeline start — metadata is fully populated.
final Result meta = await testDb!.connection.execute( final Result meta = await testDb!.connection.execute(
Sql.named( Sql.named(
''' '''
SELECT metadata->>'spelling_item_id' SELECT metadata->>'spelling_item_id',
metadata->>'displayed_word',
metadata->>'canonical_word'
FROM questions FROM questions
WHERE assigned_user_id = @uid WHERE assigned_user_id = @uid
''', ''',
), ),
parameters: <String, dynamic>{'uid': uid}, parameters: <String, dynamic>{'uid': uid},
); );
expect(meta.first[0], isNull); expect(meta.first[0], isNotNull);
expect(meta.first[1], isNotNull);
expect(meta.first[2], isNotNull);
}); });
test('prepareSpellingQuestion selects word and creates links at tile load', test('prepareSpellingQuestion on an eagerly-prepared question returns payload without re-picking',
() async { () async {
if (testDb == null) { if (testDb == null) {
markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests'); markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests');
@ -332,8 +337,10 @@ void main() {
), ),
parameters: <String, dynamic>{'uid': uid}, parameters: <String, dynamic>{'uid': uid},
); );
expect(progressAfter.first[0], 0); // Eager prep: maintainSpellingForUser immediately creates a progress record
expect(linksAfter.first[0], 0); // and link for the freshly re-seeded question, so counts are 1 not 0.
expect(progressAfter.first[0], 1);
expect(linksAfter.first[0], 1);
final Result spellingQuestions = await testDb!.connection.execute( final Result spellingQuestions = await testDb!.connection.execute(
Sql.named( Sql.named(

View File

@ -33,7 +33,7 @@ void main() {
const String uid = 'trading-schema-test-uid'; const String uid = 'trading-schema-test-uid';
await testDb!.seedUser(uid); await testDb!.seedUser(uid);
final Connection connection = testDb!.connection; final Pool connection = testDb!.connection;
final DateTime asOf = DateTime.utc(2026, 5, 23, 12); final DateTime asOf = DateTime.utc(2026, 5, 23, 12);
await connection.execute( await connection.execute(

View File

@ -94,6 +94,33 @@ void main() {
expect(spelling.enabled, isTrue); expect(spelling.enabled, isTrue);
}); });
test('isSpellingFullyWeighted when spelling owns entire mix', () {
final List<QuestionSourcePreference> sources =
QuestionSourceConfigRepository.mergeWithDefaults(
<QuestionSourcePreference>[
const QuestionSourcePreference(
id: 'market_history',
label: 'Market question',
description: 'Prospective price-move guessing game',
enabled: true,
ratio: 0,
),
const QuestionSourcePreference(
id: 'spelling',
label: 'Spelling',
description: 'Spell-check quiz from curated word lists',
enabled: true,
ratio: 100,
),
],
);
expect(
QuestionSourceConfigRepository.isSpellingFullyWeighted(sources),
isTrue,
);
});
test('effectiveMix excludes disabled and comingSoon sources', () { test('effectiveMix excludes disabled and comingSoon sources', () {
final List<QuestionSourcePreference> sources = final List<QuestionSourcePreference> sources =
QuestionSourceConfigRepository.defaultPreferences; QuestionSourceConfigRepository.defaultPreferences;