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

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', () {