aljaz/blastr
1
0
Fork 0
mirror of https://github.com/aljazceru/blastr.git synced 2025-12-19 06:54:24 +01:00
Code Issues Packages Projects Releases Wiki Activity

Don't broadcast events more than once

Browse Source
This commit is contained in:
Tony Giorgio
2023-02-18 23:10:45 -06:00
parent 0fca0de944
commit de5d42d62f
4 changed files with 132 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.toml
View File

@@ -15,6 +15,7 @@ cfg-if = "0.1.2"
worker = { git = "https://github.com/cloudflare/workers-rs", rev = "b4b9cd1f15feac412b6f9e9e9209458cd3b98430", features = ["queue"] }
futures = "0.3.26"
nostr = { git = "https://github.com/rust-nostr/nostr", rev = "c333544b1e58b89acb786695d5ca08759ebf5e69" }
serde = { version = "^1.0", features = ["derive"] }
# The `console_error_panic_hook` crate provides better debugging of panics by
# logging them with `console.error`. This is great for development, but requires

108
src/lib.rs
View File

@@ -1,9 +1,10 @@
pub(crate) use crate::nostr::{
try_queue_client_msg, NOSTR_QUEUE, NOSTR_QUEUE_2, NOSTR_QUEUE_3, NOSTR_QUEUE_4, NOSTR_QUEUE_5,
try_queue_event, NOSTR_QUEUE, NOSTR_QUEUE_2, NOSTR_QUEUE_3, NOSTR_QUEUE_4, NOSTR_QUEUE_5,
NOSTR_QUEUE_6,
};
use ::nostr::{ClientMessage, Event};
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use worker::*;
mod error;
@@ -18,6 +19,11 @@ fn log_request(req: &Request) {
);
}
#[derive(Serialize, Deserialize, Clone)]
pub struct PublishedNote {
date: String,
}
/// Main function for the Cloudflare Worker that triggers off of a HTTP req
#[event(fetch)]
pub async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
@@ -40,6 +46,25 @@ pub async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
match req.text().await {
Ok(request_text) => {
if let Ok(client_msg) = ClientMessage::from_json(request_text) {
match client_msg {
ClientMessage::Event(event) => {
console_log!("got an event from client: {}", event.id);
// check if we've already published it before
let published_notes = ctx.kv("PUBLISHED_NOTES")?;
match published_notes
.get(event.id.to_string().as_str())
.json::<PublishedNote>()
.await?
{
Some(_) => {
console_log!("event already published: {}", event.id);
return fetch();
}
None => {}
};
// broadcast it to all queues
let nostr_queues = vec![
ctx.env.queue(NOSTR_QUEUE).expect("get queue"),
ctx.env.queue(NOSTR_QUEUE_2).expect("get queue"),
@@ -48,7 +73,33 @@ pub async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
ctx.env.queue(NOSTR_QUEUE_5).expect("get queue"),
ctx.env.queue(NOSTR_QUEUE_6).expect("get queue"),
];
try_queue_client_msg(client_msg, nostr_queues).await
try_queue_event(*event.clone(), nostr_queues).await;
console_log!("queued up nostr event: {}", event.id);
match published_notes
.put(
event.id.to_string().as_str(),
PublishedNote {
date: Date::now().to_string(),
},
)?
.execute()
.await
{
Ok(_) => {
console_log!("saved published note: {}", event.id);
}
Err(e) => {
console_log!(
"could not save published note: {} - {e:?}",
event.id
);
}
}
}
_ => {
console_log!("ignoring other nostr client message types");
}
}
}
}
Err(e) => {
@@ -74,6 +125,30 @@ pub async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
continue;
};
if let Ok(client_msg) = ClientMessage::from_json(msg.text().unwrap()) {
match client_msg {
ClientMessage::Event(event) => {
console_log!("got an event from client: {}", event.id);
// check if we've already published it before
let published_notes =
ctx.kv("PUBLISHED_NOTES").expect("get kv");
match published_notes
.get(event.id.to_string().as_str())
.json::<PublishedNote>()
.await
.expect("lookup value")
{
Some(_) => {
console_log!(
"event already published: {}",
event.id
);
continue;
}
None => {}
};
// broadcast it to all queues
let nostr_queues = vec![
ctx.env.queue(NOSTR_QUEUE).expect("get queue"),
ctx.env.queue(NOSTR_QUEUE_2).expect("get queue"),
@@ -82,7 +157,34 @@ pub async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
ctx.env.queue(NOSTR_QUEUE_5).expect("get queue"),
ctx.env.queue(NOSTR_QUEUE_6).expect("get queue"),
];
try_queue_client_msg(client_msg, nostr_queues).await
try_queue_event(*event.clone(), nostr_queues).await;
console_log!("queued up nostr event: {}", event.id);
match published_notes
.put(
event.id.to_string().as_str(),
PublishedNote {
date: Date::now().to_string(),
},
)
.expect("saved note")
.execute()
.await
{
Ok(_) => {
console_log!("saved published note: {}", event.id);
}
Err(e) => {
console_log!(
"could not save published note: {} - {e:?}",
event.id
);
}
}
}
_ => {
console_log!("ignoring other nostr client message types");
}
}
}
}
WebsocketEvent::Close(_) => {

13
src/nostr.rs
View File

@@ -22,24 +22,15 @@ const RELAYS: [&str; 8] = [
"wss://nostr.wine",
];
pub async fn try_queue_client_msg(client_msg: ClientMessage, nostr_queues: Vec<Queue>) {
match client_msg {
ClientMessage::Event(event) => {
console_log!("got an event from client: {}", event.id);
pub async fn try_queue_event(event: Event, nostr_queues: Vec<Queue>) {
for nostr_queue in nostr_queues.iter() {
match queue_nostr_event_with_queue(nostr_queue, *event.clone()).await {
match queue_nostr_event_with_queue(nostr_queue, event.clone()).await {
Ok(_) => {}
Err(Error::WorkerError(e)) => {
console_log!("worker error: {e}");
}
}
}
console_log!("queued up nostr event: {}", event.id)
}
_ => {
console_log!("ignoring other nostr client message types");
}
}
}
pub async fn queue_nostr_event_with_queue(nostr_queue: &Queue, event: Event) -> Result<(), Error> {

4
wrangler.toml
View File

@@ -8,6 +8,10 @@ routes = [
{ pattern = "nostr.mutinywallet.com/event", zone_id = "2b9268714ce8d1c4431e8046d4ba55d3" }
]
kv_namespaces = [
{ binding = "PUBLISHED_NOTES", id = "afa24a392a5a41f6b1655507dfd9b97a", preview_id = "0b334aece8d74c3ab90e3e99db569ce8" }
]
[vars]
WORKERS_RS_VERSION = "0.0.11"