store: Replace event queue on expiry

Fixes: #185

Author: gnprice

lib/model/store.dart

@@ -8,6 +8,7 @@ import 'package:path/path.dart' as p;
 import 'package:path_provider/path_provider.dart';
 
 import '../api/core.dart';
+import '../api/exception.dart';
 import '../api/model/events.dart';
 import '../api/model/initial_snapshot.dart';
 import '../api/model/model.dart';
@@ -96,12 +97,26 @@ abstract class GlobalStore extends ChangeNotifier {
     future = loadPerAccount(account!);
     _perAccountStoresLoading[accountId] = future;
     store = await future;
-    _perAccountStores[accountId] = store;
+    _setPerAccount(accountId, store);
     _perAccountStoresLoading.remove(accountId);
-    notifyListeners();
     return store;
   }
 
+  Future<void> _reloadPerAccount(Account account) async {
+    assert(identical(_accounts[account.id], account));
+    assert(_perAccountStores.containsKey(account.id));
+    assert(!_perAccountStoresLoading.containsKey(account.id));
+    final store = await loadPerAccount(account);
+    _setPerAccount(account.id, store);
+  }
+
+  void _setPerAccount(int accountId, PerAccountStore store) {
+    final oldStore = _perAccountStores[accountId];
+    _perAccountStores[accountId] = store;
+    notifyListeners();
+    oldStore?.dispose();
+  }
+
   /// Load per-account data for the given account, unconditionally.
   ///
   /// This method should be called only by the implementation of [perAccount].
@@ -199,8 +214,6 @@ class PerAccountStore extends ChangeNotifier with StreamStore {
   }) : _globalStore = globalStore,
        _streams = streams;
 
-  // We'll use this in an upcoming commit.
-  // ignore: unused_field
   final GlobalStore _globalStore;
 
   final Account account;
@@ -569,9 +582,26 @@ class UpdateMachine {
         }());
       }
 
-      final result = await getEvents(store.connection,
-        queueId: queueId, lastEventId: lastEventId);
-      // TODO handle errors on get-events; retry with backoff
+      final GetEventsResult result;
+      try {
+        result = await getEvents(store.connection,
+          queueId: queueId, lastEventId: lastEventId);
+      } catch (e) {
+        switch (e) {
+          case ZulipApiException(code: 'BAD_EVENT_QUEUE_ID'):
+            assert(debugLog('Lost event queue for $store.  Replacing…'));
+            await store._globalStore._reloadPerAccount(store.account);
+            dispose();
+            debugLog('… Event queue replaced.');
+            return;
+
+          default:
+            assert(debugLog('Error polling event queue for $store: $e'));
+            // TODO(#184) handle errors on get-events; retry with backoff
+            rethrow;
+        }
+      }
+
       final events = result.events;
       for (final event in events) {
         store.handleEvent(event);

lib/widgets/message_list.dart

@@ -186,8 +186,7 @@ class _MessageListState extends State<MessageList> with PerAccountStoreAwareStat
   }
 
   @override
