liminal-server 0.25.4

//! One verified agent per connection, with departure told by the transport.

use std::collections::BTreeMap;
use std::sync::Mutex;

use serde_json::json;

use super::error::AgentError;
use super::records::{AgentEvent, AgentTransition};
use super::registry::AgentRegistry;
use super::retained::Retained;

/// Why a person's older window lost the seat to a newer one.
const SUPERSEDED: &str = "superseded by a newer window of the same person";

/// Live seats only. An inbox unsubscribe does not pretend the connection closed.
#[derive(Debug, Default)]
pub struct Presence {
    connections: Mutex<BTreeMap<String, Seat>>,
}

/// The connection holding one agent's seat, and when it was admitted: what a
/// refused second connection is told, so the holder is named.
#[derive(Clone, Copy, Debug)]
struct Seat {
    connection: u64,
    since_ms: u64,
}

/// Identity and time already proved by the subscription's admission.
#[derive(Clone, Copy)]
pub struct ConnectionObservation<'a> {
    /// The agent named by the pass.
    pub name: &'a str,
    /// The verified actor.
    pub holder: &'a str,
    /// The connection process.
    pub connection: u64,
    /// Epoch milliseconds of admission.
    pub at_ms: u64,
}

impl Presence {
    /// The agent actually present on this connection, if one.
    ///
    /// # Errors
    /// The presence lock is poisoned.
    pub fn agent(&self, connection: u64) -> Result<Option<String>, AgentError> {
        Ok(self
            .connections
            .lock()
            .map_err(|_| AgentError::Store("presence lock poisoned".into()))?
            .iter()
            .find(|(_, own)| own.connection == connection)
            .map(|(name, _)| name.clone()))
    }
    /// The first successful inbox subscription records presence once.
    /// `attach` installs the real inbox before a presence transition is landed.
    /// The lock covers duplicate admission, installation, and durable recording.
    ///
    /// # Errors
    /// A duplicate seat, unknown agent, failed subscription, or unresolved write.
    pub fn connect<T>(
        &self,
        registry: &AgentRegistry,
        retained: &Retained,
        observation: ConnectionObservation<'_>,
        attach: impl FnOnce() -> Result<T, AgentError>,
    ) -> Result<T, AgentError> {
        let ConnectionObservation {
            name,
            holder,
            connection,
            at_ms,
        } = observation;
        if connection == 0 {
            return Err(AgentError::Invalid("connection id"));
        }
        if holder != name {
            return Err(AgentError::Identity {
                holder: holder.into(),
                agent: name.into(),
            });
        }
        let mut connections = self
            .connections
            .lock()
            .map_err(|_| AgentError::Store("presence lock poisoned".into()))?;
        if let Some(&existing) = connections.get(name) {
            if existing.connection == connection {
                return attach();
            }
            // An agent holds one seat. A person opens windows: the newest one takes the seat and
            // the older connection is recorded gone, so a second tab never refuses the person.
            if !name.starts_with("person-") {
                return Err(AgentError::Present {
                    agent: name.into(),
                    connection: existing.connection,
                    since_ms: existing.since_ms,
                });
            }
            registry.observe(
                name,
                AgentTransition {
                    holder: "server".into(),
                    at_ms,
                    change: AgentEvent::Gone {
                        connection: existing.connection,
                        reason: SUPERSEDED.into(),
                    },
                },
            )?;
            connections.remove(name);
        }
        if connections
            .values()
            .any(|existing| existing.connection == connection)
        {
            return Err(AgentError::Transition(
                "one connection cannot claim two agents",
            ));
        }
        let record = registry.record(name)?;
        if !record.active() {
            return Err(AgentError::Transition("agent is retired"));
        }
        let subscription = attach()?;
        registry.observe(
            name,
            AgentTransition {
                holder: holder.into(),
                at_ms,
                change: AgentEvent::Present { connection },
            },
        )?;
        connections.insert(
            name.into(),
            Seat {
                connection,
                since_ms: at_ms,
            },
        );
        retained.publish(&[
            (
                format!("agents.{name}.presence"),
                json!({"present":true,"at_ms":at_ms,"connection":connection}),
            ),
            (
                format!("agents.{name}.state"),
                presence_state(retained, name, false)?,
            ),
        ])?;
        drop(connections);
        Ok(subscription)
    }

    /// Publishes absent with the transport's stable reason. A stale close
    /// cannot remove another connection, and repeated closes land nothing.
    ///
    /// # Errors
    /// A malformed reason, unavailable store, or failed delivery.
    pub fn close(
        &self,
        registry: &AgentRegistry,
        retained: &Retained,
        connection: u64,
        reason: &str,
        at_ms: u64,
    ) -> Result<(), AgentError> {
        let mut connections = self
            .connections
            .lock()
            .map_err(|_| AgentError::Store("presence lock poisoned".into()))?;
        let name = connections
            .iter()
            .find(|(_, own)| own.connection == connection)
            .map(|(name, _)| name.clone());
        let Some(name) = name else {
            return Ok(());
        };
        registry.observe(
            &name,
            AgentTransition {
                holder: "server".into(),
                at_ms,
                change: AgentEvent::Gone {
                    connection,
                    reason: reason.into(),
                },
            },
        )?;
        connections.remove(&name);
        let result = retained.publish(&[
            (
                format!("agents.{name}.presence"),
                json!({"present":false,"at_ms":at_ms,"reason":reason}),
            ),
            (
                format!("agents.{name}.state"),
                presence_state(retained, &name, true)?,
            ),
        ]);
        drop(connections);
        result
    }

    /// The observed seat, without reading the durable history.
    ///
    /// # Errors
    /// The presence lock is unavailable.
    pub fn connection(&self, name: &str) -> Result<Option<u64>, AgentError> {
        Ok(self
            .connections
            .lock()
            .map_err(|_| AgentError::Store("presence lock poisoned".into()))?
            .get(name)
            .map(|seat| seat.connection))
    }
}

fn presence_state(
    retained: &Retained,
    name: &str,
    gone: bool,
) -> Result<serde_json::Value, AgentError> {
    let mut state = retained
        .value(&format!("agents.{name}.state"))?
        .unwrap_or_else(|| json!({}));
    if state["state"] != "waiting" {
        state["state"] = json!(if gone { "gone" } else { "present" });
    }
    Ok(state)
}