sneedstr/src/bussy/mod.rs

use crate::{
    noose::user::{User, UserRow},
    noose::sled::BanInfo,
    utils::{error::Error, structs::Subscription},
};
use nostr::secp256k1::XOnlyPublicKey;
use std::{collections::HashMap, fmt::Debug};
use tokio::sync::{broadcast, Mutex};

pub mod channels {
    pub static MSG_NOOSE: &str = "MSG_NOOSE";
    pub static MSG_NIP05: &str = "MSG_NIP05";
    pub static MSG_RELAY: &str = "MSG_RELAY";
    pub static MSG_PIPELINE: &str = "MSG_PIPELINE";
    pub static MSG_SLED: &str = "MSG_SLED";
}

#[derive(Debug, Clone, PartialEq)]
pub enum Command {
    // DbRequest
    DbReqWriteEvent(/* client_id */ uuid::Uuid, Box<nostr::Event>),
    DbReqFindEvent(/* client_id*/ uuid::Uuid, Subscription),
    DbReqDeleteEvents(/* client_id*/ uuid::Uuid, Box<nostr::Event>),
    DbReqEventCounts(/* client_id*/ uuid::Uuid, Subscription),

    // Old messages
    DbReqInsertUser(UserRow),
    DbReqGetUser(User),
    DbReqCreateAccount(XOnlyPublicKey, String, String),
    DbReqGetAccount(String),
    DbReqClear,
    // DbResponse
    DbResRelayMessages(
        /* client_id*/ uuid::Uuid,
        /* Vec<RelayMessage::Event> */ Vec<nostr::RelayMessage>,
    ),
    DbResInfo,
    DbResOk,
    DbResOkWithStatus(/* client_id */ uuid::Uuid, nostr::RelayMessage),
    DbResAccount, // TODO: Add Account DTO as a param
    DbResUser(UserRow),
    DbResEventCounts(/* client_id */ uuid::Uuid, nostr::RelayMessage),
    // Event Pipeline
    PipelineReqEvent(/* client_id */ uuid::Uuid, Box<nostr::Event>),
    PipelineResRelayMessageOk(/* client_id */ uuid::Uuid, nostr::RelayMessage),
    PipelineResStreamOutEvent(Box<nostr::Event>),
    PipelineResOk,
    // Subscription Errors
    ClientSubscriptionError(/* error message */ String),
    // Sled
    SledReqBanUser(Box<BanInfo>),
    SledReqBanInfo(/* pubkey */ String),
    SledReqUnbanUser(/* pubkey */ String),
    SledReqGetBans,
    SledResBan(Option<BanInfo>),
    SledResBans(Vec<BanInfo>),
    SledResSuccess(bool),
    // Other
    Str(String),
    ServiceError(Error),
    Noop,
}

#[derive(Debug, Clone)]
pub struct Message {
    pub source: &'static str,
    pub content: Command,
}

#[derive(Debug)]
pub struct PubSub {
    subscribers: Mutex<HashMap<String, Vec<broadcast::Sender<Message>>>>,
}

impl Default for PubSub {
    fn default() -> Self {
        panic!("Use PubSub::new() to initialize PubSub");
    }
}

impl PubSub {
    pub fn new() -> Self {
        PubSub {
            subscribers: Mutex::new(HashMap::new()),
        }
    }

    pub async fn subscribe(&self, topic: &str) -> broadcast::Receiver<Message> {
        let (tx, _rx) = broadcast::channel(32); // 32 is the channel capacity
        let mut subscribers = self.subscribers.lock().await;
        subscribers
            .entry(topic.to_string())
            .or_insert_with(Vec::new)
            .push(tx.clone());
        tx.subscribe()
    }

    pub async fn publish(&self, topic: &str, message: Message) {
        let mut subscribers = self.subscribers.lock().await;
        if let Some(queue) = subscribers.get_mut(topic) {
            for sender in queue.iter() {
                sender.send(message.clone()).ok();
            }
        }
    }
}

// #[cfg(test)]
// mod tests {
//     use super::channels;
//     use crate::bussy::{Command, Message, PubSub};
//     use std::sync::Arc;

//     #[tokio::test]
//     async fn create_bus() {
//         let pubsub = Arc::new(PubSub::new());

//         let mut subscriber1 = pubsub.subscribe(channels::MSG_NIP05).await;
//         let mut subscriber2 = pubsub.subscribe(channels::MSG_NOOSE).await;

//         tokio::spawn(async move {
//             while let Ok(message) = subscriber1.recv().await {
//                 println!("Subscriber1 received: {:?}", message);
//             }
//         });

//         tokio::spawn(async move {
//             while let Ok(message) = subscriber2.recv().await {
//                 println!("Subscriber2 received: {:?}", message);
//             }
//         });

//         pubsub
//             .publish(
//                 channels::MSG_NIP05,
//                 Message {
//                     source: "test",
//                     content: Command::Str("Hello S1".to_string()),
//                 },
//             )
//             .await;
//         pubsub
//             .publish(
//                 channels::MSG_NOOSE,
//                 Message {
//                     source: "test",
//                     content: Command::Str("Hello S2".to_string()),
//                 },
//             )
//             .await;

//         dbg!(pubsub);

//         // Sleep to keep the main thread alive while messages are processed in the background.
//         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
//     }
// }