feat: validate events async, transform messages in async way

lib/src/crypto/signature_verifier.dart

@@ -1,33 +1,29 @@
 import 'package:nostr_dart/src/crypto/utils.dart';
 
 mixin SignatureVerifier {
-
-  bool verify({
+  Future<bool> verify({
     required String signature,
     required String message,
     required String publicKey,
   });
-
 }
 
 class SchnorrSignatureVerifier with SignatureVerifier {
-
   SchnorrSignatureVerifier();
 
   /// Verifies a schnorr signature using the BIP-340 scheme.
   ///
   /// It returns true if the signature is valid, false otherwise.
   @override
-  bool verify({
+  Future<bool> verify({
     required String signature,
     required String message,
     required String publicKey,
-  }) {
+  }) async {
     return verifyBip340Signature(
       signature: signature,
       publicKey: publicKey,
       message: message,
     );
   }
-
-}
+}

lib/src/model/event_message.dart

@@ -94,11 +94,7 @@ class EventMessage extends RelayMessage {
     required this.content,
     required this.sig,
     this.subscriptionId,
-  }) {
-    if (!_validate()) {
-      throw InvalidEventException(this);
-    }
-  }
+  });
 
   factory EventMessage.fromData({
     required EventSigner signer,
@@ -170,7 +166,7 @@ class EventMessage extends RelayMessage {
     }
   }
 
-  bool _validate() {
+  Future<bool> validate() async {
     final String calculatedId = calculateEventId(
       publicKey: pubkey,
       createdAt: createdAt,
@@ -183,7 +179,7 @@ class EventMessage extends RelayMessage {
       return false;
     }
 
-    return getIt<SignatureVerifier>().verify(
+    return await getIt<SignatureVerifier>().verify(
       signature: sig,
       message: id,
       publicKey: pubkey,

lib/src/relay.dart

@@ -36,7 +36,7 @@ class NostrRelay {
   }) {
     final controller = StreamController<RelayMessage>.broadcast();
     controller
-        .addStream(socket.messages.transform(_messageTransformer))
+        .addStream(_transformMessages(socket.messages))
         .whenComplete(controller.close);
     messages = controller.stream;
     socket.connection.listen(_onConnectionStateChange);
@@ -144,6 +144,38 @@ class NostrRelay {
     }
   }
 
+  Stream<RelayMessage> _transformMessages(Stream<dynamic> messages) {
+    return messages
+        .asyncMap<RelayMessage?>((message) async {
+          try {
+            final jsonMessage = jsonDecode(message as String) as List<dynamic>;
+            switch (jsonMessage[0]) {
+              case EventMessage.type:
+                final eventMessage = EventMessage.fromJson(jsonMessage);
+                if (await eventMessage.validate()) {
+                  return eventMessage;
+                } else {
+                  return null;
+                }
+              case EoseMessage.type:
+                return EoseMessage.fromJson(jsonMessage);
+              case OkMessage.type:
+                return OkMessage.fromJson(jsonMessage);
+              case NoticeMessage.type:
+                return NoticeMessage.fromJson(jsonMessage);
+              default:
+                logWarning(() => 'Unknown message $message');
+                return null;
+            }
+          } catch (error, stack) {
+            logWarning(() => 'Stream transform error', error, stack);
+            return null;
+          }
+        })
+        .where((event) => event != null)
+        .cast<RelayMessage>();
+  }
+
   /// Creates a new Relay and waits for it to be connected.
   static Future<NostrRelay> connect(
     String url, [
@@ -155,27 +187,3 @@ class NostrRelay {
     return relay;
   }
 }
-
-StreamTransformer<dynamic, RelayMessage> _messageTransformer =
-    StreamTransformer.fromHandlers(
-  handleData: (message, sink) {
-    try {
-      final jsonMessage = jsonDecode(message as String) as List<dynamic>;
-      switch (jsonMessage[0]) {
-        case EventMessage.type:
-          sink.add(EventMessage.fromJson(jsonMessage));
-        case EoseMessage.type:
-          sink.add(EoseMessage.fromJson(jsonMessage));
-        case OkMessage.type:
-          sink.add(OkMessage.fromJson(jsonMessage));
-        case NoticeMessage.type:
-          sink.add(NoticeMessage.fromJson(jsonMessage));
-        default:
-          logWarning(() => 'Unknown message $message');
-      }
-    } catch (error, stack) {
-      logWarning(() => 'Stream transform error', error, stack);
-      sink.addError(error);
-    }
-  },
-);

test/event_message_test.dart

@@ -98,22 +98,27 @@ void main() {
       expect(message.toString(), equals(rawEvent));
     });
 
-    test('must throw an exception if id is incorrect', () {
-      expect(
-        () => EventMessage.fromJson(
-          jsonDecode(rawEvent.replaceFirst(eventId, '123')) as List<dynamic>,
-        ),
-        throwsException,
-      );
+    test('validate() should return true for a valid event', () async {
+      final eventValid =
+          await EventMessage.fromJson(jsonDecode(rawEvent) as List<dynamic>)
+              .validate();
+      expect(eventValid, isTrue);
     });
 
-    test('must throw an exception if signature is incorrect', () {
-      expect(
-        () => EventMessage.fromJson(
-          jsonDecode(rawEvent.replaceFirst(sig, '123')) as List<dynamic>,
-        ),
-        throwsException,
-      );
+    test('validate() should return false for event with incorrect id',
+        () async {
+      final eventValid = await EventMessage.fromJson(
+        jsonDecode(rawEvent.replaceFirst(eventId, '123')) as List<dynamic>,
+      ).validate();
+      expect(eventValid, isFalse);
+    });
+
+    test('validate() should return false for event with incorrect signature',
+        () async {
+      final eventValid = await EventMessage.fromJson(
+        jsonDecode(rawEvent.replaceFirst(eventId, '123')) as List<dynamic>,
+      ).validate();
+      expect(eventValid, isFalse);
     });
   });
 }