tinytown 0.10.0

/*
 * Copyright (c) 2024-Present, Jeremy Plichta
 * Licensed under the MIT License
 */

//! Town - the central orchestration hub.

use std::collections::HashMap;
use std::path::Path;
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};

use redis::Client;
use redis::aio::ConnectionManager;
use tokio::process::Child;
use tokio::sync::RwLock;
use tracing::{debug, info, warn};

use crate::agent::{Agent, AgentId, AgentState, AgentType};
use crate::channel::Channel;
use crate::config::Config;
use crate::error::{Error, Result};
use crate::events::EventStream;
use crate::global_config::GlobalConfig;
use crate::message::{Message, MessageType};
use crate::task::{Task, TaskId};

/// Town directory structure - all artifacts go under .tt/
pub const TT_DIR: &str = ".tt";
const AGENTS_DIR: &str = ".tt/agents";
const LOGS_DIR: &str = ".tt/logs";
const TASKS_DIR: &str = ".tt/tasks";

/// Minimum required Redis version
const MIN_MANAGED_REDIS_VERSION: (u32, u32) = (8, 0);
const MIN_HEALTHCHECK_REDIS_VERSION: (u32, u32) = (7, 0);

/// The Town orchestrates agents and message passing.
#[derive(Clone)]
pub struct Town {
    config: Config,
    channel: Channel,
    agents: Arc<RwLock<HashMap<AgentId, Agent>>>,
    #[expect(dead_code)]
    processes: Arc<RwLock<HashMap<AgentId, Child>>>,
}

/// PID file name for tracking Redis process (under .tt/)
const REDIS_PID_FILE: &str = ".tt/redis.pid";
static CENTRAL_REDIS_START_LOCK: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();

/// Find the redis-server binary, preferring ~/.tt/bin over PATH.
fn find_redis_server() -> std::path::PathBuf {
    // First, check ~/.tt/bin/redis-server (bootstrapped version)
    if let Some(home) = dirs::home_dir() {
        let tt_redis = home.join(".tt/bin/redis-server");
        if tt_redis.exists() {
            debug!("Using bootstrapped Redis: {}", tt_redis.display());
            return tt_redis;
        }
    }
    // Fall back to PATH
    std::path::PathBuf::from("redis-server")
}

impl Town {
    /// Check that Redis is installed and meets minimum version requirements.
    fn check_managed_redis_version() -> Result<()> {
        use std::process::Command as StdCommand;

        let redis_bin = find_redis_server();

        // Check if redis-server is available
        let output = StdCommand::new(&redis_bin)
            .arg("--version")
            .output()
            .map_err(|_| Error::RedisNotInstalled)?;

        if !output.status.success() {
            return Err(Error::RedisNotInstalled);
        }

        let version_str = String::from_utf8_lossy(&output.stdout);

        // Parse version from output like: "Redis server v=8.0.0 sha=..."
        let version = Self::parse_redis_version(&version_str)?;

        if version < MIN_MANAGED_REDIS_VERSION {
            return Err(Error::RedisVersionTooOld(format!(
                "{}.{}",
                version.0, version.1
            )));
        }

        info!("Redis version {}.{} detected ✓", version.0, version.1);
        Ok(())
    }

    /// Parse Redis version from --version output.
    fn parse_redis_version(version_str: &str) -> Result<(u32, u32)> {
        // Format: "Redis server v=8.0.0 sha=..." or "Redis server v=7.2.4 sha=..."
        let version_part = version_str
            .split("v=")
            .nth(1)
            .and_then(|s| s.split_whitespace().next())
            .ok_or_else(|| Error::RedisVersionTooOld("unknown".to_string()))?;

        let parts: Vec<&str> = version_part.split('.').collect();
        if parts.len() < 2 {
            return Err(Error::RedisVersionTooOld(version_part.to_string()));
        }

        let major = parts[0]
            .parse::<u32>()
            .map_err(|_| Error::RedisVersionTooOld(version_part.to_string()))?;
        let minor = parts[1]
            .parse::<u32>()
            .map_err(|_| Error::RedisVersionTooOld(version_part.to_string()))?;

        Ok((major, minor))
    }

    /// Initialize a new town at the given path.
    pub async fn init(path: impl AsRef<Path>, name: impl Into<String>) -> Result<Self> {
        let path = path.as_ref();
        let name = name.into();

        let config = Config::new(&name, path);
        Self::init_with_config(config).await
    }

