//! The daemon's state (spec §7.1): the ledger it alone writes, the

//! principals kept current from the bus, the authenticator, and presence

//! with its timers. HTTP lives elsewhere (`spectra-mcp`, `radar`); this is

//! what they share.

use crate::authn::{Authenticator, Caller, Refusal};

use crate::event::{Auth, Event, Method, NewEvent};

use crate::keys::KeyPair;

use crate::ledger::{KeyResolver, Ledger};

use crate::presence::Presence;

use crate::principals::{Principal, Principals};

use crate::proof::Proof;

use anyhow::{bail, Context, Result};

use serde::{Deserialize, Serialize};

use std::path::{Path, PathBuf};

use std::sync::Arc;

use tokio::sync::{Mutex, RwLock};

/// Kinds the owner's CLI may hand the daemon, already signed over their

/// payload (spec §8.5, "One writer").

pub const OWNER_KINDS: &[&str] = &[

    "identity.rotated",

    "agent.added",

    "agent.disabled",

    "agent.scoped",

    "project.added",

    "project.linked",

    "project.unlinked",

    "brief.set",

];

pub struct Daemon {

    pub data: PathBuf,

    pub ledger: Ledger,

    pub principals: RwLock<Principals>,

    pub authn: Authenticator,

    pub presence: Mutex<Presence>,

    /// Per agent: empty `wait`s in a row, and until when `wait` is refused.

    pub idle_waits: Mutex<std::collections::HashMap<String, (u32, u64)>>,

}

impl Daemon {

    /// Open the data directory's ledger for writing and start the timers.

    pub async fn start(data: &Path) -> Result<Arc<Daemon>> {

        let db = data.join("spectra.db");

        if !db.exists() {

            bail!("no Spectra in {} (run `spectra init`)", data.display());

        }

        let instance = KeyPair::instance(data, "spectra")?;

        let ledger = Ledger::open(&db, Some(instance)).await?;

        let principals = Principals::load(&ledger).await?;

        let d = Arc::new(Daemon {

            data: data.to_path_buf(),

            ledger,

            principals: RwLock::new(principals),

            authn: Authenticator::default(),

            presence: Mutex::new(Presence::default()),

            idle_waits: Mutex::new(Default::default()),

        });

        d.clone().watch_principals();

        d.clone().tick_presence();

        Ok(d)

    }

    /// Reload who is who whenever an identity or agent event is appended.

    fn watch_principals(self: Arc<Self>) {

        let mut rx = self.ledger.subscribe();

        tokio::spawn(async move {

            loop {

                match rx.recv().await {

                    Ok(e)

                        if e.kind.starts_with("identity.")

                            || e.kind.starts_with("agent.") && e.kind != "agent.presence" && e.kind != "agent.status" =>

                    {

                        if let Ok(p) = Principals::load(&self.ledger).await {

                            *self.principals.write().await = p;

                        }

                    }

                    Ok(_) => {}

                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {

                        if let Ok(p) = Principals::load(&self.ledger).await {

                            *self.principals.write().await = p;

                        }

                    }

                    Err(_) => return,

                }

            }

        });

    }

    /// Record presence changes and quiet sessions every 15 s, and write a

    /// checkpoint when one is due by time.

    fn tick_presence(self: Arc<Self>) {

        tokio::spawn(async move {

            let mut every = tokio::time::interval(std::time::Duration::from_secs(15));

            loop {

                every.tick().await;

                let events = self.presence.lock().await.tick(crate::ids::now_ms());

                if !events.is_empty() {

                    let _ = self.ledger.append(events).await;

                }

                let _ = self.ledger.append(vec![]).await;

            }

        });

    }

    /// Authenticate a request (spec §8.5); an agent used from the wrong Unix

    /// user is recorded as `guard.impersonation`, and a first connection

    /// that binds an agent's uid as `agent.uid_bound`.

