build: bump version to 0.3.1

improvement: logging failed queries and timing
build: bump version to 0.3.0
2022-01-01 19:26:15 -06:00 · 2022-01-01 19:25:09 -06:00 · 2022-01-01 18:40:57 -06:00 · 2022-01-01 18:38:52 -06:00 · 2022-01-01 12:27:09 -06:00 · 2022-01-01 09:13:13 -06:00
8 changed files with 394 additions and 88 deletions
@@ -66,6 +66,12 @@ dependencies = [
 "serde 1.0.131",
 ]

+[[package]]
+name = "bitcoin_hashes"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "006cc91e1a1d99819bc5b8214be3555c1f0611b169f527a1fdc54ed1f2b745b0"
+
 [[package]]
 name = "bitflags"
 version = "1.3.2"
@@ -355,6 +361,25 @@ dependencies = [
 "smallvec",
 ]

+[[package]]
+name = "h2"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f072413d126e57991455e0a922b31e4c8ba7c2ffbebf6b78b4f8521397d65cd"
+dependencies = [
+ "bytes",
+ "fnv",
+ "futures-core",
+ "futures-sink",
+ "futures-util",
+ "http",
+ "indexmap",
+ "slab",
+ "tokio",
+ "tokio-util",
+ "tracing",
+]
+
 [[package]]
 name = "hashbrown"
 version = "0.11.2"
@@ -399,18 +424,59 @@ dependencies = [
 "itoa",
 ]

