Improvements from other PRs

Cargo.lock

@@ -42,6 +42,55 @@ dependencies = [
 ]
 
 [[package]]
+name = "anstream"
+version = "0.6.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "64e15c1ab1f89faffbf04a634d5e1962e9074f2741eef6d97f3c4e322426d526"
+dependencies = [
+ "anstyle",
+ "anstyle-parse",
+ "anstyle-query",
+ "anstyle-wincon",
+ "colorchoice",
+ "is_terminal_polyfill",
+ "utf8parse",
+]
+
+[[package]]
+name = "anstyle"
+version = "1.0.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1bec1de6f59aedf83baf9ff929c98f2ad654b97c9510f4e70cf6f661d49fd5b1"
+
+[[package]]
+name = "anstyle-parse"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eb47de1e80c2b463c735db5b217a0ddc39d612e7ac9e2e96a5aed1f57616c1cb"
+dependencies = [
+ "utf8parse",
+]
+
+[[package]]
+name = "anstyle-query"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6d36fc52c7f6c869915e99412912f22093507da8d9e942ceaf66fe4b7c14422a"
+dependencies = [
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "anstyle-wincon"
+version = "3.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5bf74e1b6e971609db8ca7a9ce79fd5768ab6ae46441c572e46cf596f59e57f8"
+dependencies = [
+ "anstyle",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
 name = "async-trait"
 version = "0.1.81"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -215,6 +264,12 @@ dependencies = [
 ]
 
 [[package]]
+name = "colorchoice"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3fd119d74b830634cea2a0f58bbd0d54540518a14397557951e79340abc28c0"
+
+[[package]]
 name = "core-foundation"
 version = "0.9.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -316,6 +371,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "406ac2a8c9eedf8af9ee1489bee9e50029278a6456c740f7454cf8a158abc816"
 
 [[package]]
+name = "env_filter"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f2c92ceda6ceec50f43169f9ee8424fe2db276791afde7b2cd8bc084cb376ab"
+dependencies = [
+ "log",
+ "regex",
+]
+
+[[package]]
 name = "env_logger"
 version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -329,6 +394,19 @@ dependencies = [
 ]
 
 [[package]]
+name = "env_logger"
+version = "0.11.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e13fa619b91fb2381732789fc5de83b45675e882f66623b7d8cb4f643017018d"
+dependencies = [
+ "anstream",
+ "anstyle",
+ "env_filter",
+ "humantime",
+ "log",
+]
+
+[[package]]
 name = "equivalent"
 version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -705,6 +783,12 @@ dependencies = [
 ]
 
 [[package]]
+name = "is_terminal_polyfill"
+version = "1.70.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7943c866cc5cd64cbc25b2e01621d07fa8eb2a1a23160ee81ce38704e97b8ecf"
+
+[[package]]
 name = "itoa"
 version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -810,12 +894,9 @@ dependencies = [
 
 [[package]]
 name = "log"
-version = "0.4.17"
+version = "0.4.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "abb12e687cfb44aa40f41fc3978ef76448f9b6038cad6aef4259d3c095a2382e"
-dependencies = [
- "cfg-if",
-]
+checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"
 
 [[package]]
 name = "lz4-sys"
@@ -874,7 +955,7 @@ dependencies = [
 name = "nostr"
 version = "0.1.0"
 dependencies = [
- "env_logger",
+ "env_logger 0.10.0",
  "futures",
  "futures-util",
  "instant-acme",
@@ -913,7 +994,7 @@ dependencies = [
 name = "nostr-rs-dump"
 version = "0.1.0"
 dependencies = [
- "env_logger",
+ "env_logger 0.10.0",
  "futures",
  "futures-util",
  "instant-acme",
@@ -942,7 +1023,9 @@ dependencies = [
 name = "nostr-rs-personal-relayer"
 version = "0.1.0"
 dependencies = [
+ "env_logger 0.11.5",
  "futures",
+ "log",
  "nostr-rs-client",
  "nostr-rs-memory",
  "nostr-rs-relayer",
@@ -1743,6 +1826,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
 
 [[package]]
+name = "utf8parse"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
+
+[[package]]
 name = "vcpkg"
 version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"

crates/client/src/client.rs

@@ -1,4 +1,4 @@
-use crate::Error;
+use crate::{pool::DEFAULT_CHANNEL_BUFFER_SIZE, Error};
 use futures_util::{SinkExt, StreamExt};
 use nostr_rs_types::{
     client::{self, subscribe},
@@ -73,7 +73,7 @@ impl Drop for Client {
 impl Client {
     /// Creates a new relayer
     pub fn new(send_message_to_listener: mpsc::Sender<(Response, Url)>, url: Url) -> Self {
-        let (sender_to_socket, send_to_socket) = mpsc::channel(100_000);
+        let (sender_to_socket, send_to_socket) = mpsc::channel(DEFAULT_CHANNEL_BUFFER_SIZE);
         let is_connected = Arc::new(AtomicBool::new(false));
 
         let subscriptions = Arc::new(RwLock::new(HashMap::new()));
@@ -113,7 +113,7 @@ impl Client {
             let mut connection_attempts = 0;
 
             loop {
-                log::warn!("{}: Connect attempt {}", url, connection_attempts);
+                log::info!("{}: Connect attempt {}", url, connection_attempts);
                 connection_attempts += 1;
                 let mut socket = match connect_async(url.clone()).await {
                     Ok(x) => x.0,
@@ -195,13 +195,15 @@ impl Client {
 
                             log::info!("New message: {}", msg);
 
-                            let msg: Result<Response, _> = serde_json::from_str(&msg);
+                            let event: Result<Response, _> = serde_json::from_str(&msg);
 
-                            if let Ok(msg) = msg {
+                            if let Ok(msg) = event {
                                 if let Err(error) = send_message_to_listener.try_send((msg, url.clone())) {
                                     log::error!("{}: Reconnecting client because of {}", url, error);
                                     break;
                                 }
+                            } else {
+                                log::error!("Failed to parse message: {:?} {}", event, msg);
                             }
                         }
                         else => {

crates/client/src/pool.rs

@@ -28,7 +28,7 @@ pub struct Pool {
 }
 
 /// Default channel buffer size for the pool
-pub const DEFAULT_CHANNEL_BUFFER_SIZE: usize = 1_000;
+pub const DEFAULT_CHANNEL_BUFFER_SIZE: usize = 100_000;
 
 impl Default for Pool {
     fn default() -> Self {
@@ -168,7 +168,7 @@ impl Pool {
                 });
             }
 
-            clients.insert(url.clone(), client);
+            clients.insert(url, client);
         }
     }
 }
@@ -189,7 +189,7 @@ mod test {
         let local_addr = listener.local_addr().expect("addr");
 
         let relayer = Relayer::new(Some(Memory::default()), None).expect("valid dummy server");
-        let stopper = relayer.main(listener).expect("valid main loop");
+        let (_, stopper) = relayer.main(listener).expect("valid main loop");
         (
             Url::parse(&format!("ws://{}", local_addr)).expect("valid url"),
             stopper,

crates/personal-relayer/Cargo.toml

@@ -13,3 +13,5 @@ thiserror = "1.0.39"
 url = { version = "2.5.2", features = ["serde"] }
 futures = "0.3.30"
 tokio = { version = "1.39.2", features = ["full"] }
+log = "0.4.22"
+env_logger = "0.11.5"

crates/personal-relayer/src/lib.rs

@@ -67,7 +67,8 @@ impl<T: Storage + Send + Sync + 'static> PersonalRelayer<T> {
     }
 
     pub fn main(self, server: TcpListener) -> Result<Stoppable, Error> {
-        let tasks = vec![self.relayer.main(server)?, tokio::spawn(async move {})];
+        let (relayer, handle) = self.relayer.main(server)?;
+        let tasks = vec![handle, tokio::spawn(async move {})];
         Ok(tasks.into())
     }
 }

crates/relayer/src/connection/mod.rs

@@ -2,7 +2,7 @@ use crate::{subscription::ActiveSubscription, Error};
 use futures_util::{SinkExt, StreamExt};
 use nostr_rs_client::PoolSubscription;
 use nostr_rs_types::{
-    relayer::ROk,
+    relayer::{ok::ROkStatus, ROk},
     types::{Addr, SubscriptionId},
     Request, Response,
 };
@@ -149,8 +149,7 @@ impl Connection {
                                     log::error!("Error parsing message from client: {}", err);
                                     let reply: Response = ROk {
                                         id: Addr::default(),
-                                        status: false,
-                                        message: "Error parsing message".to_owned(),
+                                        status: ROkStatus::Error("Error parsing message".to_owned()),
                                     }.into();
                                     let reply = if let Ok(reply) = serde_json::to_string(&reply) {
                                         reply

crates/relayer/src/error.rs

@@ -28,6 +28,10 @@ pub enum Error {
     #[error("Nostr client error: {0}")]
     Client(#[from] nostr_rs_client::Error),
 
+    /// No client connected
+    #[error("Nostr relayer is not connected to any client")]
+    NoClient,
+
     /// Unknown connections
     #[error("Unknown connection: {0}")]
     UnknownConnection(u128),

crates/relayer/src/relayer.rs

@@ -4,9 +4,13 @@ use crate::{
     Connection, Error,
 };
 use futures_util::StreamExt;
-use nostr_rs_client::{Error as ClientError, Pool};
+use nostr_rs_client::{Error as ClientError, Pool, Url};
 use nostr_rs_storage_base::Storage;
-use nostr_rs_types::{relayer, types::Event, Request, Response};
+use nostr_rs_types::{
+    relayer::{self, ROk, ROkStatus},
+    types::{Addr, Event},
+    Request, Response,
+};
 use std::{
     collections::{HashMap, HashSet},
     ops::Deref,
@@ -76,6 +80,13 @@ impl<T: Storage + Send + Sync + 'static> Relayer<T> {
         })
     }
 
+    /// Connects to the relayer pool
+    pub async fn connect_to_relayer(&self, url: Url) -> Result<(), Error> {
+        let (client_pool, _) = self.client_pool.as_ref().ok_or(Error::NoClient)?;
+        client_pool.connect_to(url).await;
+        Ok(())
+    }
+
     /// Total number of subscribers requests that actively listening for new events
     pub fn total_subscribers(&self) -> usize {
         self.subscriptions.total_subscribers()
@@ -91,9 +102,12 @@ impl<T: Storage + Send + Sync + 'static> Relayer<T> {
     ///
     /// This function consumes the object and takes the ownership. The returned
     /// JoinHandle() can be used to stop the main loop
-    pub fn main(self, server: TcpListener) -> Result<JoinHandle<()>, Error> {
+    pub fn main(self, server: TcpListener) -> Result<(Arc<Self>, JoinHandle<()>), Error> {
         let (this, mut receiver) = self.split()?;
-        Ok(tokio::spawn(async move {
+        let _self = Arc::new(this);
+        let this = _self.clone();
+
+        let handle = tokio::spawn(async move {
             loop {
                 tokio::select! {
                     Ok((stream, _)) = server.accept() => {
@@ -103,12 +117,12 @@ impl<T: Storage + Send + Sync + 'static> Relayer<T> {
                     Some((conn_id, request)) = receiver.recv() => {
                         // receive messages from the connection pool
                         if conn_id.is_empty() {
-                            // connection pool
+                            // message received from client pool
                             if let Request::Event(event) = request {
+                                let _ = this.broadcast(event.deref()).await;
                                 if let Some(storage) = this.storage.as_ref() {
                                     let _ = storage.store_local_event(&event).await;
                                 }
-                                this.broadcast(event.deref()).await;
                             }
                             continue;
                         }
@@ -128,7 +142,9 @@ impl<T: Storage + Send + Sync + 'static> Relayer<T> {
                     }
                 }
             }
-        }))
+        });
+
+        Ok((_self, handle))
     }
 
     /// Handle the client pool
@@ -136,21 +152,22 @@ impl<T: Storage + Send + Sync + 'static> Relayer<T> {
     /// Main loop to consume messages from the client pool and broadcast them to the local subscribers
     fn handle_client_pool(
         client_pool: Pool,
-        sender: Sender<(ConnectionId, Request)>,
+        send_message_to_relayer: Sender<(ConnectionId, Request)>,
     ) -> Result<(Pool, JoinHandle<()>), ClientError> {
         let (mut receiver, client_pool) = client_pool.split()?;
 
         let handle = tokio::spawn(async move {
             loop {
+                if receiver.len() > 500 {
+                    println!("{}", receiver.len());
+                }
                 if let Some((response, _)) = receiver.recv().await {
                     match response {
                         Response::Event(event) => {
-                            let _ = sender
-                                .send((
-                                    ConnectionId::new_empty(),
-                                    Request::Event(event.event.into()),
-                                ))
-                                .await;
+                            let _ = send_message_to_relayer.try_send((
+                                ConnectionId::new_empty(),
+                                Request::Event(event.event.into()),
+                            ));
                         }
                         Response::EndOfStoredEvents(_) => {}
                         x => {
@@ -200,32 +217,45 @@ impl<T: Storage + Send + Sync + 'static> Relayer<T> {
         &self,
         connection: &Connection,
         request: Request,
-    ) -> Result<Option<Request>, Error> {
-        match &request {
+    ) -> Result<(), Error> {
+        match request {
             Request::Event(event) => {
-                if let Some(storage) = self.storage.as_ref() {
-                    let _ = storage.store(event).await;
-                    let _ = storage.store_local_event(event).await;
+                let event_id: Addr = event.id.clone().into();
+                if !self.broadcast(&event).await? {
+                    connection.send(
+                        ROk {
+                            id: event_id,
+                            status: ROkStatus::Duplicate,
+                        }
+                        .into(),
+                    )?;
+                    return Ok(());
                 }
 
-                self.broadcast(event).await;
+                if let Some(storage) = self.storage.as_ref() {
+                    let _ = storage.store_local_event(&event).await;
+                }
 
                 if let Some((client_pool, _)) = self.client_pool.as_ref() {
                     // pass the event to the pool of clients, so this relayer can relay
                     // their local events to the clients in the network of relayers
-                    let _ = client_pool.post(event.clone()).await;
+                    let _ = client_pool.post(event).await;
                 }
+
+                connection.send(
+                    ROk {
+                        id: event_id,
+                        status: ROkStatus::Ok,
+                    }
+                    .into(),
+                )?;
             }
             Request::Request(request) => {
                 let foreign_subscription = if let Some((client_pool, _)) = self.client_pool.as_ref()
                 {
                     // pass the subscription request to the pool of clients, so this relayer
                     // can relay any unknown event to the clients through their subscriptions
-                    Some(
-                        client_pool
-                            .subscribe(request.filters.clone().into())
-                            .await?,
-                    )
+                    Some(client_pool.subscribe(request.clone()).await?)
                 } else {
                     None
                 };
@@ -272,21 +302,24 @@ impl<T: Storage + Send + Sync + 'static> Relayer<T> {
                     .await;
             }
             Request::Close(close) => {
-                connection.unsubscribe(close).await;
+                connection.unsubscribe(&close).await;
             }
         };
 
-        Ok(Some(request))
+        Ok(())
     }
 
     #[inline]
     /// Broadcast a given event to all local subscribers
-    pub async fn broadcast(&self, event: &Event) {
+    pub async fn broadcast(&self, event: &Event) -> Result<bool, Error> {
         if let Some(storage) = self.storage.as_ref() {
-            let _ = storage.store(event).await;
+            if !storage.store(event).await? {
+                return Ok(false);
+            }
         }
 
         self.subscriptions.broadcast(event.clone());
+        Ok(true)
     }
 }
 
@@ -314,7 +347,7 @@ mod test {
 
         let relayer =
             Relayer::new(Some(Memory::default()), client_pool).expect("valid dummy server");
-        let stopper = relayer.main(listener).expect("valid main loop");
+        let (_, stopper) = relayer.main(listener).expect("valid main loop");
         (
             Url::parse(&format!("ws://{}", local_addr)).expect("valid url"),
             stopper,
@@ -393,6 +426,18 @@ mod test {
         let _ = relayer
             .process_request_from_client(&connection, request)
             .await;
+
+        // ev1
+        assert_eq!(
+            ROkStatus::Ok,
+            recv.try_recv()
+                .expect("valid")
+                .as_ok()
+                .cloned()
+                .unwrap()
+                .status,
+        );
+
         // ev1
         assert_eq!(
             note,
@@ -475,7 +520,6 @@ mod test {
                 .expect("valid")
                 .as_event()
                 .expect("event")
-                .event
                 .id
                 .to_string()
         );
@@ -486,7 +530,6 @@ mod test {
                 .expect("valid")
                 .as_event()
                 .expect("event")
-                .event
                 .id
                 .to_string()
         );
@@ -497,7 +540,6 @@ mod test {
                 .expect("valid")
                 .as_event()
                 .expect("event")
-                .event
                 .id
                 .to_string()
         );
@@ -567,6 +609,14 @@ mod test {
 
         sleep(Duration::from_millis(100)).await;
 
+        // ok from posting
+        let msg = recv.try_recv();
+        assert!(msg.is_ok());
+        assert_eq!(
+            msg.expect("is ok").as_ok().expect("valid").id.to_hex(),
+            "e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9".to_owned(),
+        );
+
         // It is not empty
         let msg = recv.try_recv();
         assert!(msg.is_ok());
@@ -651,6 +701,14 @@ mod test {
 
         sleep(Duration::from_millis(100)).await;
 
+        // ok from posting
+        let msg = recv.try_recv();
+        assert!(msg.is_ok());
+        assert_eq!(
+            msg.expect("is ok").as_ok().expect("valid").id.to_hex(),
+            "e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9".to_owned(),
+        );
+
         // It is not empty
         let msg = recv.try_recv();
         assert!(msg.is_ok());
@@ -688,7 +746,7 @@ mod test {
         .expect("valid object");
 
         let relayer = Relayer::new(Some(get_db(false).await), None).expect("valid relayer");
-        let (publisher, _) = Connection::new_local_connection();
+        let (publisher, mut recv) = Connection::new_local_connection();
 
         let mut set1 = (0..1000)
             .map(|_| Connection::new_local_connection())
@@ -746,6 +804,13 @@ mod test {
 
         sleep(Duration::from_millis(10)).await;
 
+        let msg = recv.try_recv();
+        assert!(msg.is_ok());
+        assert_eq!(
+            msg.expect("is ok").as_ok().expect("valid").id.to_hex(),
+            "e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9".to_owned(),
+        );
+
         for (_, recv) in set1.iter_mut() {
             assert!(recv.try_recv().is_err());
         }
@@ -776,6 +841,33 @@ mod test {
     }
 
     #[tokio::test]
+    async fn posting_event_replies_ok() {
+        let relayer = Relayer::new(Some(get_db(false).await), None).expect("valid relayer");
+        let (connection, mut recv) = Connection::new_local_connection();
+
+        let note = get_note();
+        let note_id = note.as_event().map(|x| x.id.clone()).unwrap();
+
+        relayer
+            .process_request_from_client(&connection, note)
+            .await
+            .expect("process event");
+
+        sleep(Duration::from_millis(10)).await;
+
+        assert_eq!(
+            Some(
+                ROk {
+                    id: note_id.into(),
+                    status: ROkStatus::Ok,
+                }
+                .into()
+            ),
+            recv.try_recv().ok()
+        );
+    }
+
+    #[tokio::test]
     async fn subscribe_to_all() {
         let request: Request =
             serde_json::from_value(json!(["REQ", "1298169700973717", {}])).expect("valid object");
@@ -807,6 +899,14 @@ mod test {
 
         sleep(Duration::from_millis(10)).await;
 
+        // ok from posting
+        let msg = recv.try_recv();
+        assert!(msg.is_ok());
+        assert_eq!(
+            msg.expect("is ok").as_ok().expect("valid").id.to_hex(),
+            "e862fe23daf52ab09b36a37fa91ca3743e0c323e630e8627891212ca147c2da9".to_owned(),
+        );
+
         // It is not empty
         let msg = recv.try_recv();
         assert!(msg.is_ok());
@@ -894,19 +994,19 @@ mod test {
         assert_eq!(
             responses
                 .get(&relayer1.port().expect("port"))
-                .map(|x| x.event.id.clone()),
+                .map(|x| x.id.clone()),
             Some(signed_content.id.clone())
         );
         assert_eq!(
             responses
                 .get(&relayer2.port().expect("port"))
-                .map(|x| x.event.id.clone()),
+                .map(|x| x.id.clone()),
             Some(signed_content.id.clone())
         );
         assert_eq!(
             responses
                 .get(&relayer3.port().expect("port"))
-                .map(|x| x.event.id.clone()),
+                .map(|x| x.id.clone()),
             Some(signed_content.id)
         );
     }
@@ -942,7 +1042,7 @@ mod test {
 
         let account1 = Account::default();
         let signed_content = account1
-            .sign_content(vec![], Content::ShortTextNote("test 0".to_owned()), None)
+            .sign_content(vec![], Content::ShortTextNote("test 01".to_owned()), None)
             .expect("valid signed content");
 
         // account1 posts a new note into the relayer1, and the main relayer
@@ -956,8 +1056,8 @@ mod test {
             Some((signed_content.id, signed_content.signature)),
             main_client
                 .try_recv()
-                .and_then(|(r, _)| r.as_event().cloned().map(|x| x.event))
-                .map(|x| (x.id, x.signature))
+                .and_then(|(r, _)| r.as_event().cloned())
+                .map(|x| (x.id.clone(), x.signature.clone()))
         );
         assert!(main_client.try_recv().is_none());
     }

crates/types/src/lib.rs

@@ -13,6 +13,21 @@ pub mod types;
 
 pub use self::{request::Request, response::Response};
 
+#[cfg(test)]
+mod regression {
+    use crate::types::Event;
+
+    #[test]
+    fn event() {
+        include_str!("../tests/regression_parsing.json")
+            .lines()
+            .for_each(|line| {
+                let event: Event = serde_json::from_str(line).unwrap();
+                assert!(event.is_valid().is_ok(), "Failed to parse: {}", line);
+            });
+    }
+}
+
 #[macro_use]
 extern crate custom_derive;
 #[macro_use]

crates/types/src/relayer/auth.rs

@@ -10,7 +10,7 @@ use std::{collections::VecDeque, ops::Deref};
 
 /// This is how a relayer sends an authentication challenge to the client. The
 /// challenge must be returned signed by the client
-#[derive(Clone, Debug)]
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub struct Auth(pub String);
 
 impl Default for Auth {

crates/types/src/relayer/eose.rs

@@ -13,7 +13,7 @@ use std::{collections::VecDeque, ops::Deref};
 /// This is how to the relayer signals the client they gave all the stored
 /// messages with the requested filter. Future events will be sent, as they
 /// appear to this relayer
-#[derive(Clone, Debug)]
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub struct EndOfStoredEvents(pub SubscriptionId);
 
 impl Deref for EndOfStoredEvents {

crates/types/src/relayer/event.rs

@@ -4,10 +4,10 @@ use crate::{
     types::{self, subscription_id::Error, SubscriptionId},
 };
 use serde_json::Value;
-use std::collections::VecDeque;
+use std::{collections::VecDeque, ops::Deref};
 
 /// An event sent to clients
-#[derive(Clone, Debug)]
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub struct Event {
     /// The subscription ID that matched the inner event
     pub subscription_id: SubscriptionId,
@@ -15,6 +15,14 @@ pub struct Event {
     pub event: types::Event,
 }
 
+impl Deref for Event {
+    type Target = types::Event;
+
+    fn deref(&self) -> &Self::Target {
+        &self.event
+    }
+}
+
 impl From<(&SubscriptionId, &types::Event)> for Event {
     fn from((subscription_id, event): (&SubscriptionId, &types::Event)) -> Self {
         Self {

crates/types/src/relayer/mod.rs

@@ -8,4 +8,10 @@ pub mod event;
 pub mod notice;
 pub mod ok;
 
-pub use self::{auth::Auth, eose::EndOfStoredEvents, event::Event, notice::Notice, ok::ROk};
+pub use self::{
+    auth::Auth,
+    eose::EndOfStoredEvents,
+    event::Event,
+    notice::Notice,
+    ok::{ROk, ROkStatus},
+};

crates/types/src/relayer/notice.rs

@@ -8,7 +8,7 @@ use serde_json::Value;
 use std::{collections::VecDeque, ops::Deref};
 
 /// Notices are errors
-#[derive(Clone, Debug)]
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub struct Notice(pub String);
 
 impl Deref for Notice {

crates/types/src/relayer/ok.rs

@@ -11,15 +11,79 @@ use crate::{
 use serde_json::Value;
 use std::collections::VecDeque;
 
+#[derive(Clone, PartialEq, Eq, Debug)]
+/// ROkStatus
+pub enum ROkStatus {
+    /// Everything is ok
+    Ok,
+    /// Proof of work failed
+    Pow,
+    /// Event is known already
+    Duplicate,
+    /// User is blocked
+    Blocked,
+    /// User is blacklisted
+    Blacklisted,
+    /// Rate limit exceeded
+    RateLimit,
+    /// Invalid event
+    Invalid,
+    /// Error
+    Error(String),
+}
+
+impl From<(Addr, ROkStatus)> for ROk {
+    fn from((id, status): (Addr, ROkStatus)) -> Self {
+        Self { id, status }
+    }
+}
+
+impl ROkStatus {
+    /// Compute readable status string
+    pub fn computer_readable(&self) -> String {
+        match self {
+            ROkStatus::Ok => "",
+            ROkStatus::Pow => "pow",
+            ROkStatus::Duplicate => "duplicate",
+            ROkStatus::Blocked => "blocked",
+            ROkStatus::Blacklisted => "blocked",
+            ROkStatus::RateLimit => "rate-limit",
+            ROkStatus::Invalid => "invalid",
+            ROkStatus::Error(_) => "error",
+        }
+        .to_owned()
+    }
+
+    /// Human readable status
+    pub fn to_string(&self) -> String {
+        let message = match self {
+            ROkStatus::Ok => "",
+            ROkStatus::Pow => "difficulty",
+            ROkStatus::Duplicate => "already have this event",
+            ROkStatus::Blocked => "try again later",
+            ROkStatus::Blacklisted => "you are banned from posting here",
+            ROkStatus::RateLimit => "slow down there chief",
+            ROkStatus::Invalid => "event creation date is too far off from the current time",
+            ROkStatus::Error(e) => e,
+        }
+        .to_owned();
+
+        match self {
+            ROkStatus::Ok => "".to_owned(),
+            _ => {
+                format!("{}:{}", self.computer_readable(), message)
+            }
+        }
+    }
+}
+
 /// OK messages
-#[derive(Clone, Debug)]
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub struct ROk {
     /// Event Id
     pub id: Addr,
-    /// Whether the event has been successful or not
-    pub status: bool,
-    /// Some message
-    pub message: String,
+    /// Status
+    pub status: ROkStatus,
 }
 
 impl SerializeDeserialize for ROk {
@@ -31,8 +95,8 @@ impl SerializeDeserialize for ROk {
         Ok(vec![
             Value::String(Self::get_tag().to_owned()),
             Value::String(self.id.to_hex()),
-            Value::Bool(self.status),
-            Value::String(self.message.clone()),
+            Value::Bool(self.status == ROkStatus::Ok),
+            Value::String(self.status.computer_readable()),
         ])
     }
 
@@ -52,8 +116,25 @@ impl SerializeDeserialize for ROk {
             .to_owned();
         Ok(Self {
             id,
-            status,
-            message,
+            status: if status {
+                ROkStatus::Ok
+            } else {
+                let part = message
+                    .split(":")
+                    .next()
+                    .map(|x| x.trim().to_owned())
+                    .unwrap_or_default();
+                match part.as_str() {
+                    "pow" => ROkStatus::Pow,
+                    "duplicate" => ROkStatus::Duplicate,
+                    "blocked" => ROkStatus::Blocked,
+                    "blacklisted" => ROkStatus::Blacklisted,
+                    "rate-limit" => ROkStatus::RateLimit,
+                    "invalid" => ROkStatus::Invalid,
+                    "error" => ROkStatus::Error(message),
+                    _ => ROkStatus::Error(message),
+                }
+            },
         })
     }
 }
@@ -74,12 +155,11 @@ mod test {
     fn serialize() {
         let ok_ = ROk {
             id: "a0b0".try_into().expect("valid_id"),
-            status: true,
-            message: "Some test".into(),
+            status: ROkStatus::Ok,
         };
         let m: Response = ok_.into();
         assert_eq!(
-            r#"["OK","a0b0",true,"Some test"]"#,
+            r#"["OK","a0b0",true,""]"#,
             serde_json::to_string(&m).expect("valid json")
         );
     }

crates/types/src/response.rs

@@ -10,7 +10,7 @@ use serde::{
 use std::collections::VecDeque;
 
 custom_derive! {
-#[derive(Debug, Clone, EnumFromInner)]
+#[derive(Debug, Clone, PartialEq, Eq, EnumFromInner)]
     /// Response Message
     ///
     /// All responses from relayers to clients are abstracted in this struct

crates/types/src/types/addr.rs

@@ -13,6 +13,7 @@ use std::{
     fmt::{self, Display},
     hash::Hash,
     ops::Deref,
+    str::FromStr,
 };
 use thiserror::Error;
 
@@ -173,6 +174,14 @@ impl Addr {
 /// The first approach is try decoding the the Addr as a hex-encoded string.
 ///
 /// The second approach is decoding with bech32
+impl FromStr for Addr {
+    type Err = Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        s.try_into()
+    }
+}
+
 impl TryFrom<&str> for Addr {
     type Error = Error;
 
@@ -195,6 +204,11 @@ impl TryFrom<&str> for Addr {
     }
 }
 
+/// Converts an Addr from an string
+///
+/// The first approach is try decoding the the Addr as a hex-encoded string.
+///
+/// The second approach is decoding with bech32
 impl TryFrom<String> for Addr {
     type Error = Error;
 

crates/types/src/types/event.rs

@@ -202,6 +202,11 @@ impl Event {
         &self.inner.content
     }
 
+    /// Consumes the event and returns the content
+    pub fn take_content(self) -> Content {
+        self.inner.content
+    }
+
     /// Verifies if the Id, Public Key and Signature are correct
     fn verify_signature(
         public_key: &Id,

crates/types/src/types/tag.rs

@@ -30,23 +30,7 @@ pub enum Marker {
     Unknown(String),
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
-/// Url content or empty string
-pub enum UrlOrEmpty {
-    /// Url
-    Url(Url),
-    /// Empty string
-    Empty,
-}
-
-impl Display for UrlOrEmpty {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        match self {
-            Self::Url(url) => write!(f, "{}", url),
-            Self::Empty => write!(f, ""),
-        }
-    }
-}
+type ParsedUrl = (Option<Url>, String);
 
 impl Display for Marker {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
@@ -96,11 +80,11 @@ pub enum RelayAccessType {
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub enum Tag {
     /// Tag another event
-    Event(Id, Option<UrlOrEmpty>, Option<Marker>),
+    Event(Id, Option<ParsedUrl>, Option<Marker>),
     /// Tag another public key
-    PubKey(Id, Option<UrlOrEmpty>, Option<String>),
+    PubKey(Id, Option<ParsedUrl>, Option<String>),
     /// Tag a relay
-    Relay(Url, RelayAccessType),
+    Relay(ParsedUrl, RelayAccessType),
     /// Tag a hashtag
     Hashtag(String),
     /// Tag with an external content id
@@ -116,7 +100,7 @@ pub enum Tag {
     /// Image - NIP-23, NIP-52, NIP-58
     Image(Url, Option<String>),
     /// Zap Goal - NIP-75
-    ZapGoal(Id, Option<UrlOrEmpty>),
+    ZapGoal(Id, Option<ParsedUrl>),
     /// Weird, supported nonetheless
     Empty,
 }
@@ -157,7 +141,7 @@ impl Tag {
             Tag::PubKey(key, _, _) => Some(TagValue::Id(key.clone())),
             Tag::ExternalContentId(id, _) => Some(TagValue::String(id.clone())),
             Tag::Hashtag(content) | Tag::Title(content) => Some(TagValue::String(content.clone())),
-            Tag::Relay(url, _) => Some(TagValue::String(url.to_string())),
+            Tag::Relay((_, url), _) => Some(TagValue::String(url.to_string())),
             Tag::Unknown(_, args) => {
                 let value = args.get(0).cloned().unwrap_or_default();
                 Some(
@@ -200,8 +184,8 @@ impl ser::Serialize for Tag {
             }
             Tag::Event(event, relayer_url, marker) => {
                 seq.serialize_element(&event.to_string())?;
-                if let Some(relayer) = &relayer_url {
-                    seq.serialize_element(&relayer.to_string())?;
+                if let Some((_, relayer)) = relayer_url {
+                    seq.serialize_element(&relayer)?;
                     if let Some(marker) = &marker {
                         seq.serialize_element(&marker.to_string())?;
                     }
@@ -209,8 +193,8 @@ impl ser::Serialize for Tag {
             }
             Tag::ZapGoal(event, relayer_url) => {
                 seq.serialize_element(&event.to_string())?;
-                if let Some(relayer) = &relayer_url {
-                    seq.serialize_element(&relayer.to_string())?;
+                if let Some((_, relayer)) = relayer_url {
+                    seq.serialize_element(&relayer)?;
                 }
             }
             Tag::ExternalContentId(content, url) => {
@@ -224,14 +208,14 @@ impl ser::Serialize for Tag {
             }
             Tag::PubKey(key, relayer_url, pet_name) => {
                 seq.serialize_element(&key.to_string())?;
-                if let Some(relayer) = &relayer_url {
-                    seq.serialize_element(&relayer.to_string())?;
+                if let Some((_, relayer)) = relayer_url {
+                    seq.serialize_element(&relayer)?;
                     if let Some(pet_name) = &pet_name {
                         seq.serialize_element(pet_name)?;
                     }
                 }
             }
-            Tag::Relay(url, access) => {
+            Tag::Relay((_, url), access) => {
                 seq.serialize_element(url)?;
 
                 if let Some(access) = match access {
@@ -288,26 +272,14 @@ impl<'de> Deserialize<'de> for Tag {
                 .ok_or_else::<D::Error, _>(|| de::Error::custom("missing argument"))
                 .and_then(|id| id.parse().map_err(de::Error::custom))
                 .and_then(|id| {
-                    let relayer_url = parts
-                        .pop_front()
-                        .map(|value| {
-                            if value.is_empty() {
-                                Ok(UrlOrEmpty::Empty)
-                            } else {
-                                value.parse().map(UrlOrEmpty::Url)
-                            }
-                        })
-                        .transpose()
-                        .map_err(de::Error::custom);
-
-                    relayer_url.map(|relayer_url| {
-                        let extra = parts.pop_front();
-                        match tag_type.as_str() {
-                            "e" => Tag::Event(id, relayer_url, extra.map(|x| x.as_str().into())),
-                            "goal" => Tag::ZapGoal(id, relayer_url),
-                            "p" => Tag::PubKey(id, relayer_url, extra),
-                            _ => unreachable!(),
-                        }
+                    let relayer_url = parts.pop_front().map(|value| (value.parse().ok(), value));
+
+                    let extra = parts.pop_front();
+                    Ok(match tag_type.as_str() {
+                        "e" => Tag::Event(id, relayer_url, extra.map(|x| x.as_str().into())),
+                        "goal" => Tag::ZapGoal(id, relayer_url),
+                        "p" => Tag::PubKey(id, relayer_url, extra),
+                        _ => unreachable!(),
                     })
                 }),
             "expiration" => {
@@ -348,7 +320,11 @@ impl<'de> Deserialize<'de> for Tag {
                 let url = parts
                     .pop_front()
                     .ok_or_else::<D::Error, _>(|| de::Error::custom("missing url"))
-                    .and_then(|url| url.parse().map_err(de::Error::custom));
+                    .and_then(|url| {
+                        url.parse::<Url>()
+                            .map_err(de::Error::custom)
+                            .map(|parsed_url| (Some(parsed_url), url))
+                    });
 
                 let access = parts
                     .pop_front()
@@ -437,7 +413,7 @@ mod test {
                 "8fe53b37518e3dbe9bab26d912292001d8b882de9456b7b08b615f912dc8bf4a"
                     .parse()
                     .unwrap(),
-                Some(UrlOrEmpty::Empty),
+                Some((None, "".to_owned())),
                 Some("mention".to_owned()),
             ),
             Tag::Event(
@@ -461,10 +437,13 @@ mod test {
 
     #[test]
     fn test_relay() {
-        let json = json!(["r", "https://example.com", "read"]);
+        let json = json!(["r", "https://example.com/", "read"]);
         assert_eq!(
             Tag::Relay(
-                "https://example.com".parse().expect("valid url"),
+                (
+                    Some("https://example.com".parse().expect("valid url")),
+                    "https://example.com/".to_owned()
+                ),
                 RelayAccessType::Read
             ),
             serde_json::from_value(json).expect("valid json"),
@@ -473,7 +452,10 @@ mod test {
         let json = json!(["r", "https://example.com", "write"]);
         assert_eq!(
             Tag::Relay(
-                "https://example.com".parse().expect("valid url"),
+                (
+                    Some("https://example.com".parse().expect("valid url")),
+                    "https://example.com".to_owned()
+                ),
                 RelayAccessType::Write
             ),
             serde_json::from_value(json).expect("valid json"),
@@ -482,7 +464,10 @@ mod test {
         let json = json!(["r", "https://example.com"]);
         assert_eq!(
             Tag::Relay(
-                "https://example.com".parse().expect("valid url"),
+                (
+                    Some("https://example.com".parse().expect("valid url")),
+                    "https://example.com".to_owned()
+                ),
                 RelayAccessType::Both,
             ),
             serde_json::from_value(json).expect("valid json"),
@@ -611,7 +596,7 @@ mod test {
                 "d45a98f898820258a3313f5cb14f5fe8a9263437931ac6309f23ae0324833f39"
                     .parse()
                     .unwrap(),
-                None
+                None,
             ),
         );
     }

crates/types/tests/regression_parsing.json

@@ -0,0 +1,2 @@
+{"content":"🤙","created_at":1724806560,"id":"f4b1461f37dc847f6dea3dbc1b5470c4dceb23c4046ebbcb80807f62f520f371","kind":7,"pubkey":"b2815682cfc83fcd2c3add05785cf4573dd388457069974cc6d8cca06b3c3b78","sig":"5d6bcf31b72d8a189e0cd2095bf64bf052f0ec88ec0e5c6a669f546b14306204feffa0b453a9591639710fe6341ae53589dcbd0cda2752428bd986ccaf99bf6e","tags":[["e","078fc44df2ba89ae08e2d327d5ab43f31e60765b7517342a74e8a180b943fd71","wss://relay.primal.net","root"],["p","b2815682cfc83fcd2c3add05785cf4573dd388457069974cc6d8cca06b3c3b78"],["e","0612860a3a567a531bde1076a407c50f086f56246f5a0f8e99e061d23cd040b5"],["p","d7df5567015930b17c125b3a7cf29bef23aa5a68d09cd6518d291359606aab7b"]]}
+{"content":"🤣 https://image.nostr.build/f0f21864cb22642fc5d83cd96fc55d5d85a2e469e46e191be50bc71a7a1d9740.jpg ","created_at":1719979685,"id":"85616bf509f2612677a1e59f94cceab07f06fb37240597b260fee42c0295992b","kind":1,"pubkey":"b2815682cfc83fcd2c3add05785cf4573dd388457069974cc6d8cca06b3c3b78","sig":"f3e993413b2b4f4579e535c1559e982bb460446d47c02a7ab2e884fd996da31ea999ad89f2ad707800ab4dbf6966819dfd2c93f2d0f4ce64f79430f00b16c8ed","tags":[["e","4668bb2cc0a1fdb5b328810a22b9bf2e8425ff82a228347ddf97e2d9797cf41b","","root"],["p","32e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245"],["imeta","url https://image.nostr.build/f0f21864cb22642fc5d83cd96fc55d5d85a2e469e46e191be50bc71a7a1d9740.jpg","blurhash eEFOc0.R0hH[n#%E${%JD+Mzb=xoxZxtIq.7-:xtRjRQxZWUxWn$NH","dim 1280x720"],["r","https://image.nostr.build/f0f21864cb22642fc5d83cd96fc55d5d85a2e469e46e191be50bc71a7a1d9740.jpg"]]}

src/main.rs

@@ -72,7 +72,7 @@ async fn main() {
     let addr = "127.0.0.1:3000";
     let listener: TcpListener = TcpListener::bind(&addr).await.unwrap();
 
-    let _ = relayer.main(listener).expect("valid main").await;
+    let _ = relayer.main(listener).expect("valid main");
 
     /*
     loop {