    /// Initialize a new town using an explicit configuration.
    pub async fn init_with_config(config: Config) -> Result<Self> {
        let path = &config.root;

        // Check Redis version first
        if !config.is_remote_redis() {
            Self::check_managed_redis_version()?;
        }

        info!("Initializing town '{}' at {}", config.name, path.display());

        // Create directory structure - all artifacts go under .tt/
        std::fs::create_dir_all(path)?;
        std::fs::create_dir_all(path.join(TT_DIR))?;
        std::fs::create_dir_all(path.join(AGENTS_DIR))?;
        std::fs::create_dir_all(path.join(LOGS_DIR))?;
        std::fs::create_dir_all(path.join(TASKS_DIR))?;

        // Persist config
        config.save()?;

        // Start Redis (daemonized - stays running after we exit) and connect
        Self::start_redis(&config).await?;
        let channel = Self::connect_redis(&config).await?;

        Ok(Self {
            config,
            channel,
            agents: Arc::new(RwLock::new(HashMap::new())),
            processes: Arc::new(RwLock::new(HashMap::new())),
        })
    }

    /// Connect to an existing town.
    pub async fn connect(path: impl AsRef<Path>) -> Result<Self> {
        // Check Redis version first (skip for remote Redis)
        let config = Config::load(&path)?;

        if !config.is_remote_redis() {
            Self::check_managed_redis_version()?;
        }

        // Determine if Redis appears to be running
        let redis_appears_ready = if config.is_remote_redis() {
            // For remote Redis, always try to connect first
            true
        } else if config.redis.use_socket {
            // Unix socket mode - check if socket file exists
            config.socket_path().exists()
        } else {
            // Local TCP mode - try to connect to the port
            std::net::TcpStream::connect(format!("{}:{}", config.redis.bind, config.redis.port))
                .is_ok()
        };

        // Try to connect to Redis, start if needed
        let channel = if redis_appears_ready {
            // Redis appears to be running - try to connect
            match Self::connect_redis(&config).await {
                Ok(ch) => ch,
                Err(_) if !config.is_remote_redis() => {
                    // Local Redis not responding - restart it
                    warn!("Redis not responding, restarting...");
                    Self::start_redis(&config).await?;
                    Self::connect_redis(&config).await?
                }
                Err(e) => {
                    // Remote Redis failed - can't restart, propagate error
                    return Err(e);
                }
            }
        } else {
            // Redis not running - start it (only for local Redis)
            debug!("Redis not found, starting...");
            Self::start_redis(&config).await?;
            Self::connect_redis(&config).await?
        };

        Ok(Self {
            config,
            channel,
            agents: Arc::new(RwLock::new(HashMap::new())),
            processes: Arc::new(RwLock::new(HashMap::new())),
        })
    }

