righvalor 0.1.0

use std::sync::Arc;

use ractor::{concurrency::Duration, Actor, ActorProcessingErr, ActorRef};
use ractor_cluster::NodeServerMessage;
use righ_dm_rs::RighIpv4Addr;
use tokio::sync::RwLock;

use crate::{
    common::task::ValorTaskId,
    config::ValorConfig,
    master::{ValorMasterMessage, ValorTaskManager, ValorWorkerRegistry},
    types::{ValorID, ValorIdExt},
    worker::{
        CapacitySnapshot, ServicesSnapshot, ValorWorkerEvent, ValorWorkerMessage, WorkerHeartbeat,
    },
};

/// # Valor Master Actor
///
/// Core actor implementation of the Valor Master within the RighValor Framework.
/// The Master serves as the central coordinator for distributed computational
/// infrastructure, managing worker nodes and orchestrating task execution.
///
/// ## Master Responsibilities
///
/// - **Worker Orchestration**: Register, monitor, and coordinate Valor Workers
/// - **Task Distribution**: Assign computational tasks based on worker capabilities  
/// - **Service Management**: Control service lifecycle across the distributed network
/// - **Resource Optimization**: Optimize task placement and resource utilization
/// - **Health Monitoring**: Monitor system health and perform automatic recovery
///
/// ## Communication Pattern
///
/// The master implements southbound communication with Valor Workers for:
/// - Worker registration and capability discovery
/// - Task assignment and execution monitoring
/// - Service management command distribution
/// - Health check coordination and status collection
pub struct Master;

/// # Valor Master State
///
/// Maintains the runtime state of the Valor Master including its identity,
/// network configuration, and operational status within the distributed framework.
pub struct ValorMasterState {
    #[allow(unused)]
    id: ValorID,
    node_server: ActorRef<NodeServerMessage>,
    config: ValorConfig,

    // Centralized worker registry
    registry: Arc<RwLock<ValorWorkerRegistry>>,

    // Task manager
    task_manager: Arc<ValorTaskManager>,
}

impl ValorMasterState {
    /// Connect to a worker with retry mechanism and background processing
    pub async fn connect_worker(&self, worker_id: ValorID, ipv4_addr: RighIpv4Addr) {
        // Spawn connection attempt in background task
        let node_server = self.node_server.clone();
        let config = self.config.clone();

        let span = tracing::info_span!(
            "flow.master.connect_worker",
            worker_id = %worker_id,
            ipv4 = %ipv4_addr
        );
        tokio::spawn({
            let span = span.clone();
            async move {
                let _e = span.enter();
                Self::connect_worker_with_retry(node_server, worker_id, ipv4_addr, config).await;
            }
        });
    }

    /// Internal method with retry and fallback logic
    async fn connect_worker_with_retry(
        node_server: ActorRef<NodeServerMessage>,
        worker_id: ValorID,
        ipv4_addr: RighIpv4Addr,
        config: ValorConfig,
    ) {
        const MAX_RETRIES: u32 = 3;
        const BASE_DELAY_MS: u64 = 100;

        let primary_address = format!("{}:{}", ipv4_addr, config.app.worker_port());
        let fallback_address = format!("127.0.0.1:{}", config.app.worker_port());

        tracing::info!(
            "Starting connection attempt to worker {}@{} (fallback: {} )",
            worker_id,
            primary_address,
            fallback_address
        );

        // Try primary address first
        for attempt in 1..=MAX_RETRIES {
            match ractor_cluster::node::client::connect(&node_server, primary_address.clone()).await
            {
                Ok(_) => {
                    tracing::info!(
                        "Successfully connected to worker {}@{} on attempt {}",
                        worker_id,
                        ipv4_addr,
                        attempt
                    );

                    // After successful TCP connection, we can now send messages to the worker
                    // using its stable actor name. The worker actor is already registered
                    // with the name from config.app.worker_node_name(&worker_id.to_string())
                    let worker_actor_name = worker_id.worker_actor_name();
                    tracing::info!(
                        "Worker {} is now reachable via actor name '{}'",
                        worker_id,
                        worker_actor_name
                    );

                    // Send confirmation via shared workers PG with retry until worker joins
                    for attempt in 1..=10u32 {
                        let sent = {
                            let pg_name = ValorID::workers_pg_name();
                            if let Some(cell) = ractor::pg::get_members(&pg_name).into_iter().next()
                            {
                                let worker_ref: ActorRef<ValorWorkerMessage> = cell.into();
                                let _ = worker_ref
                                    .cast(ValorWorkerMessage::NorthboundRegisterMasterConfirmed);
                                true
                            } else {
                                false
                            }
                        };
                        if sent {
                            tracing::info!(
                                "Sent register-confirm to worker {} on attempt {}",
                                worker_id,
                                attempt
                            );
                            break;
                        } else {
                            ractor::concurrency::sleep(Duration::from_millis(200)).await;
                        }
                    }
                    return;
                }
                Err(error) => {
                    if attempt >= MAX_RETRIES {
                        tracing::warn!(
                            "Failed connecting to worker {} after {} attempts: {}",
                            worker_id,
                            attempt,
                            error
                        );
                    }
                    // Try fallback address after failure
                    tracing::info!(
                        "Attempting fallback connection to worker {}@{}",
                        worker_id,
                        fallback_address
                    );
                    if let Err(error) = ractor_cluster::node::client::connect(
                        &node_server,
                        fallback_address.clone(),
                    )
                    .await
                    {
                        tracing::warn!(
                            "Fallback connection failed for worker {}: {}",
                            worker_id,
                            error
                        );
                        // On connection closed, proactively mark worker as unreachable and trigger requeue
                        // (PG will also remove the member asynchronously.)
                        // Note: We do not hold registry here; this is only a log hook.
                        tracing::info!(
                            "Worker {} connection closed; PG will remove member soon",
                            worker_id
                        );
                        let delay_ms = BASE_DELAY_MS.saturating_mul(2u64.pow(attempt - 1));
                        ractor::concurrency::sleep(Duration::from_millis(delay_ms)).await;
                        continue;
                    } else {
                        tracing::info!(
                            "Fallback connection succeeded to worker {}@{}",
                            worker_id,
                            fallback_address
                        );
                        Self::send_to_worker(
                            &worker_id,
                            ValorWorkerMessage::NorthboundRegisterMasterConfirmed,
                        )
                        .await;
                        return;
                    }
                }
            }
        }
    }

