rship-sdk 0.1.19

use std::{collections::HashMap, sync::Arc};

use crate::instance::{InstanceArgs, InstanceProxy};
use myko_rs::{
    client::{ConnectionStatus, MykoClient},
    event::{MEvent, MEventType},
    item::Eventable,
};

use rship_entities::pulse::Pulse;
use schemars::JsonSchema;
use serde::Serialize;
use tokio::sync::Mutex;
use tokio_stream::StreamExt;
use uuid::Uuid;

#[derive(Clone)]
pub struct SdkClient {
    pub(crate) client: MykoClient,
    pub(crate) instances: Arc<Mutex<HashMap<String, InstanceProxy>>>,
}

impl SdkClient {
    pub fn new(client: MykoClient) -> Self {
        let instances = Arc::new(Mutex::new(HashMap::<String, InstanceProxy>::new()));

        let client_ref = client.clone();
        let instances_ref = instances.clone();

        tokio::spawn(async move {
            let mut connction_status = client_ref.watch_connection_status();
            while let Some(msg) = connction_status.next().await {
                if let ConnectionStatus::Connected(_) = &msg {
                    let instances = instances_ref.lock().await;

                    for instance in instances.iter() {
                        instance.1.save().await;
                    }
                }
            }
        });

        Self { client, instances }
    }

    pub async fn add_instance(&self, args: InstanceArgs) -> InstanceProxy {
        let p = InstanceProxy {
            args,
            client: self.clone(),
        };

        p.save().await;

        self.instances
            .lock()
            .await
            .insert(p.clone().args.short_id, p.clone());

        p
    }

    pub(crate) async fn save_item<T: Eventable<T, PT> + PartialEq, PT: Clone>(
        &self,
        item: &T,
    ) -> Result<(), String> {
        let event = MEvent::from_item(item, MEventType::SET, Uuid::new_v4().to_string());

        self.client.send_event(event).await
    }

    pub(crate) async fn delete_item<T: Eventable<T, PT> + PartialEq, PT: Clone>(
        &self,
        item: &T,
    ) -> Result<(), String> {
        let event = MEvent::from_item(item, MEventType::DEL, Uuid::new_v4().to_string());

        self.client.send_event(event).await
    }

    pub async fn pulse_emitter(
        &self,
        service_short_id: String,
        target_short_id: String,
        emitter_short_id: String,
        data: impl JsonSchema + Serialize,
    ) -> Result<(), String> {
        let full_emitter_id = format!("{service_short_id}:{target_short_id}:{emitter_short_id}");

        let pulse = Pulse::new(full_emitter_id.clone(), serde_json::to_value(data).unwrap());

        let event = MEvent::from_item(&pulse, MEventType::SET, uuid::Uuid::new_v4().to_string());

        self.client.send_event(event).await
    }

    pub async fn await_connection(&self) {
        loop {
            let connection_status = self.client.get_connection_status().await;
            match connection_status {
                ConnectionStatus::Connected(_) => {
                    break;
                }
                _ => {
                    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                }
            }
        }
        log::info!("Client Connected");
    }

    pub fn init() -> Self {
        Self::new(MykoClient::new())
    }

    pub fn set_address(&self, address: Option<String>) {
        self.client.set_address(address);
    }
}