wip: tests, clippy

Cargo.toml

@@ -1,15 +1,7 @@
 [package]
 name = "nostr-rs-relay"
-version = "0.7.13"
+version = "0.6.2"
 edition = "2021"
-authors = ["Greg Heartsfield <scsibug@imap.cc>"]
-description = "A relay implementation for the Nostr protocol"
-readme = "README.md"
-homepage = "https://sr.ht/~gheartsfield/nostr-rs-relay/"
-repository = "https://git.sr.ht/~gheartsfield/nostr-rs-relay"
-license = "MIT"
-keywords = ["nostr", "server"]
-categories = ["network-programming", "web-programming"]
 
 [dependencies]
 tracing = "0.1.36"

Dockerfile

@@ -1,24 +1,18 @@
-FROM docker.io/library/rust:1.66.0@sha256:359949280cebefe93ccb33089fe25111a3aadfe99eac4b6cbe8ec3e1b571dacb as builder
+FROM docker.io/library/rust:1.64.0@sha256:5cf09a76cb9baf4990d121221bbad64927cc5690ee54f246487e302ddc2ba300 as builder
 
-RUN USER=root cargo install cargo-auditable
 RUN USER=root cargo new --bin nostr-rs-relay
 WORKDIR ./nostr-rs-relay
 COPY ./Cargo.toml ./Cargo.toml
 COPY ./Cargo.lock ./Cargo.lock