    /// Start a local Redis server (daemonized).
    /// Supports both Unix socket (default) and TCP modes with security options.
    /// Redis will continue running after tinytown exits.
    async fn start_redis(config: &Config) -> Result<()> {
        // Skip starting local server for external/remote Redis
        if config.is_remote_redis() {
            info!("Using external Redis: {}", config.redis_url_redacted());
            return Ok(());
        }

        // Check if using central Redis mode
        let is_central = config.is_central_redis();

        let _central_guard = if is_central {
            Some(
                CENTRAL_REDIS_START_LOCK
                    .get_or_init(|| tokio::sync::Mutex::new(()))
                    .lock()
                    .await,
            )
        } else {
            None
        };

        // For central Redis, another test may have started the daemon just before
        // we acquired the lock. Wait for it to become reachable before returning.
        if is_central && GlobalConfig::is_central_redis_running() {
            debug!("Central Redis already running");
            Self::wait_for_redis_ready(config).await?;
            return Ok(());
        }

        // Determine PID file and working directory
        let (pid_file, work_dir) = if is_central {
            let global_dir = GlobalConfig::config_dir()?;
            std::fs::create_dir_all(&global_dir)?;
            (GlobalConfig::redis_pid_path()?, global_dir)
        } else {
            (config.root.join(REDIS_PID_FILE), config.root.clone())
        };

        let redis_bin = find_redis_server();

        debug!("Using Redis binary: {}", redis_bin.display());

        // Build args dynamically based on config
        let mut args: Vec<String> = vec![
            "--daemonize".to_string(),
            "yes".to_string(),
            "--pidfile".to_string(),
            pid_file.to_str().unwrap().to_string(),
            "--loglevel".to_string(),
            "warning".to_string(),
        ];

        if config.redis.use_socket {
            // Unix socket mode (default, current behavior)
            let socket_path = config.socket_path();

            // Remove stale socket if exists
            if socket_path.exists() {
                std::fs::remove_file(&socket_path)?;
            }

            info!("Starting Redis with socket: {}", socket_path.display());
            args.extend([
                "--unixsocket".to_string(),
                socket_path.to_str().unwrap().to_string(),
                "--unixsocketperm".to_string(),
                "700".to_string(),
                "--port".to_string(),
                "0".to_string(), // Disable TCP
            ]);
        } else {
            // TCP mode with security
            info!(
                "Starting Redis with TCP on {}:{}",
                config.redis.bind, config.redis.port
            );

            // TLS configuration
            if config.redis.tls_enabled {
                args.extend([
                    "--tls-port".to_string(),
                    config.redis.port.to_string(),
                    "--port".to_string(),
                    "0".to_string(), // Disable non-TLS port when TLS is enabled
                ]);

                if let Some(ref cert) = config.redis.tls_cert {
                    args.extend(["--tls-cert-file".to_string(), cert.clone()]);
                }
                if let Some(ref key) = config.redis.tls_key {
                    args.extend(["--tls-key-file".to_string(), key.clone()]);
                }
                if let Some(ref ca_cert) = config.redis.tls_ca_cert {
                    args.extend(["--tls-ca-cert-file".to_string(), ca_cert.clone()]);
                }
            } else {
                // Plain TCP
                args.extend(["--port".to_string(), config.redis.port.to_string()]);
            }

            // Bind address
            args.extend(["--bind".to_string(), config.redis.bind.clone()]);

            // Password authentication (check env var first via redis_password())
            if let Some(ref password) = config.redis_password() {
                args.extend(["--requirepass".to_string(), password.clone()]);
            }

            // Protected mode: Redis requires this when binding to non-localhost without password
            if config.redis.bind != "127.0.0.1" && config.redis_password().is_none() {
                warn!(
                    "Binding to {} without password - enabling protected mode",
                    config.redis.bind
                );
                args.extend(["--protected-mode".to_string(), "yes".to_string()]);
            }
        }

        // Start Redis daemonized
        if is_central {
            info!(
                "Starting central Redis on {}:{}",
                config.redis.host, config.redis.port
            );
        }
        let status = std::process::Command::new(&redis_bin)
            .args(&args)
            .current_dir(&work_dir)
            .status()?;

        if !status.success() {
            return Err(Error::Timeout("Redis failed to start".into()));
        }

        Self::wait_for_redis_ready(config).await
    }

    /// Connect to Redis.
    async fn connect_redis(config: &Config) -> Result<Channel> {
        let url = config.redis_url();
        // Use redacted URL for logging to avoid exposing password
        debug!("Connecting to Redis: {}", config.redis_url_redacted());

        let client = Client::open(url).map_err(|err| {
            Error::Config(format!(
                "Invalid Redis configuration for {}: {}",
                config.redis_url_redacted(),
                err
            ))
        })?;
        Self::run_redis_startup_health_check(&client, config).await?;

        // Short timeout - Redis should connect in milliseconds if healthy
        // Stale sockets can hang indefinitely, so fail fast and restart Redis
        let conn = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            ConnectionManager::new(client),
        )
        .await
        .map_err(|_| Error::Timeout("Redis connection timed out".into()))??;

