MotorSignal-App/lib/services/group_rides/live_broadcast_controller.dart

557 lines
17 KiB
Dart

import 'dart:async';
import 'package:flutter/widgets.dart';
import 'package:flutter_riverpod/flutter_riverpod.dart';
import 'package:geolocator/geolocator.dart';
import '../auth/auth_providers.dart';
import '../location_service.dart';
import 'group_ride_models.dart';
import 'group_ride_providers.dart';
import 'group_ride_service.dart';
/// State of the local rider's live presence broadcast for a specific Group Ride.
@immutable
class LiveBroadcastState {
final bool isOptedIn;
final bool isMasterAllowed;
final bool isEligible;
final bool isBroadcasting;
final bool isPaused;
final int retryCount;
final bool isBackground;
final LivePresenceStatus status;
final Position? lastPosition;
final DateTime? lastPublishedAt;
final String? errorMessage;
final bool isRegroupActive;
const LiveBroadcastState({
this.isOptedIn = false,
this.isMasterAllowed = false,
this.isEligible = false,
this.isBroadcasting = false,
this.isPaused = false,
this.retryCount = 0,
this.isBackground = false,
this.status = LivePresenceStatus.riding,
this.lastPosition,
this.lastPublishedAt,
this.errorMessage,
this.isRegroupActive = false,
});
LiveBroadcastState copyWith({
bool? isOptedIn,
bool? isMasterAllowed,
bool? isEligible,
bool? isBroadcasting,
bool? isPaused,
int? retryCount,
bool? isBackground,
LivePresenceStatus? status,
Position? lastPosition,
DateTime? lastPublishedAt,
String? errorMessage,
bool? isRegroupActive,
}) {
return LiveBroadcastState(
isOptedIn: isOptedIn ?? this.isOptedIn,
isMasterAllowed: isMasterAllowed ?? this.isMasterAllowed,
isEligible: isEligible ?? this.isEligible,
isBroadcasting: isBroadcasting ?? this.isBroadcasting,
isPaused: isPaused ?? this.isPaused,
retryCount: retryCount ?? this.retryCount,
isBackground: isBackground ?? this.isBackground,
status: status ?? this.status,
lastPosition: lastPosition ?? this.lastPosition,
lastPublishedAt: lastPublishedAt ?? this.lastPublishedAt,
errorMessage: errorMessage,
isRegroupActive: isRegroupActive ?? this.isRegroupActive,
);
}
}
/// Controller responsible for ephemeral live location broadcasts in an active Group Ride.
///
/// Implements dual-gate privacy (global + per-ride), 5s moving / 30s stopped heartbeat,
/// 30m accuracy filtering, and immediate termination on authorization loss.
class LiveBroadcastController extends StateNotifier<LiveBroadcastState> {
static final Set<String> _persistedOptedInRideIds = <String>{};
final String rideId;
final String callerUid;
final GroupRideService service;
final Stream<Position>? positionStream;
final Stream<GroupRide?>? rideStream;
final Stream<List<GroupRideParticipant>>? participantsStream;
final Stream<List<LiveGroupPresence>>? presenceStream;
StreamSubscription<Position>? _positionSub;
StreamSubscription<GroupRide?>? _rideSub;
StreamSubscription<List<GroupRideParticipant>>? _participantsSub;
StreamSubscription<List<LiveGroupPresence>>? _presenceSub;
Timer? _heartbeatTimer;
DateTime? _nextRetryTime;
int _consecutiveErrors = 0;
Position? _lastBroadcastPosition;
DateTime? _lastBroadcastTime;
bool _isDisposed = false;
final DateTime Function() _now;
LiveBroadcastController({
required this.rideId,
required this.callerUid,
required this.service,
this.positionStream,
this.rideStream,
this.participantsStream,
this.presenceStream,
bool initialMasterAllowed = false,
bool? initialOptIn,
DateTime Function()? now,
}) : _now = now ?? DateTime.now,
super(LiveBroadcastState(
isMasterAllowed: initialMasterAllowed,
isOptedIn: initialOptIn ?? _persistedOptedInRideIds.contains(rideId),
)) {
_initSubscriptions();
}
void _initSubscriptions() {
// 1. Listen to Ride Lifecycle
final activeRideStream = rideStream ?? service.groupRideStream(rideId);
_rideSub = activeRideStream.listen(_handleRideUpdate, onError: (e) {
debugPrint('[LiveBroadcastController] Ride stream error: $e');
});
// 2. Listen to Participants (authorization check)
// Uses caller point-listener instead of broad collection query to adhere to allow list: if false security rule
final activeParticipantsStream = participantsStream ??
(callerUid.isNotEmpty
? service
.participantDocStream(
groupRideId: rideId,
participantUid: callerUid,
)
.map((p) => p != null ? [p] : <GroupRideParticipant>[])
: const Stream<List<GroupRideParticipant>>.empty());
_participantsSub =
activeParticipantsStream.listen(_handleParticipantsUpdate, onError: (e) {
debugPrint('[LiveBroadcastController] Participants stream error: $e');
});
// 3. Listen to Live Presence (for regroup awareness)
final activePresenceStream =
presenceStream ?? const Stream<List<LiveGroupPresence>>.empty();
_presenceSub = activePresenceStream.listen(_handlePresenceUpdate, onError: (e) {
debugPrint('[LiveBroadcastController] Presence stream error: $e');
});
}
void _handleRideUpdate(GroupRide? ride) {
if (_isDisposed) return;
if (ride == null) {
_revokeBroadcasting('Ride not found');
return;
}
// Ride must be live-eligible (scheduled, assembling, or underway)
final isRideActive = ride.isLiveEligible;
if (!isRideActive) {
// Completed or canceled -> stop broadcast
_revokeBroadcasting('Ride is not in active state (${ride.status.name})');
return;
}
_evaluateEligibility();
}
void _handleParticipantsUpdate(List<GroupRideParticipant> participants) {
if (_isDisposed) return;
final myParticipant = participants.where((p) => p.userId == callerUid).toList();
if (myParticipant.isEmpty ||
myParticipant.first.status != ParticipantStatus.accepted) {
// User removed, declined, or left -> immediate revocation
_revokeBroadcasting('Caller is not an accepted participant');
return;
}
_evaluateEligibility(hasAcceptedMembership: true);
}
void _handlePresenceUpdate(List<LiveGroupPresence> presences) {
if (_isDisposed) return;
final regroupActive = presences.any((p) =>
p.status == LivePresenceStatus.regrouping && p.isFresh);
if (regroupActive != state.isRegroupActive) {
state = state.copyWith(isRegroupActive: regroupActive);
}
}
void _evaluateEligibility({bool? hasAcceptedMembership}) {
final eligible = hasAcceptedMembership ?? state.isEligible;
final broadcasting = state.isOptedIn && state.isMasterAllowed && eligible;
state = state.copyWith(
isEligible: eligible,
isBroadcasting: broadcasting,
);
if (broadcasting && _positionSub == null) {
_startBroadcasting();
} else if (!broadcasting && _positionSub != null) {
_stopBroadcasting();
}
}
/// Updates the master privacy setting (global app setting).
void updateMasterAllowed(bool allowed) {
if (_isDisposed || state.isMasterAllowed == allowed) return;
state = state.copyWith(isMasterAllowed: allowed);
_evaluateEligibility();
if (!allowed) {
service.clearLivePresence(
callerUid: callerUid,
groupRideId: rideId,
);
}
}
/// Toggles per-ride opt-in for location broadcasting.
Future<void> setOptIn(bool optIn) async {
if (_isDisposed) return;
if (state.isOptedIn == optIn) return;
_consecutiveErrors = 0;
_nextRetryTime = null;
state = state.copyWith(
isOptedIn: optIn,
isPaused: false,
retryCount: 0,
);
_evaluateEligibility();
if (!optIn) {
_persistedOptedInRideIds.remove(rideId);
// Immediately clear presence document in Firestore
await service.clearLivePresence(
callerUid: callerUid,
groupRideId: rideId,
);
} else {
_persistedOptedInRideIds.add(rideId);
if (state.isBroadcasting) {
if (state.lastPosition != null) {
// Force immediate publish on opt-in
await _publishPosition(state.lastPosition!, force: true);
} else {
await refreshAndPublishCurrentLocation();
}
}
}
}
/// Forces an immediate location sample and presence publish if broadcasting.
Future<void> refreshAndPublishCurrentLocation() async {
if (_isDisposed) return;
try {
Position? pos;
try {
pos = await Geolocator.getLastKnownPosition();
pos ??= await Geolocator.getCurrentPosition(
locationSettings: const LocationSettings(
accuracy: LocationAccuracy.high,
timeLimit: Duration(seconds: 4),
),
);
} catch (_) {
// Fallback or ignore in headless test environments
}
if (pos != null && !_isDisposed) {
_onPositionUpdate(pos);
if (state.isBroadcasting) {
await _publishPosition(pos, force: true);
}
}
} catch (e) {
debugPrint('[LiveBroadcastController] refreshAndPublishCurrentLocation error: $e');
}
}
/// Updates rider quick status (e.g. riding, needFuel, regrouping, stopped, arrived).
Future<void> setStatus(LivePresenceStatus newStatus) async {
if (_isDisposed || state.status == newStatus) return;
state = state.copyWith(status: newStatus);
if (state.isBroadcasting && state.lastPosition != null) {
await _publishPosition(state.lastPosition!, force: true);
}
}
/// Calls for a group regroup.
Future<void> triggerRegroup() async {
await setStatus(LivePresenceStatus.regrouping);
}
/// Clears regroup call by switching back to riding.
Future<void> clearRegroup() async {
await setStatus(LivePresenceStatus.riding);
}
/// Updates app lifecycle state (e.g. background vs foreground).
void updateAppLifecycle(AppLifecycleState lifecycle) {
if (_isDisposed) return;
final isBg = lifecycle == AppLifecycleState.paused ||
lifecycle == AppLifecycleState.hidden;
if (state.isBackground != isBg) {
state = state.copyWith(isBackground: isBg);
}
}
void _startBroadcasting() {
_positionSub?.cancel();
final stream = positionStream ?? Geolocator.getPositionStream(
locationSettings: const LocationSettings(
accuracy: LocationAccuracy.high,
distanceFilter: 5,
),
);
_positionSub = stream.listen(
_onPositionUpdate,
onError: (e) {
state = state.copyWith(errorMessage: 'Location stream error: $e');
},
);
_heartbeatTimer?.cancel();
_heartbeatTimer = Timer.periodic(const Duration(seconds: 10), (_) {
_checkHeartbeat();
});
}
void _stopBroadcasting() {
_positionSub?.cancel();
_positionSub = null;
_heartbeatTimer?.cancel();
_heartbeatTimer = null;
}
void _onPositionUpdate(Position position) {
if (_isDisposed || !state.isBroadcasting) return;
// Reject invalid coordinates, out of bounds, or Null Island (0,0)
if (position.latitude.isNaN ||
position.longitude.isNaN ||
position.latitude < -90.0 ||
position.latitude > 90.0 ||
position.longitude < -180.0 ||
position.longitude > 180.0 ||
(position.latitude == 0.0 && position.longitude == 0.0)) {
return;
}
// Filter accuracy: reject fixes with accuracy > 30m
if (position.accuracy > 30.0) {
return;
}
state = state.copyWith(lastPosition: position);
final now = _now();
// Respect exponential backoff throttle during transient errors
if (_nextRetryTime != null && now.isBefore(_nextRetryTime!)) {
return;
}
final speedMph = position.speed * 2.23694;
final isMoving = speedMph > 3.0;
final lastTime = _lastBroadcastTime;
if (lastTime == null) {
_publishPosition(position, force: true);
return;
}
final elapsedSeconds = now.difference(lastTime).inSeconds;
if (isMoving) {
// Moving cadence: 5 seconds with displacement > 15m or >= 10 seconds interval
if (elapsedSeconds >= 5) {
double displacement = 0;
if (_lastBroadcastPosition != null) {
displacement = Geolocator.distanceBetween(
_lastBroadcastPosition!.latitude,
_lastBroadcastPosition!.longitude,
position.latitude,
position.longitude,
);
}
if (displacement >= 15.0 || elapsedSeconds >= 10) {
_publishPosition(position);
}
}
} else {
// Stopped cadence: 30-second heartbeat
if (elapsedSeconds >= 30) {
_publishPosition(position);
}
}
}
void _checkHeartbeat() {
if (_isDisposed || !state.isBroadcasting || state.lastPosition == null) return;
final lastTime = _lastBroadcastTime;
if (lastTime == null) return;
final now = _now();
if (_nextRetryTime != null && now.isBefore(_nextRetryTime!)) {
return;
}
final elapsed = now.difference(lastTime).inSeconds;
// If stopped or no fix received, keep heartbeat within 30s
if (elapsed >= 30) {
_publishPosition(state.lastPosition!, force: true);
}
}
Future<void> _publishPosition(Position position, {bool force = false}) async {
if (!state.isBroadcasting && !force) return;
final speedMph = position.speed * 2.23694;
final now = _now();
try {
await service.publishLivePresence(
callerUid: callerUid,
groupRideId: rideId,
latitude: position.latitude,
longitude: position.longitude,
heading: position.heading,
speedMph: speedMph,
accuracyMeters: position.accuracy,
status: state.status,
sharingEnabled: true,
);
_lastBroadcastPosition = position;
_lastBroadcastTime = now;
_consecutiveErrors = 0;
_nextRetryTime = null;
state = state.copyWith(
lastPublishedAt: now,
errorMessage: null,
isPaused: false,
retryCount: 0,
);
} catch (e) {
_handlePublishError(e);
}
}
void _handlePublishError(Object error) {
if (_isDisposed) return;
final errStr = error.toString().toLowerCase();
final isPermissionDenied = errStr.contains('permission-denied') ||
errStr.contains('permission_denied') ||
errStr.contains('unauthenticated');
if (isPermissionDenied) {
// Terminal: user removed, blocked, ride completed, or rules denied write.
// Immediately stop broadcasting, cancel timers, and do not retry.
_revokeBroadcasting('Authorization lost: permission denied');
return;
}
// Transient network/offline error:
_consecutiveErrors++;
// Exponential backoff capped at 30 seconds
final backoffSeconds = (5 * (1 << (_consecutiveErrors - 1))).clamp(5, 30);
_nextRetryTime = _now().add(Duration(seconds: backoffSeconds));
state = state.copyWith(
errorMessage: 'Publish failed: $error',
isPaused: true,
retryCount: _consecutiveErrors,
);
}
void _revokeBroadcasting(String reason) {
_persistedOptedInRideIds.remove(rideId);
state = state.copyWith(
isEligible: false,
isBroadcasting: false,
isOptedIn: false,
isPaused: false,
retryCount: 0,
errorMessage: reason,
);
_stopBroadcasting();
// Clean up presence document
service.clearLivePresence(
callerUid: callerUid,
groupRideId: rideId,
);
}
@override
void dispose() {
_isDisposed = true;
_stopBroadcasting();
_rideSub?.cancel();
_participantsSub?.cancel();
_presenceSub?.cancel();
// Clean up ephemeral presence on exit
service.clearLivePresence(
callerUid: callerUid,
groupRideId: rideId,
);
super.dispose();
}
}
/// Provider for LiveBroadcastController family parameterized by group ride ID.
final liveBroadcastControllerProvider = StateNotifierProvider
.family<LiveBroadcastController, LiveBroadcastState, String>(
(ref, rideId) {
final user = ref.watch(currentUserProvider);
final service = ref.watch(groupRideServiceProvider);
final masterAllowed = ref.watch(groupRideLiveSharingAllowedProvider);
final locationService = ref.watch(locationServiceProvider);
final callerUid = user?.uid ?? '';
final controller = LiveBroadcastController(
rideId: rideId,
callerUid: callerUid,
service: service,
participantsStream: callerUid.isNotEmpty
? service
.participantDocStream(
groupRideId: rideId,
participantUid: callerUid,
)
.map((p) => p != null ? [p] : <GroupRideParticipant>[])
: null,
positionStream: locationService.getFilteredPositionStream(),
initialMasterAllowed: masterAllowed,
);
// Sync master privacy changes to controller
ref.listen<bool>(groupRideLiveSharingAllowedProvider, (prev, next) {
controller.updateMasterAllowed(next);
});
return controller;
});