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

//! Per-tool dispatch — each daemon request from the MCP bridge resolves
//! to one of these handlers. Returns a `DaemonResponse` shaped for the
//! wire protocol (the MCP bridge unwraps `Ok { payload }` and returns
//! the payload to claude; `Error { message }` becomes the tool-call
//! error message claude sees).
//!
//! Tool surface mirrors damocles-daemon's v0 set per mara on #548:
//! `send_message`, `send_dm`, `send_reaction`, `send_reply`, `mark_read`,
//! `list_rooms`, `list_room_members`, `read_room`. Plus a `ping` for the
//! MCP bridge's liveness probe.

use matrix_sdk::{
    Client,
    room::reply::{EnforceThread, Reply},
    ruma::{
        OwnedEventId, OwnedRoomId, OwnedUserId, RoomOrAliasId,
        api::client::receipt::create_receipt::v3::ReceiptType,
        events::{
            receipt::ReceiptThread,
            reaction::ReactionEventContent,
            relation::Annotation,
            room::message::{MessageType, RoomMessageEventContent},
        },
    },
};
use serde::Serialize;

use crate::protocol::DaemonResponse;

/// JSON-shape for `list_rooms`: one row per joined room.
#[derive(Debug, Serialize)]
pub struct RoomInfo {
    pub room_id: String,
    pub canonical_alias: Option<String>,
    pub name: String,
    pub member_count: u64,
}

/// JSON-shape for `list_room_members`: one row per joined member.
#[derive(Debug, Serialize)]
pub struct MemberInfo {
    pub user_id: String,
    pub display_name: Option<String>,
}

/// JSON-shape for `read_room`: one timeline event, flattened for
/// claude consumption.
#[derive(Debug, Serialize)]
pub struct TimelineEvent {
    pub event_id: String,
    pub sender: String,
    pub origin_server_ts: i64,
    pub event_type: String,
    /// Best-effort plain-text body for `m.room.message` events
    /// (text / notice / emote); empty for events without a textual
    /// body (state changes, reactions, etc.).
    pub body: String,
}

/// Resolve a room reference (id `!abc:server` OR alias `#name:server`)
/// to a joined `Room`. Returns an `Error` response if the room isn't
/// joined / the reference is malformed.
async fn resolve_room(
    client: &Client,
    reference: &str,
) -> Result<matrix_sdk::Room, DaemonResponse> {
    let parsed: &RoomOrAliasId = reference.try_into().map_err(|e| {
        DaemonResponse::error(format!("invalid room reference {reference}: {e}"))
    })?;
    let room_id: OwnedRoomId = if parsed.is_room_id() {
        OwnedRoomId::try_from(reference)
            .map_err(|e| DaemonResponse::error(format!("invalid room_id: {e}")))?
    } else {
        client
            .resolve_room_alias(parsed.as_str().try_into().map_err(|e| {
                DaemonResponse::error(format!("invalid alias: {e}"))
            })?)
            .await
            .map(|r| r.room_id)
            .map_err(|e| DaemonResponse::error(format!("resolve_room_alias {reference}: {e}")))?
    };
    client
        .get_room(&room_id)
        .ok_or_else(|| DaemonResponse::error(format!("room {room_id} not joined")))
}

/// Best-effort plain-text body for a (non-sync) timeline event.
/// Returns "" for non-text events (state changes, reactions,
/// redactions) — claude can still see the `event_type` field to
/// disambiguate. `AnyTimelineEvent` (not `AnySync...`) because
/// `read_room` pulls events via the `/messages` endpoint which
/// returns the full-form variant.
fn extract_body(event: &matrix_sdk::ruma::events::AnyTimelineEvent) -> String {
    use matrix_sdk::ruma::events::{AnyMessageLikeEvent, AnyTimelineEvent};
    match event {
        AnyTimelineEvent::MessageLike(AnyMessageLikeEvent::RoomMessage(ev)) => {
            ev.as_original().map_or_else(String::new, |orig| {
                match &orig.content.msgtype {
                    MessageType::Text(t) => t.body.clone(),
                    MessageType::Notice(n) => n.body.clone(),
                    MessageType::Emote(e) => format!("* {}", e.body),
                    _ => String::new(),
                }
            })
        }
        _ => String::new(),
    }
}

pub async fn send_message(client: &Client, room_ref: &str, body: &str) -> DaemonResponse {
    let room = match resolve_room(client, room_ref).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    let content = RoomMessageEventContent::text_markdown(body);
    match room.send(content).await {
        Ok(resp) => DaemonResponse::ok(&serde_json::json!({
            "event_id": resp.event_id.to_string(),
            "room_id": room.room_id().to_string(),
        })),
        Err(e) => DaemonResponse::error(format!("send to {}: {e}", room.room_id())),
    }
}

pub async fn send_dm(client: &Client, user_id: &str, body: &str) -> DaemonResponse {
    let uid: OwnedUserId = match user_id.parse() {
        Ok(u) => u,
        Err(e) => return DaemonResponse::error(format!("invalid user_id {user_id}: {e}")),
    };
    // Find existing DM or create one.
    let room = client
        .joined_rooms()
        .into_iter()
        .find(|r| {
            // is_direct() is async; check direct_targets() instead which
            // reads from cached state.
            r.direct_targets().iter().any(|t| t.as_str() == uid.as_str())
        });
    let room = match room {
        Some(r) => r,
        None => match client.create_dm(&uid).await {
            Ok(r) => r,
            Err(e) => return DaemonResponse::error(format!("create_dm {uid}: {e}")),
        },
    };
    let content = RoomMessageEventContent::text_markdown(body);
    match room.send(content).await {
        Ok(resp) => DaemonResponse::ok(&serde_json::json!({
            "event_id": resp.event_id.to_string(),
            "room_id": room.room_id().to_string(),
            "user_id": uid.to_string(),
        })),
        Err(e) => DaemonResponse::error(format!("send DM to {uid}: {e}")),
    }
}

