import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../data/models/conversation.dart'; import '../data/models/message.dart'; import '../data/repositories/chat_repository.dart'; import 'api_client_provider.dart'; // Separate NotifierProvider.family so UI re-builds immediately when streaming starts/stops. final isStreamingProvider = NotifierProvider.family<_IsStreamingNotifier, bool, int>( (convId) => _IsStreamingNotifier(convId), ); class _IsStreamingNotifier extends Notifier { // ignore: avoid_unused_constructor_parameters _IsStreamingNotifier(int convId); @override bool build() => false; } // Tracks the current tool status text during generation (empty = no status). final streamingStatusProvider = NotifierProvider.family<_StreamingStatusNotifier, String, int>( (convId) => _StreamingStatusNotifier(convId), ); class _StreamingStatusNotifier extends Notifier { // ignore: avoid_unused_constructor_parameters _StreamingStatusNotifier(int convId); @override String build() => ''; } final conversationsProvider = AsyncNotifierProvider>( ConversationsNotifier.new); class ConversationsNotifier extends AsyncNotifier> { @override Future> build() async { return ref.watch(chatRepositoryProvider).getConversations(); } Future create(String title) async { final conv = await ref.read(chatRepositoryProvider).createConversation(title); state = AsyncData([conv, ...state.value ?? []]); return conv; } Future delete(int id) async { await ref.read(chatRepositoryProvider).deleteConversation(id); state = AsyncData([ for (final c in state.value ?? []) if (c.id != id) c, ]); } /// Re-fetch conversations without clearing the current list (no flicker). Future refresh() async { final fresh = await ref.read(chatRepositoryProvider).getConversations(); state = AsyncData(fresh); } // Called after a message is sent to patch the server-generated title // in-place without triggering a full reload or loading state. void patchConversation(Conversation updated) { final list = state.value; if (list == null) return; state = AsyncData([ for (final c in list) if (c.id == updated.id) updated else c, ]); } } final messagesProvider = AsyncNotifierProvider.family, int>( (convId) => MessagesNotifier(convId), ); class MessagesNotifier extends AsyncNotifier> { final int _convId; MessagesNotifier(this._convId); @override Future> build() async { final (_, messages) = await ref.watch(chatRepositoryProvider).getMessages(_convId); return messages; } /// Re-fetch messages without clearing the current list (no flicker). Future refresh() async { final (_, messages) = await ref.read(chatRepositoryProvider).getMessages(_convId); state = AsyncData(messages); } Future sendMessage(String content) async { final convId = _convId; final repo = ref.read(chatRepositoryProvider); final previousMessages = state.value ?? []; // Optimistic UI: show the user message + assistant placeholder immediately. final userMsg = Message( conversationId: convId, role: MessageRole.user, content: content, ); final placeholder = Message( conversationId: convId, role: MessageRole.assistant, content: '', status: 'generating', ); state = AsyncData([...previousMessages, userMsg, placeholder]); ref.read(isStreamingProvider(convId).notifier).state = true; try { // ── Step 1: POST the message. If this fails the server never got it. ── await repo.sendMessage(convId, content); } catch (e) { state = AsyncData(previousMessages); ref.read(isStreamingProvider(convId).notifier).state = false; rethrow; } // ── Step 2: Stream the response (best effort — silent on failure). ── bool streamedContent = false; try { await for (final event in repo.streamGeneration(convId)) { if (event is ChatTextChunk) { streamedContent = true; ref.read(streamingStatusProvider(convId).notifier).state = ''; final msgs = state.requireValue; if (msgs.isEmpty) continue; final updated = msgs.last.copyWith(content: msgs.last.content + event.text); state = AsyncData([...msgs.sublist(0, msgs.length - 1), updated]); } else if (event is ChatStatusUpdate) { ref.read(streamingStatusProvider(convId).notifier).state = event.status; } } } catch (_) { // SSE failed — fall through to the polling reload below. } // ── Step 3: Poll the API until we have a completed assistant response. // // We always retry up to 20 times regardless of whether SSE streamed // content. SSE ending does NOT mean the server has finished writing to // the DB — there can be a brief gap between the stream closing and the // row being marked complete. // // To avoid clobbering SSE-streamed text with a stale empty DB row, we // only replace state when the polled data is complete OR has content. try { for (var attempt = 0; attempt < 20; attempt++) { if (attempt > 0) await Future.delayed(const Duration(seconds: 2)); final (conv, fresh) = await repo.getMessages(convId); ref.read(conversationsProvider.notifier).patchConversation(conv); final done = fresh.any( (m) => m.role == MessageRole.assistant && m.status != 'generating', ); final polledHasContent = fresh.any( (m) => m.role == MessageRole.assistant && m.content.isNotEmpty, ); // Update UI state when: // • SSE never delivered anything (polling is the only data source), or // • the polled response is complete or has content to show. if (!streamedContent || done || polledHasContent) { state = AsyncData(fresh); } if (done) break; } } catch (_) { // Polling failed entirely — clear the generating placeholder so the UI // doesn't spin forever. final msgs = state.value; if (msgs != null && msgs.isNotEmpty && msgs.last.status == 'generating') { state = AsyncData([ ...msgs.sublist(0, msgs.length - 1), msgs.last.copyWith(status: 'complete'), ]); } } finally { ref.read(isStreamingProvider(convId).notifier).state = false; ref.read(streamingStatusProvider(convId).notifier).state = ''; } } }