Merge pull request 'Roster Rework' (#15) from fix/roster-rework into master

Reviewed-on: https://codeberg.org/moxxy/moxxmpp/pulls/15

packages/moxxmpp/lib/moxxmpp.dart

@@ -29,6 +29,7 @@ export 'package:moxxmpp/src/rfcs/rfc_2782.dart';
 export 'package:moxxmpp/src/rfcs/rfc_4790.dart';
 export 'package:moxxmpp/src/roster/errors.dart';
 export 'package:moxxmpp/src/roster/roster.dart';
+export 'package:moxxmpp/src/roster/state.dart';
 export 'package:moxxmpp/src/settings.dart';
 export 'package:moxxmpp/src/socket.dart';
 export 'package:moxxmpp/src/stanza.dart';

packages/moxxmpp/lib/src/connection.dart

@@ -94,6 +94,7 @@ class XmppConnection {
     _awaitingResponseLock = Lock(),
     _xmppManagers = {},
     _incomingStanzaHandlers = List.empty(growable: true),
+    _incomingPreStanzaHandlers = List.empty(growable: true),
     _outgoingPreStanzaHandlers = List.empty(growable: true),
     _outgoingPostStanzaHandlers = List.empty(growable: true),
     _reconnectionPolicy = reconnectionPolicy,
@@ -133,6 +134,7 @@ class XmppConnection {
   /// Helpers
   ///
   final List<StanzaHandler> _incomingStanzaHandlers;
+  final List<StanzaHandler> _incomingPreStanzaHandlers;
   final List<StanzaHandler> _outgoingPreStanzaHandlers;
   final List<StanzaHandler> _outgoingPostStanzaHandlers;
   final StreamController<XmppEvent> _eventStreamController;
@@ -223,11 +225,13 @@ class XmppConnection {
     }
 
     _incomingStanzaHandlers.addAll(manager.getIncomingStanzaHandlers());
+    _incomingPreStanzaHandlers.addAll(manager.getIncomingPreStanzaHandlers());
     _outgoingPreStanzaHandlers.addAll(manager.getOutgoingPreStanzaHandlers());
     _outgoingPostStanzaHandlers.addAll(manager.getOutgoingPostStanzaHandlers());
     
     if (sortHandlers) {
       _incomingStanzaHandlers.sort(stanzaHandlerSortComparator);
+      _incomingPreStanzaHandlers.sort(stanzaHandlerSortComparator);
       _outgoingPreStanzaHandlers.sort(stanzaHandlerSortComparator);
       _outgoingPostStanzaHandlers.sort(stanzaHandlerSortComparator);
     }
@@ -630,8 +634,12 @@ class XmppConnection {
     return state;
   }
 
-  Future<StanzaHandlerData> _runIncomingStanzaHandlers(Stanza stanza) async {
-    return _runStanzaHandlers(_incomingStanzaHandlers, stanza);
+  Future<StanzaHandlerData> _runIncomingStanzaHandlers(Stanza stanza, { StanzaHandlerData? initial }) async {
+    return _runStanzaHandlers(_incomingStanzaHandlers, stanza, initial: initial);
+  }
+
+  Future<StanzaHandlerData> _runIncomingPreStanzaHandlers(Stanza stanza) async {
+    return _runStanzaHandlers(_incomingPreStanzaHandlers, stanza);
   }
 
   Future<StanzaHandlerData> _runOutgoingPreStanzaHandlers(Stanza stanza, { StanzaHandlerData? initial }) async {
@@ -673,18 +681,18 @@ class XmppConnection {
 
     // Run the incoming stanza handlers and bounce with an error if no manager handled
     // it.
-    final incomingHandlers = await _runIncomingStanzaHandlers(stanza);
-    final prefix = incomingHandlers.encrypted ?
+    final incomingPreHandlers = await _runIncomingPreStanzaHandlers(stanza);
+    final prefix = incomingPreHandlers.encrypted && incomingPreHandlers.other['encryption_error'] == null ?
       '(Encrypted) ' :
       '';
-    _log.finest('<== $prefix${incomingHandlers.stanza.toXml()}');
+    _log.finest('<== $prefix${incomingPreHandlers.stanza.toXml()}');
 
     // See if we are waiting for this stanza
-    final id = stanza.attributes['id'] as String?;
+    final id = incomingPreHandlers.stanza.attributes['id'] as String?;
     var awaited = false;
     await _awaitingResponseLock.synchronized(() async {
       if (id != null && _awaitingResponse.containsKey(id)) {
-        _awaitingResponse[id]!.complete(incomingHandlers.stanza);
+        _awaitingResponse[id]!.complete(incomingPreHandlers.stanza);
         _awaitingResponse.remove(id);
         awaited = true;
       }
@@ -695,8 +703,19 @@ class XmppConnection {
     }
 
     // Only bounce if the stanza has neither been awaited, nor handled.
+    final incomingHandlers = await _runIncomingStanzaHandlers(
+      incomingPreHandlers.stanza,
+      initial: StanzaHandlerData(
+        false,
+        incomingPreHandlers.cancel,
+        incomingPreHandlers.cancelReason,
+        incomingPreHandlers.stanza,
+        encrypted: incomingPreHandlers.encrypted,
+        other: incomingPreHandlers.other,
+      ),
+    );
     if (!incomingHandlers.done) {
-      handleUnhandledStanza(this, stanza);
+      handleUnhandledStanza(this, incomingPreHandlers.stanza);
     }
   }
 

packages/moxxmpp/lib/src/events.dart

@@ -1,6 +1,7 @@
 import 'package:moxxmpp/src/connection.dart';
 import 'package:moxxmpp/src/jid.dart';
 import 'package:moxxmpp/src/managers/data.dart';
+import 'package:moxxmpp/src/roster/roster.dart';
 import 'package:moxxmpp/src/stanza.dart';
 import 'package:moxxmpp/src/xeps/xep_0030/types.dart';
 import 'package:moxxmpp/src/xeps/xep_0060/xep_0060.dart';
@@ -53,6 +54,22 @@ class StreamResumedEvent extends XmppEvent {
 /// Triggered when stream resumption failed
 class StreamResumeFailedEvent extends XmppEvent {}
 
+/// Triggered when the roster has been modified
+class RosterUpdatedEvent extends XmppEvent {
+  RosterUpdatedEvent(this.removed, this.modified, this.added);
+
+  /// A list of bare JIDs that are removed from the roster
+  final List<String> removed;
+
+  /// A list of XmppRosterItems that are modified. Can be correlated with one's cache
+  /// using the jid attribute.
+  final List<XmppRosterItem> modified;
+
+  /// A list of XmppRosterItems that are added to the roster.
+  final List<XmppRosterItem> added;
+}
+
+/// Triggered when a message is received
 class MessageEvent extends XmppEvent {
   MessageEvent({
     required this.body,

packages/moxxmpp/lib/src/managers/base.dart

@@ -31,6 +31,11 @@ abstract class XmppManagerBase {
   /// Return the StanzaHandlers associated with this manager that deal with stanzas we
   /// receive.
   List<StanzaHandler> getIncomingStanzaHandlers() => [];
+
+  /// Return the StanzaHandlers associated with this manager that deal with stanza handlers
+  /// that have to run before the main ones run. This is useful, for example, for OMEMO
+  /// as we have to decrypt the stanza before we do anything else.
+  List<StanzaHandler> getIncomingPreStanzaHandlers() => [];
   
   /// Return the NonzaHandlers associated with this manager.
   List<NonzaHandler> getNonzaHandlers() => [];

packages/moxxmpp/lib/src/roster/roster.dart

@@ -1,5 +1,8 @@
-import 'package:moxxmpp/src/events.dart';
+import 'dart:async';
+import 'package:collection/collection.dart';
+import 'package:meta/meta.dart';
 import 'package:moxxmpp/src/jid.dart';
+import 'package:moxxmpp/src/managers/attributes.dart';
 import 'package:moxxmpp/src/managers/base.dart';
 import 'package:moxxmpp/src/managers/data.dart';
 import 'package:moxxmpp/src/managers/handlers.dart';
@@ -8,17 +11,42 @@ import 'package:moxxmpp/src/namespaces.dart';
 import 'package:moxxmpp/src/negotiators/namespaces.dart';
 import 'package:moxxmpp/src/negotiators/negotiator.dart';
 import 'package:moxxmpp/src/roster/errors.dart';
+import 'package:moxxmpp/src/roster/state.dart';
 import 'package:moxxmpp/src/stanza.dart';
 import 'package:moxxmpp/src/stringxml.dart';
 import 'package:moxxmpp/src/types/result.dart';
 
+@immutable
 class XmppRosterItem {
-  XmppRosterItem({ required this.jid, required this.subscription, this.ask, this.name, this.groups = const [] });
+  const XmppRosterItem({ required this.jid, required this.subscription, this.ask, this.name, this.groups = const [] });
   final String jid;
   final String? name;
   final String subscription;
   final String? ask;
   final List<String> groups;
+
+  @override
+  bool operator==(Object other) {
+    return other is XmppRosterItem &&
+           other.jid == jid &&
+           other.name == name &&
+           other.subscription == subscription &&
+           other.ask == ask &&
+           const ListEquality<String>().equals(other.groups, groups);
+  }
+
+  @override
+  int get hashCode => jid.hashCode ^ name.hashCode ^ subscription.hashCode ^ ask.hashCode ^ groups.hashCode;
+  
+  @override
+  String toString() {
+    return 'XmppRosterItem('
+           'jid: $jid, '
+           'name: $name, '
+           'subscription: $subscription, '
+           'ask: $ask, '
+           'groups: $groups)';
+  }
 }
 
 enum RosterRemovalResult {
@@ -28,13 +56,13 @@ enum RosterRemovalResult {
 }
 
 class RosterRequestResult {
-  RosterRequestResult({ required this.items, this.ver });
+  RosterRequestResult(this.items, this.ver);
   List<XmppRosterItem> items;
   String? ver;
 }
 
-class RosterPushEvent extends XmppEvent {
-  RosterPushEvent({ required this.item, this.ver });
+class RosterPushResult {
+  RosterPushResult(this.item, this.ver);
   final XmppRosterItem item;
   final String? ver;
 }
@@ -65,9 +93,16 @@ class RosterFeatureNegotiator extends XmppFeatureNegotiatorBase {
 
 /// This manager requires a RosterFeatureNegotiator to be registered.
 class RosterManager extends XmppManagerBase {
+  RosterManager(this._stateManager) : super();
 
-  RosterManager() : _rosterVersion = null, super();
-  String? _rosterVersion;
+  /// The class managing the entire roster state.
+  final BaseRosterStateManager _stateManager;
+
+  @override
+  void register(XmppManagerAttributes attributes) {
+    super.register(attributes);
+    _stateManager.register(attributes.sendEvent); 
+  }
   
   @override
   String getId() => rosterManager;
@@ -87,17 +122,7 @@ class RosterManager extends XmppManagerBase {
 
   @override
   Future<bool> isSupported() async => true; 
-  
-  /// Override-able functions
-  Future<void> commitLastRosterVersion(String version) async {}
-  Future<void> loadLastRosterVersion() async {}
 
-  void setRosterVersion(String ver) {
-    assert(_rosterVersion == null, 'A roster version must not be empty');
-
-    _rosterVersion = ver;
-  }
- 
   Future<StanzaHandlerData> _onRosterPush(Stanza stanza, StanzaHandlerData state) async {
     final attrs = getAttributes();
     final from = stanza.attributes['from'] as String?;
@@ -114,6 +139,7 @@ class RosterManager extends XmppManagerBase {
     }
 
     final query = stanza.firstTag('query', xmlns: rosterXmlns)!;
+    logger.fine('Roster push: ${query.toXml()}');
     final item = query.firstTag('item');
 
     if (item == null) {
@@ -121,21 +147,20 @@ class RosterManager extends XmppManagerBase {
       return state.copyWith(done: true);
     }
 
-    if (query.attributes['ver'] != null) {
-      final ver = query.attributes['ver']! as String;
-      await commitLastRosterVersion(ver);
-      _rosterVersion = ver;
-    }
-    
-    attrs.sendEvent(RosterPushEvent(
-      item: XmppRosterItem(
-        jid: item.attributes['jid']! as String,
-        subscription: item.attributes['subscription']! as String,
-        ask: item.attributes['ask'] as String?,
-        name: item.attributes['name'] as String?, 
+    unawaited(
+      _stateManager.handleRosterPush(
+        RosterPushResult(
+          XmppRosterItem(
+            jid: item.attributes['jid']! as String,
+            subscription: item.attributes['subscription']! as String,
+            ask: item.attributes['ask'] as String?,
+            name: item.attributes['name'] as String?, 
+          ),
+          query.attributes['ver'] as String?,
+        ),
       ),
-      ver: query.attributes['ver'] as String?,
-    ),);
+    );
+    
     await attrs.sendStanza(stanza.reply());
 
     return state.copyWith(done: true);
@@ -145,71 +170,69 @@ class RosterManager extends XmppManagerBase {
   /// the server deems a regular roster response more efficient than n roster pushes.
   Future<Result<RosterRequestResult, RosterError>> _handleRosterResponse(XMLNode? query) async {
     final List<XmppRosterItem> items;
+    String? rosterVersion;
     if (query != null) {
-      items = query.children.map((item) => XmppRosterItem(
-        name: item.attributes['name'] as String?,
-        jid: item.attributes['jid']! as String,
-        subscription: item.attributes['subscription']! as String,
-        ask: item.attributes['ask'] as String?,
-        groups: item.findTags('group').map((groupNode) => groupNode.innerText()).toList(),
-      ),).toList();
+      items = query.children.map(
+        (item) => XmppRosterItem(
+          name: item.attributes['name'] as String?,
+          jid: item.attributes['jid']! as String,
+          subscription: item.attributes['subscription']! as String,
+          ask: item.attributes['ask'] as String?,
+          groups: item.findTags('group').map((groupNode) => groupNode.innerText()).toList(),
+        ),
+      ).toList();
 
-      if (query.attributes['ver'] != null) {
-        final ver_ = query.attributes['ver']! as String;
-        await commitLastRosterVersion(ver_);
-        _rosterVersion = ver_;
-      }
+      rosterVersion = query.attributes['ver'] as String?;
     } else {
       logger.warning('Server response to roster request without roster versioning does not contain a <query /> element, while the type is not error. This violates RFC6121');
       return Result(NoQueryError());
     }
 
-    final ver = query.attributes['ver'] as String?;
-    if (ver != null) {
-      _rosterVersion = ver;
-      await commitLastRosterVersion(ver);
-    }
-    
-    return Result(
-      RosterRequestResult(
-        items: items,
-        ver: ver,
-      ),
+    final result = RosterRequestResult(
+      items,
+      rosterVersion,
     );
 
+    unawaited(
+      _stateManager.handleRosterFetch(result),
+    );
+
+    return Result(result);
   }
   
-  /// Requests the roster following RFC 6121 without using roster versioning.
+  /// Requests the roster following RFC 6121.
   Future<Result<RosterRequestResult, RosterError>> requestRoster() async {
     final attrs = getAttributes();
+    final query = XMLNode.xmlns(
+      tag: 'query',
+      xmlns: rosterXmlns,
+    );
+    final rosterVersion = await _stateManager.getRosterVersion();
+    if (rosterVersion != null && rosterVersioningAvailable()) {
+      query.attributes['ver'] = rosterVersion;
+    }
+
     final response = await attrs.sendStanza(
       Stanza.iq(
         type: 'get',
         children: [
-          XMLNode.xmlns(
-            tag: 'query',
-            xmlns: rosterXmlns,
-          )
+          query,
         ],
       ),
     );
 
     if (response.attributes['type'] != 'result') {
-      logger.warning('Error requesting roster without roster versioning: ${response.toXml()}');
+      logger.warning('Error requesting roster: ${response.toXml()}');
       return Result(UnknownError());
     }
 
-    final query = response.firstTag('query', xmlns: rosterXmlns);
-    return _handleRosterResponse(query);
+    final responseQuery = response.firstTag('query', xmlns: rosterXmlns);
+    return _handleRosterResponse(responseQuery);
   }
 
   /// Requests a series of roster pushes according to RFC6121. Requires that the server
   /// advertises urn:xmpp:features:rosterver in the stream features.
   Future<Result<RosterRequestResult?, RosterError>> requestRosterPushes() async {
-    if (_rosterVersion == null) {
-      await loadLastRosterVersion();
-    }
-
     final attrs = getAttributes();
     final result = await attrs.sendStanza(
       Stanza.iq(
@@ -219,7 +242,7 @@ class RosterManager extends XmppManagerBase {
             tag: 'query',
             xmlns: rosterXmlns,
             attributes: {
-              'ver': _rosterVersion ?? ''
+              'ver': await _stateManager.getRosterVersion() ?? '',
             },
           )
         ],

packages/moxxmpp/lib/src/roster/state.dart

@@ -0,0 +1,220 @@
+import 'package:meta/meta.dart';
+import 'package:moxxmpp/src/events.dart';
+import 'package:moxxmpp/src/roster/roster.dart';
+import 'package:synchronized/synchronized.dart';
+
+class _RosterProcessTriple {
+  const _RosterProcessTriple(this.removed, this.modified, this.added);
+  final String? removed;
+  final XmppRosterItem? modified;
+  final XmppRosterItem? added;
+}
+
+class RosterCacheLoadResult {
+  const RosterCacheLoadResult(this.version, this.roster);
+  final String? version;
+  final List<XmppRosterItem> roster;
+}
+
+/// This class manages the roster state in order to correctly process and persist
+/// roster pushes and facilitate roster versioning requests.
+abstract class BaseRosterStateManager {
+  /// The cached version of the roster. If null, then it has not been loaded yet.
+  List<XmppRosterItem>? _currentRoster;
+
+  /// The cached version of the roster version.
+  String? _currentVersion;
+
+  /// A critical section locking both _currentRoster and _currentVersion.
+  final Lock _lock = Lock();
+
+  /// A function to send an XmppEvent to moxxmpp's main event bus
+  late void Function(XmppEvent) _sendEvent;
+  
+  /// Overrideable function
+  /// Loads the old cached version of the roster and optionally that roster version
+  /// from persistent storage into a RosterCacheLoadResult object.
+  Future<RosterCacheLoadResult> loadRosterCache();
+
+  /// Overrideable function
+  /// Commits the roster data to persistent storage.
+  ///
+  /// [version] is the roster version string. If none was provided, then this value
+  /// is null.
+  ///
+  /// [removed] is a (possibly empty) list of bare JIDs that are removed from the
+  /// roster.
+  ///
+  /// [modified] is a (possibly empty) list of XmppRosterItems that are modified. Correlation with
+  /// the cache is done using its jid attribute.
+  ///
+  /// [added] is a (possibly empty) list of XmppRosterItems that are added by the
+  /// roster push or roster fetch request.
+  Future<void> commitRoster(String? version, List<String> removed, List<XmppRosterItem> modified, List<XmppRosterItem> added);
+
+  /// Internal function. Registers functions from the RosterManger against this
+  /// instance.
+  void register(void Function(XmppEvent) sendEvent) {
+    _sendEvent = sendEvent;
+  }
+  
+  /// Load and cache or return the cached roster version.
+  Future<String?> getRosterVersion() async {
+    return _lock.synchronized(() async {
+      await _loadRosterCache();
+
+      return _currentVersion;
+    });
+  }
+
+  /// A wrapper around _commitRoster that also sends an event to moxxmpp's event
+  /// bus.
+  Future<void> _commitRoster(String? version, List<String> removed, List<XmppRosterItem> modified, List<XmppRosterItem> added) async {
+    _sendEvent(
+      RosterUpdatedEvent(
+        removed,
+        modified,
+        added,
+      ),
+    );
+    
+    await commitRoster(version, removed, modified, added);
+  }
+  
+  /// Loads the cached roster data into memory, if that has not already happened.
+  /// NOTE: Must be called from within the _lock critical section.
+  Future<void> _loadRosterCache() async {
+    if (_currentRoster == null) {
+      final result = await loadRosterCache();
+
+      _currentRoster = result.roster;
+      _currentVersion = result.version;
+    }
+  }
+
+  /// Processes only single XmppRosterItem [item].
+  /// NOTE: Requires to be called from within the _lock critical section.
+  _RosterProcessTriple _handleRosterItem(XmppRosterItem item) {
+    if (item.subscription == 'remove') {
+      // The item has been removed
+      _currentRoster!.removeWhere((i) => i.jid == item.jid);
+      return _RosterProcessTriple(
+        item.jid,
+        null,
+        null,
+      );
+    }
+    
+    final index = _currentRoster!.indexWhere((i) => i.jid == item.jid);
+    if (index == -1) {
+      // The item does not exist
+      _currentRoster!.add(item);
+      return _RosterProcessTriple(
+        null,
+        null,
+        item,
+      );
+    } else if (_currentRoster![index] != item) {
+      // The item is updated
+      _currentRoster![index] = item;
+      return _RosterProcessTriple(
+        null,
+        item,
+        null,
+      );
+    }
+
+    // Item has not been modified or added
+    return const _RosterProcessTriple(
+      null,
+      null,
+      null,
+    );
+  }
+
+  /// Handles a roster push from the RosterManager.
+  Future<void> handleRosterPush(RosterPushResult event) async {
+    await _lock.synchronized(() async {
+      await _loadRosterCache();
+
+      _currentVersion = event.ver;
+      final result = _handleRosterItem(event.item);
+
+      if (result.removed != null) {
+        return _commitRoster(
+          _currentVersion,
+          [result.removed!],
+          [],
+          [],
+        );
+      } else if (result.modified != null) {
+        return _commitRoster(
+          _currentVersion,
+          [],
+          [result.modified!],
+          [],
+        );
+      } else if (result.added != null) {
+        return _commitRoster(
+          _currentVersion,
+          [],
+          [],
+          [result.added!],
+        );
+      }
+    });
+  }
+
+  /// Handles the result from a roster fetch.
+  Future<void> handleRosterFetch(RosterRequestResult result) async {
+    await _lock.synchronized(() async {
+      final removed = List<String>.empty(growable: true);
+      final modified = List<XmppRosterItem>.empty(growable: true);
+      final added = List<XmppRosterItem>.empty(growable: true);
+
+      await _loadRosterCache();
+      
+      _currentVersion = result.ver;
+      for (final item in result.items) {
+        final result = _handleRosterItem(item);
+
+        if (result.removed != null) removed.add(result.removed!);
+        if (result.modified != null) modified.add(result.modified!);
+        if (result.added != null) added.add(result.added!);
+      }
+
+      await _commitRoster(
+        _currentVersion,
+        removed,
+        modified,
+        added,
+      );
+    });
+  }
+
+  @visibleForTesting
+  List<XmppRosterItem> getRosterItems() => _currentRoster!;
+}
+
+@visibleForTesting
+class TestingRosterStateManager extends BaseRosterStateManager {
+  TestingRosterStateManager(
+    this.initialRosterVersion,
+    this.initialRoster,
+  );
+  final String? initialRosterVersion;
+  final List<XmppRosterItem> initialRoster;
+  int loadCount = 0;
+
+  @override
+  Future<RosterCacheLoadResult> loadRosterCache() async {
+    loadCount++;
+    return RosterCacheLoadResult(
+      initialRosterVersion,
+      initialRoster,
+    );
+  }
+
+  @override
+  Future<void> commitRoster(String? version, List<String> removed, List<XmppRosterItem> modified, List<XmppRosterItem> added) async {}
+}

packages/moxxmpp/lib/src/xeps/xep_0280.dart

@@ -25,13 +25,12 @@ class CarbonsManager extends XmppManagerBase {
   String getName() => 'CarbonsManager';
 
   @override
-  List<StanzaHandler> getIncomingStanzaHandlers() => [
+  List<StanzaHandler> getIncomingPreStanzaHandlers() => [
     StanzaHandler(
       stanzaTag: 'message',
       tagName: 'received',
       tagXmlns: carbonsXmlns,
       callback: _onMessageReceived,
-      // Before all managers the message manager depends on
       priority: -98,
     ),
     StanzaHandler(
@@ -39,7 +38,6 @@ class CarbonsManager extends XmppManagerBase {
       tagName: 'sent',
       tagXmlns: carbonsXmlns,
       callback: _onMessageSent,
-      // Before all managers the message manager depends on
       priority: -98,
     )
   ];

packages/moxxmpp/lib/src/xeps/xep_0384/errors.dart

@@ -7,3 +7,5 @@ class InvalidAffixElementsException with Exception {}
 class OmemoNotSupportedForContactException extends OmemoError {}
 
 class EncryptionFailedException with Exception {}
+
+class InvalidEnvelopePayloadException with Exception {}

packages/moxxmpp/lib/src/xeps/xep_0384/xep_0384.dart

@@ -24,6 +24,7 @@ import 'package:moxxmpp/src/xeps/xep_0384/errors.dart';
 import 'package:moxxmpp/src/xeps/xep_0384/helpers.dart';
 import 'package:moxxmpp/src/xeps/xep_0384/types.dart';
 import 'package:omemo_dart/omemo_dart.dart';
+import 'package:xml/xml.dart';
 
 const _doNotEncryptList = [
   // XEP-0033
@@ -53,27 +54,24 @@ abstract class BaseOmemoManager extends XmppManagerBase {
   Future<bool> isSupported() async => true;
 
   @override
-  List<StanzaHandler> getIncomingStanzaHandlers() => [
+  List<StanzaHandler> getIncomingPreStanzaHandlers() => [
     StanzaHandler(
       stanzaTag: 'iq',
       tagXmlns: omemoXmlns,
       tagName: 'encrypted',
       callback: _onIncomingStanza,
-      priority: 9999,
     ),
     StanzaHandler(
       stanzaTag: 'presence',
       tagXmlns: omemoXmlns,
       tagName: 'encrypted',
       callback: _onIncomingStanza,
-      priority: 9999,
     ),
     StanzaHandler(
       stanzaTag: 'message',
       tagXmlns: omemoXmlns,
       tagName: 'encrypted',
       callback: _onIncomingStanza,
-      priority: -98,
     ),
   ];
 
@@ -432,16 +430,29 @@ abstract class BaseOmemoManager extends XmppManagerBase {
       ),
     );
 
-    final children = stanza.children.where(
-      (child) => child.tag != 'encrypted' || child.attributes['xmlns'] != omemoXmlns,
-    ).toList();
     final other = Map<String, dynamic>.from(state.other);
+    var children = stanza.children;
     if (result.error != null) {
       other['encryption_error'] = result.error;
+    } else {
+      children = stanza.children.where(
+        (child) => child.tag != 'encrypted' || child.attributes['xmlns'] != omemoXmlns,
+      ).toList();
     }
 
     if (result.payload != null) {
-      final envelope = XMLNode.fromString(result.payload!);
+      XMLNode envelope;
+      try {
+        envelope = XMLNode.fromString(result.payload!);
+      } on XmlParserException catch (_) {
+        logger.warning('Failed to parse envelope payload: ${result.payload!}');
+        other['encryption_error'] = InvalidEnvelopePayloadException();
+        return state.copyWith(
+          encrypted: true,
+          other: other,
+        );
+      }
+
       children.addAll(
         envelope.firstTag('content')!.children,
       );

packages/moxxmpp/pubspec.yaml

@@ -8,6 +8,7 @@ environment:
   sdk: '>=2.17.5 <3.0.0'
 
 dependencies:
+  collection: ^1.16.0
   cryptography: ^2.0.5
   freezed: ^2.1.0+1
   freezed_annotation: ^2.1.0
@@ -20,7 +21,7 @@ dependencies:
    version: ^0.1.5
   omemo_dart:
     hosted: https://git.polynom.me/api/packages/PapaTutuWawa/pub
-    version: ^0.4.1
+    version: ^0.4.2
   random_string: ^2.3.1
   saslprep: ^1.0.2
   synchronized: ^3.0.0+2

packages/moxxmpp/test/negotiator_test.dart

@@ -61,7 +61,7 @@ void main() {
     ])
     ..registerManagers([
       PresenceManager('http://moxxmpp.example'),
-      RosterManager(),
+      RosterManager(TestingRosterStateManager('', [])),
       DiscoManager(),
       PingManager(),
     ])

packages/moxxmpp/test/roster_state_test.dart

@@ -0,0 +1,153 @@
+import 'package:moxxmpp/moxxmpp.dart';
+import 'package:test/test.dart';
+
+void main() {
+  test('Test receiving a roster push', () async {
+    final rs = TestingRosterStateManager(null, []);
+    rs.register((_) {});
+
+    await rs.handleRosterPush(
+      RosterPushResult(
+        XmppRosterItem(
+          jid: 'testuser@server.example',
+          subscription: 'both',
+        ),
+        null,
+      ),
+    );
+
+    expect(
+      rs.getRosterItems().indexWhere((item) => item.jid == 'testuser@server.example') != -1,
+      true,
+    );
+    expect(rs.loadCount, 1);
+    expect(rs.getRosterItems().length, 1);
+
+    // Receive another roster push
+    await rs.handleRosterPush(
+      RosterPushResult(
+        XmppRosterItem(
+          jid: 'testuser2@server2.example',
+          subscription: 'to',
+        ),
+        null,
+      ),
+    );
+
+    expect(
+      rs.getRosterItems().indexWhere((item) => item.jid == 'testuser2@server2.example') != -1,
+      true,
+    );
+    expect(rs.loadCount, 1);
+    expect(rs.getRosterItems().length, 2);
+
+    // Remove one of the items
+     await rs.handleRosterPush(
+      RosterPushResult(
+        XmppRosterItem(
+          jid: 'testuser2@server2.example',
+          subscription: 'remove',
+        ),
+        null,
+      ),
+    );
+
+    expect(
+      rs.getRosterItems().indexWhere((item) => item.jid == 'testuser2@server2.example') == -1,
+      true,
+    );
+    expect(
+      rs.getRosterItems().indexWhere((item) => item.jid == 'testuser@server.example') != 1,
+      true,
+    );
+    expect(rs.loadCount, 1);
+    expect(rs.getRosterItems().length, 1);   
+  });
+
+  test('Test a roster fetch', () async {
+    final rs = TestingRosterStateManager(null, []);
+    rs.register((_) {});
+
+    // Fetch the roster
+    await rs.handleRosterFetch(
+      RosterRequestResult(
+        [
+          XmppRosterItem(
+            jid: 'testuser@server.example',
+            subscription: 'both',
+          ),
+          XmppRosterItem(
+            jid: 'testuser2@server2.example',
+            subscription: 'to',
+          ),
+          XmppRosterItem(
+            jid: 'testuser3@server3.example',
+            subscription: 'from',
+          ),
+        ],
+        'aaaaaaaa',
+      ),
+    );
+
+    expect(rs.loadCount, 1);
+    expect(rs.getRosterItems().length, 3);
+    expect(rs.getRosterItems().indexWhere((item) => item.jid == 'testuser@server.example') != -1, true);
+    expect(rs.getRosterItems().indexWhere((item) => item.jid == 'testuser2@server2.example') != -1, true);
+    expect(rs.getRosterItems().indexWhere((item) => item.jid == 'testuser3@server3.example') != -1, true);
+  });
+
+  test('Test a roster fetch if we already have a roster', () async {
+    XmppEvent? event;
+    final rs = TestingRosterStateManager('aaaaa', [
+      XmppRosterItem(
+        jid: 'testuser@server.example',
+        subscription: 'both',
+      ),
+      XmppRosterItem(
+        jid: 'testuser2@server2.example',
+        subscription: 'to',
+      ),
+      XmppRosterItem(
+        jid: 'testuser3@server3.example',
+        subscription: 'from',
+      ),
+    ]);
+    rs.register((_event) {
+      event = _event;
+    });
+
+    // Fetch the roster
+    await rs.handleRosterFetch(
+      RosterRequestResult(
+        [
+          XmppRosterItem(
+            jid: 'testuser@server.example',
+            subscription: 'both',
+          ),
+          XmppRosterItem(
+            jid: 'testuser2@server2.example',
+            subscription: 'to',
+          ),
+          XmppRosterItem(
+            jid: 'testuser3@server3.example',
+            subscription: 'both',
+          ),
+          XmppRosterItem(
+            jid: 'testuser4@server4.example',
+            subscription: 'both',
+          ),
+        ],
+        'bbbbb',
+      ),
+    );
+
+    expect(event is RosterUpdatedEvent, true);
+    final updateEvent = event as RosterUpdatedEvent;
+
+    expect(updateEvent.added.length, 1);
+    expect(updateEvent.added.first.jid, 'testuser4@server4.example');
+    expect(updateEvent.modified.length, 1);
+    expect(updateEvent.modified.first.jid, 'testuser3@server3.example');
+    expect(updateEvent.removed.isEmpty, true);
+  });
+}

packages/moxxmpp/test/roster_test.dart

@@ -1,322 +0,0 @@
-import 'package:moxxmpp/moxxmpp.dart';
-import 'package:test/test.dart';
-
-// TODO(PapaTutuWawa): Fix tests
-
-typedef AddRosterItemFunction = Future<RosterItem> Function(
-  String avatarUrl,
-  String avatarHash,
-  String jid,
-  String title,
-  String subscription,
-  String ask,
-  {
-    List<String> groups,
-  }
-);
-
-typedef UpdateRosterItemFunction = Future<RosterItem> Function(
-  int id, {
-    String? avatarUrl,
-    String? avatarHash,
-    String? title,
-    String? subscription,
-    String? ask,
-    List<String>? groups,
-  }
-);
-
-AddRosterItemFunction mkAddRosterItem(void Function(String) callback) {
-  return (
-    String avatarUrl,
-    String avatarHash,
-    String jid,
-    String title,
-    String subscription,
-    String ask,
-    {
-      List<String> groups = const [],
-    }
-  ) async {
-    callback(jid);
-    return await addRosterItemFromData(
-      avatarUrl,
-      avatarHash,
-      jid,
-      title,
-      subscription,
-      ask,
-      groups: groups,
-    );
-  };
-}
-
-Future<RosterItem> addRosterItemFromData(
-  String avatarUrl,
-  String avatarHash,
-  String jid,
-  String title,
-  String subscription,
-  String ask,
-  {
-    List<String> groups = const [],
-  }
-) async => RosterItem(
-  0,
-  avatarUrl,
-  avatarHash,
-  jid,
-  title,
-  subscription,
-  ask,
-  groups,
-);
-
-UpdateRosterItemFunction mkRosterUpdate(List<RosterItem> roster) {
-  return (
-    int id, {
-      String? avatarUrl,
-      String? avatarHash,
-      String? title,
-      String? subscription,
-      String? ask,
-      List<String>? groups,
-    }
-  ) async {
-    final item = firstWhereOrNull(roster, (RosterItem item) => item.id == id)!;
-    return item.copyWith(
-      avatarUrl: avatarUrl ?? item.avatarUrl,
-      avatarHash: avatarHash ?? item.avatarHash,
-      title: title ?? item.title,
-      subscription: subscription ?? item.subscription,
-      ask: ask ?? item.ask,
-      groups: groups ?? item.groups,
-    );
-  };
-}
-
-void main() {
-  final localRosterSingle = [
-    RosterItem(
-      0,
-      '',
-      '',
-      'hallo@server.example',
-      'hallo',
-      'none',
-      '',
-      [],
-    )
-  ];
-  final localRosterDouble = [
-    RosterItem(
-      0,
-      '',
-      '',
-      'hallo@server.example',
-      'hallo',
-      'none',
-      '',
-      [],
-    ),
-    RosterItem(
-      1,
-      '',
-      '',
-      'welt@different.server.example',
-      'welt',
-      'from',
-      '',
-      [ 'Friends' ],
-    )
-  ];
-
-  group('Test roster pushes', () {
-      test('Test removing an item', () async {
-          var removeCalled = false;
-          var addCalled = false;
-          final result = await processRosterDiff(
-            localRosterDouble,
-            [
-              XmppRosterItem(
-                jid: 'hallo@server.example', subscription: 'remove',
-              ) 
-            ],
-            true,
-            mkAddRosterItem((_) { addCalled = true; }),
-            mkRosterUpdate(localRosterDouble),
-            (jid) async {
-              if (jid == 'hallo@server.example') {
-                removeCalled = true;
-              }
-            },
-            (_) async => null,
-            (_, { String? id }) async {},
-          );
-
-          expect(result.removed, [ 'hallo@server.example' ]);
-          expect(result.modified.length, 0);
-          expect(result.added.length, 0);
-          expect(removeCalled, true);
-          expect(addCalled, false);
-      });
-
-      test('Test adding an item', () async {
-          var removeCalled = false;
-          var addCalled = false;
-          final result = await processRosterDiff(
-            localRosterSingle,
-            [
-              XmppRosterItem(
-                jid: 'welt@different.server.example',
-                subscription: 'from',
-              ) 
-            ],
-            true,
-            mkAddRosterItem(
-              (jid) {
-                if (jid == 'welt@different.server.example') {
-                  addCalled = true;
-                }
-              }
-            ),
-            mkRosterUpdate(localRosterSingle),
-            (_) async { removeCalled = true; },
-            (_) async => null,
-            (_, { String? id }) async {},
-          );
-
-          expect(result.removed, [ ]);
-          expect(result.modified.length, 0);
-          expect(result.added.length, 1);
-          expect(result.added.first.subscription, 'from');
-          expect(removeCalled, false);
-          expect(addCalled, true);
-      });
-
-      test('Test modifying an item', () async {
-          var removeCalled = false;
-          var addCalled = false;
-          final result = await processRosterDiff(
-            localRosterDouble,
-            [
-              XmppRosterItem(
-                jid: 'welt@different.server.example',
-                subscription: 'both',
-                name: 'The World',
-              ) 
-            ],
-            true,
-            mkAddRosterItem((_) { addCalled = false; }),
-            mkRosterUpdate(localRosterDouble),
-            (_) async { removeCalled = true; },
-            (_) async => null,
-            (_, { String? id }) async {},
-          );
-
-          expect(result.removed, [ ]);
-          expect(result.modified.length, 1);
-          expect(result.added.length, 0);
-          expect(result.modified.first.subscription, 'both');
-          expect(result.modified.first.jid, 'welt@different.server.example');
-          expect(result.modified.first.title, 'The World');
-          expect(removeCalled, false);
-          expect(addCalled, false);
-      });
-  });
-
-  group('Test roster requests', () {
-      test('Test removing an item', () async {
-          var removeCalled = false;
-          var addCalled = false;
-          final result = await processRosterDiff(
-            localRosterSingle,
-            [],
-            false,
-            mkAddRosterItem((_) { addCalled = true; }),
-            mkRosterUpdate(localRosterDouble),
-            (jid) async {
-              if (jid == 'hallo@server.example') {
-                removeCalled = true;
-              }
-            },
-            (_) async => null,
-            (_, { String? id }) async {},
-          );
-
-          expect(result.removed, [ 'hallo@server.example' ]);
-          expect(result.modified.length, 0);
-          expect(result.added.length, 0);
-          expect(removeCalled, true);
-          expect(addCalled, false);
-      });
-
-      test('Test adding an item', () async {
-          var removeCalled = false;
-          var addCalled = false;
-          final result = await processRosterDiff(
-            localRosterSingle,
-            [
-              XmppRosterItem(
-                jid: 'hallo@server.example',
-                name: 'hallo',
-                subscription: 'none',
-              ),
-              XmppRosterItem(
-                jid: 'welt@different.server.example',
-                subscription: 'both',
-              )
-            ],
-            false,
-            mkAddRosterItem(
-              (jid) {
-                if (jid == 'welt@different.server.example') {
-                  addCalled = true;
-                }
-              }
-            ),
-            mkRosterUpdate(localRosterSingle),
-            (_) async { removeCalled = true; },
-            (_) async => null,
-            (_, { String? id }) async {},
-          );
-
-          expect(result.removed, [ ]);
-          expect(result.modified.length, 0);
-          expect(result.added.length, 1);
-          expect(result.added.first.subscription, 'both');
-          expect(removeCalled, false);
-          expect(addCalled, true);
-      });
-
-      test('Test modifying an item', () async {
-          var removeCalled = false;
-          var addCalled = false;
-          final result = await processRosterDiff(
-            localRosterSingle,
-            [
-              XmppRosterItem(
-                jid: 'hallo@server.example',
-                subscription: 'both',
-                name: 'Hallo Welt',
-              ) 
-            ],
-            false,
-            mkAddRosterItem((_) { addCalled = false; }),
-            mkRosterUpdate(localRosterDouble),
-            (_) async { removeCalled = true; },
-            (_) async => null,
-            (_, { String? id }) async {},
-          );
-
-          expect(result.removed, [ ]);
-          expect(result.modified.length, 1);
-          expect(result.added.length, 0);
-          expect(result.modified.first.subscription, 'both');
-          expect(result.modified.first.jid, 'hallo@server.example');
-          expect(result.modified.first.title, 'Hallo Welt');
-          expect(removeCalled, false);
-          expect(addCalled, false);
-      });
-  });
-}

packages/moxxmpp/test/xeps/xep_0198_test.dart

@@ -243,7 +243,7 @@ void main() {
       final sm = StreamManagementManager();
       conn.registerManagers([
           PresenceManager('http://moxxmpp.example'),
-          RosterManager(),
+          RosterManager(TestingRosterStateManager('', [])),
           DiscoManager(),
           PingManager(),
           sm,
@@ -365,7 +365,7 @@ void main() {
       final sm = StreamManagementManager();
       conn.registerManagers([
           PresenceManager('http://moxxmpp.example'),
-          RosterManager(),
+          RosterManager(TestingRosterStateManager('', [])),
           DiscoManager(),
           PingManager(),
           sm,
@@ -519,7 +519,7 @@ void main() {
       ),);
       conn.registerManagers([
           PresenceManager('http://moxxmpp.example'),
-          RosterManager(),
+          RosterManager(TestingRosterStateManager('', [])),
           DiscoManager(),
           PingManager(),
           StreamManagementManager(),
@@ -611,7 +611,7 @@ void main() {
       ),);
       conn.registerManagers([
           PresenceManager('http://moxxmpp.example'),
-          RosterManager(),
+          RosterManager(TestingRosterStateManager('', [])),
           DiscoManager(),
           PingManager(),
           StreamManagementManager(),
@@ -703,7 +703,7 @@ void main() {
       ),);
       conn.registerManagers([
           PresenceManager('http://moxxmpp.example'),
-          RosterManager(),
+          RosterManager(TestingRosterStateManager('', [])),
           DiscoManager(),
           PingManager(),
           StreamManagementManager(),

packages/moxxmpp/test/xmpp_test.dart

@@ -7,7 +7,7 @@ import 'helpers/xmpp.dart';
 /// Returns true if the roster manager triggeres an event for a given stanza
 Future<bool> testRosterManager(String bareJid, String resource, String stanzaString) async {
   var eventTriggered = false;
-  final roster = RosterManager();
+  final roster = RosterManager(TestingRosterStateManager('', []));
   roster.register(XmppManagerAttributes(
       sendStanza: (_, { StanzaFromType addFrom = StanzaFromType.full, bool addId = true, bool retransmitted = false, bool awaitable = true, bool encrypted = false }) async => XMLNode(tag: 'hallo'),
       sendEvent: (event) {
@@ -127,7 +127,7 @@ void main() {
       ),);
       conn.registerManagers([
         PresenceManager('http://moxxmpp.example'),
-        RosterManager(),
+        RosterManager(TestingRosterStateManager('', [])),
         DiscoManager(),
         PingManager(),
         StreamManagementManager(),
@@ -181,7 +181,7 @@ void main() {
       ),);
       conn.registerManagers([
         PresenceManager('http://moxxmpp.example'),
-        RosterManager(),
+        RosterManager(TestingRosterStateManager('', [])),
         DiscoManager(),
         PingManager(),
       ]);
@@ -235,7 +235,7 @@ void main() {
       ),);
       conn.registerManagers([
         PresenceManager('http://moxxmpp.example'),
-        RosterManager(),
+        RosterManager(TestingRosterStateManager('', [])),
         DiscoManager(),
         PingManager(),
       ]);
@@ -290,7 +290,7 @@ void main() {
       ),);
       conn.registerManagers([
         PresenceManager('http://moxxmpp.example'),
-        RosterManager(),
+        RosterManager(TestingRosterStateManager('', [])),
         DiscoManager(),
         PingManager(),
       ]);
@@ -308,7 +308,7 @@ void main() {
   group('Test roster pushes', () {
       test('Test for a CVE-2015-8688 style vulnerability', () async {
           var eventTriggered = false;
-          final roster = RosterManager();
+          final roster = RosterManager(TestingRosterStateManager('', []));
           roster.register(XmppManagerAttributes(
               sendStanza: (_, { StanzaFromType addFrom = StanzaFromType.full, bool addId = true, bool retransmitted = false, bool awaitable = true, bool encrypted = false }) async => XMLNode(tag: 'hallo'),
               sendEvent: (event) {