348 lines
11 KiB
Dart
348 lines
11 KiB
Dart
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<GroupRideService>((ref) {
|
|
return GroupRideService();
|
|
});
|
|
|
|
/// Stream of a specific Group Ride by ID.
|
|
final groupRideStreamProvider =
|
|
StreamProvider.family<GroupRide?, String>((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<List<GroupRideParticipant>, String>((ref, rideId) {
|
|
final service = ref.watch(groupRideServiceProvider);
|
|
final rideAsync = ref.watch(groupRideStreamProvider(rideId));
|
|
final currentUser = ref.watch(currentUserProvider);
|
|
|
|
final controller = StreamController<List<GroupRideParticipant>>();
|
|
final participantSubscriptions = <String, StreamSubscription<GroupRideParticipant?>>{};
|
|
final activeParticipants = <String, GroupRideParticipant>{};
|
|
|
|
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<String> 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<List<GroupRideJoinRequest>, 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<List<GroupRide>>((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<List<GroupRide>>((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<List<LiveGroupPresence>, String>((ref, rideId) {
|
|
final service = ref.watch(groupRideServiceProvider);
|
|
final rideAsync = ref.watch(groupRideStreamProvider(rideId));
|
|
|
|
final controller = StreamController<List<LiveGroupPresence>>();
|
|
final activeSubscriptions = <String, StreamSubscription<LiveGroupPresence?>>{};
|
|
final activePresences = <String, LiveGroupPresence>{};
|
|
final retryTimers = <String, Timer>{};
|
|
final participantSubscriptions = <String, StreamSubscription<GroupRideParticipant?>>{};
|
|
final acceptedUids = <String>{};
|
|
|
|
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<String> 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<bool>((ref) {
|
|
return false;
|
|
});
|
|
|
|
/// Active Group Ride ID currently being broadcasted to by the local rider, if any.
|
|
final activeLiveRideSharingIdProvider = StateProvider<String?>((ref) {
|
|
return null;
|
|
});
|
|
|
|
/// Provider that returns the first active group ride for the current user (assembling or underway).
|
|
final activeUserGroupRideProvider = Provider<GroupRide?>((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,
|
|
);
|
|
});
|
|
|