filter: add forwarder filter

Fixes: damus-io#10
Changelog-Added: Add forwarder filter
Signed-off-by: William Casarin <[email protected]>

Author: jb55

Makefile

@@ -1,2 +1,4 @@
 tags:
 	find src -name '*.rs' | xargs ctags 
+
+.PHONY: tags

README.md

@@ -131,6 +131,18 @@ There are no config options, but an empty config entry is still needed:
 
 `[filters.protected_events]`
 
+### Forwarder
+
+* name: `forwarder`
+
+The forwarder filter allows you to forward notes to another relay. Notes will
+be queued if the connection goes down (up to the `queue_size` buffer limit)
+
+- `relay` - the relay to forward notes to, eg: `ws://localhost:8080`
+
+- `queue_size` *optional* - size of the note queue, this is used to buffer notes if the connection goes down
+
+
 ## Testing
 
 You can test your filters like so:

noteguard-forwarder.toml

@@ -0,0 +1,8 @@
+
+pipeline = ["ratelimit", "forwarder"]
+
+[filters.forwarder]
+relay = "ws://127.0.0.1:8080"
+
+[filters.ratelimit]
+posts_per_minute = 3

src/filters/forwarder.rs

@@ -0,0 +1,107 @@
+use serde::Deserialize;
+use crate::{Note, Action, NoteFilter, InputMessage, OutputMessage};
+use futures_util::{SinkExt, StreamExt};
+use tokio::sync::mpsc::{self, Sender, Receiver};
+use tokio_tungstenite::connect_async;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::WebSocketStream;
+use tokio::time::{sleep, timeout, Duration};
+use serde_json::json;
+use log::{error, info, debug};
+
+#[derive(Default, Deserialize)]
+pub struct Forwarder {
+    relay: String,
+
+    /// the size of our bounded queue
+    queue_size: Option<u32>,
+
+    /// The channel used for communicating with the forwarder thread
+    #[serde(skip)]
+    channel: Option<Sender<Note>>,
+}
+
+async fn client_reconnect(relay: &str) -> WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>> {
+    loop {
+        match connect_async(relay).await {
+            Err(e) => {
+                error!("failed to connect to relay {}: {}", relay, e);
+                sleep(Duration::from_secs(5)).await;
+                continue;
+            }
+            Ok((ws, _)) => {
+                info!("connected to relay: {}", relay);
+                return ws;
+            }
+        }
+    }
+}
+
+async fn forwarder_task(relay: String, mut rx: Receiver<Note>) {
+    let stream = client_reconnect(&relay).await;
+    let (mut writer, mut reader) = stream.split();
+
+    loop {
+        tokio::select! {
+            result = timeout(Duration::from_secs(10), rx.recv()) => {
+                match result {
+                    Ok(Some(note)) => {
+                        if let Err(e) = writer.send(Message::Text(serde_json::to_string(&json!(["EVENT", note])).unwrap())).await {
+                            error!("got error: '{}', reconnecting...", e);
+                            let (w, r) = client_reconnect(&relay).await.split();
+                            writer = w;
+                            reader = r;
+                        }
+                    },
+                    Ok(None) => {
+                        // Channel has been closed, exit the loop
+                        error!("channel closed, stopping forwarder_task");
+                        break;
+                    }
+                    Err(_) => {
+                        // Timeout occurred, send a ping
+                        // try reading for pongs, etc
+                        let _r = reader.next();
+                        debug!("timeout reading note queue, sending ping");
+
+                        if let Err(e) = writer.send(Message::Ping(vec![])).await {
+                            error!("error during ping ({}), reconnecting...", e);
+                            let (w, r) = client_reconnect(&relay).await.split();
+                            writer = w;
+                            reader = r;
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl NoteFilter for Forwarder {
+    fn name(&self) -> &'static str {
+        "forwarder"
+    }
+
+    fn filter_note(&mut self, input: &InputMessage) -> OutputMessage {
+        if self.channel.is_none() {
+            let (tx, rx) = mpsc::channel(self.queue_size.unwrap_or(1000) as usize);
+            let relay = self.relay.clone();
+
+            tokio::task::spawn(async move {
+                forwarder_task(relay, rx).await;
+            });
+
+            self.channel = Some(tx);
+        }
+
+        // Add code to process input and send through channel
+        if let Some(ref channel) = self.channel {
+            if let Err(e) = channel.try_send(input.event.clone()) {
+                eprintln!("could not forward note: {}", e);
+            }
+        }
+
+        // Create and return an appropriate OutputMessage
+        OutputMessage::new(input.event.id.clone(), Action::Accept, None)
+    }
+}

src/filters/mod.rs

@@ -3,7 +3,13 @@ mod protected_events;
 mod ratelimit;
 mod whitelist;
 
+#[cfg(feature = "forwarder")]
+mod forwarder;
+
 pub use kinds::Kinds;
 pub use protected_events::ProtectedEvents;
 pub use ratelimit::RateLimit;
 pub use whitelist::Whitelist;
+
+#[cfg(feature = "forwarder")]
+pub use forwarder::Forwarder;

src/main.rs

@@ -1,9 +1,14 @@
 use noteguard::filters::{Kinds, ProtectedEvents, RateLimit, Whitelist};
+
+#[cfg(feature = "forwarder")]
+use noteguard::filters::Forwarder;
+
 use noteguard::{Action, InputMessage, NoteFilter, OutputMessage};
 use serde::de::DeserializeOwned;
 use serde::Deserialize;
 use std::collections::HashMap;
 use std::io::{self, BufRead, Read, Write};
+use log::info;
 
 #[derive(Deserialize)]
 struct Config {
@@ -44,6 +49,9 @@ impl Noteguard {
         self.register_filter::<Whitelist>();
         self.register_filter::<ProtectedEvents>();
         self.register_filter::<Kinds>();
+
+        #[cfg(feature = "forwarder")]
+        self.register_filter::<Forwarder>();
     }
 
     /// Run the loaded filters. You must call `load_config` before calling this, otherwise
@@ -94,7 +102,21 @@ impl Noteguard {
     }
 }
 
+#[cfg(feature = "forwarder")]
+#[tokio::main]
+async fn main() {
+    noteguard();
+}
+
+#[cfg(not(feature = "forwarder"))]
 fn main() {
+    noteguard();
+}
+
+fn noteguard() {
+    env_logger::init();
+    info!("running noteguard");
+
     let config_path = "noteguard.toml";
     let mut noteguard = Noteguard::new();
 

src/note_filter.rs

@@ -1,7 +1,7 @@
 use crate::{InputMessage, OutputMessage};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Serialize, Clone)]
 pub struct Note {
     pub id: String,
     pub pubkey: String,

test/delayed-nostril

@@ -0,0 +1,10 @@
+#!/usr/bin/env sh
+
+while true
+do
+	note="$(nostril --silent --content hello)"
+	echo "{\"type\": \"new\",\"receivedAt\":12345,\"sourceType\":\"IP4\",\"sourceInfo\": \"127.0.0.2\",\"event\":$note}"
+
+	sleep ${1:-0.1}
+done
+