+[[package]]
+name = "http-body"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ff4f84919677303da5f147645dbea6b1881f368d03ac84e1dc09031ebd7b2c6"
+dependencies = [
+ "bytes",
+ "http",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "httparse"
 version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "acd94fdbe1d4ff688b67b04eee2e17bd50995534a61539e45adfefb45e5e5503"

+[[package]]
+name = "httpdate"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c4a1e36c821dbe04574f602848a19f742f4fb3c98d40449f11bcad18d6b17421"
+
 [[package]]
 name = "humantime"
 version = "2.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"

+[[package]]
+name = "hyper"
+version = "0.14.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7ec3e62bdc98a2f0393a5048e4c30ef659440ea6e0e572965103e72bd836f55"
+dependencies = [
+ "bytes",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "h2",
+ "http",
+ "http-body",
+ "httparse",
+ "httpdate",
+ "itoa",
+ "pin-project-lite",
+ "socket2",
+ "tokio",
+ "tower-service",
+ "tracing",
+ "want",
+]
+
 [[package]]
 name = "idna"
 version = "0.2.3"
@@ -422,6 +488,16 @@ dependencies = [
 "unicode-normalization",
 ]

+[[package]]
+name = "indexmap"
+version = "1.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc633605454125dec4b66843673f01c7df2b89479b32e0ed634e43a91cff62a5"
+dependencies = [
+ "autocfg 1.0.1",
+ "hashbrown",
+]
+
 [[package]]
 name = "instant"
 version = "0.1.12"
@@ -573,15 +649,16 @@ checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"

 [[package]]
 name = "nostr-rs-relay"
-version = "0.2.1"
+version = "0.3.1"
 dependencies = [
- "bitcoin_hashes",
+ "bitcoin_hashes 0.9.7",
 "config",
 "env_logger",
 "futures",
 "futures-util",
 "governor",
 "hex",
+ "hyper",
 "lazy_static",
 "log",
 "nonzero_ext",
@@ -960,10 +1037,9 @@ checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"
 [[package]]
 name = "secp256k1"
 version = "0.20.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97d03ceae636d0fed5bae6a7f4f664354c5f4fcedf6eef053fef17e49f837d0a"
+source = "git+https://github.com/rust-bitcoin/rust-secp256k1.git?rev=50034ccb18fdd84904ab3aa6c84a12fcced33209#50034ccb18fdd84904ab3aa6c84a12fcced33209"
 dependencies = [
- "bitcoin_hashes",
+ "bitcoin_hashes 0.10.0",
 "rand 0.6.5",
 "secp256k1-sys",
 "serde 1.0.131",
@@ -972,8 +1048,7 @@ dependencies = [
 [[package]]
 name = "secp256k1-sys"
 version = "0.4.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "827cb7cce42533829c792fc51b82fbf18b125b45a702ef2c8be77fce65463a7b"
+source = "git+https://github.com/rust-bitcoin/rust-secp256k1.git?rev=50034ccb18fdd84904ab3aa6c84a12fcced33209#50034ccb18fdd84904ab3aa6c84a12fcced33209"
 dependencies = [
 "cc",
 ]
@@ -1061,6 +1136,16 @@ version = "1.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1ecab6c735a6bb4139c0caafd0cc3635748bbb3acf4550e8138122099251f309"

+[[package]]
+name = "socket2"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5dc90fe6c7be1a323296982db1836d1ea9e47b6839496dde9a541bc496df3516"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
 [[package]]
 name = "static_assertions"
 version = "1.1.0"
@@ -1165,6 +1250,20 @@ dependencies = [
 "tungstenite",
 ]

+[[package]]
+name = "tokio-util"
+version = "0.6.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e99e1983e5d376cd8eb4b66604d2e99e79f5bd988c3055891dcd8c9e2604cc0"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "log",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "toml"
 version = "0.5.8"
@@ -1174,6 +1273,38 @@ dependencies = [
 "serde 1.0.131",
 ]

+[[package]]
+name = "tower-service"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "360dfd1d6d30e05fda32ace2c8c70e9c0a9da713275777f5a4dbb8a1893930c6"
+
+[[package]]
+name = "tracing"
+version = "0.1.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "375a639232caf30edfc78e8d89b2d4c375515393e7af7e16f01cd96917fb2105"
+dependencies = [
+ "cfg-if",
+ "pin-project-lite",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-core"
+version = "0.1.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1f4ed65637b8390770814083d20756f87bfa2c21bf2f110babdc5438351746e4"
+dependencies = [
+ "lazy_static",
+]
+
+[[package]]
+name = "try-lock"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59547bce71d9c38b83d9c0e92b6066c4253371f15005def0c30d9657f50c7642"
+
 [[package]]
 name = "tungstenite"
 version = "0.16.0"
@@ -1259,6 +1390,16 @@ version = "0.9.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5fecdca9a5291cc2b8dcf7dc02453fee791a280f3743cb0905f8822ae463b3fe"

+[[package]]
+name = "want"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ce8a968cb1cd110d136ff8b819a556d6fb6d919363c61534f6860c7eb172ba0"
+dependencies = [
+ "log",
+ "try-lock",
+]
+
 [[package]]
 name = "wasi"
 version = "0.10.2+wasi-snapshot-preview1"
@@ -1,6 +1,6 @@
 [package]
 name = "nostr-rs-relay"
-version = "0.2.1"
+version = "0.3.1"
 edition = "2021"

 [dependencies]
@@ -15,7 +15,7 @@ thiserror = "^1"
 uuid = { version = "^0.8", features = ["v4"] }
 config = { version = "0.11", features = ["toml"] }
 bitcoin_hashes = { version = "^0.9", features = ["serde"] }
-secp256k1 = { version = "^0.20", features = ["rand", "rand-std", "serde", "bitcoin_hashes"] }
+secp256k1 = {git = "https://github.com/rust-bitcoin/rust-secp256k1.git", rev = "50034ccb18fdd84904ab3aa6c84a12fcced33209", features = ["rand", "rand-std", "serde", "bitcoin_hashes"] }
 serde = { version = "^1.0", features = ["derive"] }
 serde_json = "^1.0"
 hex = "^0.4"
@@ -23,3 +23,4 @@ rusqlite = "^0.26"
 lazy_static = "^1.4"
 governor = "^0.4"
 nonzero_ext = "^0.3"
+hyper={ version="0.14", features=["server","http1","http2","tcp"] }
@@ -39,6 +39,9 @@ Text Note [81cf...2652] from 296a...9b92 5 seconds ago
  hello world
 ```

+A pre-built container is also available on DockerHub:
+https://hub.docker.com/repository/docker/scsibug/nostr-rs-relay
+
 ## Configuration

 The sample `[config.toml](config.toml)` file demonstrates the
@@ -13,6 +13,7 @@ use rusqlite::OpenFlags;
 use crate::config::SETTINGS;
 use std::path::Path;
 use std::thread;
+use std::time::Instant;
 use tokio::task;

 /// Database file
@@ -157,12 +158,17 @@ pub async fn db_writer(
            }
            let mut event_write = false;
            let event = next_event.unwrap();
+            let start = Instant::now();
            match write_event(&mut conn, &event) {
                Ok(updated) => {
                    if updated == 0 {
                        debug!("ignoring duplicate event");
                    } else {
-                        info!("persisted event: {}", event.get_event_id_prefix());
+                        info!(
+                            "persisted event: {} in {:?}",
+                            event.get_event_id_prefix(),
+                            start.elapsed()
+                        );
                        event_write = true;
                        // send this out to all clients
                        bcast_tx.send(event.clone()).ok();
@@ -302,35 +308,52 @@ fn query_from_sub(sub: &Subscription) -> String {
            filter_components.push(authors_clause);
        }
        // Query for Kind
-        if f.kind.is_some() {
+        if let Some(ks) = &f.kinds {
            // kind is number, no escaping needed
-            let kind_clause = format!("kind = {}", f.kind.unwrap());
+            let str_kinds: Vec<String> = ks.iter().map(|x| x.to_string()).collect();
+            let kind_clause = format!("kind IN ({})", str_kinds.join(", "));
            filter_components.push(kind_clause);
        }
        // Query for event
-        if f.id.is_some() {
-            let id_str = f.id.as_ref().unwrap();
-            if is_hex(id_str) {
-                let id_clause = format!("event_hash = x'{}'", id_str);
+        if f.ids.is_some() {
+            let ids_escaped: Vec<String> = f
+                .ids
+                .as_ref()
+                .unwrap()
+                .iter()
+                .filter(|&x| is_hex(x))
+                .map(|x| format!("x'{}'", x))
+                .collect();
+            let id_clause = format!("event_hash IN ({})", ids_escaped.join(", "));
            filter_components.push(id_clause);
        }
-        }
        // Query for referenced event
-        if f.event.is_some() {
-            let ev_str = f.event.as_ref().unwrap();
-            if is_hex(ev_str) {
-                let ev_clause = format!("referenced_event = x'{}'", ev_str);
-                filter_components.push(ev_clause);
-            }
-        }
-        // Query for referenced pet name pubkey
-        if f.pubkey.is_some() {
-            let pet_str = f.pubkey.as_ref().unwrap();
-            if is_hex(pet_str) {
-                let pet_clause = format!("referenced_pubkey = x'{}'", pet_str);
-                filter_components.push(pet_clause);
+        if f.events.is_some() {
+            let events_escaped: Vec<String> = f
+                .events
+                .as_ref()
+                .unwrap()
+                .iter()
+                .filter(|&x| is_hex(x))
+                .map(|x| format!("x'{}'", x))
+                .collect();
+            let events_clause = format!("referenced_event IN ({})", events_escaped.join(", "));
+            filter_components.push(events_clause);
        }
+        // Query for referenced pubkey
+        if f.pubkeys.is_some() {
+            let pubkeys_escaped: Vec<String> = f
+                .pubkeys
+                .as_ref()
+                .unwrap()
+                .iter()
+                .filter(|&x| is_hex(x))
+                .map(|x| format!("x'{}'", x))
+                .collect();
+            let pubkeys_clause = format!("referenced_pubkey IN ({})", pubkeys_escaped.join(", "));
+            filter_components.push(pubkeys_clause);
        }
+
        // Query for timestamp
        if f.since.is_some() {
            let created_clause = format!("created_at > {}", f.since.unwrap());
@@ -350,7 +373,7 @@ fn query_from_sub(sub: &Subscription) -> String {
            filter_clauses.push(fc);
        } else {
            // never display hidden events
-            filter_clauses.push("hidden!=FALSE".to_owned());
+            filter_clauses.push("hidden!=TRUE".to_owned());
        }
    }

@@ -385,6 +408,8 @@ pub async fn db_query(
            Connection::open_with_flags(&full_path, OpenFlags::SQLITE_OPEN_READ_ONLY).unwrap();
        debug!("opened database for reading");
        debug!("going to query for: {:?}", sub);
+        let mut row_count: usize = 0;
+        let start = Instant::now();
        // generate SQL query
        let q = query_from_sub(&sub);
        // execute the query
@@ -396,6 +421,7 @@ pub async fn db_query(
                debug!("query aborted");
                return;
            }
+            row_count += 1;
            // TODO: check before unwrapping
            let event_json = row.get(0).unwrap();
            query_tx
@@ -405,6 +431,10 @@ pub async fn db_query(
                })
                .ok();
        }
-        debug!("query completed");
+        debug!(
+            "query completed ({} rows) in {:?}",
+            row_count,
+            start.elapsed()
+        );
    });
 }
@@ -3,14 +3,19 @@ use crate::config;
 use crate::error::Error::*;
 use crate::error::Result;
 use bitcoin_hashes::{sha256, Hash};
+use lazy_static::lazy_static;
 use log::*;
-use secp256k1::{schnorrsig, Secp256k1};
+use secp256k1::{schnorr, Secp256k1, VerifyOnly, XOnlyPublicKey};
 use serde::{Deserialize, Deserializer, Serialize};
 use serde_json::value::Value;
 use serde_json::Number;
 use std::str::FromStr;
 use std::time::SystemTime;

+lazy_static! {
+    pub static ref SECP: Secp256k1<VerifyOnly> = Secp256k1::verification_only();
+}
+
 /// Event command in network format
 #[derive(Serialize, Deserialize, PartialEq, Debug, Clone)]
 pub struct EventCmd {
@@ -109,12 +114,15 @@ impl Event {
            return false;
        }
        // * validate the message digest (sig) using the pubkey & computed sha256 message hash.
-        let secp = Secp256k1::new();
-        let sig = schnorrsig::Signature::from_str(&self.sig).unwrap();
-        let message = secp256k1::Message::from(digest);
-        let pubkey = schnorrsig::PublicKey::from_str(&self.pubkey).unwrap();
-        let verify = secp.schnorrsig_verify(&sig, &message, &pubkey);
+        let sig = schnorr::Signature::from_str(&self.sig).unwrap();
+        if let Ok(msg) = secp256k1::Message::from_slice(digest.as_ref()) {
+            let pubkey = XOnlyPublicKey::from_str(&self.pubkey).unwrap();
+            let verify = SECP.verify_schnorr(&sig, &msg, &pubkey);
            matches!(verify, Ok(()))
+        } else {
+            warn!("Error converting digest to secp256k1 message");
+            false
+        }
    }

    /// Convert event to canonical representation for signing.
@@ -1,6 +1,11 @@
 //! Server process
 use futures::SinkExt;
 use futures::StreamExt;
+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 log::*;
 use nostr_rs_relay::close::Close;
 use nostr_rs_relay::config;
@@ -12,14 +17,17 @@ use nostr_rs_relay::protostream;
 use nostr_rs_relay::protostream::NostrMessage::*;
 use nostr_rs_relay::protostream::NostrResponse::*;
 use std::collections::HashMap;
+use std::convert::Infallible;
 use std::env;
+use std::net::SocketAddr;
 use std::path::Path;
-use tokio::net::{TcpListener, TcpStream};
 use tokio::runtime::Builder;
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::{Receiver, Sender};
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
+use tokio_tungstenite::WebSocketStream;
+use tungstenite::handshake;
 use tungstenite::protocol::WebSocketConfig;

 fn db_from_args(args: Vec<String>) -> Option<String> {
@@ -28,6 +36,88 @@ fn db_from_args(args: Vec<String>) -> Option<String> {
    }
    None
 }
+async fn handle_web_request(
+    mut request: Request<Body>,
+    remote_addr: SocketAddr,
+    broadcast: Sender<Event>,
+    event_tx: tokio::sync::mpsc::Sender<Event>,
+    shutdown: Receiver<()>,
+) -> Result<Response<Body>, Infallible> {
+    match (
+        request.uri().path(),
+        request.headers().contains_key(header::UPGRADE),
+    ) {
+        //if the request is ws_echo and the request headers contains an Upgrade key
+        ("/", true) => {
+            debug!("websocket with upgrade request");
+            //assume request is a handshake, so create the handshake response
+            let response = match handshake::server::create_response_with_body(&request, || {
+                Body::empty()
+            }) {
+                Ok(response) => {
+                    //in case the handshake response creation succeeds,
+                    //spawn a task to handle the websocket connection
+                    tokio::spawn(async move {
+                        //using the hyper feature of upgrading a connection
+                        match upgrade::on(&mut request).await {
+                            //if successfully upgraded
+                            Ok(upgraded) => {
+                                //create a websocket stream from the upgraded object
+                                let ws_stream = WebSocketStream::from_raw_socket(
+                                    //pass the upgraded object
+                                    //as the base layer stream of the Websocket
+                                    upgraded,
+                                    tokio_tungstenite::tungstenite::protocol::Role::Server,
+                                    None,
+                                )
+                                .await;
+                                tokio::spawn(nostr_server(
+                                    ws_stream, broadcast, event_tx, shutdown,
+                                ));
+                            }
+                            Err(e) => println!(
+                                "error when trying to upgrade connection \
+                                 from address {} to websocket connection. \
+                                 Error is: {}",
+                                remote_addr, e
+                            ),
+                        }
+                    });
+                    //return the response to the handshake request
+                    response
+                }
+                Err(error) => {
+                    warn!("websocket response failed");
+                    let mut res =
+                        Response::new(Body::from(format!("Failed to create websocket: {}", error)));
+                    *res.status_mut() = StatusCode::BAD_REQUEST;
+                    return Ok(res);
+                }
+            };
+            Ok::<_, Infallible>(response)
+        }
+        ("/", false) => {
+            // handle request at root with no upgrade header
+            Ok(Response::new(Body::from(
+                "This is a Nostr relay.\n".to_string(),
+            )))
+        }
+        (_, _) => {
+            //handle any other url
+            Ok(Response::builder()
+                .status(StatusCode::NOT_FOUND)
+                .body(Body::from("Nothing here."))
+                .unwrap())
+        }
+    }
+}
+
+async fn shutdown_signal() {
+    // Wait for the CTRL+C signal
+    tokio::signal::ctrl_c()
+        .await
+        .expect("failed to install CTRL+C signal handler");
+}

 /// Start running a Nostr relay server.
 fn main() -> Result<(), Error> {
@@ -46,6 +136,7 @@ fn main() -> Result<(), Error> {
        }
        *settings = c;
    }
+
    let config = config::SETTINGS.read().unwrap();
    // do some config validation.
    if !Path::new(&config.database.data_directory).is_dir() {
@@ -54,6 +145,7 @@ fn main() -> Result<(), Error> {
    }
    debug!("config: {:?}", config);
    let addr = format!("{}:{}", config.network.address.trim(), config.network.port);
+    let socket_addr = addr.parse().expect("listening address not valid");
    // configure tokio runtime
    let rt = Builder::new_multi_thread()
        .enable_all()
@@ -63,8 +155,7 @@ fn main() -> Result<(), Error> {
    // start tokio
    rt.block_on(async {
        let settings = config::SETTINGS.read().unwrap();
-        let listener = TcpListener::bind(&addr).await.expect("Failed to bind");
-        info!("listening on: {}", addr);
+        info!("listening on: {}", socket_addr);
        // all client-submitted valid events are broadcast to every
        // other client on this channel.  This should be large enough
        // to accomodate slower readers (messages are dropped if
@@ -77,7 +168,7 @@ fn main() -> Result<(), Error> {
        // requested server shutdown.
        let (invoke_shutdown, _) = broadcast::channel::<()>(1);
        let ctrl_c_shutdown = invoke_shutdown.clone();
-        // listen for ctrl-c interruupts
+        // // listen for ctrl-c interruupts
        tokio::spawn(async move {
            tokio::signal::ctrl_c().await.unwrap();
            info!("shutting down due to SIGINT");
@@ -87,28 +178,35 @@ fn main() -> Result<(), Error> {
        // writing events, and for publishing events that have been
        // written (to all connected clients).
        db::db_writer(event_rx, bcast_tx.clone(), invoke_shutdown.subscribe()).await;
-
-        // track unique client connection count
-        let mut client_accept_count: usize = 0;
-        let mut stop_listening = invoke_shutdown.subscribe();
-        // handle new client connection requests, or SIGINT signals.
-        loop {
-            tokio::select! {
-                _ = stop_listening.recv() => {
-                    break;
-                }
-                Ok((stream, _)) = listener.accept() => {
-                    client_accept_count += 1;
-                    info!("creating new connection for client #{}",client_accept_count);
-                    tokio::spawn(nostr_server(
-                        stream,
-                        bcast_tx.clone(),
-                        event_tx.clone(),
-                        invoke_shutdown.subscribe(),
-                    ));
-                }
+        info!("db writer created");
+        // A `Service` is needed for every connection, so this
+        // creates one from our `handle_request` function.
+        let make_svc = make_service_fn(|conn: &AddrStream| {
+            let remote_addr = conn.remote_addr();
+            let bcast = bcast_tx.clone();
+            let event = event_tx.clone();
+            let stop = invoke_shutdown.clone();
+            async move {
+                // service_fn converts our function into a `Service`
+                Ok::<_, Infallible>(service_fn(move |request: Request<Body>| {
+                    handle_web_request(
+                        request,
+                        remote_addr,
+                        bcast.clone(),
+                        event.clone(),
+                        stop.subscribe(),
+                    )
+                }))
            }
+        });
+        let server = Server::bind(&socket_addr)
+            .serve(make_svc)
+            .with_graceful_shutdown(shutdown_signal());
+        // run hyper
+        if let Err(e) = server.await {
+            eprintln!("server error: {}", e);
        }
+        // our code
    });
    Ok(())
 }
@@ -116,7 +214,7 @@ fn main() -> Result<(), Error> {
 /// Handle new client connections.  This runs through an event loop
 /// for all client communication.
 async fn nostr_server(
-    stream: TcpStream,
+    ws_stream: WebSocketStream<Upgraded>,
    broadcast: Sender<Event>,
    event_tx: tokio::sync::mpsc::Sender<Event>,
    mut shutdown: Receiver<()>,
@@ -130,8 +228,8 @@ async fn nostr_server(
        config.max_frame_size = settings.limits.max_ws_frame_bytes;
    }
    // upgrade the TCP connection to WebSocket
-    let conn = tokio_tungstenite::accept_async_with_config(stream, Some(config)).await;
-    let ws_stream = conn.expect("websocket handshake error");
+    //let conn = tokio_tungstenite::accept_async_with_config(stream, Some(config)).await;
+    //let ws_stream = conn.expect("websocket handshake error");
    // wrap websocket into a stream & sink of Nostr protocol messages
    let mut nostr_stream = protostream::wrap_ws_in_nostr(ws_stream);
    // Track internal client state
@@ -9,9 +9,9 @@ use futures::sink::Sink;
 use futures::stream::Stream;
 use futures::task::Context;
 use futures::task::Poll;
+use hyper::upgrade::Upgraded;
 use log::*;
 use serde::{Deserialize, Serialize};
-use tokio::net::TcpStream;
 use tokio_tungstenite::WebSocketStream;
 use tungstenite::error::Error as WsError;
 use tungstenite::protocol::Message;
@@ -40,11 +40,11 @@ pub enum NostrResponse {

 /// A Nostr protocol stream is layered on top of a Websocket stream.
 pub struct NostrStream {
-    ws_stream: WebSocketStream<TcpStream>,
+    ws_stream: WebSocketStream<Upgraded>,
 }

 /// Given a websocket, return a protocol stream wrapper.
-pub fn wrap_ws_in_nostr(ws: WebSocketStream<TcpStream>) -> NostrStream {
+pub fn wrap_ws_in_nostr(ws: WebSocketStream<Upgraded>) -> NostrStream {
    NostrStream { ws_stream: ws }
 }

@@ -71,6 +71,7 @@ impl Stream for NostrStream {
                }
                Err(e) => {
                    debug!("proto parse error: {:?}", e);
+                    debug!("parse error on message: {}", msg.trim());
                    Err(Error::ProtoParseError)
                }
            }
@@ -2,6 +2,7 @@
 use crate::error::Result;
 use crate::event::Event;
 use serde::{Deserialize, Deserializer, Serialize};
+use std::collections::HashSet;

 /// Subscription identifier and set of request filters
 #[derive(Serialize, PartialEq, Debug, Clone)]
@@ -17,16 +18,16 @@ pub struct Subscription {
 /// absent ([`None`]) if it should be ignored.
 #[derive(Serialize, Deserialize, PartialEq, Debug, Clone)]
 pub struct ReqFilter {
-    /// Event hash
-    pub id: Option<String>,
-    /// Event kind
-    pub kind: Option<u64>,
+    /// Event hashes
+    pub ids: Option<Vec<String>>,
+    /// Event kinds
+    pub kinds: Option<Vec<u64>>,
    /// Referenced event hash
    #[serde(rename = "#e")]
-    pub event: Option<String>,
+    pub events: Option<Vec<String>>,
    /// Referenced public key for a petname
    #[serde(rename = "#p")]
-    pub pubkey: Option<String>,
+    pub pubkeys: Option<Vec<String>>,
    /// Events published after this time
    pub since: Option<u64>,
    /// Events published before this time
@@ -105,8 +106,13 @@ impl Subscription {

 impl ReqFilter {
    /// Check for a match within the authors list.
-    // TODO: Ambiguity; what if the array is empty?  Should we
-    // consider that the same as null?
+    fn ids_match(&self, event: &Event) -> bool {
+        self.ids
+            .as_ref()
+            .map(|vs| vs.contains(&event.id.to_owned()))
+            .unwrap_or(true)
+    }
+
    fn authors_match(&self, event: &Event) -> bool {
        self.authors
            .as_ref()
@@ -115,29 +121,47 @@ impl ReqFilter {
    }
    /// Check if this filter either matches, or does not care about the event tags.
    fn event_match(&self, event: &Event) -> bool {
-        self.event
-            .as_ref()
-            .map(|t| event.event_tag_match(t))
-            .unwrap_or(true)
+        // This needs to be analyzed for performance; building these
+        // hash sets for each active subscription isn't great.
+        if let Some(es) = &self.events {
+            let event_refs =
+                HashSet::<_>::from_iter(event.get_event_tags().iter().map(|x| x.to_owned()));
+            let filter_refs = HashSet::<_>::from_iter(es.iter().map(|x| &x[..]));
+            let cardinality = event_refs.intersection(&filter_refs).count();
+            cardinality > 0
+        } else {
+            true
+        }
    }

    /// Check if this filter either matches, or does not care about
    /// the pubkey/petname tags.
    fn pubkey_match(&self, event: &Event) -> bool {
-        self.pubkey
-            .as_ref()
-            .map(|t| event.pubkey_tag_match(t))
-            .unwrap_or(true)
+        // This needs to be analyzed for performance; building these
+        // hash sets for each active subscription isn't great.
+        if let Some(ps) = &self.pubkeys {
+            let pubkey_refs =
+                HashSet::<_>::from_iter(event.get_pubkey_tags().iter().map(|x| x.to_owned()));
+            let filter_refs = HashSet::<_>::from_iter(ps.iter().map(|x| &x[..]));
+            let cardinality = pubkey_refs.intersection(&filter_refs).count();
+            cardinality > 0
+        } else {
+            true
+        }
    }

    /// Check if this filter either matches, or does not care about the kind.
    fn kind_match(&self, kind: u64) -> bool {
-        self.kind.map(|v| v == kind).unwrap_or(true)
+        self.kinds
+            .as_ref()
+            .map(|ks| ks.contains(&kind))
+            .unwrap_or(true)
    }

    /// Determine if all populated fields in this filter match the provided event.
    pub fn interested_in_event(&self, event: &Event) -> bool {
-        self.id.as_ref().map(|v| v == &event.id).unwrap_or(true)
+        //        self.id.as_ref().map(|v| v == &event.id).unwrap_or(true)
+        self.ids_match(event)
            && self.since.map(|t| event.created_at > t).unwrap_or(true)
            && self.kind_match(event.kind)
            && self.authors_match(event)
Author	SHA1	Message	Date
Greg Heartsfield	19637d612e	build: bump version to 0.3.1	2022-01-01 19:26:15 -06:00
Greg Heartsfield	afc9a0096a	improvement: logging failed queries and timing	2022-01-01 19:25:09 -06:00
Greg Heartsfield	3d56262386	build: bump version to 0.3.0	2022-01-01 18:40:57 -06:00
Greg Heartsfield	6673fcfd11	feat: implement multi-valued filter searching NIP-01 now uses arrays instead of scalars. Fixes https://todo.sr.ht/~gheartsfield/nostr-rs-relay/17	2022-01-01 18:38:52 -06:00
Greg Heartsfield	b5da3fa2b0	docs: link to docker hub	2022-01-01 12:27:09 -06:00
Greg Heartsfield	850957213e	build: bump version to 0.2.3	2022-01-01 09:13:13 -06:00
Greg Heartsfield	1aa5a5458d	improvement: event signature validation is 100x faster Switched to latest (git) release of secp256k1, which has more efficient verification-only context for Schnorr. Switched to single pre-instantiated instance of the verifier.	2022-01-01 09:08:19 -06:00
Greg Heartsfield	620e227699	fix: connection issues with Firefox This adds Hyper, and a 200 response code. Prior to this, Firefox would fail to connect. There is also a text document displayed at the root URL to indicate this is a Nostr relay. Fixes https://todo.sr.ht/~gheartsfield/nostr-rs-relay/15	2022-01-01 08:11:20 -06:00
Greg Heartsfield	14e59ed278	build: bump version to 0.2.2	2021-12-31 16:34:52 -06:00
Greg Heartsfield	5ad383f257	fix: incorrect logic on empty filters for hidden events	2021-12-31 16:34:10 -06:00