import 'dart:async'; import 'dart:io'; import 'dart:math'; import '../question_dispatcher.dart'; import '../question_service.dart'; import '../question_source_preferences.dart'; import '../question_source_preferences_db.dart'; import '../questions_db.dart'; import '../spelling/spelling_answer_policy.dart'; import '../spelling/spelling_item_record.dart'; import '../spelling/spelling_item_selector.dart'; import '../spelling/spelling_items_db.dart'; import '../spelling/spelling_question_links_db.dart'; import '../spelling/user_spelling_progress_db.dart'; import '../trading/trading_pipeline.dart'; import 'branch_decision.dart'; import 'external_data_fetcher.dart'; /// Test-mode question shape for local pipeline testing. abstract final class PipelineTestQuestions { static const String text = 'Starter question: enter a whole number between -10 and 10.'; static int randomCorrectAnswer() => Random().nextInt(21) - 10; } /// Pipeline keys stored in [user_pipeline_state] and question rows. abstract final class PipelineKeys { static const String root = 'root'; static const String geography = 'geography'; static const String weather = 'weather'; static const String trading = 'trading'; static const String spelling = 'spelling'; } /// Steps within each pipeline branch. abstract final class PipelineSteps { static const String chooseTrack = 'choose_track'; static const String populationQuiz = 'population_quiz'; static const String capitalFollowUp = 'capital_follow_up'; static const String capitalRecovery = 'capital_recovery'; static const String temperatureQuiz = 'temperature_quiz'; static const String temperatureFollowUp = 'temperature_follow_up'; static const String spellingWordQuiz = 'spelling_word_quiz'; static const String spellingFollowUp = 'spelling_follow_up'; static const String idle = 'idle'; } /// Per-rule phases stored in [user_trading_state.context.rules.{ruleId}.phase]. abstract final class TradingPhases { /// No active question; rule engine evaluates each tick. static const String idle = 'idle'; /// Question delivered, awaiting +10/-10 from the user. static const String awaitConfirm = 'await_confirm'; /// User said +10; order staged in pending_orders for the actuator. static const String submitOrder = 'submit_order'; /// Guess-the-move question awaiting +10/-10 (scoring only, no order). static const String awaitAnswer = 'await_answer'; /// Outcome recorded; rule returns to [idle] after cooldown rolls off. static const String done = 'done'; } /// Orchestrates API-driven question creation and branches on user answers. class QuestionPipeline { QuestionPipeline({ required QuestionsDb questionsDb, required QuestionService questionService, ExternalDataFetcher? fetcher, TradingPipeline? tradingPipeline, QuestionSourcePreferencesDb? questionSourcePreferencesDb, SpellingItemsDb? spellingItemsDb, SpellingItemSelector? spellingItemSelector, UserSpellingProgressDb? spellingProgressDb, SpellingQuestionLinksDb? spellingQuestionLinksDb, this.maxQueuedQuestions = 3, this.testMode = false, }) : _questionsDb = questionsDb, _questionService = questionService, _fetcher = fetcher ?? ExternalDataFetcher(), _tradingPipeline = tradingPipeline, _questionSourcePreferencesDb = questionSourcePreferencesDb, _spellingItemsDb = spellingItemsDb ?? SpellingItemsDb(questionsDb.connection), _spellingProgressDb = spellingProgressDb ?? UserSpellingProgressDb(questionsDb.connection), _spellingQuestionLinksDb = spellingQuestionLinksDb ?? SpellingQuestionLinksDb(questionsDb.connection), _spellingSelector = spellingItemSelector ?? SpellingItemSelector( itemsDb: spellingItemsDb ?? SpellingItemsDb(questionsDb.connection), ); final QuestionsDb _questionsDb; final QuestionService _questionService; final ExternalDataFetcher _fetcher; final TradingPipeline? _tradingPipeline; final QuestionSourcePreferencesDb? _questionSourcePreferencesDb; final SpellingItemsDb _spellingItemsDb; final SpellingItemSelector _spellingSelector; final UserSpellingProgressDb _spellingProgressDb; final SpellingQuestionLinksDb _spellingQuestionLinksDb; final int maxQueuedQuestions; final bool testMode; QuestionDispatcher? _questionDispatcher; /// Wired after construction in [bin/server.dart] to avoid a circular dependency. void attachQuestionDispatcher(QuestionDispatcher dispatcher) { _questionDispatcher = dispatcher; } /// Periodic tick: start pipelines for idle users and top up shallow queues. Future runMaintenanceCycle() async { final List userIds = await _questionsDb.listAllUserFirebaseUids(); for (final String uid in userIds) { try { await _maintainUser(uid); } catch (e, st) { stderr.writeln('Pipeline maintenance failed for $uid: $e\n$st'); } } } Future _maintainUser(String firebaseUid) async { final QuestionDispatcher? dispatcher = _questionDispatcher; if (dispatcher != null) { await dispatcher.topUpQueue(firebaseUid); } var queued = await _questionsDb.countUnansweredQuestions(firebaseUid); if (queued >= maxQueuedQuestions) { return; } final Map? state = await _questionsDb.getPipelineState(firebaseUid); if (state == null) { return; } final String step = state['step'] as String; if (step == PipelineSteps.idle && queued == 0) { await _advanceFromIdle(firebaseUid, state); } } Future> _deliverPipelineQuestion({ required String firebaseUid, required String questionText, required num correctAnswer, required String sourceTag, required String pipelineKey, required String pipelineStep, Map? metadata, }) async { return _questionService.createAndDeliverQuestion( assignedUserId: firebaseUid, questionText: testMode ? PipelineTestQuestions.text : questionText, correctAnswer: testMode ? PipelineTestQuestions.randomCorrectAnswer() : correctAnswer, sourceTag: sourceTag, pipelineKey: pipelineKey, pipelineStep: pipelineStep, metadata: metadata, ); } BranchOutcome _evaluateAnswer({ required num userResponse, required num correctAnswer, required BranchOutcome Function() production, }) { if (testMode) { return userResponse.round() == correctAnswer.round() ? BranchOutcome.match : BranchOutcome.miss; } return production(); } Future _canEnqueue(String firebaseUid) async { final int queued = await _enabledUnansweredCount(firebaseUid); return queued < maxQueuedQuestions; } Future _enabledUnansweredCount(String firebaseUid) async { final QuestionSourcePreferencesDb? preferencesDb = _questionSourcePreferencesDb; if (preferencesDb == null) { return _questionsDb.countUnansweredQuestions(firebaseUid); } final List prefs = await preferencesDb.getResolved(firebaseUid); final Map mix = QuestionSourcePreferences.effectiveMix(prefs); final Set allowed = prefs .where( (ResolvedQuestionSourcePreference s) => s.enabled && !s.comingSoon && (mix[s.id] ?? 0) > 0, ) .map((ResolvedQuestionSourcePreference s) => s.id) .toSet(); return _questionsDb.countUnansweredForAllowedSources(firebaseUid, allowed); } /// Called after a question is answered; may enqueue the next branched question. Future onAnswerSubmitted({ required String firebaseUid, required Map answeredQuestion, required num userResponse, }) async { final String? pipelineKey = answeredQuestion['pipelineKey'] as String?; final String? pipelineStep = answeredQuestion['pipelineStep'] as String?; if (pipelineKey == null || pipelineStep == null) { return; } // Trading answers (including scoring) must run even when the queue is full. if (pipelineKey == PipelineKeys.trading && _tradingPipeline != null) { await _tradingPipeline.handleAnswer( firebaseUid: firebaseUid, answeredQuestion: answeredQuestion, userResponse: userResponse, ); } if (pipelineKey != PipelineKeys.spelling && !await _canEnqueue(firebaseUid)) { return; } final num correctAnswer = answeredQuestion['correctAnswer'] as num; final Map context = await _loadContext(firebaseUid, pipelineKey); switch (pipelineKey) { case PipelineKeys.root: await _handleRootAnswer( firebaseUid: firebaseUid, step: pipelineStep, userResponse: userResponse, ); case PipelineKeys.geography: await _handleGeographyAnswer( firebaseUid: firebaseUid, step: pipelineStep, userResponse: userResponse, correctAnswer: correctAnswer, context: context, ); case PipelineKeys.weather: await _handleWeatherAnswer( firebaseUid: firebaseUid, step: pipelineStep, userResponse: userResponse, correctAnswer: correctAnswer, context: context, ); case PipelineKeys.trading: break; case PipelineKeys.spelling: await _handleSpellingAnswer( firebaseUid: firebaseUid, step: pipelineStep, userResponse: userResponse, correctAnswer: correctAnswer, context: context, answeredQuestion: answeredQuestion, ); } } /// Selects the next word, links progress, and returns the prepared client payload. /// /// Called when the spelling tile loads (word selection runs here, not at enqueue). Future?> prepareSpellingQuestion({ required String firebaseUid, required String questionId, }) async { if (!await _isSpellingEnabled(firebaseUid)) { stderr.writeln( 'prepareSpelling: spelling source disabled for $firebaseUid', ); return null; } Map? question = await _questionsDb.findQuestionById( questionId: questionId, assignedUserId: firebaseUid, ); if (question == null) { stderr.writeln( 'prepareSpelling: question not found id=$questionId uid=$firebaseUid', ); return null; } if (QuestionsDb.resolveSourceId(question) != 'spelling') { stderr.writeln( 'prepareSpelling: not a spelling question id=$questionId', ); return null; } if (question['userResponse'] != null) { stderr.writeln( 'prepareSpelling: already answered id=$questionId', ); return null; } final String? step = question['pipelineStep'] as String?; final String? sourceTag = question['sourceTag'] as String?; if (step == PipelineSteps.spellingFollowUp) { stderr.writeln( 'prepareSpelling: follow-up question is not preparable id=$questionId', ); return null; } final bool preparable = step == PipelineSteps.spellingWordQuiz || sourceTag == 'spelling:word_quiz'; if (!preparable) { stderr.writeln( 'prepareSpelling: unsupported step=$step sourceTag=$sourceTag id=$questionId', ); return null; } final Map metadata = Map.from( question['metadata'] as Map? ?? {}, ); final String? existingWord = metadata['displayed_word'] as String?; if (existingWord != null && existingWord.isNotEmpty) { return _finalizePreparedSpellingQuestion( firebaseUid: firebaseUid, questionId: questionId, question: question, metadata: metadata, ); } final bool needsPick = metadata['spelling_pending'] == true || metadata['spelling_item_id'] == null; SpellingItemRecord? item; if (needsPick) { await _spellingItemsDb.ensureCatalogSeeded(); item = await _spellingSelector.pickForUser(firebaseUid); if (item == null) { stderr.writeln( 'prepareSpelling: no spelling items available for $firebaseUid', ); return null; } final Map itemMetadata = item.toQuestionMetadata(); final String questionText = testMode ? (question['text'] as String? ?? PipelineTestQuestions.text) : 'Spell the word you hear.'; question = await _questionsDb.updateSpellingQuizQuestion( questionId: questionId, assignedUserId: firebaseUid, questionText: questionText, correctAnswer: item.correctAnswer, metadata: itemMetadata, ); if (question == null) { stderr.writeln( 'prepareSpelling: update failed id=$questionId step=$step tag=$sourceTag', ); return null; } await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: PipelineKeys.spelling, step: PipelineSteps.spellingWordQuiz, context: item.toPipelineContext(), ); } else { item = _itemFromMetadata(metadata, question); if (item == null) { stderr.writeln( 'prepareSpelling: incomplete metadata id=$questionId', ); return null; } } return _finalizePreparedSpellingQuestion( firebaseUid: firebaseUid, questionId: questionId, question: question, metadata: metadata, item: item, ); } SpellingItemRecord? _itemFromMetadata( Map metadata, Map question, ) { final String? itemId = metadata['spelling_item_id'] as String?; final String? wordId = metadata['spelling_word_id'] as String?; final String? canonical = metadata['canonical_word'] as String?; final String? displayed = metadata['displayed_word'] as String?; final Object? isCorrect = metadata['is_correctly_spelled']; if (itemId == null || itemId.isEmpty || wordId == null || canonical == null || displayed == null || isCorrect is! bool) { return null; } return SpellingItemRecord( id: itemId, spellingWordId: wordId, canonicalWord: canonical, displayedWord: displayed, isCorrectlySpelled: isCorrect, correctAnswer: question['correctAnswer'] as num, ); } Future?> _finalizePreparedSpellingQuestion({ required String firebaseUid, required String questionId, required Map question, required Map metadata, SpellingItemRecord? item, }) async { final SpellingItemRecord? resolved = item ?? _itemFromMetadata(metadata, question); if (resolved == null) { return null; } await _spellingProgressDb.ensureLink( firebaseUid: firebaseUid, spellingItemId: resolved.id, ); await _spellingQuestionLinksDb.linkQuestion( questionId: questionId, firebaseUid: firebaseUid, spellingItemId: resolved.id, ); final int unansweredCount = await _questionDispatcher?.countEnabledUnansweredQuestions(firebaseUid) ?? await _questionsDb.countUnansweredQuestions(firebaseUid); return _questionsDb.toClientPayload( question, unansweredCount: unansweredCount, ); } /// Clears spelling links, progress, questions, and pipeline state; re-seeds when enabled. Future> resetSpellingProgressForUser( String firebaseUid, ) async { await _spellingQuestionLinksDb.deleteAllForUser(firebaseUid); await _spellingProgressDb.deleteAllForUser(firebaseUid); await _questionsDb.deleteAllSpellingQuestionsForUser(firebaseUid); await _questionsDb.clearSpellingPipelineState(firebaseUid); await maintainSpellingForUser(firebaseUid); final Map? question = await _questionDispatcher?.dispatchNextQuestion(firebaseUid); final int unansweredCount = await _questionDispatcher?.countEnabledUnansweredQuestions(firebaseUid) ?? await _questionsDb.countUnansweredQuestions(firebaseUid); return { 'unansweredCount': unansweredCount, if (question != null) 'question': question, }; } /// Ensures an unanswered spelling word quiz exists when the source is enabled. Future maintainSpellingForUser(String firebaseUid) async { if (!await _isSpellingEnabled(firebaseUid)) { return; } if (await _questionsDb.hasUnansweredSpellingWordQuiz(firebaseUid)) { return; } if (!await _canEnqueue(firebaseUid)) { return; } await _startSpellingPipeline(firebaseUid); } Future> _loadContext( String firebaseUid, String pipelineKey, ) async { final Map? state = await _questionsDb.getPipelineState(firebaseUid); if (state == null || state['pipelineKey'] != pipelineKey) { return {}; } return Map.from( state['context'] as Map? ?? {}, ); } Future _handleRootAnswer({ required String firebaseUid, required String step, required num userResponse, }) async { if (step != PipelineSteps.chooseTrack) { return; } final BranchOutcome outcome = BranchDecision.trackPreference(userResponse); if (outcome == BranchOutcome.preferHigh) { await _startWeatherPipeline(firebaseUid); } else { await _startGeographyPipeline(firebaseUid); } } Future _startGeographyPipeline(String firebaseUid) async { if (!await _canEnqueue(firebaseUid)) { return; } final CountryFacts? country = testMode ? null : await _fetcher.fetchRandomCountry(); if (!testMode && country == null) { await _setIdle(firebaseUid, PipelineKeys.geography); return; } final Map context = testMode ? { 'country': 'Testland', 'populationMillions': 40, 'capital': 'Test City', } : { 'country': country!.name, 'populationMillions': country.populationMillions, 'capital': country.capital, }; await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: PipelineKeys.geography, step: PipelineSteps.populationQuiz, context: context, ); const int thresholdMillions = 30; final bool overThreshold = (context['populationMillions'] as int) >= thresholdMillions; await _deliverPipelineQuestion( firebaseUid: firebaseUid, questionText: 'Does ${context['country']} have a population over $thresholdMillions million? ' 'Swipe toward +10 for yes, toward -10 for no.', correctAnswer: overThreshold ? 10 : -10, sourceTag: 'pipeline:geography:population', pipelineKey: PipelineKeys.geography, pipelineStep: PipelineSteps.populationQuiz, ); } Future _handleGeographyAnswer({ required String firebaseUid, required String step, required num userResponse, required num correctAnswer, required Map context, }) async { final String country = context['country'] as String? ?? 'this country'; final String capital = context['capital'] as String? ?? ''; switch (step) { case PipelineSteps.populationQuiz: final BranchOutcome outcome = _evaluateAnswer( userResponse: userResponse, correctAnswer: correctAnswer, production: () => BranchDecision.yesNo( userResponse: userResponse, correctAnswer: correctAnswer, ), ); if (outcome == BranchOutcome.match) { await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: PipelineKeys.geography, step: PipelineSteps.capitalFollowUp, context: context, ); await _deliverPipelineQuestion( firebaseUid: firebaseUid, questionText: 'Nice estimate! What is the capital of $country? ' 'Enter 1 if it is $capital, or -1 if you are unsure.', correctAnswer: 1, sourceTag: 'pipeline:geography:capital', pipelineKey: PipelineKeys.geography, pipelineStep: PipelineSteps.capitalFollowUp, ); } else { await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: PipelineKeys.geography, step: PipelineSteps.capitalRecovery, context: context, ); await _deliverPipelineQuestion( firebaseUid: firebaseUid, questionText: 'The population of $country is about ' '${context['populationMillions']} million. ' 'Try again: enter 1 if the capital is $capital, -1 otherwise.', correctAnswer: 1, sourceTag: 'pipeline:geography:recovery', pipelineKey: PipelineKeys.geography, pipelineStep: PipelineSteps.capitalRecovery, ); } case PipelineSteps.capitalFollowUp: case PipelineSteps.capitalRecovery: await _setIdle(firebaseUid, PipelineKeys.geography); } } Future _startWeatherPipeline(String firebaseUid) async { if (!await _canEnqueue(firebaseUid)) { return; } final WeatherFacts? weather = testMode ? null : await _fetcher.fetchRandomCityWeather(); if (!testMode && weather == null) { await _setIdle(firebaseUid, PipelineKeys.weather); return; } final Map context = testMode ? {'city': 'Test City', 'temperatureCelsius': 20} : { 'city': weather!.city, 'temperatureCelsius': weather.temperatureCelsius, }; await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: PipelineKeys.weather, step: PipelineSteps.temperatureQuiz, context: context, ); const int warmThresholdC = 15; final bool isWarm = (context['temperatureCelsius'] as int) >= warmThresholdC; await _deliverPipelineQuestion( firebaseUid: firebaseUid, questionText: 'Is it $warmThresholdC°C or warmer in ${context['city']} right now? ' 'Swipe toward +10 for yes, toward -10 for no.', correctAnswer: isWarm ? 10 : -10, sourceTag: 'pipeline:weather:temperature', pipelineKey: PipelineKeys.weather, pipelineStep: PipelineSteps.temperatureQuiz, ); } Future _handleWeatherAnswer({ required String firebaseUid, required String step, required num userResponse, required num correctAnswer, required Map context, }) async { final String city = context['city'] as String? ?? 'the city'; final int actual = (context['temperatureCelsius'] as num?)?.round() ?? 0; switch (step) { case PipelineSteps.temperatureQuiz: final BranchOutcome outcome = _evaluateAnswer( userResponse: userResponse, correctAnswer: correctAnswer, production: () => BranchDecision.yesNo( userResponse: userResponse, correctAnswer: correctAnswer, ), ); final String followUp = switch (outcome) { BranchOutcome.match => 'Spot on! It is about $actual°C in $city. ' 'Enter +1 to get another weather question, -1 to switch to geography.', BranchOutcome.close => 'Close — it is about $actual°C in $city. ' 'Enter +1 for another weather round, -1 for geography instead.', _ => 'It is about $actual°C in $city right now. ' 'Enter +1 to try another city, -1 to try geography questions.', }; await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: PipelineKeys.weather, step: PipelineSteps.temperatureFollowUp, context: context, ); await _deliverPipelineQuestion( firebaseUid: firebaseUid, questionText: followUp, correctAnswer: 1, sourceTag: 'pipeline:weather:follow_up', pipelineKey: PipelineKeys.weather, pipelineStep: PipelineSteps.temperatureFollowUp, ); case PipelineSteps.temperatureFollowUp: if (userResponse > 0) { await _startWeatherPipeline(firebaseUid); } else { await _startGeographyPipeline(firebaseUid); } } } /// 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, {bool force = false}) async { if (!force && !await _canEnqueue(firebaseUid)) { return; } if (!await _isSpellingEnabled(firebaseUid)) { 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: item.toPipelineContext(), ); await _spellingProgressDb.ensureLink( firebaseUid: firebaseUid, 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: item.toQuestionMetadata(), ); await _spellingQuestionLinksDb.linkQuestion( questionId: payload['id'] as String, firebaseUid: firebaseUid, spellingItemId: item.id, ); } Future _handleSpellingAnswer({ required String firebaseUid, required String step, required num userResponse, required num correctAnswer, required Map context, required Map answeredQuestion, }) async { final String canonicalWord = context['canonicalWord'] as String? ?? 'the word'; switch (step) { case PipelineSteps.spellingWordQuiz: final String typed = (answeredQuestion['userResponseText'] as String?)?.trim() ?? ''; final BranchOutcome outcome = SpellingAnswerPolicy.matches(typed, canonicalWord) ? BranchOutcome.match : BranchOutcome.miss; final String? spellingItemId = context['spellingItemId'] as String?; if (spellingItemId != null) { await _spellingProgressDb.recordQuizOutcome( firebaseUid: firebaseUid, spellingItemId: spellingItemId, correct: outcome == BranchOutcome.match, ); } final String followUp = switch (outcome) { BranchOutcome.match => "Correct — '$canonicalWord' is the proper spelling. " 'Enter +1 for another word, -1 to pause.', _ => "The proper spelling is '$canonicalWord'. " 'Enter +1 to try another word, -1 to pause.', }; await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: PipelineKeys.spelling, step: PipelineSteps.spellingFollowUp, context: context, ); await _deliverPipelineQuestion( firebaseUid: firebaseUid, questionText: followUp, correctAnswer: 1, sourceTag: 'spelling:follow_up', pipelineKey: PipelineKeys.spelling, pipelineStep: PipelineSteps.spellingFollowUp, ); case PipelineSteps.spellingFollowUp: if (userResponse > 0) { await _startSpellingPipeline(firebaseUid, force: true); } else { await _setIdle(firebaseUid, PipelineKeys.spelling); } } } Future _isSpellingEnabled(String firebaseUid) async { final QuestionSourcePreferencesDb? preferencesDb = _questionSourcePreferencesDb; if (preferencesDb == null) { return true; } return preferencesDb.isSourceEnabled(firebaseUid, 'spelling'); } Future _advanceFromIdle( String firebaseUid, Map state, ) async { final String pipelineKey = state['pipelineKey'] as String; switch (pipelineKey) { case PipelineKeys.geography: await _startGeographyPipeline(firebaseUid); case PipelineKeys.weather: await _startWeatherPipeline(firebaseUid); case PipelineKeys.spelling: await _startSpellingPipeline(firebaseUid); default: await _questionDispatcher?.topUpQueue(firebaseUid); } } Future _setIdle(String firebaseUid, String pipelineKey) async { await _questionsDb.upsertPipelineState( assignedUserId: firebaseUid, pipelineKey: pipelineKey, step: PipelineSteps.idle, context: await _loadContext(firebaseUid, pipelineKey), ); } void close() => _fetcher.close(); }