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 the operator's call:
//! `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, OwnedServerName, OwnedUserId, RoomOrAliasId,
        api::client::receipt::create_receipt::v3::ReceiptType,
        events::{
            reaction::ReactionEventContent,
            receipt::ReceiptThread,
            relation::Annotation,
            room::message::{MessageType, RoomMessageEventContent},
        },
    },
};
use serde::Serialize;

use crate::protocol::{DaemonResponse, InviteAction};

/// JSON-shape for `list_invites`: one row per pending room invite.
#[derive(Debug, Serialize)]
pub struct InviteInfo {
    pub room_id: String,
    pub canonical_alias: Option<String>,
    pub name: Option<String>,
}

/// 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(),
    }
}

/// Same as [`extract_body`] but for the `AnySyncTimelineEvent` variant
/// returned by `TimelineEvent::raw()` (the sdk-level decrypted form from
/// `room.messages()`).
fn extract_body_sync(event: &matrix_sdk::ruma::events::AnySyncTimelineEvent) -> String {
    use matrix_sdk::ruma::events::{AnySyncMessageLikeEvent, AnySyncTimelineEvent};
    match event {
        AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::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(),
    }
}

/// Human-facing label for a room: its canonical alias when set,
/// otherwise the raw room id. Used wherever a room is named in
/// agent-facing text (wake bodies, loose-ends, the unread guard).
#[must_use]
pub fn room_label(room: &matrix_sdk::Room) -> String {
    room.canonical_alias()
        .map_or_else(|| room.room_id().to_string(), |a| a.to_string())
}

/// Refuse to post into a room the agent hasn't caught up on. Returns
/// `Some(error)` with a helpful hint when the room still has unread
/// notifications (the agent must `read_room` then `mark_read` the
/// latest event first), or `None` when the send may proceed.
///
/// Read-state is the matrix unread-notification count, the same signal
/// the wake path and `get_loose_ends` use, so "caught up" here means
/// exactly what those surfaces mean. Reactions and `mark_read` are not
/// gated — only message-posting tools (`send_message`, `send_reply`,
/// `send_dm`) so an agent can't talk over messages it hasn't seen.
fn unread_guard(room: &matrix_sdk::Room) -> Option<DaemonResponse> {
    let count = room.unread_notification_counts().notification_count;
    if count == 0 {
        return None;
    }
    let label = room_label(room);
    Some(DaemonResponse::error(format!(
        "refusing to send: {count} unread message(s) in {label}. \
         use read_room to view them, then mark_read the latest event before \
         sending so you don't talk over messages you haven't seen."
    )))
}

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,
    };
    if let Some(reject) = unread_guard(&room) {
        return reject;
    }
    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}")),
        },
    };
    if let Some(reject) = unread_guard(&room) {
        return reject;
    }
    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,
    };
    if let Some(reject) = unread_guard(&room) {
        return reject;
    }
    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 fn list_invites(client: &Client) -> DaemonResponse {
    let invites: Vec<InviteInfo> = client
        .invited_rooms()
        .into_iter()
        .map(|room| InviteInfo {
            room_id: room.room_id().to_string(),
            canonical_alias: room.canonical_alias().map(|a| a.to_string()),
            name: room.name(),
        })
        .collect();
    DaemonResponse::ok(&invites)
}

/// Rewrite `mcp-loose-ends/matrix.json` with a summary of all pending
/// room invites. The harness scans this directory generically in
/// `get_loose_ends` — no matrix-specific code needed there.
///
/// Called after an invite arrives (from the sync handler) and after a
/// room is joined (to remove the accepted invite from loose-ends).
/// Atomic write (tmp + rename) so the harness never reads a partial file.
pub async fn refresh_invite_loose_ends(client: &Client) {
    let invites = client.invited_rooms();
    let dir = crate::paths::mcp_loose_ends_dir();
    if let Err(e) = tokio::fs::create_dir_all(&dir).await {
        tracing::warn!(error = ?e, "matrix: create mcp-loose-ends dir failed");
        return;
    }
    let items: Vec<String> = invites
        .iter()
        .map(|room| {
            let label = room_label(room);
            format!(
                "[matrix] pending invite: {label} — use list_invites to see, resolve_invite to accept or reject"
            )
        })
        .collect();
    let dest = dir.join("matrix.json");
    let tmp = dest.with_extension("json.tmp");
    let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned());
    match tokio::fs::write(&tmp, &json).await {
        Ok(()) => {
            if let Err(e) = tokio::fs::rename(&tmp, &dest).await {
                tracing::warn!(error = ?e, "matrix: rename mcp-loose-ends/matrix.json failed");
            }
        }
        Err(e) => {
            tracing::warn!(error = ?e, "matrix: write mcp-loose-ends/matrix.json.tmp failed");
        }
    }
}

pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse {
    let parsed: &RoomOrAliasId = match room_ref.try_into() {
        Ok(p) => p,
        Err(e) => {
            return DaemonResponse::error(format!("invalid room reference {room_ref}: {e}"));
        }
    };
    let server_names: Vec<OwnedServerName> = vec![];
    match client.join_room_by_id_or_alias(parsed, &server_names).await {
        Ok(room) => {
            // Refresh loose-ends so the accepted invite is removed from
            // `get_loose_ends` output immediately after the agent joins.
            refresh_invite_loose_ends(client).await;
            DaemonResponse::ok(&serde_json::json!({
                "joined": true,
                "room_id": room.room_id().to_string(),
            }))
        }
        Err(e) => DaemonResponse::error(format!("join room {room_ref}: {e}")),
    }
}

/// Accept or reject a pending invite to `room_ref` (id or alias).
///
/// `Accept` is exactly `join_room` (joining a room you were invited to
/// accepts the invite). `Reject` resolves the invited room and leaves
/// it, declining the invite. Either way the resolved invite drops out
/// of `get_loose_ends` immediately.
pub async fn resolve_invite(
    client: &Client,
    room_ref: &str,
    action: InviteAction,
) -> DaemonResponse {
    match action {
        InviteAction::Accept => join_room(client, room_ref).await,
        InviteAction::Reject => {
            // `resolve_room` -> `get_room` returns the room in any
            // membership state the store knows, including `Invited`, so
            // this works for a pending invite that was never joined.
            let room = match resolve_room(client, room_ref).await {
                Ok(r) => r,
                Err(e) => return e,
            };
            match room.leave().await {
                Ok(()) => {
                    refresh_invite_loose_ends(client).await;
                    DaemonResponse::ok(&serde_json::json!({
                        "rejected": true,
                        "room_id": room.room_id().to_string(),
                    }))
                }
                Err(e) => DaemonResponse::error(format!("reject invite {room_ref}: {e}")),
            }
        }
    }
}

/// Invite `user_id` to `room_ref`. The calling agent must be a member of
/// the room with a power level high enough to invite; otherwise the
/// homeserver rejects it and the error is surfaced verbatim.
pub async fn invite_user(client: &Client, room_ref: &str, user_id: &str) -> DaemonResponse {
    let uid: OwnedUserId = match user_id.parse() {
        Ok(u) => u,
        Err(e) => return DaemonResponse::error(format!("invalid user_id {user_id}: {e}")),
    };
    let room = match resolve_room(client, room_ref).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    match room.invite_user_by_id(&uid).await {
        Ok(()) => DaemonResponse::ok(&serde_json::json!({
            "invited": true,
            "room_id": room.room_id().to_string(),
            "user_id": uid.to_string(),
        })),
        Err(e) => DaemonResponse::error(format!("invite {uid} to {room_ref}: {e}")),
    }
}