    pub async fn authenticate(&self, proof: Option<&Proof>, method: &str, body: &[u8], peer: Option<Peer>) -> Result<Caller, Refusal> {

        let now = crate::ids::now_ms();

        let r = {

            let p = self.principals.read().await;

            if p.owner().is_none() {

                return Err(Refusal::NotAPrincipal("this Spectra has no owner yet: its owner claims it with `spectra connect`".into()));

            }

            self.authn.authenticate(&p, proof, method, body, now, peer.map(|x| x.uid))

        };

        match &r {

            Err(Refusal::WrongUid { agent, bound, from }) => {

                let e = NewEvent::system(

                    "guard.impersonation",

                    serde_json::json!({ "agent": agent, "bound_uid": bound, "from_uid": from, "from_pid": peer.map(|x| x.pid), "method": method }),

                )

                .subject(agent.clone());

                let _ = self.ledger.append_one(e).await;

                tracing::warn!(agent, bound, from, "a key used from the wrong Unix user");

            }

            Err(Refusal::BoundOverNetwork { agent, bound }) => {

                let e = NewEvent::system(

                    "guard.impersonation",

                    serde_json::json!({ "agent": agent, "bound_uid": bound, "from_uid": null, "over": "tcp", "method": method }),

                )

                .subject(agent.clone());

                let _ = self.ledger.append_one(e).await;

                tracing::warn!(agent, bound, "a uid-bound key used over the network");

            }

            Ok(Caller { principal: Principal::Agent { name, .. }, learn_uid: Some(uid), .. }) => {

                let e = NewEvent::system("agent.uid_bound", serde_json::json!({ "name": name, "uid": uid })).subject(name.clone());

                let _ = self.ledger.append_one(e).await;

            }

            _ => {}

        }

        r

    }

    /// Note a request from an agent: its session, and the event that

    /// starts one when this is the first in ten minutes.

    pub async fn touch(&self, who: &Principal, new_session: bool) -> Option<String> {

        let Principal::Agent { name, .. } = who else { return None };

        let (id, started) = self.presence.lock().await.request(name, crate::ids::now_ms(), new_session);

        if let Some(e) = started {

            let _ = self.ledger.append_one(e).await;

        }

        Some(id)

    }

    /// Append events the owner's CLI signed over their payload, after

    /// checking each as `spectra admin verify` would.

    pub async fn accept_owner_events(&self, events: Vec<WireEvent>) -> Result<Vec<u64>> {

        let mut principals = self.principals.read().await.clone();

        let projects: Vec<String> = events

            .iter()

            .filter(|e| e.kind == "project.added")

            .filter_map(|e| e.payload.get("name").and_then(|n| n.as_str()).map(String::from))

            .collect();

        let mut out = Vec::new();

        for w in events {

            let n = w.into_new_event()?;

            if n.kind == "channel.created" {

                // A project's channel comes with its project, unsigned, as `project add` makes it.

                let p = &n.payload;

                let ok = p.get("kind").and_then(|k| k.as_str()) == Some("project")

                    && p.get("name").and_then(|x| x.as_str()).is_some_and(|x| projects.iter().any(|q| q == x));

                if !ok || n.auth.method != Method::System {

                    bail!("a channel is only made with its project");

                }

                out.push(n);

                continue;

            }

            if !OWNER_KINDS.contains(&n.kind.as_str()) {

                bail!("{} cannot be handed to the daemon", n.kind);

            }

            if n.actor != "human" {

                bail!("only the owner hands events to the daemon");

            }

            if n.kind == "brief.set" {

                crate::rules::check(n.payload.get("text").and_then(|t| t.as_str()).unwrap_or(""))?;

            }

            // Check it as verify will, against who is who including the

            // batch so far.

            let probe = probe_event(&n);

            principals.check(&probe).map_err(|why| anyhow::anyhow!("refused {}: {why}", n.kind))?;

            if n.kind != "identity.rotated" {

                let key = crate::identity::KeyId(n.auth.key.context("unsigned")?);

                if !matches!(principals.principal(&key, crate::ids::now_ms()), Ok(Principal::Owner { .. })) {

                    bail!("refused {}: not signed by the owner", n.kind);

                }

            }

            out.push(n);

        }

        let appended = self.ledger.append(out).await?;

        let _ = self.ledger.checkpoint().await;

        // Who is who has changed: know it before the next request arrives.

        *self.principals.write().await = Principals::load(&self.ledger).await?;

        Ok(appended.iter().map(|e| e.seq).collect())

    }

    /// The owner `spectra init --owner` named, while none is recorded.

