Fixed bug with real time event notification

.gitignore

@@ -1,2 +1,3 @@
 /target
 crates/storage/tests/db/
+crates/*/tests/db/

crates/relayer/src/connection.rs

@@ -1,15 +1,18 @@
 use crate::{get_id, Error};
 use futures_util::{SinkExt, StreamExt};
-use nostr_rs_types::{Request, Response};
+use nostr_rs_types::{relayer::ROk, types::Addr, Request, Response};
 use parking_lot::RwLock;
 use std::collections::HashMap;
 use tokio::{
     net::TcpStream,
     sync::mpsc::{channel, Receiver, Sender},
 };
+#[allow(unused_imports)]
 use tokio_tungstenite::{accept_async, tungstenite::Message, WebSocketStream};
 
+#[derive(Debug)]
 pub struct Connection {
+    #[allow(dead_code)]
     pub(crate) conn_id: u128,
     sender: Sender<Response>,
     subscriptions: RwLock<HashMap<String, u128>>,
@@ -18,6 +21,20 @@ pub struct Connection {
 const MAX_SUBSCRIPTIONS_BUFFER: usize = 100;
 
 impl Connection {
+    #[cfg(test)]
+    pub fn new() -> (Self, Receiver<Response>) {
+        let (sender, receiver) = channel(MAX_SUBSCRIPTIONS_BUFFER);
+        (
+            Self {
+                conn_id: 0,
+                sender,
+                subscriptions: RwLock::new(HashMap::new()),
+            },
+            receiver,
+        )
+    }
+
+    #[cfg(not(test))]
     pub async fn new(
         broadcast_request: Sender<(u128, Request)>,
         stream: TcpStream,
@@ -33,7 +50,8 @@ impl Connection {
         })
     }
 
-    pub fn spawn(
+    #[cfg(not(test))]
+    fn spawn(
         broadcast_request: Sender<(u128, Request)>,
         websocket: WebSocketStream<TcpStream>,
         mut receiver: Receiver<Response>,
@@ -58,9 +76,28 @@ impl Connection {
                     Some(msg) = reader.next() => {
                         if let Ok(Message::Text(msg)) = msg {
                             let msg: Result<Request, _> = serde_json::from_str(&msg);
-                            if let Ok(msg) = msg {
-                                let _ = broadcast_request.send((conn_id, msg)).await;
-                            }
+                            match msg {
+                                Ok(msg) => {
+                                    let _ = broadcast_request.send((conn_id, msg)).await;
+                                },
+                                Err(err) => {
+                                    log::error!("Error parsing message from client: {}", err);
+                                    let reply: Response = ROk {
+                                        id: Addr::default(),
+                                        status: false,
+                                        message: "Error parsing message".to_owned(),
+                                    }.into();
+                                    let reply = if let Ok(reply) = serde_json::to_string(&reply) {
+                                        reply
+                                    } else {
+                                        continue;
+                                    };
+                                    if let Err(err) =  writer.send(Message::Text(reply.into())).await {
+                                        log::error!("Error sending message to client: {}", err);
+                                        break;
+                                    }
+                                }
+                            };
                         }
                     }
                     else => {

crates/relayer/src/lib.rs

@@ -3,8 +3,11 @@ use rand::Rng;
 mod connection;
 mod error;
 mod relayer;
+mod subscription;
 
-pub use self::{connection::Connection, error::Error, relayer::Relayer};
+pub use self::{
+    connection::Connection, error::Error, relayer::Relayer, subscription::Subscription,
+};
 
 // Get current nanoseconds and use the last 3 digits as a random number (because
 // sometimes it comes as 0)
@@ -14,7 +17,7 @@ pub(crate) fn get_id() -> u128 {
 
     let ts = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
-        .unwrap()
+        .expect("time")
         .as_nanos();
 
     ts.checked_add(random_number).unwrap_or(ts)

crates/relayer/src/relayer.rs

@@ -1,24 +1,18 @@
-use crate::{Connection, Error};
+use crate::{Connection, Error, Subscription};
 use nostr_rs_storage::RocksDb;
 use nostr_rs_types::{
     relayer,
-    types::{Event, Filter, SubscriptionId},
+    types::{Event, SubscriptionId},
     Request, Response,
 };
 use parking_lot::{RwLock, RwLockReadGuard};
 use std::{collections::HashMap, ops::Deref, sync::Arc};
+#[allow(unused_imports)]
 use tokio::{
     net::TcpStream,
     sync::mpsc::{channel, Receiver, Sender},
 };
 
-#[derive(Clone, Debug, Default, Eq, PartialEq, Hash)]
-pub struct SubscriptionType {
-    public_key: Option<Vec<u8>>,
-    id: Option<Vec<u8>>,
-    kind: Option<u32>,
-}
-
 type SubId = u128;
 
 type Subscriptions = HashMap<SubId, (SubscriptionId, Sender<Response>)>;
@@ -31,14 +25,15 @@ pub struct Relayer {
     /// Each connection keeps a list of the subscription ID provided by the user
     /// (String) and the internal, globally recognized subscription ID which is
     /// internal (SubId)
-    subscriptions_ids_index: RwLock<HashMap<SubId, Vec<SubscriptionType>>>,
+    subscriptions_ids_index: RwLock<HashMap<SubId, Vec<Subscription>>>,
     /// Each subscription type that is active has a list of subscriptions.
     ///
     /// A single REQ can be subscribed to multiple subscription types, specially
     /// when it is translated in OR filters. It is designed this way to allow a
     /// fast iteration and match quickly filters.
-    subscriptions: RwLock<HashMap<SubscriptionType, RwLock<Subscriptions>>>,
+    subscriptions: RwLock<HashMap<Subscription, RwLock<Subscriptions>>>,
     clients: RwLock<HashMap<u128, Connection>>,
+    #[allow(dead_code)]
     sender: Sender<(u128, Request)>,
 }
 
@@ -62,6 +57,7 @@ impl Relayer {
         &self.storage
     }
 
+    #[cfg(not(test))]
     pub async fn add_connection(&self, stream: TcpStream) -> Result<(), Error> {
         let client = Connection::new(self.sender.clone(), stream).await?;
         let mut clients = self.clients.write();
@@ -70,62 +66,52 @@ impl Relayer {
         Ok(())
     }
 
-    pub async fn recv(
+    fn recv_request_from_client(
         &self,
-        receiver: &mut Receiver<(u128, Request)>,
+        connection: &Connection,
+        request: Request,
     ) -> Result<Option<Request>, Error> {
-        let (conn_id, request) = if let Some(request) = receiver.recv().await {
-            request
-        } else {
-            return Ok(None);
-        };
-
-        let connections = self.clients.read();
-        let connection = connections
-            .get(&conn_id)
-            .ok_or(Error::UnknownConnection(conn_id))?;
-
         match &request {
             Request::Event(event) => {
                 self.store_and_broadcast_local_event(event.deref());
             }
             Request::Request(request) => {
-                for filter in request.filters.clone().into_iter() {
-                    // Create subscription
-                    let (sub_id, receiver) =
-                        connection.create_subscription(request.subscription_id.deref().to_owned());
-                    let mut sub_index = self.subscriptions_ids_index.write();
-                    let mut subscriptions = self.subscriptions.write();
-                    if let Some(prev_subs) = sub_index.remove(&sub_id) {
-                        // remove any previous subscriptions
-                        prev_subs.iter().for_each(|index| {
-                            if let Some(subscriptions) = subscriptions.get_mut(&index) {
-                                subscriptions.write().remove(&sub_id);
-                            }
-                        });
-                    }
-                    sub_index.insert(
-                        sub_id,
-                        Self::get_indexes_from_filter(&filter)
-                            .into_iter()
-                            .map(|index| {
-                                subscriptions
-                                    .entry(index.clone())
-                                    .or_insert_with(|| RwLock::new(HashMap::new()))
-                                    .write()
-                                    .insert(
-                                        sub_id,
-                                        (request.subscription_id.clone(), receiver.clone()),
-                                    );
-                                index
-                            })
-                            .collect::<Vec<_>>(),
-                    );
+                // Create subscription
+                let (sub_id, receiver) =
+                    connection.create_subscription(request.subscription_id.deref().to_owned());
+                let mut sub_index = self.subscriptions_ids_index.write();
+                let mut subscriptions = self.subscriptions.write();
+                if let Some(prev_subs) = sub_index.remove(&sub_id) {
+                    // remove any previous subscriptions
+                    prev_subs.iter().for_each(|index| {
+                        if let Some(subscriptions) = subscriptions.get_mut(&index) {
+                            subscriptions.write().remove(&sub_id);
+                        }
+                    });
+                }
+                sub_index.insert(
+                    sub_id,
+                    Subscription::from_filters(&request.filters)
+                        .into_iter()
+                        .map(|index| {
+                            subscriptions
+                                .entry(index.clone())
+                                .or_insert_with(|| RwLock::new(HashMap::new()))
+                                .write()
+                                .insert(
+                                    sub_id,
+                                    (request.subscription_id.clone(), receiver.clone()),
+                                );
+                            index
+                        })
+                        .collect::<Vec<_>>(),
+                );
 
-                    drop(subscriptions);
-                    drop(sub_index);
+                drop(subscriptions);
+                drop(sub_index);
 
-                    // Sent all events that match the filter that are stored in our database
+                // Sent all events that match the filter that are stored in our database
+                for filter in request.filters.clone().into_iter() {
                     self.storage
                         .get_by_filter(filter)?
                         .into_iter()
@@ -139,6 +125,7 @@ impl Relayer {
                             );
                         });
                 }
+
                 let _ = connection
                     .send(relayer::EndOfStoredEvents(request.subscription_id.clone()).into());
             }
@@ -160,73 +147,21 @@ impl Relayer {
         Ok(Some(request))
     }
 
-    #[inline]
-    fn get_indexes_from_filter(filter: &Filter) -> Vec<SubscriptionType> {
-        let public_keys = if filter.references_to_public_key.is_empty() {
-            vec![None]
-        } else {
-            filter
-                .references_to_public_key
-                .iter()
-                .map(|public_key| Some((*public_key).to_vec()))
-                .collect()
-        };
-        let id = if filter.references_to_event.is_empty() {
-            vec![None]
-        } else {
-            filter
-                .references_to_event
-                .iter()
-                .map(|id| Some((*id).to_vec()))
-                .collect()
-        };
-        let kind = if filter.kinds.is_empty() {
-            vec![None]
+    pub async fn recv(
+        &self,
+        receiver: &mut Receiver<(u128, Request)>,
+    ) -> Result<Option<Request>, Error> {
+        let (conn_id, request) = if let Some(request) = receiver.recv().await {
+            request
         } else {
-            filter
-                .kinds
-                .iter()
-                .map(|kind| Some((*kind).into()))
-                .collect()
+            return Ok(None);
         };
-        let mut subs = vec![];
-
-        for public_key in public_keys.iter() {
-            for id in id.iter() {
-                for kind in kind.iter() {
-                    subs.push(SubscriptionType {
-                        public_key: public_key.clone(),
-                        id: id.clone(),
-                        kind: *kind,
-                    });
-                }
-            }
-        }
-
-        subs
-    }
-
-    #[inline]
-    fn get_possible_listeners_from_event(event: &Event) -> Vec<SubscriptionType> {
-        let kind = event.kind().into();
-        let public_keys = [None, Some(event.author().as_ref().to_vec())];
-        let id = [None, Some(event.id.as_ref().to_vec())];
-        let kind = [None, Some(kind)];
-        let mut subs = vec![];
-
-        for public_key in public_keys.iter() {
-            for id in id.iter() {
-                for kind in kind.iter() {
-                    subs.push(SubscriptionType {
-                        public_key: public_key.clone(),
-                        id: id.clone(),
-                        kind: *kind,
-                    });
-                }
-            }
-        }
+        let connections = self.clients.read();
+        let connection = connections
+            .get(&conn_id)
+            .ok_or(Error::UnknownConnection(conn_id))?;
 
-        subs
+        self.recv_request_from_client(connection, request)
     }
 
     #[inline]
@@ -247,7 +182,7 @@ impl Relayer {
         let _ = self.storage.store_local_event(event);
         let subscriptions = self.subscriptions.read();
 
-        for subscription_type in Self::get_possible_listeners_from_event(event) {
+        for subscription_type in Subscription::from_event(event) {
             if let Some(subscribers) = subscriptions.get(&subscription_type) {
                 Self::broadcast_to_subscribers(subscribers.read(), event);
             }
@@ -259,10 +194,113 @@ impl Relayer {
         let _ = self.storage.store(event);
         let subscriptions = self.subscriptions.read();
 
-        for subscription_type in Self::get_possible_listeners_from_event(event) {
+        for subscription_type in Subscription::from_event(event) {
             if let Some(subscribers) = subscriptions.get(&subscription_type) {
                 Self::broadcast_to_subscribers(subscribers.read(), event);
             }
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::get_id;
+    use nostr_rs_storage::RocksDb;
+    use nostr_rs_types::Request;
+    use std::{
+        fs::File,
+        io::{BufRead, BufReader},
+    };
+
+    fn get_db(prefill: bool) -> RocksDb {
+        let db = RocksDb::new(format!("tests/db/{}", get_id())).expect("db");
+        if prefill {
+            let file = File::open("./tests/events.json").expect("file");
+            let events = BufReader::new(file)
+                .lines()
+                .map(|line| serde_json::from_str(&line.expect("line")).expect("valid"))
+                .collect::<Vec<Event>>();
+
+            for event in events {
+                assert!(db.store(&event).expect("valid"));
+            }
+        }
+        db
+    }
+
+    #[tokio::test]
+    async fn serve_listener_from_local_db() {
+        let request: Request = serde_json::from_str("[\"REQ\",\"1298169700973717\",{\"authors\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"since\":1681939304},{\"#p\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[1,3,6,7,9735],\"since\":1681939304},{\"#p\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[4]},{\"authors\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[4]},{\"#e\":[\"2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a\",\"a5e3369c43daf2675ecbce18831e5f4e07db0d4dde0ef4f5698e645e4c46eed1\"],\"kinds\":[1,6,7,9735]}]").expect("valid object");
+        let (relayer, _) = Relayer::new(get_db(true));
+        let (connection, mut recv) = Connection::new();
+        let _ = relayer.recv_request_from_client(&connection, request);
+        // ev1
+        assert_eq!(
+            "9508850d7ddc8ef58c8b392236c49d472dc23fa11f4e73eb5475dfb099ddff42",
+            recv.try_recv()
+                .expect("valid")
+                .as_event()
+                .expect("event")
+                .event
+                .id
+                .to_string()
+        );
+        // ev2
+        assert_eq!(
+            "2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a",
+            recv.try_recv()
+                .expect("valid")
+                .as_event()
+                .expect("event")
+                .event
+                .id
+                .to_string()
+        );
+        // ev3
+        assert_eq!(
+            "e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9",
+            recv.try_recv()
+                .expect("valid")
+                .as_event()
+                .expect("event")
+                .event
+                .id
+                .to_string()
+        );
+        // eod
+        assert!(recv
+            .try_recv()
+            .expect("valid")
+            .as_end_of_stored_events()
+            .is_some());
+
+        assert!(recv.try_recv().is_err());
+    }
+
+    #[tokio::test]
+    async fn server_listener_real_time() {
+        let request: Request = serde_json::from_str("[\"REQ\",\"1298169700973717\",{\"authors\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"since\":1681939304},{\"#p\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[1,3,6,7,9735],\"since\":1681939304},{\"#p\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[4]},{\"authors\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[4]},{\"#e\":[\"2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a\",\"a5e3369c43daf2675ecbce18831e5f4e07db0d4dde0ef4f5698e645e4c46eed1\"],\"kinds\":[1,6,7,9735]}]").expect("valid object");
+        let (relayer, _) = Relayer::new(get_db(false));
+        let (connection, mut recv) = Connection::new();
+        let _ = relayer.recv_request_from_client(&connection, request);
+        // eod
+        assert!(recv
+            .try_recv()
+            .expect("valid")
+            .as_end_of_stored_events()
+            .is_some());
+
+        // It is empty
+        assert!(recv.try_recv().is_err());
+
+        let new_event: Request = serde_json::from_str(r#"["EVENT", {"kind":1,"content":"Pong","tags":[["e","9508850d7ddc8ef58c8b392236c49d472dc23fa11f4e73eb5475dfb099ddff42","","root"],["e","2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a","","reply"],["p","39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb"],["p","ee7202ad91459e013bfef263c59e47deb0163a5e7651b026673765488bfaf102"]],"created_at":1681938616,"pubkey":"a42007e33cfa25673b26f46f39df039aa6003258a68dc88f1f1e0447607aedb3","id":"e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9","sig":"9036150a6c8a32933cffcc42aec4d2109a22e9f10d1c3860c0435a925e6386babb7df5c95fcf68c8ed6a9726a1f07225af663d0b068eb555014130aad21674fc","meta":{"revision":0,"created":1681939266488,"version":0},"$loki":108}]"#).expect("value");
+
+        relayer
+            .recv_request_from_client(&connection, new_event)
+            .expect("process event");
+
+        // It is not empty
+        assert!(recv.try_recv().is_ok());
+    }
+}

crates/relayer/src/subscription.rs

@@ -0,0 +1,222 @@
+use nostr_rs_types::types::{Event, Filter, Tag};
+
+#[derive(Clone, Debug, Default, Eq, PartialEq, PartialOrd, Ord, Hash)]
+pub struct Subscription {
+    author: Option<Vec<u8>>,
+    ref_public_key: Option<Vec<u8>>,
+    ref_id: Option<Vec<u8>>,
+    id: Option<Vec<u8>>,
+    kind: Option<u32>,
+}
+
+impl Subscription {
+    #[inline]
+    pub fn from_filters(filters: &[Filter]) -> Vec<Subscription> {
+        let mut subs = vec![];
+        filters.iter().for_each(|filter| {
+            let authors: Vec<Option<Vec<u8>>> = if filter.authors.is_empty() {
+                vec![None]
+            } else {
+                filter
+                    .authors
+                    .iter()
+                    .map(|author| Some(author.to_vec()))
+                    .collect()
+            };
+            let ref_public_keys = if filter.references_to_public_key.is_empty() {
+                vec![None]
+            } else {
+                filter
+                    .references_to_public_key
+                    .iter()
+                    .map(|public_key| Some((*public_key).to_vec()))
+                    .collect()
+            };
+            let ref_ids = if filter.references_to_event.is_empty() {
+                vec![None]
+            } else {
+                filter
+                    .references_to_event
+                    .iter()
+                    .map(|id| Some((*id).to_vec()))
+                    .collect()
+            };
+            let kind = if filter.kinds.is_empty() {
+                vec![None]
+            } else {
+                filter
+                    .kinds
+                    .iter()
+                    .map(|kind| Some((*kind).into()))
+                    .collect()
+            };
+
+            let ids = if filter.ids.is_empty() {
+                vec![None]
+            } else {
+                filter.ids.iter().map(|id| Some((*id).to_vec())).collect()
+            };
+
+            for author in authors.iter() {
+                for id in ids.iter() {
+                    for ref_public_key in ref_public_keys.iter() {
+                        for ref_id in ref_ids.iter() {
+                            for kind in kind.iter() {
+                                subs.push(Subscription {
+                                    id: id.clone(),
+                                    ref_public_key: ref_public_key.clone(),
+                                    author: author.clone(),
+                                    ref_id: ref_id.clone(),
+                                    kind: *kind,
+                                });
+                            }
+                        }
+                    }
+                }
+            }
+        });
+
+        subs
+    }
+
+    #[inline]
+    pub fn from_event(event: &Event) -> Vec<Subscription> {
+        let kind = event.kind().into();
+        let public_keys = vec![None, Some(event.author().as_ref().to_vec())];
+        let id = vec![None, Some(event.id.as_ref().to_vec())];
+        let kind = [None, Some(kind)];
+        let mut ref_public_keys = vec![None];
+        let mut ref_ids = vec![None];
+
+        event.tags().iter().for_each(|tag| match tag {
+            Tag::Event(x) => {
+                ref_ids.push(Some(x.id.as_ref().to_vec()));
+            }
+            Tag::PubKey(x) => {
+                ref_public_keys.push(Some(x.id.as_ref().to_vec()));
+            }
+            _ => {}
+        });
+
+        let mut subs = vec![];
+
+        for ref_public_key in ref_public_keys.iter() {
+            for ref_id in ref_ids.iter() {
+                for public_key in public_keys.iter() {
+                    for id in id.iter() {
+                        for kind in kind.iter() {
+                            subs.push(Subscription {
+                                ref_id: ref_id.clone(),
+                                ref_public_key: ref_public_key.clone(),
+                                author: public_key.clone(),
+                                id: id.clone(),
+                                kind: *kind,
+                            });
+                        }
+                    }
+                }
+            }
+        }
+
+        subs
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use nostr_rs_types::{types::Addr, Request};
+
+    #[test]
+    fn test_no_listen_to_all() {
+        let request: Request = serde_json::from_str("[\"REQ\",\"6440d5279e350\",{\"authors\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"since\":1681967101},{\"#p\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[1,3,6,7,9735],\"since\":1681967101},{\"#p\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[4]},{\"authors\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[4]},{\"#e\":[\"2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a\",\"a5e3369c43daf2675ecbce18831e5f4e07db0d4dde0ef4f5698e645e4c46eed1\"],\"kinds\":[1,6,7,9735]}]").expect("valid");
+        let mut subscriptions =
+            Subscription::from_filters(&request.as_request().expect("req").filters);
+        subscriptions.sort();
+
+        assert!(subscriptions
+            .binary_search(&Subscription::default())
+            .is_err())
+    }
+
+    #[test]
+    fn from_filters() {
+        let request: Request = serde_json::from_str("[\"REQ\",\"1298169700973717\",{\"authors\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"since\":1681939304},{\"#p\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[1,3,6,7,9735],\"since\":1681939304},{\"#p\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[4]},{\"authors\":[\"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb\"],\"kinds\":[4]},{\"#e\":[\"2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a\",\"a5e3369c43daf2675ecbce18831e5f4e07db0d4dde0ef4f5698e645e4c46eed1\"],\"kinds\":[1,6,7,9735]}]").expect("valid object");
+        let mut subscriptions =
+            Subscription::from_filters(&request.as_request().expect("req").filters);
+        subscriptions.sort();
+
+        assert!(subscriptions
+            .binary_search(&Subscription::default())
+            .is_err())
+    }
+
+    #[test]
+    fn from_event() {
+        let new_event: Request = serde_json::from_str(r#"["EVENT", {"kind":1,"content":"Pong","tags":[["e","9508850d7ddc8ef58c8b392236c49d472dc23fa11f4e73eb5475dfb099ddff42","","root"],["e","2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a","","reply"],["p","39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb"],["p","ee7202ad91459e013bfef263c59e47deb0163a5e7651b026673765488bfaf102"]],"created_at":1681938616,"pubkey":"a42007e33cfa25673b26f46f39df039aa6003258a68dc88f1f1e0447607aedb3","id":"e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9","sig":"9036150a6c8a32933cffcc42aec4d2109a22e9f10d1c3860c0435a925e6386babb7df5c95fcf68c8ed6a9726a1f07225af663d0b068eb555014130aad21674fc","meta":{"revision":0,"created":1681939266488,"version":0},"$loki":108}]"#).expect("value");
+        let mut subscriptions = Subscription::from_event(&new_event.as_event().expect("event"));
+        subscriptions.sort();
+
+        let pk: Addr = "39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb"
+            .try_into()
+            .expect("id");
+
+        let id: Addr = "e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9"
+            .try_into()
+            .expect("id");
+
+        let ref_id: Addr = "2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a"
+            .try_into()
+            .expect("id");
+
+        let author: Addr = "a42007e33cfa25673b26f46f39df039aa6003258a68dc88f1f1e0447607aedb3"
+            .try_into()
+            .expect("id");
+
+        let expected = vec![
+            Subscription {
+                ref_public_key: Some(pk.as_ref().to_vec()),
+                ..Subscription::default()
+            },
+            Subscription {
+                id: Some(id.as_ref().to_vec()),
+                ref_public_key: Some(pk.as_ref().to_vec()),
+                ..Subscription::default()
+            },
+            Subscription {
+                id: Some(id.as_ref().to_vec()),
+                ref_id: Some(ref_id.as_ref().to_vec()),
+                ref_public_key: Some(pk.as_ref().to_vec()),
+                ..Subscription::default()
+            },
+            Subscription {
+                author: Some(author.as_ref().to_vec()),
+                kind: Some(1),
+                ..Subscription::default()
+            },
+            Subscription {
+                author: Some(author.as_ref().to_vec()),
+                ..Subscription::default()
+            },
+            Subscription {
+                id: Some(id.as_ref().to_vec()),
+                ref_id: Some(ref_id.as_ref().to_vec()),
+                author: Some(author.as_ref().to_vec()),
+                ..Subscription::default()
+            },
+        ];
+
+        assert_eq!(subscriptions.len(), 72);
+        expected.iter().enumerate().for_each(|(i, sub)| {
+            assert!(
+                subscriptions.binary_search(sub).is_ok(),
+                "{} -> {:?}",
+                i,
+                sub
+            );
+        });
+        assert!(subscriptions
+            .binary_search(&Subscription::default())
+            .is_ok());
+    }
+}

crates/relayer/tests/events.json

@@ -0,0 +1,5 @@
+{"id":"9508850d7ddc8ef58c8b392236c49d472dc23fa11f4e73eb5475dfb099ddff42","pubkey":"39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb","created_at":1681934835,"kind":1,"tags":[],"content":"This is a test to check if my relayer will store this event locally and eventually rebroadcast. ","sig":"57f7441771c273011dee3aac2ced426f605767f58a3c3eabdf0aedd6e6e399890ca86df48007d93eec85368df90aace870531c8bbc6706f8e86bf3bd5ca54d0b"}
+{"content":"Ping","created_at":1681938126,"id":"2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a","kind":1,"pubkey":"ee7202ad91459e013bfef263c59e47deb0163a5e7651b026673765488bfaf102","sig":"b5bd0ad13e54b6a832265570ae5e59065bc810c745d177d94ceabe182bfadbffe42dd86b2c4d9ffbbecc4b0ca2476e92935117308c43968b1fbb636ffab008b9","tags":[["e","9508850d7ddc8ef58c8b392236c49d472dc23fa11f4e73eb5475dfb099ddff42","","root"],["e","a5e3369c43daf2675ecbce18831e5f4e07db0d4dde0ef4f5698e645e4c46eed1","","reply"],["p","39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb"]],"meta":{"revision":0,"created":1681939308340,"version":0},"$loki":132}
+{"kind":1,"content":"Pong","tags":[["e","9508850d7ddc8ef58c8b392236c49d472dc23fa11f4e73eb5475dfb099ddff42","","root"],["e","2e72250d80e9b3fd30230b3db3ed7d22f15d266ed345c36700b01ec153c9e28a","","reply"],["p","39a7d06e824c0c2523bedb93f0cef84245e4401fee03b6257a1c6dfd18b57efb"],["p","ee7202ad91459e013bfef263c59e47deb0163a5e7651b026673765488bfaf102"]],"created_at":1681938616,"pubkey":"a42007e33cfa25673b26f46f39df039aa6003258a68dc88f1f1e0447607aedb3","id":"e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9","sig":"9036150a6c8a32933cffcc42aec4d2109a22e9f10d1c3860c0435a925e6386babb7df5c95fcf68c8ed6a9726a1f07225af663d0b068eb555014130aad21674fc","meta":{"revision":0,"created":1681939266488,"version":0},"$loki":108}
+{"content":"🤙","created_at":1681966769,"id":"7136d1c95f84f67558f2f7ed359e4eaa747a5aae74664b8802b4b47362d311be","kind":7,"pubkey":"df94169263fec6182c66e54b63a8e3a1a5f060b742c4512764aee2fe55fc0191","sig":"e9bb3175f100a4349ac68a9ba703e4922b734267a736884da763d29ef23204673b34e17cf8b78f46827aaf626cd55f37782a1531f88e87da5470646c37520622","tags":[["e","e65a6d91fd741a7f9d0b1167da8e32334cbdc00b758ebc15b7997aec3907778a"],["p","4d0068338af5ee79c06513deaaf02492247bbf7abd29f116e6e50c158ab6a677"]]}
+{"content":"🤙","created_at":1681966768,"id":"4b5c4cf53022e2d48a48098d414546479bebe72a96c8b67c4e9450333091aedb","kind":7,"pubkey":"9984188a6578eb513fddcf658f389dbd532e54b82b628ad36666f7aa8f731b79","sig":"a389b5c42b1970b3c470f09b5c1fa360cde3a6224d83048df95d9fdc9f65ef7c47c3981dea414fcae4308cc8d49cd156fdefda282b8e7ff9ec5d0160ea180507","tags":[["e","3f3b150949f38d9bdbe846793a3a92b69748a8c3e346c3b3bad2845cd763714d"],["p","1e067bfb58820576df3daf7cb051d4411b80a0b8fa12fc253cd0ab41cf1a2069"]]}

crates/storage/src/rocksdb.rs

@@ -401,7 +401,7 @@ impl RocksDb {
         // It should be sorted until an internal HashSet can be created, one that
         // respects the inserts time, because the database is sorted already by time
         // desc
-        events.sort_by_key(|b| std::cmp::Reverse(b.created_at()));
+        events.sort_by_key(|b| b.created_at());
 
         Ok(events)
     }
@@ -507,7 +507,7 @@ mod test {
     }
 
     #[test]
-    fn records_are_sorted_by_date_desc() {
+    fn records_are_sorted_by_date_asc() {
         let db = get_db();
 
         let pk: Addr = "7bdef7be22dd8e59f4600e044aa53a1cf975a9dc7d27df5833bc77db784a5805"
@@ -524,7 +524,7 @@ mod test {
 
         let dates = vec.iter().map(|e| e.created_at()).collect::<Vec<_>>();
         let mut sorted_dates = dates.clone();
-        sorted_dates.sort_by(|a, b| b.cmp(a));
+        sorted_dates.sort_by(|a, b| a.cmp(b));
 
         assert_eq!(vec.len(), 10);
         assert_eq!(dates, sorted_dates);

crates/types/src/types/addr.rs

@@ -67,6 +67,15 @@ pub struct Addr {
     pub hrp: Option<HumanReadablePart>,
 }
 
+impl Default for Addr {
+    fn default() -> Self {
+        Addr {
+            bytes: vec![],
+            hrp: None,
+        }
+    }
+}
+
 impl AsRef<[u8]> for Addr {
     fn as_ref(&self) -> &[u8] {
         &self.bytes

src/main.rs

@@ -67,9 +67,12 @@ async fn main() {
     tokio::spawn(async move {
         loop {
             let (event, _) = client_receiver.recv().await.unwrap();
+            println!(
+                "received event: {}",
+                serde_json::to_string(&event).expect("valid json")
+            );
             match &event {
                 Response::Event(event) => {
-                    //println!("Sending event: {:?}", event.event);
                     let _ = relayer_for_client.store_and_broadcast(&event.event);
                 }
                 _msg => {
@@ -83,8 +86,11 @@ async fn main() {
     tokio::spawn(async move {
         loop {
             let x = relayer_for_recv.recv(&mut server_receiver).await.unwrap();
-            //println!("Received: {:?}", x);
             if let Some(x) = x {
+                println!(
+                    "received event: {}",
+                    serde_json::to_string(&x).expect("valid json")
+                );
                 clients.send(x).await;
             }
         }