pub async fn read_room(client: &Client, room_ref: &str, limit: Option<usize>) -> DaemonResponse {
    use matrix_sdk::room::MessagesOptions;

    let room = match resolve_room(client, room_ref).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    let limit_val = u32::try_from(limit.unwrap_or(50).min(200)).unwrap_or(50);
    let mut opts = MessagesOptions::backward();
    opts.limit = matrix_sdk::ruma::UInt::from(limit_val);
    // room.messages() transparently decrypts events in encrypted rooms.
    // UTD (unable to decrypt) events surface via `ev.kind.is_utd()` and
    // get a sentinel body so claude knows decryption failed rather than
    // seeing the raw encrypted blob.
    let msgs = match room.messages(opts).await {
        Ok(m) => m,
        Err(e) => return DaemonResponse::error(format!("messages: {e}")),
    };
    let events: Vec<TimelineEvent> = msgs
        .chunk
        .iter()
        .filter_map(|ev| {
            let parsed = ev.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 = if ev.kind.is_utd() {
                "[unable to decrypt]".to_owned()
            } else {
                extract_body_sync(&parsed)
            };
            Some(TimelineEvent {
                event_id,
                sender,
                origin_server_ts,
                event_type,
                body,
            })
        })
        .collect();
    DaemonResponse::ok(&events)
}

/// Return the number of joined rooms with at least one unread
/// notification according to the server-side push notification counts
/// cached by the matrix-sdk client.
///
/// The harness uses [`unread_summary`] (richer, with per-room body
/// snippets) rather than this endpoint. `unread_count` is kept as a
/// lightweight public endpoint for callers that only need the count
/// and want to avoid the `/messages` network round-trips that
/// [`collect_unread`] issues for count==1 rooms.
#[must_use]
pub fn unread_count(client: &Client) -> DaemonResponse {
    let rooms = u32::try_from(
        client
            .joined_rooms()
            .into_iter()
            .filter(|r| r.unread_notification_counts().notification_count > 0)
            .count(),
    )
    .unwrap_or(u32::MAX);
    DaemonResponse::ok(&serde_json::json!({ "rooms": rooms }))
}

/// Collect all rooms with unread notifications and build per-room
/// [`crate::protocol::RoomUnread`] entries. For rooms with exactly
/// one unread notification the last message body is fetched via the
/// `/messages` endpoint (best-effort; failures leave `last_body` as
/// `None`). Rooms with zero unreads are omitted.
///
/// **Latency note**: each count==1 room triggers a live `/messages`
/// network request to the matrix homeserver to retrieve the message
/// body. This adds per-room round-trip latency to `get_loose_ends`
/// and the wake-signal path. Acceptable in practice (rooms with
/// unread are few; request is best-effort), but worth bearing in
/// mind if latency becomes a concern.
#[must_use]
pub async fn collect_unread(client: &Client) -> Vec<crate::protocol::RoomUnread> {
    use crate::protocol::RoomUnread;
    let mut result = Vec::new();
    for room in client.joined_rooms() {
        let count =
            u32::try_from(room.unread_notification_counts().notification_count).unwrap_or(u32::MAX);
        if count == 0 {
            continue;
        }
        let label = room_label(&room);
        let (last_body, last_sender) = if count == 1 {
            fetch_last_message(client, &room).await
        } else {
            (None, None)
        };
        result.push(RoomUnread {
            label,
            count,
            last_body,
            last_sender,
        });
    }
    result
}

/// Fetch the body + sender of the most recent room message. Returns
/// `(None, None)` on any error or when the timeline contains no text
/// events.
async fn fetch_last_message(
    client: &Client,
    room: &matrix_sdk::Room,
) -> (Option<String>, Option<String>) {
    use matrix_sdk::ruma::api::Direction;
    use matrix_sdk::ruma::api::client::message::get_message_events;
    let mut req =
        get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward);
    req.limit = matrix_sdk::ruma::UInt::from(1u32);
    let Ok(resp) = client.send(req).await else {
        return (None, None);
    };
    for raw in &resp.chunk {
        if let Ok(ev) = raw.deserialize() {
            let body = extract_body(&ev);
            if !body.is_empty() {
                return (
                    Some(crate::wake::truncate_chars(&body, 100)),
                    Some(ev.sender().to_string()),
                );
            }
        }
    }
    (None, None)
}

/// Return per-room unread summaries. For rooms with exactly one
/// unread notification, attempts to include the sender + truncated
/// body; rooms with multiple unreads carry only the count.
#[must_use]
pub async fn unread_summary(client: &Client) -> DaemonResponse {
    let rooms = collect_unread(client).await;
    DaemonResponse::ok(&rooms)
}