520 lines
21 KiB
Rust
520 lines
21 KiB
Rust
//! 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: omitting `account` on
|
|
//! a tool call resolves to `main`. When **multiple** accounts are
|
|
//! configured, omitting `account` is rejected with the list of names (see
|
|
//! `Registry::resolve` / `pick_name`) — the agent can't tell more than one
|
|
//! account exists, so it must choose explicitly.
|
|
|
|
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 mut accounts = vec![hive];
|
|
if let Some(raw) = std::env::var_os("HIVE_MATRIX_ACCOUNTS") {
|
|
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}"))?;
|
|
accounts.extend(extras);
|
|
}
|
|
// Reject duplicate names among the *configured* set (hive + 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
|
|
);
|
|
}
|
|
}
|
|
// Append dashboard-provisioned accounts (a `matrix-token-<name>` file +
|
|
// its `matrix-account-<name>.json` homeserver sidecar) that aren't
|
|
// already declared in config, so an account logged in via the dashboard
|
|
// form works without a `matrixAccounts` edit + rebuild. Explicit config
|
|
// wins on name collision.
|
|
let configured: std::collections::HashSet<String> =
|
|
accounts.iter().map(|a| a.name.clone()).collect();
|
|
for disc in discover_token_accounts() {
|
|
if !configured.contains(&disc.name) {
|
|
accounts.push(disc);
|
|
}
|
|
}
|
|
Ok(accounts)
|
|
}
|
|
|
|
/// Scan the agent state dir for extra matrix accounts provisioned via the
|
|
/// dashboard login form: each is a `matrix-token-<name>` file plus a
|
|
/// `matrix-account-<name>.json` sidecar carrying the homeserver. Returns one
|
|
/// [`AccountCfg`] per discovered account that has BOTH files — a token
|
|
/// without a sidecar is skipped, because the homeserver is then unknown and
|
|
/// defaulting to the hive homeserver would be wrong for an external account.
|
|
/// Best-effort: an unreadable dir or malformed sidecar yields fewer
|
|
/// accounts, never an error — explicit `HIVE_MATRIX_ACCOUNTS` config stays
|
|
/// authoritative.
|
|
fn discover_token_accounts() -> Vec<AccountCfg> {
|
|
let token_path = paths::token_file();
|
|
let Some(state_dir) = token_path.parent() else {
|
|
return Vec::new();
|
|
};
|
|
let Ok(rd) = std::fs::read_dir(state_dir) else {
|
|
return Vec::new();
|
|
};
|
|
let mut out = Vec::new();
|
|
for entry in rd.flatten() {
|
|
let fname = entry.file_name();
|
|
let Some(fname) = fname.to_str() else {
|
|
continue;
|
|
};
|
|
// `matrix-token` (no suffix) is the hive account, handled separately;
|
|
// only `matrix-token-<name>` files are extra accounts.
|
|
let Some(name) = fname.strip_prefix("matrix-token-") else {
|
|
continue;
|
|
};
|
|
if name.is_empty() {
|
|
continue;
|
|
}
|
|
let sidecar = state_dir.join(format!("matrix-account-{name}.json"));
|
|
let Some(homeserver) = read_account_homeserver(&sidecar) else {
|
|
tracing::warn!(
|
|
account = name,
|
|
sidecar = %sidecar.display(),
|
|
"matrix: discovered token but no homeserver sidecar; skipping account \
|
|
(re-login via the dashboard to write it)"
|
|
);
|
|
continue;
|
|
};
|
|
out.push(AccountCfg {
|
|
name: name.to_owned(),
|
|
token_file: entry.path(),
|
|
state_dir: state_dir.join(format!("matrix-sdk-state-{name}")),
|
|
homeserver: Some(homeserver),
|
|
});
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Read `{"homeserver": "<url>"}` from a sidecar file. `None` when the file
|
|
/// is missing, unreadable, not valid JSON, or the `homeserver` field is
|
|
/// absent / empty.
|
|
fn read_account_homeserver(path: &Path) -> Option<String> {
|
|
let raw = std::fs::read_to_string(path).ok()?;
|
|
let json: serde_json::Value = serde_json::from_str(&raw).ok()?;
|
|
json.get("homeserver")?
|
|
.as_str()
|
|
.filter(|s| !s.is_empty())
|
|
.map(ToOwned::to_owned)
|
|
}
|
|
|
|
/// 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>>,
|
|
}
|
|
|
|
/// Pick which account name a request resolves to, or an error. Pure (no
|
|
/// `Client`) so the ambiguity rule is unit-testable. `names` is the sorted
|
|
/// set of configured account names.
|
|
///
|
|
/// - `Some(name)` → that name (membership is checked by the caller).
|
|
/// - `None` with a single account → the primary (zero-config default).
|
|
/// - `None` with multiple accounts → an error: the agent can't tell more
|
|
/// than one account exists, so force an explicit `account` and list the
|
|
/// choices.
|
|
fn pick_name<'a>(
|
|
account: Option<&'a str>,
|
|
primary: &'a str,
|
|
names: &[&str],
|
|
) -> Result<&'a str, String> {
|
|
match account {
|
|
Some(name) => Ok(name),
|
|
None if names.len() <= 1 => Ok(primary),
|
|
None => Err(format!(
|
|
"multiple matrix accounts configured; pass `account` explicitly — available: [{}]",
|
|
names.join(", ")
|
|
)),
|
|
}
|
|
}
|
|
|
|
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` selects the
|
|
/// primary account *only when it's unambiguous* — a single configured
|
|
/// account. With multiple accounts, omitting `account` is an error
|
|
/// (see `pick_name`): the agent can't tell more than one exists, so we
|
|
/// force an explicit choice and list the names. Returns a
|
|
/// human-readable error (listing known accounts) surfaced to the agent
|
|
/// as a tool error.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Errors when `account` is omitted but multiple accounts are
|
|
/// configured, or 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 mut names: Vec<&str> = self.by_name.keys().map(String::as_str).collect();
|
|
names.sort_unstable();
|
|
let name = pick_name(account, &self.primary, &names)?;
|
|
self.by_name.get(name).ok_or_else(|| {
|
|
format!(
|
|
"unknown matrix account {name:?}; available accounts: [{}]",
|
|
names.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, pick_name, write_accounts_snapshot};
|
|
|
|
#[test]
|
|
fn pick_name_single_account_defaults_to_primary() {
|
|
assert_eq!(pick_name(None, "main", &["main"]).unwrap(), "main");
|
|
}
|
|
|
|
#[test]
|
|
fn pick_name_multi_account_omitted_errors_with_menu() {
|
|
let err = pick_name(None, "main", &["main", "public"]).unwrap_err();
|
|
assert!(err.contains("pass `account` explicitly"));
|
|
assert!(err.contains("main"));
|
|
assert!(err.contains("public"));
|
|
}
|
|
|
|
#[test]
|
|
fn pick_name_explicit_passes_through_even_with_multiple() {
|
|
assert_eq!(
|
|
pick_name(Some("public"), "main", &["main", "public"]).unwrap(),
|
|
"public"
|
|
);
|
|
}
|
|
|
|
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();
|
|
}
|
|
}
|