feat: Remove events from the OmemoManager

lib/omemo_dart.dart

@@ -8,9 +8,9 @@ export 'src/omemo/bundle.dart';
 export 'src/omemo/device.dart';
 export 'src/omemo/encrypted_key.dart';
 export 'src/omemo/encryption_result.dart';
-export 'src/omemo/events.dart';
 export 'src/omemo/fingerprint.dart';
 export 'src/omemo/omemo.dart';
+export 'src/omemo/ratchet_data.dart';
 export 'src/omemo/ratchet_map_key.dart';
 export 'src/omemo/stanza.dart';
 export 'src/trust/base.dart';

lib/src/helpers.dart

@@ -83,3 +83,20 @@ extension BeforeAfterListDiff<T> on List<T> {
     );
   }
 }
+
+extension AppendToListOrCreateExtension<K, V> on Map<K, List<V>> {
+  /// Create or append [value] to the list identified with key [key].
+  void appendOrCreate(K key, V value) {
+    if (containsKey(key)) {
+      this[key]!.add(value);
+    } else {
+      this[key] = [value];
+    }
+  }
+}
+
+extension StringFromBase64Extension on String {
+  /// Base64-decode this string. Useful for doing `someString?.fromBase64()` instead
+  /// of `someString != null ? base64Decode(someString) : null`.
+  List<int> fromBase64() => base64Decode(this);
+}

lib/src/omemo/events.dart

@@ -1,59 +0,0 @@
-import 'package:omemo_dart/omemo_dart.dart';
-
-abstract class OmemoEvent {}
-
-/// Triggered when (possibly multiple) ratchets have been created at sending time.
-class RatchetsAddedEvent extends OmemoEvent {
-  RatchetsAddedEvent(this.ratchets);
-
-  /// The mapping of the newly created ratchets.
-  final Map<RatchetMapKey, OmemoDoubleRatchet> ratchets;
-}
-
-/// Triggered when a ratchet has been modified
-class RatchetModifiedEvent extends OmemoEvent {
-  RatchetModifiedEvent(
-    this.jid,
-    this.deviceId,
-    this.ratchet,
-    this.added,
-    this.replaced,
-  );
-  final String jid;
-  final int deviceId;
-  final OmemoDoubleRatchet ratchet;
-
-  /// Indicates whether the ratchet has just been created (true) or just modified (false).
-  final bool added;
-
-  /// Indicates whether the ratchet has been replaced (true) or not.
-  final bool replaced;
-}
-
-/// Triggered when a ratchet has been removed and should be removed from storage.
-class RatchetRemovedEvent extends OmemoEvent {
-  RatchetRemovedEvent(this.jid, this.deviceId);
-  final String jid;
-  final int deviceId;
-}
-
-/// Triggered when the device map has been modified
-class DeviceListModifiedEvent extends OmemoEvent {
-  DeviceListModifiedEvent(this.jid, this.added, this.removed);
-
-  /// The JID of the user.
-  final String jid;
-
-  /// The list of added devices for [jid].
-  final List<int> added;
-
-  /// The list of removed devices for [jid].
-  final List<int> removed;
-}
-
-/// Triggered by the OmemoSessionManager when our own device bundle was modified
-/// and thus should be republished.
-class DeviceModifiedEvent extends OmemoEvent {
-  DeviceModifiedEvent(this.device);
-  final OmemoDevice device;
-}

lib/src/omemo/omemo.dart

@@ -17,9 +17,9 @@ import 'package:omemo_dart/src/omemo/device.dart';
 import 'package:omemo_dart/src/omemo/encrypted_key.dart';
 import 'package:omemo_dart/src/omemo/encryption_result.dart';
 import 'package:omemo_dart/src/omemo/errors.dart';
-import 'package:omemo_dart/src/omemo/events.dart';
 import 'package:omemo_dart/src/omemo/fingerprint.dart';
 import 'package:omemo_dart/src/omemo/queue.dart';
+import 'package:omemo_dart/src/omemo/ratchet_data.dart';
 import 'package:omemo_dart/src/omemo/ratchet_map_key.dart';
 import 'package:omemo_dart/src/omemo/stanza.dart';
 import 'package:omemo_dart/src/protobuf/schema.pb.dart';
@@ -27,22 +27,62 @@ import 'package:omemo_dart/src/trust/base.dart';
 import 'package:omemo_dart/src/x3dh/x3dh.dart';
 import 'package:synchronized/synchronized.dart';
 
-extension AppendToListOrCreateExtension<K, V> on Map<K, List<V>> {
+/// Callback type definitions
-  /// Create or append [value] to the list identified with key [key].
-  void appendOrCreate(K key, V value) {
-    if (containsKey(key)) {
-      this[key]!.add(value);
-    } else {
-      this[key] = [value];
-    }
-  }
-}
 
