lavina/crates/lavina-core/src/clustering.rs

use std::collections::HashMap;
use std::net::SocketAddr;

use anyhow::{anyhow, Result};
use reqwest::Client;
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
use reqwest_tracing::{DefaultSpanBackend, TracingMiddleware};
use serde::{Deserialize, Serialize};

pub mod broadcast;
pub mod room;

type Addresses = Vec<SocketAddr>;

#[derive(Deserialize, Debug, Clone)]
pub struct ClusterConfig {
    pub metadata: ClusterMetadata,
    pub addresses: Addresses,
}

#[derive(Deserialize, Debug, Clone)]
pub struct ClusterMetadata {
    /// The node id of the current node.
    pub node_id: u32,
    /// Owns all rooms in the cluster except the ones specified in `rooms`.
    pub main_owner: u32,
    pub rooms: HashMap<String, u32>,
}

pub struct LavinaClient {
    addresses: Addresses,
    client: ClientWithMiddleware,
}

impl LavinaClient {
    pub fn new(addresses: Addresses) -> Self {
        let client = ClientBuilder::new(Client::new()).with(TracingMiddleware::<DefaultSpanBackend>::new()).build();
        Self { addresses, client }
    }

    async fn send_request(&self, node_id: u32, path: &str, req: impl Serialize) -> Result<()> {
        let Some(address) = self.addresses.get(node_id as usize) else {
            return Err(anyhow!("Unknown node"));
        };
        match self.client.post(format!("http://{}{}", address, path)).json(&req).send().await {
            Ok(res) => {
                if res.status().is_server_error() || res.status().is_client_error() {
                    tracing::error!("Cluster request failed: {:?}", res);
                    return Err(anyhow!("Server error"));
                }
                Ok(())
            }
            Err(e) => Err(e.into()),
        }
    }
}