diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 83018bf1b..c31af7ff4 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/packages/stream_chat/CHANGELOG.md @@ -24,6 +24,7 @@ - Fixed a `StateError` (`Cannot add new events after calling close`) thrown when the client is disposed while a reconnect recovery is still in flight. - Fixed `Message.deleteMyReaction` dropping an entire reaction group when its summed scores reached zero even though other users' reactions kept the count positive; the group is now retained as long as its count stays above zero. - Fixed reaction groups synthesized from legacy `reaction_counts`/`reaction_scores` payloads being discarded at parse time when their score total was zero or negative despite a positive count. +- Fixed live location expiry emitting repeated `location.expired` events for the same expired location. ## 10.2.0 diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart index 6bacb60a9..78ab80cda 100644 --- a/packages/stream_chat/lib/src/client/channel.dart +++ b/packages/stream_chat/lib/src/client/channel.dart @@ -5,6 +5,7 @@ import 'dart:math' as math; import 'package:collection/collection.dart'; import 'package:rxdart/rxdart.dart'; +import 'package:stream_chat/src/client/live_location_expiration_scheduler.dart'; import 'package:stream_chat/src/client/retry_queue.dart'; import 'package:stream_chat/src/core/util/utils.dart'; import 'package:stream_chat/stream_chat.dart'; @@ -2553,7 +2554,7 @@ class ChannelClientState { _startCleaningStalePinnedMessages(); - _startCleaningExpiredLocations(); + _startSchedulingLocationExpiration(); _listenChannelPushPreferenceUpdated(); @@ -4045,40 +4046,38 @@ class ChannelClientState { ); } - Timer? _staleLiveLocationsCleanerTimer; - void _startCleaningExpiredLocations() { - _staleLiveLocationsCleanerTimer?.cancel(); - _staleLiveLocationsCleanerTimer = Timer.periodic( - const Duration(seconds: 1), - (_) { - final currentUserId = _channel._client.state.currentUser?.id; - if (currentUserId == null) return; - - final expired = activeLiveLocations.where((it) => it.isExpired); - if (expired.isEmpty) return; - - for (final sharedLocation in expired) { - // Skip if the location is shared by the current user, - // as we are already handling them in the client. - if (sharedLocation.userId == currentUserId) continue; - - final lastUpdatedAt = DateTime.timestamp(); - final locationExpiredEvent = Event( - type: EventType.locationExpired, - cid: sharedLocation.channelCid, - message: Message( - id: sharedLocation.messageId, - updatedAt: lastUpdatedAt, - sharedLocation: sharedLocation.copyWith( - updatedAt: lastUpdatedAt, - ), - ), - ); + late final _locationExpirationScheduler = LiveLocationExpirationScheduler( + onExpired: _handleLocationExpired, + ); - _channel._client.handleEvent(locationExpiredEvent); - } - }, + // Listens for changes to the active live locations and (re)schedules the + // one-shot expiry timers accordingly. + void _startSchedulingLocationExpiration() { + _subscriptions.add( + activeLiveLocationsStream.listen(_locationExpirationScheduler.schedule), + ); + } + + // Emits a synthetic `location.expired` event for the expired [location]. + void _handleLocationExpired(Location location) { + // The current user's own live locations are handled by the client-level + // scheduler in [ClientState]; skip them here. + final currentUserId = _channel._client.state.currentUser?.id; + if (currentUserId == null || location.userId == currentUserId) return; + + final lastUpdatedAt = DateTime.timestamp(); + + final locationExpiredEvent = Event( + type: EventType.locationExpired, + cid: location.channelCid, + message: Message( + id: location.messageId, + updatedAt: lastUpdatedAt, + sharedLocation: location.copyWith(updatedAt: lastUpdatedAt), + ), ); + + _channel._client.handleEvent(locationExpiredEvent); } // Listens to channel push preference update events and updates the state @@ -4578,7 +4577,7 @@ class ChannelClientState { _threadsController.close(); _staleTypingEventsCleanerTimer?.cancel(); _stalePinnedMessagesCleanerTimer?.cancel(); - _staleLiveLocationsCleanerTimer?.cancel(); + _locationExpirationScheduler.cancel(); _typingEventsController.close(); } } diff --git a/packages/stream_chat/lib/src/client/client.dart b/packages/stream_chat/lib/src/client/client.dart index d56a37c21..1b80793b9 100644 --- a/packages/stream_chat/lib/src/client/client.dart +++ b/packages/stream_chat/lib/src/client/client.dart @@ -8,6 +8,7 @@ import 'package:rxdart/rxdart.dart'; import 'package:stream_chat/src/client/channel.dart'; import 'package:stream_chat/src/client/channel_delivery_reporter.dart'; import 'package:stream_chat/src/client/event_resolvers.dart' as event_resolvers; +import 'package:stream_chat/src/client/live_location_expiration_scheduler.dart'; import 'package:stream_chat/src/client/query_channels_result.dart'; import 'package:stream_chat/src/client/retry_policy.dart'; import 'package:stream_chat/src/core/api/attachment_file_uploader.dart'; @@ -2625,8 +2626,6 @@ class ClientState { _listenLocationUpdated(); _listenLocationExpired(); // endregion - - _startCleaningExpiredLocations(); } /// Stops listening to the client events. @@ -2814,34 +2813,25 @@ class ClientState { ); } - Timer? _staleLiveLocationsCleanerTimer; - void _startCleaningExpiredLocations() { - _staleLiveLocationsCleanerTimer?.cancel(); - _staleLiveLocationsCleanerTimer = Timer.periodic( - const Duration(seconds: 1), - (_) { - final expired = activeLiveLocations.where((it) => it.isExpired); - if (expired.isEmpty) return; - - for (final sharedLocation in expired) { - final lastUpdatedAt = DateTime.timestamp(); - - final locationExpiredEvent = Event( - type: EventType.locationExpired, - cid: sharedLocation.channelCid, - message: Message( - id: sharedLocation.messageId, - updatedAt: lastUpdatedAt, - sharedLocation: sharedLocation.copyWith( - updatedAt: lastUpdatedAt, - ), - ), - ); + late final _locationExpirationScheduler = LiveLocationExpirationScheduler( + onExpired: _handleLocationExpired, + ); - _client.handleEvent(locationExpiredEvent); - } - }, + // Emits a synthetic `location.expired` event for the expired [location]. + void _handleLocationExpired(Location location) { + final lastUpdatedAt = DateTime.timestamp(); + + final locationExpiredEvent = Event( + type: EventType.locationExpired, + cid: location.channelCid, + message: Message( + id: location.messageId, + updatedAt: lastUpdatedAt, + sharedLocation: location.copyWith(updatedAt: lastUpdatedAt), + ), ); + + _client.handleEvent(locationExpiredEvent); } final StreamChatClient _client; @@ -2891,8 +2881,11 @@ class ClientState { @internal set activeLiveLocations(List locations) { // For safe-keeping, we filter out any inactive locations before update. - final activeLocations = locations.where((it) => it.isActive); - _activeLiveLocationsController.safeAdd(activeLocations.toList()); + final activeLocations = locations.where((it) => it.isActive).toList(); + _activeLiveLocationsController.safeAdd(activeLocations); + + // Reschedule the expiry timers for the updated set of active locations. + _locationExpirationScheduler.schedule(activeLocations); } /// The current unread channels count @@ -2980,7 +2973,7 @@ class ClientState { _unreadThreadsController.close(); _totalUnreadCountController.close(); _activeLiveLocationsController.close(); - _staleLiveLocationsCleanerTimer?.cancel(); + _locationExpirationScheduler.cancel(); _channelsController.close(); for (final channel in channels.values) { diff --git a/packages/stream_chat/lib/src/client/live_location_expiration_scheduler.dart b/packages/stream_chat/lib/src/client/live_location_expiration_scheduler.dart new file mode 100644 index 000000000..521673bbb --- /dev/null +++ b/packages/stream_chat/lib/src/client/live_location_expiration_scheduler.dart @@ -0,0 +1,120 @@ +import 'dart:async'; + +import 'package:stream_chat/src/core/models/location.dart'; + +/// {@template liveLocationExpirationScheduler} +/// Schedules a one-shot [Timer] per live [Location] that fires once at the +/// location's `endAt`, invoking [onExpired] for that location. +/// +/// Timers are keyed by message id and only (re)scheduled when a location's +/// `endAt` changes, so coordinate-only updates leave the existing timers +/// untouched. +/// {@endtemplate} +class LiveLocationExpirationScheduler { + /// {@macro liveLocationExpirationScheduler} + LiveLocationExpirationScheduler({required this.onExpired}); + + /// Called once when a scheduled live location reaches its `endAt`. + final void Function(Location location) onExpired; + + // The currently running expiry timers, keyed by message id. + final _scheduledTimers = {}; + + /// Reconciles the running expiry timers with the given live [locations]. + /// + /// Timers for locations that are no longer present are cancelled; the rest + /// are reused when their `endAt` is unchanged (refreshing the coordinates) or + /// (re)armed when new or when their `endAt` changed. + void schedule(Iterable locations) { + // The live locations that should have an expiry timer, keyed by message id. + final locationsToSchedule = { + for (final location in locations) + if (_shouldSchedule(location)) location.messageId!: location, + }; + + _cancelTimersForRemovedLocations(locationsToSchedule); + for (final entry in locationsToSchedule.entries) { + _reuseOrRescheduleTimer(entry.key, entry.value); + } + } + + /// Cancels all scheduled expiry timers. + void cancel() { + for (final scheduled in _scheduledTimers.values) { + scheduled.timer.cancel(); + } + _scheduledTimers.clear(); + } + + // Whether the [location] is a live location that should get an expiry timer. + bool _shouldSchedule(Location location) { + return location.messageId != null && location.endAt != null && !location.isExpired; + } + + // Cancels the timers of locations that are no longer scheduled. + void _cancelTimersForRemovedLocations( + Map locationsToSchedule, + ) { + _scheduledTimers.removeWhere((messageId, scheduled) { + if (locationsToSchedule.containsKey(messageId)) return false; + scheduled.timer.cancel(); + return true; + }); + } + + // Reuses the existing timer when the [location]'s expiry time is unchanged, + // otherwise cancels the stale timer and arms a fresh one. + void _reuseOrRescheduleTimer(String messageId, Location location) { + final scheduled = _scheduledTimers[messageId]; + + // Same expiry time: keep the timer, just refresh the coordinates so an + // expiry reports the latest position. + if (scheduled != null && !_endAtChanged(location, scheduled)) { + scheduled.location = location; + return; + } + + // New location, or its expiry time changed: arm a fresh timer. + scheduled?.timer.cancel(); + _scheduledTimers[messageId] = _ScheduledExpiration( + location, + _createTimer(messageId, location.endAt!), + ); + } + + // Whether the [location]'s `endAt` differs from the one currently scheduled. + bool _endAtChanged(Location location, _ScheduledExpiration scheduled) { + return location.endAt != scheduled.location.endAt; + } + + Timer _createTimer(String messageId, DateTime endAt) { + final delay = endAt.difference(DateTime.now()); + return Timer( + delay.isNegative ? Duration.zero : delay, + () => _onTimerFired(messageId), + ); + } + + void _onTimerFired(String messageId) { + final scheduled = _scheduledTimers[messageId]; + if (scheduled == null) return; + + final location = scheduled.location; + // Re-check against the current clock; re-arm rather than fire if it isn't + // actually expired yet (e.g. the clock moved back or `endAt` was extended). + if (!location.isExpired) { + scheduled.timer = _createTimer(messageId, location.endAt!); + return; + } + + _scheduledTimers.remove(messageId); + onExpired(location); + } +} + +class _ScheduledExpiration { + _ScheduledExpiration(this.location, this.timer); + + Location location; + Timer timer; +} diff --git a/packages/stream_chat/test/src/client/channel_test.dart b/packages/stream_chat/test/src/client/channel_test.dart index d3330511d..52c8f9338 100644 --- a/packages/stream_chat/test/src/client/channel_test.dart +++ b/packages/stream_chat/test/src/client/channel_test.dart @@ -7729,6 +7729,57 @@ void main() { expect(channel.state?.activeLiveLocations, isEmpty); }); + test("should auto-expire another user's live location once at endAt", () async { + final liveLocation = Location( + channelCid: channel.cid, + userId: 'user1', // Another user. + messageId: 'msg1', + latitude: 40.7128, + longitude: -74.0060, + createdByDeviceId: 'device1', + endAt: DateTime.now().add(const Duration(milliseconds: 200)), + ); + + channel.state?.addNewMessage( + Message(id: 'msg1', sharedLocation: liveLocation), + ); + expect(channel.state?.activeLiveLocations, hasLength(1)); + + // Before endAt no expiry event is emitted. + await Future.delayed(const Duration(milliseconds: 80)); + verifyNever(() => client.handleEvent(any())); + + // After endAt the scheduler emits exactly one location.expired event. + await Future.delayed(const Duration(milliseconds: 250)); + final captured = verify(() => client.handleEvent(captureAny())).captured; + expect(captured, hasLength(1)); + final event = captured.single as Event; + expect(event.type, EventType.locationExpired); + expect(event.message?.id, 'msg1'); + }); + + test("should not auto-expire the current user's own live location", () async { + final ownLocation = Location( + channelCid: channel.cid, + userId: 'test-user-id', // The current user (handled by the client). + messageId: 'msg-own', + latitude: 40.7128, + longitude: -74.0060, + createdByDeviceId: 'device1', + endAt: DateTime.now().add(const Duration(milliseconds: 150)), + ); + + channel.state?.addNewMessage( + Message(id: 'msg-own', sharedLocation: ownLocation), + ); + expect(channel.state?.activeLiveLocations, hasLength(1)); + + // The channel scheduler skips the current user's own locations, so no + // expiry event is emitted even after endAt passes. + await Future.delayed(const Duration(milliseconds: 300)); + verifyNever(() => client.handleEvent(any())); + }); + test('should not add static location to active locations', () async { final staticLocation = Location( channelCid: channel.cid, diff --git a/packages/stream_chat/test/src/client/client_test.dart b/packages/stream_chat/test/src/client/client_test.dart index c341e23c8..0c66ac1af 100644 --- a/packages/stream_chat/test/src/client/client_test.dart +++ b/packages/stream_chat/test/src/client/client_test.dart @@ -4254,6 +4254,40 @@ void main() { expect(client.state.activeLiveLocations, isEmpty); }); + test('should auto-expire an active live location once at endAt', () async { + final expiredEvents = []; + final sub = client.on(EventType.locationExpired).listen(expiredEvents.add); + addTearDown(sub.cancel); + + // Setting an active location schedules a one-shot expiry timer. + client.state.activeLiveLocations = [ + Location( + channelCid: 'test-channel:123', + messageId: 'message-123', + userId: userId, + latitude: 40.7128, + longitude: -74.0060, + createdByDeviceId: 'device-1', + endAt: DateTime.now().add(const Duration(milliseconds: 200)), + ), + ]; + expect(client.state.activeLiveLocations, hasLength(1)); + + // Before endAt nothing is emitted and the location stays active. + await delay(80); + expect(expiredEvents, isEmpty); + expect(client.state.activeLiveLocations, hasLength(1)); + + // After endAt the timer fires once and the location is removed. + await delay(250); + expect(expiredEvents, hasLength(1)); + expect(client.state.activeLiveLocations, isEmpty); + + // The timer is one-shot: no further events are emitted. + await delay(200); + expect(expiredEvents, hasLength(1)); + }); + test('should ignore location events for other users', () async { final location = Location( channelCid: 'test-channel:123', diff --git a/packages/stream_chat/test/src/client/live_location_expiration_scheduler_test.dart b/packages/stream_chat/test/src/client/live_location_expiration_scheduler_test.dart new file mode 100644 index 000000000..b7a2bd36a --- /dev/null +++ b/packages/stream_chat/test/src/client/live_location_expiration_scheduler_test.dart @@ -0,0 +1,125 @@ +import 'package:stream_chat/src/client/live_location_expiration_scheduler.dart'; +import 'package:stream_chat/src/core/models/location.dart'; +import 'package:test/test.dart'; + +import '../utils.dart'; + +Location _location({ + String? messageId = 'msg1', + String userId = 'user1', + double latitude = 40.7128, + double longitude = -74.0060, + Duration? endsIn = const Duration(milliseconds: 200), +}) { + return Location( + channelCid: 'messaging:123', + messageId: messageId, + userId: userId, + latitude: latitude, + longitude: longitude, + endAt: endsIn == null ? null : DateTime.now().add(endsIn), + ); +} + +void main() { + group('LiveLocationExpirationScheduler', () { + late List expired; + late LiveLocationExpirationScheduler scheduler; + + setUp(() { + expired = []; + scheduler = LiveLocationExpirationScheduler(onExpired: expired.add); + }); + + tearDown(() => scheduler.cancel()); + + test('fires onExpired exactly once at endAt', () async { + scheduler.schedule([_location(endsIn: const Duration(milliseconds: 200))]); + + // Before endAt nothing fires. + await delay(80); + expect(expired, isEmpty); + + // After endAt it fires exactly once. + await delay(250); + expect(expired, hasLength(1)); + expect(expired.single.messageId, 'msg1'); + + // The timer is one-shot: no repeats even after more time passes. + await delay(200); + expect(expired, hasLength(1)); + }); + + test('ignores static, already-expired and message-id-less locations', () async { + scheduler.schedule([ + _location(messageId: 'static', endsIn: null), + _location(messageId: 'expired', endsIn: const Duration(milliseconds: -1)), + _location(messageId: null), + ]); + + await delay(150); + expect(expired, isEmpty); + }); + + test('reports latest coordinates and keeps expiry time on update', () async { + // A moving live location updates its coordinates but keeps the same + // endAt. + final original = _location( + latitude: 1, + longitude: 1, + endsIn: const Duration(milliseconds: 250), + ); + scheduler.schedule([original]); + + await delay(80); + // Same messageId and endAt, only the coordinates differ. + final moved = original.copyWith(latitude: 2, longitude: 2); + scheduler.schedule([moved]); + + // Still fires once at the original endAt (the update didn't shift it)... + await delay(250); + expect(expired, hasLength(1)); + // ...and reports the latest coordinates, not the original ones. + expect(expired.single.latitude, 2); + }); + + test('reschedules when endAt changes', () async { + final original = _location(endsIn: const Duration(milliseconds: 500)); + scheduler.schedule([original]); + + await delay(60); + final rescheduled = original.copyWith( + endAt: DateTime.now().add(const Duration(milliseconds: 150)), + ); + scheduler.schedule([rescheduled]); + + // Fires at the new (earlier) endAt, before the original one. + await delay(250); + expect(expired, hasLength(1)); + expect(expired.single.endAt, rescheduled.endAt); + + // The original timer was cancelled, so it never fires. + await delay(350); + expect(expired, hasLength(1)); + }); + + test('does not fire for a location removed before expiry', () async { + scheduler.schedule([_location(endsIn: const Duration(milliseconds: 300))]); + + await delay(80); + scheduler.schedule([]); // Removed from the active set. + + await delay(350); + expect(expired, isEmpty); + }); + + test('cancel() prevents pending timers from firing', () async { + scheduler + ..schedule([_location(endsIn: const Duration(milliseconds: 200))]) + ..cancel(); + + await delay(300); + expect(expired, isEmpty); + }); + }); +}