Command result notices

This introduces a new notice-like event with more structure so that clients
can know if an event is sucessfully written to the database. Clients
can't really do much with the current NOTICE messages, but with these
structured result messages, it can know if an event was sucessfully
saved.

Whenever there is an error and we have an id available (event id, sub
id, etc) we return these structured OK events instead.

Example:

When saving the following event:
    ["EVENT",{"id": "event_id" }]

The server will now return events in the following format:
    ["OK", event_id, "true|false", message]

For example, on a successful save:
    ["OK", "event_id", "true"]

If we already have the event:
    ["OK", "event_id", "true", "duplicate"]

If the event is rejected:
    ["OK", "event_id", "false", "you are blocked"]

If a subscription fails:
    ["OK", "sub_id", "false", "Too many subscriptions"]

NIP coming soon!

src/conn.rs

@@ -2,7 +2,6 @@
 use crate::close::Close;
 use crate::error::Error;
 use crate::error::Result;
-use crate::event::Event;
 
 use crate::subscription::Subscription;
 use std::collections::HashMap;
@@ -43,6 +42,10 @@ impl ClientConn {
         }
     }
 
+    pub fn subscriptions(&self) -> &HashMap<String, Subscription> {
+        &self.subscriptions
+    }
+
     /// Get a short prefix of the client's unique identifier, suitable
     /// for logging.
     #[must_use]
@@ -55,18 +58,6 @@ impl ClientConn {
         &self.client_ip
     }
 
-    /// Find all matching subscriptions.
-    #[must_use]
-    pub fn get_matching_subscriptions(&self, e: &Event) -> Vec<&str> {
-        let mut v: Vec<&str> = vec![];
-        for (id, sub) in &self.subscriptions {
-            if sub.interested_in_event(e) {
-                v.push(id);
-            }
-        }
-        v
-    }
-
     /// Add a new subscription for this connection.
     /// # Errors
     ///
@@ -110,9 +101,9 @@ impl ClientConn {
         // TODO: return notice if subscription did not exist.
         self.subscriptions.remove(&c.id);
         debug!(
-            "removed subscription, currently have {} active subs (cid={:?})",
+            "removed subscription, currently have {} active subs (cid={})",
             self.subscriptions.len(),
-            self.client_id
+            self.get_client_prefix(),
         );
     }
 }

src/db.rs

@@ -6,6 +6,7 @@ use crate::event::{single_char_tagname, Event};
 use crate::hexrange::hex_range;
 use crate::hexrange::HexSearch;
 use crate::nip05;
+use crate::notice::Notice;
 use crate::schema::{upgrade_db, STARTUP_SQL};
 use crate::subscription::ReqFilter;
 use crate::subscription::Subscription;
@@ -32,7 +33,7 @@ pub type PooledConnection = r2d2::PooledConnection<r2d2_sqlite::SqliteConnection
 /// Events submitted from a client, with a return channel for notices
 pub struct SubmittedEvent {
     pub event: Event,
-    pub notice_tx: tokio::sync::mpsc::Sender<String>,
+    pub notice_tx: tokio::sync::mpsc::Sender<Notice>,
 }
 
 /// Database file
@@ -158,7 +159,9 @@ pub async fn db_writer(
                         event.get_event_id_prefix()
                     );
                     notice_tx
-                        .try_send("pubkey is not allowed to publish to this relay".to_owned())
+                        .try_send(Notice::message(
+                            "pubkey is not allowed to publish to this relay".to_owned(),
+                        ))
                         .ok();
                     continue;
                 }
@@ -189,10 +192,10 @@ pub async fn db_writer(
                                   event.get_author_prefix()
                             );
                             notice_tx
-                                .try_send(
+                                .try_send(Notice::message(
                                     "NIP-05 verification is no longer valid (expired/wrong domain)"
                                         .to_owned(),
-                                )
+                                ))
                                 .ok();
                             continue;
                         }
@@ -203,7 +206,9 @@ pub async fn db_writer(
                             event.get_author_prefix()
                         );
                         notice_tx
-                            .try_send("NIP-05 verification needed to publish events".to_owned())
+                            .try_send(Notice::message(
+                                "NIP-05 verification needed to publish events".to_owned(),
+                            ))
                             .ok();
                         continue;
                     }
