WIP: Add consecutive filters

.gitignore

@@ -20,3 +20,4 @@ build/
 /target
 /dist
 .dev.vars
+.wrangler

src/lib.rs

@@ -12,7 +12,10 @@ use ::nostr::{
 };
 use futures::StreamExt;
 use serde::{Deserialize, Serialize};
-use std::sync::Arc;
+use std::{
+    ops::Deref,
+    sync::{Arc, Mutex},
+};
 use worker::*;
 
 mod error;
@@ -181,6 +184,7 @@ pub async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
             server.accept()?;
             console_log!("accepted websocket, about to spawn event stream");
             wasm_bindgen_futures::spawn_local(async move {
+                let requested_filters = Arc::new(Mutex::new(Filter::new()));
                 let mut event_stream = server.events().expect("stream error");
                 console_log!("spawned event stream, waiting for first message..");
                 while let Some(event) = event_stream.next().await {
@@ -324,34 +328,53 @@ pub async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
                                             };
 
                                             if valid_nwc {
-                                                let ctx_clone = ctx.clone();
-                                                let sub_id = subscription_id.clone();
-                                                let server_clone = server.clone();
-                                                wasm_bindgen_futures::spawn_local(async move {
-                                                    console_log!("Got NWC filter!");
-                                                    let mut sent_events = vec![];
-                                                    loop {
-                                                        match handle_filter(
-                                                            &sent_events,
-                                                            sub_id.clone(),
-                                                            filter.clone(),
-                                                            &server_clone,
-                                                            &ctx_clone,
-                                                        )
-                                                            .await
-                                                        {
-                                                            Ok(new_event_ids) => {
-                                                                // add new events to sent events
-                                                                sent_events.extend(new_event_ids);
+                                                let master_guard = requested_filters.lock().unwrap();
+                                                let master_filter = master_guard.deref();
+                                                let fresh_filter = Filter::new();
+                                                let was_empty = *master_filter == fresh_filter;
+
+                                                // now add the new filters to the main filter
+                                                // object. This is a bit of a hack but we only
+                                                // check certain sub filters for NWC.
+                                                add_to_master_filter(master_filter, &filter);
+                                                drop(master_guard);
+
+                                                // only spin up a new one if there's not a
+                                                // spawn_local already going with filters
+                                                // when other filters are added in, it should 
+                                                // be picked up in the master filter 
+                                                if was_empty {
+                                                    let ctx_clone = ctx.clone();
+                                                    let sub_id = subscription_id.clone();
+                                                    let server_clone = server.clone();
+                                                    let master_clone = requested_filters.clone();
+                                                    wasm_bindgen_futures::spawn_local(async move {
+                                                        console_log!("Got NWC filter!");
+                                                        let mut sent_events = vec![];
+                                                        loop {
+                                                            console_log!("Looping through filter handling...");
+                                                            match handle_filter(
+                                                                &sent_events,
+                                                                sub_id.clone(),
+                                                                master_clone.clone(),
+                                                                &server_clone,
+                                                                &ctx_clone,
+                                                            )
+                                                                .await
+                                                            {
+                                                                Ok(new_event_ids) => {
+                                                                    // add new events to sent events
+                                                                    sent_events.extend(new_event_ids);
+                                                                }
+                                                                Err(e) => console_log!(
+                                                                    "error handling filter: {e}"
+                                                                ),
                                                             }
-                                                            Err(e) => console_log!(
-                                                            "error handling filter: {e}"
-                                                        ),
-                                                        }
 
-                                                        utils::delay(10_000).await;
-                                                    }
-                                                });
+                                                            utils::delay(10_000).await;
+                                                        }
+                                                    });
+                                                }
                                             }
                                         } {
                                             // if not a nwc filter, we just send EOSE
@@ -637,6 +660,57 @@ pub async fn delete_nwc_response(event: &Event, ctx: &RouteContext<()>) -> Resul
     Ok(())
 }
 
+fn add_to_master_filter(master_filter: &Filter, new_filter: &Filter) {
+    if let Some(vec) = &new_filter.ids {
+        master_filter
+            .ids
+            .get_or_insert_with(Vec::new)
+            .extend(vec.clone());
+    }
+    if let Some(vec) = &new_filter.authors {
+        master_filter
+            .authors
+            .get_or_insert_with(Vec::new)
+            .extend(vec.clone());
+    }
+    if let Some(vec) = &new_filter.kinds {
+        master_filter
+            .kinds
+            .get_or_insert_with(Vec::new)
+            .extend(vec.clone());
+    }
+    if let Some(vec) = &new_filter.events {
+        master_filter
+            .events
+            .get_or_insert_with(Vec::new)
+            .extend(vec.clone());
+    }
+    if let Some(vec) = &new_filter.pubkeys {
+        master_filter
+            .pubkeys
+            .get_or_insert_with(Vec::new)
+            .extend(vec.clone());
+    }
+    if let Some(vec) = &new_filter.hashtags {
+        master_filter
+            .hashtags
+            .get_or_insert_with(Vec::new)
+            .extend(vec.clone());
+    }
+    if let Some(vec) = &new_filter.references {
+        master_filter
+            .references
+            .get_or_insert_with(Vec::new)
+            .extend(vec.clone());
+    }
+    if let Some(vec) = &new_filter.identifiers {
+        master_filter
+            .identifiers
+            .get_or_insert_with(Vec::new)
+            .extend(vec.clone());
+    }
+}
+
 fn relay_response(msg: RelayMessage) -> worker::Result<Response> {
     Response::from_json(&msg)?.with_cors(&cors())
 }