    pub fn named_owner(&self) -> Option<String> {

        let text = std::fs::read_to_string(self.data.join("spectra.toml")).ok()?;

        let t: toml::Table = text.parse().ok()?;

        t.get("owner")?.get("key")?.as_str().map(|k| k.to_ascii_lowercase())

    }

    /// The owner claims a Spectra set up with `spectra init --owner`

    /// (spec §8.6): the named identity, signed by itself over its payload,

    /// then any successors up to the owner's current identity, each a

    /// rotation as `identity refresh` makes. Accepted only while no owner

    /// is recorded, and only from the key init named.

    pub async fn claim(&self, events: Vec<WireEvent>) -> Result<Vec<u64>> {

        let mut principals = self.principals.read().await.clone();

        if principals.owner().is_some() {

            bail!("this Spectra already has an owner");

        }

        let named = self.named_owner().context("this Spectra names no owner (spectra init --owner)")?;

        let mut out = Vec::new();

        for (i, w) in events.into_iter().enumerate() {

            let n = w.into_new_event()?;

            let want = if i == 0 { "identity.owner" } else { "identity.rotated" };

            if n.kind != want || n.actor != "human" {

                bail!("a claim is the owner's identity.owner, then their rotations");

            }

            if i == 0 {

                let key = n.payload.get("key").and_then(|k| k.as_str()).unwrap_or("").to_ascii_lowercase();

                if key != named || n.auth.key.map(hex::encode) != Some(named.clone()) {

                    bail!("this Spectra was set up for another owner");

                }

            }

            principals.check(&probe_event(&n)).map_err(|why| anyhow::anyhow!("refused the claim: {why}"))?;

            out.push(n);

        }

        if out.is_empty() {

            bail!("an empty claim");

        }

        let appended = self.ledger.append(out).await?;

        let _ = self.ledger.checkpoint().await;

        *self.principals.write().await = Principals::load(&self.ledger).await?;

        Ok(appended.iter().map(|e| e.seq).collect())

    }

    /// Write a checkpoint covering everything (at shutdown).

    pub async fn shutdown(&self) {

        let _ = self.ledger.checkpoint().await;

    }

}

/// The Unix peer of a socket connection.

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub struct Peer {

    pub uid: u32,

    pub pid: Option<i32>,

}

/// An event as the CLI hands it to the daemon: signed already.

#[derive(Clone, Debug, Serialize, Deserialize)]

pub struct WireEvent {

    pub kind: String,

    pub actor: String,

    pub subject: Option<String>,

    pub project: Option<String>,

    pub payload: serde_json::Value,

    /// Hex of the auth key and the encoded proof; absent for a project's channel.

    pub key: Option<String>,

    pub proof: Option<String>,

}

impl WireEvent {

    pub fn from_new(n: &NewEvent) -> WireEvent {

        WireEvent {

            kind: n.kind.clone(),

            actor: n.actor.clone(),

            subject: n.subject.clone(),

            project: n.project.clone(),

            payload: n.payload.clone(),

            key: n.auth.key.map(hex::encode),

            proof: n.auth.proof.as_ref().map(hex::encode),

        }

    }

    fn into_new_event(self) -> Result<NewEvent> {

        let auth = match (&self.key, &self.proof) {

            (Some(k), Some(p)) => Auth {

                key: Some(hex::decode(k).ok().and_then(|b| b.try_into().ok()).context("a key is 32 bytes of hex")?),

                method: Method::Signature,

                assurance: 2,

                proof: Some(hex::decode(p).context("a proof is hex")?),

            },

            _ => Auth::system(),

        };

        let mut n = NewEvent::system(&self.kind, self.payload).by(self.actor, auth);

        n.subject = self.subject;

        n.project = self.project;

        Ok(n)

    }

}

/// The event `n` would be, for checking before it is appended.

fn probe_event(n: &NewEvent) -> Event {

    let now = crate::ids::now_ms();

    Event {

        seq: 0,

        id: String::new(),

        at_ms: now,

        hlc: crate::hlc::Hlc { wall_ms: now, logical: 0 },

        prev: [0; 32],

        hash: [0; 32],

        auth: n.auth.clone(),

        kind: n.kind.clone(),

        actor: n.actor.clone(),

        session: n.session.clone(),

        project: n.project.clone(),

        subject: n.subject.clone(),

        payload: Some(crate::event::canonical_json(&n.payload)),

        trace: n.trace.clone(),

    }

}