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

//! Unix socket server: the daemon listens here, the stdio MCP bridge
//! `connect()`s on every tool call. One JSON request line in, one
//! JSON response line out. Connections are short-lived (per tool call)
//! so the loop is just accept → dispatch → reply → close.

use std::path::Path;
use std::sync::Arc;

use anyhow::{Context, Result};
use matrix_sdk::Client;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};

use crate::accounts::Registry;
use crate::handlers;
use crate::protocol::{DaemonOp, DaemonRequest, DaemonResponse};

/// Start listening on `socket_path` and serve forever. Removes any
/// stale socket file first (daemon restart after a non-clean shutdown
/// would otherwise hit EADDRINUSE). `registry` resolves each request's
/// `account` to the matrix client that serves it.
pub async fn serve(socket_path: &Path, registry: Arc<Registry>) -> Result<()> {
    let _ = tokio::fs::remove_file(socket_path).await;
    if let Some(parent) = socket_path.parent() {
        tokio::fs::create_dir_all(parent)
            .await
            .with_context(|| format!("mkdir {}", parent.display()))?;
    }
    let listener = UnixListener::bind(socket_path)
        .with_context(|| format!("bind unix socket {}", socket_path.display()))?;
    tracing::info!(path = %socket_path.display(), "mcp socket listener up");
    loop {
        let (stream, _) = listener
            .accept()
            .await
            .context("accept connection on mcp socket")?;
        let registry = registry.clone();
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream, &registry).await {
                tracing::warn!(error = %e, "mcp socket connection error");
            }
        });
    }
}

async fn handle_connection(stream: UnixStream, registry: &Registry) -> Result<()> {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    while let Some(line) = lines.next_line().await? {
        let response = match serde_json::from_str::<DaemonRequest>(&line) {
            Ok(req) => dispatch(req, registry).await,
            Err(e) => DaemonResponse::error(format!("parse request: {e}")),
        };
        let mut json = serde_json::to_string(&response)?;
        json.push('\n');
        writer.write_all(json.as_bytes()).await?;
        writer.flush().await?;
    }
    Ok(())
}

async fn dispatch(req: DaemonRequest, registry: &Registry) -> DaemonResponse {
    // Ping is account-agnostic — answer without resolving a client so a
    // health probe works even before any account restores.
    if matches!(req.op, DaemonOp::Ping) {
        return DaemonResponse::ok(&serde_json::json!({"ok": true}));
    }
    // ListAccounts is registry-wide, not per-account — answer before
    // resolving a single client (the `account` field is meaningless for
    // it, and it must work even if the primary failed to restore).
    if matches!(req.op, DaemonOp::ListAccounts) {
        return DaemonResponse::ok(&registry.list());
    }
    let client = match registry.resolve(req.account.as_deref()) {
        Ok(c) => c,
        Err(msg) => return DaemonResponse::error(msg),
    };
    dispatch_op(req.op, client).await
}

async fn dispatch_op(op: DaemonOp, client: &Client) -> DaemonResponse {
    match op {
        // Unreachable in practice: `dispatch` handles Ping before resolving
        // a client (so a health probe works before any account restores).
        // Kept for an exhaustive match.
        DaemonOp::Ping => DaemonResponse::ok(&serde_json::json!({"ok": true})),
        // Unreachable in practice: `dispatch` handles ListAccounts before
        // resolving a client (it is registry-wide, not per-account). Kept
        // for an exhaustive match — there is no client-scoped meaning.
        DaemonOp::ListAccounts => DaemonResponse::error(
            "list_accounts is registry-wide; handled before client resolution",
        ),
        DaemonOp::SendMessage { room, body } => handlers::send_message(client, &room, &body).await,
        DaemonOp::SendDm { user_id, body } => handlers::send_dm(client, &user_id, &body).await,
        DaemonOp::SendFile {
            room,
            path,
            caption,
        } => handlers::send_file(client, &room, &path, caption.as_deref()).await,
        DaemonOp::OpenDm { user_id } => handlers::open_dm(client, &user_id).await,
        DaemonOp::SendReaction {
            room,
            event_id,
            key,
        } => handlers::send_reaction(client, &room, &event_id, &key).await,
        DaemonOp::SendReply {
            room,
            event_id,
            body,
        } => handlers::send_reply(client, &room, &event_id, &body).await,
        DaemonOp::MarkRead { room, event_id } => {
            handlers::mark_read(client, &room, &event_id).await
        }
        DaemonOp::SendRedact {
            room,
            event_id,
            reason,
        } => handlers::send_redact(client, &room, &event_id, reason.as_deref()).await,
        DaemonOp::ListRooms => handlers::list_rooms(client).await,
        DaemonOp::ListInvites => handlers::list_invites(client),
        DaemonOp::JoinRoom { room } => handlers::join_room(client, &room).await,
        DaemonOp::ResolveInvite { room, action } => {
            handlers::resolve_invite(client, &room, action).await
        }
        DaemonOp::InviteUser { room, user_id } => {
            handlers::invite_user(client, &room, &user_id).await
        }
        DaemonOp::ListRoomMembers { room } => handlers::list_room_members(client, &room).await,
        DaemonOp::ReadRoom {
            room,
            limit,
            from,
            until,
        } => handlers::read_room(client, &room, limit, from, until).await,
        DaemonOp::DownloadFile {
            room,
            event_id,
            dest_path,
        } => handlers::download_file(client, &room, &event_id, dest_path.as_deref()).await,
        DaemonOp::UnreadCount => handlers::unread_count(client),
        DaemonOp::UnreadSummary => handlers::unread_summary(client).await,
    }
}