build: bump version to 0.6.2

Updating async-trait v0.1.56 -> v0.1.57
Removing block-buffer v0.7.3
Removing block-padding v0.1.5
Updating bumpalo v3.10.0 -> v3.11.0
Removing byte-tools v0.3.1
Updating bytes v1.1.0 -> v1.2.1
Updating cpufeatures v0.2.2 -> v0.2.3
Updating crossbeam-utils v0.8.10 -> v0.8.11
Updating crypto-common v0.1.4 -> v0.1.6
Removing digest v0.8.1
Removing fake-simd v0.1.2
Updating fastrand v1.7.0 -> v1.8.0
Updating futures v0.3.21 -> v0.3.23
Updating futures-channel v0.3.21 -> v0.3.23
Updating futures-core v0.3.21 -> v0.3.23
Updating futures-executor v0.3.21 -> v0.3.23
Updating futures-io v0.3.21 -> v0.3.23
Updating futures-macro v0.3.21 -> v0.3.23
Updating futures-sink v0.3.21 -> v0.3.23
Updating futures-task v0.3.21 -> v0.3.23
Updating futures-util v0.3.21 -> v0.3.23
Removing generic-array v0.12.4
Removing generic-array v0.14.5
Adding   generic-array v0.14.6
Updating h2 v0.3.13 -> v0.3.14
Updating hashbrown v0.12.1 -> v0.12.3
Updating hyper v0.14.19 -> v0.14.20
Updating itoa v1.0.2 -> v1.0.3
Updating js-sys v0.3.58 -> v0.3.59
Updating libc v0.2.126 -> v0.2.132
Removing maplit v1.0.2
Updating once_cell v1.12.1 -> v1.13.1
Removing opaque-debug v0.2.3
Updating openssl v0.10.40 -> v0.10.41
Updating openssl-sys v0.9.74 -> v0.9.75
Updating pest v2.1.3 -> v2.2.1
Updating pest_derive v2.1.0 -> v2.2.1
Updating pest_generator v2.1.3 -> v2.2.1
Updating pest_meta v2.1.3 -> v2.2.1
Updating proc-macro2 v1.0.40 -> v1.0.43
Updating quote v1.0.20 -> v1.0.21
Updating raw-cpuid v10.3.0 -> v10.5.0
Updating redox_syscall v0.2.13 -> v0.2.16
Updating regex v1.5.6 -> v1.6.0
Updating regex-syntax v0.6.26 -> v0.6.27
Updating ryu v1.0.10 -> v1.0.11
Updating security-framework v2.6.1 -> v2.7.0
Updating serde v1.0.138 -> v1.0.143
Updating serde_derive v1.0.138 -> v1.0.143
Updating serde_json v1.0.82 -> v1.0.83
Removing sha-1 v0.8.2
Updating slab v0.4.6 -> v0.4.7
Updating syn v1.0.98 -> v1.0.99
Updating thiserror v1.0.31 -> v1.0.32
Updating thiserror-impl v1.0.31 -> v1.0.32
Updating tokio v1.19.2 -> v1.20.1
Updating tokio-tungstenite v0.17.1 -> v0.17.2
Updating tracing v0.1.35 -> v0.1.36
Updating tracing-core v0.1.28 -> v0.1.29
Updating tungstenite v0.17.2 -> v0.17.3
Updating ucd-trie v0.1.3 -> v0.1.4
Updating unicode-ident v1.0.1 -> v1.0.3
Updating wasm-bindgen v0.2.81 -> v0.2.82
Updating wasm-bindgen-backend v0.2.81 -> v0.2.82
Updating wasm-bindgen-macro v0.2.81 -> v0.2.82
Updating wasm-bindgen-macro-support v0.2.81 -> v0.2.82
Updating wasm-bindgen-shared v0.2.81 -> v0.2.82
Updating web-sys v0.3.58 -> v0.3.59

Cargo.toml

@@ -1,28 +1,32 @@
 [package]
 name = "nostr-rs-relay"
-version = "0.4.2"
+version = "0.6.2"
 edition = "2021"
 
 [dependencies]
 log = "^0.4"
 env_logger = "^0.9"
-tokio = { version = "^1.14", features = ["full"] }
+tokio = { version = "^1.16", features = ["full"] }
 futures = "^0.3"
 futures-util = "^0.3"
-tokio-tungstenite = "^0.16"
-tungstenite = "^0.16"
+tokio-tungstenite = "^0.17"
+tungstenite = "^0.17"
 thiserror = "^1"
 uuid = { version = "^0.8", features = ["v4"] }
-config = { version = "0.11", features = ["toml"] }
-bitcoin_hashes = { version = "^0.9", features = ["serde"] }
-secp256k1 = {git = "https://github.com/rust-bitcoin/rust-secp256k1.git", rev = "50034ccb18fdd84904ab3aa6c84a12fcced33209", features = ["rand", "rand-std", "serde", "bitcoin_hashes"] }
+config = { version = "^0.12", features = ["toml"] }
+bitcoin_hashes = { version = "^0.10", features = ["serde"] }
+secp256k1 = {version = "^0.21", features = ["rand", "rand-std", "serde", "bitcoin_hashes"] }
 serde = { version = "^1.0", features = ["derive"] }
 serde_json = {version = "^1.0", features = ["preserve_order"]}
 hex = "^0.4"
-rusqlite = { version = "^0.26", features = ["limits"]}
+rusqlite = { version = "^0.26", features = ["limits","bundled"]}
 r2d2 = "^0.8"
 r2d2_sqlite = "^0.19"
 lazy_static = "^1.4"
 governor = "^0.4"
 nonzero_ext = "^0.3"
-hyper={ version="0.14", features=["server","http1","http2","tcp"] }
+hyper = { version="0.14", features=["client", "server","http1","http2","tcp"] }
+hyper-tls = "^0.5"
+http = { version = "^0.2" }
+parse_duration = "^2"
+rand = "^0.8"

Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.57 as builder
+FROM docker.io/library/rust:1.63.0@sha256:d7e3f69edcdcd03b145d8d9361765b816656755e49c1c1fe28224a4505f91b0a as builder
 
 RUN USER=root cargo new --bin nostr-rs-relay
 WORKDIR ./nostr-rs-relay
@@ -12,11 +12,11 @@ COPY ./src ./src
 RUN rm ./target/release/deps/nostr*relay*
 RUN cargo build --release
 
-FROM debian:buster-slim
+FROM docker.io/library/debian:bullseye-20220801-slim@sha256:139a42fa3bde3e5bad6ae912aaaf2103565558a7a73afe6ce6ceed6e46a6e519
 ARG APP=/usr/src/app
 ARG APP_DATA=/usr/src/app/db
 RUN apt-get update \
