fix(#355): merge prefs into drift on migration instead of discarding

The migration's already-present branch called prefs.remove(key) and threw
the prefs copy away when drift already held the key. A build that writes to
SharedPreferences (a non-drift build, or any in-between test build) collects
new messages there, so the next drift run silently gapped that history out.
It cost 574 real channel messages, recovered from backups.

Union the prefs copy into drift by element identity (messageId, else
publicKey, else canonical JSON), keeping drift's live copy on a collision and
appending prefs-only elements. Verify the merged write before removing the
source. New `merged` counter in the report.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

lib/storage/drift/blob_store.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 
 import 'package:flutter/foundation.dart';
 
@@ -176,12 +177,41 @@ class BlobStore {
         }
         final source = raw;
 
-        if (await read(key) != null) {
-          // Already migrated on a previous run. Leave the prefs copy for the
-          // sweep below rather than assuming; the read-back proved the data is
-          // present in drift.
-          report.alreadyPresent++;
+        final existing = await read(key);
+        if (existing != null) {
+          // Drift already holds this key. Do NOT discard the prefs copy: a
+          // build that writes to SharedPreferences (a non-drift build, or any
+          // in-between test build) accumulates NEW messages there, and throwing
+          // them away silently gaps the history across test cycles (#355). Union
+          // the prefs copy into drift by message identity, keeping drift's live
+          // entries, then verify before removing the source.
+          final merged = _mergeBulk(existing, source);
+          if (merged == null) {
+            // Not a JSON list (e.g. a scalar under a bulk prefix): drift's copy
+            // stands, nothing to union. Safe to drop the prefs duplicate.
+            report.alreadyPresent++;
+            await prefs.remove(key);
+            continue;
+          }
+          if (merged == existing) {
+            // Prefs added nothing new; drift is already a superset.
+            report.alreadyPresent++;
+            await prefs.remove(key);
+            continue;
+          }
+          await write(key, merged);
+          final readBack = await read(key);
+          if (readBack != merged) {
+            report.failed++;
+            appLogger.error(
+              'Blob merge FAILED for $key: wrote ${merged.length} chars, '
+              'read back ${readBack?.length ?? "null"}. Prefs copy left intact.',
+              tag: 'Storage',
+            );
+            continue;
+          }
           await prefs.remove(key);
+          report.merged++;
           continue;
         }
 
@@ -216,6 +246,7 @@ class BlobStore {
     final level = report.failed > 0 ? 'WITH FAILURES' : 'ok';
     appLogger.info(
       'Blob migration complete ($level): ${report.migrated} moved, '
+      '${report.merged} merged, '
       '${report.alreadyPresent} already present, ${report.skipped} skipped, '
       '${report.failed} failed, '
       '${(report.bytes / 1024 / 1024).toStringAsFixed(2)} MB',
@@ -230,11 +261,58 @@ class BlobStore {
     }
     return report;
   }
+
+  /// Unions the [prefs] copy of a bulk key into the [drift] copy by element
+  /// identity, keeping drift's entries where both hold the same one.
+  ///
+  /// All bulk families store a JSON list of objects (messages keyed by
+  /// `messageId`, contacts by `publicKey`); identity falls back to the object's
+  /// canonical form so an id-less element is never dropped. Order is drift's
+  /// list followed by the prefs-only elements, which mirrors how the stores
+  /// append on save.
+  ///
+  /// Returns null when either side is not a JSON list of objects (a scalar
+  /// under a bulk prefix), signalling the caller to leave drift's copy as-is.
+  /// Returns the drift JSON unchanged when prefs contributes nothing new.
+  static String? _mergeBulk(String drift, String prefs) {
+    final List<dynamic> driftList;
+    final List<dynamic> prefsList;
+    try {
+      final d = jsonDecode(drift);
+      final p = jsonDecode(prefs);
+      if (d is! List || p is! List) return null;
+      driftList = d;
+      prefsList = p;
+    } catch (_) {
+      return null;
+    }
+
+    String idOf(dynamic e) {
+      if (e is Map) {
+        final id = e['messageId'];
+        if (id is String && id.isNotEmpty) return 'm:$id';
+        final pk = e['publicKey'];
+        if (pk is String && pk.isNotEmpty) return 'p:$pk';
+      }
+      // No stable id: fall back to the element's canonical JSON so distinct
+      // elements stay distinct and true duplicates collapse.
+      return 'j:${jsonEncode(e)}';
+    }
+
+    final seen = <String>{for (final e in driftList) idOf(e)};
+    final result = List<dynamic>.from(driftList);
+    for (final e in prefsList) {
+      if (seen.add(idOf(e))) result.add(e);
+    }
+    if (result.length == driftList.length) return drift;
+    return jsonEncode(result);
+  }
 }
 
 /// Mutable tally of a migration run; surfaced in the log and used by tests.
 class MigrationReport {
   int migrated = 0;
+  int merged = 0;
   int alreadyPresent = 0;
   int skipped = 0;
   int failed = 0;

test/storage/blob_store_migration_test.dart

@@ -75,18 +75,78 @@ void main() {
     expect(await store.read('contacts_dev'), '[{"c":1}]');
   });
 
-  test('a re-added prefs key does not clobber migrated data', () async {
-    // Simulates an older build writing the key again after migration. The
-    // migrated copy must win; the stale prefs copy is discarded, not promoted.
-    await seedPrefs({'contacts_dev': '[{"c":"new"}]'});
-    await store.write('contacts_dev', '[{"c":"migrated"}]');
-
-    final report = await store.migrateFromPrefs();
-
-    expect(report.alreadyPresent, 1);
-    expect(await store.read('contacts_dev'), '[{"c":"migrated"}]');
-    expect(PrefsManager.instance.getString('contacts_dev'), isNull);
-  });
+  test(
+    '#355: prefs messages absent from drift are merged, not discarded',
+    () async {
+      // The bug: an in-between build (non-drift, or any build that writes prefs)
+      // accumulates NEW messages in prefs. On the next drift run the key is
+      // "already present", so the old code discarded the prefs copy and the new
+      // messages were gapped out of history. They must be UNIONED in instead.
+      await store.write(
+        'channel_messages_devpsk_abc',
+        '[{"messageId":"a"},{"messageId":"b"}]',
+      );
+      await seedPrefs({
+        'channel_messages_devpsk_abc':
+            '[{"messageId":"a"},{"messageId":"c"},{"messageId":"d"}]',
+      });
+
+      final report = await store.migrateFromPrefs();
+
+      expect(report.merged, 1);
+      expect(report.failed, 0);
+      final ids = (await store.read('channel_messages_devpsk_abc'))!;
+      // Drift's a,b kept; prefs-only c,d appended; shared a not duplicated.
+      expect(
+        ids,
+        '[{"messageId":"a"},{"messageId":"b"},'
+        '{"messageId":"c"},{"messageId":"d"}]',
+      );
+      expect(
+        PrefsManager.instance.getString('channel_messages_devpsk_abc'),
+        isNull,
+      );
+    },
+  );
+
+  test(
+    'merge keeps the drift copy of a shared entity, adds prefs-only ones',
+    () async {
+      // Contacts collide by publicKey: the live drift copy wins for a shared key,
+      // and a contact seen only on the in-between build is still added.
+      await store.write('contacts_dev', '[{"publicKey":"A","name":"drift"}]');
+      await seedPrefs({
+        'contacts_dev':
+            '[{"publicKey":"A","name":"stale"},{"publicKey":"B","name":"new"}]',
+      });
+
+      final report = await store.migrateFromPrefs();
+
+      expect(report.merged, 1);
+      expect(
+        await store.read('contacts_dev'),
+        '[{"publicKey":"A","name":"drift"},{"publicKey":"B","name":"new"}]',
+      );
+      expect(PrefsManager.instance.getString('contacts_dev'), isNull);
+    },
+  );
+
+  test(
+    'already-present with nothing new to add reports alreadyPresent',
+    () async {
+      // Prefs is a subset of drift: union changes nothing, and the stale prefs
+      // copy is dropped without a needless rewrite.
+      await store.write('contacts_dev', '[{"publicKey":"A"}]');
+      await seedPrefs({'contacts_dev': '[{"publicKey":"A"}]'});
+
+      final report = await store.migrateFromPrefs();
+
+      expect(report.alreadyPresent, 1);
+      expect(report.merged, 0);
+      expect(await store.read('contacts_dev'), '[{"publicKey":"A"}]');
+      expect(PrefsManager.instance.getString('contacts_dev'), isNull);
+    },
+  );
 
   test('preserves a payload larger than the 5 MiB localStorage cap', () async {
     // 4 chars per element, so >1.4M elements clears 5 MiB.