-extension StringFromBase64Extension on String {
+/// Directly "package" [result] into an OMEMO message and send it to [recipientJid].
-  /// Base64-decode this string. Useful for doing `someString?.fromBase64()` instead
+typedef SendEmptyOmemoMessageFunction = Future<void> Function(
-  /// of `someString != null ? base64Decode(someString) : null`.
+  EncryptionResult result,
-  List<int> fromBase64() => base64Decode(this);
+  String recipientJid,
-}
+);
+
+/// Fetches the device list for [jid]. If no device list could be fetched, returns null.
+typedef FetchDeviceListFunction = Future<List<int>?> Function(String jid);
+
+/// Fetch the device bundle for the device with id @id of jid. If it cannot be fetched, return null.
+typedef FetchDeviceBundleFunction = Future<OmemoBundle?> Function(
+  String jid,
+  int id,
+);
+
+/// Subscribes to the device list node of [jid].
+typedef DeviceListSubscribeFunction = Future<void> Function(String jid);
+
+/// Commits the device list for [jid] to persistent storage. [added] will be the list of
+/// devices added and [removed] will be the list of removed devices.
+typedef CommitDeviceListCallback = Future<void> Function(
+  String jid,
+  List<int> added,
+  List<int> removed,
+);
+
+/// A stub implementation of [CommitDeviceListCallback].
+Future<void> commitDeviceListStub(
+  String _,
+  List<int> __,
+  List<int> ___,
+) async {}
+
+/// Commits the mapping of the (new) ratchets in [ratchets] to persistent storage.
+typedef CommitRatchetsCallback = Future<void> Function(
+  List<OmemoRatchetData> ratchets,
+);
+
+/// A stub implementation of [CommitRatchetsCallback];
+Future<void> commitRatchetsStub(List<OmemoRatchetData> _) async {}
+
+/// Commits the device [device] to persistent storage.
+typedef CommitDeviceCallback = Future<void> Function(OmemoDevice device);
+
+/// A stub implementation of [CommitDeviceCallback].
+Future<void> commitDeviceStub(OmemoDevice device) async {}
+
+/// Removes the ratchets identified by their keys in [ratchets] from persistent storage.
+typedef RemoveRatchetsFunction = Future<void> Function(
+  List<RatchetMapKey> ratchets,
+);
+
+/// A stub implementation of [RemoveRatchetsFunction].
+Future<void> removeRatchetsStub(List<RatchetMapKey> ratchets) async {}
 
 class OmemoManager {
   OmemoManager(
@@ -51,8 +91,12 @@ class OmemoManager {
     this.sendEmptyOmemoMessageImpl,
     this.fetchDeviceListImpl,
     this.fetchDeviceBundleImpl,
-    this.subscribeToDeviceListNodeImpl,
+    this.subscribeToDeviceListNodeImpl, {
-  );
+    this.commitRatchets = commitRatchetsStub,
+    this.commitDeviceList = commitDeviceListStub,
+    this.commitDevice = commitDeviceStub,
+    this.removeRatchets = removeRatchetsStub,
+  });
 
   final Logger _log = Logger('OmemoManager');
 
@@ -60,18 +104,29 @@ class OmemoManager {
 
   /// Send an empty OMEMO:2 message using the encrypted payload @result to
   /// @recipientJid.
-  final Future<void> Function(EncryptionResult result, String recipientJid)
+  final SendEmptyOmemoMessageFunction sendEmptyOmemoMessageImpl;
-      sendEmptyOmemoMessageImpl;
 
   /// Fetch the list of device ids associated with @jid. If the device list cannot be
   /// fetched, return null.
-  final Future<List<int>?> Function(String jid) fetchDeviceListImpl;
+  final FetchDeviceListFunction fetchDeviceListImpl;
 
   /// Fetch the device bundle for the device with id @id of jid. If it cannot be fetched, return null.
-  final Future<OmemoBundle?> Function(String jid, int id) fetchDeviceBundleImpl;
+  final FetchDeviceBundleFunction fetchDeviceBundleImpl;
 
   /// Subscribe to the device list PEP node of @jid.
-  final Future<void> Function(String jid) subscribeToDeviceListNodeImpl;
+  final DeviceListSubscribeFunction subscribeToDeviceListNodeImpl;
+
+  /// Callback to commit the ratchet to persistent storage.
+  final CommitRatchetsCallback commitRatchets;
+
+  /// Callback to commit the device list to persistent storage.
+  final CommitDeviceListCallback commitDeviceList;
+
+  /// Callback to commit the device to persistent storage.
+  final CommitDeviceCallback commitDevice;
+
+  /// Callback to remove ratchets from persistent storage.
+  final RemoveRatchetsFunction removeRatchets;
 
   /// Map bare JID to its known devices
   final Map<String, List<int>> _deviceList = {};
@@ -98,11 +153,6 @@ class OmemoManager {
   // ignore: prefer_final_fields
   OmemoDevice _device;
 
-  /// The event bus of the session manager
-  final StreamController<OmemoEvent> _eventStreamController =
-      StreamController<OmemoEvent>.broadcast();
-  Stream<OmemoEvent> get eventStream => _eventStreamController.stream;
-
   Future<Result<OmemoError, String?>> _decryptAndVerifyHmac(
     List<int>? ciphertext,
     List<int> keyAndHmac,
@@ -153,8 +203,10 @@ class OmemoManager {
         _deviceList[jid] = newDeviceList;
         _deviceListRequested[jid] = true;
 
-        _eventStreamController.add(
+        await commitDeviceList(
-          DeviceListModifiedEvent(jid, newDeviceList, []),
+          jid,
+          newDeviceList,
+          [],
         );
       }
     }
@@ -359,24 +411,21 @@ class OmemoManager {
       // Commit the ratchet
       _ratchetMap[ratchetKey] = ratchet;
       _deviceList.appendOrCreate(stanza.bareSenderJid, stanza.senderDeviceId);
-      _eventStreamController.add(
+      await commitRatchets([
-        RatchetModifiedEvent(
+        OmemoRatchetData(
           stanza.bareSenderJid,
           stanza.senderDeviceId,
           ratchet,
           true,
           false,
         ),
-      );
+      ]);
 
       // Replace the OPK if we're not doing a catchup.
       if (!stanza.isCatchup) {
         await _deviceLock.synchronized(() async {
           await _device.replaceOnetimePrekey(kexMessage.pkId);
-
+          await commitDevice(_device);
-          _eventStreamController.add(
-            DeviceModifiedEvent(_device),
-          );
         });
       }
 
@@ -451,15 +500,15 @@ class OmemoManager {
 
       // Message was successfully decrypted, so commit the ratchet
       _ratchetMap[ratchetKey] = ratchet;
-      _eventStreamController.add(
+      await commitRatchets([
-        RatchetModifiedEvent(
+        OmemoRatchetData(
           stanza.bareSenderJid,
           stanza.senderDeviceId,
           ratchet,
           false,
           false,
         ),
-      );
+      ]);
 
       // Send a heartbeat, if required.
       await _maybeSendEmptyMessage(ratchetKey, false, false);
@@ -563,14 +612,16 @@ class OmemoManager {
 
     // Commit the newly created ratchets, if we created any.
     if (addedRatchetKeys.isNotEmpty) {
-      _eventStreamController.add(
+      await commitRatchets(
-        RatchetsAddedEvent(
+        addedRatchetKeys.map((key) {
-          Map<RatchetMapKey, OmemoDoubleRatchet>.fromEntries(
+          return OmemoRatchetData(
-            addedRatchetKeys
+            key.jid,
-                .map((key) => MapEntry(key, _ratchetMap[key]!))
+            key.deviceId,
-                .toList(),
+            _ratchetMap[key]!,
-          ),
+            true,
-        ),
+            false,
+          );
+        }).toList(),
       );
     }
 
@@ -702,15 +753,20 @@ class OmemoManager {
     await _ratchetQueue.synchronized(
       [jid],
       () async {
-        for (final device in _deviceList[jid] ?? <int>[]) {
+        // Remove the ratchet and commit
-          // Remove the ratchet and commit
+        final keys = (_deviceList[jid] ?? <int>[])
-          _ratchetMap.remove(RatchetMapKey(jid, device));
+            .map((device) => RatchetMapKey(jid, device));
-          _eventStreamController.add(RatchetRemovedEvent(jid, device));
+        for (final key in keys) {
+          _ratchetMap.remove(key);
         }
+        await removeRatchets(keys.toList());
 
         // Clear the device list
-        _eventStreamController
+        await commitDeviceList(
-            .add(DeviceListModifiedEvent(jid, [], _deviceList[jid]!));
+          jid,
+          [],
+          _deviceList[jid]!,
+        );
         _deviceList.remove(jid);
         _deviceListRequested.remove(jid);
       },
@@ -736,9 +792,7 @@ class OmemoManager {
         _deviceListRequested[jid] = true;
 
         // Commit the device list
-        _eventStreamController.add(
+        await commitDeviceList(jid, delta.added, delta.removed);
-          DeviceListModifiedEvent(jid, delta.added, delta.removed),
-        );
       },
     );
   }
@@ -768,9 +822,15 @@ class OmemoManager {
     } else {
       // Commit
       final ratchet = _ratchetMap[ratchetKey]!..acknowledged = true;
-      _eventStreamController.add(
+      await commitRatchets([
-        RatchetModifiedEvent(jid, device, ratchet, false, false),
+        OmemoRatchetData(
-      );
+          jid,
+          device,
+          ratchet,
+          false,
+          false,
+        ),
+      ]);
     }
   }
 

lib/src/omemo/ratchet_data.dart

@@ -0,0 +1,26 @@
+import 'package:omemo_dart/src/double_ratchet/double_ratchet.dart';
+
+class OmemoRatchetData {
+  const OmemoRatchetData(
+    this.jid,
+    this.id,
+    this.ratchet,
+    this.added,
+    this.replaced,
+  );
+
+  /// The JID we have the ratchet with.
+  final String jid;
+
+  /// The device id we have the ratchet with.
+  final int id;
+
+  /// The actual double ratchet to commit.
+  final OmemoDoubleRatchet ratchet;
+
+  /// Indicates whether the ratchet has just been created (true) or just modified (false).
+  final bool added;
+
+  /// Indicates whether the ratchet has been replaced (true) or not.
+  final bool replaced;
+}