-    && apt-get install -y ca-certificates tzdata sqlite3 \
+    && apt-get install -y ca-certificates tzdata sqlite3 libc6 \
     && rm -rf /var/lib/apt/lists/*
 
 EXPOSE 8080

README.md

@@ -17,10 +17,12 @@ NIPs with a relay-specific implementation are listed here.
 - [x] NIP-01: Id/Author prefix search (_experimental_)
 - [x] NIP-02: Hide old contact list events
 - [ ] NIP-03: OpenTimestamps
-- [ ] NIP-05: Mapping Nostr keys to DNS identifiers
-- [ ] NIP-09: Event deletion
+- [x] NIP-05: Mapping Nostr keys to DNS identifiers
+- [x] NIP-09: Event deletion
 - [x] NIP-11: Relay information document
 - [x] NIP-12: Generic tag search (_experimental_)
+- [x] NIP-15: End of stored events notice
+- [x] NIP-16: Replaceable and ephemeral events
 
 ## Quick Start
 
@@ -79,8 +81,10 @@ termination, load balancing, and other features), see [Reverse
 Proxy](reverse-proxy.md).
 
 ## Dev Channel
-The current dev discussions for this project is happening at https://discord.gg/ufG6fH52Vk.
-Drop in to query any development related questions.
+
+For development discussions, please feel free to use the [sourcehut
+mailing list](https://lists.sr.ht/~gheartsfield/nostr-rs-relay-devel).
+Or, drop by the [Nostr Telegram Channel](https://t.me/nostr_protocol).
 
 License
 ---

config.toml

@@ -22,7 +22,7 @@ description = "A newly created nostr-rs-relay.\n\nCustomize this with your own i
 # line option.
 data_directory = "."
 
-# Database connection pool settings:
+# Database connection pool settings for subscribers:
 
 # Minimum number of SQLite reader connections
 #min_conn = 4
@@ -59,17 +59,44 @@ reject_future_seconds = 1800
 #max_ws_frame_bytes = 131072
 
 # Broadcast buffer size, in number of events.  This prevents slow
-# readers from consuming memory.  Defaults to 4096.
-#broadcast_buffer = 4096
+# readers from consuming memory.
+#broadcast_buffer = 16384
 
 # Event persistence buffer size, in number of events.  This provides
-# backpressure to senders if writes are slow.  Defaults to 16.
-#event_persist_buffer = 16
+# backpressure to senders if writes are slow.
+#event_persist_buffer = 4096
 
 [authorization]
 # Pubkey addresses in this array are whitelisted for event publishing.
-# Only valid events by these authors will be accepted.
+# Only valid events by these authors will be accepted, if the variable
+# is set.
 #pubkey_whitelist = [
 #  "35d26e4690cbe1a898af61cc3515661eb5fa763b57bd0b42e45099c8b32fd50f",
 #  "887645fef0ce0c3c1218d2f5d8e6132a19304cdc57cd20281d082f38cfea0072",
 #]
+
+[verified_users]
+# NIP-05 verification of users.  Can be "enabled" to require NIP-05
+# metadata for event authors, "passive" to perform validation but
+# never block publishing, or "disabled" to do nothing.
+#mode = "disabled"
+
+# Domain names that will be prevented from publishing events.
+#domain_blacklist = ["wellorder.net"]
+
+# Domain names that are allowed to publish events.  If defined, only
+# events NIP-05 verified authors at these domains are persisted.
+#domain_whitelist = ["example.com"]
+
+# Consider an pubkey "verified" if we have a successful validation
+# from the NIP-05 domain within this amount of time.  Note, if the
+# domain provides a successful response that omits the account,
+# verification is immediately revoked.
+#verify_expiration = "1 week"
+
+# How long to wait between verification attempts for a specific author.
+#verify_update_frequency = "24 hours"
+
+# How many consecutive failed checks before we give up on verifying
+# this author.
+#max_consecutive_failures = 20

docs/user-verification-nip05.md

@@ -0,0 +1,248 @@
+# Author Verification Design Document
+
+The relay will use NIP-05 DNS-based author verification to limit which
+authors can publish events to a relay.  This document describes how
+this feature will operate.
+
+## Considerations
+
+DNS-based author verification is designed to be deployed in relays that
+want to prevent spam, so there should be strong protections to prevent
+unauthorized authors from persisting data.  This includes data needed to
+verify new authors.
+
+There should be protections in place to ensure the relay cannot be
+used to spam or flood other webservers.  Additionally, there should be
+protections against server-side request forgery (SSRF).
+
+## Design Overview
+
+### Concepts
+
+All authors are initially "unverified".  Unverified authors that submit
+appropriate `NIP-05` metadata events become "candidates" for
+verification.  A candidate author becomes verified  when the relay
+inspects a kind `0` metadata event for the author with a `nip05` field,
+and follows the procedure in `NIP-05` to successfully associate the
+author with an internet identifier.
+
+The `NIP-05` procedure verifies an author for a fixed period of time,
+configurable by the relay operator.  If this "verification expiration
+time" (`verify_expiration`) is exceeded without being refreshed, they
+are once again unverified.
+
+Verified authors have their status regularly and automatically updated
+through scheduled polling to their verified domain, this process is
+"re-verification".  It is performed based on the configuration setting
+`verify_update_frequency`, which defines how long the relay waits
+between verification attempts (whether the result was success or
+failure).
+
+Authors may change their verification data (the internet identifier from
+`NIP-05`) with a new metadata event, which then requires
+re-verification.  Their old verification remains valid until
+expiration.
+
+Performing candidate author verification is a best-effort activity and
+may be significantly rate-limited to prevent relays being used to
+attack other hosts.  Candidate verification (untrusted authors) should
+never impact re-verification (trusted authors).
+
+## Operating Modes
+
+The relay may operate in one of three modes.  "Disabled" performs no
+validation activities, and will never permit or deny events based on
+an author's NIP-05 metadata.  "Passive" performs NIP-05 validation,
+but does not permit or deny events based on the validity or presence
+of NIP-05 metadata.  "Enabled" will require current and valid NIP-05
+metadata for any events to be persisted.  "Enabled" mode will
+additionally consider domain whitelist/blacklist configuration data to
+restrict which author's events are persisted.
+
+## Design Details
+
+### Data Storage
+
+Verification is stored in a dedicated table.  This tracks:
+
+* `nip05` identifier
+* most recent verification timestamp
+* most recent verification failure timestamp
+* reference to the metadata event (used for tracking `created_at` and
+  `pubkey`)
+
+### Event Handling
+
+All events are first validated to ensure the signature is valid.
+
+Incoming events of kind _other_ than metadata (kind `0`) submitted by
+clients will be evaluated as follows.
+
+* If the event's author has a current verification, the event is
+  persisted as normal.
+* If the event's author has either no verification, or the
+  verification is expired, the event is rejected.
+
+If the event is a metadata event, we handle it differently.
+
+We first determine the verification status of the event's pubkey.
+
+* If the event author is unverified, AND the event contains a `nip05`
+  key, we consider this a verification candidate.
+* If the event author is unverified, AND the event does not contain a
+  `nip05` key, this is not a candidate, and the event is dropped.
+
+* If the event author is verified, AND the event contains a `nip05`
+  key that is identical to the currently stored value, no special
+  action is needed.
+* If the event author is verified, AND the event contains a different
+  `nip05` than was previously verified, with a more recent timestamp,
+  we need to re-verify.
+* If the event author is verified, AND the event is missing a `nip05`
+  key, and the event timestamp is more recent than what was verified,
+  we do nothing.  The current verification will be allowed to expire.
+
+### Candidate Verification
+
+When a candidate verification is requested, a rate limit will be
+utilized.  If the rate limit is exceeded, new candidate verification
+requests will be dropped.  In practice, this is implemented by a
+size-limited channel that drops events that exceed a threshold.
+
+Candidates are never persisted in the database.
+
+### Re-Verification
+
+Re-verification is straightforward when there has been no change to
+the `nip05` key.  A new request to the `nip05` domain is performed,
+and if successful, the verification timestamp is updated to the
+current time.  If the request fails due to a timeout or server error,
+the failure timestamp is updated instead.
+
+When the the `nip05` key has changed and this event is more recent, we
+will create a new verification record, and delete all other records
+for the same name.
+
+Regarding creating new records vs. updating: We never update the event
+reference or `nip05` identifier in a verification record. Every update
+either reset the last failure or last success timestamp.
+
+### Determining Verification Status
+
+In determining if an event is from a verified author, the following
+procedure should be used:
+
+Join the verification table with the event table, to provide
+verification data alongside the event `created_at` and `pubkey`
+metadata.  Find the most recent verification record for the author,
+based on the `created_at` time.
+
+Reject the record if the success timestamp is not within our
+configured expiration time.
+
+Reject records with disallowed domains, based on any whitelists or
+blacklists in effect.
+
+If a result remains, the author is treated as verified.
+
+This does give a time window for authors transitioning their verified
+status between domains.  There may be a period of time in which there
+are multiple valid rows in the verification table for a given author.
+
+### Cleaning Up Inactive Verifications
+
+After a author verification has expired, we will continue to check for
+it to become valid again.  After a configurable number of attempts, we
+should simply forget it, and reclaim the space.
+
+### Addition of Domain Whitelist/Blacklist
+
+A set of whitelisted or blacklisted domains may be provided.  If both
+are provided, only the whitelist is used.  In this context, domains
+are either "allowed" (present on a whitelist and NOT present on a
+blacklist), or "denied" (NOT present on a whitelist and present on a
+blacklist).
+
+The processes outlined so far are modified in the presence of these
+options:
+
+* Only authors with allowed domains can become candidates for
+  verification.
+* Verification status queries additionally filter out any denied
+  domains.
+* Re-verification processes only proceed with allowed domains.
+
+### Integration
+
+We have an existing database writer thread, which receives events and
+attempts to persist them to disk.  Once validated and persisted, these
+events are broadcast to all subscribers.
+
+When verification is enabled, the writer must check to ensure a valid,
+unexpired verification record exists for the auther.  All metadata
+events (regardless of verification status) are forwarded to a verifier
+module.  If the verifier determines a new verification record is
+needed, it is also responsible for persisting and broadcasting the
+event, just as the database writer would have done.
+
+## Threat Scenarios
+
+Some of these mitigations are fully implemented, others are documented
+simply to demonstrate a mitigation is possible.
+
+### Domain Spamming
+
+*Threat*: A author with a high-volume of events creates a metadata event
+with a bogus domain, causing the relay to generate significant
+unwanted traffic to a target.
+
+*Mitigation*: Rate limiting for all candidate verification will limit
+external requests to a reasonable amount.  Currently, this is a simple
+delay that slows down the HTTP task.
+
+### Denial of Service for Legitimate Authors
+
+*Threat*: A author with a high-volume of events creates a metadata event
+with a domain that is invalid for them, _but which is used by other
+legitimate authors_.  This triggers rate-limiting against the legitimate
+domain, and blocks authors from updating their own metadata.
+
+*Mitigation*: Rate limiting should only apply to candidates, so any
+existing verified authors have priority for re-verification.  New
+authors will be affected, as we can not distinguish between the threat
+and a legitimate author. _(Unimplemented)_
+
+### Denial of Service by Consuming Storage
+
+*Threat*: A author creates a high volume of random metadata events with
+unique domains, in order to cause us to store large amounts of data
+for to-be-verified authors.
+
+*Mitigation*: No data is stored for candidate authors.  This makes it
+harder for new authors to become verified, but is effective at
+preventing this attack.
+
+### Metadata Replay for Verified Author
+
+*Threat*: Attacker replays out-of-date metadata event for a author, to
+cause a verification to fail.
+
+*Mitigation*: New metadata events have their signed timestamp compared
+against the signed timestamp of the event that has most recently
+verified them.  If the metadata event is older, it is discarded.
+
+### Server-Side Request Forgery via Metadata
+
+*Threat*: Attacker includes malicious data in the `nip05` event, which
+is used to generate HTTP requests against potentially internal
+resources.  Either leaking data, or invoking webservices beyond their
+own privileges.
+
+*Mitigation*: Consider detecting and dropping when the `nip05` field
+is an IP address.  Allow the relay operator to utilize the `blacklist`
+or `whitelist` to constrain hosts that will be contacted.  Most
+importantly, the verification process is hardcoded to only make
+requests to a known url path
+(`.well-known/nostr.json?name=<LOCAL_NAME>`).  The `<LOCAL_NAME>`
+component is restricted to a basic ASCII subset (preventing additional
+URL components).

src/close.rs

@@ -1,4 +1,6 @@
 //! Subscription close request parsing
+//!
+//! Representation and parsing of `CLOSE` messages sent from clients.
 use crate::error::{Error, Result};
 use serde::{Deserialize, Serialize};
 
@@ -11,7 +13,7 @@ pub struct CloseCmd {
     id: String,
 }
 
-/// Close command parsed
+/// Identifier of the subscription to be closed.
 #[derive(Serialize, Deserialize, PartialEq, Debug, Clone)]
 pub struct Close {
     /// The subscription identifier being closed.

src/config.rs

@@ -1,7 +1,10 @@
+//! Configuration file and settings management
+use config::{Config, ConfigError, File};
 use lazy_static::lazy_static;
 use log::*;
 use serde::{Deserialize, Serialize};
 use std::sync::RwLock;
+use std::time::Duration;
 
 // initialize a singleton default configuration
 lazy_static! {
@@ -67,6 +70,60 @@ pub struct Authorization {
     pub pubkey_whitelist: Option<Vec<String>>, // If present, only allow these pubkeys to publish events
 }
 
+#[derive(Serialize, Deserialize, PartialEq, Debug, Clone)]
+#[serde(rename_all = "lowercase")]
+pub enum VerifiedUsersMode {
+    Enabled,
+    Passive,
+    Disabled,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(unused)]
+pub struct VerifiedUsers {
+    pub mode: VerifiedUsersMode, // Mode of operation: "enabled" (enforce) or "passive" (check only). If none, this is simply disabled.
+    pub domain_whitelist: Option<Vec<String>>, // If present, only allow verified users from these domains can publish events
+    pub domain_blacklist: Option<Vec<String>>, // If present, allow all verified users from any domain except these
+    pub verify_expiration: Option<String>, // how long a verification is cached for before no longer being used
+    pub verify_update_frequency: Option<String>, // how often to attempt to update verification
+    pub verify_expiration_duration: Option<Duration>, // internal result of parsing verify_expiration
+    pub verify_update_frequency_duration: Option<Duration>, // internal result of parsing verify_update_frequency
+    pub max_consecutive_failures: usize, // maximum number of verification failures in a row, before ceasing future checks
+}
+
+impl VerifiedUsers {
+    pub fn init(&mut self) {
+        self.verify_expiration_duration = self.verify_expiration_duration();
+        self.verify_update_frequency_duration = self.verify_update_duration();
+    }
+
+    pub fn is_enabled(&self) -> bool {
+        self.mode == VerifiedUsersMode::Enabled
+    }
+
+    pub fn is_active(&self) -> bool {
+        self.mode == VerifiedUsersMode::Enabled || self.mode == VerifiedUsersMode::Passive
+    }
+
+    pub fn is_passive(&self) -> bool {
+        self.mode == VerifiedUsersMode::Passive
+    }
+
+    pub fn verify_expiration_duration(&self) -> Option<Duration> {
+        self.verify_expiration
+            .as_ref()
+            .and_then(|x| parse_duration::parse(x).ok())
+    }
+    pub fn verify_update_duration(&self) -> Option<Duration> {
+        self.verify_update_frequency
+            .as_ref()
+            .and_then(|x| parse_duration::parse(x).ok())
+    }
+    pub fn is_valid(&self) -> bool {
+        self.verify_expiration_duration().is_some() && self.verify_update_duration().is_some()
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 #[allow(unused)]
 pub struct Settings {
@@ -75,33 +132,36 @@ pub struct Settings {
     pub network: Network,
     pub limits: Limits,
     pub authorization: Authorization,
+    pub verified_users: VerifiedUsers,
     pub retention: Retention,
     pub options: Options,
 }
 
 impl Settings {
     pub fn new() -> Self {
-        let d = Self::default();
+        let default_settings = Self::default();
         // attempt to construct settings with file
-        //        Self::new_from_default(&d).unwrap_or(d)
-        let from_file = Self::new_from_default(&d);
+        let from_file = Self::new_from_default(&default_settings);
         match from_file {
             Ok(f) => f,
             Err(e) => {
                 warn!("Error reading config file ({:?})", e);
-                d
+                default_settings
             }
         }
     }
 
-    fn new_from_default(default: &Settings) -> Result<Self, config::ConfigError> {
-        let config: config::Config = config::Config::new();
-        let settings: Settings = config
+    fn new_from_default(default: &Settings) -> Result<Self, ConfigError> {
+        let builder = Config::builder();
+        let config: Config = builder
             // use defaults
-            .with_merged(config::Config::try_from(default).unwrap())?
+            .add_source(Config::try_from(default)?)
             // override with file contents
-            .with_merged(config::File::with_name("config"))?
-            .try_into()?;
+            .add_source(File::with_name("config"))
+            .build()?
+            .try_into()
+            .unwrap();
+        let mut settings: Settings = config.try_deserialize()?;
         // ensure connection pool size is logical
         if settings.database.min_conn > settings.database.max_conn {
             panic!(
@@ -109,6 +169,12 @@ impl Settings {
                 settings.database.min_conn, settings.database.max_conn
             );
         }
+        // ensure durations parse
+        if !settings.verified_users.is_valid() {
+            panic!("VerifiedUsers time settings could not be parsed");
+        }
+        // initialize durations for verified users
+        settings.verified_users.init();
         Ok(settings)
     }
 }
@@ -137,12 +203,22 @@ impl Default for Settings {
                 max_event_bytes: Some(2 << 17),      // 128K
                 max_ws_message_bytes: Some(2 << 17), // 128K
                 max_ws_frame_bytes: Some(2 << 17),   // 128K
-                broadcast_buffer: 4096,
-                event_persist_buffer: 16,
+                broadcast_buffer: 16384,
+                event_persist_buffer: 4096,
             },
             authorization: Authorization {
                 pubkey_whitelist: None, // Allow any address to publish
             },
+            verified_users: VerifiedUsers {
+                mode: VerifiedUsersMode::Disabled,
+                domain_whitelist: None,
+                domain_blacklist: None,
+                verify_expiration: Some("1 week".to_owned()),
+                verify_update_frequency: Some("1 day".to_owned()),
+                verify_expiration_duration: None,
+                verify_update_frequency_duration: None,
+                max_consecutive_failures: 20,
+            },
             retention: Retention {
                 max_events: None,          // max events
                 max_bytes: None,           // max size

src/error.rs

@@ -40,8 +40,49 @@ pub enum Error {
     ConfigError(config::ConfigError),
     #[error("Data directory does not exist")]
     DatabaseDirError,
+    #[error("Database Connection Pool Error")]
+    DatabasePoolError(r2d2::Error),
     #[error("Custom Error : {0}")]
     CustomError(String),
+    #[error("Task join error")]
+    JoinError,
+    #[error("Hyper Client error")]
+    HyperError(hyper::Error),
+    #[error("Hex encoding error")]
+    HexError(hex::FromHexError),
+    #[error("Unknown/Undocumented")]
+    UnknownError,
+}
+
+//impl From<Box<dyn std::error::Error>> for Error {
+//    fn from(e: Box<dyn std::error::Error>) -> Self {
+//        Error::CustomError("error".to_owned())
+//    }
+//}
+
+impl From<hex::FromHexError> for Error {
+    fn from(h: hex::FromHexError) -> Self {
+        Error::HexError(h)
+    }
+}
+
+impl From<hyper::Error> for Error {
+    fn from(h: hyper::Error) -> Self {
+        Error::HyperError(h)
+    }
+}
+
+impl From<r2d2::Error> for Error {
+    fn from(d: r2d2::Error) -> Self {
+        Error::DatabasePoolError(d)
+    }
+}
+
+impl From<tokio::task::JoinError> for Error {
+    /// Wrap SQL error
+    fn from(_j: tokio::task::JoinError) -> Self {
+        Error::JoinError
+    }
 }
 
 impl From<rusqlite::Error> for Error {

src/event.rs

@@ -2,6 +2,8 @@
 use crate::config;
 use crate::error::Error::*;
 use crate::error::Result;
+use crate::nip05;
+use crate::utils::unix_time;
 use bitcoin_hashes::{sha256, Hash};
 use lazy_static::lazy_static;
 use log::*;
@@ -12,20 +14,20 @@ use serde_json::Number;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::str::FromStr;
-use std::time::SystemTime;
 
 lazy_static! {
+    /// Secp256k1 verification instance.
     pub static ref SECP: Secp256k1<VerifyOnly> = Secp256k1::verification_only();
 }
 
-/// Event command in network format
+/// Event command in network format.
 #[derive(Serialize, Deserialize, PartialEq, Debug, Clone)]
 pub struct EventCmd {
     cmd: String, // expecting static "EVENT"
     event: Event,
 }
 
-/// Event parsed
+/// Parsed nostr event.
 #[derive(Serialize, Deserialize, PartialEq, Debug, Clone)]
 pub struct Event {
     pub id: String,
@@ -37,9 +39,9 @@ pub struct Event {
     pub(crate) tags: Vec<Vec<String>>,
     pub(crate) content: String,
     pub(crate) sig: String,
-    // Optimization for tag search, built on demand
+    // Optimization for tag search, built on demand.
     #[serde(skip)]
-    pub(crate) tagidx: Option<HashMap<String, HashSet<String>>>,
+    pub(crate) tagidx: Option<HashMap<char, HashSet<String>>>,
 }
 
 /// Simple tag type for array of array of strings.
@@ -54,6 +56,25 @@ where
     Ok(opt.unwrap_or_else(Vec::new))
 }
 
+/// Attempt to form a single-char tag name.
+pub fn single_char_tagname(tagname: &str) -> Option<char> {
+    // We return the tag character if and only if the tagname consists
+    // of a single char.
+    let mut tagnamechars = tagname.chars();
+    let firstchar = tagnamechars.next();
+    return match firstchar {
+        Some(_) => {
+            // check second char
+            if tagnamechars.next().is_none() {
+                firstchar
+            } else {
+                None
+            }
+        }
+        None => None,
+    };
+}
+
 /// Convert network event to parsed/validated event.
 impl From<EventCmd> for Result<Event> {
     fn from(ec: EventCmd) -> Result<Event> {
@@ -70,15 +91,26 @@ impl From<EventCmd> for Result<Event> {
     }
 }
 
-/// Seconds since 1970
-fn unix_time() -> u64 {
-    SystemTime::now()
-        .duration_since(SystemTime::UNIX_EPOCH)
-        .map(|x| x.as_secs())
-        .unwrap_or(0)
-}
-
 impl Event {
+    pub fn is_kind_metadata(&self) -> bool {
+        self.kind == 0
+    }
+
+    /// Pull a NIP-05 Name out of the event, if one exists
+    pub fn get_nip05_addr(&self) -> Option<nip05::Nip05Name> {
+        if self.is_kind_metadata() {
+            // very quick check if we should attempt to parse this json
+            if self.content.contains("\"nip05\"") {
+                // Parse into JSON
+                let md_parsed: Value = serde_json::from_str(&self.content).ok()?;
+                let md_map = md_parsed.as_object()?;
+                let nip05_str = md_map.get("nip05")?.as_str()?;
+                return nip05::Nip05Name::try_from(nip05_str).ok();
+            }
+        }
+        None
+    }
+
     /// Build an event tag index
     fn build_index(&mut self) {
         // if there are no tags; just leave the index as None
@@ -86,17 +118,22 @@ impl Event {
             return;
         }
         // otherwise, build an index
-        let mut idx: HashMap<String, HashSet<String>> = HashMap::new();
+        let mut idx: HashMap<char, HashSet<String>> = HashMap::new();
         // iterate over tags that have at least 2 elements
         for t in self.tags.iter().filter(|x| x.len() > 1) {
             let tagname = t.get(0).unwrap();
+            let tagnamechar_opt = single_char_tagname(tagname);
+            if tagnamechar_opt.is_none() {
+                continue;
+            }
+            let tagnamechar = tagnamechar_opt.unwrap();
             let tagval = t.get(1).unwrap();
             // ensure a vector exists for this tag
-            if !idx.contains_key(tagname) {
-                idx.insert(tagname.clone(), HashSet::new());
+            if !idx.contains_key(&tagnamechar) {
+                idx.insert(tagnamechar.clone(), HashSet::new());
             }
             // get the tag vec and insert entry
-            let tidx = idx.get_mut(tagname).expect("could not get tag vector");
+            let tidx = idx.get_mut(&tagnamechar).expect("could not get tag vector");
             tidx.insert(tagval.clone());
         }
         // save the tag structure
@@ -107,6 +144,19 @@ impl Event {
     pub fn get_event_id_prefix(&self) -> String {
         self.id.chars().take(8).collect()
     }
+    pub fn get_author_prefix(&self) -> String {
+        self.pubkey.chars().take(8).collect()
+    }
+
+    /// Retrieve tag values
+    pub fn tag_values_by_name(&self, tag_name: &str) -> Vec<String> {
+        self.tags
+            .iter()
+            .filter(|x| x.len() > 1)
+            .filter(|x| x.get(0).unwrap() == tag_name)
+            .map(|x| x.get(1).unwrap().to_owned())
+            .collect()
+    }
 
     /// Check if this event has a valid signature.
     fn is_valid(&self) -> bool {
@@ -120,7 +170,7 @@ impl Event {
             if curr_time + (allowable_future as u64) < self.created_at {
                 let delta = self.created_at - curr_time;
                 debug!(
-                    "Event is too far in the future ({} seconds), rejecting",
+                    "event is too far in the future ({} seconds), rejecting",
                     delta
                 );
                 return false;
@@ -133,7 +183,7 @@ impl Event {
         // * serialize with no spaces/newlines
         let c_opt = self.to_canonical();
         if c_opt.is_none() {
-            info!("event could not be canonicalized");
+            debug!("event could not be canonicalized");
             return false;
         }
         let c = c_opt.unwrap();
@@ -142,6 +192,7 @@ impl Event {
         let hex_digest = format!("{:x}", digest);
         // * ensure the id matches the computed sha256sum.
         if self.id != hex_digest {
+            debug!("event id does not match digest");
             return false;
         }
         // * validate the message digest (sig) using the pubkey & computed sha256 message hash.
@@ -152,11 +203,11 @@ impl Event {
                 let verify = SECP.verify_schnorr(&sig, &msg, &pubkey);
                 matches!(verify, Ok(()))
             } else {
-                info!("Client sent malformed pubkey");
+                debug!("client sent malformed pubkey");
                 false
             }
         } else {
-            warn!("Error converting digest to secp256k1 message");
+            info!("error converting digest to secp256k1 message");
             false
         }
     }
@@ -199,9 +250,10 @@ impl Event {
     }
 
     /// Determine if the given tag and value set intersect with tags in this event.
-    pub fn generic_tag_val_intersect(&self, tagname: &str, check: &HashSet<String>) -> bool {
+    pub fn generic_tag_val_intersect(&self, tagname: char, check: &HashSet<String>) -> bool {
         match &self.tagidx {
-            Some(idx) => match idx.get(tagname) {
+            // check if this is indexable tagname
+            Some(idx) => match idx.get(&tagname) {
                 Some(valset) => {
                     let common = valset.intersection(check);
                     common.count() > 0
@@ -249,7 +301,7 @@ mod tests {
     fn empty_event_tag_match() -> Result<()> {
         let event = simple_event();
         assert!(!event
-            .generic_tag_val_intersect("e", &HashSet::from(["foo".to_owned(), "bar".to_owned()])));
+            .generic_tag_val_intersect('e', &HashSet::from(["foo".to_owned(), "bar".to_owned()])));
         Ok(())
     }
 
@@ -260,7 +312,7 @@ mod tests {
         event.build_index();
         assert_eq!(
             event.generic_tag_val_intersect(
-                "e",
+                'e',
                 &HashSet::from(["foo".to_owned(), "bar".to_owned()])
             ),
             true
@@ -315,6 +367,32 @@ mod tests {
         assert_eq!(c, expected);
     }
 
+    #[test]
+    fn event_tag_select() {
+        let e = Event {
+            id: "999".to_owned(),
+            pubkey: "012345".to_owned(),
+            created_at: 501234,
+            kind: 1,
+            tags: vec![
+                vec!["j".to_owned(), "abc".to_owned()],
+                vec!["e".to_owned(), "foo".to_owned()],
+                vec!["e".to_owned(), "bar".to_owned()],
+                vec!["e".to_owned(), "baz".to_owned()],
+                vec![
+                    "p".to_owned(),
+                    "aaaa".to_owned(),
+                    "ws://example.com".to_owned(),
+                ],
+            ],
+            content: "this is a test".to_owned(),
+            sig: "abcde".to_owned(),
+            tagidx: None,
+        };
+        let v = e.tag_values_by_name("e");
+        assert_eq!(v, vec!["foo", "bar", "baz"]);
+    }
+
     #[test]
     fn event_canonical_with_tags() {
         let e = Event {

src/hexrange.rs

@@ -0,0 +1,159 @@
+//! Utilities for searching hexadecimal
+use crate::utils::is_hex;
+use hex;
+
+/// Types of hexadecimal queries.
+#[derive(PartialEq, Debug, Clone)]
+pub enum HexSearch {
+    // when no range is needed, exact 32-byte
+    Exact(Vec<u8>),
+    // lower (inclusive) and upper range (exclusive)
+    Range(Vec<u8>, Vec<u8>),
+    // lower bound only, upper bound is MAX inclusive
+    LowerOnly(Vec<u8>),
+}
+
+/// Check if a string contains only f chars
+fn is_all_fs(s: &str) -> bool {
+    s.chars().all(|x| x == 'f' || x == 'F')
+}
+
+/// Find the next hex sequence greater than the argument.
+pub fn hex_range(s: &str) -> Option<HexSearch> {
+    // handle special cases
+    if !is_hex(s) || s.len() > 64 {
+        return None;
+    }
+    if s.len() == 64 {
+        return Some(HexSearch::Exact(hex::decode(s).ok()?));
+    }
+    // if s is odd, add a zero
+    let mut hash_base = s.to_owned();
+    let mut odd = hash_base.len() % 2 != 0;
+    if odd {
+        // extend the string to make it even
+        hash_base.push('0');
+    }
+    let base = hex::decode(hash_base).ok()?;
+    // check for all ff's
+    if is_all_fs(s) {
+        // there is no higher bound, we only want to search for blobs greater than this.
+        return Some(HexSearch::LowerOnly(base));
+    }
+
+    // return a range
+    let mut upper = base.clone();
+    let mut byte_len = upper.len();
+
+    // for odd strings, we made them longer, but we want to increment the upper char (+16).
+    // we know we can do this without overflowing because we explicitly set the bottom half to 0's.
+    while byte_len > 0 {
+        byte_len -= 1;
+        // check if byte can be incremented, or if we need to carry.
+        let b = upper[byte_len];
+        if b == u8::MAX {
+            // reset and carry
+            upper[byte_len] = 0;
+        } else if odd {
+            // check if first char in this byte is NOT 'f'
+            if b < 240 {
+                upper[byte_len] = b + 16; // bump up the first character in this byte
+                                          // increment done, stop iterating through the vec
+                break;
+            } else {
+                // if it is 'f', reset the byte to 0 and do a carry
+                // reset and carry
+                upper[byte_len] = 0;
+            }
+            // done with odd logic, so don't repeat this
+            odd = false;
+        } else {
+            // bump up the first character in this byte
+            upper[byte_len] = b + 1;
+            // increment done, stop iterating
+            break;
+        }
+    }
+    Some(HexSearch::Range(base, upper))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::error::Result;
+
+    #[test]
+    fn hex_range_exact() -> Result<()> {
+        let hex = "abcdef00abcdef00abcdef00abcdef00abcdef00abcdef00abcdef00abcdef00";
+        let r = hex_range(hex);
+        assert_eq!(
+            r,
+            Some(HexSearch::Exact(hex::decode(hex).expect("invalid hex")))
+        );
+        Ok(())
+    }
+    #[test]
+    fn hex_full_range() -> Result<()> {
+        let hex = "aaaa";
+        let hex_upper = "aaab";
+        let r = hex_range(hex);
+        assert_eq!(
+            r,
+            Some(HexSearch::Range(
+                hex::decode(hex).expect("invalid hex"),
+                hex::decode(hex_upper).expect("invalid hex")
+            ))
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn hex_full_range_odd() -> Result<()> {
+        let r = hex_range("abc");
+        assert_eq!(
+            r,
+            Some(HexSearch::Range(
+                hex::decode("abc0").expect("invalid hex"),
+                hex::decode("abd0").expect("invalid hex")
+            ))
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn hex_full_range_odd_end_f() -> Result<()> {
+        let r = hex_range("abf");
+        assert_eq!(
+            r,
+            Some(HexSearch::Range(
+                hex::decode("abf0").expect("invalid hex"),
+                hex::decode("ac00").expect("invalid hex")
+            ))
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn hex_no_upper() -> Result<()> {
+        let r = hex_range("ffff");
+        assert_eq!(
+            r,
+            Some(HexSearch::LowerOnly(
+                hex::decode("ffff").expect("invalid hex")
+            ))
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn hex_no_upper_odd() -> Result<()> {
+        let r = hex_range("fff");
+        assert_eq!(
+            r,
+            Some(HexSearch::LowerOnly(
+                hex::decode("fff0").expect("invalid hex")
+            ))
+        );
+        Ok(())
+    }
+}

src/info.rs

@@ -1,8 +1,9 @@
-use crate::config;
+//! Relay metadata using NIP-11
 /// Relay Info
+use crate::config;
 use serde::{Deserialize, Serialize};
 
-const CARGO_PKG_VERSION: Option<&'static str> = option_env!("CARGO_PKG_VERSION");
+pub const CARGO_PKG_VERSION: Option<&'static str> = option_env!("CARGO_PKG_VERSION");
 
 #[derive(Debug, Serialize, Deserialize)]
 #[allow(unused)]
@@ -34,7 +35,7 @@ impl From<config::Info> for RelayInfo {
             description: i.description,
             pubkey: i.pubkey,
             contact: i.contact,
-            supported_nips: Some(vec![1, 2, 11]),
+            supported_nips: Some(vec![1, 2, 11, 15, 16]),
             software: Some("https://git.sr.ht/~gheartsfield/nostr-rs-relay".to_owned()),
             version: CARGO_PKG_VERSION.map(|x| x.to_owned()),
         }

src/lib.rs

@@ -4,7 +4,9 @@ pub mod conn;
 pub mod db;
 pub mod error;
 pub mod event;
+pub mod hexrange;
 pub mod info;
-pub mod protostream;
+pub mod nip05;
+pub mod schema;
 pub mod subscription;
-pub mod tags;
+pub mod utils;

src/main.rs

@@ -9,27 +9,34 @@ use hyper::{
 };
 use log::*;
 use nostr_rs_relay::close::Close;
+use nostr_rs_relay::close::CloseCmd;
 use nostr_rs_relay::config;
 use nostr_rs_relay::conn;
 use nostr_rs_relay::db;
+use nostr_rs_relay::db::SubmittedEvent;
 use nostr_rs_relay::error::{Error, Result};
 use nostr_rs_relay::event::Event;
+use nostr_rs_relay::event::EventCmd;
 use nostr_rs_relay::info::RelayInfo;
-use nostr_rs_relay::protostream;
-use nostr_rs_relay::protostream::NostrMessage::*;
-use nostr_rs_relay::protostream::NostrResponse::*;
+use nostr_rs_relay::nip05;
+use nostr_rs_relay::subscription::Subscription;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
 use std::collections::HashMap;
 use std::convert::Infallible;
 use std::env;
 use std::net::SocketAddr;
 use std::path::Path;
+use std::time::Duration;
+use std::time::Instant;
 use tokio::runtime::Builder;
-use tokio::sync::broadcast;
-use tokio::sync::broadcast::{Receiver, Sender};
+use tokio::sync::broadcast::{self, Receiver, Sender};
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio_tungstenite::WebSocketStream;
+use tungstenite::error::Error as WsError;
 use tungstenite::handshake;
+use tungstenite::protocol::Message;
 use tungstenite::protocol::WebSocketConfig;
 
 /// Return a requested DB name from command line arguments.
@@ -46,7 +53,7 @@ async fn handle_web_request(
     pool: db::SqlitePool,
     remote_addr: SocketAddr,
     broadcast: Sender<Event>,
-    event_tx: tokio::sync::mpsc::Sender<Event>,
+    event_tx: tokio::sync::mpsc::Sender<SubmittedEvent>,
     shutdown: Receiver<()>,
 ) -> Result<Response<Body>, Infallible> {
     match (
@@ -55,7 +62,7 @@ async fn handle_web_request(
     ) {
         // Request for / as websocket
         ("/", true) => {
-            debug!("websocket with upgrade request");
+            trace!("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()
@@ -127,14 +134,17 @@ async fn handle_web_request(
                         return Ok(Response::builder()
                             .status(200)
                             .header("Content-Type", "application/nostr+json")
+                            .header("Access-Control-Allow-Origin", "*")
                             .body(b)
                             .unwrap());
                     }
                 }
             }
-            Ok(Response::new(Body::from(
-                "Please use a Nostr client to connect.",
-            )))
+            Ok(Response::builder()
+                .status(200)
+                .header("Content-Type", "text/plain")
+                .body(Body::from("Please use a Nostr client to connect."))
+                .unwrap())
         }
         (_, _) => {
             //handle any other url
@@ -171,21 +181,45 @@ fn main() -> Result<(), Error> {
         *settings = c;
     }
 
-    let config = config::SETTINGS.read().unwrap();
+    let settings = config::SETTINGS.read().unwrap();
+    trace!("Config: {:?}", settings);
     // do some config validation.
-    if !Path::new(&config.database.data_directory).is_dir() {
+    if !Path::new(&settings.database.data_directory).is_dir() {
         error!("Database directory does not exist");
         return Err(Error::DatabaseDirError);
     }
-    trace!("config: {:?}", config);
-    let addr = format!("{}:{}", config.network.address.trim(), config.network.port);
+    let addr = format!(
+        "{}:{}",
+        settings.network.address.trim(),
+        settings.network.port
+    );
     let socket_addr = addr.parse().expect("listening address not valid");
-    if let Some(addr_whitelist) = &config.authorization.pubkey_whitelist {
+    // address whitelisting settings
+    if let Some(addr_whitelist) = &settings.authorization.pubkey_whitelist {
         info!(
             "Event publishing restricted to {} pubkey(s)",
             addr_whitelist.len()
         );
     }
+    // check if NIP-05 enforced user verification is on
+    if settings.verified_users.is_active() {
+        info!(
+            "NIP-05 user verification mode:{:?}",
+            settings.verified_users.mode
+        );
+        if let Some(d) = settings.verified_users.verify_update_duration() {
+            info!("NIP-05 check user verification every:   {:?}", d);
+        }
+        if let Some(d) = settings.verified_users.verify_expiration_duration() {
+            info!("NIP-05 user verification expires after: {:?}", d);
+        }
+        if let Some(wl) = &settings.verified_users.domain_whitelist {
+            info!("NIP-05 domain whitelist: {:?}", wl);
+        }
+        if let Some(bl) = &settings.verified_users.domain_blacklist {
+            info!("NIP-05 domain blacklist: {:?}", bl);
+        }
+    }
     // configure tokio runtime
     let rt = Builder::new_multi_thread()
         .enable_all()
@@ -203,15 +237,42 @@ fn main() -> Result<(), Error> {
         let (bcast_tx, _) = broadcast::channel::<Event>(settings.limits.broadcast_buffer);
         // validated events that need to be persisted are sent to the
         // database on via this channel.
-        let (event_tx, event_rx) = mpsc::channel::<Event>(settings.limits.event_persist_buffer);
+        let (event_tx, event_rx) =
+            mpsc::channel::<SubmittedEvent>(settings.limits.event_persist_buffer);
         // establish a channel for letting all threads now about a
         // requested server shutdown.
-        let (invoke_shutdown, _) = broadcast::channel::<()>(1);
+        let (invoke_shutdown, shutdown_listen) = broadcast::channel::<()>(1);
+        // create a channel for sending any new metadata event.  These
+        // will get processed relatively slowly (a potentially
+        // multi-second blocking HTTP call) on a single thread, so we
+        // buffer requests on the channel.  No harm in dropping events
+        // here, since we are protecting against DoS.  This can make
+        // it difficult to setup initial metadata in bulk, since
+        // overwhelming this will drop events and won't register
+        // metadata events.
+        let (metadata_tx, metadata_rx) = broadcast::channel::<Event>(4096);
         // start the database writer thread.  Give it a channel for
         // 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;
+        db::db_writer(
+            event_rx,
+            bcast_tx.clone(),
+            metadata_tx.clone(),
+            shutdown_listen,
+        )
+        .await;
         info!("db writer created");
+
+        // create a nip-05 verifier thread
+        let verifier_opt = nip05::Verifier::new(metadata_rx, bcast_tx.clone());
+        if let Ok(mut v) = verifier_opt {
+            if settings.verified_users.is_active() {
+                tokio::task::spawn(async move {
+                    info!("starting up NIP-05 verifier...");
+                    v.run().await;
+                });
+            }
+        }
         // // listen for ctrl-c interruupts
         let ctrl_c_shutdown = invoke_shutdown.clone();
         tokio::spawn(async move {
@@ -220,7 +281,14 @@ fn main() -> Result<(), Error> {
             ctrl_c_shutdown.send(()).ok();
         });
         // build a connection pool for sqlite connections
-        let pool = db::build_read_pool();
+        let pool = db::build_pool(
+            "client query",
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY
+                | rusqlite::OpenFlags::SQLITE_OPEN_SHARED_CACHE,
+            settings.database.min_conn,
+            settings.database.max_conn,
+            true,
+        );
         // A `Service` is needed for every connection, so this
         // creates one from our `handle_request` function.
         let make_svc = make_service_fn(|conn: &AddrStream| {
@@ -255,49 +323,122 @@ fn main() -> Result<(), Error> {
     Ok(())
 }
 
+/// Nostr protocol messages from a client
+#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)]
+#[serde(untagged)]
+pub enum NostrMessage {
+    /// An `EVENT` message
+    EventMsg(EventCmd),
+    /// A `REQ` message
+    SubMsg(Subscription),
+    /// A `CLOSE` message
+    CloseMsg(CloseCmd),
+}
+
+/// Convert Message to NostrMessage
+fn convert_to_msg(msg: String) -> Result<NostrMessage> {
+    let config = config::SETTINGS.read().unwrap();
+    let parsed_res: Result<NostrMessage> = serde_json::from_str(&msg).map_err(|e| e.into());
+    match parsed_res {
+        Ok(m) => {
+            if let NostrMessage::EventMsg(_) = m {
+                if let Some(max_size) = config.limits.max_event_bytes {
+                    // check length, ensure that some max size is set.
+                    if msg.len() > max_size && max_size > 0 {
+                        return Err(Error::EventMaxLengthError(msg.len()));
+                    }
+                }
+            }
+            Ok(m)
+        }
+        Err(e) => {
+            debug!("proto parse error: {:?}", e);
+            debug!("parse error on message: {}", msg.trim());
+            Err(Error::ProtoParseError)
+        }
+    }
+}
+
+/// Turn a string into a NOTICE message ready to send over a WebSocket
+fn make_notice_message(msg: &str) -> Message {
+    Message::text(json!(["NOTICE", msg]).to_string())
+}
+
 /// Handle new client connections.  This runs through an event loop
 /// for all client communication.
 async fn nostr_server(
     pool: db::SqlitePool,
-    ws_stream: WebSocketStream<Upgraded>,
+    mut ws_stream: WebSocketStream<Upgraded>,
     broadcast: Sender<Event>,
-    event_tx: tokio::sync::mpsc::Sender<Event>,
+    event_tx: mpsc::Sender<SubmittedEvent>,
     mut shutdown: Receiver<()>,
 ) {
     // get a broadcast channel for clients to communicate on
     let mut bcast_rx = broadcast.subscribe();
-    // 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");
-    // 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
     let mut conn = conn::ClientConn::new();
     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.
     let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(256);
+    // Create channel for receiving NOTICEs
+    let (notice_tx, mut notice_rx) = mpsc::channel::<String>(32);
+
+    // 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(300);
+
+    // disconnect after 20 minutes without a ping response or event.
+    let max_quiet_time = Duration::from_secs(60 * 20);
+
+    let start = tokio::time::Instant::now() + default_ping_dur;
+    let mut ping_interval = tokio::time::interval_at(start, default_ping_dur);
+
     // maintain a hashmap of a oneshot channel for active subscriptions.
     // 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;
-    info!("new connection for client: {}", cid);
+    info!("new connection for client: {:?}", cid);
     loop {
         tokio::select! {
             _ = shutdown.recv() => {
                 // server shutting down, exit loop
                 break;
             },
+            _ = ping_interval.tick() => {
+                // check how long since we talked to client
+                // if it has been too long, disconnect
+                if last_message_time.elapsed() > max_quiet_time {
+                    debug!("ending connection due to lack of client ping response");
+                    break;
+                }
+                // Send a ping
+                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();
+            },
             Some(query_result) = query_rx.recv() => {
                 // database informed us of a query result we asked for
-                let res = EventRes(query_result.sub_id,query_result.event);
-                client_received_event_count += 1;
-                nostr_stream.send(res).await.ok();
+                let subesc = query_result.sub_id.replace("\"", "");
+                if query_result.event == "EOSE" {
+                    let send_str = format!("[\"EOSE\",\"{}\"]", subesc);
+                    ws_stream.send(Message::Text(send_str)).await.ok();
+                } else {
+                    client_received_event_count += 1;
+                    // send a result
+                    let send_str = format!("[\"EVENT\",\"{}\",{}]", subesc, &query_result.event);
+                    ws_stream.send(Message::Text(send_str)).await.ok();
+                }
             },
+            // TODO: consider logging the LaggedRecv error
             Ok(global_event) = bcast_rx.recv() => {
                 // an event has been broadcast to all clients
                 // first check if there is a subscription for this event.
@@ -306,39 +447,78 @@ async fn nostr_server(
                     // TODO: serialize at broadcast time, instead of
                     // once for each consumer.
                     if let Ok(event_str) = serde_json::to_string(&global_event) {
-                        debug!("sub match: client: {}, sub: {}, event: {}",
+                        debug!("sub match: client: {:?}, sub: {:?}, event: {:?}",
                                cid, s,
                                global_event.get_event_id_prefix());
                         // create an event response and send it
-                        let res = EventRes(s.to_owned(),event_str);
-                        nostr_stream.send(res).await.ok();
+                        let subesc = s.replace("\"", "");
+                        ws_stream.send(Message::Text(format!("[\"EVENT\",\"{}\",{}]", subesc, event_str))).await.ok();
+                        //nostr_stream.send(res).await.ok();
                     } else {
-                        warn!("could not convert event to string");
+                        warn!("could not serialize event {:?}", global_event.get_event_id_prefix());
                     }
                 }
             },
-            // check if this client has a subscription
-            proto_next = nostr_stream.next() => {
-                match proto_next {
-                    Some(Ok(EventMsg(ec))) => {
+            ws_next = ws_stream.next() => {
+                // update most recent message time for client
+                last_message_time = Instant::now();
+                // Consume text messages from the client, parse into Nostr messages.
+                let nostr_msg = match ws_next {
+                    Some(Ok(Message::Text(m))) => {
+                        convert_to_msg(m)
+                    },
+                    Some(Ok(Message::Binary(_))) => {
+                        ws_stream.send(make_notice_message("binary messages are not accepted")).await.ok();
+                        continue;
+                    },
+                    Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {
+                        // get a ping/pong, ignore.  tungstenite will
+                        // send responses automatically.
+                        continue;
+                    },
+                    None |
+                    Some(Ok(Message::Close(_))) |
+                    Some(Err(WsError::AlreadyClosed)) |
+                    Some(Err(WsError::ConnectionClosed)) |
+                    Some(Err(WsError::Protocol(tungstenite::error::ProtocolError::ResetWithoutClosingHandshake)))
+                        => {
+                        debug!("websocket close from client: {:?}",cid);
+                        break;
+                    },
+                    Some(Err(WsError::Io(e))) => {
+                        // IO errors are considered fatal
+                        warn!("IO error (client: {:?}): {:?}", cid, e);
+                        break;
+                    }
+                    x => {
+                        // default condition on error is to close the client connection
+                        info!("unknown error (client: {:?}): {:?} (closing conn)", cid, x);
+                        break;
+                    }
+                };
+
+                // convert ws_next into proto_next
+                match nostr_msg {
+                    Ok(NostrMessage::EventMsg(ec)) => {
                         // An EventCmd needs to be validated to be converted into an Event
                         // handle each type of message
                         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: {} from client: {}", id_prefix, cid);
-                                // Write this to the database
-                                event_tx.send(e.clone()).await.ok();
+                                debug!("successfully parsed/validated event: {:?} from client: {:?}", id_prefix, cid);
+                                // Write this to the database.
+                                let submit_event = SubmittedEvent { event: e.clone(), notice_tx: notice_tx.clone() };
+                                event_tx.send(submit_event).await.ok();
                                 client_published_event_count += 1;
                             },
                             Err(_) => {
-                                info!("client {} sent an invalid event", cid);
-                                nostr_stream.send(NoticeRes("event was invalid".to_owned())).await.ok();
+                                info!("client {:?} sent an invalid event", cid);
+                                ws_stream.send(make_notice_message("event was invalid")).await.ok();
                             }
                         }
                     },
-                    Some(Ok(SubMsg(s))) => {
+                    Ok(NostrMessage::SubMsg(s)) => {
                         debug!("client {} requesting a subscription", cid);
                         // subscription handling consists of:
                         // * registering the subscription so future events can be matched
@@ -352,18 +532,15 @@ async fn nostr_server(
                                     previous_query.send(()).ok();
                                 }
                                 // start a database query
-                                // show pool stats
-                                debug!("DB pool stats: {:?}", pool.state());
-                                db::db_query(s, pool.get().expect("could not get connection"), query_tx.clone(), abandon_query_rx).await;
+                                db::db_query(s, cid.to_owned(), pool.clone(), query_tx.clone(), abandon_query_rx).await;
                             },
                             Err(e) => {
                                 info!("Subscription error: {}", e);
-                                nostr_stream.send(NoticeRes(format!("{}",e))).await.ok();
-
+                                ws_stream.send(make_notice_message(&e.to_string())).await.ok();
                             }
                         }
                     },
-                    Some(Ok(CloseMsg(cc))) => {
+                    Ok(NostrMessage::CloseMsg(cc)) => {
                         // closing a request simply removes the subscription.
                         let parsed : Result<Close> = Result::<Close>::from(cc);
                         match parsed {
@@ -380,24 +557,24 @@ async fn nostr_server(
                             },
                             Err(_) => {
                                 info!("invalid command ignored");
-
+                                ws_stream.send(make_notice_message("could not parse command")).await.ok();
                             }
                         }
                     },
-                    None => {
-                        debug!("normal websocket close from client: {}",cid);
-                        break;
-                    },
-                    Some(Err(Error::ConnError)) => {
-                        debug!("got connection close/error, disconnecting client: {}",cid);
+                    Err(Error::ConnError) => {
+                        debug!("got connection close/error, disconnecting client: {:?}",cid);
                         break;
                     }
-                    Some(Err(Error::EventMaxLengthError(s))) => {
-                        info!("client {} sent event larger ({} bytes) than max size", cid, s);
-                        nostr_stream.send(NoticeRes("event exceeded max size".to_owned())).await.ok();
+                    Err(Error::EventMaxLengthError(s)) => {
+                        info!("client {:?} sent event larger ({} bytes) than max size", cid, s);
+                        ws_stream.send(make_notice_message("event exceeded max size")).await.ok();
                     },
-                    Some(Err(e)) => {
-                        info!("got non-fatal error from client: {}, error: {:?}", cid, e);
+                    Err(Error::ProtoParseError) => {
+                        info!("client {:?} sent event that could not be parsed", cid);
+                        ws_stream.send(make_notice_message("could not parse command")).await.ok();
+                    },
+                    Err(e) => {
+                        info!("got non-fatal error from client: {:?}, error: {:?}", cid, e);
                     },
                 }
             },
@@ -408,7 +585,7 @@ async fn nostr_server(
         stop_tx.send(()).ok();
     }
     info!(
-        "stopping connection for client: {} (client sent {} event(s), received {})",
+        "stopping connection for client: {:?} (client sent {} event(s), received {})",
         cid, client_published_event_count, client_received_event_count
     );
 }

src/nip05.rs

@@ -0,0 +1,818 @@
+//! User verification using NIP-05 names
+//!
+//! NIP-05 defines a mechanism for authors to associate an internet
+//! address with their public key, in metadata events.  This module
+//! consumes a stream of metadata events, and keeps a database table
+//! updated with the current NIP-05 verification status.
+use crate::config::SETTINGS;
+use crate::db;
+use crate::error::{Error, Result};
+use crate::event::Event;
+use crate::utils::unix_time;
+use hyper::body::HttpBody;
+use hyper::client::connect::HttpConnector;
+use hyper::Client;
+use hyper_tls::HttpsConnector;
+use log::*;
+use rand::Rng;
+use rusqlite::params;
+use std::time::Duration;
+use std::time::Instant;
+use std::time::SystemTime;
+use tokio::time::Interval;
+
+/// NIP-05 verifier state
+pub struct Verifier {
+    /// Metadata events for us to inspect
+    metadata_rx: tokio::sync::broadcast::Receiver<Event>,
+    /// Newly validated events get written and then broadcast on this channel to subscribers
+    event_tx: tokio::sync::broadcast::Sender<Event>,
+    /// SQLite read query pool
+    read_pool: db::SqlitePool,
+    /// SQLite write query pool
+    write_pool: db::SqlitePool,
+    /// HTTP client
+    client: hyper::Client<HttpsConnector<HttpConnector>, hyper::Body>,
+    /// After all accounts are updated, wait this long before checking again.
+    wait_after_finish: Duration,
+    /// Minimum amount of time between HTTP queries
+    http_wait_duration: Duration,
+    /// Interval for updating verification records
+    reverify_interval: Interval,
+}
+
+/// A NIP-05 identifier is a local part and domain.
+#[derive(PartialEq, Debug, Clone)]
+pub struct Nip05Name {
+    local: String,
+    domain: String,
+}
+
+impl Nip05Name {
+    /// Does this name represent the entire domain?
+    pub fn is_domain_only(&self) -> bool {
+        self.local == "_"
+    }
+
+    /// Determine the URL to query for verification
+    fn to_url(&self) -> Option<http::Uri> {
+        format!(
+            "https://{}/.well-known/nostr.json?name={}",
+            self.domain, self.local
+        )
+        .parse::<http::Uri>()
+        .ok()
+    }
+}
+
+// Parsing Nip05Names from strings
+impl std::convert::TryFrom<&str> for Nip05Name {
+    type Error = Error;
+    fn try_from(inet: &str) -> Result<Self, Self::Error> {
+        // break full name at the @ boundary.
+        let components: Vec<&str> = inet.split('@').collect();
+        if components.len() != 2 {
+            Err(Error::CustomError("too many/few components".to_owned()))
+        } else {
+            // check if local name is valid
+            let local = components[0];
+            let domain = components[1];
+            if local
+                .chars()
+                .all(|x| x.is_alphanumeric() || x == '_' || x == '-' || x == '.')
+            {
+                if domain
+                    .chars()
+                    .all(|x| x.is_alphanumeric() || x == '-' || x == '.')
+                {
+                    Ok(Nip05Name {
+                        local: local.to_owned(),
+                        domain: domain.to_owned(),
+                    })
+                } else {
+                    Err(Error::CustomError(
+                        "invalid character in domain part".to_owned(),
+                    ))
+                }
+            } else {
+                Err(Error::CustomError(
+                    "invalid character in local part".to_owned(),
+                ))
+            }
+        }
+    }
+}
+
+impl std::fmt::Display for Nip05Name {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}@{}", self.local, self.domain)
+    }
+}
+
+// Current time, with a slight foward jitter in seconds
+fn now_jitter(sec: u64) -> u64 {
+    // random time between now, and 10min in future.
+    let mut rng = rand::thread_rng();
+    let jitter_amount = rng.gen_range(0..sec);
+    let now = unix_time();
+    now.saturating_add(jitter_amount)
+}
+
+/// Check if the specified username and address are present and match in this response body
+fn body_contains_user(username: &str, address: &str, bytes: hyper::body::Bytes) -> Result<bool> {
+    // convert the body into json
+    let body: serde_json::Value = serde_json::from_slice(&bytes)?;
+    // ensure we have a names object.
+    let names_map = body
+        .as_object()
+        .and_then(|x| x.get("names"))
+        .and_then(|x| x.as_object())
+        .ok_or_else(|| Error::CustomError("not a map".to_owned()))?;
+    // get the pubkey for the requested user
+    let check_name = names_map.get(username).and_then(|x| x.as_str());
+    // ensure the address is a match
+    Ok(check_name.map(|x| x == address).unwrap_or(false))
+}
+
+impl Verifier {
+    pub fn new(
+        metadata_rx: tokio::sync::broadcast::Receiver<Event>,
+        event_tx: tokio::sync::broadcast::Sender<Event>,
+    ) -> Result<Self> {
+        info!("creating NIP-05 verifier");
+        // build a database connection for reading and writing.
+        let write_pool = db::build_pool(
+            "nip05 writer",
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE,
+            1,    // min conns
+            4,    // max conns
+            true, // wait for DB
+        );
+        let read_pool = db::build_pool(
+            "nip05 reader",
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
+            1,    // min conns
+            8,    // max conns
+            true, // wait for DB
+        );
+        // setup hyper client
+        let https = HttpsConnector::new();
+        let client = Client::builder().build::<_, hyper::Body>(https);
+
+        // After all accounts have been re-verified, don't check again
+        // for this long.
+        let wait_after_finish = Duration::from_secs(60 * 10);
+        // when we have an active queue of accounts to validate, we
+        // will wait this duration between HTTP requests.
+        let http_wait_duration = Duration::from_secs(1);
+        // setup initial interval for re-verification.  If we find
+        // there is no work to be done, it will be reset to a longer
+        // duration.
+        let reverify_interval = tokio::time::interval(http_wait_duration);
+        Ok(Verifier {
+            metadata_rx,
+            event_tx,
+            read_pool,
+            write_pool,
+            client,
+            wait_after_finish,
+            http_wait_duration,
+            reverify_interval,
+        })
+    }
+
+    /// Perform web verification against a NIP-05 name and address.
+    pub async fn get_web_verification(
+        &mut self,
+        nip: &Nip05Name,
+        pubkey: &str,
+    ) -> UserWebVerificationStatus {
+        self.get_web_verification_res(nip, pubkey)
+            .await
+            .unwrap_or(UserWebVerificationStatus::Unknown)
+    }
+
+    /// Perform web verification against an `Event` (must be metadata).
+    pub async fn get_web_verification_from_event(
+        &mut self,
+        e: &Event,
+    ) -> UserWebVerificationStatus {
+        let nip_parse = e.get_nip05_addr();
+        if let Some(nip) = nip_parse {
+            self.get_web_verification_res(&nip, &e.pubkey)
+                .await
+                .unwrap_or(UserWebVerificationStatus::Unknown)
+        } else {
+            UserWebVerificationStatus::Unknown
+        }
+    }
+
+    /// Perform web verification, with a `Result` return.
+    async fn get_web_verification_res(
+        &mut self,
+        nip: &Nip05Name,
+        pubkey: &str,
+    ) -> Result<UserWebVerificationStatus> {
+        // determine if this domain should be checked
+        if !is_domain_allowed(&nip.domain) {
+            return Ok(UserWebVerificationStatus::DomainNotAllowed);
+        }
+        let url = nip
+            .to_url()
+            .ok_or_else(|| Error::CustomError("invalid NIP-05 URL".to_owned()))?;
+        let req = hyper::Request::builder()
+            .method(hyper::Method::GET)
+            .uri(url)
+            .header("Accept", "application/json")
+            .header(
+                "User-Agent",
+                format!(
+                    "nostr-rs-relay/{} NIP-05 Verifier",
+                    crate::info::CARGO_PKG_VERSION.unwrap()
+                ),
+            )
+            .body(hyper::Body::empty())
+            .expect("request builder");
+
+        let response_fut = self.client.request(req);
+
+        // HTTP request with timeout
+        match tokio::time::timeout(Duration::from_secs(5), response_fut).await {
+            Ok(response_res) => {
+                let response = response_res?;
+                // limit size of verification document to 1MB.
+                const MAX_ALLOWED_RESPONSE_SIZE: u64 = 1024 * 1024;
+                // determine content length from response
+                let response_content_length = match response.body().size_hint().upper() {
+                    Some(v) => v,
+                    None => MAX_ALLOWED_RESPONSE_SIZE + 1, // reject missing content length
+                };
+                // TODO: test how hyper handles the client providing an inaccurate content-length.
+                if response_content_length <= MAX_ALLOWED_RESPONSE_SIZE {
+                    let (parts, body) = response.into_parts();
+                    // TODO: consider redirects
+                    if parts.status == http::StatusCode::OK {
+                        // parse body, determine if the username / key / address is present
+                        let body_bytes = hyper::body::to_bytes(body).await?;
+                        let body_matches = body_contains_user(&nip.local, pubkey, body_bytes)?;
+                        if body_matches {
+                            return Ok(UserWebVerificationStatus::Verified);
+                        } else {
+                            // successful response, parsed as a nip-05
+                            // document, but this name/pubkey was not
+                            // present.
+                            return Ok(UserWebVerificationStatus::Unverified);
+                        }
+                    }
+                } else {
+                    info!(
+                        "content length missing or exceeded limits for account: {:?}",
+                        nip.to_string()
+                    );
+                }
+            }
+            Err(_) => {
+                info!("timeout verifying account {:?}", nip);
+                return Ok(UserWebVerificationStatus::Unknown);
+            }
+        }
+        Ok(UserWebVerificationStatus::Unknown)
+    }
+
+    /// Perform NIP-05 verifier tasks.
+    pub async fn run(&mut self) {
+        // use this to schedule periodic re-validation tasks
+        // run a loop, restarting on failure
+        loop {
+            let res = self.run_internal().await;
+            if let Err(e) = res {
+                info!("error in verifier: {:?}", e);
+            }
+        }
+    }
+
+    /// Internal select loop for performing verification
+    async fn run_internal(&mut self) -> Result<()> {
+        tokio::select! {
+            m = self.metadata_rx.recv() => {
+                match m {
+                    Ok(e) => {
+                        if let Some(naddr) = e.get_nip05_addr() {
+                            info!("got metadata event for ({:?},{:?})", naddr.to_string() ,e.get_author_prefix());
+                            // Process a new author, checking if they are verified:
+                            let check_verified = get_latest_user_verification(self.read_pool.get().expect("could not get connection"), &e.pubkey).await;
+                            // ensure the event we got is more recent than the one we have, otherwise we can ignore it.
+                            if let Ok(last_check) = check_verified {
+                                if e.created_at <= last_check.event_created {
+                                    // this metadata is from the same author as an existing verification.
+                                    // it is older than what we have, so we can ignore it.
+                                    debug!("received older metadata event for author {:?}", e.get_author_prefix());
+                                    return Ok(());
+                                }
+                            }
+                            // old, or no existing record for this user.  In either case, we just create a new one.
+                            let start = Instant::now();
+                            let v = self.get_web_verification_from_event(&e).await;
+                            info!(
+                                "checked name {:?}, result: {:?}, in: {:?}",
+                                naddr.to_string(),
+                                v,
+                                start.elapsed()
+                            );
+                            // sleep to limit how frequently we make HTTP requests for new metadata events.  This should limit us to 4 req/sec.
+                            tokio::time::sleep(Duration::from_millis(250)).await;
+                            // if this user was verified, we need to write the
+                            // record, persist the event, and broadcast.
+                            if let UserWebVerificationStatus::Verified = v {
+                                self.create_new_verified_user(&naddr.to_string(), &e).await?;
+                            }
+                        }
+                    },
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(c)) => {
+                        warn!("incoming metadata events overwhelmed buffer, {} events dropped",c);
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                        info!("metadata broadcast channel closed");
+                    }
+                }
+            },
+            _ = self.reverify_interval.tick() => {
+                // check and see if there is an old account that needs
+                // to be reverified
+                self.do_reverify().await?;
+            },
+        }
+        Ok(())
+    }
+
+    /// Reverify the oldest user verification record.
+    async fn do_reverify(&mut self) -> Result<()> {
+        let reverify_setting;
+        let max_failures;
+        {
+            // this block prevents a read handle to settings being
+            // captured by the async DB call (guard is not Send)
+            let settings = SETTINGS.read().unwrap();
+            reverify_setting = settings.verified_users.verify_update_frequency_duration;
+            max_failures = settings.verified_users.max_consecutive_failures;
+        }
+        // get from settings, but default to 6hrs between re-checking an account
+        let reverify_dur = reverify_setting.unwrap_or_else(|| Duration::from_secs(60 * 60 * 6));
+        // find all verification records that have success or failure OLDER than the reverify_dur.
+        let now = SystemTime::now();
+        let earliest = now - reverify_dur;
+        let earliest_epoch = earliest
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .map(|x| x.as_secs())
+            .unwrap_or(0);
+        let vr = get_oldest_user_verification(self.read_pool.get()?, earliest_epoch).await;
+        match vr {
+            Ok(ref v) => {
+                let new_status = self.get_web_verification(&v.name, &v.address).await;
+                match new_status {
+                    UserWebVerificationStatus::Verified => {
+                        // freshly verified account, update the
+                        // timestamp.
+                        self.update_verification_record(self.write_pool.get()?, v)
+                            .await?;
+                    }
+                    UserWebVerificationStatus::DomainNotAllowed
+                    | UserWebVerificationStatus::Unknown => {
+                        // server may be offline, or temporarily
+                        // blocked by the config file.  Note the
+                        // failure so we can process something
+                        // else.
+
+                        // have we had enough failures to give up?
+                        if v.failure_count >= max_failures as u64 {
+                            info!(
+                                "giving up on verifying {:?} after {} failures",
+                                v.name, v.failure_count
+                            );
+                            self.delete_verification_record(self.write_pool.get()?, v)
+                                .await?;
+                        } else {
+                            // record normal failure, incrementing failure count
+                            self.fail_verification_record(self.write_pool.get()?, v)
+                                .await?;
+                        }
+                    }
+                    UserWebVerificationStatus::Unverified => {
+                        // domain has removed the verification, drop
+                        // the record on our side.
+                        self.delete_verification_record(self.write_pool.get()?, v)
+                            .await?;
+                    }
+                }
+            }
+            Err(Error::SqlError(rusqlite::Error::QueryReturnedNoRows)) => {
+                // No users need verification.  Reset the interval to
+                // the next verification attempt.
+                let start = tokio::time::Instant::now() + self.wait_after_finish;
+                self.reverify_interval = tokio::time::interval_at(start, self.http_wait_duration);
+            }
+            Err(ref e) => {
+                warn!(
+                    "Error when checking for NIP-05 verification records: {:?}",
+                    e
+                );
+            }
+        }
+        Ok(())
+    }
+
+    /// Reset the verification timestamp on a VerificationRecord
+    pub async fn update_verification_record(
+        &mut self,
+        mut conn: db::PooledConnection,
+        vr: &VerificationRecord,
+    ) -> Result<()> {
+        let vr_id = vr.rowid;
+        let vr_str = vr.to_string();
+        tokio::task::spawn_blocking(move || {
+            // add some jitter to the verification to prevent everything from stacking up together.
+            let verif_time = now_jitter(600);
+            let tx = conn.transaction()?;
+            {
+                // update verification time and reset any failure count
+                let query =
+                    "UPDATE user_verification SET verified_at=?, failure_count=0 WHERE id=?";
+                let mut stmt = tx.prepare(query)?;
+                stmt.execute(params![verif_time, vr_id])?;
+            }
+            tx.commit()?;
+            info!("verification updated for {}", vr_str);
+            let ok: Result<()> = Ok(());
+            ok
+        })
+        .await?
+    }
+    /// Reset the failure timestamp on a VerificationRecord
+    pub async fn fail_verification_record(
+        &mut self,
+        mut conn: db::PooledConnection,
+        vr: &VerificationRecord,
+    ) -> Result<()> {
+        let vr_id = vr.rowid;
+        let vr_str = vr.to_string();
+        let fail_count = vr.failure_count.saturating_add(1);
+        tokio::task::spawn_blocking(move || {
+            // add some jitter to the verification to prevent everything from stacking up together.
+            let fail_time = now_jitter(600);
+            let tx = conn.transaction()?;
+            {
+                let query = "UPDATE user_verification SET failed_at=?, failure_count=? WHERE id=?";
+                let mut stmt = tx.prepare(query)?;
+                stmt.execute(params![fail_time, fail_count, vr_id])?;
+            }
+            tx.commit()?;
+            info!("verification failed for {}", vr_str);
+            let ok: Result<()> = Ok(());
+            ok
+        })
+        .await?
+    }
+    /// Delete a VerificationRecord that is no longer valid
+    pub async fn delete_verification_record(
+        &mut self,
+        mut conn: db::PooledConnection,
+        vr: &VerificationRecord,
+    ) -> Result<()> {
+        let vr_id = vr.rowid;
+        let vr_str = vr.to_string();
+        tokio::task::spawn_blocking(move || {
+            let tx = conn.transaction()?;
+            {
+                let query = "DELETE FROM user_verification WHERE id=?;";
+                let mut stmt = tx.prepare(query)?;
+                stmt.execute(params![vr_id])?;
+            }
+            tx.commit()?;
+            info!("verification rescinded for {}", vr_str);
+            let ok: Result<()> = Ok(());
+            ok
+        })
+        .await?
+    }
+
+    /// Persist an event, create a verification record, and broadcast.
+    // TODO: have more event-writing logic handled in the db module.
+    // Right now, these events avoid the rate limit.  That is
+    // acceptable since as soon as the user is registered, this path
+    // is no longer used.
+    // TODO: refactor these into spawn_blocking
+    // calls to get them off the async executors.
+    async fn create_new_verified_user(&mut self, name: &str, event: &Event) -> Result<()> {
+        let start = Instant::now();
+        // we should only do this if we are enabled.  if we are
+        // disabled/passive, the event has already been persisted.
+        let should_write_event;
+        {
+            let settings = SETTINGS.read().unwrap();
+            should_write_event = settings.verified_users.is_enabled()
+        }
+        if should_write_event {
+            match db::write_event(&mut self.write_pool.get()?, event) {
+                Ok(updated) => {
+                    if updated != 0 {
+                        info!(
+                            "persisted event: {:?} in {:?}",
+                            event.get_event_id_prefix(),
+                            start.elapsed()
+                        );
+                        self.event_tx.send(event.clone()).ok();
+                    }
+                }
+                Err(err) => {
+                    warn!("event insert failed: {:?}", err);
+                    if let Error::SqlError(r) = err {
+                        warn!("because: : {:?}", r);
+                    }
+                }
+            }
+        }
+        // write the verification record
+        save_verification_record(self.write_pool.get()?, event, name).await?;
+        Ok(())
+    }
+}
+
+/// Result of checking user's verification status against DNS/HTTP.
+#[derive(PartialEq, Debug, Clone)]
+pub enum UserWebVerificationStatus {
+    Verified,         // user is verified, as of now.
+    DomainNotAllowed, // domain blacklist or whitelist denied us from attempting a verification
+    Unknown,          // user's status could not be determined (timeout, server error)
+    Unverified,       // user's status is not verified (successful check, name / addr do not match)
+}
+
+/// A NIP-05 verification record.
+#[derive(PartialEq, Debug, Clone)]
+// Basic information for a verification event.  Gives us all we need to assert a NIP-05 address is good.
+pub struct VerificationRecord {
+    pub rowid: u64,                // database row for this verification event
+    pub name: Nip05Name,           // address being verified
+    pub address: String,           // pubkey
+    pub event: String,             // event ID hash providing the verification
+    pub event_created: u64,        // when the metadata event was published
+    pub last_success: Option<u64>, // the most recent time a verification was provided.  None if verification under this name has never succeeded.
+    pub last_failure: Option<u64>, // the most recent time verification was attempted, but could not be completed.
+    pub failure_count: u64,        // how many consecutive failures have been observed.
+}
+
+/// Check with settings to determine if a given domain is allowed to
+/// publish.
+pub fn is_domain_allowed(domain: &str) -> bool {
+    let settings = SETTINGS.read().unwrap();
+    // if there is a whitelist, domain must be present in it.
+    if let Some(wl) = &settings.verified_users.domain_whitelist {
+        // workaround for Vec contains not accepting &str
+        return wl.iter().any(|x| x == domain);
+    }
+    // otherwise, check that user is not in the blacklist
+    if let Some(bl) = &settings.verified_users.domain_blacklist {
+        return !bl.iter().any(|x| x == domain);
+    }
+    true
+}
+
+impl VerificationRecord {
+    /// Check if the record is recent enough to be considered valid,
+    /// and the domain is allowed.
+    pub fn is_valid(&self) -> bool {
+        let settings = SETTINGS.read().unwrap();
+        // how long a verification record is good for
+        let nip05_expiration = &settings.verified_users.verify_expiration_duration;
+        if let Some(e) = nip05_expiration {
+            if !self.is_current(e) {
+                return false;
+            }
+        }
+        // check domains
+        is_domain_allowed(&self.name.domain)
+    }
+
+    /// Check if this record has been validated since the given
+    /// duration.
+    fn is_current(&self, d: &Duration) -> bool {
+        match self.last_success {
+            Some(s) => {
+                // current time - duration
+                let now = SystemTime::now();
+                let cutoff = now - *d;
+                let cutoff_epoch = cutoff
+                    .duration_since(SystemTime::UNIX_EPOCH)
+                    .map(|x| x.as_secs())
+                    .unwrap_or(0);
+                s > cutoff_epoch
+            }
+            None => false,
+        }
+    }
+}
+
+impl std::fmt::Display for VerificationRecord {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "({:?},{:?})",
+            self.name.to_string(),
+            self.address.chars().take(8).collect::<String>()
+        )
+    }
+}
+
+/// Create a new verification record based on an event
+pub async fn save_verification_record(
+    mut conn: db::PooledConnection,
+    event: &Event,
+    name: &str,
+) -> Result<()> {
+    let e = hex::decode(&event.id).ok();
+    let n = name.to_owned();
+    let a_prefix = event.get_author_prefix();
+    tokio::task::spawn_blocking(move || {
+        let tx = conn.transaction()?;
+        {
+            // if we create a /new/ one, we should get rid of any old ones.  or group the new ones by name and only consider the latest.
+            let query = "INSERT INTO user_verification (metadata_event, name, verified_at) VALUES ((SELECT id from event WHERE event_hash=?), ?, strftime('%s','now'));";
+            let mut stmt = tx.prepare(query)?;
+            stmt.execute(params![e, n])?;
+            // get the row ID
+            let v_id = tx.last_insert_rowid();
+            // delete everything else by this name
+            let del_query = "DELETE FROM user_verification WHERE name = ? AND id != ?;";
+            let mut del_stmt = tx.prepare(del_query)?;
+            let count = del_stmt.execute(params![n,v_id])?;
+            if count > 0 {
+                info!("removed {} old verification records for ({:?},{:?})", count, n, a_prefix);
+            }
+        }
+        tx.commit()?;
+        info!("saved new verification record for ({:?},{:?})", n, a_prefix);
+        let ok: Result<()> = Ok(());
+        ok
+    }).await?
+}
+
+/// Retrieve the most recent verification record for a given pubkey (async).
+pub async fn get_latest_user_verification(
+    conn: db::PooledConnection,
+    pubkey: &str,
+) -> Result<VerificationRecord> {
+    let p = pubkey.to_owned();
+    tokio::task::spawn_blocking(move || query_latest_user_verification(conn, p)).await?
+}
+
+/// Query database for the latest verification record for a given pubkey.
+pub fn query_latest_user_verification(
+    mut conn: db::PooledConnection,
+    pubkey: String,
+) -> Result<VerificationRecord> {
+    let tx = conn.transaction()?;
+    let query = "SELECT v.id, v.name, e.event_hash, 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 e.author=? ORDER BY e.created_at DESC, v.verified_at DESC, v.failed_at DESC LIMIT 1;";
+    let mut stmt = tx.prepare_cached(query)?;
+    let fields = stmt.query_row(params![hex::decode(&pubkey).ok()], |r| {
+        let rowid: u64 = r.get(0)?;
+        let rowname: String = r.get(1)?;
+        let eventid: Vec<u8> = r.get(2)?;
+        let created_at: u64 = r.get(3)?;
+        // create a tuple since we can't throw non-rusqlite errors in this closure
+        Ok((
+            rowid,
+            rowname,
+            eventid,
+            created_at,
+            r.get(4).ok(),
+            r.get(5).ok(),
+            r.get(6)?,
+        ))
+    })?;
+    Ok(VerificationRecord {
+        rowid: fields.0,
+        name: Nip05Name::try_from(&fields.1[..])?,
+        address: pubkey,
+        event: hex::encode(fields.2),
+        event_created: fields.3,
+        last_success: fields.4,
+        last_failure: fields.5,
+        failure_count: fields.6,
+    })
+}
+
+/// Retrieve the oldest user verification (async)
+pub async fn get_oldest_user_verification(
+    conn: db::PooledConnection,
+    earliest: u64,
+) -> Result<VerificationRecord> {
+    let res =
+        tokio::task::spawn_blocking(move || query_oldest_user_verification(conn, earliest)).await?;
+    res
+}
+
+pub fn query_oldest_user_verification(
+    mut conn: db::PooledConnection,
+    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 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)?;
+        let rowname: String = r.get(1)?;
+        let eventid: Vec<u8> = r.get(2)?;
+        let pubkey: Vec<u8> = r.get(3)?;
+        let created_at: u64 = r.get(4)?;
+        // create a tuple since we can't throw non-rusqlite errors in this closure
+        Ok((
+            rowid,
+            rowname,
+            eventid,
+            pubkey,
+            created_at,
+            r.get(5).ok(),
+            r.get(6).ok(),
+            r.get(7)?,
+        ))
+    })?;
+    let vr = VerificationRecord {
+        rowid: fields.0,
+        name: Nip05Name::try_from(&fields.1[..])?,
+        address: hex::encode(fields.3),
+        event: hex::encode(fields.2),
+        event_created: fields.4,
+        last_success: fields.5,
+        last_failure: fields.6,
+        failure_count: fields.7,
+    };
+    Ok(vr)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn local_from_inet() {
+        let addr = "bob@example.com";
+        let parsed = Nip05Name::try_from(addr);
+        assert!(!parsed.is_err());
+        let v = parsed.unwrap();
+        assert_eq!(v.local, "bob");
+        assert_eq!(v.domain, "example.com");
+    }
+
+    #[test]
+    fn not_enough_sep() {
+        let addr = "bob_example.com";
+        let parsed = Nip05Name::try_from(addr);
+        assert!(parsed.is_err());
+    }
+
+    #[test]
+    fn too_many_sep() {
+        let addr = "foo@bob@example.com";
+        let parsed = Nip05Name::try_from(addr);
+        assert!(parsed.is_err());
+    }
+
+    #[test]
+    fn invalid_local_name() {
+        // non-permitted ascii chars
+        assert!(Nip05Name::try_from("foo!@example.com").is_err());
+        assert!(Nip05Name::try_from("foo @example.com").is_err());
+        assert!(Nip05Name::try_from(" foo@example.com").is_err());
+        assert!(Nip05Name::try_from("f oo@example.com").is_err());
+        assert!(Nip05Name::try_from("foo<@example.com").is_err());
+        // unicode dash
+        assert!(Nip05Name::try_from("foo‐bar@example.com").is_err());
+        // emoji
+        assert!(Nip05Name::try_from("foo😭bar@example.com").is_err());
+    }
+    #[test]
+    fn invalid_domain_name() {
+        // non-permitted ascii chars
+        assert!(Nip05Name::try_from("foo@examp!e.com").is_err());
+        assert!(Nip05Name::try_from("foo@ example.com").is_err());
+        assert!(Nip05Name::try_from("foo@exa mple.com").is_err());
+        assert!(Nip05Name::try_from("foo@example .com").is_err());
+        assert!(Nip05Name::try_from("foo@exa<mple.com").is_err());
+        // unicode dash
+        assert!(Nip05Name::try_from("foobar@exa‐mple.com").is_err());
+        // emoji
+        assert!(Nip05Name::try_from("foobar@ex😭ample.com").is_err());
+    }
+
+    #[test]
+    fn to_url() {
+        let nip = Nip05Name::try_from("foobar@example.com").unwrap();
+        assert_eq!(
+            nip.to_url(),
+            Some(
+                "https://example.com/.well-known/nostr.json?name=foobar"
+                    .parse()
+                    .unwrap()
+            )
+        );
+    }
+}

src/protostream.rs

@@ -1,139 +0,0 @@
-//! Nostr protocol layered over WebSocket
-use crate::close::CloseCmd;
-use crate::config;
-use crate::error::{Error, Result};
-use crate::event::EventCmd;
-use crate::subscription::Subscription;
-use core::pin::Pin;
-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_tungstenite::WebSocketStream;
-use tungstenite::error::Error as WsError;
-use tungstenite::protocol::Message;
-
-/// Nostr protocol messages from a client
-#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)]
-#[serde(untagged)]
-pub enum NostrMessage {
-    /// An `EVENT` message
-    EventMsg(EventCmd),
-    /// A `REQ` message
-    SubMsg(Subscription),
-    /// A `CLOSE` message
-    CloseMsg(CloseCmd),
-}
-
-/// Nostr protocol messages from a relay/server
-#[derive(Deserialize, Serialize, Clone, PartialEq, Debug)]
-pub enum NostrResponse {
-    /// A `NOTICE` response
-    NoticeRes(String),
-    /// An `EVENT` response, composed of the subscription identifier,
-    /// and serialized event JSON
-    EventRes(String, String),
-}
-
-/// A Nostr protocol stream is layered on top of a Websocket stream.
-pub struct NostrStream {
-    ws_stream: WebSocketStream<Upgraded>,
-}
-
-/// Given a websocket, return a protocol stream wrapper.
-pub fn wrap_ws_in_nostr(ws: WebSocketStream<Upgraded>) -> NostrStream {
-    NostrStream { ws_stream: ws }
-}
-
-/// Implement the [`Stream`] interface to produce Nostr messages.
-impl Stream for NostrStream {
-    type Item = Result<NostrMessage>;
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        // get the configuration
-        /// Convert Message to NostrMessage
-        fn convert(msg: String) -> Result<NostrMessage> {
-            let config = config::SETTINGS.read().unwrap();
-            let parsed_res: Result<NostrMessage> = serde_json::from_str(&msg).map_err(|e| e.into());
-            match parsed_res {
-                Ok(m) => {
-                    if let NostrMessage::EventMsg(_) = m {
-                        if let Some(max_size) = config.limits.max_event_bytes {
-                            // check length, ensure that some max size is set.
-                            if msg.len() > max_size && max_size > 0 {
-                                return Err(Error::EventMaxLengthError(msg.len()));
-                            }
-                        }
-                    }
-                    Ok(m)
-                }
-                Err(e) => {
-                    debug!("proto parse error: {:?}", e);
-                    debug!("parse error on message: {}", msg.trim());
-                    Err(Error::ProtoParseError)
-                }
-            }
-        }
-        match Pin::new(&mut self.ws_stream).poll_next(cx) {
-            Poll::Pending => Poll::Pending,
-            Poll::Ready(None) => Poll::Ready(None),
-            Poll::Ready(Some(v)) => match v {
-                Ok(Message::Text(vs)) => Poll::Ready(Some(convert(vs))),
-                Ok(Message::Ping(_x)) => {
-                    debug!("client ping");
-                    //Pin::new(&mut self.ws_stream).start_send(Message::Pong(x));
-                    //info!("sent pong");
-                    Poll::Pending
-                }
-                Ok(Message::Binary(_)) => Poll::Ready(Some(Err(Error::ProtoParseError))),
-                Ok(Message::Pong(_)) => Poll::Pending,
-                Ok(Message::Close(_)) => Poll::Ready(None),
-                Err(WsError::AlreadyClosed) | Err(WsError::ConnectionClosed) => Poll::Ready(None),
-                Err(_) => Poll::Ready(Some(Err(Error::ConnError))),
-            },
-        }
-    }
-}
-
-/// Implement the [`Sink`] interface to produce Nostr responses.
-impl Sink<NostrResponse> for NostrStream {
-    type Error = Error;
-
-    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        // map the error type
-        match Pin::new(&mut self.ws_stream).poll_ready(cx) {
-            Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
-            Poll::Ready(Err(_)) => Poll::Ready(Err(Error::ConnWriteError)),
-            Poll::Pending => Poll::Pending,
-        }
-    }
-
-    fn start_send(mut self: Pin<&mut Self>, item: NostrResponse) -> Result<(), Self::Error> {
-        // TODO: do real escaping for these - at least on NOTICE,
-        // which surely has some problems if arbitrary text is sent.
-        let send_str = match item {
-            NostrResponse::NoticeRes(msg) => {
-                let s = msg.replace("\"", "");
-                format!("[\"NOTICE\",\"{}\"]", s)
-            }
-            NostrResponse::EventRes(sub, eventstr) => {
-                let subesc = sub.replace("\"", "");
-                format!("[\"EVENT\",\"{}\",{}]", subesc, eventstr)
-            }
-        };
-        match Pin::new(&mut self.ws_stream).start_send(Message::Text(send_str)) {
-            Ok(()) => Ok(()),
-            Err(_) => Err(Error::ConnWriteError),
-        }
-    }
-
-    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        Poll::Ready(Ok(()))
-    }
-
-    fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        Poll::Ready(Ok(()))
-    }
-}

src/schema.rs

@@ -0,0 +1,290 @@
+//! Database schema and migrations
+use crate::db::PooledConnection;
+use crate::error::Result;
+use crate::event::{single_char_tagname, Event};
+use crate::utils::is_lower_hex;
+use log::*;
+use rusqlite::limits::Limit;
+use rusqlite::params;
+use rusqlite::Connection;
+use std::time::Instant;
+
+/// Startup DB Pragmas
+pub const STARTUP_SQL: &str = r##"
+PRAGMA main.synchronous=NORMAL;
+PRAGMA foreign_keys = ON;
+pragma mmap_size = 536870912; -- 512MB of mmap
+"##;
+
+/// Schema definition
+const INIT_SQL: &str = r##"
+-- Database settings
+PRAGMA encoding = "UTF-8";
+PRAGMA journal_mode=WAL;
+PRAGMA main.synchronous=NORMAL;
+PRAGMA foreign_keys = ON;
+PRAGMA application_id = 1654008667;
+PRAGMA user_version = 6;
+
+-- Event Table
+CREATE TABLE IF NOT EXISTS event (
+id INTEGER PRIMARY KEY,
+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
+kind INTEGER NOT NULL, -- event kind
+hidden INTEGER, -- relevant for queries
+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 created_at_index ON event(created_at);
+CREATE INDEX IF NOT EXISTS author_index ON event(author);
+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
+-- hex-string), or TEXT otherwise.
+-- This means that searches need to select the appropriate column.
+CREATE TABLE IF NOT EXISTS tag (
+id INTEGER PRIMARY KEY,
+event_id INTEGER NOT NULL, -- an event ID that contains a tag.
+name TEXT, -- the tag name ("p", "e", whatever)
+value TEXT, -- the tag value, if not hex.
+value_hex BLOB, -- the tag value, if it can be interpreted as a lowercase hex string.
+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);
+
+-- NIP-05 User Validation
+CREATE TABLE IF NOT EXISTS user_verification (
+id INTEGER PRIMARY KEY,
+metadata_event INTEGER NOT NULL, -- the metadata event used for this validation.
+name TEXT NOT NULL, -- the nip05 field value (user@domain).
+verified_at INTEGER, -- timestamp this author/nip05 was most recently verified.
+failed_at INTEGER, -- timestamp a verification attempt failed (host down).
+failure_count INTEGER DEFAULT 0, -- number of consecutive failures.
+FOREIGN KEY(metadata_event) REFERENCES event(id) ON UPDATE CASCADE ON DELETE CASCADE
+);
+CREATE INDEX IF NOT EXISTS user_verification_name_index ON user_verification(name);
+CREATE INDEX IF NOT EXISTS user_verification_event_index ON user_verification(metadata_event);
+"##;
+
+/// Determine the current application database schema version.
+pub fn db_version(conn: &mut Connection) -> Result<usize> {
+    let query = "PRAGMA user_version;";
+    let curr_version = conn.query_row(query, [], |row| row.get(0))?;
+    Ok(curr_version)
+}
+
+/// Upgrade DB to latest version, and execute pragma settings
+pub fn upgrade_db(conn: &mut PooledConnection) -> Result<()> {
+    // check the version.
+    let mut curr_version = db_version(conn)?;
+    info!("DB version = {:?}", curr_version);
+
+    debug!(
+        "SQLite max query parameters: {}",
+        conn.limit(Limit::SQLITE_LIMIT_VARIABLE_NUMBER)
+    );
+    debug!(
+        "SQLite max table/blob/text length: {} MB",
+        (conn.limit(Limit::SQLITE_LIMIT_LENGTH) as f64 / (1024 * 1024) as f64).floor()
+    );
+    debug!(
+        "SQLite max SQL length: {} MB",
+        (conn.limit(Limit::SQLITE_LIMIT_SQL_LENGTH) as f64 / (1024 * 1024) as f64).floor()
+    );
+
+    // initialize from scratch
+    if curr_version == 0 {
+        match conn.execute_batch(INIT_SQL) {
+            Ok(()) => {
+                info!("database pragma/schema initialized to v6, and ready");
+            }
+            Err(err) => {
+                error!("update failed: {}", err);
+                panic!("database could not be initialized");
+            }
+        }
+    }
+    if curr_version == 1 {
+        // only change is adding a hidden column to events.
+        let upgrade_sql = r##"
+ALTER TABLE event ADD hidden INTEGER;
+UPDATE event SET hidden=FALSE;
+PRAGMA user_version = 2;
+"##;
+        match conn.execute_batch(upgrade_sql) {
+            Ok(()) => {
+                info!("database schema upgraded v1 -> v2");
+                curr_version = 2;
+            }
+            Err(err) => {
+                error!("update failed: {}", err);
+                panic!("database could not be upgraded");
+            }
+        }
+    }
+    if curr_version == 2 {
+        // this version lacks the tag column
+        info!("database schema needs update from 2->3");
+        let upgrade_sql = r##"
+CREATE TABLE IF NOT EXISTS tag (
+id INTEGER PRIMARY KEY,
+event_id INTEGER NOT NULL, -- an event ID that contains a tag.
+name TEXT, -- the tag name ("p", "e", whatever)
+value TEXT, -- the tag value, if not hex.
+value_hex BLOB, -- the tag value, if it can be interpreted as a hex string.
+FOREIGN KEY(event_id) REFERENCES event(id) ON UPDATE CASCADE ON DELETE CASCADE
+);
+PRAGMA user_version = 3;
+"##;
+        // TODO: load existing refs into tag table
+        match conn.execute_batch(upgrade_sql) {
+            Ok(()) => {
+                info!("database schema upgraded v2 -> v3");
+                curr_version = 3;
+            }
+            Err(err) => {
+                error!("update failed: {}", err);
+                panic!("database could not be upgraded");
+            }
+        }
+        // iterate over every event/pubkey tag
+        let tx = conn.transaction()?;
+        {
+            let mut stmt = tx.prepare("select event_id, \"e\", lower(hex(referenced_event)) from event_ref union select event_id, \"p\", lower(hex(referenced_pubkey)) from pubkey_ref;")?;
+            let mut tag_rows = stmt.query([])?;
+            while let Some(row) = tag_rows.next()? {
+                // we want to capture the event_id that had the tag, the tag name, and the tag hex value.
+                let event_id: u64 = row.get(0)?;
+                let tag_name: String = row.get(1)?;
+                let tag_value: String = row.get(2)?;
+                // this will leave behind p/e tags that were non-hex, but they are invalid anyways.
+                if is_lower_hex(&tag_value) {
+                    tx.execute(
+                        "INSERT INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3);",
+                        params![event_id, tag_name, hex::decode(&tag_value).ok()],
+                    )?;
+                }
+            }
+        }
+        tx.commit()?;
+        info!("Upgrade complete");
+    }
+    if curr_version == 3 {
+        info!("database schema needs update from 3->4");
+        let upgrade_sql = r##"
+-- incoming metadata events with nip05
+CREATE TABLE IF NOT EXISTS user_verification (
+id INTEGER PRIMARY KEY,
+metadata_event INTEGER NOT NULL, -- the metadata event used for this validation.
+name TEXT NOT NULL, -- the nip05 field value (user@domain).
+verified_at INTEGER, -- timestamp this author/nip05 was most recently verified.
+failed_at INTEGER, -- timestamp a verification attempt failed (host down).
+failure_count INTEGER DEFAULT 0, -- number of consecutive failures.
+FOREIGN KEY(metadata_event) REFERENCES event(id) ON UPDATE CASCADE ON DELETE CASCADE
+);
+CREATE INDEX IF NOT EXISTS user_verification_name_index ON user_verification(name);
+CREATE INDEX IF NOT EXISTS user_verification_event_index ON user_verification(metadata_event);
+PRAGMA user_version = 4;
+"##;
+        match conn.execute_batch(upgrade_sql) {
+            Ok(()) => {
+                info!("database schema upgraded v3 -> v4");
+                curr_version = 4;
+            }
+            Err(err) => {
+                error!("update failed: {}", err);
+                panic!("database could not be upgraded");
+            }
+        }
+    }
+
+    if curr_version == 4 {
+        info!("database schema needs update from 4->5");
+        let upgrade_sql = r##"
+DROP TABLE IF EXISTS event_ref;
+DROP TABLE IF EXISTS pubkey_ref;
+PRAGMA user_version=5;
+"##;
+        match conn.execute_batch(upgrade_sql) {
+            Ok(()) => {
+                info!("database schema upgraded v4 -> v5");
+                // uncomment if we have a newer version
+                //curr_version = 5;
+            }
+            Err(err) => {
+                error!("update failed: {}", err);
+                panic!("database could not be upgraded");
+            }
+        }
+    } else if curr_version == 5 {
+        info!("database schema needs update from 5->6");
+        // We need to rebuild the tags table.  iterate through the
+        // event table.  build event from json, insert tags into a
+        // fresh tag table.  This was needed due to a logic error in
+        // how hex-like tags got indexed.
+        let start = Instant::now();
+        let tx = conn.transaction()?;
+        {
+            // Clear out table
+            tx.execute("DELETE FROM tag;", [])?;
+            let mut stmt = tx.prepare("select id, content from event order by id;")?;
+            let mut tag_rows = stmt.query([])?;
+            while let Some(row) = tag_rows.next()? {
+                // we want to capture the event_id that had the tag, the tag name, and the tag hex value.
+                let event_id: u64 = row.get(0)?;
+                let event_json: String = row.get(1)?;
+                let event: Event = serde_json::from_str(&event_json)?;
+                // look at each event, and each tag, creating new tag entries if appropriate.
+                for t in event.tags.iter().filter(|x| x.len() > 1) {
+                    let tagname = t.get(0).unwrap();
+                    let tagnamechar_opt = single_char_tagname(tagname);
+                    if tagnamechar_opt.is_none() {
+                        continue;
+                    }
+                    // safe because len was > 1
+                    let tagval = t.get(1).unwrap();
+                    // insert as BLOB if we can restore it losslessly.
+                    // this means it needs to be even length and lowercase.
+                    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()],
+                        )?;
+                    } else {
+                        // otherwise, insert as text
+                        tx.execute(
+                            "INSERT INTO tag (event_id, name, value) VALUES (?1, ?2, ?3);",
+                            params![event_id, tagname, &tagval],
+                        )?;
+                    }
+                }
+            }
+            tx.execute("PRAGMA user_version = 6;", [])?;
+        }
+        tx.commit()?;
+        info!("database schema upgraded v5 -> v6 in {:?}", start.elapsed());
+        // vacuum after large table modification
+        let start = Instant::now();
+        conn.execute("VACUUM;", [])?;
+        info!("vacuumed DB after tags rebuild in {:?}", start.elapsed());
+    } else if curr_version == 6 {
+        debug!("Database version was already current (v6)");
+    } else if curr_version > 7 {
+        panic!(
+            "Database version is newer than supported by this executable (v{})",
+            curr_version
+        );
+    }
+
+    // Setup PRAGMA
+    conn.execute_batch(STARTUP_SQL)?;
+    debug!("SQLite PRAGMA startup completed");
+    Ok(())
+}

src/subscription.rs

@@ -31,9 +31,13 @@ pub struct ReqFilter {
     pub until: Option<u64>,
     /// List of author public keys
     pub authors: Option<Vec<String>>,
+    /// Limit number of results
+    pub limit: Option<u64>,
     /// Set of tags
     #[serde(skip)]
-    pub tags: Option<HashMap<String, HashSet<String>>>,
+    pub tags: Option<HashMap<char, HashSet<String>>>,
+    /// Force no matches due to malformed data
+    pub force_no_match: bool,
 }
 
 impl<'de> Deserialize<'de> for ReqFilter {
@@ -54,7 +58,9 @@ impl<'de> Deserialize<'de> for ReqFilter {
             since: None,
             until: None,
             authors: None,
+            limit: None,
             tags: None,
+            force_no_match: false,
         };
         let mut ts = None;
         // iterate through each key, and assign values that exist
@@ -68,22 +74,28 @@ impl<'de> Deserialize<'de> for ReqFilter {
                 rf.since = Deserialize::deserialize(val).ok();
             } else if key == "until" {
                 rf.until = Deserialize::deserialize(val).ok();
+            } else if key == "limit" {
+                rf.limit = Deserialize::deserialize(val).ok();
             } else if key == "authors" {
                 rf.authors = Deserialize::deserialize(val).ok();
             } else if key.starts_with('#') && key.len() > 1 && val.is_array() {
-                // remove the prefix
-                let tagname = &key[1..];
-                if ts.is_none() {
-                    // Initialize the tag if necessary
-                    ts = Some(HashMap::new());
-                }
-                if let Some(m) = ts.as_mut() {
-                    let tag_vals: Option<Vec<String>> = Deserialize::deserialize(val).ok();
-                    if let Some(v) = tag_vals {
-                        let hs = HashSet::from_iter(v.into_iter());
-                        m.insert(tagname.to_owned(), hs);
+                if let Some(tag_search) = tag_search_char_from_filter(key) {
+                    if ts.is_none() {
+                        // Initialize the tag if necessary
+                        ts = Some(HashMap::new());
                     }
-                };
+                    if let Some(m) = ts.as_mut() {
+                        let tag_vals: Option<Vec<String>> = Deserialize::deserialize(val).ok();
+                        if let Some(v) = tag_vals {
+                            let hs = HashSet::from_iter(v.into_iter());
+                            m.insert(tag_search.to_owned(), hs);
+                        }
+                    };
+                } else {
+                    // tag search that is multi-character, don't add to subscription
+                    rf.force_no_match = true;
+                    continue;
+                }
             }
         }
         rf.tags = ts;
@@ -91,6 +103,26 @@ impl<'de> Deserialize<'de> for ReqFilter {
     }
 }
 
+/// Attempt to form a single-char identifier from a tag search filter
+fn tag_search_char_from_filter(tagname: &str) -> Option<char> {
+    let tagname_nohash = &tagname[1..];
+    // We return the tag character if and only if the tagname consists
+    // of a single char.
+    let mut tagnamechars = tagname_nohash.chars();
+    let firstchar = tagnamechars.next();
+    return match firstchar {
+        Some(_) => {
+            // check second char
+            if tagnamechars.next().is_none() {
+                firstchar
+            } else {
+                None
+            }
+        }
+        None => None,
+    };
+}
+
 impl<'de> Deserialize<'de> for Subscription {
     /// Custom deserializer for subscriptions, which have a more
     /// complex structure than the other message types.
@@ -189,7 +221,7 @@ impl ReqFilter {
         // get the hashset from the filter.
         if let Some(map) = &self.tags {
             for (key, val) in map.iter() {
-                let tag_match = event.generic_tag_val_intersect(key, val);
+                let tag_match = event.generic_tag_val_intersect(*key, val);
                 // if there is no match for this tag, the match fails.
                 if !tag_match {
                     return false;
@@ -214,9 +246,11 @@ impl ReqFilter {
         //        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.until.map(|t| event.created_at < t).unwrap_or(true)
             && self.kind_match(event.kind)
             && self.authors_match(event)
             && self.tag_match(event)
+            && !self.force_no_match
     }
 }
 
@@ -321,6 +355,50 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn interest_until() -> Result<()> {
+        // subscription with a filter for ID and time
+        let s: Subscription =
+            serde_json::from_str(r#"["REQ","xyz",{"ids": ["abc"], "until": 1000}]"#)?;
+        let e = Event {
+            id: "abc".to_owned(),
+            pubkey: "".to_owned(),
+            created_at: 50,
+            kind: 0,
+            tags: Vec::new(),
+            content: "".to_owned(),
+            sig: "".to_owned(),
+            tagidx: None,
+        };
+        assert!(s.interested_in_event(&e));
+        Ok(())
+    }
+
+    #[test]
+    fn interest_range() -> Result<()> {
+        // subscription with a filter for ID and time
+        let s_in: Subscription =
+            serde_json::from_str(r#"["REQ","xyz",{"ids": ["abc"], "since": 100, "until": 200}]"#)?;
+        let s_before: Subscription =
+            serde_json::from_str(r#"["REQ","xyz",{"ids": ["abc"], "since": 100, "until": 140}]"#)?;
+        let s_after: Subscription =
+            serde_json::from_str(r#"["REQ","xyz",{"ids": ["abc"], "since": 160, "until": 200}]"#)?;
+        let e = Event {
+            id: "abc".to_owned(),
+            pubkey: "".to_owned(),
+            created_at: 150,
+            kind: 0,
+            tags: Vec::new(),
+            content: "".to_owned(),
+            sig: "".to_owned(),
+            tagidx: None,
+        };
+        assert!(s_in.interested_in_event(&e));
+        assert!(!s_before.interested_in_event(&e));
+        assert!(!s_after.interested_in_event(&e));
+        Ok(())
+    }
+
     #[test]
     fn interest_time_and_id() -> Result<()> {
         // subscription with a filter for ID and time

src/tags.rs

@@ -1,315 +0,0 @@
-//! Tags used in events to link to another event or a pubkey
-//!
-//! Reference specification NIP01: https://github.com/fiatjaf/nostr/blob/master/nips/01.md#events-and-signatures
-//!
-
-use bitcoin_hashes::hex::ToHex;
-use bitcoin_hashes::sha256;
-use secp256k1::XOnlyPublicKey;
-use std::str::FromStr;
-
-use serde::de::Unexpected;
-use serde::ser::SerializeSeq;
-use serde::{Deserialize, Deserializer, Serializer};
-use serde_json::Value;
-
-use crate::error::Error;
-
-type EventId = sha256::Hash;
-
-#[derive(Debug, PartialEq)]
-struct EventTag {
-    event_id: EventId,
-    recommended_url: Option<String>,
-}
-
-#[derive(Debug, PartialEq)]
-struct PubkeyTag {
-    pubkey: XOnlyPublicKey,
-    recommended_url: Option<String>,
-}
-
-// Tag structure representing two possible types of tags
-#[derive(Debug, PartialEq)]
-enum Tag {
-    Event(EventTag),
-    Pubkey(PubkeyTag),
-}
-
-// Custom json serialization into protocol network format
-// Event tag : ["e", "<32 byte event-id>", "optional<url>"]
-// Pubkey tag : ["p", "<32 byte Xonly Pubkey>", "optional<url>"]
-impl serde::Serialize for Tag {
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        match self {
-            Self::Event(event_tag) => {
-                let mut seq = serializer.serialize_seq(None)?;
-                seq.serialize_element("e")?;
-                seq.serialize_element(&event_tag.event_id.to_hex())?;
-                if let Some(url) = &event_tag.recommended_url {
-                    seq.serialize_element(url)?;
-                } else {
-                }
-                seq.end()
-            }
-            Self::Pubkey(pubkey_tag) => {
-                let mut seq = serializer.serialize_seq(None)?;
-                seq.serialize_element("p")?;
-                seq.serialize_element(&pubkey_tag.pubkey.to_hex())?;
-                if let Some(url) = &pubkey_tag.recommended_url {
-                    seq.serialize_element(url)?;
-                } else {
-                }
-                seq.end()
-            }
-        }
-    }
-}
-
-// Custom json deserialization from protocol network format
-impl<'de> serde::Deserialize<'de> for Tag {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        // Receive incoming data in a json object
-        let received: Value = Deserialize::deserialize(deserializer)?;
-
-        // Check received data is a json array
-        let values = received.as_array().ok_or_else(|| {
-            serde::de::Error::invalid_type(Unexpected::Other("tag json object"), &"json array")
-        })?;
-
-        // Check json array contains only string
-        let values = values
-            .iter()
-            .map(|value| {
-                value.as_str().ok_or_else(|| {
-                    serde::de::Error::invalid_type(
-                        Unexpected::Other("tag json data"),
-                        &"json string",
-                    )
-                })
-            })
-            .collect::<Result<Vec<_>, _>>()?;
-
-        // Check length is not more tha 3
-        if values.len() > 3 {
-            Err(serde::de::Error::invalid_length(
-                values.len(),
-                &"tag length is 2 or 3",
-            ))
-        } else {
-            // Parse the json array into appropriate types
-            match values[0] {
-                // This denotes an event type tag
-                "e" => {
-                    let event_id = EventId::from_str(values[1])
-                        .map_err(|e| serde::de::Error::custom(e.to_string()))?;
-                    let recomended_url = if values.len() == 3 {
-                        Some(values[2].into())
-                    } else {
-                        None
-                    };
-                    Ok(Tag::Event(EventTag {
-                        event_id,
-                        recommended_url: recomended_url,
-                    }))
-                }
-                // This denotes a pubkey type tag
-                "p" => {
-                    let pubkey = XOnlyPublicKey::from_str(values[1])
-                        .map_err(|e| serde::de::Error::custom(e.to_string()))?;
-                    let recomended_url = if values.len() == 3 {
-                        Some(values[2].into())
-                    } else {
-                        None
-                    };
-
-                    Ok(Tag::Pubkey(PubkeyTag {
-                        pubkey,
-                        recommended_url: recomended_url,
-                    }))
-                }
-                // Any other tag type is currently not supported
-                _ => Err(serde::de::Error::invalid_value(
-                    Unexpected::Other("tag type flag"),
-                    &"'e' or 'p'",
-                )),
-            }
-        }
-    }
-}
-
-#[allow(dead_code)]
-// Some api (not public currently) to use Tags
-impl Tag {
-    fn new_event_tag(event_id: EventId, recomended_url: Option<String>) -> Self {
-        Self::Event(EventTag {
-            event_id,
-            recommended_url: recomended_url,
-        })
-    }
-
-    fn new_pubkey_tag(pubkey: XOnlyPublicKey, recomended_url: Option<String>) -> Self {
-        Self::Pubkey(PubkeyTag {
-            pubkey,
-            recommended_url: recomended_url,
-        })
-    }
-
-    fn get_recomended_url(&self) -> Option<String> {
-        match self {
-            Self::Event(EventTag {
-                event_id: _,
-                recommended_url,
-            }) => recommended_url.clone(),
-            Self::Pubkey(PubkeyTag {
-                pubkey: _,
-                recommended_url,
-            }) => recommended_url.clone(),
-        }
-    }
-
-    fn get_event_id(&self) -> Result<EventId, Error> {
-        match self {
-            Self::Event(EventTag {
-                event_id,
-                recommended_url: _,
-            }) => Ok(*event_id),
-            _ => Err(Error::CustomError("Expected event tag".to_string())),
-        }
-    }
-
-    fn get_pubkey(&self) -> Result<XOnlyPublicKey, Error> {
-        match self {
-            Self::Pubkey(PubkeyTag {
-                pubkey,
-                recommended_url: _,
-            }) => Ok(*pubkey),
-            _ => Err(Error::CustomError("Expected pubkey tag".to_string())),
-        }
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::*;
-    use bitcoin_hashes::Hash;
-
-    #[test]
-    fn serde_roundtrip() {
-        let pubkey = XOnlyPublicKey::from_str(
-            "18845781f631c48f1c9709e23092067d06837f30aa0cd0544ac887fe91ddd166",
-        )
-        .unwrap();
-        let url = "wss://rsslay.fiatjaf.com";
-        static HASH_BYTES: [u8; 32] = [
-            0xef, 0x53, 0x7f, 0x25, 0xc8, 0x95, 0xbf, 0xa7, 0x82, 0x52, 0x65, 0x29, 0xa9, 0xb6,
-            0x3d, 0x97, 0xaa, 0x63, 0x15, 0x64, 0xd5, 0xd7, 0x89, 0xc2, 0xb7, 0x65, 0x44, 0x8c,
-            0x86, 0x35, 0xfb, 0x6c,
-        ];
-        let event_id = EventId::from_slice(&HASH_BYTES).expect("right number of bytes");
-
-        let event_tag = Tag::new_event_tag(event_id, Some(url.to_string()));
-        let pubkey_tag = Tag::new_pubkey_tag(pubkey, Some(url.to_string()));
-
-        let ser_event_tag = serde_json::to_string(&event_tag).unwrap();
-        let ser_pubkey_tag = serde_json::to_string(&pubkey_tag).unwrap();
-
-        let deser_event_tag: Tag = serde_json::from_str(&ser_event_tag).unwrap();
-        let deser_pubkey_tag: Tag = serde_json::from_str(&ser_pubkey_tag).unwrap();
-
-        assert_eq!(deser_event_tag, event_tag);
-        assert_eq!(deser_pubkey_tag, pubkey_tag);
-    }
-
-    #[test]
-    fn invalid_pubkey() {
-        let test_string = r#"["p","845781f631c48f1c9709e23092067d06837f30aa0cd0544ac887fe91ddd166","wss://rsslay.fiatjaf.com"]"#;
-        let tag: Result<Tag, _> = serde_json::from_str(test_string);
-        assert_eq!(
-            tag.err().expect("expect error").to_string(),
-            "secp: malformed public key".to_string()
-        );
-    }
-
-    #[test]
-    fn invalid_datatype() {
-        let test_string = r#"["p", 188457, "wss://rsslay.fiatjaf.com"]"#;
-        let tag: Result<Tag, _> = serde_json::from_str(test_string);
-        assert_eq!(
-            tag.err().expect("expect error").to_string(),
-            "invalid type: tag json data, expected json string".to_string()
-        );
-    }
-
-    #[test]
-    fn invalid_tagbyte() {
-        let test_string = r#"["q", "18845781f631c48f1c9709e23092067d06837f30aa0cd0544ac887fe91ddd166", "wss://rsslay.fiatjaf.com"]"#;
-        let tag: Result<Tag, _> = serde_json::from_str(test_string);
-        assert_eq!(
-            tag.err().expect("expect error").to_string(),
-            "invalid value: tag type flag, expected 'e' or 'p'".to_string()
-        );
-    }
-
-    #[test]
-    fn invalid_event_id() {
-        let test_string = r#"["e","ef537f25c895bfa782526529a9b63d97aa631564d5d78c8635fb6c","wss://rsslay.fiatjaf.com"]"#;
-        let tag: Result<Tag, _> = serde_json::from_str(test_string);
-        assert_eq!(
-            tag.err().expect("expect error").to_string(),
-            "bad hex string length 54 (expected 64)".to_string()
-        );
-    }
-
-    #[test]
-    fn invalid_url_type() {
-        let test_string = r#"["e","ef537f25c895bfa782526529a9b63d97aa631564d5d789c2b765448c8635fb6c", 123456788]"#;
-        let tag: Result<Tag, _> = serde_json::from_str(test_string);
-        assert_eq!(
-            tag.err().expect("expect error").to_string(),
-            "invalid type: tag json data, expected json string".to_string()
-        );
-    }
-
-    #[test]
-    fn invalid_length() {
-        let test_string = r#"["e","ef537f25c895bfa782526529a9b63d97aa631564d5d789c2b765448c8635fb6c","wss://rsslay.fiatjaf.com", "Random extra data"]"#;
-        let tag: Result<Tag, _> = serde_json::from_str(test_string);
-        assert_eq!(
-            tag.err().expect("expect error").to_string(),
-            "invalid length 4, expected tag length is 2 or 3".to_string()
-        );
-    }
-
-    #[test]
-    fn invalid_json_type() {
-        let test_string = r#"{"type": "e","event": "ef537f25c895bfa782526529a9b63d97aa631564d5d789c2b765448c8635fb6c"}"#;
-        let tag: Result<Tag, _> = serde_json::from_str(test_string);
-        assert_eq!(
-            tag.err().expect("expect error").to_string(),
-            "invalid type: tag json object, expected json array".to_string()
-        );
-    }
-
-    #[test]
-    fn evenrt_tag_missing_url() {
-        let test_string =
-            r#"["e","ef537f25c895bfa782526529a9b63d97aa631564d5d789c2b765448c8635fb6c"]"#;
-        let tag: Tag = serde_json::from_str(test_string).unwrap();
-        assert!(tag.get_recomended_url().is_none());
-    }
-
-    #[test]
-    fn pubkey_tag_missing_url() {
-        let test_string =
-            r#"["p","18845781f631c48f1c9709e23092067d06837f30aa0cd0544ac887fe91ddd166"]"#;
-        let tag: Tag = serde_json::from_str(test_string).unwrap();
-        assert!(tag.get_recomended_url().is_none());
-    }
-}

src/utils.rs

@@ -0,0 +1,33 @@
+//! Common utility functions
+use std::time::SystemTime;
+
+/// Seconds since 1970.
+pub fn unix_time() -> u64 {
+    SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .map(|x| x.as_secs())
+        .unwrap_or(0)
+}
+
+/// Check if a string contains only hex characters.
+pub fn is_hex(s: &str) -> bool {
+    s.chars().all(|x| char::is_ascii_hexdigit(&x))
+}
+
+/// Check if a string contains only lower-case hex chars.
+pub fn is_lower_hex(s: &str) -> bool {
+    s.chars().all(|x| {
+        (char::is_ascii_lowercase(&x) || char::is_ascii_digit(&x)) && char::is_ascii_hexdigit(&x)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn lower_hex() {
+        let hexstr = "abcd0123";
+        assert_eq!(is_lower_hex(hexstr), true);
+    }
+}