hyperhive/hive-matrix-mcp/src/accounts.rs

//! Multi-account configuration + the dispatch registry.
//!
//! A single `hive-matrix-daemon` can serve N matrix accounts (one
//! matrix-sdk `Client` each, with its own session/store dir + sync
//! loop). The **hive-internal account** (named `main`) is always
//! present and is the primary: it is synthesized from the per-agent
//! single-account paths (`<state>/matrix-token` +
//! `<state>/matrix-sdk-state`) and the daemon-wide `HIVE_MATRIX_URL`.
//! Any **extra** accounts come from the `HIVE_MATRIX_ACCOUNTS` env var
//! (JSON, written by the nix harness module from
//! `hyperhive.matrixAccounts`) and are appended after `main`.
//!
//! So a single-account agent (no `HIVE_MATRIX_ACCOUNTS`) gets exactly
//! `main` — zero config, same behaviour as before. The **primary** is
//! `main`: it is selected when a tool call omits `account`, so callers
//! never specify one for the hive account.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::Context as _;
use matrix_sdk::Client;
use serde::{Deserialize, Serialize};

use crate::paths;

/// One declared matrix account. `homeserver` is optional per account
/// (defaults to the daemon-wide `HIVE_MATRIX_URL`) so accounts on the
/// same homeserver need not repeat it.
#[derive(Debug, Clone, Deserialize)]
pub struct AccountCfg {
    /// Logical name the agent uses to address this account
    /// (`account` arg on the MCP tools). Unique within the daemon.
    pub name: String,
    /// Path to the bearer-token file for this account (hive-c0re
    /// writes it; the daemon reads it).
    pub token_file: PathBuf,
    /// Per-account matrix-sdk sqlite store dir (crypto keys + cache).
    pub state_dir: PathBuf,
    /// Homeserver URL; falls back to [`paths::homeserver_url`] when absent.
    #[serde(default)]
    pub homeserver: Option<String>,
}

impl AccountCfg {
    /// Resolve the effective homeserver URL (per-account override or
    /// the daemon-wide default).
    #[must_use]
    pub fn homeserver(&self) -> String {
        self.homeserver
            .clone()
            .unwrap_or_else(paths::homeserver_url)
    }
}

/// Build the account list: the always-present hive-internal `main`
/// account (primary, synthesized from the per-agent single-account
/// paths) followed by any extras declared in `HIVE_MATRIX_ACCOUNTS`.
///
/// With no `HIVE_MATRIX_ACCOUNTS` set this returns just `main`, so a
/// single-account agent is unchanged. When set, the env var carries
/// only the *extra* accounts (the nix `matrixAccounts` option never
/// redeclares the hive account); they are appended after `main`, which
/// stays index 0 = primary.
///
/// # Errors
///
/// Returns an error if `HIVE_MATRIX_ACCOUNTS` is set but is not valid
/// JSON, or if an extra account's name collides with `main` or another
/// extra.
pub fn configured() -> anyhow::Result<Vec<AccountCfg>> {
    // The hive-internal account: always primary, named `main`, built
    // from the legacy single-account paths + the daemon-wide homeserver.
    let hive = AccountCfg {
        name: "main".to_owned(),
        token_file: paths::token_file(),
        state_dir: paths::matrix_state_dir(),
        homeserver: None,
    };
    let Some(raw) = std::env::var_os("HIVE_MATRIX_ACCOUNTS") else {
        return Ok(vec![hive]);
    };
    let extras: Vec<AccountCfg> = serde_json::from_str(&raw.to_string_lossy())
        .map_err(|e| anyhow::anyhow!("parse HIVE_MATRIX_ACCOUNTS as JSON array: {e}"))?;
    let mut accounts = Vec::with_capacity(extras.len() + 1);
    accounts.push(hive);
    accounts.extend(extras);
    let mut seen = std::collections::HashSet::new();
    for a in &accounts {
        if !seen.insert(a.name.as_str()) {
            anyhow::bail!(
                "duplicate matrix account name {:?} (the hive-internal account is named \"main\")",
                a.name
            );
        }
    }
    Ok(accounts)
}

/// Live status of one matrix account, as reported by [`Registry::list`]
/// (the `list_accounts` daemon op). Only accounts that successfully
/// restored a session appear, so `live` is always `true` today; the
/// field is kept so a future "configured but down" entry can report
/// `false` without a wire-shape change.
#[derive(Debug, Serialize)]
pub struct AccountStatus {
    /// Logical account name (the `account` arg on the MCP tools).
    pub name: String,
    /// Effective homeserver URL the restored client is talking to.
    pub homeserver: String,
    /// The account's own matrix user id (`@user:server`), when known.
    pub user_id: Option<String>,
    /// Whether the account has a live, restored client. Always `true`
    /// for registry entries today (the registry only holds restored
    /// accounts); reserved for future configured-but-down reporting.
    pub live: bool,
    /// Whether this is the primary account (selected when a tool call
    /// omits `account`).
    pub is_primary: bool,
}

