mail4agent_messenger_shell/
cmd.rs

1//! M5: rich agent commands over the client's local socket (mail/Slack-like).
2//! One JSON line `{"cmd":"rooms.list","as":"<nick>", ...}` in, one JSON line
3//! out. No crypto in the request or the answer: the shell encrypts, decrypts
4//! and tracks devices itself. A line without `cmd` is the legacy send request.
5
6use serde::{Deserialize, Serialize};
7use serde_json::{json, Value};
8
9use crate::{
10    mxid_localpart, CreateRoomKind, ItemContent, MessageKind, MessengerCommand, OpenedStore,
11    OutgoingMessage, RoomId, UserId,
12};
13
14/// One command request.
15#[derive(Debug, Clone, Default, Serialize, Deserialize)]
16pub struct CmdRequest {
17    /// Command name, e.g. `rooms.list`.
18    pub cmd: String,
19    /// Session nick that runs the command.
20    #[serde(rename = "as")]
21    pub as_nick: String,
22    /// Command arguments (command specific).
23    #[serde(default)]
24    pub args: Value,
25}
26
27/// The answer to one [`CmdRequest`].
28#[derive(Debug, Clone, Default, Serialize, Deserialize)]
29pub struct CmdReply {
30    /// Success flag.
31    pub ok: bool,
32    /// Why it failed. No secrets.
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    pub error: Option<String>,
35    /// Command result.
36    #[serde(default)]
37    pub data: Value,
38}
39
40impl CmdReply {
41    /// A failed reply.
42    pub fn failed(error: impl Into<String>) -> Self {
43        Self { ok: false, error: Some(error.into()), data: Value::Null }
44    }
45    fn ok(data: Value) -> Self {
46        Self { ok: true, error: None, data }
47    }
48}
49
50fn arg_str<'a>(args: &'a Value, key: &str) -> Option<&'a str> {
51    args.get(key).and_then(Value::as_str).map(str::trim).filter(|v| !v.is_empty())
52}
53
54impl OpenedStore {
55    /// Display title of a room: its name, or the other members for a DM/group.
56    fn room_title(&self, room_id: &RoomId) -> String {
57        let me = self.driver.core.user_id().clone();
58        let Some(state) = self.driver.core.room_state(room_id) else { return String::new() };
59        if let Some(name) = state.name.as_deref().filter(|n| !n.is_empty()) {
60            return name.to_string();
61        }
62        let others: Vec<String> = state
63            .members
64            .iter()
65            .filter(|(id, _)| **id != me)
66            .map(|(id, m)| m.displayname.clone().unwrap_or_else(|| mxid_localpart(id.as_str()).to_string()))
67            .collect();
68        others.join(", ")
69    }
70
71    /// Resolves `room` (id, `#name`, name, or a nick for a DM) to a joined room id.
72    fn resolve_room(&mut self, spec: &str, now_ms: i64) -> Result<RoomId, String> {
73        if spec.starts_with('!') {
74            return RoomId::parse(spec).map_err(|e| e.to_string());
75        }
76        let wanted = spec.trim_start_matches(['#', '@']).to_ascii_lowercase();
77        let ids: Vec<RoomId> = self.driver.core.room_ids().cloned().collect();
78        let by_name: Vec<RoomId> = ids
79            .iter()
80            .filter(|id| self.room_title(id).to_ascii_lowercase() == wanted)
81            .cloned()
82            .collect();
83        if by_name.len() == 1 {
84            return Ok(by_name[0].clone());
85        }
86        if by_name.len() > 1 {
87            return Err(format!("{spec} matches several rooms; use the room id"));
88        }
89        if spec.starts_with('#') {
90            return Err(format!("no room named {spec}"));
91        }
92        let id = self.ensure_dm(spec, now_ms).map_err(|e| format!("{spec}: {e}"))?;
93        RoomId::parse(&id).map_err(|e| e.to_string())
94    }
95
96    fn settle_cmd(&mut self, now_ms: i64) {
97        for i in 1..=6 {
98            let _ = self.drive(now_ms + i * 1_000, false);
99            std::thread::sleep(std::time::Duration::from_millis(200));
100        }
101    }
102
103    fn rows(&self, room_id: &RoomId, limit: usize) -> Vec<Value> {
104        let Some(timeline) = self.driver.core.timeline(room_id) else { return Vec::new() };
105        let mut out: Vec<Value> = timeline
106            .items()
107            .iter()
108            .filter_map(|item| {
109                let (kind, body) = match &item.content {
110                    ItemContent::Text(t) => ("text", t.body.clone()),
111                    ItemContent::Notice(t) => ("notice", t.body.clone()),
112                    ItemContent::Emote(t) => ("emote", t.body.clone()),
113                    ItemContent::Undecryptable { .. } => ("undecryptable", String::new()),
114                    _ => return None,
115                };
116                Some(json!({
117                    "event_id": item.event_id.as_ref().map(|e| e.as_str()),
118                    "from": mxid_localpart(item.sender.as_str()),
119                    "from_id": item.sender.as_str(),
120                    "ts": item.origin_server_ts,
121                    "kind": kind,
122                    "body": body,
123                    "reply_to": item.relations.reply_to.as_ref().map(|e| e.as_str()),
124                    "thread": item.relations.thread_root.as_ref().map(|e| e.as_str()),
125                    "state": crate::outcome_name(&item.send_state),
126                }))
127            })
128            .collect();
129        let skip = out.len().saturating_sub(limit);
130        out.drain(..skip);
131        out
132    }
133
134    fn send_into(&mut self, room: &RoomId, text: &str, reply_to: Option<&str>, now_ms: i64) -> CmdReply {
135        let reply_to = match reply_to {
136            Some(id) => match crate::EventId::parse(id) {
137                Ok(id) => Some(id),
138                Err(e) => return CmdReply::failed(format!("event id: {e}")),
139            },
140            None => None,
141        };
142        // Refuse here, not at the server: a send into a room this session is not in only earns a refusal
143        // that would be stored as a failed row.
144        let me = self.driver.core.user_id().as_str().to_string();
145        if !self.member_joined(room.as_str(), &me) {
146            return CmdReply::failed("this session has not joined that room (join it first)");
147        }
148        let before = self.texts().len();
149        if let Err(e) = self.dispatch(
150            MessengerCommand::SendMessage {
151                room_id: room.clone(),
152                message: OutgoingMessage { kind: MessageKind::Text, body: text.to_string(), reply_to, edit_of: None },
153                txn_id: None,
154            },
155            now_ms,
156        ) {
157            return CmdReply::failed(format!("send: {e}"));
158        }
159        let mut t = now_ms;
160        for _ in 0..30 {
161            t += 1_000;
162            let _ = self.drive(t, false);
163            std::thread::sleep(std::time::Duration::from_millis(200));
164            if let Some(row) = self.texts().into_iter().rev().find(|r| r.room_id == room.as_str() && r.body == text) {
165                if row.outcome == "sent" {
166                    return CmdReply::ok(json!({ "room": room.as_str(), "event_id": row.event_id }));
167                }
168                if row.outcome.starts_with("failed") {
169                    return CmdReply::failed(format!("send {}", row.outcome));
170                }
171            }
172        }
173        let _ = before;
174        CmdReply::failed("send not confirmed yet (queued)")
175    }
176
177    /// Runs one rich command for this session.
178    pub fn run_command(&mut self, req: &CmdRequest, now_ms: i64) -> CmdReply {
179        let a = &req.args;
180        match req.cmd.as_str() {
181            "rooms.list" => {
182                let rooms: Vec<Value> = self
183                    .rooms()
184                    .into_iter()
185                    .filter(|r| r.membership != "absent" && r.membership != "leave")
186                    .map(|r| {
187                        let id = RoomId::parse(&r.room_id).ok();
188                        let title = id.as_ref().map(|i| self.room_title(i)).unwrap_or_default();
189                        let last = id.as_ref().and_then(|i| self.rows(i, 1).into_iter().next());
190                        json!({ "room": r.room_id, "title": title, "membership": r.membership,
191                                "encrypted": r.encrypted, "last": last })
192                    })
193                    .collect();
194                CmdReply::ok(json!({ "rooms": rooms }))
195            }
196            "rooms.read" => {
197                let Some(spec) = arg_str(a, "room") else { return CmdReply::failed("room is required") };
198                let limit = a.get("limit").and_then(Value::as_u64).unwrap_or(20).min(200) as usize;
199                match self.resolve_room(spec, now_ms) {
200                    Ok(id) => CmdReply::ok(json!({ "room": id.as_str(), "title": self.room_title(&id), "messages": self.rows(&id, limit) })),
201                    Err(e) => CmdReply::failed(e),
202                }
203            }
204            "rooms.send" | "rooms.reply" => {
205                let (Some(spec), Some(text)) = (arg_str(a, "room"), arg_str(a, "text")) else {
206                    return CmdReply::failed("room and text are required");
207                };
208                let reply_to = if req.cmd == "rooms.reply" {
209                    match arg_str(a, "event") {
210                        Some(e) => Some(e),
211                        None => return CmdReply::failed("event is required for a reply"),
212                    }
213                } else {
214                    None
215                };
216                match self.resolve_room(spec, now_ms) {
217                    Ok(id) => self.send_into(&id, text, reply_to, now_ms),
218                    Err(e) => CmdReply::failed(e),
219                }
220            }
221            "rooms.join" | "rooms.leave" => {
222                let Some(spec) = arg_str(a, "room") else { return CmdReply::failed("room is required") };
223                let id = match self.resolve_room(spec, now_ms) {
224                    Ok(id) => id,
225                    Err(e) => return CmdReply::failed(e),
226                };
227                let command = if req.cmd == "rooms.join" {
228                    MessengerCommand::JoinRoom { room_id: id.clone() }
229                } else {
230                    MessengerCommand::LeaveRoom { room_id: id.clone() }
231                };
232                match self.dispatch(command, now_ms) {
233                    Ok(()) => {
234                        self.settle_cmd(now_ms);
235                        CmdReply::ok(json!({ "room": id.as_str() }))
236                    }
237                    Err(e) => CmdReply::failed(e.to_string()),
238                }
239            }
240            "rooms.create" => {
241                let Some(name) = arg_str(a, "name") else { return CmdReply::failed("name is required") };
242                let invite: Vec<UserId> = a
243                    .get("invite")
244                    .and_then(Value::as_array)
245                    .map(|v| v.iter().filter_map(Value::as_str).filter_map(|u| UserId::parse(u).ok()).collect())
246                    .unwrap_or_default();
247                let wanted = arg_str(a, "kind").unwrap_or("group");
248                // `public` is an ordinary public room: the server's public shape (a channel) with the
249                // write level lowered to 0 once the room exists, so everyone who joins can write.
250                let open_for_all = wanted == "public";
251                let kind = match wanted {
252                    "channel" | "public" => CreateRoomKind::Channel { name: name.to_string(), topic: arg_str(a, "topic").map(str::to_string) },
253                    "group" => CreateRoomKind::Group { name: name.to_string(), invite, members_can_invite: true },
254                    other => return CmdReply::failed(format!("kind must be group, channel or public, not {other}")),
255                };
256                match self.dispatch(MessengerCommand::CreateRoom { kind }, now_ms) {
257                    Ok(()) => {
258                        let mut found = None;
259                        for i in 1..=25 {
260                            let _ = self.drive(now_ms + i * 1_000, false);
261                            std::thread::sleep(std::time::Duration::from_millis(200));
262                            found = self.rooms().into_iter().find(|r| {
263                                r.membership == "join"
264                                    && RoomId::parse(&r.room_id).map(|i| self.room_title(&i) == name).unwrap_or(false)
265                            });
266                            if found.is_some() {
267                                break;
268                            }
269                        }
270                        if open_for_all {
271                            let Some(room) = found.as_ref().and_then(|r| RoomId::parse(&r.room_id).ok()) else {
272                                return CmdReply::failed("the room was created but did not show up; set its write level with rooms.power");
273                            };
274                            let lowered = MessengerCommand::SetPowerLevels { room_id: room, events_default: Some(0), users: Vec::new() };
275                            if let Err(e) = self.dispatch(lowered, now_ms) {
276                                return CmdReply::failed(format!("the room was created but its write level was not lowered: {e}"));
277                            }
278                            self.settle_cmd(now_ms);
279                        }
280                        CmdReply::ok(json!({ "room": found.map(|r| r.room_id) }))
281                    }
282                    Err(e) => CmdReply::failed(e.to_string()),
283                }
284            }
285            "rooms.power" => {
286                let Some(spec) = arg_str(a, "room") else { return CmdReply::failed("room is required") };
287                let id = match self.resolve_room(spec, now_ms) {
288                    Ok(id) => id,
289                    Err(e) => return CmdReply::failed(e),
290                };
291                let events_default = a.get("events_default").and_then(Value::as_i64);
292                let mut users: Vec<(UserId, i64)> = Vec::new();
293                if let Some(who) = arg_str(a, "user") {
294                    let Some(level) = a.get("level").and_then(Value::as_i64) else {
295                        return CmdReply::failed("level is required with user");
296                    };
297                    let user = if who.starts_with('@') && who.contains(':') {
298                        who.to_string()
299                    } else {
300                        match self.find_nick(who, now_ms) {
301                            Ok(found) => found.user_id,
302                            Err(e) => return CmdReply::failed(format!("{who}: {e}")),
303                        }
304                    };
305                    let Ok(user_id) = UserId::parse(&user) else { return CmdReply::failed("bad user id") };
306                    users.push((user_id, level));
307                }
308                if events_default.is_none() && users.is_empty() {
309                    return CmdReply::failed("nothing to change: give events_default, or user and level");
310                }
311                match self.dispatch(MessengerCommand::SetPowerLevels { room_id: id.clone(), events_default, users }, now_ms) {
312                    Ok(()) => {
313                        self.settle_cmd(now_ms);
314                        CmdReply::ok(json!({ "room": id.as_str() }))
315                    }
316                    Err(e) => CmdReply::failed(e.to_string()),
317                }
318            }
319            "rooms.invite" => {
320                let (Some(spec), Some(who)) = (arg_str(a, "room"), arg_str(a, "user")) else {
321                    return CmdReply::failed("room and user are required");
322                };
323                let id = match self.resolve_room(spec, now_ms) {
324                    Ok(id) => id,
325                    Err(e) => return CmdReply::failed(e),
326                };
327                let user = if who.starts_with('@') && who.contains(':') {
328                    who.to_string()
329                } else {
330                    match self.find_nick(who, now_ms) {
331                        Ok(found) => found.user_id,
332                        Err(e) => return CmdReply::failed(format!("{who}: {e}")),
333                    }
334                };
335                let Ok(user_id) = UserId::parse(&user) else { return CmdReply::failed("bad user id") };
336                match self.dispatch(MessengerCommand::Invite { room_id: id.clone(), user_id }, now_ms) {
337                    Ok(()) => {
338                        self.settle_cmd(now_ms);
339                        CmdReply::ok(json!({ "room": id.as_str() }))
340                    }
341                    Err(e) => CmdReply::failed(e.to_string()),
342                }
343            }
344            "mentions" => {
345                let limit = a.get("limit").and_then(Value::as_u64).unwrap_or(20).min(200) as usize;
346                let me = self.nick().unwrap_or("").to_ascii_lowercase();
347                let my_id = self.driver.core.user_id().as_str().to_string();
348                let ids: Vec<RoomId> = self.driver.core.room_ids().cloned().collect();
349                let mut hits = Vec::new();
350                for id in ids {
351                    for mut row in self.rows(&id, 200) {
352                        let body = row["body"].as_str().unwrap_or("").to_ascii_lowercase();
353                        let from_me = row["from_id"].as_str() == Some(my_id.as_str());
354                        if !from_me && !me.is_empty() && body.contains(&format!("@{me}")) {
355                            row["room"] = json!(id.as_str());
356                            hits.push(row);
357                        }
358                    }
359                }
360                hits.sort_by_key(|r| r["ts"].as_i64().unwrap_or(0));
361                let skip = hits.len().saturating_sub(limit);
362                hits.drain(..skip);
363                CmdReply::ok(json!({ "mentions": hits }))
364            }
365            "threads.read" => {
366                let (Some(spec), Some(root)) = (arg_str(a, "room"), arg_str(a, "event")) else {
367                    return CmdReply::failed("room and event are required");
368                };
369                match self.resolve_room(spec, now_ms) {
370                    Ok(id) => {
371                        let rows: Vec<Value> = self
372                            .rows(&id, 500)
373                            .into_iter()
374                            .filter(|r| r["event_id"].as_str() == Some(root) || r["thread"].as_str() == Some(root) || r["reply_to"].as_str() == Some(root))
375                            .collect();
376                        CmdReply::ok(json!({ "room": id.as_str(), "messages": rows }))
377                    }
378                    Err(e) => CmdReply::failed(e),
379                }
380            }
381            other => CmdReply::failed(format!("unknown command {other}")),
382        }
383    }
384}
mail4agent_messenger_shell/cmd.rs

mail4agent_messenger_shell/
cmd.rs
