//! 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");

    }

}