cyberhybridhub/lib/services/questions_hub_service.dart
nathanAndrinoid a7c370b743
Some checks failed
Admin portal tests / test (push) Has been cancelled
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 <cursoragent@cursor.com>
2026-06-05 22:04:06 +00:00

553 lines
18 KiB
Dart

import 'dart:async';
import 'package:flutter/foundation.dart';
import 'package:signalr_netcore/signalr_client.dart';
import '../config/api_config.dart';
import '../models/guess_score_reset_result.dart';
import '../models/guess_score_summary.dart';
import '../models/incoming_question.dart';
import '../models/question_defer_result.dart';
import '../models/question_source_preference.dart';
import '../models/question_submit_result.dart';
import '../repositories/question_source_config_repository.dart';
import 'auth_service.dart';
import 'questions_api_service.dart';
/// Maintains a SignalR connection to receive incoming questions from the API.
class QuestionsHubService {
QuestionsHubService._();
static final QuestionsHubService instance = QuestionsHubService._();
final QuestionsApiService _api = QuestionsApiService();
final ValueNotifier<bool> hasPendingQuestion = ValueNotifier<bool>(false);
final ValueNotifier<IncomingQuestion?> pendingQuestion =
ValueNotifier<IncomingQuestion?>(null);
final ValueNotifier<int> pendingQuestionCount = ValueNotifier<int>(0);
final ValueNotifier<bool> questionPanelOpen = ValueNotifier<bool>(false);
final ValueNotifier<List<IncomingQuestion>> questionQueue =
ValueNotifier<List<IncomingQuestion>>(<IncomingQuestion>[]);
final ValueNotifier<bool> questionActionBusy = ValueNotifier<bool>(false);
final ValueNotifier<bool> scoreResetBusy = ValueNotifier<bool>(false);
final ValueNotifier<bool> spellingResetBusy = ValueNotifier<bool>(false);
final ValueNotifier<GuessScoreSummary> guessScoreSummary =
ValueNotifier<GuessScoreSummary>(GuessScoreSummary.empty);
HubConnection? _connection;
bool _connecting = false;
IncomingQuestion? get currentQuestion {
final IncomingQuestion? pending = pendingQuestion.value;
final List<IncomingQuestion> queue = questionQueue.value;
// After an answer, [pendingQuestion] advances but the old card can remain at
// queue[0] until the queue is replaced — prefer the active pending target.
if (pending != null &&
(queue.isEmpty || queue.first.id != pending.id)) {
return pending;
}
if (queue.isNotEmpty) {
return queue.first;
}
return pending;
}
/// Login hook: load persisted score, bootstrap question state, then SignalR.
Future<void> onLogin() async {
await _refreshGuessScoreSummary();
final ({IncomingQuestion? question, GuessScoreSummary? score}) boot =
await _api.bootstrapOnLogin();
if (boot.score != null) {
guessScoreSummary.value = boot.score!;
} else {
await _refreshGuessScoreSummary();
}
if (boot.question != null) {
_applyIncoming(boot.question!);
}
await connect();
}
Future<void> connect() async {
if (_connecting) {
return;
}
if (_connection?.state == HubConnectionState.Connected) {
return;
}
final String? token = await AuthService.instance.getIdToken();
if (token == null) {
return;
}
_connecting = true;
try {
await disconnect();
final HubConnection connection = HubConnectionBuilder()
.withUrl(
questionsHubUrl,
options: HttpConnectionOptions(
accessTokenFactory: () async =>
await AuthService.instance.getIdToken() ?? '',
requestTimeout: 30000,
),
)
.withAutomaticReconnect(
retryDelays: <int>[0, 2000, 5000, 10000],
)
.build();
connection.on('ReceiveQuestion', _onReceiveQuestion);
_connection = connection;
await connection.start();
} catch (e, st) {
debugPrint('Questions hub connect failed: $e\n$st');
await disconnect();
} finally {
_connecting = false;
}
}
void _onReceiveQuestion(List<Object?>? arguments) {
if (arguments == null || arguments.isEmpty) {
return;
}
final Object? raw = arguments.first;
final Map<String, dynamic> json;
if (raw is Map<String, dynamic>) {
json = raw;
} else if (raw is Map) {
json = Map<String, dynamic>.from(raw);
} else {
debugPrint('ReceiveQuestion: unexpected payload type ${raw.runtimeType}');
return;
}
final IncomingQuestion question = IncomingQuestion.fromJson(json);
_applyIncoming(question);
}
void _applyIncoming(IncomingQuestion question) {
if (!_acceptsQuestionSource(question)) {
return;
}
final int count =
question.unansweredCount < 1 ? 1 : question.unansweredCount;
pendingQuestion.value = question;
pendingQuestionCount.value = count;
hasPendingQuestion.value = count >= 1;
if (questionPanelOpen.value) {
final List<IncomingQuestion> queue = List<IncomingQuestion>.from(
questionQueue.value,
);
if (!queue.any((IncomingQuestion q) => q.id == question.id)) {
queue.add(question);
questionQueue.value = queue;
}
}
}
/// Opens the inline question panel and loads the full pending queue.
Future<void> openQuestionPanel() async {
questionActionBusy.value = true;
try {
// Align client filters with server-side source preferences before fetch.
await QuestionSourceConfigRepository.instance.loadFromRemote();
final List<IncomingQuestion> active = await _fetchUnansweredForPanel();
questionPanelOpen.value = true;
if (active.isNotEmpty) {
questionQueue.value = active;
pendingQuestion.value = active.first;
pendingQuestionCount.value = active.length;
hasPendingQuestion.value = true;
} else if (pendingQuestion.value != null &&
_acceptsQuestionSource(pendingQuestion.value!)) {
questionQueue.value = <IncomingQuestion>[pendingQuestion.value!];
pendingQuestionCount.value = 1;
hasPendingQuestion.value = true;
} else {
pendingQuestion.value = null;
questionQueue.value = <IncomingQuestion>[];
hasPendingQuestion.value = _shouldKeepPanelOpenWithoutQuestion();
}
} finally {
questionActionBusy.value = false;
}
}
bool _shouldKeepPanelOpenWithoutQuestion() {
final List<QuestionSourcePreference> sources =
QuestionSourceConfigRepository.instance.current;
return QuestionSourceConfigRepository.isSpellingOnlyEnabled(sources) ||
QuestionSourceConfigRepository.isMarketOnlyEnabled(sources);
}
bool _acceptsQuestionSource(IncomingQuestion question) {
final List<QuestionSourcePreference> sources =
QuestionSourceConfigRepository.instance.current;
if (QuestionSourceConfigRepository.isSpellingOnlyEnabled(sources)) {
return question.isSpelling;
}
if (QuestionSourceConfigRepository.isMarketOnlyEnabled(sources)) {
return question.isMarketHistory;
}
if (QuestionSourceConfigRepository.isSpellingFullyWeighted(sources) &&
question.isMarketHistory) {
return false;
}
if (!QuestionSourceConfigRepository.isMarketHistoryEnabled(sources)) {
return question.sourceId != 'market_history';
}
if (!QuestionSourceConfigRepository.isSpellingEnabled(sources)) {
return !question.isSpelling;
}
return true;
}
List<IncomingQuestion> _filterForEnabledSources(
List<IncomingQuestion> questions,
) {
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() {
questionPanelOpen.value = false;
}
/// Replaces a queue row after tile-load spelling prepare succeeds.
void applyPreparedQuestion(IncomingQuestion prepared) {
final List<IncomingQuestion> queue =
List<IncomingQuestion>.from(questionQueue.value);
final int index = queue.indexWhere((IncomingQuestion q) => q.id == prepared.id);
if (index >= 0) {
queue[index] = prepared;
questionQueue.value = queue;
}
if (pendingQuestion.value?.id == prepared.id) {
pendingQuestion.value = prepared;
}
}
/// Swipe left: move current question to end of queue without answering.
Future<void> deferCurrentQuestion() async {
final IncomingQuestion? question = currentQuestion;
if (question == null || questionActionBusy.value) {
return;
}
final String skippedQuestionId = question.id;
questionActionBusy.value = true;
try {
final QuestionDeferResult? result = await _api.deferQuestion(
questionId: skippedQuestionId,
);
if (result == null) {
return;
}
if (result.unansweredCount == 0) {
_clearPendingUi();
return;
}
final List<IncomingQuestion> refreshed = await _fetchUnansweredForPanel();
if (refreshed.isEmpty) {
if (result.nextQuestion != null &&
_acceptsQuestionSource(result.nextQuestion!)) {
_setActiveQuestion(
result.nextQuestion!,
unansweredCount: result.unansweredCount,
);
} else if (_shouldKeepPanelOpenWithoutQuestion()) {
pendingQuestion.value = null;
questionQueue.value = <IncomingQuestion>[];
hasPendingQuestion.value = true;
questionPanelOpen.value = true;
} else {
_clearPendingUi();
}
return;
}
IncomingQuestion next = refreshed.first;
if (refreshed.length > 1 && next.id == skippedQuestionId) {
next = refreshed.firstWhere(
(IncomingQuestion q) => q.id != skippedQuestionId,
orElse: () => refreshed.first,
);
} else if (result.nextQuestion != null &&
result.nextQuestion!.id != skippedQuestionId &&
_acceptsQuestionSource(result.nextQuestion!)) {
next = result.nextQuestion!;
}
_setActiveQuestion(
next,
unansweredCount: result.unansweredCount,
queue: refreshed,
);
} finally {
questionActionBusy.value = false;
}
}
/// Submit a typed spelling answer.
Future<void> submitCurrentSpellingAnswer(String answerText) async {
final IncomingQuestion? question = currentQuestion;
if (question == null ||
questionActionBusy.value ||
!question.usesSpellingWordTile) {
return;
}
final String trimmed = answerText.trim();
if (trimmed.isEmpty) {
return;
}
await _submitAnswer(
answeredQuestionId: question.id,
question: question,
answerText: trimmed,
);
}
/// Swipe right: submit slider value as the answer.
Future<void> submitCurrentAnswer({num answer = 0}) async {
final IncomingQuestion? question = currentQuestion;
if (question == null || questionActionBusy.value) {
return;
}
await _submitAnswer(
answeredQuestionId: question.id,
question: question,
answer: answer,
);
}
Future<void> _submitAnswer({
required String answeredQuestionId,
required IncomingQuestion question,
num? answer,
String? answerText,
}) async {
questionActionBusy.value = true;
try {
final QuestionSubmitResult? result = await _api.submitAnswer(
questionId: answeredQuestionId,
answer: answer,
answerText: answerText,
);
if (result == null) {
return;
}
if (question.isMarketHistory) {
guessScoreSummary.value = result.score;
}
if (result.nextQuestion != null &&
_acceptsQuestionSource(result.nextQuestion!)) {
_setActiveQuestion(
result.nextQuestion!,
unansweredCount: result.unansweredCount,
);
return;
}
if (result.unansweredCount == 0) {
if (_shouldKeepPanelOpenWithoutQuestion()) {
pendingQuestion.value = null;
questionQueue.value = <IncomingQuestion>[];
hasPendingQuestion.value = true;
questionPanelOpen.value = true;
} else {
_clearPendingUi();
}
return;
}
final List<IncomingQuestion> refreshed = await _fetchUnansweredForPanel();
final List<IncomingQuestion> remaining = refreshed
.where((IncomingQuestion q) => q.id != answeredQuestionId)
.toList();
if (remaining.isEmpty) {
if (_shouldKeepPanelOpenWithoutQuestion()) {
pendingQuestion.value = null;
questionQueue.value = <IncomingQuestion>[];
hasPendingQuestion.value = true;
questionPanelOpen.value = true;
} else {
_clearPendingUi();
}
return;
}
_setActiveQuestion(
remaining.first,
unansweredCount: result.unansweredCount,
queue: remaining,
);
} finally {
questionActionBusy.value = false;
}
}
/// Makes [question] the sole active card (queue head matches [pendingQuestion]).
void _setActiveQuestion(
IncomingQuestion question, {
required int unansweredCount,
List<IncomingQuestion>? queue,
}) {
final int count = unansweredCount < 1 ? 1 : unansweredCount;
pendingQuestion.value = question;
pendingQuestionCount.value = count;
hasPendingQuestion.value = true;
if (questionPanelOpen.value) {
questionQueue.value = queue ?? <IncomingQuestion>[question];
}
}
Future<void> _refreshGuessScoreSummary() async {
final GuessScoreSummary? score = await _api.fetchGuessScoreSummary();
if (score == null) {
return;
}
guessScoreSummary.value = score;
}
/// Clears spelling links, familiarity progress, and spelling questions on the server.
Future<bool> resetSpellingProgress() async {
spellingResetBusy.value = true;
try {
final result = await _api.resetSpellingProgress();
if (result == null) {
return false;
}
final List<IncomingQuestion> refreshed = await _fetchUnansweredForPanel();
final List<IncomingQuestion> active =
refreshed.where(_acceptsQuestionSource).toList(growable: false);
if (active.isNotEmpty) {
pendingQuestion.value = active.first;
pendingQuestionCount.value = active.length;
hasPendingQuestion.value = true;
if (questionPanelOpen.value) {
questionQueue.value = active;
}
} else {
pendingQuestion.value = null;
questionQueue.value = <IncomingQuestion>[];
hasPendingQuestion.value = _shouldKeepPanelOpenWithoutQuestion();
}
return true;
} finally {
spellingResetBusy.value = false;
}
}
/// Clears cumulative guess score and answer statistics on the server.
Future<bool> resetGuessScoreSummary() async {
scoreResetBusy.value = true;
try {
final GuessScoreResetResult? result = await _api.resetGuessScoreSummary();
if (result == null) {
return false;
}
guessScoreSummary.value = result.score;
_clearPendingUi();
if (result.question != null) {
_applyIncoming(result.question!);
}
return true;
} finally {
scoreResetBusy.value = false;
}
}
/// Clears pending question UI; keeps the SignalR connection alive.
void _clearPendingUi() {
hasPendingQuestion.value = false;
pendingQuestion.value = null;
pendingQuestionCount.value = 0;
questionQueue.value = <IncomingQuestion>[];
questionPanelOpen.value = false;
}
void dismissPending() {
_clearPendingUi();
}
Future<void> disconnect() async {
final HubConnection? connection = _connection;
_connection = null;
if (connection != null) {
try {
await connection.stop();
} catch (_) {
// Ignore shutdown errors.
}
}
_clearPendingUi();
}
/// Clears in-memory score (call on sign-out; score remains on server per UID).
void clearGuessScoreCache() {
guessScoreSummary.value = GuessScoreSummary.empty;
}
}