//! Presence and sessions (spec §8.1, §8.2), kept in memory by the daemon
//! from what agents do, and recorded as events when they change, so the
//! ledger says when an agent went quiet.

use crate::event::NewEvent;
use std::collections::HashMap;

/// A request within this long makes an agent online.
pub const ONLINE_MS: u64 = 60_000;
/// Silence this long ends a session and makes an agent offline.
pub const IDLE_AFTER_MS: u64 = 10 * 60_000;

#[derive(Clone, Debug, Default)]
struct Activity {
    last_request_ms: Option<u64>,
    waiting: u32,
    shims: u32,
    status_ms: Option<u64>,
    blocked: bool,
    paused: bool,
    session: Option<String>,
    state: Option<&'static str>,
}

/// Every agent's activity, and the state last recorded for each.
#[derive(Debug, Default)]
pub struct Presence {
    agents: HashMap<String, Activity>,
}

impl Presence {
    /// A request from `agent`: returns its session id, and a
    /// `agent.session_started` event when this begins a new one.
    pub fn request(&mut self, agent: &str, now: u64, new_session: bool) -> (String, Option<NewEvent>) {
        let a = self.agents.entry(agent.to_string()).or_default();
        let lapsed = a.last_request_ms.is_none_or(|t| now.saturating_sub(t) >= IDLE_AFTER_MS);
        a.last_request_ms = Some(now);
        if a.session.is_some() && !lapsed && !new_session {
            return (a.session.clone().unwrap_or_default(), None);
        }
        let id = crate::ids::ulid();
        a.session = Some(id.clone());
        let e = NewEvent::system("agent.session_started", serde_json::json!({ "agent": agent, "session": id }))
            .subject(agent)
            .session(id.clone());
        (id, Some(e))
    }
    pub fn wait_started(&mut self, agent: &str) {
        self.agents.entry(agent.to_string()).or_default().waiting += 1;
    }
    pub fn wait_ended(&mut self, agent: &str) {
        let a = self.agents.entry(agent.to_string()).or_default();
        a.waiting = a.waiting.saturating_sub(1);
    }
    pub fn shim_connected(&mut self, agent: &str) {
        self.agents.entry(agent.to_string()).or_default().shims += 1;
    }
    pub fn shim_gone(&mut self, agent: &str) {
        let a = self.agents.entry(agent.to_string()).or_default();
        a.shims = a.shims.saturating_sub(1);
    }
    /// The agent set its status (the `status` tool); `None` clears it.
    pub fn status(&mut self, agent: &str, at: Option<u64>) {
        self.agents.entry(agent.to_string()).or_default().status_ms = at;
    }
    pub fn blocked(&mut self, agent: &str, blocked: bool) {
        self.agents.entry(agent.to_string()).or_default().blocked = blocked;
    }
    pub fn paused(&mut self, agent: &str, paused: bool) {
        self.agents.entry(agent.to_string()).or_default().paused = paused;
    }

    /// The state `agent` is in at `now`.
    pub fn state(&self, agent: &str, now: u64) -> &'static str {
        self.agents.get(agent).map(|a| compute(a, now)).unwrap_or("offline")
    }

    /// Events for every change since the last call: `agent.presence` for a
    /// new state, `agent.session_ended` for a session gone quiet.
    pub fn tick(&mut self, now: u64) -> Vec<NewEvent> {
        let mut out = Vec::new();
        let mut names: Vec<&String> = self.agents.keys().collect();
        names.sort();
        let names: Vec<String> = names.into_iter().cloned().collect();
        for name in names {
            let a = self.agents.get_mut(&name).expect("listed");
            let s = compute(a, now);
            if a.state != Some(s) {
                let from = a.state.unwrap_or("offline");
                a.state = Some(s);
                if from != s {
                    out.push(
                        NewEvent::system("agent.presence", serde_json::json!({ "agent": name, "state": s, "from": from }))
                            .subject(name.clone()),
                    );
                }
            }
            let quiet = a.last_request_ms.is_none_or(|t| now.saturating_sub(t) >= IDLE_AFTER_MS);
            if quiet && a.waiting == 0 {
                if let Some(id) = a.session.take() {
                    out.push(
                        NewEvent::system("agent.session_ended", serde_json::json!({ "agent": name, "session": id }))
                            .subject(name.clone())
                            .session(id),
                    );
                }
            }
        }
        out
    }
}

