import 'dart:async'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../auth/auth_providers.dart'; import 'group_ride_models.dart'; import 'group_ride_service.dart'; /// Provider for GroupRideService. final groupRideServiceProvider = Provider((ref) { return GroupRideService(); }); /// Stream of a specific Group Ride by ID. final groupRideStreamProvider = StreamProvider.family((ref, rideId) { final service = ref.watch(groupRideServiceProvider); return service.groupRideStream(rideId); }); /// Stream of participants for a specific Group Ride. /// Uses bounded point-document listeners per UID discovered via parent GroupRide.accessUids /// (plus the signed-in user's own UID if available) to ensure zero collection list queries /// and complete peer block isolation. final groupRideParticipantsProvider = StreamProvider.family.autoDispose, String>((ref, rideId) { final service = ref.watch(groupRideServiceProvider); final rideAsync = ref.watch(groupRideStreamProvider(rideId)); final currentUser = ref.watch(currentUserProvider); final controller = StreamController>(); final participantSubscriptions = >{}; final activeParticipants = {}; void emitCurrent() { if (controller.isClosed) return; final list = activeParticipants.values.toList() ..sort((a, b) { if (a.role == GroupRideRole.organizer) return -1; if (b.role == GroupRideRole.organizer) return 1; return a.createdAt.compareTo(b.createdAt); }); controller.add(list); } void reconcileCandidates(List candidateUids) { final candidateSet = candidateUids.toSet(); if (currentUser != null && !currentUser.isAnonymous) { candidateSet.add(currentUser.uid); } final existingUids = participantSubscriptions.keys.toList(); for (final uid in existingUids) { if (!candidateSet.contains(uid)) { participantSubscriptions.remove(uid)?.cancel(); activeParticipants.remove(uid); } } for (final uid in candidateSet) { if (!participantSubscriptions.containsKey(uid)) { final stream = service.participantDocStream(groupRideId: rideId, participantUid: uid); participantSubscriptions[uid] = stream.listen( (participant) { if (participant != null) { activeParticipants[uid] = participant; } else { activeParticipants.remove(uid); } emitCurrent(); }, onError: (_) { activeParticipants.remove(uid); emitCurrent(); }, ); } } emitCurrent(); } rideAsync.when( data: (ride) { if (ride == null) { for (final sub in participantSubscriptions.values) { sub.cancel(); } participantSubscriptions.clear(); activeParticipants.clear(); emitCurrent(); return; } reconcileCandidates(ride.accessUids); }, error: (e, st) { if (!controller.isClosed) controller.addError(e, st); }, loading: () {}, ); ref.onDispose(() { for (final sub in participantSubscriptions.values) { sub.cancel(); } participantSubscriptions.clear(); activeParticipants.clear(); controller.close(); }); return controller.stream; }); /// FutureProvider fetching pending join requests for a GroupRide (organizer or club leadership only). final groupRideJoinRequestsProvider = FutureProvider.family.autoDispose, String>((ref, rideId) async { final user = ref.watch(currentUserProvider); if (user == null || user.isAnonymous) { return const []; } final authService = ref.watch(authServiceProvider); final token = await authService.getIdToken(); final service = ref.watch(groupRideServiceProvider); const definedUrl = String.fromEnvironment('MOTORSIGNAL_API_BASE_URL'); final baseUrl = definedUrl.isNotEmpty ? definedUrl : null; try { return await service.fetchJoinRequests( groupRideId: rideId, idToken: token, baseUrl: baseUrl, ); } catch (e) { return const []; } }); /// Stream of upcoming Group Rides for the current user. final upcomingGroupRidesProvider = StreamProvider>((ref) { final user = ref.watch(currentUserProvider); if (user == null || user.isAnonymous) { return Stream.value([]); } final service = ref.watch(groupRideServiceProvider); return service.upcomingRidesStream(callerUid: user.uid); }); /// Stream of Group Rides organized by the current user. final myGroupRidesProvider = StreamProvider>((ref) { final user = ref.watch(currentUserProvider); if (user == null || user.isAnonymous) { return Stream.value([]); } final service = ref.watch(groupRideServiceProvider); return service.myRidesStream(callerUid: user.uid); }); /// Stream of live presence records for all accepted participants in a specific Group Ride. /// Uses bounded point-document listeners per candidate UID discovered via parent GroupRide.accessUids. /// Isolates peer permission-denied / sharing-off events to the specific peer and recovers cleanly. final groupRideLivePresenceProvider = StreamProvider.family.autoDispose, String>((ref, rideId) { final service = ref.watch(groupRideServiceProvider); final rideAsync = ref.watch(groupRideStreamProvider(rideId)); final controller = StreamController>(); final activeSubscriptions = >{}; final activePresences = {}; final retryTimers = {}; final participantSubscriptions = >{}; final acceptedUids = {}; void emitCurrent() { if (controller.isClosed) return; final now = DateTime.now(); final freshList = activePresences.values .where((p) => p.sharingEnabled && p.updatedAt.isAfter(now.subtract(const Duration(seconds: 120)))) .toList(); controller.add(freshList); } // Periodic sweep for stale presences (> 120s) final ttlTimer = Timer.periodic(const Duration(seconds: 15), (_) => emitCurrent()); void stopPresenceListener(String uid) { retryTimers.remove(uid)?.cancel(); activeSubscriptions.remove(uid)?.cancel(); activePresences.remove(uid); emitCurrent(); } void startPresenceListener(String uid) { if (activeSubscriptions.containsKey(uid) || !acceptedUids.contains(uid)) return; retryTimers.remove(uid)?.cancel(); final stream = service.livePresenceDocStream(groupRideId: rideId, targetUid: uid); activeSubscriptions[uid] = stream.listen( (presence) { if (presence != null && presence.sharingEnabled) { activePresences[uid] = presence; } else { activePresences.remove(uid); } emitCurrent(); }, onError: (e) { // Peer error (sharing disabled, mutual block, or permission-denied) activePresences.remove(uid); emitCurrent(); activeSubscriptions.remove(uid)?.cancel(); // Bounded retry to recover when sharing is toggled back ON if (!controller.isClosed && acceptedUids.contains(uid)) { retryTimers[uid] = Timer(const Duration(seconds: 10), () { if (!controller.isClosed && acceptedUids.contains(uid)) { startPresenceListener(uid); } }); } }, ); } void reconcileCandidates(List candidateUids) { final candidateSet = candidateUids.toSet(); // 1. Remove participant listeners and presence listeners for candidates no longer present final existingUids = participantSubscriptions.keys.toList(); for (final uid in existingUids) { if (!candidateSet.contains(uid)) { participantSubscriptions.remove(uid)?.cancel(); acceptedUids.remove(uid); stopPresenceListener(uid); } } // 2. Add participant listener for new candidate UIDs for (final uid in candidateSet) { if (!participantSubscriptions.containsKey(uid)) { final pStream = service.participantDocStream(groupRideId: rideId, participantUid: uid); participantSubscriptions[uid] = pStream.listen( (participant) { final wasAccepted = acceptedUids.contains(uid); final isNowAccepted = participant?.status == ParticipantStatus.accepted; if (isNowAccepted) { acceptedUids.add(uid); if (!wasAccepted || !activeSubscriptions.containsKey(uid)) { startPresenceListener(uid); } } else { acceptedUids.remove(uid); stopPresenceListener(uid); } }, onError: (err) { // If participant document itself is denied (e.g. mutual block on roster), exclude UID acceptedUids.remove(uid); stopPresenceListener(uid); }, ); } } } rideAsync.when( data: (ride) { if (ride == null || ride.isCompleted || ride.isCanceled) { for (final sub in participantSubscriptions.values) { sub.cancel(); } participantSubscriptions.clear(); for (final sub in activeSubscriptions.values) { sub.cancel(); } activeSubscriptions.clear(); for (final timer in retryTimers.values) { timer.cancel(); } retryTimers.clear(); acceptedUids.clear(); activePresences.clear(); emitCurrent(); return; } reconcileCandidates(ride.accessUids); }, error: (e, st) { if (!controller.isClosed) controller.addError(e, st); }, loading: () {}, ); ref.onDispose(() { ttlTimer.cancel(); for (final timer in retryTimers.values) { timer.cancel(); } retryTimers.clear(); for (final sub in participantSubscriptions.values) { sub.cancel(); } participantSubscriptions.clear(); for (final sub in activeSubscriptions.values) { sub.cancel(); } activeSubscriptions.clear(); controller.close(); }); return controller.stream; }); /// Master privacy setting: whether group ride live location sharing is permitted by user. /// Defaults to false (explicit opt-in required). final groupRideLiveSharingAllowedProvider = StateProvider((ref) { return false; }); /// Active Group Ride ID currently being broadcasted to by the local rider, if any. final activeLiveRideSharingIdProvider = StateProvider((ref) { return null; }); /// Provider that returns the first active group ride for the current user (assembling or underway). final activeUserGroupRideProvider = Provider((ref) { final upcomingAsync = ref.watch(upcomingGroupRidesProvider); final activeSharingId = ref.watch(activeLiveRideSharingIdProvider); return upcomingAsync.when( data: (rides) { if (activeSharingId != null) { try { return rides.firstWhere((r) => r.id == activeSharingId); } catch (_) {} } try { return rides.firstWhere((r) => r.status == GroupRideStatus.underway || r.status == GroupRideStatus.assembling); } catch (_) { return null; } }, loading: () => null, error: (_, _) => null, ); });