        Ok(Channel::new(conn, &config.name))
    }

    async fn run_redis_startup_health_check(client: &Client, config: &Config) -> Result<()> {
        let mut conn = tokio::time::timeout(
            Duration::from_secs(5),
            client.get_multiplexed_async_connection(),
        )
        .await
        .map_err(|_| {
            Self::redis_connection_error(
                config,
                "Timed out while opening the configured Redis connection",
            )
        })?
        .map_err(|err| Self::redis_connection_error(config, &err.to_string()))?;

        let response: String = tokio::time::timeout(
            Duration::from_secs(2),
            redis::cmd("PING").query_async(&mut conn),
        )
        .await
        .map_err(|_| Self::redis_connection_error(config, "Timed out waiting for Redis PING"))?
        .map_err(|err| Self::redis_connection_error(config, &err.to_string()))?;

        if response != "PONG" {
            return Err(Self::redis_connection_error(
                config,
                &format!("Unexpected PING response: {}", response),
            ));
        }

        let info: String = tokio::time::timeout(
            Duration::from_secs(2),
            redis::cmd("INFO").arg("server").query_async(&mut conn),
        )
        .await
        .map_err(|_| {
            Self::redis_connection_error(config, "Timed out while fetching Redis server info")
        })?
        .map_err(|err| Self::redis_connection_error(config, &err.to_string()))?;

        let version = Self::parse_info_redis_version(&info)?;
        if version < MIN_HEALTHCHECK_REDIS_VERSION {
            return Err(Error::Config(format!(
                "Configured Redis at {} is version {}.{}. Tinytown requires Redis {}.{}+ for connectivity.",
                config.redis_url_redacted(),
                version.0,
                version.1,
                MIN_HEALTHCHECK_REDIS_VERSION.0,
                MIN_HEALTHCHECK_REDIS_VERSION.1
            )));
        }

        Ok(())
    }

    async fn wait_for_redis_ready(config: &Config) -> Result<()> {
        let deadline = Instant::now() + Duration::from_secs(10);
        while Instant::now() < deadline {
            let endpoint_ready = if config.redis.use_socket {
                config.socket_path().exists()
            } else {
                std::net::TcpStream::connect(format!("{}:{}", config.redis.bind, config.redis.port))
                    .is_ok()
            };

            if endpoint_ready && Self::ping_redis(config).await.is_ok() {
                debug!("Redis ready");
                return Ok(());
            }

            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        Err(Error::Timeout("Redis failed to start".into()))
    }

    async fn ping_redis(config: &Config) -> Result<()> {
        let client = Client::open(config.redis_url())?;
        let mut conn = tokio::time::timeout(
            Duration::from_secs(1),
            client.get_multiplexed_async_connection(),
        )
        .await
        .map_err(|_| Error::Timeout("Redis connection timed out".into()))??;

        let response: String = tokio::time::timeout(
            Duration::from_secs(1),
            redis::cmd("PING").query_async(&mut conn),
        )
        .await
        .map_err(|_| Error::Timeout("Redis ping timed out".into()))??;

        if response == "PONG" {
            Ok(())
        } else {
            Err(Error::Timeout(format!(
                "Unexpected Redis PING response: {}",
                response
            )))
        }
    }

    fn parse_info_redis_version(info: &str) -> Result<(u32, u32)> {
        let version = info
            .lines()
            .find_map(|line| line.strip_prefix("redis_version:"))
            .ok_or_else(|| {
                Error::Config("Redis INFO response did not include redis_version".into())
            })?;
        let parts: Vec<&str> = version.split('.').collect();
        if parts.len() < 2 {
            return Err(Error::Config(format!(
                "Redis INFO reported an invalid redis_version value: {version}"
            )));
        }

        let major = parts[0].parse::<u32>().map_err(|_| {
            Error::Config(format!(
                "Redis INFO reported an invalid redis_version value: {version}"
            ))
        })?;
        let minor = parts[1].parse::<u32>().map_err(|_| {
            Error::Config(format!(
                "Redis INFO reported an invalid redis_version value: {version}"
            ))
        })?;

        Ok((major, minor))
    }

    fn redis_connection_error(config: &Config, detail: &str) -> Error {
        let detail = detail.trim();
        let uppercase = detail.to_ascii_uppercase();
        let prefix = if uppercase.contains("WRONGPASS") || uppercase.contains("NOAUTH") {
            "Redis authentication failed"
        } else {
            "Failed to connect to configured Redis"
        };

        Error::Config(format!(
            "{prefix} at {}: {detail}",
            config.redis_url_redacted()
        ))
    }

    /// Spawn a new worker agent.
    pub async fn spawn_agent(&self, name: &str, cli: &str) -> Result<AgentHandle> {
        let normalized = name.trim().to_lowercase();
        if normalized == "supervisor" || normalized == "conductor" {
            return Err(Error::Config(format!(
                "'{}' is reserved for the well-known supervisor/conductor mailbox",
                name
            )));
        }

        let agent = Agent::new(name, cli, AgentType::Worker);
        let id = agent.id;

        // Store agent state
        self.channel.set_agent_state(&agent).await?;
        self.agents.write().await.insert(id, agent);

        info!("Spawned agent '{}' ({})", name, id);

        Ok(AgentHandle {
            id,
            channel: self.channel.clone(),
        })
    }

    /// Get a handle to an existing agent.
    pub async fn agent(&self, name: &str) -> Result<AgentHandle> {
        // Look up agent in Redis (persisted across process restarts)
        if let Some(agent) = self.channel.get_agent_by_name(name).await? {
            return Ok(AgentHandle {
                id: agent.id,
                channel: self.channel.clone(),
            });
        }

        let normalized = name.trim().to_lowercase();
        if normalized == "supervisor" || normalized == "conductor" {
            return Ok(AgentHandle {
                id: AgentId::supervisor(),
                channel: self.channel.clone(),
            });
        }
        Err(Error::AgentNotFound(name.to_string()))
    }

    /// List all agents.
    pub async fn list_agents(&self) -> Vec<Agent> {
        // Get agents from Redis (persisted across process restarts)
        self.channel.list_agents().await.unwrap_or_default()
    }

    /// Get the communication channel.
    pub fn channel(&self) -> &Channel {
        &self.channel
    }

    /// Get the town configuration.
    pub fn config(&self) -> &Config {
        &self.config
    }

    /// Get the town root directory.
    pub fn root(&self) -> &Path {
        &self.config.root
    }

    /// Create an EventStream for emitting/reading structured events.
    pub fn event_stream(&self) -> EventStream {
        EventStream::new(self.channel.conn().clone(), self.channel.town_name())
    }
}