/// Paused, then blocked, then busy, then online, then idle, then offline.
fn compute(a: &Activity, now: u64) -> &'static str {
    let within = |t: Option<u64>, ms: u64| t.is_some_and(|t| now.saturating_sub(t) < ms);
    if a.paused {
        "paused"
    } else if a.blocked {
        "blocked"
    } else if within(a.status_ms, IDLE_AFTER_MS) {
        "busy"
    } else if a.waiting > 0 || within(a.last_request_ms, ONLINE_MS) {
        "online"
    } else if a.shims > 0 || within(a.last_request_ms, IDLE_AFTER_MS) {
        "idle"
    } else {
        "offline"
    }
}

/// Apply presence, status and session events to their tables.
pub async fn project(conn: &mut sqlx::SqliteConnection, e: &crate::event::Event) -> sqlx::Result<()> {
    let p = e.json();
    let s = |k: &str| p.get(k).and_then(|v| v.as_str()).map(String::from);
    // A projection skips what it cannot use: a malformed payload must never
    // stop the ledger from taking events.
    match e.kind.as_str() {
        "agent.presence" if s("agent").is_some() && s("state").is_some() => {
            sqlx::query("INSERT INTO presence (agent, state, since_ms) VALUES (?, ?, ?) ON CONFLICT(agent) DO UPDATE SET state = excluded.state, since_ms = excluded.since_ms")
                .bind(s("agent"))
                .bind(s("state"))
                .bind(e.at_ms as i64)
                .execute(&mut *conn)
                .await?;
        }
        "agent.status" if e.actor.starts_with("agent:") => {
            let agent = e.actor.strip_prefix("agent:").map(String::from);
            sqlx::query(
                "INSERT INTO presence (agent, state, since_ms, status_state, status_text, status_task, status_ms) VALUES (?, 'online', ?, ?, ?, ?, ?) \
                 ON CONFLICT(agent) DO UPDATE SET status_state = excluded.status_state, status_text = excluded.status_text, status_task = excluded.status_task, status_ms = excluded.status_ms",
            )
            .bind(agent)
            .bind(e.at_ms as i64)
            .bind(s("state"))
            .bind(s("text"))
            .bind(s("task"))
            .bind(e.at_ms as i64)
            .execute(&mut *conn)
            .await?;
        }
        "agent.session_started" if s("agent").is_some() && s("session").is_some() => {
            sqlx::query("INSERT OR IGNORE INTO sessions (id, agent, started_ms) VALUES (?, ?, ?)")
                .bind(s("session"))
                .bind(s("agent"))
                .bind(e.at_ms as i64)
                .execute(&mut *conn)
                .await?;
        }
        "agent.session_ended" => {
            sqlx::query("UPDATE sessions SET ended_ms = ? WHERE id = ?")
                .bind(e.at_ms as i64)
                .bind(s("session"))
                .execute(&mut *conn)
                .await?;
        }
        _ => {}
    }
    Ok(())
}

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

    fn kinds(v: &[NewEvent]) -> Vec<String> {
        v.iter().map(|e| format!("{} {}", e.kind, e.payload.get("state").and_then(|s| s.as_str()).unwrap_or(""))).collect()
    }

    #[test]
    fn presence_moves_through_its_states_and_records_each_change() {
        let mut p = Presence::default();
        let t = 1_000_000;
        let (s1, started) = p.request("claude", t, false);
        assert!(started.is_some());
        assert_eq!(kinds(&p.tick(t)), ["agent.presence online"]);
        assert!(p.tick(t + 1).is_empty(), "no change, no event");
        assert_eq!(p.state("claude", t + ONLINE_MS + 1), "idle");
        assert_eq!(kinds(&p.tick(t + ONLINE_MS + 1)), ["agent.presence idle"]);
        p.status("claude", Some(t + ONLINE_MS + 2));
        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "busy");
        p.blocked("claude", true);
        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "blocked");
        p.blocked("claude", false);
        p.status("claude", None);
        p.paused("claude", true);
        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "paused");
        p.paused("claude", false);
        let late = t + IDLE_AFTER_MS + 5;
        let ev = p.tick(late);
        assert_eq!(kinds(&ev), ["agent.presence offline", "agent.session_ended "]);
        let (s2, started) = p.request("claude", late + 1, false);
        assert!(started.is_some() && s2 != s1, "a new session after ten minutes' silence");
        let (s3, none) = p.request("claude", late + 2, false);
        assert!(none.is_none() && s3 == s2);
        let (s4, again) = p.request("claude", late + 3, true);
        assert!(again.is_some() && s4 != s3, "brief(new_session: true)");
    }

    #[test]
    fn a_waiting_or_connected_agent_is_not_offline() {
        let mut p = Presence::default();
        p.request("kimi", 0, false);
        p.wait_started("kimi");
        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "online");
        p.wait_ended("kimi");
        p.shim_connected("kimi");
        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "idle");
        p.shim_gone("kimi");
        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "offline");
    }
}