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`,
//! `send_redact`, `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| attachment_marker(&orig.content.msgtype)),
        _ => String::new(),
    }
}

/// Best-effort plain-text body / attachment marker for a message
/// `MessageType`. Text-like messages return their body; media messages
/// (`m.file` / `m.image` / `m.audio` / `m.video`) return a
/// `[file: name]`-style marker so an agent reading the room sees that an
/// attachment is present (and can fetch it with `download_file`); all
/// other types return "".
fn attachment_marker(msgtype: &MessageType) -> String {
    match msgtype {
        MessageType::Text(t) => t.body.clone(),
        MessageType::Notice(n) => n.body.clone(),
        MessageType::Emote(e) => format!("* {}", e.body),
        MessageType::File(f) => format!("[file: {}]", f.body),
        MessageType::Image(i) => format!("[image: {}]", i.body),
        MessageType::Audio(a) => format!("[audio: {}]", a.body),
        MessageType::Video(v) => format!("[video: {}]", v.body),
        _ => 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| attachment_marker(&orig.content.msgtype)),
        _ => 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())),
    }
}

/// Find the existing DM room with `user_id` or create one. Shared by
/// `send_dm` and `open_dm`. `direct_targets()` (cached state) is
/// used rather than the async `is_direct()`.
async fn resolve_or_create_dm(
    client: &Client,
    user_id: &str,
) -> Result<matrix_sdk::Room, DaemonResponse> {
    let uid: OwnedUserId = user_id
        .parse()
        .map_err(|e| DaemonResponse::error(format!("invalid user_id {user_id}: {e}")))?;
    if let Some(room) = client.joined_rooms().into_iter().find(|r| {
        r.direct_targets()
            .iter()
            .any(|t| t.as_str() == uid.as_str())
    }) {
        return Ok(room);
    }
    client
        .create_dm(&uid)
        .await
        .map_err(|e| DaemonResponse::error(format!("create_dm {uid}: {e}")))
}

pub async fn send_dm(client: &Client, user_id: &str, body: &str) -> DaemonResponse {
    let room = match resolve_or_create_dm(client, user_id).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(),
            "user_id": user_id,
        })),
        Err(e) => DaemonResponse::error(format!("send DM to {user_id}: {e}")),
    }
}

/// Maximum file size accepted by `send_file`.
const MAX_UPLOAD_BYTES: u64 = 50 * 1024 * 1024;

/// Read a local file and post it as a matrix attachment to `room`,
/// inferring the MIME type from the file extension. An optional
/// `caption` is sent as a best-effort follow-up text message (the
/// attachment has already landed, so a caption failure is non-fatal).
/// Used by `send_file`.
async fn upload_attachment(
    room: &matrix_sdk::Room,
    path: &str,
    caption: Option<&str>,
) -> DaemonResponse {
    let p = std::path::Path::new(path);
    let meta = match tokio::fs::metadata(p).await {
        Ok(m) => m,
        Err(e) => return DaemonResponse::error(format!("stat {path}: {e}")),
    };
    if !meta.is_file() {
        return DaemonResponse::error(format!("{path} is not a regular file"));
    }
    if meta.len() > MAX_UPLOAD_BYTES {
        return DaemonResponse::error(format!(
            "{path} is {} bytes, over the {MAX_UPLOAD_BYTES}-byte upload cap",
            meta.len()
        ));
    }
    let data = match tokio::fs::read(p).await {
        Ok(d) => d,
        Err(e) => return DaemonResponse::error(format!("read {path}: {e}")),
    };
    let filename = p
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("file")
        .to_owned();
    let content_type: mime::Mime = mime_guess::from_path(p).first_or_octet_stream();
    let resp = match room
        .send_attachment(
            filename.clone(),
            &content_type,
            data,
            matrix_sdk::attachment::AttachmentConfig::new(),
        )
        .await
    {
        Ok(r) => r,
        Err(e) => {
            return DaemonResponse::error(format!("send attachment to {}: {e}", room.room_id()));
        }
    };
    if let Some(c) = caption.filter(|c| !c.is_empty()) {
        let _ = room.send(RoomMessageEventContent::text_markdown(c)).await;
    }
    DaemonResponse::ok(&serde_json::json!({
        "event_id": resp.event_id.to_string(),
        "room_id": room.room_id().to_string(),
        "filename": filename,
        "mime": content_type.essence_str(),
    }))
}

