feat: MQTT client replaces WebSocket (Phase 2)

Matthias Nott

2026-03-22 c4ce6380fbfa55f22e9c20bb2ccffe4456ed9683

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
import 'dart:async';
import 'dart:convert';
import 'dart:io';

import 'package:path_provider/path_provider.dart';

import '../models/message.dart';

/// Per-session JSON file persistence with debounced saves.
class MessageStore {
  MessageStore._();

  static Directory? _baseDir;
  static Timer? _debounceTimer;
  static final Map<String, List<Message>> _pendingSaves = {};

  /// Initialize the base directory for message storage.
  static Future<Directory> _getBaseDir() async {
    if (_baseDir != null) return _baseDir!;
    final appDir = await getApplicationDocumentsDirectory();
    _baseDir = Directory('${appDir.path}/messages');
    if (!await _baseDir!.exists()) {
      await _baseDir!.create(recursive: true);
    }
    return _baseDir!;
  }

  static String _fileForSession(String sessionId) {
    // Sanitize session ID for filename
    final safe = sessionId.replaceAll(RegExp(r'[^\w\-]'), '_');
    return 'session_$safe.json';
  }

  /// Save messages for a session with 1-second debounce.
  static void save(String sessionId, List<Message> messages) {
    _pendingSaves[sessionId] = messages;
    _debounceTimer?.cancel();
    _debounceTimer = Timer(const Duration(seconds: 1), _flushAll);
  }

  /// Immediately flush all pending saves.
  static Future<void> flush() async {
    _debounceTimer?.cancel();
    await _flushAll();
  }

  static Future<void> _flushAll() async {
    final entries = Map<String, List<Message>>.from(_pendingSaves);
    _pendingSaves.clear();

    for (final entry in entries.entries) {
      await _writeSession(entry.key, entry.value);
    }
  }

  static Future<void> _writeSession(
      String sessionId, List<Message> messages) async {
    try {
      final dir = await _getBaseDir();
      final file = File('${dir.path}/${_fileForSession(sessionId)}');
      // Strip heavy fields for persistence
      final lightMessages = messages.map((m) => m.toJsonLight()).toList();
      await file.writeAsString(jsonEncode(lightMessages));
    } catch (e) {
      // Silently fail - message persistence is best-effort
    }
  }

  /// Load messages for a session.
  /// [limit] controls how many recent messages to return (default: 50).
  /// [offset] is the number of messages to skip from the end (for pagination).
  static Future<List<Message>> load(
    String sessionId, {
    int limit = 50,
    int offset = 0,
  }) async {
    try {
      final dir = await _getBaseDir();
      final file = File('${dir.path}/${_fileForSession(sessionId)}');
      if (!await file.exists()) return [];

      final jsonStr = await file.readAsString();
      final List<dynamic> jsonList = jsonDecode(jsonStr) as List<dynamic>;
      final allMessages = jsonList
          .map((j) => _messageFromJson(j as Map<String, dynamic>))
          .where((m) => !m.isEmptyVoice && !m.isEmptyText)
          .toList();

      // Paginate from the end (newest messages first in storage)
      if (offset >= allMessages.length) return [];
      final end = allMessages.length - offset;
      final start = (end - limit).clamp(0, end);
      return allMessages.sublist(start, end);
    } catch (e) {
      return [];
    }
  }

  /// Load all messages for a session (no pagination).
  static Future<List<Message>> loadAll(String sessionId) async {
    try {
      final dir = await _getBaseDir();
      final file = File('${dir.path}/${_fileForSession(sessionId)}');
      if (!await file.exists()) return [];

      final jsonStr = await file.readAsString();
      final List<dynamic> jsonList = jsonDecode(jsonStr) as List<dynamic>;
      return jsonList
          .map((j) => _messageFromJson(j as Map<String, dynamic>))
          .where((m) => !m.isEmptyVoice && !m.isEmptyText)
          .toList();
    } catch (e) {
      return [];
    }
  }

  /// Deserialize a message from JSON, applying migration rules:
  /// - Voice messages without audioUri are downgraded to text (transcript only).
  ///   This handles messages saved before a restart, where the temp audio file
  ///   is no longer available. The transcript (content) is preserved.
  static Message _messageFromJson(Map<String, dynamic> json) {
    final raw = Message.fromJson(json);
    if (raw.type == MessageType.voice &&
        (raw.audioUri == null || raw.audioUri!.isEmpty)) {
      // Downgrade to text so the bubble shows the transcript instead of a
      // broken play button.
      return Message(
        id: raw.id,
        role: raw.role,
        type: MessageType.text,
        content: raw.content,
        timestamp: raw.timestamp,
        status: raw.status,
        duration: raw.duration,
      );
    }
    return raw;
  }

  /// Delete stored messages for a session.
  static Future<void> delete(String sessionId) async {
    try {
      final dir = await _getBaseDir();
      final file = File('${dir.path}/${_fileForSession(sessionId)}');
      if (await file.exists()) {
        await file.delete();
      }
    } catch (_) {}
  }

  /// Clear all stored messages.
  static Future<void> clearAll() async {
    try {
      final dir = await _getBaseDir();
      if (await dir.exists()) {
        await dir.delete(recursive: true);
        await dir.create(recursive: true);
      }
    } catch (_) {}
  }
}