path: root/src/server.rs

use crate::resp_parser::RespType;
use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::sync::{Arc, Mutex};
use std::{env, thread};

use crate::shared_cache::SharedCache;

#[derive(Debug, Clone)]
pub struct MasterServer {
    pub dir: Option<String>,
    pub dbfilename: Option<String>,
    pub replid: Option<String>,
    pub current_offset: Option<String>,
    pub port: String,
    pub cache: SharedCache,
    replicas: Vec<SlaveServer>,
}

impl MasterServer {
    fn new() -> Self {
        Self {
            dir: None,
            dbfilename: None,
            port: "6379".to_string(),
            replid: Some("8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb".to_string()),
            current_offset: Some("0".to_string()),
            cache: Arc::new(Mutex::new(HashMap::new())),
            replicas: vec![],
        }
    }

    fn port(&self) -> &str {
        &self.port
    }

    pub fn broadcast_command(&mut self, command: &[u8]) {
        println!("Hello from brodcast");
        self.replicas.retain(|replica| {
            if let Some(conn) = &replica.connection {
                let mut conn = conn.lock().unwrap();
                if let Err(e) = conn.write_all(command) {
                    eprintln!("Failed to send to replica {}: {}", replica.port, e);
                    false // Drop dead connections
                } else {
                    true
                }
            } else {
                false
            }
        });
    }
}

#[derive(Debug, Clone)]
pub struct SlaveServer {
    pub dir: Option<String>,
    pub dbfilename: Option<String>,
    pub port: String,
    pub master_replid: Option<String>,
    pub master_repl_offset: Option<String>,
    pub master_host: String,
    pub master_port: String,
    pub connection: Option<Arc<Mutex<TcpStream>>>,
    pub cache: SharedCache,
}

impl SlaveServer {
    fn new(
        port: String,
        master_host: String,
        master_port: String,
        connection: Option<Arc<Mutex<TcpStream>>>,
    ) -> Self {
        Self {
            dir: None,
            dbfilename: None,
            port,
            master_replid: Some("8371b4fb1155b71f4a04d3e1bc3e18c4a990aeeb".to_string()),
            master_repl_offset: Some("0".to_string()),
            master_host,
            master_port,
            connection,
            cache: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    fn connect(&self) -> Result<TcpStream, std::io::Error> {
        let master_address = format!("{}:{}", self.master_host, self.master_port);
        return TcpStream::connect(master_address);
    }

    fn handshake(&mut self) -> Result<(), String> {
        match self.connect() {
            Ok(mut stream) => {
                let mut buffer = [0; 512];

                let mut send_command = |command: &[u8]| -> Result<(), String> {
                    stream
                        .write_all(command)
                        .map_err(|e| format!("Failed to send: {}", e))?;

                    match stream.read(&mut buffer) {
                        Ok(0) | Err(_) => return Ok(()), // connection closed or error
                        Ok(_) => Ok(()),
                    }
                };

                // PING
                send_command(&resp_bytes!(array => [resp!(bulk "PING")]))?;

                // REPLCONF listening-port <PORT>
                send_command(&resp_bytes!(array => [
                    resp!(bulk "REPLCONF"),
                    resp!(bulk "listening-port"),
                    resp!(bulk self.port.clone())
                ]))?;

                // REPLCONF capa psync2
                send_command(&resp_bytes!(array => [
                    resp!(bulk "REPLCONF"),
                    resp!(bulk "capa"),
                    resp!(bulk "psync2")
                ]))?;

                // PSYNC <REPL_ID> <REPL_OFFSSET>
                send_command(&resp_bytes!(array => [
                    resp!(bulk "PSYNC"),
                    resp!(bulk "?"),
                    resp!(bulk "-1")
                ]))?;

                // Store the persistent connection
                let shared_stream = Arc::new(Mutex::new(stream));
                self.connection = Some(shared_stream.clone());

                // Spawn the background listener thread
                thread::spawn(move || {
                    let mut buffer = [0u8; 1024];
                    loop {
                        let mut stream = shared_stream.lock().unwrap();
                        match stream.read(&mut buffer) {
                            Ok(0) => {
                                println!("Master disconnected");
                                break;
                            }
                            Ok(n) => {
                                println!(
                                    "REPLICA received: {}",
                                    String::from_utf8_lossy(&buffer[..n])
                                );
                            }
                            Err(e) => {
                                eprintln!("Error reading from master: {}", e);
                                break;
                            }
                        }
                    }
                });

                Ok(())
            }
            Err(e) => Err(format!("Master node doesn't exist: {}", e)),
        }
    }
}

#[derive(Debug, Clone)]
pub enum RedisServer {
    Master(MasterServer),
    Slave(SlaveServer),
}

impl RedisServer {
    pub fn master() -> Self {
        RedisServer::Master(MasterServer::new())
    }

    pub fn slave(port: String, master_host: String, master_port: String) -> Self {
        RedisServer::Slave(SlaveServer::new(port, master_host, master_port, None))
    }

    // Helper methods to access common fields regardless of variant
    pub fn port(&self) -> &str {
        match self {
            RedisServer::Master(master) => &master.port,
            RedisServer::Slave(slave) => &slave.port,
        }
    }

    pub fn set_port(&mut self, port: String) {
        match self {
            RedisServer::Master(master) => master.port = port,
            RedisServer::Slave(slave) => slave.port = port,
        }
    }

    pub fn dir(&self) -> &Option<String> {
        match self {
            RedisServer::Master(master) => &master.dir,
            RedisServer::Slave(slave) => &slave.dir,
        }
    }

    pub fn set_dir(&mut self, dir: Option<String>) {
        match self {
            RedisServer::Master(master) => master.dir = dir,
            RedisServer::Slave(slave) => slave.dir = dir,
        }
    }

    pub fn dbfilename(&self) -> &Option<String> {
        match self {
            RedisServer::Master(master) => &master.dbfilename,
            RedisServer::Slave(slave) => &slave.dbfilename,
        }
    }

    pub fn set_dbfilename(&mut self, dbfilename: Option<String>) {
        match self {
            RedisServer::Master(master) => master.dbfilename = dbfilename,
            RedisServer::Slave(slave)