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::handlers;
use crate::protocol::{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).
pub async fn serve(socket_path: &Path, client: Client) -> 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");
    let client = Arc::new(client);
    loop {
        let (stream, _) = listener
            .accept()
            .await
            .context("accept connection on mcp socket")?;
        let client = client.clone();
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream, &client).await {
                tracing::warn!(error = %e, "mcp socket connection error");
            }
        });
    }
}

async fn handle_connection(stream: UnixStream, client: &Client) -> 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, client).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, client: &Client) -> DaemonResponse {
    match req {
        DaemonRequest::Ping => DaemonResponse::ok(&serde_json::json!({"ok": true})),
        DaemonRequest::SendMessage { room, body } => {
            handlers::send_message(client, &room, &body).await
        }
        DaemonRequest::SendDm { user_id, body } => handlers::send_dm(client, &user_id, &body).await,
        DaemonRequest::SendReaction {
            room,
            event_id,
            key,
        } => handlers::send_reaction(client, &room, &event_id, &key).await,
        DaemonRequest::SendReply {
            room,
            event_id,
            body,
        } => handlers::send_reply(client, &room, &event_id, &body).await,
        DaemonRequest::MarkRead { room, event_id } => {
            handlers::mark_read(client, &room, &event_id).await
        }
        DaemonRequest::ListRooms => handlers::list_rooms(client).await,
        DaemonRequest::ListInvites => handlers::list_invites(client),
        DaemonRequest::JoinRoom { room } => handlers::join_room(client, &room).await,
        DaemonRequest::InviteUser { room, user_id } => {
            handlers::invite_user(client, &room, &user_id).await
        }
        DaemonRequest::ListRoomMembers { room } => handlers::list_room_members(client, &room).await,
        DaemonRequest::ReadRoom { room, limit } => handlers::read_room(client, &room, limit).await,
        DaemonRequest::UnreadCount => handlers::unread_count(client),
        DaemonRequest::UnreadSummary => handlers::unread_summary(client).await,
    }
}