/// Account name → live `Client` map plus the primary-account name used
/// when a request omits `account`. Built once at daemon startup from
/// the accounts that successfully restored a session.
pub struct Registry {
    primary: String,
    by_name: HashMap<String, Arc<Client>>,
}

impl Registry {
    /// Build an empty registry whose primary is `primary`. Clients are
    /// added with [`Registry::insert`] as each account restores.
    #[must_use]
    pub fn new(primary: String) -> Self {
        Self {
            primary,
            by_name: HashMap::new(),
        }
    }

    /// Register a restored client under `name`.
    pub fn insert(&mut self, name: String, client: Client) {
        self.by_name.insert(name, Arc::new(client));
    }

    /// Whether any account restored successfully.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.by_name.is_empty()
    }

    /// Snapshot every restored account: name, homeserver, user id, and
    /// primary flag. Registry membership == a session restored, so every
    /// entry is reported `live`. Sorted primary-first then by name for a
    /// stable order in the dashboard. Account-agnostic — the caller does
    /// not resolve a single client (see `socket::dispatch`).
    #[must_use]
    pub fn list(&self) -> Vec<AccountStatus> {
        let mut out: Vec<AccountStatus> = self
            .by_name
            .iter()
            .map(|(name, client)| AccountStatus {
                name: name.clone(),
                homeserver: client.homeserver().to_string(),
                user_id: client.user_id().map(ToString::to_string),
                live: true,
                is_primary: *name == self.primary,
            })
            .collect();
        out.sort_by(|a, b| {
            b.is_primary
                .cmp(&a.is_primary)
                .then_with(|| a.name.cmp(&b.name))
        });
        out
    }

    /// Resolve a request's `account` to a client. `None` → the primary
    /// account. Returns a human-readable error (listing known accounts)
    /// when the name is unknown — surfaced to the agent as a tool error.
    ///
    /// # Errors
    ///
    /// Errors when the named account (or the primary, if `None`) has no
    /// live client — e.g. an unknown name, or the primary failed to
    /// restore at startup.
    pub fn resolve(&self, account: Option<&str>) -> Result<&Arc<Client>, String> {
        let name = account.unwrap_or(&self.primary);
        self.by_name.get(name).ok_or_else(|| {
            let mut known: Vec<&str> = self.by_name.keys().map(String::as_str).collect();
            known.sort_unstable();
            format!(
                "unknown matrix account {name:?}; available accounts: [{}]",
                known.join(", ")
            )
        })
    }

    /// Publish the live-account snapshot ([`Registry::list`]) to `path`.
    /// Thin wrapper over [`write_accounts_snapshot`] (split out so the
    /// atomic-write logic is unit-testable without a live `Client`).
    /// Called once at daemon startup after all restores.
    ///
    /// # Errors
    /// Propagates filesystem errors from the atomic write.
    pub fn write_snapshot(&self, path: &Path) -> anyhow::Result<()> {
        write_accounts_snapshot(path, &self.list())
    }

    /// Heartbeat the snapshot: rewrite it unconditionally so the file
    /// mtime advances even when the account set is unchanged. The daemon
    /// calls this on a timer (see `main`) so the dashboard's `as_of`
    /// (derived from the file mtime) tracks real daemon liveness instead
    /// of freezing at the boot-time write — a stalled mtime then means
    /// the daemon is down, which the dashboard can render as stale/dimmed.
    ///
    /// # Errors
    /// Propagates filesystem errors from the atomic write.
    pub fn heartbeat_snapshot(&self, path: &Path) -> anyhow::Result<()> {
        heartbeat_accounts_snapshot(path, &self.list())
    }
}

/// Atomically write `accounts` as pretty-printed JSON to `path`
/// (`<path>.tmp` + rename) so a dashboard reader never sees a partial
/// file. Idempotent — skips the rewrite when the on-disk content already
/// matches, keeping the mtime stable. Used for the boot-time publish.
///
/// The daemon rebuilds this file fresh on every boot (and is restarted
/// by the `matrix-token*` path-watcher when a new account is
/// provisioned), so the file lists exactly the accounts that restored at
/// the last start. Real-time liveness is conveyed by the file mtime,
/// which the daemon advances on a timer via
/// [`heartbeat_accounts_snapshot`] — a stalled mtime means the daemon is
/// down, so a reader can treat an old snapshot as stale.
///
/// # Errors
/// Returns an error if the parent dir can't be created or the
/// write/rename fails.
pub fn write_accounts_snapshot(path: &Path, accounts: &[AccountStatus]) -> anyhow::Result<()> {
    write_accounts_snapshot_inner(path, accounts, false)
}

