import 'package:flutter/foundation.dart' show debugPrint; import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:supabase_flutter/supabase_flutter.dart'; import 'package:uuid/uuid.dart'; import '../brick/cache_helpers.dart'; import '../models/chat_message.dart'; import '../utils/app_time.dart'; import '../utils/snackbar.dart' show isOfflineSaveError; import 'connectivity_provider.dart'; import 'supabase_provider.dart'; import 'stream_recovery.dart'; import 'realtime_controller.dart'; // --------------------------------------------------------------------------- // Offline-pending state // --------------------------------------------------------------------------- /// Pending chat messages keyed by threadId. /// Messages sync on reconnect when the thread's provider is active. final offlinePendingChatMessagesProvider = StateProvider>>((ref) => const {}); // --------------------------------------------------------------------------- // Read provider // --------------------------------------------------------------------------- /// Real-time chat messages for a swap request thread. final chatMessagesProvider = StreamProvider.family, String>(( ref, threadId, ) { final client = ref.watch(supabaseClientProvider); final pendingByThread = ref.watch(offlinePendingChatMessagesProvider); final pendingForThread = pendingByThread[threadId] ?? const []; List applyPending(List rows) { if (pendingForThread.isEmpty) return rows; final byId = {for (final r in rows) r.id: r}; for (final p in pendingForThread) { byId.putIfAbsent(p.id, () => p); } return byId.values.toList() ..sort((a, b) => a.createdAt.compareTo(b.createdAt)); } final wrapper = StreamRecoveryWrapper( stream: client .from('chat_messages') .stream(primaryKey: ['id']) .eq('thread_id', threadId) .order('created_at'), onPollData: () async { final data = await client .from('chat_messages') .select() .eq('thread_id', threadId) .order('created_at'); return data.map(ChatMessage.fromMap).toList(); }, fromMap: ChatMessage.fromMap, channelName: 'chat_messages_$threadId', onStatusChanged: ref.read(realtimeControllerProvider).handleChannelStatus, onOfflineData: () async { final all = await cachedListFromBrick(); final filtered = all .where((m) => m.threadId == threadId) .toList() ..sort((a, b) => a.createdAt.compareTo(b.createdAt)); return applyPending(filtered); }, onCacheMirror: (rows) => mirrorBatchToBrick(rows, tag: 'chat_messages'), ); ref.onDispose(wrapper.dispose); // Reconnect-replay: fires when this thread's screen is active. ref.listen(isOnlineProvider, (wasOnline, isNowOnline) async { if (wasOnline == true || !isNowOnline) return; final pending = List.from( (ref.read(offlinePendingChatMessagesProvider)[threadId] ?? []), ); if (pending.isEmpty) return; debugPrint( '[chatMessagesProvider($threadId)] reconnected — syncing ${pending.length} message(s)', ); final synced = []; for (final msg in pending) { try { await client.from('chat_messages').insert({ 'id': msg.id, 'thread_id': msg.threadId, 'sender_id': msg.senderId, 'body': msg.body, }); synced.add(msg.id); } catch (e) { final s = e.toString(); if (s.contains('23505') || s.contains('duplicate key')) { synced.add(msg.id); } else { debugPrint( '[chatMessagesProvider] failed to sync message id=${msg.id}: $e', ); } } } if (synced.isNotEmpty) { final current = Map>.from( ref.read(offlinePendingChatMessagesProvider), ); final remaining = (current[threadId] ?? []).where((m) => !synced.contains(m.id)).toList(); if (remaining.isEmpty) { current.remove(threadId); } else { current[threadId] = remaining; } ref.read(offlinePendingChatMessagesProvider.notifier).state = Map.unmodifiable(current); } }); return wrapper.stream.map((result) => applyPending(result.data)); }); // --------------------------------------------------------------------------- // Controller // --------------------------------------------------------------------------- final chatControllerProvider = Provider((ref) { final client = ref.watch(supabaseClientProvider); return ChatController(client, ref); }); class ChatController { ChatController(this._client, [this._ref]); final SupabaseClient _client; final Ref? _ref; /// Send a message to a thread. Offline: queues locally. Future sendMessage({ required String threadId, required String body, }) async { final userId = _client.auth.currentUser?.id; if (userId == null) throw Exception('Not authenticated'); final id = const Uuid().v4(); try { await _client.from('chat_messages').insert({ 'id': id, 'thread_id': threadId, 'sender_id': userId, 'body': body, }); } catch (e) { if (!isOfflineSaveError(e)) rethrow; final ref = _ref; if (ref != null) { final existing = ref.read(offlinePendingChatMessagesProvider)[threadId] ?? []; // Idempotency guard: suppress if the same body was queued within the // last 10 s — prevents duplicate messages from rapid taps while offline. final cutoff = AppTime.now().subtract(const Duration(seconds: 10)); if (existing.any((m) => m.body == body && m.createdAt.isAfter(cutoff))) { debugPrint( '[ChatController] duplicate offline message suppressed for thread=$threadId', ); return; } } final message = ChatMessage( id: id, threadId: threadId, senderId: userId, body: body, createdAt: AppTime.now(), ); final ref2 = _ref; if (ref2 != null) { final current = Map>.from( ref2.read(offlinePendingChatMessagesProvider), ); current[threadId] = [...(current[threadId] ?? []), message]; ref2.read(offlinePendingChatMessagesProvider.notifier).state = Map.unmodifiable(current); } mirrorBatchToBrick([message], tag: 'offline_chat'); debugPrint('[ChatController] sendMessage queued offline: $id'); } } }