sdk: init reconcilie method for Relay

Author: yukibtc

Cargo.lock

@@ -0,0 +1,29 @@
+// Copyright (c) 2022-2023 Yuki Kishimoto
+// Distributed under the MIT software license
+
+use nostr_sdk::prelude::*;
+
+const BECH32_SK: &str = "nsec1ufnus6pju578ste3v90xd5m2decpuzpql2295m3sknqcjzyys9ls0qlc85";
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt::init();
+
+    let secret_key = SecretKey::from_bech32(BECH32_SK)?;
+    let my_keys = Keys::new(secret_key);
+
+    let client = Client::new(&my_keys);
+    client.add_relay("wss://relay.damus.io", None).await?;
+
+    client.connect().await;
+
+    let filter = Filter::new()
+        .author(my_keys.public_key().to_string())
+        .limit(10);
+    let relay = client.relay("wss://relay.damus.io").await?;
+    relay
+        .reconcilie(filter, vec![(EventId::all_zeros(), Timestamp::now())])
+        .await?;
+
+    Ok(())
+}

crates/nostr-sdk/examples/reconcilie.rs

@@ -19,6 +19,8 @@ use std::time::Instant;
 use async_utility::futures_util::stream::AbortHandle;
 use async_utility::{futures_util, thread, time};
 use nostr::message::MessageHandleError;
+use nostr::negentropy::hex;
+use nostr::negentropy::{self, Bytes, Negentropy};
 #[cfg(feature = "nip11")]
 use nostr::nips::nip11::RelayInformationDocument;
 use nostr::{ClientMessage, Event, EventId, Filter, RelayMessage, SubscriptionId, Timestamp, Url};
@@ -41,6 +43,12 @@ type Message = (RelayEvent, Option<oneshot::Sender<bool>>);
 /// [`Relay`] error
 #[derive(Debug, Error)]
 pub enum Error {
+    /// Negentropy error
+    #[error(transparent)]
+    Negentropy(#[from] negentropy::Error),
+    /// Hex error
+    #[error(transparent)]
+    Hex(#[from] hex::Error),
     /// Channel timeout
     #[error("channel timeout")]
     ChannelTimeout,
@@ -1546,4 +1554,89 @@ impl Relay {
             }
         });
     }
+
+    /// Negentropy reconciliation
+    pub async fn reconcilie(
+        &self,
+        filter: Filter,
+        my_items: Vec<(EventId, Timestamp)>,
+    ) -> Result<(), Error> {
+        if !self.opts.read() {
+            return Err(Error::ReadDisabled);
+        }
+
+        let id_size: usize = 16;
+
+        let mut negentropy = Negentropy::new(id_size, Some(5_000))?;
+
+        for (id, timestamp) in my_items.into_iter() {
+            let cutted_id: &[u8] = &id.as_bytes()[..id_size];
+            let cutted_id = Bytes::from_slice(cutted_id);
+            negentropy.add_item(timestamp.as_u64(), cutted_id)?;
+        }
+
+        negentropy.seal()?;
+
+        let id = SubscriptionId::generate();
+        let open_msg = ClientMessage::neg_open(&mut negentropy, &id, filter)?;
+
+        self.send_msg(open_msg, Some(Duration::from_secs(10)))
+            .await?;
+
+        let mut notifications = self.notification_sender.subscribe();
+        while let Ok(notification) = notifications.recv().await {
+            if let RelayPoolNotification::Message(url, msg) = notification {
+                if url == self.url {
+                    match msg {
+                        RelayMessage::NegMsg {
+                            subscription_id,
+                            message,
+                        } => {
+                            if subscription_id == id {
+                                let query: Bytes = Bytes::from_hex(message)?;
+                                let mut need_ids: Vec<Bytes> = Vec::new();
+                                let msg: Option<Bytes> = negentropy.reconcile_with_ids(
+                                    &query,
+                                    &mut Vec::new(),
+                                    &mut need_ids,
+                                )?;
+
+                                // TODO: request ids to relay
+                                println!("IDs: {need_ids:?}");
+
+                                match msg {
+                                    Some(query) => {
+                                        self.send_msg(
+                                            ClientMessage::NegMsg {
+                                                subscription_id: id.clone(),
+                                                message: query.to_hex(),
+                                            },
+                                            None,
+                                        )
+                                        .await?;
+                                    }
+                                    None => {
+                                        tracing::info!("Reconciliation terminated");
+                                        break;
+                                    }
+                                }
+                            }
+                        }
+                        RelayMessage::NegErr {
+                            subscription_id,
+                            code,
+                        } => {
+                            if subscription_id == id {
+                                tracing::error!("Negentropy syncing error: {code}");
+                                break;
+                            }
+                        }
+                        _ => (),
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
 }

crates/nostr-sdk/src/relay/mod.rs

@@ -56,7 +56,7 @@ bip39 = { version = "2.0", default-features = false, optional = true }
 bitcoin = { version = "0.30", default-features = false, features = ["rand", "serde"] }
 cbc = { version = "0.1", optional = true }
 chacha20 = { version = "0.9", optional = true }
-negentropy = { git = "https://github.com/yukibtc/rust-negentropy", rev = "62a65db938c7088d521980d3fc1e1593713a899d", default-features = false }
+negentropy = { git = "https://github.com/yukibtc/rust-negentropy", rev = "547592168356f423dd10773384b798e7c7139893", default-features = false }
 nostr-ots = { version = "0.2", optional = true }
 once_cell = { workspace = true, optional = true }
 reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls-webpki-roots", "socks"], optional = true }