/// Like [`write_accounts_snapshot`] but always rewrites (tmp + rename)
/// even when the on-disk content is byte-identical, so the file mtime
/// advances. The daemon calls this on a periodic heartbeat so the
/// dashboard's `as_of` (the file mtime) reflects daemon liveness rather
/// than freezing at the boot-time write.
///
/// # Errors
/// Returns an error if the parent dir can't be created or the
/// write/rename fails.
pub fn heartbeat_accounts_snapshot(path: &Path, accounts: &[AccountStatus]) -> anyhow::Result<()> {
    write_accounts_snapshot_inner(path, accounts, true)
}

/// Shared body for [`write_accounts_snapshot`] (idempotent) and
/// [`heartbeat_accounts_snapshot`] (`force`). When `force` is false the
/// rewrite is skipped if the on-disk content already matches, keeping the
/// mtime stable; when true the tmp + rename always runs so the mtime
/// advances.
fn write_accounts_snapshot_inner(
    path: &Path,
    accounts: &[AccountStatus],
    force: bool,
) -> anyhow::Result<()> {
    let body =
        serde_json::to_string_pretty(accounts).expect("Vec<AccountStatus> is always serialisable");
    if !force && std::fs::read_to_string(path).ok().as_deref() == Some(body.as_str()) {
        return Ok(());
    }
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
    }
    let tmp = path.with_extension("json.tmp");
    std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
    std::fs::rename(&tmp, path).with_context(|| {
        format!(
            "rename {} -> {} (atomic publish)",
            tmp.display(),
            path.display()
        )
    })?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{AccountStatus, heartbeat_accounts_snapshot, write_accounts_snapshot};

    fn sample() -> Vec<AccountStatus> {
        vec![
            AccountStatus {
                name: "main".to_owned(),
                homeserver: "http://localhost:8008".to_owned(),
                user_id: Some("@agent:hive".to_owned()),
                live: true,
                is_primary: true,
            },
            AccountStatus {
                name: "public".to_owned(),
                homeserver: "https://matrix.org".to_owned(),
                user_id: None,
                live: true,
                is_primary: false,
            },
        ]
    }

    #[test]
    fn snapshot_writes_json_and_cleans_up_tmp() {
        let dir = std::env::temp_dir().join(format!(
            "hh-acct-snap-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("matrix-accounts.json");

        write_accounts_snapshot(&path, &sample()).unwrap();
        let written = std::fs::read_to_string(&path).unwrap();
        // Round-trips to the same shape the dashboard consumes.
        let parsed: serde_json::Value = serde_json::from_str(&written).unwrap();
        assert_eq!(parsed[0]["name"], "main");
        assert_eq!(parsed[0]["live"], true);
        assert_eq!(parsed[0]["is_primary"], true);
        assert_eq!(parsed[1]["homeserver"], "https://matrix.org");
        assert!(parsed[1]["user_id"].is_null());
        // No leftover temp file after the atomic publish.
        assert!(!path.with_extension("json.tmp").exists());

        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn snapshot_is_idempotent_when_unchanged() {
        let dir = std::env::temp_dir().join(format!(
            "hh-acct-idem-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("matrix-accounts.json");

        write_accounts_snapshot(&path, &sample()).unwrap();
        let mtime1 = std::fs::metadata(&path).unwrap().modified().unwrap();
        // Second write with identical content must skip the rename (mtime
        // stays put), so inotify watchers don't see a spurious change.
        write_accounts_snapshot(&path, &sample()).unwrap();
        let mtime2 = std::fs::metadata(&path).unwrap().modified().unwrap();
        assert_eq!(mtime1, mtime2);

        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn heartbeat_advances_mtime_even_when_unchanged() {
        let dir = std::env::temp_dir().join(format!(
            "hh-acct-hb-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("matrix-accounts.json");

        write_accounts_snapshot(&path, &sample()).unwrap();
        let mtime1 = std::fs::metadata(&path).unwrap().modified().unwrap();
        // The heartbeat must rewrite (tmp + rename) even when the content
        // is byte-identical, so the mtime advances and the dashboard reads
        // a fresh `as_of`. Small sleep so the new mtime is strictly later
        // than the first (tmpfs/ext4 have sub-second mtime resolution).
        std::thread::sleep(std::time::Duration::from_millis(20));
        heartbeat_accounts_snapshot(&path, &sample()).unwrap();
        let mtime2 = std::fs::metadata(&path).unwrap().modified().unwrap();
        assert!(mtime2 > mtime1, "heartbeat should advance mtime");
        // Content is still the same shape, and no leftover temp file.
        let written = std::fs::read_to_string(&path).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&written).unwrap();
        assert_eq!(parsed[0]["name"], "main");
        assert!(!path.with_extension("json.tmp").exists());

        std::fs::remove_dir_all(&dir).ok();
    }
}