feat: MQTT migration, offline catch_up, clean session, image support

Matthias Nott

2026-03-24 cb470d33d2665fcc6f8448d2736777656cf0cbe7

 lib/screens/chat_screen.dart
..
..
@@ -56,6 +56,9 @@
56
56
   bool _isCatchingUp = false;
57
57
   bool _screenshotForChat = false;
58
58
   final Set<int> _seenSeqs = {};
59
+  bool _sessionReady = false;
60
+  final List<Map<String, dynamic>> _pendingMessages = [];
61
+  final Map<String, List<Message>> _catchUpPending = {};
59
62
 
60
63
   @override
61
64
   void initState() {
..
..
@@ -66,9 +69,14 @@
66
69
   }
67
70
 
68
71
   Future<void> _initAll() async {
69
-    // Load lastSeq BEFORE connecting so catch_up sends the right value
72
+    // Load persisted state BEFORE connecting
70
73
     final prefs = await SharedPreferences.getInstance();
71
74
     _lastSeq = prefs.getInt('lastSeq') ?? 0;
75
+    // Restore last active session so catch_up routes to the right session
76
+    final savedSessionId = prefs.getString('activeSessionId');
77
+    if (savedSessionId != null && mounted) {
78
+      ref.read(activeSessionIdProvider.notifier).state = savedSessionId;
79
+    }
72
80
     if (!mounted) return;
73
81
 
74
82
     // Listen for playback state changes to reset play button UI
..
..
@@ -146,10 +154,11 @@
146
154
     };
147
155
     _ws!.onMessage = _handleMessage;
148
156
     _ws!.onOpen = () {
157
+      _sessionReady = false; // Gate messages until sessions arrive
158
+      _pendingMessages.clear();
149
159
       final activeId = ref.read(activeSessionIdProvider);
150
160
       _sendCommand('sync', activeId != null ? {'activeSessionId': activeId} : null);
151
-      // catch_up is still available during the transition period
152
-      _sendCommand('catch_up', {'lastSeq': _lastSeq});
161
+      // catch_up is sent after sessions arrive (in _handleSessions)
153
162
     };
154
163
     _ws!.onError = (error) {
155
164
       debugPrint('MQTT error: $error');
..
..
@@ -168,6 +177,14 @@
168
177
   }
169
178
 
170
179
   void _handleMessage(Map<String, dynamic> msg) {
180
+    final type = msg['type'] as String?;
181
+    // Sessions and catch_up always process immediately
182
+    // Content messages (text, voice, image) wait until session is ready
183
+    if (!_sessionReady && type != 'sessions' && type != 'catch_up' && type != 'status' && type != 'typing') {
184
+      _pendingMessages.add(msg);
185
+      return;
186
+    }
187
+
171
188
     // Track sequence numbers for catch_up protocol
172
189
     final seq = msg['seq'] as int?;
173
190
     if (seq != null) {
..
..
@@ -184,8 +201,6 @@
184
201
         _saveLastSeq();
185
202
       }
186
203
     }
187
-
188
-    final type = msg['type'] as String?;
189
204
 