    async fn send_to_worker(id: &ValorID, msg: ValorWorkerMessage) {
        let pg_name = ValorID::workers_pg_name();
        if let Some(cell) = ractor::pg::get_members(&pg_name).into_iter().next() {
            let worker_ref: ActorRef<ValorWorkerMessage> = cell.into();
            let _ = worker_ref.cast(msg);
        } else {
            tracing::warn!(
                "No workers in PG '{}' when trying to send to {}",
                pg_name,
                id
            );
        }
    }
}

impl Actor for Master {
    type Msg = ValorMasterMessage;
    type State = ValorMasterState;
    type Arguments = (
        ValorID,
        ActorRef<NodeServerMessage>,
        ValorConfig,
        Arc<RwLock<ValorWorkerRegistry>>,
        Arc<ValorTaskManager>,
    );

    async fn pre_start(
        &self,
        myself: ActorRef<Self::Msg>,
        (id, node_server, config, registry, task_manager): Self::Arguments,
    ) -> Result<Self::State, ActorProcessingErr> {
        // Join masters process group for workers to discover master via PG
        let master_pg = ValorID::masters_pg_name();
        let cell = myself.get_cell();
        ractor::pg::join(master_pg, vec![cell]);
        // Single-master self-check: if more than 1 member in masters PG, exit
        let masters_pg_name = ValorID::masters_pg_name();
        let members = ractor::pg::get_members(&masters_pg_name);
        if members.len() > 1 {
            tracing::error!(
                "Multiple masters detected in PG '{}' ({} members). Refusing to start this master.",
                masters_pg_name,
                members.len()
            );
            std::process::exit(1);
        }
        Ok(ValorMasterState {
            id,
            node_server,
            config,
            registry,
            task_manager,
        })
    }

