From a7c370b7430fc04075cd6a71ab8d449038c747eb Mon Sep 17 00:00:00 2001 From: nathanAndrinoid Date: Fri, 5 Jun 2026 22:04:06 +0000 Subject: [PATCH] Fix database connection pooling and UI question filtering - 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 --- .../question_source_config_repository.dart | 8 ++ lib/screens/home_screen.dart | 6 +- lib/screens/landing_screen.dart | 3 - lib/services/questions_hub_service.dart | 65 +++++++++-- server/bin/server.dart | 4 - server/lib/db.dart | 26 ++--- .../market_history_admin_handler.dart | 2 +- server/lib/handlers/questions_handler.dart | 20 ++-- server/lib/pipeline/question_pipeline.dart | 83 ++++++++------ server/lib/question_dispatcher.dart | 102 ++++++++++++------ server/lib/question_service.dart | 1 - .../lib/question_source_preferences_db.dart | 2 +- server/lib/questions_db.dart | 4 +- server/lib/spelling/spelling_items_db.dart | 2 +- .../spelling/spelling_question_links_db.dart | 2 +- .../spelling/user_spelling_progress_db.dart | 2 +- server/lib/trading/guess_score_store.dart | 8 +- server/lib/trading/market_data_db.dart | 2 +- server/lib/trading/market_data_history.dart | 2 +- server/lib/trading/market_data_retention.dart | 4 +- .../trading/market_history_admin_actions.dart | 4 +- .../market_history_prospective_questions.dart | 18 +--- server/lib/trading/market_history_query.dart | 4 +- .../market_history_question_audit.dart | 4 +- .../trading/market_history_week_coverage.dart | 4 +- .../prospective_guess_assignments_db.dart | 2 +- .../trading/prospective_guess_selection.dart | 4 +- server/lib/trading/rule_engine.dart | 2 - server/lib/trading/sync_run_recorder.dart | 2 +- server/lib/trading/tradable_assets_db.dart | 2 +- server/lib/trading/tradable_assets_sync.dart | 2 +- server/lib/trading/trade_orders_db.dart | 2 +- server/lib/trading/trading_config_db.dart | 2 +- server/lib/trading/user_trading_state_db.dart | 2 +- .../lib/workers/market_history_scheduler.dart | 4 +- server/test/helpers/test_db.dart | 10 +- .../market_data_history_sync_test.dart | 2 +- .../market_history_admin_handler_test.dart | 2 +- .../market_history_admin_risk_test.dart | 2 +- .../market_history_scheduler_test.dart | 2 +- .../market_history_schema_test.dart | 10 +- server/test/integration/migration_test.dart | 2 +- .../question_dispatcher_mix_test.dart | 44 ++++++++ .../integration/spelling_only_panel_test.dart | 86 +++++++++++++++ .../integration/spelling_pipeline_test.dart | 17 ++- .../test/integration/trading_schema_test.dart | 2 +- ...uestion_source_config_repository_test.dart | 27 +++++ 47 files changed, 431 insertions(+), 181 deletions(-) diff --git a/lib/repositories/question_source_config_repository.dart b/lib/repositories/question_source_config_repository.dart index f586214..1721f3b 100644 --- a/lib/repositories/question_source_config_repository.dart +++ b/lib/repositories/question_source_config_repository.dart @@ -142,6 +142,14 @@ class QuestionSourceConfigRepository { return false; } + /// True when spelling is enabled and its effective mix is 100%. + static bool isSpellingFullyWeighted(List sources) { + if (!isSpellingEnabled(sources)) { + return false; + } + return (effectiveMix(sources)['spelling'] ?? 0) >= 100; + } + /// Effective share (0–100) for each enabled source; empty when none enabled. static Map effectiveMix( List sources, diff --git a/lib/screens/home_screen.dart b/lib/screens/home_screen.dart index af74bee..d05c0da 100644 --- a/lib/screens/home_screen.dart +++ b/lib/screens/home_screen.dart @@ -170,9 +170,13 @@ class HomeScreen extends StatelessWidget { ); final bool spellingEnabled = QuestionSourceConfigRepository.isSpellingEnabled(sources); + final bool spellingFullyWeighted = + QuestionSourceConfigRepository.isSpellingFullyWeighted( + sources, + ); final bool showSpellingPlaceholder = panelOpen && spellingEnabled && - !marketEnabled && + (spellingFullyWeighted || !marketEnabled) && (question == null || !question.isSpelling); final bool showMarketPlaceholder = panelOpen && marketEnabled && diff --git a/lib/screens/landing_screen.dart b/lib/screens/landing_screen.dart index 51fe415..b3c700f 100644 --- a/lib/screens/landing_screen.dart +++ b/lib/screens/landing_screen.dart @@ -105,9 +105,6 @@ class _LandingScreenState extends State { case 'cancelled-popup-request': case 'auth/popup-closed-by-user': 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.'; } } diff --git a/lib/services/questions_hub_service.dart b/lib/services/questions_hub_service.dart index fb43a38..18b2a9e 100644 --- a/lib/services/questions_hub_service.dart +++ b/lib/services/questions_hub_service.dart @@ -156,9 +156,7 @@ class QuestionsHubService { try { // Align client filters with server-side source preferences before fetch. await QuestionSourceConfigRepository.instance.loadFromRemote(); - final List fetched = await _api.fetchUnanswered(); - // GET /questions already returns only server-enabled sources. - final List active = fetched; + final List active = await _fetchUnansweredForPanel(); questionPanelOpen.value = true; @@ -198,6 +196,10 @@ class QuestionsHubService { if (QuestionSourceConfigRepository.isMarketOnlyEnabled(sources)) { return question.isMarketHistory; } + if (QuestionSourceConfigRepository.isSpellingFullyWeighted(sources) && + question.isMarketHistory) { + return false; + } if (!QuestionSourceConfigRepository.isMarketHistoryEnabled(sources)) { return question.sourceId != 'market_history'; } @@ -213,6 +215,54 @@ class QuestionsHubService { 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> _fetchUnansweredForPanel() async { + final List sources = + QuestionSourceConfigRepository.instance.current; + final Map 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 fetched = await _api.fetchUnanswered(); + final List 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 finalFetch = await _api.fetchUnanswered(); + return _filterForEnabledSources(finalFetch); + } + void closeQuestionPanel() { questionPanelOpen.value = false; } @@ -253,7 +303,7 @@ class QuestionsHubService { return; } - final List refreshed = await _api.fetchUnanswered(); + final List refreshed = await _fetchUnansweredForPanel(); if (refreshed.isEmpty) { if (result.nextQuestion != null && _acceptsQuestionSource(result.nextQuestion!)) { @@ -279,7 +329,8 @@ class QuestionsHubService { orElse: () => refreshed.first, ); } else if (result.nextQuestion != null && - result.nextQuestion!.id != skippedQuestionId) { + result.nextQuestion!.id != skippedQuestionId && + _acceptsQuestionSource(result.nextQuestion!)) { next = result.nextQuestion!; } @@ -370,7 +421,7 @@ class QuestionsHubService { return; } - final List refreshed = await _api.fetchUnanswered(); + final List refreshed = await _fetchUnansweredForPanel(); final List remaining = refreshed .where((IncomingQuestion q) => q.id != answeredQuestionId) .toList(); @@ -428,7 +479,7 @@ class QuestionsHubService { return false; } - final List refreshed = await _api.fetchUnanswered(); + final List refreshed = await _fetchUnansweredForPanel(); final List active = refreshed.where(_acceptsQuestionSource).toList(growable: false); if (active.isNotEmpty) { diff --git a/server/bin/server.dart b/server/bin/server.dart index fe4d911..e13b8d6 100644 --- a/server/bin/server.dart +++ b/server/bin/server.dart @@ -155,10 +155,6 @@ Future main() async { final TradableAssetsDb tradableAssetsDb = TradableAssetsDb(db.connection); final AlpacaAssetsClient assetsClient = AlpacaAssetsClient(env: env.alpaca); - final MarketHistoryApiRateLimiter historyRateLimiter = - MarketHistoryApiRateLimiter( - requestsPerMinute: mh.apiRequestsPerMinute, - ); final AlpacaMarketDataClient historyMarketDataClient = AlpacaMarketDataClient(env: env.alpaca); final TradableAssetsSync tradableAssetsSync = TradableAssetsSync( diff --git a/server/lib/db.dart b/server/lib/db.dart index fc8d584..ef54190 100644 --- a/server/lib/db.dart +++ b/server/lib/db.dart @@ -3,14 +3,14 @@ import 'dart:io'; import 'package:postgres/postgres.dart'; class ProfileDb { - ProfileDb(this._connection); + ProfileDb(this._pool); - final Connection _connection; + final Pool _pool; static Future connect(String databaseUrl) async { final Uri uri = Uri.parse(databaseUrl); - final Connection connection = await Connection.open( - Endpoint( + final Pool pool = Pool.withEndpoints( + [Endpoint( host: uri.host.isEmpty ? 'localhost' : uri.host, port: uri.hasPort ? uri.port : 5432, database: uri.pathSegments.isNotEmpty ? uri.pathSegments.last : 'postgres', @@ -18,13 +18,13 @@ class ProfileDb { password: uri.userInfo.contains(':') ? uri.userInfo.split(':').skip(1).join(':') : 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 migrate() async { final List files = Directory('migrations') @@ -43,13 +43,13 @@ class ProfileDb { .toList(); for (final String statement in statements) { - await _connection.execute(statement); + await _pool.execute(statement); } } } Future?> getProfile(String firebaseUid) async { - final Result result = await _connection.execute( + final Result result = await _pool.execute( Sql.named( 'SELECT firebase_uid, email, display_name, photo_url, locale, timezone, ' 'onboarding_done, revision, updated_at ' @@ -69,7 +69,7 @@ class ProfileDb { required Map body, required int clientRevision, }) async { - final Result existing = await _connection.execute( + final Result existing = await _pool.execute( Sql.named('SELECT revision FROM users WHERE firebase_uid = @uid'), parameters: {'uid': firebaseUid}, ); @@ -82,7 +82,7 @@ class ProfileDb { } } - await _connection.execute( + await _pool.execute( Sql.named( ''' INSERT INTO users ( @@ -119,7 +119,7 @@ class ProfileDb { return (await getProfile(firebaseUid))!; } - Future close() => _connection.close(); + Future close() => _pool.close(); Map _rowToJson(ResultRow row) { final DateTime updatedAt = row[8]! as DateTime; diff --git a/server/lib/handlers/market_history_admin_handler.dart b/server/lib/handlers/market_history_admin_handler.dart index e5c8322..b7225e0 100644 --- a/server/lib/handlers/market_history_admin_handler.dart +++ b/server/lib/handlers/market_history_admin_handler.dart @@ -41,7 +41,7 @@ class MarketHistoryAdminPortalConfig { Handler marketHistoryAdminHandler({ required FirebaseAuthVerifier auth, - required Connection connection, + required Pool connection, required Set adminFirebaseUids, MarketHistoryAdminActions? actions, MarketHistoryAdminPortalConfig? portalConfig, diff --git a/server/lib/handlers/questions_handler.dart b/server/lib/handlers/questions_handler.dart index 8133117..5576835 100644 --- a/server/lib/handlers/questions_handler.dart +++ b/server/lib/handlers/questions_handler.dart @@ -32,7 +32,7 @@ Handler questionsHandler({ final Map? question = await questionDispatcher.dispatchNextQuestion(firebaseUid); final int unansweredCount = - await questionsDb.countUnansweredQuestions(firebaseUid); + await questionDispatcher.countEnabledUnansweredQuestions(firebaseUid); final Map score = await questionsDb.getGuessScoreSummary(firebaseUid); return _jsonResponse(200, { @@ -92,7 +92,7 @@ Handler questionsHandler({ final Map? question = await questionDispatcher.dispatchNextQuestion(firebaseUid); final int unansweredCount = - await questionsDb.countUnansweredQuestions(firebaseUid); + await questionDispatcher.countEnabledUnansweredQuestions(firebaseUid); return _jsonResponse(200, { 'score': score, 'unansweredCount': unansweredCount, @@ -136,7 +136,7 @@ Handler questionsHandler({ ); } var unansweredCount = - await questionsDb.countUnansweredQuestions(firebaseUid); + await questionDispatcher.countEnabledUnansweredQuestions(firebaseUid); final Map? nextQuestion = await questionDispatcher.nextQuestionAfterAnswer( firebaseUid: firebaseUid, @@ -226,18 +226,12 @@ Handler questionsHandler({ return _jsonResponse(404, {'error': 'Not found'}); } final int unansweredCount = - await questionsDb.countUnansweredQuestions(firebaseUid); - final Map? nextRow = - await questionsDb.findNextUnansweredAfterDefer( - assignedUserId: firebaseUid, + await questionDispatcher.countEnabledUnansweredQuestions(firebaseUid); + final Map? nextQuestion = + await questionDispatcher.nextQuestionAfterDefer( + firebaseUid: firebaseUid, deferredQuestionId: id, ); - final Map? nextQuestion = nextRow == null - ? null - : questionsDb.toClientPayload( - nextRow, - unansweredCount: unansweredCount, - ); return _jsonResponse(200, { 'question': updated, 'unansweredCount': unansweredCount, diff --git a/server/lib/pipeline/question_pipeline.dart b/server/lib/pipeline/question_pipeline.dart index e08520c..10121b9 100644 --- a/server/lib/pipeline/question_pipeline.dart +++ b/server/lib/pipeline/question_pipeline.dart @@ -150,7 +150,7 @@ class QuestionPipeline { } } - Future _deliverPipelineQuestion({ + Future> _deliverPipelineQuestion({ required String firebaseUid, required String questionText, required num correctAnswer, @@ -159,7 +159,7 @@ class QuestionPipeline { required String pipelineStep, Map? metadata, }) async { - await _questionService.createAndDeliverQuestion( + return _questionService.createAndDeliverQuestion( assignedUserId: firebaseUid, questionText: testMode ? PipelineTestQuestions.text : questionText, @@ -452,6 +452,7 @@ class QuestionPipeline { ); final int unansweredCount = + await _questionDispatcher?.countEnabledUnansweredQuestions(firebaseUid) ?? await _questionsDb.countUnansweredQuestions(firebaseUid); return _questionsDb.toClientPayload( question, @@ -472,6 +473,7 @@ class QuestionPipeline { final Map? question = await _questionDispatcher?.dispatchNextQuestion(firebaseUid); final int unansweredCount = + await _questionDispatcher?.countEnabledUnansweredQuestions(firebaseUid) ?? await _questionsDb.countUnansweredQuestions(firebaseUid); return { 'unansweredCount': unansweredCount, @@ -507,27 +509,6 @@ class QuestionPipeline { ); } - Future _startRootPipeline(String firebaseUid) async { - if (!await _canEnqueue(firebaseUid)) { - return; - } - await _questionsDb.upsertPipelineState( - assignedUserId: firebaseUid, - pipelineKey: PipelineKeys.root, - step: PipelineSteps.chooseTrack, - context: {}, - ); - 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 _handleRootAnswer({ 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 _startSpellingPipeline(String firebaseUid) async { if (!await _canEnqueue(firebaseUid)) { return; @@ -752,22 +738,57 @@ class QuestionPipeline { 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: {}, + ); + await _deliverPipelineQuestion( + firebaseUid: firebaseUid, + questionText: 'Spelling question loading…', + correctAnswer: 0, + sourceTag: 'spelling:word_quiz', + pipelineKey: PipelineKeys.spelling, + pipelineStep: PipelineSteps.spellingWordQuiz, + metadata: {'spelling_pending': true}, + ); + return; + } + + // Word selected — store context and create a fully-prepared question. await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: PipelineKeys.spelling, step: PipelineSteps.spellingWordQuiz, - context: {}, + context: item.toPipelineContext(), ); - await _deliverPipelineQuestion( + + await _spellingProgressDb.ensureLink( firebaseUid: firebaseUid, - questionText: testMode - ? PipelineTestQuestions.text - : 'Spelling question loading…', - correctAnswer: 0, + spellingItemId: item.id, + ); + + final Map payload = await _deliverPipelineQuestion( + firebaseUid: firebaseUid, + questionText: 'Spell the word you hear.', + correctAnswer: item.correctAnswer, sourceTag: 'spelling:word_quiz', pipelineKey: PipelineKeys.spelling, pipelineStep: PipelineSteps.spellingWordQuiz, - metadata: {'spelling_pending': true}, + metadata: item.toQuestionMetadata(), + ); + + await _spellingQuestionLinksDb.linkQuestion( + questionId: payload['id'] as String, + firebaseUid: firebaseUid, + spellingItemId: item.id, ); } diff --git a/server/lib/question_dispatcher.dart b/server/lib/question_dispatcher.dart index 7eb2cca..cc6c7a8 100644 --- a/server/lib/question_dispatcher.dart +++ b/server/lib/question_dispatcher.dart @@ -24,19 +24,18 @@ class QuestionDispatcher { final QuestionSourcePreferencesDb _preferencesDb; 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. Future?> dispatchNextQuestion(String firebaseUid) async { await _questionsDb.ensureUserExists(firebaseUid); var rows = await _listEnabledUnansweredOrdered(firebaseUid); - if (rows.isEmpty) { - if (!await _enqueueOneFromMix(firebaseUid)) { - await _ensureSpellingQuestionWhenNeeded(firebaseUid); - } - rows = await _listEnabledUnansweredOrdered(firebaseUid); - } - if (rows.isEmpty) { - await _ensureSpellingQuestionWhenNeeded(firebaseUid); + for (var attempt = 0; + rows.isEmpty && attempt < _maxQueueAttempts; + attempt++) { + await _runQueueMaintenance(firebaseUid); rows = await _listEnabledUnansweredOrdered(firebaseUid); } @@ -46,41 +45,40 @@ class QuestionDispatcher { return _toClientPayload(rows.first, rows.length); } - /// Ensures an enabled-source question exists before the question panel opens. - Future 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. + /// + /// 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>> listPanelQuestions( String firebaseUid, ) async { - await preparePanelQueue(firebaseUid); + 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); + for (var attempt = 0; + rows.isEmpty && attempt < _maxQueueAttempts; + attempt++) { + await _runQueueMaintenance(firebaseUid); rows = await _listEnabledUnansweredOrdered(firebaseUid); } + return rows - .map( - (Map row) => - _toClientPayload(row, rows.length), - ) + .map((Map row) => _toClientPayload(row, rows.length)) .toList(growable: false); } + /// Counts the number of unanswered questions that belong to enabled sources. + Future countEnabledUnansweredQuestions(String firebaseUid) async { + final List prefs = + await _preferencesDb.getResolved(firebaseUid); + final Map mix = QuestionSourceMix.effectiveMixFromResolved(prefs); + final Set allowed = _enabledSourceIds(prefs) + .where((String id) => (mix[id] ?? 0) > 0) + .toSet(); + return _questionsDb.countUnansweredForAllowedSources(firebaseUid, allowed); + } + /// Next unanswered enabled-source question after an answer. Future?> nextQuestionAfterAnswer({ required String firebaseUid, @@ -96,7 +94,9 @@ class QuestionDispatcher { final Map mix = QuestionSourceMix.effectiveMixFromResolved(prefs); final Map delivered = await _questionsDb.countDeliveredBySourceId(firebaseUid); - final Set allowed = _enabledSourceIds(prefs); + final Set allowed = _enabledSourceIds(prefs) + .where((String id) => (mix[id] ?? 0) > 0) + .toSet(); final Map? row = await _questionsDb.findHeadUnansweredForMix( firebaseUid, effectiveMixPercent: mix, @@ -111,6 +111,24 @@ class QuestionDispatcher { return _toClientPayload(row, rows.length); } + /// Next unanswered enabled-source question after deferring a question. + Future?> nextQuestionAfterDefer({ + required String firebaseUid, + required String deferredQuestionId, + }) async { + final List> rows = + await _listEnabledUnansweredOrdered(firebaseUid); + if (rows.isEmpty) { + return dispatchNextQuestion(firebaseUid); + } + for (final Map 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. Future topUpQueue(String firebaseUid) async { 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 _runQueueMaintenance(String firebaseUid) async { + if (!await _enqueueOneFromMix(firebaseUid)) { + await _ensureSpellingQuestionWhenNeeded(firebaseUid); + } + } + Future _enqueueOneFromMix(String firebaseUid) async { final List prefs = await _preferencesDb.getResolved(firebaseUid); @@ -144,7 +172,7 @@ class QuestionDispatcher { final List order = _sourceTryOrderForEnqueue( prefs: prefs, primarySourceId: primary, - ); + ).where((String id) => (mix[id] ?? 0) > 0).toList(growable: false); final int? marketBatchCap = _marketBatchCapForMix(prefs); @@ -167,7 +195,11 @@ class QuestionDispatcher { final List prefs = await _preferencesDb.getResolved(firebaseUid); final Map mix = QuestionSourceMix.effectiveMixFromResolved(prefs); - final Set 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 allowed = _enabledSourceIds(prefs) + .where((String id) => (mix[id] ?? 0) > 0) + .toSet(); if (allowed.isEmpty) { return >[]; } diff --git a/server/lib/question_service.dart b/server/lib/question_service.dart index 5ce3fce..c27bac5 100644 --- a/server/lib/question_service.dart +++ b/server/lib/question_service.dart @@ -200,7 +200,6 @@ class QuestionService { return payload; } - /// On SignalR connect: deliver an existing unanswered question only (no create). Future deliverPendingQuestionOnConnect( QuestionsHubConnection connection, ) async { diff --git a/server/lib/question_source_preferences_db.dart b/server/lib/question_source_preferences_db.dart index e746a08..313643d 100644 --- a/server/lib/question_source_preferences_db.dart +++ b/server/lib/question_source_preferences_db.dart @@ -7,7 +7,7 @@ import 'question_source_preferences.dart'; class QuestionSourcePreferencesDb { QuestionSourcePreferencesDb(this._connection); - final Connection _connection; + final Pool _connection; Future> getResolved( String firebaseUid, diff --git a/server/lib/questions_db.dart b/server/lib/questions_db.dart index fd91ec0..c065b02 100644 --- a/server/lib/questions_db.dart +++ b/server/lib/questions_db.dart @@ -15,9 +15,9 @@ import 'trading/user_trading_state_db.dart'; class QuestionsDb { QuestionsDb(this._connection); - final Connection _connection; + final Pool _connection; - Connection get connection => _connection; + Pool get connection => _connection; static const Uuid _uuid = Uuid(); diff --git a/server/lib/spelling/spelling_items_db.dart b/server/lib/spelling/spelling_items_db.dart index 0deb741..fc4eae6 100644 --- a/server/lib/spelling/spelling_items_db.dart +++ b/server/lib/spelling/spelling_items_db.dart @@ -7,7 +7,7 @@ import 'spelling_item_record.dart'; class SpellingItemsDb { SpellingItemsDb(this._connection); - final Connection _connection; + final Pool _connection; /// Inserts curriculum from [SpellingWordList] when the catalog is empty. Future ensureCatalogSeeded() async { diff --git a/server/lib/spelling/spelling_question_links_db.dart b/server/lib/spelling/spelling_question_links_db.dart index ced46bf..c04bcc3 100644 --- a/server/lib/spelling/spelling_question_links_db.dart +++ b/server/lib/spelling/spelling_question_links_db.dart @@ -4,7 +4,7 @@ import 'package:postgres/postgres.dart'; class SpellingQuestionLinksDb { SpellingQuestionLinksDb(this._connection); - final Connection _connection; + final Pool _connection; /// Creates a link when the spelling tile loads; idempotent per [questionId]. Future linkQuestion({ diff --git a/server/lib/spelling/user_spelling_progress_db.dart b/server/lib/spelling/user_spelling_progress_db.dart index 17aaf4d..3f8b1a3 100644 --- a/server/lib/spelling/user_spelling_progress_db.dart +++ b/server/lib/spelling/user_spelling_progress_db.dart @@ -6,7 +6,7 @@ import 'spelling_progress_policy.dart'; class UserSpellingProgressDb { UserSpellingProgressDb(this._connection); - final Connection _connection; + final Pool _connection; /// Ensures a per-user link to [spellingItemId] with scores starting at zero. Future ensureLink({ diff --git a/server/lib/trading/guess_score_store.dart b/server/lib/trading/guess_score_store.dart index 1b4cfbd..d2e901a 100644 --- a/server/lib/trading/guess_score_store.dart +++ b/server/lib/trading/guess_score_store.dart @@ -11,7 +11,7 @@ abstract final class GuessScoreStore { /// Counters and total are reconciled from [market_history_prospective_answers] /// when stored JSON is missing or behind answer history (cross-device safe). static Future> loadSummary( - Connection connection, + Pool connection, String firebaseUid, ) async { final UserTradingStateDb tradingStateDb = UserTradingStateDb(connection); @@ -94,7 +94,7 @@ abstract final class GuessScoreStore { } static Future<_AnswerStats> _fetchAnswerStats( - Connection connection, + Pool connection, String firebaseUid, ) async { final Result result = await connection.execute( @@ -127,7 +127,7 @@ abstract final class GuessScoreStore { } static Future _recomputeTotalScore( - Connection connection, + Pool connection, String firebaseUid, ) async { final Result rows = await connection.execute( @@ -153,7 +153,7 @@ abstract final class GuessScoreStore { } static Future _latestSlotWire( - Connection connection, + Pool connection, String firebaseUid, ) async { final Result result = await connection.execute( diff --git a/server/lib/trading/market_data_db.dart b/server/lib/trading/market_data_db.dart index 46968a5..4f4c248 100644 --- a/server/lib/trading/market_data_db.dart +++ b/server/lib/trading/market_data_db.dart @@ -36,7 +36,7 @@ class MarketDataSnapshot { class MarketDataDb { MarketDataDb(this._connection); - final Connection _connection; + final Pool _connection; Future insertSnapshot({ required String symbol, diff --git a/server/lib/trading/market_data_history.dart b/server/lib/trading/market_data_history.dart index 7e23f75..f26a3c3 100644 --- a/server/lib/trading/market_data_history.dart +++ b/server/lib/trading/market_data_history.dart @@ -122,7 +122,7 @@ class MarketDataHistorySync { required AlpacaMarketDataClient marketDataClient, required TradableAssetsDb tradableAssetsDb, required MarketDataDb marketDataDb, - required Connection connection, + required Pool connection, this.batchSize = MarketHistoryConfig.historySyncBatchSize, this.maxSymbols = MarketHistoryConfig.historySyncMaxSymbols, this.windowDays = MarketHistoryConfig.windowDays, diff --git a/server/lib/trading/market_data_retention.dart b/server/lib/trading/market_data_retention.dart index e79665f..80eeb5f 100644 --- a/server/lib/trading/market_data_retention.dart +++ b/server/lib/trading/market_data_retention.dart @@ -29,7 +29,7 @@ class MarketDataRetentionResult { /// [market_data_archive] inside the same transaction, then delete. class MarketDataRetention { MarketDataRetention({ - required Connection connection, + required Pool connection, this.windowDays = MarketHistoryConfig.windowDays, this.batchSize = MarketHistoryConfig.retentionBatchSize, void Function(String sql)? onExecute, @@ -37,7 +37,7 @@ class MarketDataRetention { _recorder = SyncRunRecorder(connection), _onExecute = onExecute; - final Connection _connection; + final Pool _connection; final SyncRunRecorder _recorder; final void Function(String sql)? _onExecute; diff --git a/server/lib/trading/market_history_admin_actions.dart b/server/lib/trading/market_history_admin_actions.dart index c4874cb..49fc8f7 100644 --- a/server/lib/trading/market_history_admin_actions.dart +++ b/server/lib/trading/market_history_admin_actions.dart @@ -15,7 +15,7 @@ class AdminTriggerResult { /// On-demand market-history operations for the admin portal. class MarketHistoryAdminActions { MarketHistoryAdminActions({ - required Connection connection, + required Pool connection, required AdminRunStage runUniverse, required AdminRunStage runBackfill, required Future Function(DateTime now, bool archive, int windowDays) @@ -28,7 +28,7 @@ class MarketHistoryAdminActions { _runBackfill = runBackfill, _runCleanup = runCleanup; - final Connection _connection; + final Pool _connection; final SyncRunRecorder _recorder; final AdminRunStage _runUniverse; final AdminRunStage _runBackfill; diff --git a/server/lib/trading/market_history_prospective_questions.dart b/server/lib/trading/market_history_prospective_questions.dart index a9e0f4c..8dfd4b3 100644 --- a/server/lib/trading/market_history_prospective_questions.dart +++ b/server/lib/trading/market_history_prospective_questions.dart @@ -34,7 +34,7 @@ class ProspectiveQuestionsRefreshResult { /// (and at refresh) when bar history is purged. class MarketHistoryProspectiveQuestions { MarketHistoryProspectiveQuestions({ - required Connection connection, + required Pool connection, MarketHistoryQuestionAudit? questionAudit, }) : _connection = connection, _questionAudit = @@ -46,7 +46,7 @@ class MarketHistoryProspectiveQuestions { /// Single-flight refresh across worker ticks (`pg_advisory_xact_lock`). static const int refreshAdvisoryLockKey = 824238151; - final Connection _connection; + final Pool _connection; final MarketHistoryQuestionAudit _questionAudit; /// Upserts top-50% volume questions for the current default slot pair. @@ -227,20 +227,6 @@ class MarketHistoryProspectiveQuestions { ); } - Future _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) => 'What was the percent price change for $symbol from the prior ' diff --git a/server/lib/trading/market_history_query.dart b/server/lib/trading/market_history_query.dart index fea68a2..6e9c822 100644 --- a/server/lib/trading/market_history_query.dart +++ b/server/lib/trading/market_history_query.dart @@ -28,12 +28,12 @@ class WeeklyMover { /// Read-side queries over 4-hour history bars for question rules. class MarketHistoryQuery { MarketHistoryQuery({ - required Connection connection, + required Pool connection, TradableAssetsDb? tradableAssetsDb, }) : _connection = connection, _tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection); - final Connection _connection; + final Pool _connection; final TradableAssetsDb _tradableAssetsDb; /// Symbols with ≥ [minBars] 4-hour closes in [`asOf` − window, `asOf`), whose diff --git a/server/lib/trading/market_history_question_audit.dart b/server/lib/trading/market_history_question_audit.dart index b3e0cd8..47cc65f 100644 --- a/server/lib/trading/market_history_question_audit.dart +++ b/server/lib/trading/market_history_question_audit.dart @@ -290,12 +290,12 @@ DateTime questionAuditStepNewerCompareUntil({ /// Read-side audit of the two newest 4-hour bars per active tradable symbol. class MarketHistoryQuestionAudit { MarketHistoryQuestionAudit({ - required Connection connection, + required Pool connection, TradableAssetsDb? tradableAssetsDb, }) : _connection = connection, _tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection); - final Connection _connection; + final Pool _connection; final TradableAssetsDb _tradableAssetsDb; Future page({ diff --git a/server/lib/trading/market_history_week_coverage.dart b/server/lib/trading/market_history_week_coverage.dart index 6715945..d97b7c8 100644 --- a/server/lib/trading/market_history_week_coverage.dart +++ b/server/lib/trading/market_history_week_coverage.dart @@ -87,7 +87,7 @@ class MarketHistoryWeekCoverageReport { /// Validates session-half bar coverage per Eastern day for the admin week view. class MarketHistoryWeekCoverage { MarketHistoryWeekCoverage({ - required Connection connection, + required Pool connection, TradableAssetsDb? tradableAssetsDb, this.windowDays = MarketHistoryConfig.windowDays, this.timeframe = MarketHistoryConfig.barTimeframe, @@ -95,7 +95,7 @@ class MarketHistoryWeekCoverage { }) : _tradableAssetsDb = tradableAssetsDb ?? TradableAssetsDb(connection), _connection = connection; - final Connection _connection; + final Pool _connection; final TradableAssetsDb _tradableAssetsDb; final int windowDays; final String timeframe; diff --git a/server/lib/trading/prospective_guess_assignments_db.dart b/server/lib/trading/prospective_guess_assignments_db.dart index 4af2b35..fe906e9 100644 --- a/server/lib/trading/prospective_guess_assignments_db.dart +++ b/server/lib/trading/prospective_guess_assignments_db.dart @@ -6,7 +6,7 @@ import 'market_history_session_slot.dart'; class ProspectiveGuessAssignmentsDb { ProspectiveGuessAssignmentsDb(this._connection); - final Connection _connection; + final Pool _connection; static const String statusPending = 'pending'; static const String statusAnswered = 'answered'; diff --git a/server/lib/trading/prospective_guess_selection.dart b/server/lib/trading/prospective_guess_selection.dart index 6bce8e9..b02445c 100644 --- a/server/lib/trading/prospective_guess_selection.dart +++ b/server/lib/trading/prospective_guess_selection.dart @@ -28,7 +28,7 @@ class ProspectiveSlotBatch { /// that batch). [slot_start] advances after every top-half symbol is answered. class ProspectiveGuessSelection { ProspectiveGuessSelection({ - required Connection connection, + required Pool connection, UserTradingStateDb? tradingStateDb, ProspectiveGuessAssignmentsDb? assignmentsDb, MarketHistoryQuestionAudit? questionAudit, @@ -40,7 +40,7 @@ class ProspectiveGuessSelection { _questionAudit = questionAudit ?? MarketHistoryQuestionAudit(connection: connection); - final Connection _connection; + final Pool _connection; final UserTradingStateDb _tradingStateDb; final ProspectiveGuessAssignmentsDb _assignmentsDb; final MarketHistoryQuestionAudit _questionAudit; diff --git a/server/lib/trading/rule_engine.dart b/server/lib/trading/rule_engine.dart index d209651..39f8763 100644 --- a/server/lib/trading/rule_engine.dart +++ b/server/lib/trading/rule_engine.dart @@ -183,8 +183,6 @@ class RuleEngine { DateTime? lastFiredAt, DateTime? now, }) { - final DateTime evaluatedAt = (now ?? _clock()).toUtc(); - if (rule.type != 'guess_weekly_move') { return RuleEvaluation( rule: rule, diff --git a/server/lib/trading/sync_run_recorder.dart b/server/lib/trading/sync_run_recorder.dart index 8db77d2..f884775 100644 --- a/server/lib/trading/sync_run_recorder.dart +++ b/server/lib/trading/sync_run_recorder.dart @@ -75,7 +75,7 @@ class SyncRunCounts { class SyncRunRecorder { SyncRunRecorder(this._connection); - final Connection _connection; + final Pool _connection; static const String abortSupersededMessage = 'aborted: superseded by new worker sync'; diff --git a/server/lib/trading/tradable_assets_db.dart b/server/lib/trading/tradable_assets_db.dart index fc58162..465b722 100644 --- a/server/lib/trading/tradable_assets_db.dart +++ b/server/lib/trading/tradable_assets_db.dart @@ -36,7 +36,7 @@ class TradableAssetRow { class TradableAssetsDb { TradableAssetsDb(this._connection); - final Connection _connection; + final Pool _connection; /// Upserts every entry in [assets] (PK on `symbol`) using [now] as the /// shared `refreshed_at`, then marks any rows whose `refreshed_at` is diff --git a/server/lib/trading/tradable_assets_sync.dart b/server/lib/trading/tradable_assets_sync.dart index 6fe661a..4cc0a45 100644 --- a/server/lib/trading/tradable_assets_sync.dart +++ b/server/lib/trading/tradable_assets_sync.dart @@ -42,7 +42,7 @@ class TradableAssetsSync { TradableAssetsSync({ required AlpacaAssetsClient assetsClient, required TradableAssetsDb assetsDb, - required Connection connection, + required Pool connection, }) : _assetsClient = assetsClient, _assetsDb = assetsDb, _recorder = SyncRunRecorder(connection); diff --git a/server/lib/trading/trade_orders_db.dart b/server/lib/trading/trade_orders_db.dart index e496dff..8684fdc 100644 --- a/server/lib/trading/trade_orders_db.dart +++ b/server/lib/trading/trade_orders_db.dart @@ -46,7 +46,7 @@ class TradeOrder { class TradeOrdersDb { TradeOrdersDb(this._connection); - final Connection _connection; + final Pool _connection; static const Uuid _uuid = Uuid(); /// Counts non-terminal orders submitted at or after [since]. diff --git a/server/lib/trading/trading_config_db.dart b/server/lib/trading/trading_config_db.dart index d77a2b7..249d1ba 100644 --- a/server/lib/trading/trading_config_db.dart +++ b/server/lib/trading/trading_config_db.dart @@ -8,7 +8,7 @@ import 'trading_config.dart'; class TradingConfigDb { TradingConfigDb(this._connection); - final Connection _connection; + final Pool _connection; Future resolveEffectiveConfig(String firebaseUid) async { final Result result = await _connection.execute( diff --git a/server/lib/trading/user_trading_state_db.dart b/server/lib/trading/user_trading_state_db.dart index 457cb04..8cc99bd 100644 --- a/server/lib/trading/user_trading_state_db.dart +++ b/server/lib/trading/user_trading_state_db.dart @@ -8,7 +8,7 @@ import 'market_history_session_slot.dart'; class UserTradingStateDb { UserTradingStateDb(this._connection); - final Connection _connection; + final Pool _connection; static const String ingestContextKey = 'ingest_last_fetch'; static const String rulesContextKey = 'rules'; diff --git a/server/lib/workers/market_history_scheduler.dart b/server/lib/workers/market_history_scheduler.dart index 71b838c..324d385 100644 --- a/server/lib/workers/market_history_scheduler.dart +++ b/server/lib/workers/market_history_scheduler.dart @@ -28,7 +28,7 @@ class MarketHistorySchedulerReport { /// are aborted so a hung prior sync cannot block the worker. class MarketHistoryScheduler { MarketHistoryScheduler({ - required Connection connection, + required Pool connection, this.config = const MarketHistorySchedulerConfig(), Future Function(DateTime now)? runUniverse, Future Function(DateTime now)? runBackfill, @@ -43,7 +43,7 @@ class MarketHistoryScheduler { _runProspectiveQuestions = runProspectiveQuestions, _backfillIsDue = backfillIsDue; - final Connection _connection; + final Pool _connection; final SyncRunRecorder _recorder; final MarketHistorySchedulerConfig config; final Future Function(DateTime now)? _runUniverse; diff --git a/server/test/helpers/test_db.dart b/server/test/helpers/test_db.dart index d5f0ea1..29d629b 100644 --- a/server/test/helpers/test_db.dart +++ b/server/test/helpers/test_db.dart @@ -20,10 +20,10 @@ class TestDb { TestDb._(this.db, this._connection, this.databaseUrl); final ProfileDb db; - final Connection _connection; + final Pool _connection; final String databaseUrl; - Connection get connection => _connection; + Pool get connection => _connection; static const String testDatabaseName = 'cyberhybridhub_test'; @@ -68,9 +68,9 @@ class TestDb { : 'postgres'; final String adminUrl = uri.replace(path: '/$adminDb').toString(); - final Connection admin = await Connection.open( - _endpointFromUrl(adminUrl), - settings: const ConnectionSettings(sslMode: SslMode.disable), + final Pool admin = Pool.withEndpoints( + [_endpointFromUrl(adminUrl)], + settings: const PoolSettings(sslMode: SslMode.disable), ); try { final Result exists = await admin.execute( diff --git a/server/test/integration/market_data_history_sync_test.dart b/server/test/integration/market_data_history_sync_test.dart index df87ddf..078c7a3 100644 --- a/server/test/integration/market_data_history_sync_test.dart +++ b/server/test/integration/market_data_history_sync_test.dart @@ -18,7 +18,7 @@ import '../helpers/mock_http_client.dart'; import '../helpers/test_db.dart'; Future _seedTradables( - Connection connection, + Pool connection, List symbols, ) async { final TradableAssetsDb db = TradableAssetsDb(connection); diff --git a/server/test/integration/market_history_admin_handler_test.dart b/server/test/integration/market_history_admin_handler_test.dart index 851f846..b9b06fc 100644 --- a/server/test/integration/market_history_admin_handler_test.dart +++ b/server/test/integration/market_history_admin_handler_test.dart @@ -68,7 +68,7 @@ Future _post( } Future _insertRun( - Connection connection, { + Pool connection, { required String kind, required DateTime startedAt, DateTime? finishedAt, diff --git a/server/test/integration/market_history_admin_risk_test.dart b/server/test/integration/market_history_admin_risk_test.dart index 12b130e..22b9172 100644 --- a/server/test/integration/market_history_admin_risk_test.dart +++ b/server/test/integration/market_history_admin_risk_test.dart @@ -39,7 +39,7 @@ Future _get(Handler handler, {required String bearer}) async { } Future _insertRecord( - Connection connection, + Pool connection, AdminSyncRunRecord run, ) async { await connection.execute( diff --git a/server/test/integration/market_history_scheduler_test.dart b/server/test/integration/market_history_scheduler_test.dart index 656159e..1eded8e 100644 --- a/server/test/integration/market_history_scheduler_test.dart +++ b/server/test/integration/market_history_scheduler_test.dart @@ -218,7 +218,7 @@ void main() { } final DateTime t0 = DateTime.utc(2026, 5, 26, 12); - final Connection connection = testDb!.connection; + final Pool connection = testDb!.connection; await scheduler( runUniverse: (DateTime now) => recordStage(TradableAssetsSync.kind, now), diff --git a/server/test/integration/market_history_schema_test.dart b/server/test/integration/market_history_schema_test.dart index 414ea25..a4405d4 100644 --- a/server/test/integration/market_history_schema_test.dart +++ b/server/test/integration/market_history_schema_test.dart @@ -36,7 +36,7 @@ void main() { return; } - final Connection connection = testDb!.connection; + final Pool connection = testDb!.connection; final DateTime asOf = DateTime.utc(2026, 5, 23, 13); // First insert. @@ -97,7 +97,7 @@ void main() { return; } - final Connection connection = testDb!.connection; + final Pool connection = testDb!.connection; final DateTime asOfBase = DateTime.utc(2026, 5, 23, 14); // Insert without naming timeframe — must default to 'tick'. @@ -169,7 +169,7 @@ void main() { return; } - final Connection connection = testDb!.connection; + final Pool connection = testDb!.connection; await connection.execute( ''' @@ -199,7 +199,7 @@ void main() { return; } - final Connection connection = testDb!.connection; + final Pool connection = testDb!.connection; // Seed enough rows that the planner has a real choice. for (int i = 0; i < 60; i++) { @@ -260,7 +260,7 @@ void main() { return; } - final Connection connection = testDb!.connection; + final Pool connection = testDb!.connection; final DateTime started = DateTime.utc(2026, 5, 26, 10); final DateTime finished = DateTime.utc(2026, 5, 26, 10, 0, 5); diff --git a/server/test/integration/migration_test.dart b/server/test/integration/migration_test.dart index 8a79550..5a47878 100644 --- a/server/test/integration/migration_test.dart +++ b/server/test/integration/migration_test.dart @@ -23,7 +23,7 @@ void main() { return; } - final Connection connection = testDb!.connection; + final Pool connection = testDb!.connection; final Result tables = await connection.execute( ''' SELECT table_name diff --git a/server/test/integration/question_dispatcher_mix_test.dart b/server/test/integration/question_dispatcher_mix_test.dart index 86c1288..faab5ca 100644 --- a/server/test/integration/question_dispatcher_mix_test.dart +++ b/server/test/integration/question_dispatcher_mix_test.dart @@ -3,6 +3,7 @@ library; import 'package:cyberhybridhub_server/pipeline/question_pipeline.dart'; import 'package:cyberhybridhub_server/question_dispatcher.dart'; +import 'package:cyberhybridhub_server/questions_db.dart'; import 'package:postgres/postgres.dart'; import 'package:test/test.dart'; @@ -50,6 +51,49 @@ void main() { 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, + >[ + {'id': 'market_history', 'enabled': true, 'ratio': 0}, + {'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: {'uid': uid}, + ); + + expect(rows, isNotEmpty); + for (final ResultRow row in rows) { + expect( + QuestionsDb.resolveSourceId({ + 'sourceTag': row[0], + 'pipelineKey': row[1], + }), + 'spelling', + ); + } + }); + test('70/30 mix yields spelling and respects fall-through over cycles', () async { if (testDb == null) { markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests'); diff --git a/server/test/integration/spelling_only_panel_test.dart b/server/test/integration/spelling_only_panel_test.dart index 1888802..7738fc0 100644 --- a/server/test/integration/spelling_only_panel_test.dart +++ b/server/test/integration/spelling_only_panel_test.dart @@ -24,6 +24,92 @@ void main() { 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, + >[ + {'id': 'market_history', 'enabled': true, 'ratio': 0}, + {'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> panel = + await dispatcher.listPanelQuestions(uid); + + expect(panel, isNotEmpty); + for (final Map 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, + >[ + {'id': 'market_history', 'enabled': true, 'ratio': 0}, + {'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 fakeAnswered = { + 'sourceTag': 'spelling:word_quiz', + 'pipelineKey': 'spelling', + 'pipelineStep': 'spelling_word_quiz', + }; + + final Map? 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 { if (testDb == null) { markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests'); diff --git a/server/test/integration/spelling_pipeline_test.dart b/server/test/integration/spelling_pipeline_test.dart index 39a21fb..0cf88aa 100644 --- a/server/test/integration/spelling_pipeline_test.dart +++ b/server/test/integration/spelling_pipeline_test.dart @@ -63,20 +63,25 @@ void main() { expect(rows.first[1], PipelineSteps.spellingWordQuiz); 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( Sql.named( ''' - SELECT metadata->>'spelling_item_id' + SELECT metadata->>'spelling_item_id', + metadata->>'displayed_word', + metadata->>'canonical_word' FROM questions WHERE assigned_user_id = @uid ''', ), parameters: {'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 { if (testDb == null) { markTestSkipped('Set DATABASE_URL or TEST_DATABASE_URL for integration tests'); @@ -332,8 +337,10 @@ void main() { ), parameters: {'uid': uid}, ); - expect(progressAfter.first[0], 0); - expect(linksAfter.first[0], 0); + // Eager prep: maintainSpellingForUser immediately creates a progress record + // 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( Sql.named( diff --git a/server/test/integration/trading_schema_test.dart b/server/test/integration/trading_schema_test.dart index e523ded..2811016 100644 --- a/server/test/integration/trading_schema_test.dart +++ b/server/test/integration/trading_schema_test.dart @@ -33,7 +33,7 @@ void main() { const String uid = 'trading-schema-test-uid'; await testDb!.seedUser(uid); - final Connection connection = testDb!.connection; + final Pool connection = testDb!.connection; final DateTime asOf = DateTime.utc(2026, 5, 23, 12); await connection.execute( diff --git a/test/repositories/question_source_config_repository_test.dart b/test/repositories/question_source_config_repository_test.dart index e27c6d0..30e2cdf 100644 --- a/test/repositories/question_source_config_repository_test.dart +++ b/test/repositories/question_source_config_repository_test.dart @@ -94,6 +94,33 @@ void main() { expect(spelling.enabled, isTrue); }); + test('isSpellingFullyWeighted when spelling owns entire mix', () { + final List sources = + QuestionSourceConfigRepository.mergeWithDefaults( + [ + 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', () { final List sources = QuestionSourceConfigRepository.defaultPreferences;