190
205
     switch (type) {
191
206
       case 'sessions':
..
..
@@ -231,7 +246,8 @@
231
246
         if (sessionId != null) _incrementUnread(sessionId);
232
247
       case 'catch_up':
233
248
         final serverSeq = msg['serverSeq'] as int?;
234
-        if (serverSeq != null && serverSeq > _lastSeq) {
249
+        if (serverSeq != null) {
250
+          // Always sync to server's seq — if server restarted, its seq may be lower
235
251
           _lastSeq = serverSeq;
236
252
           _saveLastSeq();
237
253
         }
..
..
@@ -241,19 +257,91 @@
241
257
         final catchUpMsgs = msg['messages'] as List<dynamic>?;
242
258
         if (catchUpMsgs != null && catchUpMsgs.isNotEmpty) {
243
259
           _isCatchingUp = true;
260
+          final activeId = ref.read(activeSessionIdProvider);
244
261
           final existing = ref.read(messagesProvider);
245
262
           final existingContents = existing
246
263
               .where((m) => m.role == MessageRole.assistant)
247
264
               .map((m) => m.content)
248
265
               .toSet();
249
266
           for (final m in catchUpMsgs) {
250
-            final content = (m as Map<String, dynamic>)['content'] as String? ?? '';
251
-            // Skip if we already have this message locally
252
-            if (content.isNotEmpty && existingContents.contains(content)) continue;
253
-            _handleMessage(m);
254
-            if (content.isNotEmpty) existingContents.add(content);
267
+            final map = m as Map<String, dynamic>;
268
+            final msgType = map['type'] as String? ?? 'text';
269
+            final content = map['content'] as String? ?? map['transcript'] as String? ?? map['caption'] as String? ?? '';
270
+            final msgSessionId = map['sessionId'] as String?;
271
+            final imageData = map['imageBase64'] as String?;
272
+
273
+            // Skip empty text messages (images with no caption are OK)
274
+            if (content.isEmpty && imageData == null) continue;
275
+            // Dedup by content (skip images from dedup — they have unique msgIds)
276
+            if (imageData == null && content.isNotEmpty && existingContents.contains(content)) continue;
277
+
278
+            final Message message;
279
+            if (msgType == 'image' && imageData != null) {
280
+              message = Message.image(
281
+                role: MessageRole.assistant,
282
+                imageBase64: imageData,
283
+                content: content,
284
+                status: MessageStatus.sent,
285
+              );
286
+            } else {
287
+              message = Message.text(
288
+                role: MessageRole.assistant,
289
+                content: content,
290
+                status: MessageStatus.sent,
291
+              );
292
+            }
293
+
294
+            if (msgSessionId == null || msgSessionId == activeId) {
295
+              // Active session or no session: add directly to chat
296
+              ref.read(messagesProvider.notifier).addMessage(message);
297
+            } else {
298
+              // Different session: store + unread badge + toast
299
+              // Collect for batch storage below to avoid race condition
300
+              _catchUpPending.putIfAbsent(msgSessionId, () => []).add(message);
301
+              _incrementUnread(msgSessionId);
302
+            }
303
+            existingContents.add(content);
255
304
           }
256
305
           _isCatchingUp = false;
306
+          _scrollToBottom();
307
+          // Batch-store cross-session messages (sequential to avoid race condition)
308
+          if (_catchUpPending.isNotEmpty) {
309
+            final pending = Map<String, List<Message>>.from(_catchUpPending);
310
+            _catchUpPending.clear();
311
+            // Show one toast per session with message count
312
+            if (mounted) {
313
+              final sessions = ref.read(sessionsProvider);
314
+              for (final entry in pending.entries) {
315
+                final session = sessions.firstWhere(
316
+                  (s) => s.id == entry.key,
317
+                  orElse: () => Session(id: entry.key, index: 0, name: 'Unknown', type: 'claude'),
318
+                );
319
+                final count = entry.value.length;
320
+                final preview = count == 1
321
+                    ? entry.value.first.content
322
+                    : '$count messages';
323
+                ToastManager.show(
324
+                  context,
325
+                  sessionName: session.name,
326
+                  preview: preview.length > 100 ? '${preview.substring(0, 100)}...' : preview,
327
+                  onTap: () => _switchSession(entry.key),
328
+                );
329
+              }
330
+            }
331
+            () async {
332
+              for (final entry in pending.entries) {
333
+                final existing = await MessageStore.loadAll(entry.key);
334
+                MessageStore.save(entry.key, [...existing, ...entry.value]);
335
+                await MessageStore.flush();
336
+              }
337
+            }();
338
+          }
339
+          // Clear unread for active session
340
+          if (activeId != null) {
341
+            final counts = Map<String, int>.from(ref.read(unreadCountsProvider));
342
+            counts.remove(activeId);
343
+            ref.read(unreadCountsProvider.notifier).state = counts;
344
+          }
257
345
         }
258
346
       case 'pong':
259
347
         break; // heartbeat response, ignore
..
..
@@ -284,6 +372,22 @@
284
372
       );
285
373
       ref.read(activeSessionIdProvider.notifier).state = active.id;
286
374
       ref.read(messagesProvider.notifier).switchSession(active.id);
375
+      SharedPreferences.getInstance().then((p) => p.setString('activeSessionId', active.id));
376
+    }
377
+
378
+    // Session is ready — process any pending messages that arrived before sessions list
379
+    if (!_sessionReady) {
380
+      _sessionReady = true;
381
+      // Request catch_up now that session is set
382
+      _sendCommand('catch_up', {'lastSeq': _lastSeq});
383
+      // Drain messages that arrived before sessions list
384
+      if (_pendingMessages.isNotEmpty) {
385
+        final pending = List<Map<String, dynamic>>.from(_pendingMessages);
386
+        _pendingMessages.clear();
387
+        for (final m in pending) {
388
+          _handleMessage(m);
389
+        }
390
+      }
287
391
     }
288
392
   }
289
393
 
..
..
@@ -507,6 +611,7 @@
507
611
 
508
612
     ref.read(activeSessionIdProvider.notifier).state = sessionId;
509
613
     await ref.read(messagesProvider.notifier).switchSession(sessionId);
614
+    SharedPreferences.getInstance().then((p) => p.setString('activeSessionId', sessionId));
510
615
 
511
616
     final counts = Map<String, int>.from(ref.read(unreadCountsProvider));
512
617
     counts.remove(sessionId);