    async fn handle(
        &self,
        _myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        match message {
            ValorMasterMessage::SouthboundRegisterWorker(worker_id, ipv4_addr) => {
                // PG-based duplicate guard: if worker_id already present in 'valor.workers', skip
                let members = ractor::pg::get_members(&ValorID::workers_pg_name());
                if members
                    .iter()
                    .any(|cell| cell.get_name() == Some(worker_id.to_string()))
                {
                    tracing::warn!("Worker {} already registered", worker_id);
                    return Ok(());
                }

                tracing::info!("Registering worker {} at {}", worker_id, ipv4_addr);
                state.registry.write().await.on_join(worker_id.clone());
                state.connect_worker(worker_id, ipv4_addr).await;
            }
            ValorMasterMessage::SouthboundUnregisterWorker(worker_id) => {
                tracing::warn!("Unregistering worker {}", worker_id);
                state.registry.write().await.on_leave(&worker_id);
            }
            ValorMasterMessage::SouthboundWorkerStateUpdate(event) => match event {
                ValorWorkerEvent::Heartbeat(WorkerHeartbeat {
                    id,
                    ts_mono_ms,
                    seq_no,
                }) => {
                    let (became_ready, gap) = state
                        .registry
                        .write()
                        .await
                        .on_heartbeat(&id, ts_mono_ms, seq_no);
                    // If worker transitions to Ready, consider re-assigning pending tasks later
                    if became_ready {
                        tracing::info!("Worker {} is now READY (missed={})", id, gap);
                    } else if gap > 0 {
                        tracing::debug!("Worker {} heartbeat gap detected: {}", id, gap);
                    }
                }
                ValorWorkerEvent::CapacityReport(CapacitySnapshot { id, capacity, .. }) => {
                    let short = id.to_string();
                    let prev = state.registry.read().await.get_capacity(&id);
                    let state_now = state.registry.read().await.get_state(&id);

                    // Compute percents
                    let cpu_used_pct = capacity.cpu_usage_pct.clamp(0.0, 100.0);
                    let mem_free_pct = if capacity.total_mem_mb > 0 {
                        (capacity.free_mem_mb as f32 * 100.0) / (capacity.total_mem_mb as f32)
                    } else {
                        0.0
                    };

                    // Deltas vs previous snapshot
                    let (delta_free_cpu, delta_free_mem) = if let Some(p) = prev {
                        (
                            capacity.free_cpu as i32 - p.free_cpu as i32,
                            capacity.free_mem_mb as i32 - p.free_mem_mb as i32,
                        )
                    } else {
                        (0, 0)
                    };

                    state.registry.write().await.update_capacity(&id, capacity);

                    tracing::debug!(
                        "Capacity update from {} | state={:?} | cpu: used {:.1}% (free {} / total {}) [Δfree {}] | mem: free {}MB / {}MB ({:.1}%) [Δ {}MB] | workers={} ",
                        short,
                        state_now.unwrap_or(super::registry::ValorWorkerState::Joining),
                        cpu_used_pct,
                        capacity.free_cpu,
                        capacity.total_cpu,
                        delta_free_cpu,
                        capacity.free_mem_mb,
                        capacity.total_mem_mb,
                        mem_free_pct,
                        delta_free_mem,
                        state.registry.read().await.len()
                    );
                }
                ValorWorkerEvent::ServicesReport(ServicesSnapshot {
                    id,
                    ts_mono_ms: _ts,
                    services,
                    version,
                }) => {
                    tracing::info!(
                        "Services report from {} (v{}): {} service(s)",
                        id,
                        version,
                        services.len()
                    );
                    state.registry.write().await.update_services(&id, services);
                }
            },
            // Task management messages
            ValorMasterMessage::CreateTask(task) => {
                let mspan = tracing::info_span!(
                    "flow.master.create_task",
                    task_id = %task.task_id
                );
                tracing::info!(parent: &mspan, "Master: received CreateTask");
                match state.task_manager.create_task(task).await {
                    Ok(task_id) => {
                        tracing::info!(parent: &mspan, "Master: task created (pending)");
                        // schedule by simple strategy
                        let reg_snapshot = state.registry.read().await;
                        let maybe_task = state.task_manager.get_task(&task_id).await;
                        let Some(task_for_sched) = maybe_task.as_ref() else {
                            tracing::error!("Task {} disappeared before scheduling", task_id);
                            return Ok(());
                        };
                        let chosen = crate::master::ValorMasterScheduler::select_worker(
                            task_for_sched,
                            &reg_snapshot,
                        );
                        // debug-level scheduler details removed per request
                        if let Some(worker_id) = chosen {
                            if let Err(e) =
                                state.task_manager.assign_task(&task_id, &worker_id).await
                            {
                                tracing::error!("Failed to assign task {}: {}", task_id, e);
                            } else {
                                tracing::info!(parent: &mspan, "Master: task assigned -> {}", worker_id);
                                // send task to worker via shared workers PG
                                let members = ractor::pg::get_members(
                                    &crate::types::ValorID::workers_pg_name(),
                                );
                                if members.is_empty() {
                                    tracing::warn!(
                                        "No workers found in PG to send task {}",
                                        task_id
                                    );
                                } else if let Some(task) =
                                    state.task_manager.get_task(&task_id).await
                                {
                                    for cell in members {
                                        let worker_ref: ActorRef<
                                            crate::worker::ValorWorkerMessage,
                                        > = cell.into();
                                        let _ = worker_ref.cast(
                                            crate::worker::ValorWorkerMessage::NorthboundMasterTask(
                                                Box::new(task.clone()),
                                            ),
                                        );
                                    }
                                }
                            }
                        } else {
                            tracing::warn!("No workers available to assign task {}", task_id);
                        }
                    }
                    Err(e) => {
                        tracing::error!("Failed to create task: {}", e);
                    }
                }
            }
            ValorMasterMessage::UpdateTaskStatus(update) => {
                let task_id_str = update.task_id.clone();
                let task_id = ValorTaskId::new(update.task_id);
                match state
                    .task_manager
                    .update_task_status(
                        &task_id,
                        &update.worker_id,
                        update.status,
                        update.output,
                        update.error,
                    )
                    .await
                {
                    Ok(()) => {
                        tracing::debug!(
                            "Updated task {} status to {:?}",
                            task_id_str,
                            update.status
                        );
                        // When task transitions to Running, start watchdog if configured
                        if update.status == crate::common::task::ValorTaskStatus::Running {
                            state.task_manager.maybe_spawn_watchdog(&task_id).await;
                        }
                    }
                    Err(e) => {
                        tracing::error!("Failed to update task status: {}", e);
                    }
                }
            }
            ValorMasterMessage::CancelTask(task_id) => {
                let task_id = ValorTaskId::new(task_id);
                match state.task_manager.cancel_task(&task_id).await {
                    Ok(()) => {
                        tracing::info!("Cancelled task: {}", task_id);
                    }
                    Err(e) => {
                        tracing::error!("Failed to cancel task: {}", e);
                    }
                }
            }
            ValorMasterMessage::GetTask(task_id) => {
                let task_id = ValorTaskId::new(task_id);
                let task = state.task_manager.get_task(&task_id).await;
                tracing::debug!("Get task {}: {:?}", task_id, task.is_some());
                // TODO: Implement response mechanism
            }
            ValorMasterMessage::ListTasks => {
                let tasks = state.task_manager.list_tasks().await;
                tracing::debug!("List tasks: {} tasks found", tasks.len());
                // TODO: Implement response mechanism
            }
            ValorMasterMessage::InternalWorkerRefResolved(_, _) => {
                tracing::debug!("Internal message received (worker ref resolved)");
            }
            ValorMasterMessage::InternalCleanupTick => {
                // Periodic liveness cleanup and task fallback
                let mut registry = state.registry.write().await;
                let (_before, _after, removed_ids) = registry.cleanup_stale_detailed();
                drop(registry);
                for wid in removed_ids {
                    let requeued = state
                        .task_manager
                        .requeue_tasks_for_unreachable_worker(&wid)
                        .await;
                    if requeued > 0 {
                        tracing::warn!(
                            worker = %wid,
                            requeued,
                            "Worker evicted; tasks requeued to Pending"
                        );
                    }
                }
            }
        }
        Ok(())
    }