pub async fn send_file(
    client: &Client,
    room_ref: &str,
    path: &str,
    caption: Option<&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;
    }
    upload_attachment(&room, path, caption).await
}

/// Resolve (find-or-create) the DM room with `user_id` and return its
/// room id without sending anything. The caller then uses the room-based
/// tools (`send_file`, `send_message`, …) against that id — so there is
/// no per-tool `_dm` variant.
pub async fn open_dm(client: &Client, user_id: &str) -> DaemonResponse {
    let room = match resolve_or_create_dm(client, user_id).await {
        Ok(r) => r,
        Err(e) => return e,
    };
    DaemonResponse::ok(&serde_json::json!({
        "room_id": room.room_id().to_string(),
        "user_id": user_id,
    }))
}

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 send_redact(
    client: &Client,
    room_ref: &str,
    event_id: &str,
    reason: Option<&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}")),
    };
    // Redaction is irreversible: the homeserver drops the event content and
    // keeps only the shell + reason. A fresh txn id (None) is fine — this is
    // a one-shot operator/agent action, not a retried send. Power level is
    // enforced server-side (own events always; others' need a moderator PL),
    // so a redact that races concurrent delivery is resolved by the server.
    // No client-side pre-check — we let the server be the authority on both
    // permission and ordering.
    match room.redact(&eid, reason, None).await {
        Ok(resp) => DaemonResponse::ok(&serde_json::json!({
            "event_id": resp.event_id.to_string(),
            "target": eid.to_string(),
        })),
        Err(e) => DaemonResponse::error(format!("redact {eid}: {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)
}

/// Download the media attachment carried by `event_id` in `room` and
/// write it to a local file, returning the path. Resolves the event,
/// extracts the `MediaSource` from its `m.file` / `m.image` / `m.audio`
/// / `m.video` content, fetches the bytes via the media API, and writes
/// them to `dest_path` (or a temp file named after the attachment when
/// omitted). The agent then reads the file from the returned path.
pub async fn download_file(
    client: &Client,
    room_ref: &str,
    event_id: &str,
    dest_path: Option<&str>,
) -> DaemonResponse {
    use matrix_sdk::media::{MediaFormat, MediaRequestParameters};
    use matrix_sdk::ruma::events::{AnySyncMessageLikeEvent, AnySyncTimelineEvent};

    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 ev = match room.event(&eid, None).await {
        Ok(e) => e,
        Err(e) => return DaemonResponse::error(format!("fetch event {eid}: {e}")),
    };
    let parsed = match ev.raw().deserialize() {
        Ok(p) => p,
        Err(e) => return DaemonResponse::error(format!("deserialize event {eid}: {e}")),
    };
    // Pull the media source + filename from the message content.
    let AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(msg)) = parsed
    else {
        return DaemonResponse::error(format!("event {eid} is not a room message"));
    };
    let Some(orig) = msg.as_original() else {
        return DaemonResponse::error(format!("event {eid} is redacted"));
    };
    let (source, filename) = match &orig.content.msgtype {
        MessageType::File(f) => (f.source.clone(), f.body.clone()),
        MessageType::Image(i) => (i.source.clone(), i.body.clone()),
        MessageType::Audio(a) => (a.source.clone(), a.body.clone()),
        MessageType::Video(v) => (v.source.clone(), v.body.clone()),
        _ => {
            return DaemonResponse::error(format!(
                "event {eid} carries no file/image/audio/video attachment"
            ));
        }
    };
    let req = MediaRequestParameters {
        source,
        format: MediaFormat::File,
    };
    let bytes = match client.media().get_media_content(&req, true).await {
        Ok(b) => b,
        Err(e) => return DaemonResponse::error(format!("download media for {eid}: {e}")),
    };
    let dest = if let Some(p) = dest_path {
        std::path::PathBuf::from(p)
    } else {
        let name = std::path::Path::new(&filename)
            .file_name()
            .and_then(|n| n.to_str())
            .filter(|n| !n.is_empty())
            .unwrap_or("matrix-attachment");
        std::env::temp_dir().join(name)
    };
    if let Err(e) = tokio::fs::write(&dest, &bytes).await {
        return DaemonResponse::error(format!("write {}: {e}", dest.display()));
    }
    DaemonResponse::ok(&serde_json::json!({
        "path": dest.display().to_string(),
        "filename": filename,
        "bytes": bytes.len(),
        "room_id": room.room_id().to_string(),
    }))
}

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