diff --git a/lib/storage/channel_message_store.dart b/lib/storage/channel_message_store.dart index 842a2d5..b80016f 100644 --- a/lib/storage/channel_message_store.dart +++ b/lib/storage/channel_message_store.dart @@ -73,43 +73,49 @@ class ChannelMessageStore { // Deletion has its own path (removeChannelMessage) so this never // resurrects a message the user deleted. final key = _storageKey(channelIndex); - final byKey = {}; - - final existing = await BlobStore.instance.readWithPrefsFallback(key); - if (existing != null && existing.isNotEmpty) { - try { - for (final e in jsonDecode(existing) as List) { - final m = _messageFromJson(e as Map); - byKey[_mergeKey(m)] = m; + final blobs = BlobStore.instance; + // Serialise the whole read-modify-write against other saves/deletes on this + // key so two concurrent saves cannot clobber each other (Gemini review). + await blobs.synchronized(key, () async { + final byKey = {}; + final existing = await blobs.readWithPrefsFallback(key); + if (existing != null && existing.isNotEmpty) { + try { + for (final e in jsonDecode(existing) as List) { + final m = _messageFromJson(e as Map); + byKey[_mergeKey(m)] = m; + } + } catch (e) { + // SAFELANE 6: never merge into an empty base and truncate silently. + appLogger.error( + 'Failed to decode existing channel $channelIndex history before ' + 'merge; aborting save to avoid truncation: $e', + tag: 'Storage', + ); + return; } - } catch (e) { - // SAFELANE 6: a decode failure here must be loud, not silently drop the - // persisted history by merging into an empty base. - appLogger.error( - 'Failed to decode existing channel $channelIndex history before ' - 'merge; aborting save to avoid truncation: $e', - tag: 'Storage', - ); - return; } - } - for (final m in messages) { - byKey[_mergeKey(m)] = m; - } - - final merged = byKey.values.toList() - ..sort((a, b) => a.timestamp.compareTo(b.timestamp)); - await BlobStore.instance.write( - key, - jsonEncode(merged.map(_messageToJson).toList()), - ); + for (final m in messages) { + byKey[_mergeKey(m)] = m; + } + final merged = byKey.values.toList() + ..sort((a, b) => a.timestamp.compareTo(b.timestamp)); + await blobs.write(key, jsonEncode(merged.map(_messageToJson).toList())); + }); } /// Stable identity for merge/dedupe. messageId when present, else a composite /// that distinguishes distinct messages that share no id. String _mergeKey(ChannelMessage m) { if (m.messageId.isNotEmpty) return 'id:${m.messageId}'; - return 'x:${m.packetHash ?? ''}:${m.timestamp.millisecondsSinceEpoch}:${m.text}'; + // Include the sender: two different senders can post identical text at the + // same timestamp without a messageId, and would otherwise collide and lose + // one (Gemini review, 2026-07-20). + final sender = m.senderKey == null + ? '' + : m.senderKey!.map((b) => b.toRadixString(16)).join(); + return 'x:$sender:${m.packetHash ?? ''}:' + '${m.timestamp.millisecondsSinceEpoch}:${m.text}'; } /// Removes a single message from the persisted history. The explicit delete @@ -121,28 +127,27 @@ class ChannelMessageStore { ) async { if (publicKeyHex.isEmpty) return; final key = _storageKey(channelIndex); - final existing = await BlobStore.instance.readWithPrefsFallback(key); - if (existing == null || existing.isEmpty) return; - - final List raw; - try { - raw = jsonDecode(existing) as List; - } catch (e) { - appLogger.error( - 'Failed to decode channel $channelIndex history for delete: $e', - tag: 'Storage', - ); - return; - } - final target = _mergeKey(message); - final kept = raw - .map((e) => _messageFromJson(e as Map)) - .where((m) => _mergeKey(m) != target) - .toList(); - await BlobStore.instance.write( - key, - jsonEncode(kept.map(_messageToJson).toList()), - ); + final blobs = BlobStore.instance; + await blobs.synchronized(key, () async { + final existing = await blobs.readWithPrefsFallback(key); + if (existing == null || existing.isEmpty) return; + final List raw; + try { + raw = jsonDecode(existing) as List; + } catch (e) { + appLogger.error( + 'Failed to decode channel $channelIndex history for delete: $e', + tag: 'Storage', + ); + return; + } + final target = _mergeKey(message); + final kept = raw + .map((e) => _messageFromJson(e as Map)) + .where((m) => _mergeKey(m) != target) + .toList(); + await blobs.write(key, jsonEncode(kept.map(_messageToJson).toList())); + }); } /// Load messages for a specific channel @@ -177,9 +182,30 @@ class ChannelMessageStore { appLogger.info( 'Migrating channel messages $legacyKey -> $key (PSK-keyed, #194)', ); - await blobs.write(key, legacy); + // Under the key lock, and MERGE rather than overwrite: a save may + // have landed on the PSK key between the read above and here, and a + // blind write would clobber it (Gemini review). Union keeps both the + // adopted legacy history and any freshly-saved message. + jsonString = await blobs.synchronized(key, () async { + final byKey = {}; + for (final srcJson in [await blobs.read(key), legacy]) { + if (srcJson == null || srcJson.isEmpty) continue; + try { + for (final e in jsonDecode(srcJson) as List) { + final m = _messageFromJson(e as Map); + byKey[_mergeKey(m)] = m; + } + } catch (_) { + // Skip an undecodable source rather than aborting the adoption. + } + } + final merged = byKey.values.toList() + ..sort((a, b) => a.timestamp.compareTo(b.timestamp)); + final encoded = jsonEncode(merged.map(_messageToJson).toList()); + await blobs.write(key, encoded); + return encoded; + }); await blobs.deleteEverywhere(legacyKey); - jsonString = legacy; break; } } diff --git a/lib/storage/drift/blob_store.dart b/lib/storage/drift/blob_store.dart index 8239c72..c8ccf9f 100644 --- a/lib/storage/drift/blob_store.dart +++ b/lib/storage/drift/blob_store.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter/foundation.dart'; import '../../utils/app_logger.dart'; @@ -46,6 +48,24 @@ class BlobStore { static bool isBulkKey(String key) => migratedPrefixes.any(key.startsWith); + /// Per-key operation chain. Merge-on-save and the legacy-key migration are + /// read-modify-write sequences with an await gap; two of them racing on the + /// same key would let the second clobber the first and silently lose + /// messages (Gemini review, 2026-07-20). Every RMW on a key runs through + /// [synchronized], which serialises operations per key while leaving + /// different keys concurrent. + final Map> _keyChains = {}; + + /// Serialises [action] against other synchronized actions on the same [key]. + Future synchronized(String key, Future Function() action) { + final prior = _keyChains[key] ?? Future.value(); + final result = prior.then((_) => action()); + // Next op waits for this one; swallow errors so one failure does not wedge + // the chain for the key. + _keyChains[key] = result.then((_) {}, onError: (_) {}); + return result; + } + /// Reads a bulk key, falling back to SharedPreferences if drift does not /// have it. /// diff --git a/lib/storage/drift/offband_database.dart b/lib/storage/drift/offband_database.dart index 2d14954..2d7d209 100644 --- a/lib/storage/drift/offband_database.dart +++ b/lib/storage/drift/offband_database.dart @@ -76,26 +76,59 @@ class OffbandDatabase extends _$OffbandDatabase { final docs = await getApplicationDocumentsDirectory(); final oldPath = p.join(docs.path, '$_dbName.sqlite'); if (File(oldPath).existsSync() && oldPath != newPath) { - for (final suffix in ['', '-wal', '-shm']) { - final src = File('$oldPath$suffix'); - if (src.existsSync()) { - src.copySync('$newPath$suffix'); - src.deleteSync(); + // Copy ALL files first, verify, and only then delete the sources. + // Deleting per-file mid-loop (Gemini review) could strand the main + // .sqlite in one place and its -wal in another if a later copy + // failed, corrupting the DB. Copy-all-then-delete-all makes a + // partial failure recoverable: the original stays intact and the + // half-written destination is cleaned up. + final suffixes = ['', '-wal', '-shm']; + final present = suffixes + .where((s) => File('$oldPath$s').existsSync()) + .toList(); + try { + for (final s in present) { + File('$oldPath$s').copySync('$newPath$s'); } + // Verify every copy exists with a matching size before deleting. + for (final s in present) { + final src = File('$oldPath$s'); + final dst = File('$newPath$s'); + if (!dst.existsSync() || dst.lengthSync() != src.lengthSync()) { + throw StateError('copy verification failed for $newPath$s'); + } + } + for (final s in present) { + File('$oldPath$s').deleteSync(); + } + appLogger.info( + 'Relocated drift DB out of the documents dir (OneDrive on ' + 'Windows) into the app-support dir: $oldPath -> $newPath', + tag: 'Storage', + ); + } catch (e) { + // Roll back any partial destination so drift does not open a + // half-copied DB; the original in the documents dir is untouched. + for (final s in present) { + final dst = File('$newPath$s'); + if (dst.existsSync()) { + try { + dst.deleteSync(); + } catch (_) {} + } + } + rethrow; } - appLogger.info( - 'Relocated drift DB out of the documents dir (OneDrive on ' - 'Windows) into the app-support dir: $oldPath -> $newPath', - tag: 'Storage', - ); } } catch (e) { - // Relocation is best-effort: if it fails, drift opens a fresh DB at the - // correct location and the migration re-runs from SharedPreferences. - // Loud, never silent (SAFELANE 6). + // Relocation is best-effort. On failure the ORIGINAL DB is left intact + // in the documents dir (the destination was rolled back), so no data is + // lost - drift opens a fresh DB at the support location and the user's + // history remains recoverable from the old path. Loud, never silent. appLogger.error( - 'Failed to relocate the drift DB from the documents dir: $e. ' - 'A fresh DB will be created at the app-support location.', + 'Failed to relocate the drift DB from the documents dir: $e. The ' + 'original at the old path is intact; a fresh DB will open at the ' + 'app-support location. Recover the old DB manually if needed.', tag: 'Storage', ); } diff --git a/lib/storage/message_store.dart b/lib/storage/message_store.dart index b912d0e..37b0575 100644 --- a/lib/storage/message_store.dart +++ b/lib/storage/message_store.dart @@ -29,34 +29,32 @@ class MessageStore { // truncate the store. Upsert by identity; deletion is explicit // (removeMessage). final key = '$keyFor$contactKeyHex'; - final byKey = {}; - - final existing = await BlobStore.instance.readWithPrefsFallback(key); - if (existing != null && existing.isNotEmpty) { - try { - for (final e in jsonDecode(existing) as List) { - final m = _messageFromJson(e as Map); - byKey[_mergeKey(m)] = m; + final blobs = BlobStore.instance; + await blobs.synchronized(key, () async { + final byKey = {}; + final existing = await blobs.readWithPrefsFallback(key); + if (existing != null && existing.isNotEmpty) { + try { + for (final e in jsonDecode(existing) as List) { + final m = _messageFromJson(e as Map); + byKey[_mergeKey(m)] = m; + } + } catch (e) { + appLogger.error( + 'Failed to decode existing DM history before merge; aborting save ' + 'to avoid truncation: $e', + tag: 'Storage', + ); + return; } - } catch (e) { - appLogger.error( - 'Failed to decode existing DM history before merge; aborting save ' - 'to avoid truncation: $e', - tag: 'Storage', - ); - return; } - } - for (final m in messages) { - byKey[_mergeKey(m)] = m; - } - - final merged = byKey.values.toList() - ..sort((a, b) => a.timestamp.compareTo(b.timestamp)); - await BlobStore.instance.write( - key, - jsonEncode(merged.map(_messageToJson).toList()), - ); + for (final m in messages) { + byKey[_mergeKey(m)] = m; + } + final merged = byKey.values.toList() + ..sort((a, b) => a.timestamp.compareTo(b.timestamp)); + await blobs.write(key, jsonEncode(merged.map(_messageToJson).toList())); + }); } String _mergeKey(Message m) { diff --git a/test/storage/channel_message_merge_test.dart b/test/storage/channel_message_merge_test.dart index 83f6144..3cba6e2 100644 --- a/test/storage/channel_message_merge_test.dart +++ b/test/storage/channel_message_merge_test.dart @@ -79,6 +79,25 @@ void main() { expect(all.single.reactions['thumbsup'], 2); }); + test( + 'concurrent saves to one channel do not lose messages (Gemini)', + () async { + await store.saveChannelMessages(0, [msg(0)]); + // Fire two saves without awaiting between them: they race on the same key. + // Serialisation must make the result the union, not last-writer-wins. + final a = store.saveChannelMessages(0, [msg(1)]); + final b = store.saveChannelMessages(0, [msg(2)]); + await Future.wait([a, b]); + final all = await store.loadChannelMessages(0); + expect( + all.map((m) => m.text), + containsAll(['m0', 'm1', 'm2']), + reason: 'neither concurrent save may clobber the other', + ); + expect(all, hasLength(3)); + }, + ); + test('deleting a message does NOT resurrect it on the next save', () async { await store.saveChannelMessages(0, [for (var i = 0; i < 5; i++) msg(i)]);