fix: move from std::sync fully to tokio

Fixes Relay adding

Closes https://github.com/rust-nostr/nostr/issues/533

Cargo.lock

@@ -166,31 +166,22 @@ dependencies = [
 
 [[package]]
 name = "async-wsocket"
-version = "0.5.2"
+version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3445f8f330db8e5f3be7912f170f32e43fec90d995c71ced1ec3b8394b4a873c"
+checksum = "5725a0615e4eb98e82e9cb963529398114e3fccfbf0e8b9111d605e2ac443e46"
 dependencies = [
  "async-utility",
+ "futures",
  "futures-util",
+ "js-sys",
  "thiserror",
  "tokio",
  "tokio-rustls 0.26.0",
  "tokio-socks",
  "tokio-tungstenite",
  "url",
- "wasm-ws",
+ "wasm-bindgen",
- "webpki-roots",
+ "web-sys",
-]
-
-[[package]]
-name = "async_io_stream"
-version = "0.3.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6d7b9decdf35d8908a7e3ef02f64c5e9b1695e230154c0e8de3969142d9b94c"
-dependencies = [
- "futures",
- "pharos",
- "rustc_version",
 ]
 
 [[package]]
@@ -669,7 +660,6 @@ checksum = "645c6916888f6cb6350d2550b80fb63e734897a8498abe35cfb732b6487804b0"
 dependencies = [
  "futures-channel",
  "futures-core",
- "futures-executor",
  "futures-io",
  "futures-sink",
  "futures-task",
@@ -692,34 +682,12 @@ version = "0.3.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dfc6580bb841c5a68e9ef15c77ccc837b40a7504914d52e47b8b0e9bbda25a1d"
 
-[[package]]
-name = "futures-executor"
-version = "0.3.30"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a576fc72ae164fca6b9db127eaa9a9dda0d61316034f33a0a0d4eda41f02b01d"
-dependencies = [
- "futures-core",
- "futures-task",
- "futures-util",
-]
-
 [[package]]
 name = "futures-io"
 version = "0.3.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a44623e20b9681a318efdd71c299b6b222ed6f231972bfe2f224ebad6311f0c1"
 
-[[package]]
-name = "futures-macro"
-version = "0.3.30"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn",
-]
-
 [[package]]
 name = "futures-sink"
 version = "0.3.30"
@@ -741,7 +709,6 @@ dependencies = [
  "futures-channel",
  "futures-core",
  "futures-io",
- "futures-macro",
  "futures-sink",
  "futures-task",
  "memchr",
@@ -1285,9 +1252,9 @@ dependencies = [
 
 [[package]]
 name = "nostr"
-version = "0.33.0"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f08db214560a34bf7c4c1fea09a8461b9412bae58ba06e99ce3177d89fa1e0a6"
+checksum = "5897e4142fcc33c4f1d58ad17f665e87dcba70de7e370c0bda1aa0fb73212c2a"
 dependencies = [
  "aes",
  "base64 0.21.7",
@@ -1315,9 +1282,9 @@ dependencies = [
 
 [[package]]
 name = "nostr-database"
-version = "0.33.1"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50eebf5020d70891e3c229128de5fc73af632b651d02742383b314d3d0c7e953"
+checksum = "1926ef55392f3eea1bbe4a1358b64bbf12dd6eb554f40f483941a102c6263fc6"
 dependencies = [
  "async-trait",
  "lru",
@@ -1329,9 +1296,9 @@ dependencies = [
 
 [[package]]
 name = "nostr-relay-pool"
-version = "0.33.0"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "afa5502a3df456790ca16d90cc688a677117d57ab56b079dcfa091390ac9f202"
+checksum = "c6480cf60564957a2a64bd050d047ee0717e08dced7a389e22ef4e9fc104edd2"
 dependencies = [
  "async-utility",
  "async-wsocket",
@@ -1340,14 +1307,15 @@ dependencies = [
  "nostr-database",
  "thiserror",
  "tokio",
+ "tokio-stream",
  "tracing",
 ]
 
 [[package]]
 name = "nostr-sdk"
-version = "0.33.0"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b427dceefbbb49a9dd98abb8c4e40d25fdd467e99821aaad88615252bdb915bd"
+checksum = "ca0c0c5f8ddbdfc064ea71883191ec53de6ed52b5dca10ab07f0810b99e91acc"
 dependencies = [
  "async-utility",
  "atomic-destructor",
@@ -1365,9 +1333,9 @@ dependencies = [
 
 [[package]]
 name = "nostr-signer"
-version = "0.33.0"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "665268b316f41cd8fa791be54b6c7935c5a239461708c380a699d6677be9af38"
+checksum = "5c30294a7be7d9d5ac777954812f5c7b4ae2a1e583a62e33537f87d98ab23729"
 dependencies = [
  "async-utility",
  "nostr",
@@ -1379,9 +1347,9 @@ dependencies = [
 
 [[package]]
 name = "nostr-zapper"
-version = "0.33.0"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69922e74f8eab1f9d287008c0c06acdec87277a2d8f44bd9d38e003422aea0ab"
+checksum = "dcf3ba30e807145e9cb924faf8fb0719e460f613088e99c753b67c2a9929c5b7"
 dependencies = [
  "async-trait",
  "nostr",
@@ -1409,9 +1377,9 @@ dependencies = [
 
 [[package]]
 name = "nwc"
-version = "0.33.0"
+version = "0.34.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb2e04b3edb5e9572e95b62842430625f1718e8a4a3596a30aeb04e6734764ea"
+checksum = "9a16ac06bc273fcd4ead47c0c5a58b6cc7db2247fc7a64dd9bc11cf18e3efeb4"
 dependencies = [
  "async-utility",
  "nostr",
@@ -1479,16 +1447,6 @@ version = "2.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
 
-[[package]]
-name = "pharos"
-version = "0.5.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e9567389417feee6ce15dd6527a8a1ecac205ef62c2932bcf3d9f6fc5b78b414"
-dependencies = [
- "futures",
- "rustc_version",
-]
-
 [[package]]
 name = "phf"
 version = "0.11.2"
@@ -1766,15 +1724,6 @@ version = "0.1.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d626bb9dae77e28219937af045c257c28bfd3f69333c512553507f5f9798cb76"
 
-[[package]]
-name = "rustc_version"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bfa0f585226d2e68097d4f95d113b15b83a82e819ab25717ec0590d9584ef366"
-dependencies = [
- "semver",
-]
-
 [[package]]
 name = "rustix"
 version = "0.38.34"
@@ -1919,18 +1868,6 @@ dependencies = [
  "cc",
 ]
 
-[[package]]
-name = "semver"
-version = "1.0.23"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "61697e0a1c7e512e84a621326239844a24d8207b4669b41bc18b32ea5cbf988b"
-
-[[package]]
-name = "send_wrapper"
-version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cd0b0ec5f1c1ca621c432a25813d8d60c88abe6d3e08a3eb9cf37d97a0fe3d73"
-
 [[package]]
 name = "serde"
 version = "1.0.199"
@@ -2176,6 +2113,17 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-stream"
+version = "0.1.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "267ac89e0bec6e691e5813911606935d77c476ff49024f98abcea3e7b15e37af"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-tungstenite"
 version = "0.23.1"
@@ -2450,23 +2398,6 @@ version = "0.2.92"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "af190c94f2773fdb3729c55b007a722abb5384da03bc0986df4c289bf5567e96"
 
-[[package]]
-name = "wasm-ws"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "688c5806d1b06b4f3d90d015e23364dc5d3af412ee64abba6dde8fdc01637e33"
-dependencies = [
- "async_io_stream",
- "futures",
- "js-sys",
- "pharos",
- "send_wrapper",
- "thiserror",
- "wasm-bindgen",
- "wasm-bindgen-futures",
- "web-sys",
-]
-
 [[package]]
 name = "web-sys"
 version = "0.3.69"

Cargo.toml

@@ -21,7 +21,7 @@ colog = "1.3"
 colored = "2.1"
 parse_datetime = "0.5.0"
 interim = { version = "0.1", features = ["chrono"] }
-nostr-sdk = "0.33" # { git = "https://github.com/rust-nostr/nostr" }
+nostr-sdk = "0.34" # { git = "https://github.com/rust-nostr/nostr" }
 tokio = { version = "1.37", features = ["rt", "rt-multi-thread", "macros"] }
 regex = "1.10.5"
 

README.md

@@ -115,7 +115,8 @@ Dot or slash can be repeated to move to parent tasks before acting.
 - `<[TEXT]` - close active task and move up, with optional status description
 - `!TEXT` - set status for current task from text and move up (empty: Open)
 - `,[TEXT]` - list notes or add text note (comment / description)
-- `*[INT]` - set priority - can also be used in task creation, with any digit
+- TBI: `*[INT]` - set priority - can also be used in task creation, with any digit
+- TBI: `;[TEXT]` - list comments or comment on task
 - TBI: show status history and creation with attribution
 - `&` - undo last action (moving in place or upwards confirms pending actions)
 - `wss://...` - switch or subscribe to relay (prefix with space to forcibly add a new one)
@@ -126,7 +127,7 @@ Property Filters:
 - `+TAG` - add tag filter
 - `-TAG` - remove tag filters by prefix
 - `?STATUS` - filter by status (type or description) - plain `?` to reset, `??` to show all
-- `@AUTHOR` - filter by author (`@` for self, id prefix, name prefix)
+- `@AUTHOR` - filter by time or author (pubkey, or `@` for self, TBI: id prefix, name prefix)
 - TBI: `**INT` - filter by priority
 - TBI: Filter by time
 
@@ -181,7 +182,6 @@ Considering to use Calendar: https://github.com/nostr-protocol/nips/blob/master/
 - Open Command characters: `_^\=$%~'"`, `{}[]`
 - Remove colon from task creation syntax
 - reassign undo to `&` and use `@` for people
-- maybe use `;` for sorting instead of `::`
   
 ### Conceptual
 
@@ -189,7 +189,7 @@ The following features are not ready to be implemented
 because they need conceptualization.
 Suggestions welcome!
 
-- Do not track time on Closed task?
+- Queueing tasks
 - Allow adding new parent via description?
 - Special commands: help, exit, tutorial, change log level
 - Duplicate task (subtasks? timetracking?)

src/kinds.rs

@@ -48,7 +48,7 @@ Utilities:
 - `path` - name including parent tasks
 - `rpath` - name including parent tasks up to active task
 - TBI `depends` - list all tasks this task depends on before it becomes actionable
-Debugging: `pubkey`, `props`, `alltags`, `descriptions`";
+Debugging: `kind`, `pubkey`, `props`, `alltags`, `descriptions`";
 
 pub(crate) fn build_tracking<I>(id: I) -> EventBuilder
 where

src/main.rs

@@ -8,18 +8,19 @@ use std::iter::once;
 use std::ops::Sub;
 use std::path::PathBuf;
 use std::str::FromStr;
-use std::sync::mpsc;
-use std::sync::mpsc::RecvTimeoutError;
-use std::sync::mpsc::Sender;
 use std::time::Duration;
 
-use colored::{ColoredString, Colorize};
+use colored::Colorize;
 use env_logger::Builder;
 use itertools::Itertools;
 use log::{debug, error, info, LevelFilter, trace, warn};
 use nostr_sdk::prelude::*;
 use nostr_sdk::TagStandard::Hashtag;
 use regex::Regex;
+use tokio::sync::mpsc;
+use tokio::sync::mpsc::Sender;
+use tokio::time::error::Elapsed;
+use tokio::time::timeout;
 use xdg::BaseDirectories;
 
 use crate::helpers::*;
@@ -80,7 +81,7 @@ impl EventSender {
         debug!("Flushing {} events from queue", self.queue.borrow().len());
         let values = self.clear();
         self.url.as_ref().map(|url| {
-            self.tx.send(MostrMessage::AddTasks(url.clone(), values)).inspect_err(|e| {
+            self.tx.try_send(MostrMessage::AddTasks(url.clone(), values)).err().map(|e| {
                 error!("Nostr communication thread failure, changes will not be persisted: {}", e)
             })
         });
@@ -180,20 +181,20 @@ async fn main() {
         },
     }
 
+    let mut notifications = client.notifications();
+    client.connect().await;
+
     let sub1 = client.subscribe(vec![
         Filter::new().kinds(KINDS.into_iter().map(|k| Kind::from(k)))
     ], None).await;
     info!("Subscribed to tasks with {:?}", sub1);
 
-    let mut notifications = client.notifications();
-    client.connect().await;
-
     let sub2 = client.subscribe(vec![
         Filter::new().kinds(PROP_KINDS.into_iter().map(|k| Kind::from(k)))
     ], None).await;
     info!("Subscribed to updates with {:?}", sub2);
 
-    let (tx, rx) = mpsc::channel::<MostrMessage>();
+    let (tx, mut rx) = mpsc::channel::<MostrMessage>(64);
     let tasks_for_url = |url: Option<Url>| Tasks::from(url, &tx, &keys);
     let mut relays: HashMap<Url, Tasks> =
         client.relays().await.into_keys().map(|url| (url.clone(), tasks_for_url(Some(url)))).collect();
@@ -201,23 +202,16 @@ async fn main() {
     let sender = tokio::spawn(async move {
         let mut queue: Option<(Url, Vec<Event>)> = None;
 
-        loop {
         if let Ok(user) = var("USER") {
             let metadata = Metadata::new()
                 .name(user);
-                //    .display_name("My Username")
-                //    .about("Description")
-                //    .picture(Url::parse("https://example.com/avatar.png")?)
-                //    .banner(Url::parse("https://example.com/banner.png")?)
-                //    .nip05("username@example.com")
-                //    .lud16("yuki@getalby.com")
-                //    .custom_field("custom_field", "my value");
             or_print(client.set_metadata(&metadata).await);
         }
 
-            let result_received = rx.recv_timeout(Duration::from_secs(INACTVITY_DELAY));
+        loop {
+            let result_received = timeout(Duration::from_secs(INACTVITY_DELAY), rx.recv()).await;
             match result_received {
-                Ok(MostrMessage::NewRelay(url)) => {
+                Ok(Some(MostrMessage::NewRelay(url))) => {
                     if client.add_relay(&url).await.unwrap() {
                         match client.connect_relay(&url).await {
                             Ok(()) => info!("Connected to {url}"),
@@ -227,7 +221,7 @@ async fn main() {
                         warn!("Relay {url} already added");
                     }
                 }
-                Ok(MostrMessage::AddTasks(url, mut events)) => {
+                Ok(Some(MostrMessage::AddTasks(url, mut events))) => {
                     trace!("Queueing {:?}", &events);
                     if let Some((queue_url, mut queue_events)) = queue {
                         if queue_url == url {
@@ -244,14 +238,14 @@ async fn main() {
                         queue = Some((url, events))
                     }
                 }
-                Ok(MostrMessage::Flush) | Err(RecvTimeoutError::Timeout) => if let Some((url, events)) = queue {
+                Ok(Some(MostrMessage::Flush)) | Err(Elapsed { .. }) => if let Some((url, events)) = queue {
                     info!("Sending {} events to {url} due to {}", events.len(),
-                        result_received.map_or_else(|e| format!("{:?}", e), |m| format!("{:?}", m)));
+                        result_received.map_or("inactivity", |_| "flush message"));
                     client.batch_event_to(vec![url], events, RelaySendOptions::new()).await;
                     queue = None;
                 }
-                Err(err) => {
+                Ok(None) => {
-                    debug!("Finalizing nostr communication thread because of {:?}: {}", err, err);
+                    debug!("Finalizing nostr communication thread because communication channel was closed");
                     break;
                 }
             }
@@ -595,7 +589,7 @@ async fn main() {
                             }
                             match Url::parse(&input) {
                                 Err(e) => warn!("Failed to parse url \"{input}\": {}", e),
-                                Ok(url) => match tx.send(MostrMessage::NewRelay(url.clone())) {
+                                Ok(url) => match tx.send(MostrMessage::NewRelay(url.clone())).await {
                                     Err(e) => error!("Nostr communication thread failure, cannot add relay \"{url}\": {e}"),
                                     Ok(_) => {
                                         info!("Connecting to {url}");

src/task.rs

@@ -157,6 +157,7 @@ impl Task {
             "name" => Some(self.event.content.clone()),
             "pubkey" => Some(self.event.pubkey.to_string()),
             "created" => Some(local_datetimestamp(&self.event.created_at)),
+            "kind" => Some(self.event.kind.to_string()),
             // Dynamic
             "status" => self.state_label().map(|c| c.to_string()),
             "desc" => self.descriptions().last().cloned(),

src/tasks.rs

@@ -4,7 +4,6 @@ use std::io::{Error, stdout, Write};
 use std::iter::{empty, once};
 use std::ops::{Div, Rem};
 use std::str::FromStr;
-use std::sync::mpsc::Sender;
 use std::time::Duration;
 
 use chrono::Local;
@@ -105,7 +104,7 @@ impl Display for StateFilter {
 }
 
 impl Tasks {
-    pub(crate) fn from(url: Option<Url>, tx: &Sender<MostrMessage>, keys: &Keys) -> Self {
+    pub(crate) fn from(url: Option<Url>, tx: &tokio::sync::mpsc::Sender<MostrMessage>, keys: &Keys) -> Self {
         Self::with_sender(EventSender {
             url,
             tx: tx.clone(),
@@ -765,7 +764,7 @@ impl Tasks {
 
     pub(crate) fn add_task(&mut self, event: Event) {
         if self.tasks.contains_key(&event.id) {
-            debug!("Did not insert duplicate event {}", event.id); // TODO warn in next sdk version
+            warn!("Did not insert duplicate event {}", event.id);
         } else {
             let id = event.id;
             let task = Task::new(event);
@@ -1093,10 +1092,10 @@ mod tasks_test {
     use super::*;
 
     fn stub_tasks() -> Tasks {
-        use std::sync::mpsc;
+        use tokio::sync::mpsc;
         use nostr_sdk::Keys;
 
-        let (tx, _rx) = mpsc::channel();
+        let (tx, _rx) = mpsc::channel(16);
         Tasks::with_sender(EventSender {
             url: None,
             tx,