fix(llc): replace live location polling with one-shot expiry timers

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
 

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();
   }
 }

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<Location> 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) {

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 = <String, _ScheduledExpiration>{};
+
+  /// 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<Location> locations) {
+    // The live locations that should have an expiry timer, keyed by message id.
+    final locationsToSchedule = <String, Location>{
+      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<String, Location> 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;
+}

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,