pub async fn send_reaction(
    client: &Client,
    room_ref: &str,
    event_id: &str,
    key: &str,
) -> DaemonResponse {
    let room = match resolve_room(client, room_ref).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    let eid: OwnedEventId = match event_id.parse() {
        Ok(e) => e,
        Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
    };
    let content = ReactionEventContent::new(Annotation::new(eid.clone(), key.to_owned()));
    match room.send(content).await {
        Ok(resp) => DaemonResponse::ok(&serde_json::json!({
            "event_id": resp.event_id.to_string(),
            "target": eid.to_string(),
        })),
        Err(e) => DaemonResponse::error(format!("send reaction to {eid}: {e}")),
    }
}

pub async fn send_reply(
    client: &Client,
    room_ref: &str,
    event_id: &str,
    body: &str,
) -> DaemonResponse {
    let room = match resolve_room(client, room_ref).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    let eid: OwnedEventId = match event_id.parse() {
        Ok(e) => e,
        Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
    };
    let content = RoomMessageEventContent::text_markdown(body).into();
    let reply = Reply {
        event_id: eid.clone(),
        enforce_thread: EnforceThread::MaybeThreaded,
    };
    let reply_content = match room.make_reply_event(content, reply).await {
        Ok(c) => c,
        Err(e) => return DaemonResponse::error(format!("make_reply_event: {e}")),
    };
    match room.send(reply_content).await {
        Ok(resp) => DaemonResponse::ok(&serde_json::json!({
            "event_id": resp.event_id.to_string(),
            "target": eid.to_string(),
        })),
        Err(e) => DaemonResponse::error(format!("send reply: {e}")),
    }
}

pub async fn mark_read(client: &Client, room_ref: &str, event_id: &str) -> DaemonResponse {
    let room = match resolve_room(client, room_ref).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    let eid: OwnedEventId = match event_id.parse() {
        Ok(e) => e,
        Err(e) => return DaemonResponse::error(format!("invalid event_id {event_id}: {e}")),
    };
    match room
        .send_single_receipt(ReceiptType::Read, ReceiptThread::Unthreaded, eid.clone())
        .await
    {
        Ok(()) => DaemonResponse::ok(&serde_json::json!({
            "marked_read": eid.to_string(),
        })),
        Err(e) => DaemonResponse::error(format!("send_single_receipt: {e}")),
    }
}

pub async fn list_rooms(client: &Client) -> DaemonResponse {
    let mut rooms = Vec::new();
    for room in client.joined_rooms() {
        let name = match room.display_name().await {
            Ok(n) => n.to_string(),
            Err(_) => room.room_id().to_string(),
        };
        let canonical_alias = room.canonical_alias().map(|a| a.to_string());
        rooms.push(RoomInfo {
            room_id: room.room_id().to_string(),
            canonical_alias,
            name,
            member_count: room.joined_members_count(),
        });
    }
    DaemonResponse::ok(&rooms)
}

pub async fn list_room_members(client: &Client, room_ref: &str) -> DaemonResponse {
    let room = match resolve_room(client, room_ref).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    let members = match room.members(matrix_sdk::RoomMemberships::JOIN).await {
        Ok(m) => m,
        Err(e) => return DaemonResponse::error(format!("members: {e}")),
    };
    let list: Vec<MemberInfo> = members
        .iter()
        .map(|m| MemberInfo {
            user_id: m.user_id().to_string(),
            display_name: m.display_name().map(ToOwned::to_owned),
        })
        .collect();
    DaemonResponse::ok(&list)
}

pub async fn read_room(client: &Client, room_ref: &str, limit: Option<usize>) -> DaemonResponse {
    use matrix_sdk::ruma::api::client::message::get_message_events;
    use matrix_sdk::ruma::api::Direction;

    let room = match resolve_room(client, room_ref).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    let limit = limit.unwrap_or(50).min(200);
    let mut req = get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward);
    req.limit = matrix_sdk::ruma::UInt::try_from(limit as u64).unwrap_or(matrix_sdk::ruma::UInt::from(50u32));
    let resp = match client.send(req).await {
        Ok(r) => r,
        Err(e) => return DaemonResponse::error(format!("get_message_events: {e}")),
    };
    let events: Vec<TimelineEvent> = resp
        .chunk
        .iter()
        .filter_map(|raw| {
            let parsed = raw.deserialize().ok()?;
            let event_id = parsed.event_id().to_string();
            let sender = parsed.sender().to_string();
            let origin_server_ts: i64 = parsed.origin_server_ts().0.into();
            let event_type = parsed.event_type().to_string();
            let body = extract_body(&parsed);
            Some(TimelineEvent {
                event_id,
                sender,
                origin_server_ts,
                event_type,
                body,
            })
        })
        .collect();
    DaemonResponse::ok(&events)
}