use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};

#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum NodeRole {
    Primary,
    Secondary,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct NodeConfig {
    pub name: String,
    pub role: NodeRole,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct HotConfig {
    pub path: PathBuf,
    pub max_gb: u64,
    pub stale_hours: u64,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct WarmConfig {
    pub projects_path: PathBuf,
    pub zfs_dataset: String,
    pub snapshot_retain_hours: u64,
    pub snapshot_retain_days: u64,
    pub snapshot_retain_weeks: u64,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ColdConfig {
    pub archive_path: PathBuf,
    pub zfs_dataset: String,
    pub retain_weeks: u64,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ReplicationConfig {
    pub receive_from_peer: Option<bool>,
    pub peer_user: Option<String>,
    pub send_to_host: Option<String>,
    pub send_to_user: Option<String>,
    pub cold_dataset_on_peer: Option<String>,
    pub nightly_at: Option<String>,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct PeerConfig {
    pub host: String,
    pub user: String,
}

/// One peer node the cluster knows about. Every peer carries up to two reachable
/// endpoints: a LAN socket (fast path, tried first) and a Tailscale socket
/// (fallback, always reachable when the tailnet is up). At least one must be set.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
pub struct PeerEntry {
    /// Stable human-facing name (e.g. `"architect"`). Must be unique in the config.
    pub name: String,
    /// Zone tag — one of `"fabric-10g"`, `"lan-1g"`, `"roaming"`, or a custom label.
    pub zone: String,
    /// LAN socket. Omit when this peer has no LAN presence from our perspective
    /// (e.g. a roaming laptop reachable only via Tailscale).
    #[serde(default)]
    pub lan_addr: Option<SocketAddr>,
    /// Tailscale socket. Omit only if the peer is LAN-only.
    #[serde(default)]
    pub tailscale_addr: Option<SocketAddr>,
}

impl PeerEntry {
    /// Sanity check: at least one address must be reachable.
    pub fn validate(&self) -> Result<()> {
        if self.name.is_empty() {
            bail!("peer entry with empty name");
        }
        if self.zone.is_empty() {
            bail!("peer {} has empty zone", self.name);
        }
        if self.lan_addr.is_none() && self.tailscale_addr.is_none() {
            bail!(
                "peer {} has no reachable address (both lan_addr and tailscale_addr unset)",
                self.name
            );
        }
        Ok(())
    }
}

/// Paths to persisted mTLS material for the RPC transport (Phase 1d).
///
/// A production node reads its identity from these three files on startup:
///   * `ca_cert` — fleet root CA cert (public, distributed to every node)
///   * `node_cert` — this node's leaf cert (signed by the CA)
///   * `node_key` — this node's private key (must be 0o600, never checked in)
///
/// The `[cluster.tls]` block is optional so pre-v2 configs keep loading;
/// callers that need mTLS (e.g. `QuicServer::bind`) fail with a clear
/// message when it's absent.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Default)]
pub struct ClusterTlsConfig {
    pub ca_cert: PathBuf,
    pub node_cert: PathBuf,
    pub node_key: PathBuf,
}

/// Cluster membership configuration. Optional at the top level so existing
/// single-node deployments (pre-v2) keep loading. Once present, describes the
/// local node's zone + bind addresses, and enumerates known peers.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Default)]
pub struct ClusterConfig {
    /// This node's zone tag.
    pub zone: String,
    /// LAN listen socket for gossip (typically `0.0.0.0:7701`). Chitchat runs
    /// on this UDP port. Omit on roaming nodes.
    #[serde(default)]
    pub bind_lan: Option<SocketAddr>,
    /// Tailscale listen socket for gossip. Omit on strictly-LAN nodes.
    #[serde(default)]
    pub bind_tailscale: Option<SocketAddr>,
    /// LAN listen socket for RPC (QUIC). Defaults to `bind_lan.port + 1`
    /// so gossip and RPC don't collide on the same UDP endpoint.
    #[serde(default)]
    pub bind_rpc_lan: Option<SocketAddr>,
    /// Tailscale listen socket for RPC (QUIC). Defaults to
    /// `bind_tailscale.port + 1`.
    #[serde(default)]
    pub bind_rpc_tailscale: Option<SocketAddr>,
    /// Static seed list of peers. Runtime membership (Phase 1b) will extend this
    /// via gossip; the config list bootstraps discovery.
    #[serde(default)]
    pub peers: Vec<PeerEntry>,
    /// Optional mTLS material paths. Required when the RPC transport is
    /// used; absent means "gossip only, no RPC" for now.
    #[serde(default)]
    pub tls: Option<ClusterTlsConfig>,
    /// Optional local blob-store root (Phase 2). When set, the daemon
    /// opens a content-addressed store at this path and serves it via
    /// the Blob* RPC methods. Absent means the node participates in
    /// gossip + peer-status but returns `NotConfigured` for Blob RPCs.
    #[serde(default)]
    pub blob_store_root: Option<PathBuf>,
    /// Phase 5j: optional bind address for the Prometheus `/metrics`
    /// endpoint. When set, the daemon spins up a tiny HTTP server on
    /// this address serving the same counters exposed via `GetMetrics`
    /// and gossip. Typical value is `127.0.0.1:7702` (Prometheus scrapes
    /// via the LAN listener the node advertises to its scrape target
    /// group). Absent means "no scrape endpoint".
    #[serde(default)]
    pub prom_bind: Option<SocketAddr>,
    /// Field finding 2026-07-12: how often the daemon runs
    /// `gc_orphan_chunks` to reclaim disk from chunks no live
    /// manifest references. `None` or `0` disables auto-GC — the
    /// operator can still invoke `claw-store cluster-gc` by hand.
    /// Typical value: `6` hours on a runner cache.
    #[serde(default)]
    pub gc_interval_hours: Option<u64>,
    /// Field finding 2026-07-12: total blob-store size cap in GiB.
    /// When set, the auto-GC ticker runs `evict_to_size_cap` after
    /// its orphan sweep, deleting oldest manifests until the
    /// live-referenced footprint sits at or below this bound.
    /// Absent means "grow unbounded".
    #[serde(default)]
    pub blob_max_gb: Option<u64>,
}

/// Compute the default RPC address for a gossip address: same IP, port + 1.
/// Used to derive `bind_rpc_lan` / `bind_rpc_tailscale` when the operator
/// hasn't set them explicitly.
fn default_rpc_addr(gossip: SocketAddr) -> Option<SocketAddr> {
    let port = gossip.port().checked_add(1)?;
    Some(SocketAddr::new(gossip.ip(), port))
}

impl ClusterConfig {
    /// Sanity check the cluster config: at least one bind address, no duplicate
    /// peer names, every peer has at least one address.
    pub fn validate(&self) -> Result<()> {
        if self.zone.is_empty() {
            bail!("cluster.zone is empty");
        }
        if self.bind_lan.is_none() && self.bind_tailscale.is_none() {
            bail!("cluster has no bind address (both bind_lan and bind_tailscale unset)");
        }
        let mut names = HashSet::new();
        for peer in &self.peers {
            peer.validate()?;
            if !names.insert(peer.name.as_str()) {
                bail!("duplicate peer name in cluster.peers: {}", peer.name);
            }
        }
        Ok(())
    }

    /// Look up a peer by name.
    pub fn peer(&self, name: &str) -> Option<&PeerEntry> {
        self.peers.iter().find(|p| p.name == name)
    }

    /// The LAN address this node listens on for RPC (QUIC). Prefers the
    /// explicit `bind_rpc_lan` override; otherwise derives from `bind_lan`
    /// with port + 1.
    pub fn rpc_lan(&self) -> Option<SocketAddr> {
        self.bind_rpc_lan
            .or_else(|| self.bind_lan.and_then(default_rpc_addr))
    }

    /// The Tailscale address this node listens on for RPC (QUIC).
    pub fn rpc_tailscale(&self) -> Option<SocketAddr> {
        self.bind_rpc_tailscale
            .or_else(|| self.bind_tailscale.and_then(default_rpc_addr))
    }
}

impl PeerEntry {
    /// The peer's LAN RPC address, if it has one.
    ///
    /// This is derived from the peer's `lan_addr` (which is their gossip
    /// address in the config, matching how they identify themselves). RPC
    /// runs on gossip port + 1 by convention. Gossip-based discovery in
    /// Phase 1b will supplant this once a peer has broadcast its own
    /// [`gossip::keys::RPC_ADDR_LAN`](crate::cluster::gossip::keys).
    pub fn rpc_lan(&self) -> Option<SocketAddr> {
        self.lan_addr.and_then(default_rpc_addr)
    }

    /// The peer's Tailscale RPC address.
    pub fn rpc_tailscale(&self) -> Option<SocketAddr> {
        self.tailscale_addr.and_then(default_rpc_addr)
    }
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct Config {
    pub node: NodeConfig,
    pub hot: HotConfig,
    pub warm: WarmConfig,
    pub cold: Option<ColdConfig>,
    pub replication: Option<ReplicationConfig>,
    pub peer: Option<PeerConfig>,
    /// v2 cluster membership (peers, zones, bind addresses). Optional so pre-v2
    /// configs still load; once populated, `Config::load` calls `validate()`.
    #[serde(default)]
    pub cluster: Option<ClusterConfig>,
    /// Optional Bearer token required on all HTTP POST endpoints.
    /// Set to a long random string, e.g. `openssl rand -hex 32`.
    /// If absent, POST endpoints are unauthenticated (internal-network use only).
    ///
    /// When set, this is treated as an **admin** token — no namespace
    /// restriction. Prefer per-app tokens under `[[aggregator.tokens]]`
    /// (below) for multi-tenant setups; `api_token` stays as the
    /// pre-Phase-9 escape hatch for single-tenant use.
    #[serde(default)]
    pub api_token: Option<String>,

    /// Aggregator-side auth: per-app Bearer tokens, each scoped to a
    /// namespace prefix on tag names. Enables safe multi-tenant use
    /// (e.g. clawmates workspace X only touches `workspace:x:*` tags).
    /// Empty by default; `api_token` above still works as a wildcard
    /// admin token.
    #[serde(default)]
    pub aggregator: Option<AggregatorConfig>,
}

/// Aggregator-side per-app auth config. See [`Config::aggregator`].
#[derive(Debug, Clone, Deserialize, Serialize, Default)]
pub struct AggregatorConfig {
    /// One entry per app that talks to the aggregator. A token with no
    /// `namespace` set is an admin token (can touch any tag); a token
    /// with `namespace = "foo"` may only write tags whose name starts
    /// with `foo:`.
    #[serde(default)]
    pub tokens: Vec<TokenEntry>,
}

/// A single Bearer token binding: `token` value → optional `namespace`
/// prefix that constrains which tag names this caller may touch.
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct TokenEntry {
    /// The Bearer value the app presents in `Authorization: Bearer …`.
    /// Long random string, e.g. `openssl rand -hex 32`.
    pub token: String,
    /// Tag-name prefix this token is allowed to write. Enforced with
    /// a mandatory `<namespace>:` separator so `workspace:42` cannot
    /// silently reach `workspace:420:*`. Absent = admin (any tag).
    #[serde(default)]
    pub namespace: Option<String>,
    /// Human note; not consumed by auth. Shown in logs / listings.
    #[serde(default)]
    pub description: Option<String>,
}

impl Config {
    pub fn load(path: &Path) -> Result<Self> {
        let content = std::fs::read_to_string(path)
            .with_context(|| format!("reading config at {}", path.display()))?;
        let cfg: Config = toml::from_str(&content)
            .with_context(|| format!("parsing config at {}", path.display()))?;
        if let Some(cluster) = cfg.cluster.as_ref() {
            cluster
                .validate()
                .with_context(|| format!("validating cluster config in {}", path.display()))?;
        }
        Ok(cfg)
    }

    pub fn default_path() -> PathBuf {
        PathBuf::from("/etc/claw-store/config.toml")
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_load_primary_config() {
        let toml = r#"
[node]
name = "architect"
role = "primary"

[hot]
path = "/hot/targets"
max_gb = 200
stale_hours = 48

[warm]
projects_path = "/slab/projects"
zfs_dataset = "slab/projects"
snapshot_retain_hours = 24
snapshot_retain_days = 7
snapshot_retain_weeks = 4

[cold]
archive_path = "/data/archive"
zfs_dataset = "data/archive"
retain_weeks = 12

[replication]
receive_from_peer = true
peer_user = "osobh"
"#;
        let cfg: Config = toml::from_str(toml).unwrap();
        assert_eq!(cfg.node.name, "architect");
        assert_eq!(cfg.node.role, NodeRole::Primary);
        assert_eq!(cfg.hot.max_gb, 200);
        assert!(cfg.cold.is_some());
        assert_eq!(cfg.cold.unwrap().retain_weeks, 12);
    }

    #[test]
    fn test_pre_v2_config_still_loads_without_cluster_section() {
        // Backwards compatibility: any config that worked before v2 must still
        // parse cleanly with `cluster` absent.
        let toml = r#"
[node]
name = "tank"
role = "secondary"

[hot]
path = "/hot/targets"
max_gb = 200
stale_hours = 48

[warm]
projects_path = "/slab/projects"
zfs_dataset = "slab/projects"
snapshot_retain_hours = 24
snapshot_retain_days = 7
snapshot_retain_weeks = 4
"#;
        let cfg: Config = toml::from_str(toml).unwrap();
        assert!(cfg.cluster.is_none(), "cluster must be optional");
    }

    #[test]
    fn test_cluster_config_parses_with_peers() {
        let toml = r#"
[node]
name = "tank"
role = "secondary"

[hot]
path = "/hot/targets"
max_gb = 200
stale_hours = 48

[warm]
projects_path = "/slab/projects"
zfs_dataset = "slab/projects"
snapshot_retain_hours = 24
snapshot_retain_days = 7
snapshot_retain_weeks = 4

[cluster]
zone = "fabric-10g"
bind_lan = "10.0.0.14:7701"
bind_tailscale = "100.64.1.2:7701"

[[cluster.peers]]
name = "architect"
zone = "fabric-10g"
lan_addr = "10.0.0.13:7701"
tailscale_addr = "100.64.1.3:7701"

[[cluster.peers]]
name = "morpheus"
zone = "lan-1g"
lan_addr = "192.168.1.50:7701"
tailscale_addr = "100.64.1.4:7701"

[[cluster.peers]]
name = "laptop"
zone = "roaming"
tailscale_addr = "100.64.1.5:7701"
"#;
        let cfg: Config = toml::from_str(toml).unwrap();
        let cluster = cfg.cluster.expect("cluster section present");
        assert_eq!(cluster.zone, "fabric-10g");
        assert_eq!(cluster.peers.len(), 3);

        let architect = cluster.peer("architect").expect("architect peer");
        assert_eq!(architect.zone, "fabric-10g");
        assert_eq!(
            architect.lan_addr.unwrap(),
            "10.0.0.13:7701".parse::<SocketAddr>().unwrap()
        );

        let laptop = cluster.peer("laptop").expect("laptop peer");
        assert!(laptop.lan_addr.is_none(), "roaming peer has no LAN address");
        assert!(laptop.tailscale_addr.is_some());

        cluster.validate().expect("valid cluster config");
    }

    #[test]
    fn test_peer_without_addresses_fails_validation() {
        let peer = PeerEntry {
            name: "ghost".into(),
            zone: "fabric-10g".into(),
            lan_addr: None,
            tailscale_addr: None,
        };
        let err = peer.validate().unwrap_err().to_string();
        assert!(
            err.contains("ghost") && err.contains("no reachable address"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn test_cluster_without_bind_addresses_fails_validation() {
        let cluster = ClusterConfig {
            zone: "fabric-10g".into(),
            bind_lan: None,
            bind_tailscale: None,
            peers: vec![],
            bind_rpc_lan: None,
            bind_rpc_tailscale: None,
            tls: None,
            blob_store_root: None,
            prom_bind: None,
            gc_interval_hours: None,
            blob_max_gb: None,
        };
        let err = cluster.validate().unwrap_err().to_string();
        assert!(err.contains("no bind address"), "unexpected error: {err}");
    }

    #[test]
    fn test_cluster_duplicate_peer_names_fails_validation() {
        let cluster = ClusterConfig {
            zone: "fabric-10g".into(),
            bind_lan: Some("10.0.0.14:7701".parse().unwrap()),
            bind_tailscale: None,
            peers: vec![
                PeerEntry {
                    name: "architect".into(),
                    zone: "fabric-10g".into(),
                    lan_addr: Some("10.0.0.13:7701".parse().unwrap()),
                    tailscale_addr: None,
                },
                PeerEntry {
                    name: "architect".into(),
                    zone: "fabric-10g".into(),
                    lan_addr: Some("10.0.0.14:7701".parse().unwrap()),
                    tailscale_addr: None,
                },
            ],
            bind_rpc_lan: None,
            bind_rpc_tailscale: None,
            tls: None,
            blob_store_root: None,
            prom_bind: None,
            gc_interval_hours: None,
            blob_max_gb: None,
        };
        let err = cluster.validate().unwrap_err().to_string();
        assert!(
            err.contains("duplicate peer name"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn test_cluster_config_load_invokes_validate() {
        // Same shape as a real config file but written to a tempfile so the
        // full path — read, parse, validate — is exercised.
        let toml = r#"
[node]
name = "tank"
role = "secondary"

[hot]
path = "/hot/targets"
max_gb = 200
stale_hours = 48

[warm]
projects_path = "/slab/projects"
zfs_dataset = "slab/projects"
snapshot_retain_hours = 24
snapshot_retain_days = 7
snapshot_retain_weeks = 4

[cluster]
zone = "fabric-10g"

[[cluster.peers]]
name = "architect"
zone = "fabric-10g"
"#;
        let tmp = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(tmp.path(), toml).unwrap();
        let err = Config::load(tmp.path()).unwrap_err().to_string();
        // Fails because cluster has no bind address AND peer has no reachable address.
        assert!(err.contains("validating cluster config"), "err: {err}");
    }

    #[test]
    fn test_load_secondary_config() {
        let toml = r#"
[node]
name = "tank"
role = "secondary"

[hot]
path = "/hot/targets"
max_gb = 300
stale_hours = 48

[warm]
projects_path = "/slab/projects"
zfs_dataset = "slab/projects"
snapshot_retain_hours = 24
snapshot_retain_days = 7
snapshot_retain_weeks = 4

[replication]
send_to_host = "10.10.0.9"
send_to_user = "osobh"
cold_dataset_on_peer = "data/archive/tank-projects"
nightly_at = "03:30"
"#;
        let cfg: Config = toml::from_str(toml).unwrap();
        assert_eq!(cfg.node.role, NodeRole::Secondary);
        assert!(cfg.cold.is_none());
        let rep = cfg.replication.unwrap();
        assert_eq!(rep.send_to_host.unwrap(), "10.10.0.9");
    }
}