fix: resume message reload, direct session writes, MQTT trace pipe to server

Matthias Nott

8 days ago 1bf6e76e31383aef77e42943fc2caf350cf7e096

fix: resume message reload, direct session writes, MQTT trace pipe to server

 lib/providers/providers.dart
..
..
@@ -99,10 +99,9 @@
99
99
 
100
100
   /// Switch to a new session and load its messages.
101
101
   Future<void> switchSession(String sessionId) async {
102
-    // Force-flush current session to disk before switching
102
+    // Write current session DIRECTLY to disk (no debounce — prevents data loss)
103
103
     if (_currentSessionId != null && state.isNotEmpty) {
104
-      MessageStore.save(_currentSessionId!, state);
105
-      await MessageStore.flush();
104
+      await MessageStore.writeDirect(_currentSessionId!, state);
106
105
     }
107
106
 
108
107
     _currentSessionId = sessionId;
 lib/screens/chat_screen.dart
..
..
@@ -65,6 +65,7 @@
65
65
   String? _playingMessageId;
66
66
   int _lastSeq = 0;
67
67
   bool _isCatchingUp = false;
68
+  bool _catchUpReceived = false;
68
69
   bool _screenshotForChat = false;
69
70
   // FIFO dedup queue: O(1) eviction by removing from front when over cap.
70
71
   final List<int> _seenSeqsList = [];
..
..
@@ -222,28 +223,20 @@
222
223
       // Re-register APNs token after reconnect so daemon always has a fresh token
223
224
       _push?.onMqttConnected();
224
225
     };
