From 06bb73662d32d65d1e775a4dd35f67d82d673e40 Mon Sep 17 00:00:00 2001
From: Matthias Nott <mnott@mnsoft.org>
Date: Mon, 06 Apr 2026 15:32:03 +0200
Subject: [PATCH] feat: rewrite message store - append-only log, sync routing, eliminates async race conditions
---
lib/services/message_store.dart | 376 +++++++++++++++++++++------------
lib/providers/providers.dart | 73 +-----
lib/screens/chat_screen.dart | 173 ++++++---------
3 files changed, 326 insertions(+), 296 deletions(-)
diff --git a/lib/providers/providers.dart b/lib/providers/providers.dart
index 6155189..06fe159 100644
--- a/lib/providers/providers.dart
+++ b/lib/providers/providers.dart
@@ -98,73 +98,48 @@
String? get currentSessionId => _currentSessionId;
- /// Switch to a new session and load its messages.
- Future<void> switchSession(String sessionId) async {
- final trace = StackTrace.current.toString().split('\n').take(4).join(' | ');
- TraceService.instance.addTrace(
- 'switchSession',
- 'from=${_currentSessionId?.substring(0, 8) ?? "null"}(${state.length}) → ${sessionId.substring(0, 8)} | $trace',
- );
- // Write current session to disk
- if (_currentSessionId != null && state.isNotEmpty) {
- await MessageStore.writeDirect(_currentSessionId!, state);
- }
-
- // Skip reload if staying on the same session — messages are already in memory
+ /// Switch to a session. SYNCHRONOUS — no async gap, no race with incoming
+ /// messages. MessageStoreV2.loadSession reads from the in-memory index.
+ void switchSession(String sessionId) {
if (_currentSessionId == sessionId) {
- TraceService.instance.addTrace('switchSession SKIP', 'already on ${sessionId.substring(0, 8)}');
+ TraceService.instance.addTrace(
+ 'switchSession SKIP', 'already on ${sessionId.substring(0, 8)}');
return;
}
-
+ TraceService.instance.addTrace(
+ 'switchSession',
+ 'from=${_currentSessionId?.substring(0, 8) ?? "null"}(${state.length}) → ${sessionId.substring(0, 8)}',
+ );
_currentSessionId = sessionId;
- final messages = await MessageStore.loadAll(sessionId);
- // Merge: if addMessage ran during loadAll and added messages for THIS session,
- // they'll be in state but not in the loaded messages. Keep the longer list.
- if (state.length > messages.length && _currentSessionId == sessionId) {
- TraceService.instance.addTrace('switchSession MERGE', 'kept ${state.length} (loaded ${messages.length})');
- } else {
- state = messages;
- }
+ state = MessageStoreV2.loadSession(sessionId);
}
- /// Add a message to the current session.
+ /// Add a message to the current session (display + append-only persist).
void addMessage(Message message) {
state = [...state, message];
if (_currentSessionId != null) {
- MessageStore.save(_currentSessionId!, state);
+ MessageStoreV2.append(_currentSessionId!, message);
}
}
- /// Update a message by ID.
+ /// Update a message by ID (in-memory only — patch is not persisted to log).
void updateMessage(String id, Message Function(Message) updater) {
state = state.map((m) => m.id == id ? updater(m) : m).toList();
- if (_currentSessionId != null) {
- MessageStore.save(_currentSessionId!, state);
- }
}
- /// Remove a message by ID.
+ /// Remove a message by ID (in-memory only).
void removeMessage(String id) {
state = state.where((m) => m.id != id).toList();
- if (_currentSessionId != null) {
- MessageStore.save(_currentSessionId!, state);
- }
}
- /// Remove all messages matching a predicate.
+ /// Remove all messages matching a predicate (in-memory only).
void removeWhere(bool Function(Message) test) {
state = state.where((m) => !test(m)).toList();
- if (_currentSessionId != null) {
- MessageStore.save(_currentSessionId!, state);
- }
}
- /// Clear all messages for the current session.
+ /// Clear all messages for the current session (in-memory only).
void clearMessages() {
state = [];
- if (_currentSessionId != null) {
- MessageStore.save(_currentSessionId!, state);
- }
}
void updateContent(String messageId, String content) {
@@ -185,22 +160,6 @@
else
m,
];
- if (_currentSessionId != null) {
- MessageStore.save(_currentSessionId!, state);
- }
- }
-
- /// Load more (older) messages for pagination.
- Future<void> loadMore() async {
- if (_currentSessionId == null) return;
- final older = await MessageStore.load(
- _currentSessionId!,
- offset: state.length,
- limit: 50,
- );
- if (older.isNotEmpty) {
- state = [...older, ...state];
- }
}
}
diff --git a/lib/screens/chat_screen.dart b/lib/screens/chat_screen.dart
index f331ab9..ee0f68b 100644
--- a/lib/screens/chat_screen.dart
+++ b/lib/screens/chat_screen.dart
@@ -72,7 +72,8 @@
final Set<int> _seenSeqs = {};
bool _sessionReady = false;
final List<Map<String, dynamic>> _pendingMessages = [];
- final Map<String, List<Message>> _catchUpPending = {};
+ // _catchUpPending removed: cross-session catch_up messages are now appended
+ // synchronously via MessageStoreV2.append() in the catch_up handler.
List<String>? _cachedSessionOrder;
Timer? _typingTimer;
bool _unreadCountsLoaded = false;
@@ -86,6 +87,9 @@
}
Future<void> _initAll() async {
+ // Initialize append-only message store (reads log, rebuilds index, compacts).
+ await MessageStoreV2.initialize();
+
// Load persisted state BEFORE connecting
final prefs = await SharedPreferences.getInstance();
_lastSeq = prefs.getInt('lastSeq') ?? 0;
@@ -104,8 +108,8 @@
final savedSessionId = prefs.getString('activeSessionId');
if (savedSessionId != null && mounted) {
ref.read(activeSessionIdProvider.notifier).state = savedSessionId;
- // Load messages for the restored session so chat isn't empty on startup
- await ref.read(messagesProvider.notifier).switchSession(savedSessionId);
+ // Synchronous: no async gap between load and any arriving messages.
+ ref.read(messagesProvider.notifier).switchSession(savedSessionId);
}
if (!mounted) return;
@@ -165,14 +169,11 @@
_persistUnreadCounts(counts);
}
+ // ignore: unused_field
bool _isLoadingMore = false;
void _onScroll() {
- if (!_isLoadingMore &&
- _scrollController.position.pixels >=
- _scrollController.position.maxScrollExtent - 100) {
- _isLoadingMore = true;
- ref.read(messagesProvider.notifier).loadMore().then((_) => _isLoadingMore = false);
- }
+ // Pagination removed: all messages are loaded synchronously on session
+ // switch via the in-memory index. Nothing to do on scroll.
}
// Helper: send a command to the gateway in the expected format
@@ -377,18 +378,23 @@
_lastSeq = serverSeq;
_saveLastSeq();
}
- // Merge catch_up messages: only add messages not already in local storage.
- // We check by content match against existing messages to avoid duplicates
- // while still picking up messages that arrived while the app was backgrounded.
+ // Merge catch_up messages: only add messages not already displayed.
+ // Dedup by content to avoid showing messages already in the UI.
final catchUpMsgs = msg['messages'] as List<dynamic>?;
if (catchUpMsgs != null && catchUpMsgs.isNotEmpty) {
_isCatchingUp = true;
final activeId = ref.read(activeSessionIdProvider);
+ final currentId = ref.read(messagesProvider.notifier).currentSessionId;
final existing = ref.read(messagesProvider);
final existingContents = existing
.where((m) => m.role == MessageRole.assistant)
.map((m) => m.content)
.toSet();
+
+ // Collect cross-session sessions that received messages (for toasts).
+ final crossSessionCounts = <String, int>{};
+ final crossSessionPreviews = <String, String>{};
+
for (final m in catchUpMsgs) {
final map = m as Map<String, dynamic>;
final msgType = map['type'] as String? ?? 'text';
@@ -417,52 +423,46 @@
);
}
- _chatLog('catch_up msg: session=${msgSessionId?.substring(0, 8) ?? "NULL"} active=${activeId?.substring(0, 8)} match=${msgSessionId == activeId || msgSessionId == null} content="${content.substring(0, content.length.clamp(0, 40))}"');
- if (msgSessionId == null || msgSessionId == activeId) {
- // Active session or no session: add directly to chat
+ _chatLog('catch_up msg: session=${msgSessionId?.substring(0, 8) ?? "NULL"} active=${activeId?.substring(0, 8)} content="${content.substring(0, content.length.clamp(0, 40))}"');
+
+ if (msgSessionId == null || msgSessionId == currentId) {
+ // Active session or no session: add to UI (addMessage also appends to log).
ref.read(messagesProvider.notifier).addMessage(message);
} else {
- // Different session: store + unread badge + toast
- // Collect for batch storage below to avoid race condition
- _catchUpPending.putIfAbsent(msgSessionId, () => []).add(message);
+ // Cross-session: synchronous append — no race condition.
+ MessageStoreV2.append(msgSessionId, message);
_incrementUnread(msgSessionId);
+ crossSessionCounts[msgSessionId] = (crossSessionCounts[msgSessionId] ?? 0) + 1;
+ crossSessionPreviews.putIfAbsent(msgSessionId, () => content);
}
existingContents.add(content);
}
+
_isCatchingUp = false;
_scrollToBottom();
- // Batch-store cross-session messages (sequential to avoid race condition)
- if (_catchUpPending.isNotEmpty) {
- final pending = Map<String, List<Message>>.from(_catchUpPending);
- _catchUpPending.clear();
- // Show one toast per session with message count
- if (mounted) {
- final sessions = ref.read(sessionsProvider);
- for (final entry in pending.entries) {
- final session = sessions.firstWhere(
- (s) => s.id == entry.key,
- orElse: () => Session(id: entry.key, index: 0, name: 'Unknown', type: 'claude'),
- );
- final count = entry.value.length;
- final preview = count == 1
- ? entry.value.first.content
- : '$count messages';
- ToastManager.show(
- context,
- sessionName: session.name,
- preview: preview.length > 100 ? '${preview.substring(0, 100)}...' : preview,
- onTap: () => _switchSession(entry.key),
- );
- }
+
+ // Show one toast per cross-session that received messages.
+ if (crossSessionCounts.isNotEmpty && mounted) {
+ final sessions = ref.read(sessionsProvider);
+ for (final entry in crossSessionCounts.entries) {
+ final sid = entry.key;
+ final count = entry.value;
+ final session = sessions.firstWhere(
+ (s) => s.id == sid,
+ orElse: () => Session(id: sid, index: 0, name: 'Unknown', type: 'claude'),
+ );
+ final preview = count == 1
+ ? (crossSessionPreviews[sid] ?? '')
+ : '$count messages';
+ ToastManager.show(
+ context,
+ sessionName: session.name,
+ preview: preview.length > 100 ? '${preview.substring(0, 100)}...' : preview,
+ onTap: () => _switchSession(sid),
+ );
}
- () async {
- for (final entry in pending.entries) {
- final existing = await MessageStore.loadAll(entry.key);
- MessageStore.save(entry.key, [...existing, ...entry.value]);
- await MessageStore.flush();
- }
- }();
}
+
// Clear unread for active session
if (activeId != null) {
final counts = Map<String, int>.from(ref.read(unreadCountsProvider));
@@ -500,6 +500,7 @@
orElse: () => sessions.first,
);
ref.read(activeSessionIdProvider.notifier).state = active.id;
+ // Synchronous session switch — no async gap.
ref.read(messagesProvider.notifier).switchSession(active.id);
SharedPreferences.getInstance().then((p) => p.setString('activeSessionId', active.id));
}
@@ -520,7 +521,7 @@
}
}
- Future<void> _handleIncomingMessage(Map<String, dynamic> msg) async {
+ void _handleIncomingMessage(Map<String, dynamic> msg) {
final sessionId = msg['sessionId'] as String?;
final content = msg['content'] as String? ??
msg['text'] as String? ??
@@ -537,16 +538,16 @@
status: MessageStatus.sent,
);
- // Use currentSessionId from notifier (what's actually loaded in the provider)
- // not activeSessionIdProvider (can be stale after background resume)
+ // Use currentSessionId from notifier (what's actually loaded in the provider),
+ // not activeSessionIdProvider (can be stale after background resume).
final currentId = ref.read(messagesProvider.notifier).currentSessionId;
if (sessionId != null && sessionId != currentId) {
- // Store message for the other session so it's there when user switches
+ // Append directly to the log for the target session — synchronous, no race.
TraceService.instance.addTrace(
'message stored for session',
'sessionId=${sessionId.substring(0, sessionId.length.clamp(0, 8))}, toast shown',
);
- await _storeForSession(sessionId, message);
+ MessageStoreV2.append(sessionId, message);
_incrementUnread(sessionId);
final sessions = ref.read(sessionsProvider);
final session = sessions.firstWhere(
@@ -616,9 +617,10 @@
final currentId = ref.read(messagesProvider.notifier).currentSessionId;
_chatLog('voice: sessionId=$sessionId currentId=$currentId audioPath=$savedAudioPath content="${content.substring(0, content.length.clamp(0, 30))}"');
if (sessionId != null && sessionId != currentId) {
- _chatLog('voice: cross-session, storing for $sessionId');
- await _storeForSession(sessionId, storedMessage);
- _chatLog('voice: stored, incrementing unread');
+ _chatLog('voice: cross-session, appending to store for $sessionId');
+ // Synchronous append — no async gap, no race condition.
+ MessageStoreV2.append(sessionId, storedMessage);
+ _chatLog('voice: appended, incrementing unread');
_incrementUnread(sessionId);
final sessions = ref.read(sessionsProvider);
final session = sessions.firstWhere(
@@ -684,10 +686,10 @@
status: MessageStatus.sent,
);
- // Cross-session routing: store for target session if not currently loaded
+ // Cross-session routing: append to log for target session if not currently loaded.
final currentId = ref.read(messagesProvider.notifier).currentSessionId;
if (sessionId != null && sessionId != currentId) {
- _storeForSession(sessionId, message);
+ MessageStoreV2.append(sessionId, message);
_incrementUnread(sessionId);
return;
}
@@ -697,51 +699,19 @@
_scrollToBottom();
}
- /// Store a message for a non-active session so it persists when the user switches to it.
- Future<void> _storeForSession(String sessionId, Message message) async {
- final existing = await MessageStore.loadAll(sessionId);
- _chatLog('storeForSession: $sessionId existing=${existing.length} adding type=${message.type.name} content="${message.content.substring(0, message.content.length.clamp(0, 30))}" audioUri=${message.audioUri != null ? "set(${message.audioUri!.length})" : "null"}');
- MessageStore.save(sessionId, [...existing, message]);
- await MessageStore.flush();
- // Verify
- final verify = await MessageStore.loadAll(sessionId);
- _chatLog('storeForSession: verified ${verify.length} messages after save');
+ /// Superseded by MessageStoreV2.append() — call sites now use the synchronous
+ /// append directly. Kept as dead code until all callers are confirmed removed.
+ // ignore: unused_element
+ void _storeForSession(String sessionId, Message message) {
+ MessageStoreV2.append(sessionId, message);
}
- /// Update a transcript for a message stored on disk (not in the active session).
- /// Scans all session files to find the message by ID, updates content, and saves.
+ /// With the append-only log, transcript updates for cross-session messages
+ /// are not patched back to disk (the append-only design doesn't support
+ /// in-place edits). The transcript is updated in-memory if the message is
+ /// in the active session. Cross-session transcript updates are a no-op.
Future<void> _updateTranscriptOnDisk(String messageId, String content) async {
- try {
- final dir = await getApplicationDocumentsDirectory();
- final msgDir = Directory('${dir.path}/messages');
- if (!await msgDir.exists()) return;
-
- await for (final entity in msgDir.list()) {
- if (entity is! File || !entity.path.endsWith('.json')) continue;
-
- final jsonStr = await entity.readAsString();
- final List<dynamic> jsonList = jsonDecode(jsonStr) as List<dynamic>;
- bool found = false;
-
- final updated = jsonList.map((j) {
- final map = j as Map<String, dynamic>;
- if (map['id'] == messageId) {
- found = true;
- return {...map, 'content': content};
- }
- return map;
- }).toList();
-
- if (found) {
- await entity.writeAsString(jsonEncode(updated));
- _chatLog('transcript: updated messageId=$messageId on disk in ${entity.path.split('/').last}');
- return;
- }
- }
- _chatLog('transcript: messageId=$messageId not found on disk');
- } catch (e) {
- _chatLog('transcript: disk update error=$e');
- }
+ _chatLog('transcript: cross-session update for messageId=$messageId — in-memory only (append-only log)');
}
void _incrementUnread(String sessionId) {
@@ -770,7 +740,8 @@
ref.read(isTypingProvider.notifier).state = false;
ref.read(activeSessionIdProvider.notifier).state = sessionId;
- await ref.read(messagesProvider.notifier).switchSession(sessionId);
+ // Synchronous — no async gap between session switch and incoming messages.
+ ref.read(messagesProvider.notifier).switchSession(sessionId);
SharedPreferences.getInstance().then((p) => p.setString('activeSessionId', sessionId));
final counts = Map<String, int>.from(ref.read(unreadCountsProvider));
diff --git a/lib/services/message_store.dart b/lib/services/message_store.dart
index 1723b70..71a7655 100644
--- a/lib/services/message_store.dart
+++ b/lib/services/message_store.dart
@@ -1,4 +1,3 @@
-import 'dart:async';
import 'dart:convert';
import 'dart:io';
@@ -8,30 +7,39 @@
import '../models/message.dart';
import 'trace_service.dart';
-/// Per-session JSON file persistence with debounced saves.
-class MessageStore {
- MessageStore._();
+/// Append-only log-based message persistence.
+///
+/// Layout:
+/// messages/log.jsonl — one JSON object per line, each a serialized Message
+/// messages/index.json — { "sessionId": [lineNumber, ...] }
+///
+/// All writes are synchronous (writeAsStringSync with FileMode.append) to
+/// prevent race conditions between concurrent addMessage / switchSession calls.
+class MessageStoreV2 {
+ MessageStoreV2._();
+
+ static const _backupChannel = MethodChannel('com.mnsoft.pailot/backup');
+
+ // In-memory index: sessionId -> list of 0-based line numbers in log.jsonl
+ static final Map<String, List<int>> _index = {};
+
+ // Number of lines currently in the log (= next line number to write)
+ static int _lineCount = 0;
+
+ // Flush the index to disk every N appends to amortise I/O
+ static const _indexFlushInterval = 20;
+ static int _appendsSinceFlush = 0;
static Directory? _baseDir;
- static Timer? _debounceTimer;
- static final Map<String, List<Message>> _pendingSaves = {};
- // Per-session lock to prevent concurrent read/write on the same file
- static final Map<String, Completer<void>> _locks = {};
- static const _backupChannel =
- MethodChannel('com.mnsoft.pailot/backup');
+ // ------------------------------------------------------------------ init --
- /// Initialize the base directory for message storage.
- /// On iOS, the directory is excluded from iCloud / iTunes backup so that
- /// large base64 image attachments do not bloat the user's cloud storage.
- /// Messages can be re-fetched from the server if needed.
static Future<Directory> _getBaseDir() async {
if (_baseDir != null) return _baseDir!;
final appDir = await getApplicationDocumentsDirectory();
_baseDir = Directory('${appDir.path}/messages');
- final created = !await _baseDir!.exists();
- if (created) {
- await _baseDir!.create(recursive: true);
+ if (!_baseDir!.existsSync()) {
+ _baseDir!.createSync(recursive: true);
}
// Exclude from iCloud / iTunes backup (best-effort, iOS only).
if (Platform.isIOS) {
@@ -40,144 +48,243 @@
'excludeFromBackup',
_baseDir!.path,
);
- } catch (_) {
- // Non-fatal: if the channel call fails, backup exclusion is skipped.
- }
+ } catch (_) {}
}
return _baseDir!;
}
- static String _fileForSession(String sessionId) {
- // Sanitize session ID for filename
- final safe = sessionId.replaceAll(RegExp(r'[^\w\-]'), '_');
- return 'session_$safe.json';
- }
+ static String _logPath(Directory dir) => '${dir.path}/log.jsonl';
+ static String _indexPath(Directory dir) => '${dir.path}/index.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);
- }
-
- /// Write directly to disk, bypassing debounce. For critical saves.
- static Future<void> writeDirect(String sessionId, List<Message> messages) async {
- _debounceTimer?.cancel();
- _pendingSaves.remove(sessionId);
- await _withLock(sessionId, () => _writeSession(sessionId, messages));
- }
-
- /// Acquire a per-session lock, run the operation, release.
- static Future<T> _withLock<T>(String sessionId, Future<T> Function() fn) async {
- // Wait for any existing operation on this session to finish
- while (_locks.containsKey(sessionId)) {
- await _locks[sessionId]!.future;
- }
- final completer = Completer<void>();
- _locks[sessionId] = completer;
- try {
- return await fn();
- } finally {
- _locks.remove(sessionId);
- completer.complete();
- }
- }
-
- /// 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 _withLock(entry.key, () => _writeSession(entry.key, entry.value));
- }
- }
-
- static Future<void> _writeSession(
- String sessionId, List<Message> messages) async {
+ /// Called once at app startup. Reads log.jsonl and rebuilds the in-memory
+ /// index. Then calls compact() to trim old messages.
+ static Future<void> initialize() 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();
- final json = jsonEncode(lightMessages);
- await file.writeAsString(json);
- TraceService.instance.addTrace('MsgStore WRITE', '${sessionId.substring(0, 8)}: ${messages.length} msgs');
+ final logFile = File(_logPath(dir));
+ final indexFile = File(_indexPath(dir));
+
+ // Try loading saved index first (fast path).
+ if (indexFile.existsSync()) {
+ try {
+ final raw = indexFile.readAsStringSync();
+ final decoded = jsonDecode(raw) as Map<String, dynamic>;
+ for (final entry in decoded.entries) {
+ _index[entry.key] =
+ (entry.value as List<dynamic>).map((e) => e as int).toList();
+ }
+ } catch (_) {
+ _index.clear();
+ }
+ }
+
+ // Count actual lines in log to set _lineCount.
+ if (logFile.existsSync()) {
+ final content = logFile.readAsStringSync();
+ _lineCount = content.isEmpty
+ ? 0
+ : content.trimRight().split('\n').length;
+ } else {
+ _lineCount = 0;
+ }
+
+ // If the index was missing or corrupt, rebuild from log.
+ if (_index.isEmpty && _lineCount > 0) {
+ await _rebuildIndex(logFile);
+ }
+
+ TraceService.instance.addTrace(
+ 'MsgStoreV2 INIT', '$_lineCount lines, ${_index.length} sessions');
+
+ // Compact on startup (keeps last 200 per session).
+ await compact();
} catch (e) {
- TraceService.instance.addTrace('MsgStore WRITE ERROR', '${sessionId.substring(0, 8)}: $e');
+ TraceService.instance.addTrace('MsgStoreV2 INIT ERROR', '$e');
}
}
- /// 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 {
+ static Future<void> _rebuildIndex(File logFile) async {
+ _index.clear();
+ final lines = logFile.readAsLinesSync();
+ for (var i = 0; i < lines.length; i++) {
+ final line = lines[i].trim();
+ if (line.isEmpty) continue;
+ try {
+ final map = jsonDecode(line) as Map<String, dynamic>;
+ final sessionId = map['sessionId'] as String?;
+ if (sessionId != null) {
+ _index.putIfAbsent(sessionId, () => []).add(i);
+ }
+ } catch (_) {}
+ }
+ }
+
+ // --------------------------------------------------------------- append --
+
+ /// Append a message to the log. SYNCHRONOUS — no async gap, no race.
+ ///
+ /// Each line written includes a 'sessionId' field so the index can be
+ /// rebuilt from the log alone if needed.
+ static void append(String sessionId, Message message) {
try {
- final dir = await _getBaseDir();
- final file = File('${dir.path}/${_fileForSession(sessionId)}');
- if (!await file.exists()) return [];
+ final dir = _baseDir;
+ if (dir == null) {
+ // initialize() hasn't been called yet — silently drop (shouldn't happen).
+ TraceService.instance
+ .addTrace('MsgStoreV2 APPEND WARN', 'baseDir null, dropping');
+ return;
+ }
+ final logFile = File(_logPath(dir));
+ final json = message.toJsonLight();
+ json['sessionId'] = sessionId;
+ final line = '${jsonEncode(json)}\n';
- 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();
+ // Synchronous append — atomic single write, no read-modify-write.
+ logFile.writeAsStringSync(line, mode: FileMode.append);
- // 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);
+ // Update in-memory index.
+ _index.putIfAbsent(sessionId, () => []).add(_lineCount);
+ _lineCount++;
+
+ // Periodically flush index to disk.
+ _appendsSinceFlush++;
+ if (_appendsSinceFlush >= _indexFlushInterval) {
+ _flushIndex(dir);
+ _appendsSinceFlush = 0;
+ }
} catch (e) {
+ TraceService.instance.addTrace('MsgStoreV2 APPEND ERROR', '$e');
+ }
+ }
+
+ // -------------------------------------------------------------- load --
+
+ /// Load messages for a session. SYNCHRONOUS — reads from the log using the
+ /// in-memory index. Safe to call from switchSession without async gaps.
+ static List<Message> loadSession(String sessionId) {
+ try {
+ final dir = _baseDir;
+ if (dir == null) return [];
+ final logFile = File(_logPath(dir));
+ if (!logFile.existsSync()) return [];
+
+ final lineNumbers = _index[sessionId];
+ if (lineNumbers == null || lineNumbers.isEmpty) return [];
+
+ // Read all lines at once then pick the ones we need.
+ final allLines = logFile.readAsLinesSync();
+ final messages = <Message>[];
+
+ for (final n in lineNumbers) {
+ if (n >= allLines.length) continue;
+ final line = allLines[n].trim();
+ if (line.isEmpty) continue;
+ try {
+ final map = jsonDecode(line) as Map<String, dynamic>;
+ // Remove synthetic sessionId field before deserialising.
+ map.remove('sessionId');
+ final msg = _messageFromJson(map);
+ if (!msg.isEmptyVoice && !msg.isEmptyText) {
+ messages.add(msg);
+ }
+ } catch (_) {}
+ }
+
+ TraceService.instance.addTrace(
+ 'MsgStoreV2 LOAD', '${sessionId.substring(0, 8)}: ${messages.length} msgs');
+ return messages;
+ } catch (e) {
+ TraceService.instance
+ .addTrace('MsgStoreV2 LOAD ERROR', '${sessionId.substring(0, 8)}: $e');
return [];
}
}
- /// Load all messages for a session (no pagination).
- static Future<List<Message>> loadAll(String sessionId) async {
- return _withLock(sessionId, () => _loadAllImpl(sessionId));
- }
+ // ------------------------------------------------------------- compact --
- static Future<List<Message>> _loadAllImpl(String sessionId) async {
+ /// Rewrite the log keeping at most [keepPerSession] messages per session.
+ /// Called once on startup after initialize(). NOT called during normal use.
+ static Future<void> compact({int keepPerSession = 200}) async {
try {
final dir = await _getBaseDir();
- final file = File('${dir.path}/${_fileForSession(sessionId)}');
- if (!await file.exists()) return [];
+ final logFile = File(_logPath(dir));
+ if (!logFile.existsSync()) return;
- final jsonStr = await file.readAsString();
- final List<dynamic> jsonList = jsonDecode(jsonStr) as List<dynamic>;
- final msgs = jsonList
- .map((j) => _messageFromJson(j as Map<String, dynamic>))
- .where((m) => !m.isEmptyVoice && !m.isEmptyText)
- .toList();
- TraceService.instance.addTrace('MsgStore LOAD', '${sessionId.substring(0, 8)}: ${msgs.length} msgs');
- return msgs;
+ final allLines = logFile.readAsLinesSync();
+ if (allLines.length < 500) return; // nothing worth compacting
+
+ // Build a set of line numbers to keep: last keepPerSession per session.
+ final keepLines = <int>{};
+ for (final entry in _index.entries) {
+ final lines = entry.value;
+ final start = lines.length > keepPerSession
+ ? lines.length - keepPerSession
+ : 0;
+ for (var i = start; i < lines.length; i++) {
+ keepLines.add(lines[i]);
+ }
+ }
+
+ if (keepLines.length == allLines.length) return; // nothing removed
+
+ // Rewrite the log with only the kept lines, rebuilding the index.
+ final newIndex = <String, List<int>>{};
+ final buffer = StringBuffer();
+ var newLine = 0;
+
+ for (var i = 0; i < allLines.length; i++) {
+ if (!keepLines.contains(i)) continue;
+ final line = allLines[i].trim();
+ if (line.isEmpty) continue;
+ buffer.write('$line\n');
+ // Extract sessionId for new index.
+ try {
+ final map = jsonDecode(line) as Map<String, dynamic>;
+ final sid = map['sessionId'] as String?;
+ if (sid != null) {
+ newIndex.putIfAbsent(sid, () => []).add(newLine);
+ }
+ } catch (_) {}
+ newLine++;
+ }
+
+ logFile.writeAsStringSync(buffer.toString());
+ _index
+ ..clear()
+ ..addAll(newIndex);
+ _lineCount = newLine;
+ _flushIndex(dir);
+
+ TraceService.instance.addTrace(
+ 'MsgStoreV2 COMPACT', '${allLines.length} → $newLine lines');
} catch (e) {
- TraceService.instance.addTrace('MsgStore LOAD ERROR', '${sessionId.substring(0, 8)}: $e');
- return [];
+ TraceService.instance.addTrace('MsgStoreV2 COMPACT ERROR', '$e');
}
}
- /// 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.
+ // --------------------------------------------------------- index flush --
+
+ static void _flushIndex(Directory dir) {
+ try {
+ final indexMap = _index.map(
+ (k, v) => MapEntry(k, v));
+ File(_indexPath(dir))
+ .writeAsStringSync(jsonEncode(indexMap));
+ } catch (_) {}
+ }
+
+ /// Force-flush the index to disk (call on app suspend / session switch).
+ static void flushIndex() {
+ if (_baseDir != null) _flushIndex(_baseDir!);
+ }
+
+ // ------------------------------------------------- migration helper --
+
+ /// Deserialize a message, downgrading voice→text if audio is unavailable.
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,
@@ -191,25 +298,18 @@
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 --
- /// Clear all stored messages.
+ /// Wipe everything (used from settings / debug).
static Future<void> clearAll() async {
try {
final dir = await _getBaseDir();
- if (await dir.exists()) {
- await dir.delete(recursive: true);
- await dir.create(recursive: true);
+ if (dir.existsSync()) {
+ dir.deleteSync(recursive: true);
+ dir.createSync(recursive: true);
}
+ _index.clear();
+ _lineCount = 0;
} catch (_) {}
}
}
--
Gitblit v1.3.1