-  void onNewStore() {
-    model?.dispose();
+  void onNewStore() { // TODO(#464) try to keep using old model until new one gets messages
     _initModel(PerAccountStoreWidget.of(context));
   }
 

lib/widgets/store.dart

@@ -135,14 +135,17 @@ class PerAccountStoreWidget extends StatefulWidget {
   /// use [State.didChangeDependencies] instead.  For discussion, see
   /// [BuildContext.dependOnInheritedWidgetOfExactType].
   ///
+  /// A [State] that calls this method in [State.didChangeDependencies]
+  /// should typically also mix in [PerAccountStoreAwareStateMixin],
+  /// in order to handle the store itself being replaced with a new store.
+  /// This happens in particular if the old store's event queue expires
+  /// on the server.
+  ///
   /// See also:
   ///  * [accountIdOf], for the account ID corresponding to the same data.
   ///  * [GlobalStoreWidget.of], for the app's data beyond that of a
   ///    particular account.
   ///  * [InheritedNotifier], which provides the "dependency" mechanism.
-  // TODO(#185): Explain in dartdoc that the returned [PerAccountStore] might
-  //   differ from one call to the next, and to handle that with
-  //   [PerAccountStoreAwareStateMixin].
   static PerAccountStore of(BuildContext context) {
     final widget = context.dependOnInheritedWidgetOfExactType<_PerAccountStoreInheritedWidget>();
     assert(widget != null, 'No PerAccountStoreWidget ancestor');
@@ -253,13 +256,11 @@ class LoadingPage extends StatelessWidget {
 /// The ambient [PerAccountStore] can be replaced in some circumstances,
 /// such as when an event queue expires. See [PerAccountStoreWidget.of].
 /// When that happens, stateful widgets should
-/// - remove listeners on the old [PerAccountStore], and
+/// - stop using the old [PerAccountStore], which will already have
+///   been disposed;
 /// - add listeners on the new one.
 ///
 /// Use this mixin, overriding [onNewStore], to do that concisely.
-// TODO(#185): Until #185, I think [PerAccountStoreWidget.of] never actually
-//   returns a different [PerAccountStore] from one call to the next.
-//   But it will, and when it does, we want our [StatefulWidgets] to handle it.
 mixin PerAccountStoreAwareStateMixin<T extends StatefulWidget> on State<T> {
   PerAccountStore? _store;
 
@@ -270,8 +271,9 @@ mixin PerAccountStoreAwareStateMixin<T extends StatefulWidget> on State<T> {
   /// and again whenever dependencies change so that [PerAccountStoreWidget.of]
   /// would return a different store from previously.
   ///
-  /// In this, remove any listeners on the old store
-  /// and add them on the new store.
+  /// In this, add any needed listeners on the new store
+  /// and drop any references to the old store, which will already
+  /// have been disposed.
   void onNewStore();
 
   @override

test/model/store_test.dart

@@ -140,18 +140,26 @@ void main() {
   });
 
   group('UpdateMachine.poll', () {
+    late TestGlobalStore globalStore;
     late UpdateMachine updateMachine;
     late PerAccountStore store;
     late FakeApiConnection connection;
 
-    void prepareStore({int? lastEventId}) {
-      updateMachine = eg.updateMachine(initialSnapshot: eg.initialSnapshot(
-        lastEventId: lastEventId,
-      ));
+    void updateFromGlobalStore() {
+      updateMachine = globalStore.updateMachines[eg.selfAccount.id]!;
       store = updateMachine.store;
+      assert(identical(store, globalStore.perAccountSync(eg.selfAccount.id)));
       connection = store.connection as FakeApiConnection;
     }
 
+    Future<void> prepareStore({int? lastEventId}) async {
+      globalStore = TestGlobalStore(accounts: []);
+      await globalStore.add(eg.selfAccount, eg.initialSnapshot(
+        lastEventId: lastEventId));
+      await globalStore.perAccount(eg.selfAccount.id);
+      updateFromGlobalStore();
+    }
+
     void checkLastRequest({required int lastEventId}) {
       check(connection.lastRequest).isA<http.Request>()
         ..method.equals('GET')
@@ -163,7 +171,7 @@ void main() {
     }
 
     test('loops on success', () async {
-      prepareStore(lastEventId: 1);
+      await prepareStore(lastEventId: 1);
       check(updateMachine.lastEventId).equals(1);
 
       updateMachine.debugPauseLoop();
@@ -191,7 +199,7 @@ void main() {
     });
 
     test('handles events', () async {
-      prepareStore();
+      await prepareStore();
       updateMachine.debugPauseLoop();
       updateMachine.poll();
 
@@ -206,6 +214,40 @@ void main() {
       await Future.delayed(Duration.zero);
       check(store.userSettings!.twentyFourHourTime).isTrue();
     });
+
+    test('handles expired queue', () async {
+      await prepareStore();
+      updateMachine.debugPauseLoop();
+      updateMachine.poll();
+      check(globalStore.perAccountSync(store.account.id)).identicalTo(store);
+
+      // Let the server expire the event queue.
+      connection.prepare(httpStatus: 400, json: {
+        'result': 'error', 'code': 'BAD_EVENT_QUEUE_ID',
+        'queue_id': updateMachine.queueId,
+        'msg': 'Bad event queue ID: ${updateMachine.queueId}',
+      });
+      updateMachine.debugAdvanceLoop();
+      await null;
+      await Future.delayed(Duration.zero);
+
+      // The global store has a new store.
+      check(globalStore.perAccountSync(store.account.id)).not(it()..identicalTo(store));
+      updateFromGlobalStore();
+
+      // The new UpdateMachine updates the new store.
+      updateMachine.debugPauseLoop();
+      updateMachine.poll();
+      check(store.userSettings!.twentyFourHourTime).isFalse();
+      connection.prepare(json: GetEventsResult(events: [
+        UserSettingsUpdateEvent(id: 2,
+          property: UserSettingName.twentyFourHourTime, value: true),
+      ], queueId: null).toJson());
+      updateMachine.debugAdvanceLoop();
+      await null;
+      await Future.delayed(Duration.zero);
+      check(store.userSettings!.twentyFourHourTime).isTrue();
+    });
   });
 
   group('UpdateMachine.registerNotificationToken', () {