-# build dependencies only (caching)
+RUN cargo build --release
-RUN cargo auditable build --release --locked
-# get rid of starter project code
 RUN rm src/*.rs
 
-# copy project source code
 COPY ./src ./src
 
-# build auditable release using locked deps
 RUN rm ./target/release/deps/nostr*relay*
-RUN cargo auditable build --release --locked
+RUN cargo build --release
-
-FROM docker.io/library/debian:bullseye-20221205-slim@sha256:25f10b4f1ded5341a3ca0a30290ff3cd5639415f0c5a2222d5e7d5dd72952aa1
 
+FROM docker.io/library/debian:bullseye-20221004-slim@sha256:8b702518a671c926b5ece4efe386a476eb4777646a36d996d4bd50944f2f11a2
 ARG APP=/usr/src/app
 ARG APP_DATA=/usr/src/app/db
 RUN apt-get update \

README.md

@@ -1,8 +1,8 @@
 # [nostr-rs-relay](https://git.sr.ht/~gheartsfield/nostr-rs-relay)
 
-This is a [nostr](https://github.com/nostr-protocol/nostr) relay,
+This is a [nostr](https://github.com/nostr-protocol/nostr) relay, written in
-written in Rust.  It currently supports the entire relay protocol, and
+Rust.  It currently supports the entire relay protocol, and has a
-persists data with SQLite.
+SQLite persistence layer.
 
 The project master repository is available on
 [sourcehut](https://sr.ht/~gheartsfield/nostr-rs-relay/), and is
@@ -26,9 +26,7 @@ mirrored on [GitHub](https://github.com/scsibug/nostr-rs-relay).
 - [x] NIP-12: [Generic Tag Queries](https://github.com/nostr-protocol/nips/blob/master/12.md)
 - [x] NIP-15: [End of Stored Events Notice](https://github.com/nostr-protocol/nips/blob/master/15.md)
 - [x] NIP-16: [Event Treatment](https://github.com/nostr-protocol/nips/blob/master/16.md)
-- [x] NIP-20: [Command Results](https://github.com/nostr-protocol/nips/blob/master/20.md)
 - [x] NIP-22: [Event `created_at` limits](https://github.com/nostr-protocol/nips/blob/master/22.md) (_future-dated events only_)
-- [x] NIP-26: [Event Delegation](https://github.com/nostr-protocol/nips/blob/master/26.md)
 
 ## Quick Start
 
@@ -37,32 +35,15 @@ application.  Use a bind mount to store the SQLite database outside of
 the container image, and map the container's 8080 port to a host port
 (7000 in the example below).
 
-The examples below start a rootless podman container, mapping a local
-data directory and config file.
-
 ```console
-$ podman build -t nostr-rs-relay .
+$ docker build -t nostr-rs-relay .
 
-$ mkdir data
+$ docker run -it -p 7000:8080 \
+  --mount src=$(pwd)/data,target=/usr/src/app/db,type=bind nostr-rs-relay
 
-$ podman unshare chown 100:100 data
+[2021-12-31T19:58:31Z INFO  nostr_rs_relay] listening on: 0.0.0.0:8080
-
+[2021-12-31T19:58:31Z INFO  nostr_rs_relay::db] opened database "/usr/src/app/db/nostr.db" for writing
-$ podman run -it --rm -p 7000:8080 \
+[2021-12-31T19:58:31Z INFO  nostr_rs_relay::db] DB version = 2
-  --user=100:100 \
-  -v $(pwd)/data:/usr/src/app/db:Z \
-  -v $(pwd)/config.toml:/usr/src/app/config.toml:ro,Z \
-  --name nostr-relay nostr-rs-relay:latest
-
-Nov 19 15:31:15.013  INFO nostr_rs_relay: Starting up from main
-Nov 19 15:31:15.017  INFO nostr_rs_relay::server: listening on: 0.0.0.0:8080
-Nov 19 15:31:15.019  INFO nostr_rs_relay::server: db writer created
-Nov 19 15:31:15.019  INFO nostr_rs_relay::server: control message listener started
-Nov 19 15:31:15.019  INFO nostr_rs_relay::db: Built a connection pool "event writer" (min=1, max=4)
-Nov 19 15:31:15.019  INFO nostr_rs_relay::db: opened database "/usr/src/app/db/nostr.db" for writing
-Nov 19 15:31:15.019  INFO nostr_rs_relay::schema: DB version = 0
-Nov 19 15:31:15.054  INFO nostr_rs_relay::schema: database pragma/schema initialized to v7, and ready
-Nov 19 15:31:15.054  INFO nostr_rs_relay::schema: All migration scripts completed successfully.  Welcome to v7.
-Nov 19 15:31:15.521  INFO nostr_rs_relay::db: Built a connection pool "client query" (min=4, max=128)
 ```
 
 Use a `nostr` client such as

config.toml

@@ -46,14 +46,6 @@ address = "0.0.0.0"
 # Listen on this port
 port = 8080
 
-# If present, read this HTTP header for logging client IP addresses.
-# Examples for common proxies, cloudflare:
-#remote_ip_header = "x-forwarded-for"
-#remote_ip_header = "cf-connecting-ip"
-
-# Websocket ping interval in seconds, defaults to 5 minutes
-#ping_interval = 300
-
 [options]
 # Reject events that have timestamps greater than this many seconds in
 # the future.  Recommended to reject anything greater than 30 minutes
@@ -62,24 +54,8 @@ reject_future_seconds = 1800
 
 [limits]
 # Limit events created per second, averaged over one minute.  Must be
-# an integer.  If not set (or set to 0), defaults to unlimited.  Note:
+# an integer.  If not set (or set to 0), defaults to unlimited.
-# this is for the server as a whole, not per-connection.
+#messages_per_sec = 0
-# messages_per_sec = 0
-
-# Limit client subscriptions created per second, averaged over one
-# minute.  Must be an integer.  If not set (or set to 0), defaults to
-# unlimited.
-#subscriptions_per_min = 0
-
-# UNIMPLEMENTED...
-# Limit how many concurrent database connections a client can have.
-# This prevents a single client from starting too many expensive
-# database queries.  Must be an integer.  If not set (or set to 0),
-# defaults to unlimited (subject to subscription limits).
-#db_conns_per_client = 0
-
-# Limit blocking threads used for database connections.  Defaults to 16.
-#max_blocking_threads = 16
 
 # Limit the maximum size of an EVENT message.  Defaults to 128 KB.
 # Set to 0 for unlimited.

rustfmt.toml

@@ -1,4 +1 @@
 edition = "2021"
-#max_width = 140
-#chain_width = 100
-#fn_call_width = 100

src/config.rs

@@ -28,10 +28,9 @@ pub struct Database {
 pub struct Network {
     pub port: u16,
     pub address: String,
-    pub remote_ip_header: Option<String>, // retrieve client IP from this HTTP header if present
-    pub ping_interval_seconds: u32,
 }
 
+//
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[allow(unused)]
 pub struct Options {
@@ -52,9 +51,6 @@ pub struct Retention {
 #[allow(unused)]
 pub struct Limits {
     pub messages_per_sec: Option<u32>, // Artificially slow down event writing to limit disk consumption (averaged over 1 minute)
-    pub subscriptions_per_min: Option<u32>, // Artificially slow down request (db query) creation to prevent abuse (averaged over 1 minute)
-    pub db_conns_per_client: Option<u32>, // How many concurrent database queries (not subscriptions) may a client have?
-    pub max_blocking_threads: usize,
     pub max_event_bytes: Option<usize>, // Maximum size of an EVENT message
     pub max_ws_message_bytes: Option<usize>,
     pub max_ws_frame_bytes: Option<usize>,
@@ -211,15 +207,10 @@ impl Default for Settings {
             },
             network: Network {
                 port: 8080,
-                ping_interval_seconds: 300,
                 address: "0.0.0.0".to_owned(),
-                remote_ip_header: None,
             },
             limits: Limits {
                 messages_per_sec: None,
-                subscriptions_per_min: None,
-                db_conns_per_client: None,
-                max_blocking_threads: 16,
                 max_event_bytes: Some(2 << 17),      // 128K
                 max_ws_message_bytes: Some(2 << 17), // 128K
                 max_ws_frame_bytes: Some(2 << 17),   // 128K

src/conn.rs

@@ -2,10 +2,11 @@
 use crate::close::Close;
 use crate::error::Error;
 use crate::error::Result;
+use crate::event::Event;
 
 use crate::subscription::Subscription;
 use std::collections::HashMap;
-use tracing::{debug, trace};
+use tracing::{debug, info};
 use uuid::Uuid;
 
 /// A subscription identifier has a maximum length
@@ -13,8 +14,6 @@ const MAX_SUBSCRIPTION_ID_LEN: usize = 256;
 
 /// State for a client connection
 pub struct ClientConn {
-    /// Client IP (either from socket, or configured proxy header
-    client_ip: String,
     /// Unique client identifier generated at connection time
     client_id: Uuid,
     /// The current set of active client subscriptions
@@ -25,32 +24,22 @@ pub struct ClientConn {
 
 impl Default for ClientConn {
     fn default() -> Self {
-        Self::new("unknown".to_owned())
+        Self::new()
     }
 }
 
 impl ClientConn {
     /// Create a new, empty connection state.
     #[must_use]
-    pub fn new(client_ip: String) -> Self {
+    pub fn new() -> Self {
         let client_id = Uuid::new_v4();
         ClientConn {
-            client_ip,
             client_id,
             subscriptions: HashMap::new(),
             max_subs: 32,
         }
     }
 
-    pub fn subscriptions(&self) -> &HashMap<String, Subscription> {
-        &self.subscriptions
-    }
-
-    /// Check if the given subscription already exists
-    pub fn has_subscription(&self, sub: &Subscription) -> bool {
-        self.subscriptions.values().any(|x| x == sub)
-    }
-
     /// Get a short prefix of the client's unique identifier, suitable
     /// for logging.
     #[must_use]
@@ -58,9 +47,16 @@ impl ClientConn {
         self.client_id.to_string().chars().take(8).collect()
     }
 
+    /// Find all matching subscriptions.
     #[must_use]
-    pub fn ip(&self) -> &str {
+    pub fn get_matching_subscriptions(&self, e: &Event) -> Vec<&str> {
-        &self.client_ip
+        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.
@@ -74,7 +70,7 @@ impl ClientConn {
         // prevent arbitrarily long subscription identifiers from
         // being used.
         if sub_id_len > MAX_SUBSCRIPTION_ID_LEN {
-            debug!(
+            info!(
                 "ignoring sub request with excessive length: ({})",
                 sub_id_len
             );
@@ -83,12 +79,8 @@ impl ClientConn {
         // check if an existing subscription exists, and replace if so
         if self.subscriptions.contains_key(&k) {
             self.subscriptions.remove(&k);
-            self.subscriptions.insert(k, s.clone());
+            self.subscriptions.insert(k, s);
-            trace!(
+            debug!("replaced existing subscription");
-                "replaced existing subscription (cid: {}, sub: {:?})",
-                self.get_client_prefix(),
-                s.get_id()
-            );
             return Ok(());
         }
 
@@ -98,10 +90,9 @@ impl ClientConn {
         }
         // add subscription
         self.subscriptions.insert(k, s);
-        trace!(
+        debug!(
-            "registered new subscription, currently have {} active subs (cid: {})",
+            "registered new subscription, currently have {} active subs",
-            self.subscriptions.len(),
+            self.subscriptions.len()
-            self.get_client_prefix(),
         );
         Ok(())
     }
@@ -110,10 +101,10 @@ impl ClientConn {
     pub fn unsubscribe(&mut self, c: &Close) {
         // TODO: return notice if subscription did not exist.
         self.subscriptions.remove(&c.id);
-        trace!(
+        debug!(
-            "removed subscription, currently have {} active subs (cid: {})",
+            "removed subscription, currently have {} active subs (cid={})",
             self.subscriptions.len(),
-            self.get_client_prefix(),
+            self.client_id
         );
     }
 }

src/db.rs

@@ -6,7 +6,6 @@ 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;
@@ -33,20 +32,12 @@ 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<Notice>,
+    pub notice_tx: tokio::sync::mpsc::Sender<String>,
 }
 
 /// Database file
 pub const DB_FILE: &str = "nostr.db";
 
-/// How frequently to run maintenance
-/// How many persisted events before DB maintenannce is triggered.
-pub const EVENT_MAINTENANCE_FREQ_SEC: u64 = 60;
-
-/// How many persisted events before we pause for backups.
-/// It isn't clear this is enough to make the online backup API work yet.
-pub const EVENT_COUNT_BACKUP_PAUSE_TRIGGER: usize = 1000;
-
 /// Build a database connection pool.
 /// # Panics
 ///
@@ -84,7 +75,6 @@ pub fn build_pool(
         .test_on_check_out(true) // no noticeable performance hit
         .min_idle(Some(min_size))
         .max_size(max_size)
-        .max_lifetime(Some(Duration::from_secs(30)))
         .build(manager)
         .unwrap();
     info!(
@@ -94,46 +84,6 @@ pub fn build_pool(
     pool
 }
 
-/// Perform normal maintenance
-pub fn optimize_db(conn: &mut PooledConnection) -> Result<()> {
-    let start = Instant::now();
-    conn.execute_batch("PRAGMA optimize;")?;
-    info!("optimize ran in {:?}", start.elapsed());
-    Ok(())
-}
-#[derive(Debug)]
-enum SqliteReturnStatus {
-    SqliteOk,
-    SqliteBusy,
-    SqliteError,
-    SqliteOther(u64),
-}
-
-/// Checkpoint/Truncate WAL
-pub fn checkpoint_db(conn: &mut PooledConnection) -> Result<()> {
-    let query = "PRAGMA wal_checkpoint(TRUNCATE);";
-    let start = Instant::now();
-    let (cp_result, wal_size, _frames_checkpointed) = conn.query_row(query, [], |row| {
-        let checkpoint_result: u64 = row.get(0)?;
-        let wal_size: u64 = row.get(1)?;
-        let frames_checkpointed: u64 = row.get(2)?;
-        Ok((checkpoint_result, wal_size, frames_checkpointed))
-    })?;
-    let result = match cp_result {
-        0 => SqliteReturnStatus::SqliteOk,
-        1 => SqliteReturnStatus::SqliteBusy,
-        2 => SqliteReturnStatus::SqliteError,
-        x => SqliteReturnStatus::SqliteOther(x),
-    };
-    info!(
-        "checkpoint ran in {:?} (result: {:?}, WAL size: {})",
-        start.elapsed(),
-        result,
-        wal_size
-    );
-    Ok(())
-}
-
 /// Spawn a database writer that persists events to the SQLite store.
 pub async fn db_writer(
     settings: Settings,
@@ -156,7 +106,7 @@ pub async fn db_writer(
             &settings,
             OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE,
             1,
-            2,
+            4,
             false,
         );
         if settings.database.in_memory {
@@ -173,10 +123,6 @@ pub async fn db_writer(
         let rps_setting = settings.limits.messages_per_sec;
         let mut most_recent_rate_limit = Instant::now();
         let mut lim_opt = None;
-        // Constant writing has interfered with online backups.  Keep
-        // track of how long since we've given the backups a chance to
-        // run.
-        let mut backup_pause_counter: usize = 0;
         let clock = governor::clock::QuantaClock::default();
         if let Some(rps) = rps_setting {
             if rps > 0 {
@@ -196,15 +142,12 @@ pub async fn db_writer(
             if next_event.is_none() {
                 break;
             }
-            // track if an event write occurred; this is used to
-            // update the rate limiter
             let mut event_write = false;
             let subm_event = next_event.unwrap();
             let event = subm_event.event;
             let notice_tx = subm_event.notice_tx;
             // check if this event is authorized.
             if let Some(allowed_addrs) = whitelist {
-                // TODO: incorporate delegated pubkeys
                 // if the event address is not in allowed_addrs.
                 if !allowed_addrs.contains(&event.pubkey) {
                     info!(
@@ -212,10 +155,7 @@ pub async fn db_writer(
                         event.get_event_id_prefix()
                     );
                     notice_tx
-                        .try_send(Notice::blocked(
+                        .try_send("pubkey is not allowed to publish to this relay".to_owned())
-                            event.id,
-                            "pubkey is not allowed to publish to this relay",
-                        ))
                         .ok();
                     continue;
                 }
@@ -241,16 +181,15 @@ pub async fn db_writer(
                                 event.get_author_prefix()
                             );
                         } else {
-                            info!(
+                            info!("rejecting event, author ({:?} / {:?}) verification invalid (expired/wrong domain)",
-                                "rejecting event, author ({:?} / {:?}) verification invalid (expired/wrong domain)",
+                                  uv.name.to_string(),
-                                uv.name.to_string(),
+                                  event.get_author_prefix()
-                                event.get_author_prefix()
                             );
                             notice_tx
-                                .try_send(Notice::blocked(
+                                .try_send(
-                                    event.id,
+                                    "NIP-05 verification is no longer valid (expired/wrong domain)"
-                                    "NIP-05 verification is no longer valid (expired/wrong domain)",
+                                        .to_owned(),
-                                ))
+                                )
                                 .ok();
                             continue;
                         }
@@ -261,10 +200,7 @@ pub async fn db_writer(
                             event.get_author_prefix()
                         );
                         notice_tx
-                            .try_send(Notice::blocked(
+                            .try_send("NIP-05 verification needed to publish events".to_owned())
-                                event.id,
-                                "NIP-05 verification needed to publish events",
-                            ))
                             .ok();
                         continue;
                     }
@@ -277,24 +213,22 @@ pub async fn db_writer(
             // TODO: cache recent list of authors to remove a DB call.
             let start = Instant::now();
             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()
                 );
+                bcast_tx.send(event.clone()).ok();
                 event_write = true
             } else {
-                log_pool_stats("writer", &pool);
                 match write_event(&mut pool.get()?, &event) {
                     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()
@@ -302,21 +236,18 @@ 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);
-                        let msg = "relay experienced an error trying to publish the latest event";
+                        notice_tx
-                        notice_tx.try_send(Notice::error(event.id, msg)).ok();
+                            .try_send(
+                                "relay experienced an error trying to publish the latest event"
+                                    .to_owned(),
+                            )
+                            .ok();
                     }
                 }
-                backup_pause_counter += 1;
-                if backup_pause_counter > EVENT_COUNT_BACKUP_PAUSE_TRIGGER {
-                    info!("pausing db write thread for a moment...");
-                    thread::sleep(Duration::from_millis(500));
-                    backup_pause_counter = 0
-                }
             }
 
             // use rate limit, if defined, and if an event was actually written.
@@ -353,18 +284,16 @@ pub fn write_event(conn: &mut PooledConnection, e: &Event) -> Result<usize> {
     let tx = conn.transaction()?;
     // get relevant fields from event and convert to blobs.
     let id_blob = hex::decode(&e.id).ok();
-    let pubkey_blob: Option<Vec<u8>> = hex::decode(&e.pubkey).ok();
+    let pubkey_blob = hex::decode(&e.pubkey).ok();
-    let delegator_blob: Option<Vec<u8>> = e.delegated_by.as_ref().and_then(|d| hex::decode(d).ok());
     let event_str = serde_json::to_string(&e).ok();
     // ignore if the event hash is a duplicate.
     let mut ins_count = tx.execute(
-        "INSERT OR IGNORE INTO event (event_hash, created_at, kind, author, delegated_by, content, first_seen, hidden) VALUES (?1, ?2, ?3, ?4, ?5, ?6, strftime('%s','now'), FALSE);",
+        "INSERT OR IGNORE INTO event (event_hash, created_at, kind, author, content, first_seen, hidden) VALUES (?1, ?2, ?3, ?4, ?5, strftime('%s','now'), FALSE);",
-        params![id_blob, e.created_at, e.kind, pubkey_blob, delegator_blob, event_str]
+        params![id_blob, e.created_at, e.kind, pubkey_blob, event_str]
     )?;
     if ins_count == 0 {
         // if the event was a duplicate, no need to insert event or
-        // pubkey references.
+        // pubkey references.  This will abort the txn.
-        tx.rollback().ok();
         return Ok(ins_count);
     }
     // remember primary key of the event most recently inserted.
@@ -382,9 +311,9 @@ pub fn write_event(conn: &mut PooledConnection, e: &Event) -> Result<usize> {
                     // if tagvalue is lowercase hex;
                     if is_lower_hex(tagval) && (tagval.len() % 2 == 0) {
                         tx.execute(
-                            "INSERT OR IGNORE INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3)",
+			    "INSERT OR IGNORE INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3)",
-                            params![ev_id, &tagname, hex::decode(tagval).ok()],
+			    params![ev_id, &tagname, hex::decode(&tagval).ok()],
-                        )?;
+			)?;
                     } else {
                         tx.execute(
                             "INSERT OR IGNORE INTO tag (event_id, name, value) VALUES (?1, ?2, ?3)",
@@ -490,13 +419,14 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
 
     // if the filter is malformed, don't return anything.
     if f.force_no_match {
-        let empty_query = "SELECT e.content, e.created_at FROM event e WHERE 1=0".to_owned();
+        let empty_query =
+            "SELECT DISTINCT(e.content), e.created_at FROM event e WHERE 1=0".to_owned();
         // query parameters for SQLite
         let empty_params: Vec<Box<dyn ToSql>> = vec![];
         return (empty_query, empty_params);
     }
 
-    let mut query = "SELECT e.content, e.created_at FROM event e".to_owned();
+    let mut query = "SELECT DISTINCT(e.content), e.created_at FROM event e  ".to_owned();
     // query parameters for SQLite
     let mut params: Vec<Box<dyn ToSql>> = vec![];
 
@@ -509,22 +439,16 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
         for auth in authvec {
             match hex_range(auth) {
                 Some(HexSearch::Exact(ex)) => {
-                    auth_searches.push("author=? OR delegated_by=?".to_owned());
+                    auth_searches.push("author=?".to_owned());
-                    params.push(Box::new(ex.clone()));
                     params.push(Box::new(ex));
                 }
                 Some(HexSearch::Range(lower, upper)) => {
-                    auth_searches.push(
+                    auth_searches.push("(author>? AND author<?)".to_owned());
-                        "(author>? AND author<?) OR (delegated_by>? AND delegated_by<?)".to_owned(),
-                    );
-                    params.push(Box::new(lower.clone()));
-                    params.push(Box::new(upper.clone()));
                     params.push(Box::new(lower));
                     params.push(Box::new(upper));
                 }
                 Some(HexSearch::LowerOnly(lower)) => {
-                    auth_searches.push("author>? OR delegated_by>?".to_owned());
+                    auth_searches.push("author>?".to_owned());
-                    params.push(Box::new(lower.clone()));
                     params.push(Box::new(lower));
                 }
                 None => {
@@ -532,14 +456,8 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
                 }
             }
         }
-        if !authvec.is_empty() {
+        let authors_clause = format!("({})", auth_searches.join(" OR "));
-            let authors_clause = format!("({})", auth_searches.join(" OR "));
+        filter_components.push(authors_clause);
-            filter_components.push(authors_clause);
-        } else {
-            // if the authors list was empty, we should never return
-            // any results.
-            filter_components.push("false".to_owned());
-        }
     }
     // Query for Kind
     if let Some(ks) = &f.kinds {
@@ -572,14 +490,8 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
                 }
             }
         }
-        if !idvec.is_empty() {
+        let id_clause = format!("({})", id_searches.join(" OR "));
-            let id_clause = format!("({})", id_searches.join(" OR "));
+        filter_components.push(id_clause);
-            filter_components.push(id_clause);
-        } else {
-            // if the ids list was empty, we should never return
-            // any results.
-            filter_components.push("false".to_owned());
-        }
     }
     // Query for tags
     if let Some(map) = &f.tags {
@@ -588,7 +500,7 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
             let mut blob_vals: Vec<Box<dyn ToSql>> = vec![];
             for v in val {
                 if (v.len() % 2 == 0) && is_lower_hex(v) {
-                    if let Ok(h) = hex::decode(v) {
+                    if let Ok(h) = hex::decode(&v) {
                         blob_vals.push(Box::new(h));
                     }
                 } else {
@@ -599,10 +511,7 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
             let str_clause = format!("value IN ({})", repeat_vars(str_vals.len()));
             let blob_clause = format!("value_hex IN ({})", repeat_vars(blob_vals.len()));
             // find evidence of the target tag name/value existing for this event.
-            let tag_clause = format!(
+            let tag_clause = format!("e.id IN (SELECT e.id FROM event e LEFT JOIN tag t on e.id=t.event_id WHERE hidden!=TRUE and (name=? AND ({} OR {})))", str_clause, blob_clause);
-                "e.id IN (SELECT e.id FROM event e LEFT JOIN tag t on e.id=t.event_id WHERE hidden!=TRUE and (name=? AND ({} OR {})))",
-                str_clause, blob_clause
-            );
             // add the tag name as the first parameter
             params.push(Box::new(key.to_string()));
             // add all tag values that are plain strings as params
@@ -655,55 +564,13 @@ fn query_from_sub(sub: &Subscription) -> (String, Vec<Box<dyn ToSql>>) {
     // encapsulate subqueries into select statements
     let subqueries_selects: Vec<String> = subqueries
         .iter()
-        .map(|s| format!("SELECT distinct content, created_at FROM ({})", s))
+        .map(|s| format!("SELECT content, created_at FROM ({})", s))
         .collect();
     let query: String = subqueries_selects.join(" UNION ");
+    debug!("final query string: {}", query);
     (query, params)
 }
 
-/// Check if the pool is fully utilized
-fn _pool_at_capacity(pool: &SqlitePool) -> bool {
-    let state: r2d2::State = pool.state();
-    state.idle_connections == 0
-}
-
-/// Log pool stats
-fn log_pool_stats(name: &str, pool: &SqlitePool) {
-    let state: r2d2::State = pool.state();
-    let in_use_cxns = state.connections - state.idle_connections;
-    trace!(
-        "DB pool {:?} usage (in_use: {}, available: {})",
-        name,
-        in_use_cxns,
-        state.connections
-    );
-    if state.connections == in_use_cxns {
-        debug!("DB pool {:?} is empty (in_use: {})", name, in_use_cxns);
-    }
-}
-
-/// Perform database maintenance on a regular basis
-pub async fn db_maintenance(pool: SqlitePool) {
-    tokio::task::spawn(async move {
-        loop {
-            tokio::select! {
-                    _ = tokio::time::sleep(Duration::from_secs(EVENT_MAINTENANCE_FREQ_SEC)) => {
-                        if let Ok(mut conn) = pool.get() {
-                // the busy timer will block writers, so don't set
-                // this any higher than you want max latency for event
-                // writes.
-                conn.busy_timeout(Duration::from_secs(1)).ok();
-                            debug!("running database optimizer");
-                            optimize_db(&mut conn).ok();
-                            debug!("running wal_checkpoint(TRUNCATE)");
-                            checkpoint_db(&mut conn).ok();
-                        }
-            }
-                };
-        }
-    });
-}
-
 /// Perform a database query using a subscription.
 ///
 /// The [`Subscription`] is converted into a SQL query.  Each result
@@ -717,116 +584,45 @@ pub async fn db_query(
     query_tx: tokio::sync::mpsc::Sender<QueryResult>,
     mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
 ) {
-    let pre_spawn_start = Instant::now();
     task::spawn_blocking(move || {
-        let db_queue_time = pre_spawn_start.elapsed();
+        debug!("going to query for: {:?}", sub);
-        // if the queue time was very long (>5 seconds), spare the DB and abort.
-        if db_queue_time > Duration::from_secs(5) {
-            info!(
-                "shedding DB query load from {:?} (cid: {}, sub: {:?})",
-                db_queue_time, client_id, sub.id
-            );
-            return Ok(());
-        }
-        // otherwise, report queuing time if it is slow
-        else if db_queue_time > Duration::from_secs(1) {
-            debug!(
-                "(slow) DB query queued for {:?} (cid: {}, sub: {:?})",
-                db_queue_time, client_id, sub.id
-            );
-        }
-        let start = Instant::now();
         let mut row_count: usize = 0;
+        let start = Instant::now();
         // generate SQL query
         let (q, p) = query_from_sub(&sub);
-        let sql_gen_elapsed = start.elapsed();
+        debug!("SQL generated in {:?}", start.elapsed());
-        if sql_gen_elapsed > Duration::from_millis(10) {
-            debug!("SQL (slow) generated in {:?}", start.elapsed());
-        }
         // show pool stats
-        log_pool_stats("reader", &pool);
+        debug!("DB pool stats: {:?}", pool.state());
-        // cutoff for displaying slow queries
-        let slow_cutoff = Duration::from_millis(2000);
-        // any client that doesn't cause us to generate new rows in 5
-        // seconds gets dropped.
-        let abort_cutoff = Duration::from_secs(5);
         let start = Instant::now();
-        let mut slow_first_event;
-        let mut last_successful_send = Instant::now();
         if let Ok(conn) = pool.get() {
             // execute the query. Don't cache, since queries vary so much.
             let mut stmt = conn.prepare(&q)?;
             let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?;
             let mut first_result = true;
             while let Some(row) = event_rows.next()? {
-                let first_event_elapsed = start.elapsed();
-                slow_first_event = first_event_elapsed >= slow_cutoff;
                 if first_result {
                     debug!(
-                        "first result in {:?} (cid: {}, sub: {:?})",
+                        "time to first result: {:?} (cid={}, sub={:?})",
-                        first_event_elapsed, client_id, sub.id
+                        start.elapsed(),
+                        client_id,
+                        sub.id
                     );
                     first_result = false;
                 }
-                // logging for slow queries; show sub and SQL.
+                // check if this is still active
-                // to reduce logging; only show 1/16th of clients (leading 0)
+                // TODO:  check every N rows
-                if row_count == 0 && slow_first_event && client_id.starts_with("0") {
+                if abandon_query_rx.try_recv().is_ok() {
-                    debug!(
+                    debug!("query aborted (sub={:?})", sub.id);
-                        "query req (slow): {:?} (cid: {}, sub: {:?})",
-                        sub, client_id, sub.id
-                    );
-                    debug!(
-                        "query string (slow): {} (cid: {}, sub: {:?})",
-                        q, client_id, sub.id
-                    );
-                } else {
-                    trace!(
-                        "query req: {:?} (cid: {}, sub: {:?})",
-                        sub,
-                        client_id,
-                        sub.id
-                    );
-                    trace!(
-                        "query string: {} (cid: {}, sub: {:?})",
-                        q,
-                        client_id,
-                        sub.id
-                    );
-                }
-                // check if this is still active; every 100 rows
-                if row_count % 100 == 0 && abandon_query_rx.try_recv().is_ok() {
-                    debug!("query aborted (cid: {}, sub: {:?})", client_id, sub.id);
                     return Ok(());
                 }
                 row_count += 1;
                 let event_json = row.get(0)?;
-                loop {
-                    if query_tx.capacity() != 0 {
-                        // we have capacity to add another item
-                        break;
-                    } else {
-                        // the queue is full
-                        trace!("db reader thread is stalled");
-                        if last_successful_send + abort_cutoff < Instant::now() {
-                            // the queue has been full for too long, abort
-                            info!("aborting database query due to slow client");
-                            let ok: Result<()> = Ok(());
-                            return ok;
-                        }
-                        // give the queue a chance to clear before trying again
-                        thread::sleep(Duration::from_millis(100));
-                    }
-                }
-                // TODO: we could use try_send, but we'd have to juggle
-                // getting the query result back as part of the error
-                // result.
                 query_tx
                     .blocking_send(QueryResult {
                         sub_id: sub.get_id(),
                         event: event_json,
                     })
                     .ok();
-                last_successful_send = Instant::now();
             }
             query_tx
                 .blocking_send(QueryResult {
@@ -835,12 +631,11 @@ pub async fn db_query(
                 })
                 .ok();
             debug!(
-                "query completed in {:?} (cid: {}, sub: {:?}, db_time: {:?}, rows: {})",
+                "query completed ({} rows) in {:?} (cid={}, sub={:?})",
-                pre_spawn_start.elapsed(),
+                row_count,
-                client_id,
-                sub.id,
                 start.elapsed(),
-                row_count
+                client_id,
+                sub.id
             );
         } else {
             warn!("Could not get a database connection for querying");

src/delegation.rs

@@ -30,6 +30,10 @@ use tracing::{debug, info};
 // different condition strings, since we do not support grouping or
 // "OR" logic.
 
+// using serde_urldecode, we can get a serde data format from the
+// condition string.  We will then map that with a deserializer that
+// maps to a ConditionQuery.
+
 lazy_static! {
     /// Secp256k1 verification instance.
     pub static ref SECP: Secp256k1<VerifyOnly> = Secp256k1::verification_only();
@@ -84,19 +88,10 @@ pub struct ConditionQuery {
 }
 
 impl ConditionQuery {
-    pub fn allows_event(&self, event: &Event) -> bool {
+    pub fn allows_event(&self, _event: &Event) -> bool {
-        // check each condition, to ensure that the event complies
+        // check each condition, to ensure that the event complies with the restriction.
-        // with the restriction.
+
-        for c in &self.conditions {
+        false
-            if !c.allows_event(event) {
-                // any failing conditions invalidates the delegation
-                // on this event
-                return false;
-            }
-        }
-        // delegation was permitted unconditionally, or all conditions
-        // were true
-        true
     }
 }
 
@@ -336,7 +331,6 @@ mod tests {
         Event {
             id: "0".to_owned(),
             pubkey: "0".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: vec![],

src/error.rs

@@ -17,16 +17,10 @@ pub enum Error {
     ConnWriteError,
     #[error("EVENT parse failed")]
     EventParseFailed,
-    #[error("CLOSE message parse failed")]
+    #[error("ClOSE message parse failed")]
     CloseParseFailed,
-    #[error("Event invalid signature")]
+    #[error("Event validation failed")]
-    EventInvalidSignature,
+    EventInvalid,
-    #[error("Event invalid id")]
-    EventInvalidId,
-    #[error("Event malformed pubkey")]
-    EventMalformedPubkey,
-    #[error("Event could not canonicalize")]
-    EventCouldNotCanonicalize,
     #[error("Event too large")]
     EventMaxLengthError(usize),
     #[error("Subscription identifier max length exceeded")]

src/event.rs

@@ -27,19 +27,11 @@ pub struct EventCmd {
     event: Event,
 }
 
-impl EventCmd {
-    pub fn event_id(&self) -> &str {
-        &self.event.id
-    }
-}
-
 /// Parsed nostr event.
 #[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Clone)]
 pub struct Event {
     pub id: String,
     pub(crate) pubkey: String,
-    #[serde(skip)]
-    pub(crate) delegated_by: Option<String>,
     pub(crate) created_at: u64,
     pub(crate) kind: u64,
     #[serde(deserialize_with = "tag_from_string")]
@@ -89,13 +81,12 @@ impl From<EventCmd> for Result<Event> {
         // ensure command is correct
         if ec.cmd != "EVENT" {
             Err(CommandUnknownError)
+        } else if ec.event.is_valid() {
+            let mut e = ec.event;
+            e.build_index();
+            Ok(e)
         } else {
-            ec.event.validate().map(|_| {
+            Err(EventInvalid)
-                let mut e = ec.event;
-                e.build_index();
-                e.update_delegation();
-                e
-            })
         }
     }
 }
@@ -125,45 +116,27 @@ impl Event {
     // if so, return an alternate author for the event
     pub fn delegated_author(&self) -> Option<String> {
         // is there a delegation tag?
-        let delegation_tag: Vec<String> = self
+        let delegation_tag = self.tag_values_by_name("delegation");
-            .tags
-            .iter()
-            .filter(|x| x.len() == 4)
-            .filter(|x| x.get(0).unwrap() == "delegation")
-            .take(1)
-            .next()?
-            .to_vec(); // get first tag
-
-        //let delegation_tag = self.tag_values_by_name("delegation");
         // delegation tags should have exactly 3 elements after the name (pubkey, condition, sig)
         // the event is signed by the delagatee
         let delegatee = &self.pubkey;
         // the delegation tag references the claimed delagator
-        let delegator: &str = delegation_tag.get(1)?;
+        let delegator = delegation_tag.get(0)?;
-        let querystr: &str = delegation_tag.get(2)?;
+        let querystr = delegation_tag.get(1)?;
-        let sig: &str = delegation_tag.get(3)?;
+        let sig = delegation_tag.get(2)?;
-
+        // pass into the validate_delegation
-        // attempt to get a condition query; this requires the delegation to have a valid signature.
         if let Some(cond_query) = validate_delegation(delegator, delegatee, querystr, sig) {
-            // The signature was valid, now we ensure the delegation
+            // check if this condition query would allow this event.
-            // condition is valid for this event:
             if cond_query.allows_event(self) {
-                // since this is allowed, we will provide the delegatee
                 Some(delegator.into())
             } else {
-                debug!("an event failed to satisfy delegation conditions");
                 None
             }
         } else {
-            debug!("event had had invalid delegation signature");
             None
         }
     }
 
-    /// Update delegation status
-    fn update_delegation(&mut self) {
-        self.delegated_by = self.delegated_author();
-    }
     /// Build an event tag index
     fn build_index(&mut self) {
         // if there are no tags; just leave the index as None
@@ -199,7 +172,7 @@ impl Event {
         self.pubkey.chars().take(8).collect()
     }
 
-    /// Retrieve tag initial values across all tags matching the name
+    /// Retrieve tag values
     pub fn tag_values_by_name(&self, tag_name: &str) -> Vec<String> {
         self.tags
             .iter()
@@ -226,7 +199,7 @@ impl Event {
     }
 
     /// Check if this event has a valid signature.
-    fn validate(&self) -> Result<()> {
+    fn is_valid(&self) -> bool {
         // TODO: return a Result with a reason for invalid events
         // validation is performed by:
         // * parsing JSON string into event fields
@@ -235,8 +208,8 @@ impl Event {
         // * serialize with no spaces/newlines
         let c_opt = self.to_canonical();
         if c_opt.is_none() {
-            debug!("could not canonicalize");
+            debug!("event could not be canonicalized");
-            return Err(EventCouldNotCanonicalize);
+            return false;
         }
         let c = c_opt.unwrap();
         // * compute the sha256sum.
@@ -245,21 +218,21 @@ impl Event {
         // * ensure the id matches the computed sha256sum.
         if self.id != hex_digest {
             debug!("event id does not match digest");
-            return Err(EventInvalidId);
+            return false;
         }
         // * validate the message digest (sig) using the pubkey & computed sha256 message hash.
         let sig = schnorr::Signature::from_str(&self.sig).unwrap();
         if let Ok(msg) = secp256k1::Message::from_slice(digest.as_ref()) {
             if let Ok(pubkey) = XOnlyPublicKey::from_str(&self.pubkey) {
-                SECP.verify_schnorr(&sig, &msg, &pubkey)
+                let verify = SECP.verify_schnorr(&sig, &msg, &pubkey);
-                    .map_err(|_| EventInvalidSignature)
+                matches!(verify, Ok(()))
             } else {
                 debug!("client sent malformed pubkey");
-                Err(EventMalformedPubkey)
+                false
             }
         } else {
             info!("error converting digest to secp256k1 message");
-            Err(EventInvalidSignature)
+            false
         }
     }
 
@@ -323,7 +296,6 @@ mod tests {
         Event {
             id: "0".to_owned(),
             pubkey: "0".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: vec![],
@@ -405,7 +377,6 @@ mod tests {
         let e = Event {
             id: "999".to_owned(),
             pubkey: "012345".to_owned(),
-            delegated_by: None,
             created_at: 501234,
             kind: 1,
             tags: vec![],
@@ -423,7 +394,6 @@ mod tests {
         let e = Event {
             id: "999".to_owned(),
             pubkey: "012345".to_owned(),
-            delegated_by: None,
             created_at: 501234,
             kind: 1,
             tags: vec![
@@ -450,7 +420,6 @@ mod tests {
         let e = Event {
             id: "999".to_owned(),
             pubkey: "012345".to_owned(),
-            delegated_by: None,
             created_at: 501234,
             kind: 1,
             tags: vec![
@@ -477,7 +446,6 @@ mod tests {
         let e = Event {
             id: "999".to_owned(),
             pubkey: "012345".to_owned(),
-            delegated_by: None,
             created_at: 501234,
             kind: 1,
             tags: vec![

src/hexrange.rs

@@ -3,7 +3,7 @@ use crate::utils::is_hex;
 use hex;
 
 /// Types of hexadecimal queries.
-#[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Clone)]
+#[derive(PartialEq, Eq, Debug, Clone)]
 pub enum HexSearch {
     // when no range is needed, exact 32-byte
     Exact(Vec<u8>),

src/info.rs

@@ -35,7 +35,7 @@ impl From<config::Info> for RelayInfo {
             description: i.description,
             pubkey: i.pubkey,
             contact: i.contact,
-            supported_nips: Some(vec![1, 2, 9, 11, 12, 15, 16, 20, 22, 26]),
+            supported_nips: Some(vec![1, 2, 9, 11, 12, 15, 16, 22]),
             software: Some("https://git.sr.ht/~gheartsfield/nostr-rs-relay".to_owned()),
             version: CARGO_PKG_VERSION.map(|x| x.to_owned()),
         }

src/lib.rs

@@ -8,7 +8,6 @@ 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/main.rs

@@ -34,11 +34,11 @@ fn main() {
         // enable tracing with tokio-console
         ConsoleLayer::builder().with_default_env().init();
     }
+
     // update with database location
     if let Some(db) = db_dir {
         settings.database.data_directory = db;
     }
-
     let (_, ctrl_rx): (MpscSender<()>, MpscReceiver<()>) = syncmpsc::channel();
     // run this in a new thread
     let handle = thread::spawn(|| {

src/nip05.rs

@@ -517,7 +517,7 @@ impl Verifier {
                 Ok(updated) => {
                     if updated != 0 {
                         info!(
-                            "persisted event (new verified pubkey): {:?} in {:?}",
+                            "persisted event: {:?} in {:?}",
                             event.get_event_id_prefix(),
                             start.elapsed()
                         );
@@ -721,7 +721,7 @@ pub fn query_oldest_user_verification(
     earliest: u64,
 ) -> Result<VerificationRecord> {
     let tx = conn.transaction()?;
-    let query = "SELECT v.id, v.name, e.event_hash, e.author, e.created_at, v.verified_at, v.failed_at, v.failure_count FROM user_verification v INNER JOIN event e ON e.id=v.metadata_event WHERE (v.verified_at < ? OR v.verified_at IS NULL) AND (v.failed_at < ? OR v.failed_at IS NULL) ORDER BY v.verified_at ASC, v.failed_at ASC LIMIT 1;";
+    let query = "SELECT v.id, v.name, e.event_hash, e.author, e.created_at, v.verified_at, v.failed_at, v.failure_count FROM user_verification v LEFT JOIN event e ON e.id=v.metadata_event WHERE (v.verified_at < ? OR v.verified_at IS NULL) AND (v.failed_at < ? OR v.failed_at IS NULL) ORDER BY v.verified_at ASC, v.failed_at ASC LIMIT 1;";
     let mut stmt = tx.prepare_cached(query)?;
     let fields = stmt.query_row(params![earliest, earliest], |r| {
         let rowid: u64 = r.get(0)?;

src/notice.rs

@@ -1,86 +0,0 @@
-pub enum EventResultStatus {
-    Saved,
-    Duplicate,
-    Invalid,
-    Blocked,
-    RateLimited,
-    Error,
-}
-
-pub struct EventResult {
-    pub id: String,
-    pub msg: String,
-    pub status: EventResultStatus,
-}
-
-pub enum Notice {
-    Message(String),
-    EventResult(EventResult),
-}
-
-impl EventResultStatus {
-    pub fn to_bool(&self) -> bool {
-        match self {
-            Self::Saved => true,
-            Self::Duplicate => true,
-            Self::Invalid => false,
-            Self::Blocked => false,
-            Self::RateLimited => false,
-            Self::Error => false,
-        }
-    }
-
-    pub fn prefix(&self) -> &'static str {
-        match self {
-            Self::Saved => "saved",
-            Self::Duplicate => "duplicate",
-            Self::Invalid => "invalid",
-            Self::Blocked => "blocked",
-            Self::RateLimited => "rate-limited",
-            Self::Error => "error",
-        }
-    }
-}
-
-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)
-    }
-
-    fn prefixed(id: String, msg: &str, status: EventResultStatus) -> Notice {
-        let msg = format!("{}: {}", status.prefix(), msg);
-        Notice::EventResult(EventResult { id, msg, status })
-    }
-
-    pub fn invalid(id: String, msg: &str) -> Notice {
-        Notice::prefixed(id, msg, EventResultStatus::Invalid)
-    }
-
-    pub fn blocked(id: String, msg: &str) -> Notice {
-        Notice::prefixed(id, msg, EventResultStatus::Blocked)
-    }
-
-    pub fn rate_limited(id: String, msg: &str) -> Notice {
-        Notice::prefixed(id, msg, EventResultStatus::RateLimited)
-    }
-
-    pub fn duplicate(id: String) -> Notice {
-        Notice::prefixed(id, "", EventResultStatus::Duplicate)
-    }
-
-    pub fn error(id: String, msg: &str) -> Notice {
-        Notice::prefixed(id, msg, EventResultStatus::Error)
-    }
-
-    pub fn saved(id: String) -> Notice {
-        Notice::EventResult(EventResult {
-            id,
-            msg: "".into(),
-            status: EventResultStatus::Saved,
-        })
-    }
-}

src/schema.rs

@@ -16,11 +16,11 @@ pub const STARTUP_SQL: &str = r##"
 PRAGMA main.synchronous=NORMAL;
 PRAGMA foreign_keys = ON;
 PRAGMA journal_size_limit=32768;
-pragma mmap_size = 17179869184; -- cap mmap at 16GB
+pragma mmap_size = 536870912; -- 512MB of mmap
 "##;
 
 /// Latest database version
-pub const DB_VERSION: usize = 11;
+pub const DB_VERSION: usize = 7;
 
 /// Schema definition
 const INIT_SQL: &str = formatcp!(
@@ -40,7 +40,7 @@ event_hash BLOB NOT NULL, -- 4-byte hash
 first_seen INTEGER NOT NULL, -- when the event was first seen (not authored!) (seconds since 1970)
 created_at INTEGER NOT NULL, -- when the event was authored
 author BLOB NOT NULL, -- author pubkey
-delegated_by BLOB, -- delegator pubkey (NIP-26)
+delegator BLOB, -- delegator pubkey (NIP-26)
 kind INTEGER NOT NULL, -- event kind
 hidden INTEGER, -- relevant for queries
 content TEXT NOT NULL -- serialized json of event object
@@ -48,10 +48,9 @@ content TEXT NOT NULL -- serialized json of event object
 
 -- Event Indexes
 CREATE UNIQUE INDEX IF NOT EXISTS event_hash_index ON event(event_hash);
-CREATE INDEX IF NOT EXISTS author_index ON event(author);
 CREATE INDEX IF NOT EXISTS created_at_index ON event(created_at);
-CREATE INDEX IF NOT EXISTS delegated_by_index ON event(delegated_by);
+CREATE INDEX IF NOT EXISTS author_index ON event(author);
-CREATE INDEX IF NOT EXISTS event_composite_index ON event(kind,created_at);
+CREATE INDEX IF NOT EXISTS kind_index ON event(kind);
 
 -- Tag Table
 -- Tag values are stored as either a BLOB (if they come in as a
@@ -67,8 +66,6 @@ FOREIGN KEY(event_id) REFERENCES event(id) ON UPDATE CASCADE ON DELETE CASCADE
 );
 CREATE INDEX IF NOT EXISTS tag_val_index ON tag(value);
 CREATE INDEX IF NOT EXISTS tag_val_hex_index ON tag(value_hex);
-CREATE INDEX IF NOT EXISTS tag_composite_index ON tag(event_id,name,value_hex,value);
-CREATE INDEX IF NOT EXISTS tag_name_eid_index ON tag(name,event_id,value_hex);
 
 -- NIP-05 User Validation
 CREATE TABLE IF NOT EXISTS user_verification (
@@ -159,19 +156,6 @@ pub fn upgrade_db(conn: &mut PooledConnection) -> Result<()> {
             if curr_version == 6 {
                 curr_version = mig_6_to_7(conn)?;
             }
-            if curr_version == 7 {
-                curr_version = mig_7_to_8(conn)?;
-            }
-            if curr_version == 8 {
-                curr_version = mig_8_to_9(conn)?;
-            }
-            if curr_version == 9 {
-                curr_version = mig_9_to_10(conn)?;
-            }
-            if curr_version == 10 {
-                curr_version = mig_10_to_11(conn)?;
-            }
-
             if curr_version == DB_VERSION {
                 info!(
                     "All migration scripts completed successfully.  Welcome to v{}.",
@@ -347,7 +331,7 @@ fn mig_5_to_6(conn: &mut PooledConnection) -> Result<usize> {
                 if (tagval.len() % 2 == 0) && is_lower_hex(tagval) {
                     tx.execute(
                         "INSERT INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3);",
-                        params![event_id, tagname, hex::decode(tagval).ok()],
+                        params![event_id, tagname, hex::decode(&tagval).ok()],
                     )?;
                 } else {
                     // otherwise, insert as text
@@ -373,8 +357,7 @@ fn mig_6_to_7(conn: &mut PooledConnection) -> Result<usize> {
     info!("database schema needs update from 6->7");
     // only change is adding a hidden column to events.
     let upgrade_sql = r##"
-ALTER TABLE event ADD delegated_by BLOB;
+ALTER TABLE event ADD delegator BLOB;
-CREATE INDEX IF NOT EXISTS delegated_by_index ON event(delegated_by);
 PRAGMA user_version = 7;
 "##;
     match conn.execute_batch(upgrade_sql) {
@@ -388,84 +371,3 @@ PRAGMA user_version = 7;
     }
     Ok(7)
 }
-
-fn mig_7_to_8(conn: &mut PooledConnection) -> Result<usize> {
-    info!("database schema needs update from 7->8");
-    // Remove redundant indexes, and add a better multi-column index.
-    let upgrade_sql = r##"
-DROP INDEX IF EXISTS created_at_index;
-DROP INDEX IF EXISTS kind_index;
-CREATE INDEX IF NOT EXISTS event_composite_index ON event(kind,created_at);
-PRAGMA user_version = 8;
-"##;
-    match conn.execute_batch(upgrade_sql) {
-        Ok(()) => {
-            info!("database schema upgraded v7 -> v8");
-        }
-        Err(err) => {
-            error!("update failed: {}", err);
-            panic!("database could not be upgraded");
-        }
-    }
-    Ok(8)
-}
-
-fn mig_8_to_9(conn: &mut PooledConnection) -> Result<usize> {
-    info!("database schema needs update from 8->9");
-    // Those old indexes were actually helpful...
-    let upgrade_sql = r##"
-CREATE INDEX IF NOT EXISTS created_at_index ON event(created_at);
-CREATE INDEX IF NOT EXISTS event_composite_index ON event(kind,created_at);
-PRAGMA user_version = 9;
-"##;
-    match conn.execute_batch(upgrade_sql) {
-        Ok(()) => {
-            info!("database schema upgraded v8 -> v9");
-        }
-        Err(err) => {
-            error!("update failed: {}", err);
-            panic!("database could not be upgraded");
-        }
-    }
-    Ok(9)
-}
-
-fn mig_9_to_10(conn: &mut PooledConnection) -> Result<usize> {
-    info!("database schema needs update from 9->10");
-    // Those old indexes were actually helpful...
-    let upgrade_sql = r##"
-CREATE INDEX IF NOT EXISTS tag_composite_index ON tag(event_id,name,value_hex,value);
-PRAGMA user_version = 10;
-"##;
-    match conn.execute_batch(upgrade_sql) {
-        Ok(()) => {
-            info!("database schema upgraded v9 -> v10");
-        }
-        Err(err) => {
-            error!("update failed: {}", err);
-            panic!("database could not be upgraded");
-        }
-    }
-    Ok(10)
-}
-
-fn mig_10_to_11(conn: &mut PooledConnection) -> Result<usize> {
-    info!("database schema needs update from 10->11");
-    // Those old indexes were actually helpful...
-    let upgrade_sql = r##"
-CREATE INDEX IF NOT EXISTS tag_name_eid_index ON tag(name,event_id,value_hex);
-reindex;
-pragma optimize;
-PRAGMA user_version = 11;
-"##;
-    match conn.execute_batch(upgrade_sql) {
-        Ok(()) => {
-            info!("database schema upgraded v10 -> v11");
-        }
-        Err(err) => {
-            error!("update failed: {}", err);
-            panic!("database could not be upgraded");
-        }
-    }
-    Ok(11)
-}

src/server.rs

@@ -10,19 +10,15 @@ use crate::event::Event;
 use crate::event::EventCmd;
 use crate::info::RelayInfo;
 use crate::nip05;
-use crate::notice::Notice;
 use crate::subscription::Subscription;
 use futures::SinkExt;
 use futures::StreamExt;
-use governor::{Jitter, Quota, RateLimiter};
-use http::header::HeaderMap;
 use hyper::header::ACCEPT;
 use hyper::service::{make_service_fn, service_fn};
 use hyper::upgrade::Upgraded;
 use hyper::{
     header, server::conn::AddrStream, upgrade, Body, Request, Response, Server, StatusCode,
 };
-use rusqlite::OpenFlags;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use std::collections::HashMap;
@@ -88,32 +84,11 @@ async fn handle_web_request(
                                     Some(config),
                                 )
                                 .await;
-                                let user_agent = get_header_string("user-agent", request.headers());
+
-                                // determine the remote IP from headers if the exist
-                                let header_ip = settings
-                                    .network
-                                    .remote_ip_header
-                                    .as_ref()
-                                    .and_then(|x| get_header_string(x, request.headers()));
-                                // use the socket addr as a backup
-                                let remote_ip =
-                                    header_ip.unwrap_or_else(|| remote_addr.ip().to_string());
-                                let client_info = ClientInfo {
-                                    remote_ip,
-                                    user_agent,
-                                };
-                                // spawn a nostr server with our websocket
                                 tokio::spawn(nostr_server(
-                                    pool,
+                                    pool, settings, ws_stream, broadcast, event_tx, shutdown,
-                                    client_info,
-                                    settings,
-                                    ws_stream,
-                                    broadcast,
-                                    event_tx,
-                                    shutdown,
                                 ));
                             }
-                            // todo: trace, don't print...
                             Err(e) => println!(
                                 "error when trying to upgrade connection \
                                  from address {} to websocket connection. \
@@ -173,12 +148,6 @@ async fn handle_web_request(
     }
 }
 
-fn get_header_string(header: &str, headers: &HeaderMap) -> Option<String> {
-    headers
-        .get(header)
-        .and_then(|x| x.to_str().ok().map(|x| x.to_string()))
-}
-
 // return on a control-c or internally requested shutdown signal
 async fn ctrl_c_or_signal(mut shutdown_signal: Receiver<()>) {
     let mut term_signal = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
@@ -198,6 +167,7 @@ async fn ctrl_c_or_signal(mut shutdown_signal: Receiver<()>) {
         info!("Shutting down webserver due to SIGTERM");
         break;
         },
+
         }
     }
 }
@@ -246,14 +216,6 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
     let rt = Builder::new_multi_thread()
         .enable_all()
         .thread_name("tokio-ws")
-        // limit concurrent SQLite blocking threads
-        .max_blocking_threads(settings.limits.max_blocking_threads)
-        .on_thread_start(|| {
-            trace!("started new thread");
-        })
-        .on_thread_stop(|| {
-            trace!("stopping thread");
-        })
         .build()
         .unwrap();
     // start tokio
@@ -311,17 +273,6 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
                 }
             }
         }
-        // build a connection pool for DB maintenance
-        let maintenance_pool = db::build_pool(
-            "maintenance writer",
-            &settings,
-            OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE,
-            1,
-            1,
-            false,
-        );
-        db::db_maintenance(maintenance_pool).await;
-
         // listen for (external to tokio) shutdown request
         let controlled_shutdown = invoke_shutdown.clone();
         tokio::spawn(async move {
@@ -332,7 +283,6 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
                     controlled_shutdown.send(()).ok();
                 }
                 Err(std::sync::mpsc::RecvError) => {
-                    // FIXME: spurious error on startup?
                     debug!("shutdown requestor is disconnected");
                 }
             };
@@ -351,7 +301,8 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
         let pool = db::build_pool(
             "client query",
             &settings,
-            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY
+                | rusqlite::OpenFlags::SQLITE_OPEN_SHARED_CACHE,
             db_min_conn,
             db_max_conn,
             true,
@@ -427,65 +378,36 @@ 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(notice: Notice) -> Message {
+fn make_notice_message(msg: &str) -> Message {
-    let json = match notice {
+    Message::text(json!(["NOTICE", msg]).to_string())
-        Notice::Message(ref msg) => json!(["NOTICE", msg]),
-        Notice::EventResult(ref res) => json!(["OK", res.id, res.status.to_bool(), res.msg]),
-    };
-
-    Message::text(json.to_string())
-}
-
-struct ClientInfo {
-    remote_ip: String,
-    user_agent: Option<String>,
 }
 
 /// Handle new client connections.  This runs through an event loop
 /// for all client communication.
 async fn nostr_server(
     pool: db::SqlitePool,
-    client_info: ClientInfo,
     settings: Settings,
     mut ws_stream: WebSocketStream<Upgraded>,
     broadcast: Sender<Event>,
     event_tx: mpsc::Sender<SubmittedEvent>,
     mut shutdown: Receiver<()>,
 ) {
-    // the time this websocket nostr server started
-    let orig_start = Instant::now();
     // get a broadcast channel for clients to communicate on
     let mut bcast_rx = broadcast.subscribe();
     // Track internal client state
-    let mut conn = conn::ClientConn::new(client_info.remote_ip);
+    let mut conn = conn::ClientConn::new();
-    // subscription creation rate limiting
-    let mut sub_lim_opt = None;
-    // 100ms jitter when the rate limiter returns
-    let jitter = Jitter::up_to(Duration::from_millis(100));
-    let sub_per_min_setting = settings.limits.subscriptions_per_min;
-    if let Some(sub_per_min) = sub_per_min_setting {
-        if sub_per_min > 0 {
-            trace!("Rate limits for sub creation ({}/min)", sub_per_min);
-            let quota_time = core::num::NonZeroU32::new(sub_per_min).unwrap();
-            let quota = Quota::per_minute(quota_time);
-            sub_lim_opt = Some(RateLimiter::direct(quota));
-        }
-    }
-    // Use the remote IP as the client identifier
     let cid = conn.get_client_prefix();
     // Create a channel for receiving query results from the database.
     // we will send out the tx handle to any query we generate.
-    // this has capacity for some of the larger requests we see, which
+    let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(256);
-    // should allow the DB thread to release the handle earlier.
-    let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(20000);
     // Create channel for receiving NOTICEs
-    let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(128);
+    let (notice_tx, mut notice_rx) = mpsc::channel::<String>(32);
 
     // last time this client sent data (message, ping, etc.)
     let mut last_message_time = Instant::now();
 
     // ping interval (every 5 minutes)
-    let default_ping_dur = Duration::from_secs(settings.network.ping_interval_seconds.into());
+    let default_ping_dur = Duration::from_secs(300);
 
     // disconnect after 20 minutes without a ping response or event.
     let max_quiet_time = Duration::from_secs(60 * 20);
@@ -497,18 +419,16 @@ async fn nostr_server(
     // when these subscriptions are cancelled, make a message
     // available to the executing query so it knows to stop.
     let mut running_queries: HashMap<String, oneshot::Sender<()>> = HashMap::new();
+
     // for stats, keep track of how many events the client published,
     // 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 client connection (cid: {}, ip: {:?})", cid, conn.ip());
+    info!("new connection for client: {:?}", cid);
-    if let Some(ua) = client_info.user_agent {
-        debug!("cid: {}, user-agent: {:?}", cid, ua);
-    }
     loop {
         tokio::select! {
             _ = shutdown.recv() => {
-        info!("Close connection down due to shutdown, client: {}, ip: {:?}, connected: {:?}", cid, conn.ip(), orig_start.elapsed());
+        info!("Shutting client connection down due to shutdown: {:?}", cid);
                 // server shutting down, exit loop
                 break;
             },
@@ -523,7 +443,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
@@ -542,15 +462,12 @@ 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.
-                for (s, sub) in conn.subscriptions() {
+                let matching_subs = conn.get_matching_subscriptions(&global_event);
-                    if !sub.interested_in_event(&global_event) {
+                for s in matching_subs {
-                        continue;
-                    }
-
                     // TODO: serialize at broadcast time, instead of
                     // once for each consumer.
                     if let Ok(event_str) = serde_json::to_string(&global_event) {
-                        trace!("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
@@ -571,7 +488,7 @@ async fn nostr_server(
                     },
             Some(Ok(Message::Binary(_))) => {
             ws_stream.send(
-                make_notice_message(Notice::message("binary messages are not accepted".into()))).await.ok();
+                make_notice_message("binary messages are not accepted")).await.ok();
                         continue;
                     },
                     Some(Ok(Message::Ping(_) | Message::Pong(_))) => {
@@ -581,7 +498,8 @@ async fn nostr_server(
                     },
             Some(Err(WsError::Capacity(MessageTooLong{size, max_size}))) => {
             ws_stream.send(
-                make_notice_message(Notice::message(format!("message too large ({} > {})",size, max_size)))).await.ok();
+                make_notice_message(
+                &format!("message too large ({} > {})",size, max_size))).await.ok();
                         continue;
             },
                     None |
@@ -589,17 +507,17 @@ async fn nostr_server(
              Err(WsError::AlreadyClosed | WsError::ConnectionClosed |
                  WsError::Protocol(tungstenite::error::ProtocolError::ResetWithoutClosingHandshake)))
                         => {
-                            debug!("websocket close from client (cid: {}, ip: {:?})",cid, conn.ip());
+                        debug!("websocket close from client: {:?}",cid);
                         break;
                     },
                     Some(Err(WsError::Io(e))) => {
                         // IO errors are considered fatal
-                        warn!("IO error (cid: {}, ip: {:?}): {:?}", cid, conn.ip(), e);
+                        warn!("IO error (client: {:?}): {:?}", cid, e);
                         break;
                     }
                     x => {
                         // default condition on error is to close the client connection
-                        info!("unknown error (cid: {}, ip: {:?}): {:?} (closing conn)", cid, conn.ip(), x);
+                        info!("unknown error (client: {:?}): {:?} (closing conn)", cid, x);
                         break;
                     }
                 };
@@ -609,12 +527,11 @@ async fn nostr_server(
                     Ok(NostrMessage::EventMsg(ec)) => {
                         // An EventCmd needs to be validated to be converted into an Event
                         // handle each type of message
-                        let evid = ec.event_id().to_owned();
                         let parsed : Result<Event> = Result::<Event>::from(ec);
                         match parsed {
                             Ok(e) => {
                                 let id_prefix:String = e.id.chars().take(8).collect();
-                                debug!("successfully parsed/validated event: {:?} (cid: {})", 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.
@@ -622,50 +539,39 @@ 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 {
-                        let msg = format!("The event created_at field is out of the acceptable range (+{}sec) for this relay.",fut_sec);
+                    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 notice = Notice::invalid(e.id, &msg);
-                        ws_stream.send(make_notice_message(notice)).await.ok();
                     }
                 }
                             },
-                            Err(e) => {
+                            Err(_) => {
-                                info!("client sent an invalid event (cid: {})", cid);
+                                info!("client {:?} sent an invalid event", cid);
-                                ws_stream.send(make_notice_message(Notice::invalid(evid, &format!("{}", e)))).await.ok();
+                                ws_stream.send(make_notice_message("event was invalid")).await.ok();
                             }
                         }
                     },
                     Ok(NostrMessage::SubMsg(s)) => {
-                        debug!("subscription requested (cid: {}, sub: {:?})", cid, s.id);
+                        debug!("client {} requesting a subscription", cid);
                         // subscription handling consists of:
-                        // * check for rate limits
                         // * registering the subscription so future events can be matched
                         // * making a channel to cancel to request later
                         // * sending a request for a SQL query
-            // Do nothing if the sub already exists.
+                        let (abandon_query_tx, abandon_query_rx) = oneshot::channel::<()>();
-            if !conn.has_subscription(&s) {
+                        match conn.subscribe(s.clone()) {
-                if let Some(ref lim) = sub_lim_opt {
+                            Ok(()) => {
-                lim.until_ready_with_jitter(jitter).await;
+                                // when we insert, if there was a previous query running with the same name, cancel it.
-                }
+                                if let Some(previous_query) = running_queries.insert(s.id.to_owned(), abandon_query_tx) {
-                            let (abandon_query_tx, abandon_query_rx) = oneshot::channel::<()>();
+                                    previous_query.send(()).ok();
-                            match conn.subscribe(s.clone()) {
+                                }
-                Ok(()) => {
+                                // start a database query
-                                    // when we insert, if there was a previous query running with the same name, cancel it.
+                                db::db_query(s, cid.to_owned(), pool.clone(), query_tx.clone(), abandon_query_rx).await;
-                                    if let Some(previous_query) = running_queries.insert(s.id.to_owned(), abandon_query_tx) {
+                            },
-                    previous_query.send(()).ok();
+                            Err(e) => {
-                                    }
+                                info!("Subscription error: {}", e);
-                                    // start a database query.  this spawns a blocking database query on a worker thread.
+                                ws_stream.send(make_notice_message(&e.to_string())).await.ok();
-                                    db::db_query(s, cid.to_owned(), pool.clone(), query_tx.clone(), abandon_query_rx).await;
-                },
-                Err(e) => {
-                    info!("Subscription error: {} (cid: {}, sub: {:?})", e, cid, s.id);
-                                    ws_stream.send(make_notice_message(Notice::message(format!("Subscription error: {}", e)))).await.ok();
-                }
                             }
-            } else {
+                        }
-        info!("client sent duplicate subscription, ignoring (cid: {}, sub: {:?})", cid, s.id);
-        }
                     },
                     Ok(NostrMessage::CloseMsg(cc)) => {
                         // closing a request simply removes the subscription.
@@ -682,23 +588,23 @@ async fn nostr_server(
                                 conn.unsubscribe(&c);
                             } else {
                                 info!("invalid command ignored");
-                                ws_stream.send(make_notice_message(Notice::message("could not parse command".into()))).await.ok();
+                                ws_stream.send(make_notice_message("could not parse command")).await.ok();
                             }
                     },
                     Err(Error::ConnError) => {
-                        debug!("got connection close/error, disconnecting cid: {}, ip: {:?}",cid, conn.ip());
+                        debug!("got connection close/error, disconnecting client: {:?}",cid);
                         break;
                     }
                     Err(Error::EventMaxLengthError(s)) => {
-                        info!("client sent event larger ({} bytes) than max size (cid: {})", s, cid);
+                        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();
+                        ws_stream.send(make_notice_message("event exceeded max size")).await.ok();
                     },
                     Err(Error::ProtoParseError) => {
-                        info!("client sent event that could not be parsed (cid: {})", cid);
+                        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();
+                        ws_stream.send(make_notice_message("could not parse command")).await.ok();
                     },
                     Err(e) => {
-                        info!("got non-fatal error from client (cid: {}, error: {:?}", cid, e);
+                        info!("got non-fatal error from client: {:?}, error: {:?}", cid, e);
                     },
                 }
             },
@@ -709,11 +615,7 @@ async fn nostr_server(
         stop_tx.send(()).ok();
     }
     info!(
-        "stopping client connection (cid: {}, ip: {:?}, sent: {} events, recv: {} events, connected: {:?})",
+        "stopping connection for client: {:?} (client sent {} event(s), received {})",
-        cid,
+        cid, client_published_event_count, client_received_event_count
-        conn.ip(),
-        client_published_event_count,
-        client_received_event_count,
-	orig_start.elapsed()
     );
 }

src/subscription.rs

@@ -37,9 +37,6 @@ pub struct ReqFilter {
     #[serde(skip)]
     pub tags: Option<HashMap<char, HashSet<String>>>,
     /// Force no matches due to malformed data
-    // we can't represent it in the req filter, so we don't want to
-    // erroneously match.  This basically indicates the req tried to
-    // do something invalid.
     pub force_no_match: bool,
 }
 
@@ -220,17 +217,6 @@ impl ReqFilter {
             .unwrap_or(true)
     }
 
-    fn delegated_authors_match(&self, event: &Event) -> bool {
-        if let Some(delegated_pubkey) = &event.delegated_by {
-            self.authors
-                .as_ref()
-                .map(|vs| prefix_match(vs, delegated_pubkey))
-                .unwrap_or(true)
-        } else {
-            false
-        }
-    }
-
     fn tag_match(&self, event: &Event) -> bool {
         // get the hashset from the filter.
         if let Some(map) = &self.tags {
@@ -262,7 +248,7 @@ impl ReqFilter {
             && self.since.map(|t| event.created_at > t).unwrap_or(true)
             && self.until.map(|t| event.created_at < t).unwrap_or(true)
             && self.kind_match(event.kind)
-            && (self.authors_match(event) || self.delegated_authors_match(event))
+            && self.authors_match(event)
             && self.tag_match(event)
             && !self.force_no_match
     }
@@ -322,7 +308,6 @@ mod tests {
         let e = Event {
             id: "foo".to_owned(),
             pubkey: "abcd".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: Vec::new(),
@@ -341,7 +326,6 @@ mod tests {
         let e = Event {
             id: "abcd".to_owned(),
             pubkey: "".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: Vec::new(),
@@ -360,7 +344,6 @@ mod tests {
         let e = Event {
             id: "abcde".to_owned(),
             pubkey: "".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: Vec::new(),
@@ -380,7 +363,6 @@ mod tests {
         let e = Event {
             id: "abc".to_owned(),
             pubkey: "".to_owned(),
-            delegated_by: None,
             created_at: 50,
             kind: 0,
             tags: Vec::new(),
@@ -404,7 +386,6 @@ mod tests {
         let e = Event {
             id: "abc".to_owned(),
             pubkey: "".to_owned(),
-            delegated_by: None,
             created_at: 150,
             kind: 0,
             tags: Vec::new(),
@@ -426,7 +407,6 @@ mod tests {
         let e = Event {
             id: "abc".to_owned(),
             pubkey: "".to_owned(),
-            delegated_by: None,
             created_at: 50,
             kind: 0,
             tags: Vec::new(),
@@ -445,7 +425,6 @@ mod tests {
         let e = Event {
             id: "abc".to_owned(),
             pubkey: "".to_owned(),
-            delegated_by: None,
             created_at: 1001,
             kind: 0,
             tags: Vec::new(),
@@ -464,7 +443,6 @@ mod tests {
         let e = Event {
             id: "abc".to_owned(),
             pubkey: "".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: Vec::new(),
@@ -483,7 +461,6 @@ mod tests {
         let e = Event {
             id: "123".to_owned(),
             pubkey: "abc".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: Vec::new(),
@@ -502,7 +479,6 @@ mod tests {
         let e = Event {
             id: "123".to_owned(),
             pubkey: "bcd".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: Vec::new(),
@@ -521,7 +497,6 @@ mod tests {
         let e = Event {
             id: "123".to_owned(),
             pubkey: "xyz".to_owned(),
-            delegated_by: None,
             created_at: 0,
             kind: 0,
             tags: Vec::new(),