@@ -218,7 +223,7 @@ pub async fn db_writer(
             if event.kind >= 20000 && event.kind < 30000 {
                 bcast_tx.send(event.clone()).ok();
                 info!(
-                    "published ephemeral event {:?} from {:?} in {:?}",
+                    "published ephemeral event: {:?} from: {:?} in: {:?}",
                     event.get_event_id_prefix(),
                     event.get_author_prefix(),
                     start.elapsed()
@@ -229,9 +234,10 @@ pub async fn db_writer(
                     Ok(updated) => {
                         if updated == 0 {
                             trace!("ignoring duplicate or deleted event");
+                            notice_tx.try_send(Notice::duplicate(event.id)).ok();
                         } else {
                             info!(
-                                "persisted event {:?} from {:?} in {:?}",
+                                "persisted event: {:?} from: {:?} in: {:?}",
                                 event.get_event_id_prefix(),
                                 event.get_author_prefix(),
                                 start.elapsed()
@@ -239,16 +245,14 @@ pub async fn db_writer(
                             event_write = true;
                             // send this out to all clients
                             bcast_tx.send(event.clone()).ok();
+                            notice_tx.try_send(Notice::saved(event.id)).ok();
                         }
                     }
                     Err(err) => {
                         warn!("event insert failed: {:?}", err);
-                        notice_tx
-                            .try_send(
-                                "relay experienced an error trying to publish the latest event"
-                                    .to_owned(),
-                            )
-                            .ok();
+                        let msg =
+                            "relay experienced an error trying to publish the latest event".into();
+                        notice_tx.try_send(Notice::err_msg(msg, event.id)).ok();
                     }
                 }
             }
@@ -577,7 +581,7 @@ fn query_from_sub(sub: &Subscription) -> (String, Vec<Box<dyn ToSql>>) {
         .map(|s| format!("SELECT content, created_at FROM ({})", s))
         .collect();
     let query: String = subqueries_selects.join(" UNION ");
-    debug!("final query string: {}", query);
+    trace!("final query string: {}", query);
     (query, params)
 }
 
@@ -595,12 +599,12 @@ pub async fn db_query(
     mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
 ) {
     task::spawn_blocking(move || {
-        debug!("going to query for: {:?}", sub);
+        trace!("going to query for: {:?}", sub);
         let mut row_count: usize = 0;
         let start = Instant::now();
         // generate SQL query
         let (q, p) = query_from_sub(&sub);
-        debug!("SQL generated in {:?}", start.elapsed());
+        trace!("SQL generated in {:?}", start.elapsed());
         // show pool stats
         debug!("DB pool stats: {:?}", pool.state());
         let start = Instant::now();

src/lib.rs

@@ -8,6 +8,7 @@ pub mod event;
 pub mod hexrange;
 pub mod info;
 pub mod nip05;
+pub mod notice;
 pub mod schema;
 pub mod subscription;
 pub mod utils;

src/notice.rs

@@ -0,0 +1,48 @@
+use crate::error;
+
+pub enum EventResultStatus {
+    Saved,
+    Duplicate,
+    Error(String),
+}
+
+pub struct EventResult {
+    pub id: String,
+    pub status: EventResultStatus,
+}
+
+pub enum Notice {
+    Message(String),
+    EventResult(EventResult),
+}
+
+impl Notice {
+    pub fn err(err: error::Error, id: String) -> Notice {
+        Notice::err_msg(format!("{}", err), id)
+    }
+
+    pub fn message(msg: String) -> Notice {
+        Notice::Message(msg)
+    }
+
+    pub fn saved(id: String) -> Notice {
+        Notice::EventResult(EventResult {
+            id,
+            status: EventResultStatus::Saved,
+        })
+    }
+
+    pub fn duplicate(id: String) -> Notice {
+        Notice::EventResult(EventResult {
+            id,
+            status: EventResultStatus::Duplicate,
+        })
+    }
+
+    pub fn err_msg(msg: String, id: String) -> Notice {
+        Notice::EventResult(EventResult {
+            id,
+            status: EventResultStatus::Error(msg),
+        })
+    }
+}

src/server.rs

@@ -10,6 +10,7 @@ use crate::event::Event;
 use crate::event::EventCmd;
 use crate::info::RelayInfo;
 use crate::nip05;
+use crate::notice::{EventResultStatus, Notice};
 use crate::subscription::Subscription;
 use futures::SinkExt;
 use futures::StreamExt;
@@ -405,8 +406,17 @@ fn convert_to_msg(msg: String, max_bytes: Option<usize>) -> Result<NostrMessage>
 }
 
 /// Turn a string into a NOTICE message ready to send over a WebSocket
-fn make_notice_message(msg: &str) -> Message {
-    Message::text(json!(["NOTICE", msg]).to_string())
+fn make_notice_message(notice: Notice) -> Message {
+    let json = match notice {
+        Notice::Message(ref msg) => json!(["NOTICE", msg]),
+        Notice::EventResult(ref res) => match &res.status {
+            EventResultStatus::Saved => json!(["OK", res.id, "true"]),
+            EventResultStatus::Duplicate => json!(["OK", res.id, "true", "duplicate"]),
+            EventResultStatus::Error(msg) => json!(["OK", res.id, "false", msg]),
+        },
+    };
+
+    Message::text(json.to_string())
 }
 
 struct ClientInfo {
@@ -435,7 +445,7 @@ async fn nostr_server(
     // we will send out the tx handle to any query we generate.
     let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(256);
     // Create channel for receiving NOTICEs
-    let (notice_tx, mut notice_rx) = mpsc::channel::<String>(32);
+    let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(32);
 
     // last time this client sent data (message, ping, etc.)
     let mut last_message_time = Instant::now();
@@ -458,14 +468,14 @@ async fn nostr_server(
     // and how many it received from queries.
     let mut client_published_event_count: usize = 0;
     let mut client_received_event_count: usize = 0;
-    debug!("new connection for client: {:?}, ip: {:?}", cid, conn.ip());
+    debug!("new connection for client: {}, ip: {:?}", cid, conn.ip());
     if let Some(ua) = client_info.user_agent {
-        debug!("client: {:?} has user-agent: {:?}", cid, ua);
+        debug!("client: {} has user-agent: {:?}", cid, ua);
     }
     loop {
         tokio::select! {
             _ = shutdown.recv() => {
-        info!("Shutting client connection down due to shutdown: {:?}, ip: {:?}", cid, conn.ip());
+        info!("Close connection down due to shutdown, client: {}, ip: {:?}", cid, conn.ip());
                 // server shutting down, exit loop
                 break;
             },
@@ -480,7 +490,7 @@ async fn nostr_server(
                 ws_stream.send(Message::Ping(Vec::new())).await.ok();
             },
             Some(notice_msg) = notice_rx.recv() => {
-                ws_stream.send(make_notice_message(&notice_msg)).await.ok();
+                ws_stream.send(make_notice_message(notice_msg)).await.ok();
             },
             Some(query_result) = query_rx.recv() => {
                 // database informed us of a query result we asked for
@@ -499,12 +509,15 @@ async fn nostr_server(
             Ok(global_event) = bcast_rx.recv() => {
                 // an event has been broadcast to all clients
                 // first check if there is a subscription for this event.
-                let matching_subs = conn.get_matching_subscriptions(&global_event);
-                for s in matching_subs {
+                for (s, sub) in conn.subscriptions() {
+                    if !sub.interested_in_event(&global_event) {
+                        continue;
+                    }
+
                     // TODO: serialize at broadcast time, instead of
                     // once for each consumer.
                     if let Ok(event_str) = serde_json::to_string(&global_event) {
-                        debug!("sub match for client: {:?}, sub: {:?}, event: {:?}",
+                        debug!("sub match for client: {}, sub: {:?}, event: {:?}",
                                cid, s,
                                global_event.get_event_id_prefix());
                         // create an event response and send it
@@ -525,7 +538,7 @@ async fn nostr_server(
                     },
             Some(Ok(Message::Binary(_))) => {
             ws_stream.send(
-                make_notice_message("binary messages are not accepted")).await.ok();
+                make_notice_message(Notice::message("binary messages are not accepted".into()))).await.ok();
                         continue;
                     },
                     Some(Ok(Message::Ping(_) | Message::Pong(_))) => {
@@ -535,8 +548,7 @@ async fn nostr_server(
                     },
             Some(Err(WsError::Capacity(MessageTooLong{size, max_size}))) => {
             ws_stream.send(
-                make_notice_message(
-                &format!("message too large ({} > {})",size, max_size))).await.ok();
+                make_notice_message(Notice::message(format!("message too large ({} > {})",size, max_size)))).await.ok();
                         continue;
             },
                     None |
@@ -544,17 +556,17 @@ async fn nostr_server(
              Err(WsError::AlreadyClosed | WsError::ConnectionClosed |
                  WsError::Protocol(tungstenite::error::ProtocolError::ResetWithoutClosingHandshake)))
                         => {
-                            debug!("websocket close from client: {:?}, ip: {:?}",cid, conn.ip());
+                            debug!("websocket close from client: {}, ip: {:?}",cid, conn.ip());
                         break;
                     },
                     Some(Err(WsError::Io(e))) => {
                         // IO errors are considered fatal
-                        warn!("IO error (client: {:?}, ip: {:?}): {:?}", cid, conn.ip(), e);
+                        warn!("IO error (client: {}, ip: {:?}): {:?}", cid, conn.ip(), e);
                         break;
                     }
                     x => {
                         // default condition on error is to close the client connection
-                        info!("unknown error (client: {:?}, ip: {:?}): {:?} (closing conn)", cid, conn.ip(), x);
+                        info!("unknown error (client: {}, ip: {:?}): {:?} (closing conn)", cid, conn.ip(), x);
                         break;
                     }
                 };
@@ -568,7 +580,7 @@ async fn nostr_server(
                         match parsed {
                             Ok(e) => {
                                 let id_prefix:String = e.id.chars().take(8).collect();
-                                debug!("successfully parsed/validated event: {:?} from client: {:?}", id_prefix, cid);
+                                debug!("successfully parsed/validated event: {:?} from client: {}", id_prefix, cid);
                                 // check if the event is too far in the future.
                                 if e.is_valid_timestamp(settings.options.reject_future_seconds) {
                                     // Write this to the database.
@@ -576,20 +588,22 @@ async fn nostr_server(
                                     event_tx.send(submit_event).await.ok();
                                     client_published_event_count += 1;
                 } else {
-                    info!("client {:?} sent a far future-dated event", cid);
+                    info!("client: {} sent a far future-dated event", cid);
                     if let Some(fut_sec) = settings.options.reject_future_seconds {
-                    ws_stream.send(make_notice_message(&format!("The event created_at field is out of the acceptable range (+{}sec) for this relay and was not stored.",fut_sec))).await.ok();
+                        let msg = format!("The event created_at field is out of the acceptable range (+{}sec) for this relay and was not stored.",fut_sec);
+                        let notice = Notice::err_msg(msg, e.id);
+                        ws_stream.send(make_notice_message(notice)).await.ok();
                     }
                 }
                             },
                             Err(_) => {
-                                info!("client: {:?} sent an invalid event", cid);
-                                ws_stream.send(make_notice_message("event was invalid")).await.ok();
+                                info!("client: {} sent an invalid event", cid);
+                                ws_stream.send(make_notice_message(Notice::message("event was invalid".into()))).await.ok();
                             }
                         }
                     },
                     Ok(NostrMessage::SubMsg(s)) => {
-                        debug!("client {} requesting a subscription", cid);
+                        debug!("client: {} requesting a subscription", cid);
                         // subscription handling consists of:
                         // * registering the subscription so future events can be matched
                         // * making a channel to cancel to request later
@@ -606,7 +620,7 @@ async fn nostr_server(
                             },
                             Err(e) => {
                                 info!("Subscription error: {}", e);
-                                ws_stream.send(make_notice_message(&e.to_string())).await.ok();
+                                ws_stream.send(make_notice_message(Notice::err(e, s.id))).await.ok();
                             }
                         }
                     },
@@ -625,23 +639,23 @@ async fn nostr_server(
                                 conn.unsubscribe(&c);
                             } else {
                                 info!("invalid command ignored");
-                                ws_stream.send(make_notice_message("could not parse command")).await.ok();
+                                ws_stream.send(make_notice_message(Notice::message("could not parse command".into()))).await.ok();
                             }
                     },
                     Err(Error::ConnError) => {
-                        debug!("got connection close/error, disconnecting client: {:?}, ip: {:?}",cid, conn.ip());
+                        debug!("got connection close/error, disconnecting client: {}, ip: {:?}",cid, conn.ip());
                         break;
                     }
                     Err(Error::EventMaxLengthError(s)) => {
-                        info!("client {:?} sent event larger ({} bytes) than max size", cid, s);
-                        ws_stream.send(make_notice_message("event exceeded max size")).await.ok();
+                        info!("client: {} sent event larger ({} bytes) than max size", cid, s);
+                        ws_stream.send(make_notice_message(Notice::message("event exceeded max size".into()))).await.ok();
                     },
                     Err(Error::ProtoParseError) => {
-                        info!("client {:?} sent event that could not be parsed", cid);
-                        ws_stream.send(make_notice_message("could not parse command")).await.ok();
+                        info!("client {} sent event that could not be parsed", cid);
+                        ws_stream.send(make_notice_message(Notice::message("could not parse command".into()))).await.ok();
                     },
                     Err(e) => {
-                        info!("got non-fatal error from client: {:?}, error: {:?}", cid, e);
+                        info!("got non-fatal error from client: {}, error: {:?}", cid, e);
                     },
                 }
             },
@@ -652,7 +666,7 @@ async fn nostr_server(
         stop_tx.send(()).ok();
     }
     info!(
-        "stopping connection for client: {:?}, ip: {:?} (client sent {} event(s), received {})",
+        "stopping connection for client: {}, ip: {:?} (client sent {} event(s), received {})",
         cid,
         conn.ip(),
         client_published_event_count,