225
-    _ws!.onResume = () {
226
-      // App came back from background — connection may or may not be alive.
227
-      // Try catch_up first, but if no response comes, force reconnect.
228
-      _chatLog('onResume: sending catch_up with lastSeq=$_lastSeq');
229
-      final seqBefore = _lastSeq;
226
+    _ws!.onResume = () async {
227
+      // App came back from background — reload messages and catch up.
228
+      _chatLog('onResume: reloading messages and sending catch_up');
230
229
       _sendCommand('catch_up', {'lastSeq': _lastSeq});
231
-      // Force UI rebuild for any buffered messages
232
-      Future.delayed(const Duration(milliseconds: 300), () {
233
-        if (mounted) {
234
-          setState(() {});
235
-          _scrollToBottom();
236
-        }
237
-      });
238
-      // If catch_up didn't produce a response in 2s, connection is dead — reconnect
239
-      Future.delayed(const Duration(seconds: 2), () {
240
-        if (!mounted) return;
241
-        if (_lastSeq == seqBefore) {
242
-          // No new messages arrived — connection likely dead
243
-          _chatLog('onResume: no catch_up response after 2s, forcing reconnect');
244
-          _ws?.forceReconnect();
245
-        }
246
-      });
230
+      // Force reload current session messages from provider (triggers rebuild)
231
+      final activeId = ref.read(activeSessionIdProvider);
232
+      if (activeId != null) {
233
+        // Re-save then re-load to ensure UI matches persisted state
234
+        await ref.read(messagesProvider.notifier).switchSession(activeId);
235
+      }
236
+      if (mounted) {
237
+        setState(() {});
238
+        _scrollToBottom();
239
+      }
247
240
     };
248
241
     _ws!.onError = (error) {
249
242
       debugPrint('MQTT error: $error');
..
..
@@ -259,6 +252,9 @@
259
252
     );
260
253
 
261
254
     await _ws!.connect();
255
+
256
+    // Attach MQTT to trace service for auto-publishing logs to server
257
+    TraceService.instance.attachMqtt(_ws!);
262
258
 
263
259
     // Initialize push notifications after MQTT is set up so token can be
264
260
     // sent immediately if already connected.
..
..
@@ -371,6 +367,7 @@
371
367
         final sessionId = msg['sessionId'] as String?;
372
368
         if (sessionId != null) _incrementUnread(sessionId);
373
369
       case 'catch_up':
370
+        _catchUpReceived = true;
374
371
         final serverSeq = msg['serverSeq'] as int?;
375
372
         if (serverSeq != null) {
376
373
           // Always sync to server's seq — if server restarted, its seq may be lower
 lib/services/message_store.dart
..
..
@@ -57,6 +57,12 @@
57
57
     _debounceTimer = Timer(const Duration(seconds: 1), _flushAll);
58
58
   }
59
59
 
60
+  /// Write directly to disk, bypassing debounce. For critical saves.
61
+  static Future<void> writeDirect(String sessionId, List<Message> messages) async {
62
+    _pendingSaves.remove(sessionId);
63
+    await _writeSession(sessionId, messages);
64
+  }
65
+
60
66
   /// Immediately flush all pending saves.
61
67
   static Future<void> flush() async {
62
68
     _debounceTimer?.cancel();
 lib/services/mqtt_service.dart
..
..
@@ -11,6 +11,7 @@
11
11
 import 'package:path_provider/path_provider.dart' as pp;
12
12
 import 'package:mqtt_client/mqtt_client.dart';
13
13
 import 'package:mqtt_client/mqtt_server_client.dart';
14
+import 'package:typed_data/typed_data.dart';
14
15
 import 'package:shared_preferences/shared_preferences.dart';
15
16
 import 'package:uuid/uuid.dart';
16
17
 
..
..
@@ -665,6 +666,15 @@
665
666
   /// Current timestamp in milliseconds.
666
667
   int _now() => DateTime.now().millisecondsSinceEpoch;
667
668
 
669
+  /// Publish raw bytes to a topic. Used by TraceService for log streaming.
670
+  void publishRaw(String topic, Uint8Buffer payload, MqttQos qos) {
671
+    final client = _client;
672
+    if (client == null || client.connectionStatus?.state != MqttConnectionState.connected) return;
673
+    try {
674
+      client.publishMessage(topic, qos, payload);
675
+    } catch (_) {}
676
+  }
677
+
668
678
   /// Publish a JSON payload to an MQTT topic.
669
679
   void _publish(String topic, Map<String, dynamic> payload, MqttQos qos) {
670
680
     final client = _client;
 lib/services/trace_service.dart
..
..
@@ -1,4 +1,10 @@
1
+import 'dart:convert';
2
+
1
3
 import 'package:flutter/foundation.dart';
4
+import 'package:mqtt_client/mqtt_client.dart';
5
+import 'package:mqtt_client/mqtt_server_client.dart';
6
+
7
+import 'mqtt_service.dart';
2
8
 
3
9
 /// A single trace entry capturing a message-handling event.
4
10
 class TraceEntry {
..
..
@@ -22,17 +28,28 @@
22
28
 /// Captures message-handling events from MQTT, chat screen, and other
23
29
 /// components. The buffer is capped at [maxEntries] (default 200).
24
30
 /// Works in both debug and release builds.
31
+///
32
+/// When an MqttService is attached via [attachMqtt], trace entries are
33
+/// automatically published to the server on `pailot/control/in` so they
34
+/// can be read from the daemon log.
25
35
 class TraceService {
26
36
   TraceService._();
27
37
   static final TraceService instance = TraceService._();
28
38
 
29
39
   static const int maxEntries = 200;
30
40
   final List<TraceEntry> _entries = [];
41
+  MqttService? _mqtt;
42
+
43
+  /// Attach an MQTT service for auto-publishing traces to the server.
44
+  void attachMqtt(MqttService mqtt) {
45
+    _mqtt = mqtt;
46
+  }
31
47
 
32
48
   /// All entries, oldest first.
33
49
   List<TraceEntry> get entries => List.unmodifiable(_entries);
34
50
 
35
51
   /// Add a trace entry. Oldest entry is evicted once the buffer is full.
52
+  /// If MQTT is attached and connected, the entry is also published to the server.
36
53
   void addTrace(String event, String details) {
37
54
     _entries.add(TraceEntry(
38
55
       timestamp: DateTime.now(),
..
..
@@ -43,6 +60,28 @@
43
60
       _entries.removeAt(0);
44
61
     }
45
62
     debugPrint('[TRACE] $event — $details');
63
+
64
+    // Auto-publish to server if MQTT is connected
65
+    _publishTrace(event, details);
66
+  }
67
+
68
+  void _publishTrace(String event, String details) {
69
+    final mqtt = _mqtt;
70
+    if (mqtt == null || !mqtt.isConnected) return;
71
+    try {
72
+      final payload = jsonEncode({
73
+        'type': 'command',
74
+        'command': 'app_trace',
75
+        'event': event,
76
+        'details': details,
77
+        'ts': DateTime.now().millisecondsSinceEpoch,
78
+      });
79
+      final builder = MqttClientPayloadBuilder();
80
+      builder.addString(payload);
81
+      mqtt.publishRaw('pailot/control/in', builder.payload!, MqttQos.atMostOnce);
82
+    } catch (_) {
83
+      // Non-fatal — don't let trace logging break the app
84
+    }
46
85
   }
47
86
 
48
87
   /// Clear all entries.