    async fn handle_supervisor_evt(
        &self,
        _myself: ActorRef<Self::Msg>,
        message: ractor::SupervisionEvent,
        _state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        tracing::warn!("Supervisor event: {:?}", message);
        Ok(())
    }
}

impl ValorMasterState {
    #[allow(unused)] // Used for on-demand registry dump during debugging/ops.
    async fn log_registry_snapshot(state: &Self) {
        let guard = state.registry.read().await;
        let workers: Vec<String> = guard
            .iter()
            .map(|(id, rec)| {
                let short = id.to_string();
                let cap = &rec.capacity;
                let mem_pct = if cap.total_mem_mb > 0 {
                    (cap.free_mem_mb as f32 * 100.0) / (cap.total_mem_mb as f32)
                } else {
                    0.0
                };
                format!(
                    "{} [cpu free {}/{} ({:.1}% used) mem free {}/{}MB ({:.1}%)]",
                    short,
                    cap.free_cpu,
                    cap.total_cpu,
                    cap.cpu_usage_pct,
                    cap.free_mem_mb,
                    cap.total_mem_mb,
                    mem_pct
                )
            })
            .collect();
        tracing::info!(
            "Registry snapshot ({} workers): {}",
            workers.len(),
            workers.join(", ")
        );
    }
}

pub(crate) async fn startup_master_node_with_registry(
    port: u16,
    config: &ValorConfig,
    registry: Arc<RwLock<ValorWorkerRegistry>>,
    task_manager: Arc<ValorTaskManager>,
) -> ActorRef<ValorMasterMessage> {
    let id = &config.cli.id;
    let server = ractor_cluster::NodeServer::new(
        port,
        config.app.cluster_cookie(),
        format!("Master-NodeServer-{id}"),
        config.app.hostname(),
        // todo: TLS
        None,
        None,
    );

    let (node_server_actor, _handle) = Actor::spawn(None, server, ())
        .await
        .expect("Failed to start Master's NodeServer");

    let master_actor_name = ValorID::new_master(id).to_string();
    let (master_actor, _test_handle) = Actor::spawn(
        Some(master_actor_name),
        Master,
        (
            ValorID::new(&config.cli.id),
            node_server_actor,
            config.clone(),
            registry,
            task_manager,
        ),
    )
    .await
    .expect("Master actor failed to start up!");

    tracing::info!(
        "Master started on port {} with cookie: {}",
        port,
        config.app.cluster_cookie()
    );

    // wait for server startup to complete
    ractor::concurrency::sleep(Duration::from_millis(1000)).await;

    master_actor
}