// Note: Redis runs daemonized and persists after Town is dropped.
// `tt stop` only requests town agents to stop gracefully; it does not shut down shared Redis.

/// Handle for interacting with an agent.
#[derive(Clone)]
pub struct AgentHandle {
    id: AgentId,
    channel: Channel,
}

impl AgentHandle {
    /// Get the agent ID.
    pub fn id(&self) -> AgentId {
        self.id
    }

    /// Assign a task to this agent.
    pub async fn assign(&self, task: Task) -> Result<TaskId> {
        let task_id = task.id;

        // Store task
        self.channel.set_task(&task).await?;

        // Send assignment message
        let msg = Message::new(
            AgentId::supervisor(),
            self.id,
            MessageType::TaskAssign {
                task_id: task_id.to_string(),
            },
        );
        self.channel.send(&msg).await?;

        Ok(task_id)
    }

    /// Send a message to this agent.
    pub async fn send(&self, msg_type: MessageType) -> Result<()> {
        let msg = Message::new(AgentId::supervisor(), self.id, msg_type);
        self.channel.send(&msg).await
    }

    /// Check agent's inbox length.
    pub async fn inbox_len(&self) -> Result<usize> {
        self.channel.inbox_len(self.id).await
    }

    /// Get agent state.
    pub async fn state(&self) -> Result<Option<Agent>> {
        self.channel.get_agent_state(self.id).await
    }

    /// Wait for agent to complete current task.
    pub async fn wait(&self) -> Result<()> {
        // Poll until agent is idle, stopped, cold, or error
        loop {
            if let Some(agent) = self.state().await? {
                match agent.state {
                    AgentState::Idle | AgentState::Stopped | AgentState::Cold => return Ok(()),
                    AgentState::Error => {
                        return Err(Error::AgentNotFound(format!(
                            "Agent {} in error state",
                            self.id
                        )));
                    }
                    _ => {}
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::Town;
    use crate::Error;

    #[test]
    fn parse_info_redis_version_rejects_invalid_values_with_config_error() {
        let err = Town::parse_info_redis_version("redis_version:not-a-version")
            .expect_err("invalid INFO version should fail");

        match err {
            Error::Config(message) => {
                assert!(message.contains("invalid redis_version value"));
                assert!(message.contains("not-a-version"));
                assert!(!message.contains("requires Redis 8.0+"));
                assert!(!message.contains("tt bootstrap"));
            }
            other => panic!("expected config error, got {other}"),
        }
    }
}