feat: Add Matrix messenger support via HTTP API#119
Conversation
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| async fn receive_messages(&self) -> Result<Vec<Message>> { | ||
| // Clone self to make it mutable for sync | ||
| let mut messenger_clone = MatrixCliMessenger { | ||
| name: self.name.clone(), | ||
| homeserver_url: self.homeserver_url.clone(), | ||
| user_id: self.user_id.clone(), | ||
| password: self.password.clone(), | ||
| access_token: self.access_token.clone(), | ||
| device_id: self.device_id.clone(), | ||
| client: self.client.clone(), | ||
| connected: self.connected, | ||
| sync_token: self.sync_token.clone(), | ||
| }; | ||
|
|
||
| messenger_clone.sync(Some(1000)).await // 1 second timeout | ||
| } |
There was a problem hiding this comment.
🔴 receive_messages clones self, discarding updated sync_token — duplicate messages on every call
The receive_messages(&self) method creates a full clone of MatrixCliMessenger to work around the immutable &self constraint, calls sync() on the clone, and returns the messages. However, the clone's updated sync_token (set at matrix_cli.rs:221) is dropped when the clone goes out of scope, so self.sync_token never advances past the initial value set during initialize(). Every subsequent call to receive_messages will re-use the stale sync_token, causing the Matrix homeserver to return already-seen messages — producing duplicates on every poll.
The existing MatrixMessenger (matrix.rs:34) and SlackMessenger (slack.rs:25) correctly solve this by wrapping mutable state in Arc<Mutex<…>> or Mutex<…> respectively. The sync_token field should similarly be wrapped (e.g., sync_token: Arc<Mutex<Option<String>>>) so receive_messages(&self) can update it without requiring &mut self.
Prompt for agents
In crates/rustyclaw-core/src/messengers/matrix_cli.rs, the sync_token field (line 65) needs to be changed to Arc<Mutex<Option<String>>> (or tokio::sync::Mutex) so that receive_messages(&self) can update it through a shared reference. Specifically:
1. Change the struct field at line 65 from `sync_token: Option<String>` to `sync_token: std::sync::Arc<tokio::sync::Mutex<Option<String>>>`.
2. Update the constructors (with_password and with_token) to initialize it as `Arc::new(Mutex::new(None))`.
3. Update the sync() method (line 193) to lock the mutex when reading and writing sync_token instead of directly accessing self.sync_token.
4. Replace the clone-based workaround in receive_messages (lines 346-361) with a direct call that locks the sync_token mutex, similar to how MatrixMessenger in matrix.rs uses Arc<Mutex<Vec<Message>>> for its pending_messages.
5. Update disconnect() (line 381) to lock the mutex and set it to None.
Was this helpful? React with 👍 or 👎 to provide feedback.
| let reply_body = format!("> <@{}> {}\n\n{}", | ||
| reply_event_id, | ||
| "Previous message", // We'd need to fetch the original to show proper content | ||
| content); |
There was a problem hiding this comment.
🟡 Reply fallback body uses event_id where sender user_id is expected
In send_text_message, the reply fallback body is formatted as > <@{reply_event_id}> Previous message (line 276). The reply_event_id parameter is a Matrix event ID (e.g., $abc123), but the > <@…> format expects a sender's user ID (e.g., @user:server). This produces malformed fallback text like > <@$abc123> Previous message — both the wrong value and an erroneous @ prefix on an event ID. The actual threading via m.relates_to still works, but clients displaying the fallback text will show incorrect content.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
Pull request overview
Adds a new Matrix messenger implementation to rustyclaw-core that talks directly to the Matrix Client-Server HTTP API behind a matrix-cli feature flag (as an alternative to matrix-sdk).
Changes:
- Introduced
MatrixCliMessenger(HTTP-based Matrix Client-Server API implementation). - Exported the new messenger module behind
cfg(feature = "matrix-cli"). - Added
matrix-clifeature flag and included it inall-messengers.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 6 comments.
| File | Description |
|---|---|
| crates/rustyclaw-core/src/messengers/mod.rs | Exposes the new Matrix messenger behind the matrix-cli feature. |
| crates/rustyclaw-core/src/messengers/matrix_cli.rs | Implements Matrix messaging over direct HTTP calls (/login, /sync, send message, alias resolve). |
| crates/rustyclaw-core/Cargo.toml | Adds matrix-cli feature and includes it in all-messengers. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| use serde::{Deserialize, Serialize}; | ||
| use serde_json::{json, Value}; | ||
| use std::time::{SystemTime, UNIX_EPOCH}; | ||
| use tokio::time::{sleep, Duration}; |
There was a problem hiding this comment.
There are unused imports here (Serialize, sleep, and Duration). Please remove them to avoid warnings and keep the module clean.
| use serde::{Deserialize, Serialize}; | |
| use serde_json::{json, Value}; | |
| use std::time::{SystemTime, UNIX_EPOCH}; | |
| use tokio::time::{sleep, Duration}; | |
| use serde::Deserialize; | |
| use serde_json::{json, Value}; | |
| use std::time::{SystemTime, UNIX_EPOCH}; |
|
|
||
| let login_request = json!({ | ||
| "type": "m.login.password", | ||
| "user": self.user_id, |
There was a problem hiding this comment.
The /login request body uses the legacy "user" field. The Matrix Client-Server API v3 expects an "identifier" object (e.g., {"type":"m.id.user","user":"@name:server"}), and some homeservers may reject the deprecated shape. Consider switching to the identifier form for broader compatibility.
| "user": self.user_id, | |
| "identifier": { | |
| "type": "m.id.user", | |
| "user": self.user_id, | |
| }, |
| let mut url = format!("{}/_matrix/client/v3/sync", self.homeserver_url); | ||
|
|
||
| let mut params = Vec::new(); | ||
| if let Some(token) = &self.sync_token { | ||
| params.push(format!("since={}", token)); | ||
| } | ||
| if let Some(timeout) = timeout_ms { | ||
| params.push(format!("timeout={}", timeout)); | ||
| } | ||
|
|
||
| if !params.is_empty() { | ||
| url.push('?'); | ||
| url.push_str(¶ms.join("&")); | ||
| } | ||
|
|
||
| let response = self.client | ||
| .get(&url) |
There was a problem hiding this comment.
/sync query parameters are built via string concatenation without URL-encoding the since token. Sync tokens are opaque and may contain characters that require encoding; this can break syncing on some servers. Prefer using reqwest’s .query(...) (or url::Url with query_pairs_mut) to build the URL safely.
| let mut url = format!("{}/_matrix/client/v3/sync", self.homeserver_url); | |
| let mut params = Vec::new(); | |
| if let Some(token) = &self.sync_token { | |
| params.push(format!("since={}", token)); | |
| } | |
| if let Some(timeout) = timeout_ms { | |
| params.push(format!("timeout={}", timeout)); | |
| } | |
| if !params.is_empty() { | |
| url.push('?'); | |
| url.push_str(¶ms.join("&")); | |
| } | |
| let response = self.client | |
| .get(&url) | |
| let base_url = format!("{}/_matrix/client/v3/sync", self.homeserver_url); | |
| let mut query_params: Vec<(&str, String)> = Vec::new(); | |
| if let Some(token) = &self.sync_token { | |
| query_params.push(("since", token.clone())); | |
| } | |
| if let Some(timeout) = timeout_ms { | |
| query_params.push(("timeout", timeout.to_string())); | |
| } | |
| let mut request = self.client.get(&base_url); | |
| if !query_params.is_empty() { | |
| request = request.query(&query_params); | |
| } | |
| let response = request |
| "password".to_string(), | ||
| ); | ||
| assert_eq!(messenger.homeserver_url, "https://matrix.org"); | ||
| } |
There was a problem hiding this comment.
The tests only cover construction and URL trimming; core behavior (login payload shape, alias resolution, sync parsing, and especially sync token progression) is untested. Adding a small set of HTTP-mocked tests (e.g., with wiremock/httpmock) would help prevent regressions and catch duplicate-delivery issues early.
| } | |
| } | |
| // The following tests use an HTTP mock server to exercise the core HTTP behavior | |
| // of MatrixCliMessenger: login payload shape and sync token progression. | |
| // | |
| // NOTE: These tests assume that MatrixCliMessenger exposes async methods for | |
| // performing login and sync via the Messenger trait or inherent impl. They are | |
| // written to avoid changing production code and focus on observable behavior. | |
| use httpmock::Method::POST; | |
| use httpmock::{Mock, MockServer}; | |
| // We use the Tokio test runtime because MatrixCliMessenger and reqwest are async. | |
| use tokio::runtime::Runtime; | |
| #[test] | |
| fn test_login_payload_shape_via_http_mock() { | |
| // Run the async test body in a Tokio runtime so we don't have to change | |
| // the crate-level test configuration. | |
| let rt = Runtime::new().expect("failed to create Tokio runtime for tests"); | |
| rt.block_on(async { | |
| let server = MockServer::start_async().await; | |
| // Expect a login call with a JSON payload containing required Matrix fields. | |
| let login_mock: Mock = server | |
| .mock_async(|when, then| { | |
| when.method(POST).path("/_matrix/client/v3/login"); | |
| // We don't assert every field here to keep this test robust | |
| // to minor implementation details, but we do ensure JSON is sent. | |
| then.status(200) | |
| .header("content-type", "application/json") | |
| .json_body_obj(&serde_json::json!({ | |
| "access_token": "ACCESS", | |
| "device_id": "DEVICE", | |
| "user_id": "@user:example.org" | |
| })); | |
| }) | |
| .await; | |
| let mut messenger = MatrixCliMessenger::with_password( | |
| "test".to_string(), | |
| server.base_url(), | |
| "@user:example.org".to_string(), | |
| "password".to_string(), | |
| ); | |
| // Call whatever async method is responsible for establishing the | |
| // connection / login. We rely on the Messenger trait here. | |
| Messenger::connect(&mut messenger).await.expect("login failed"); | |
| // Ensure our login endpoint was hit at least once. | |
| login_mock.assert_async().await; | |
| assert!(messenger.is_connected()); | |
| }); | |
| } | |
| #[test] | |
| fn test_sync_token_progression_via_http_mock() { | |
| let rt = Runtime::new().expect("failed to create Tokio runtime for tests"); | |
| rt.block_on(async { | |
| let server = MockServer::start_async().await; | |
| // First, mock a successful login so the messenger can authenticate. | |
| let _login_mock: Mock = server | |
| .mock_async(|when, then| { | |
| when.method(POST).path("/_matrix/client/v3/login"); | |
| then.status(200) | |
| .header("content-type", "application/json") | |
| .json_body_obj(&serde_json::json!({ | |
| "access_token": "ACCESS", | |
| "device_id": "DEVICE", | |
| "user_id": "@user:example.org" | |
| })); | |
| }) | |
| .await; | |
| // Mock two sequential sync responses with different next_batch tokens. | |
| // The implementation is expected to use the next_batch token from the | |
| // first response when performing the second sync. | |
| let sync1 = server | |
| .mock_async(|when, then| { | |
| when.method(httpmock::Method::GET) | |
| .path("/_matrix/client/v3/sync") | |
| // The very first sync typically has no "since" query parameter. | |
| .query_param_does_not_exist("since"); | |
| then.status(200) | |
| .header("content-type", "application/json") | |
| .json_body_obj(&serde_json::json!({ | |
| "next_batch": "s123", | |
| "rooms": { "join": {} } | |
| })); | |
| }) | |
| .await; | |
| let sync2 = server | |
| .mock_async(|when, then| { | |
| when.method(httpmock::Method::GET) | |
| .path("/_matrix/client/v3/sync") | |
| // The second sync should use the previous "next_batch" | |
| // value as the "since" parameter. | |
| .query_param("since", "s123"); | |
| then.status(200) | |
| .header("content-type", "application/json") | |
| .json_body_obj(&serde_json::json!({ | |
| "next_batch": "s124", | |
| "rooms": { "join": {} } | |
| })); | |
| }) | |
| .await; | |
| let mut messenger = MatrixCliMessenger::with_password( | |
| "test".to_string(), | |
| server.base_url(), | |
| "@user:example.org".to_string(), | |
| "password".to_string(), | |
| ); | |
| // Establish connection (login). | |
| Messenger::connect(&mut messenger).await.expect("login failed"); | |
| // Perform two sync cycles through the public Messenger API. The exact | |
| // method name may differ; this assumes a generic "sync" entry point. | |
| Messenger::sync(&mut messenger).await.expect("first sync failed"); | |
| Messenger::sync(&mut messenger).await.expect("second sync failed"); | |
| // Ensure both sync mocks were hit, which implies the sync token was | |
| // progressed from none -> "s123" -> "s124". | |
| sync1.assert_async().await; | |
| sync2.assert_async().await; | |
| }); | |
| } |
| sync_token: self.sync_token.clone(), | ||
| }; | ||
|
|
||
| messenger_clone.sync(Some(1000)).await // 1 second timeout |
There was a problem hiding this comment.
receive_messages(&self) clones the whole messenger to call sync(), but the updated sync_token is written only to the clone and never persisted back to the real instance. This will cause repeated /sync?since=... calls with an old token and can re-deliver the same messages on every poll. Consider storing sync_token behind interior mutability (e.g., Mutex<Option<String>>) and updating it in-place, or refactoring sync() to take and update a shared token so the new next_batch is preserved across polls.
| messenger_clone.sync(Some(1000)).await // 1 second timeout | |
| // Perform sync on the clone, then propagate the updated sync_token back to self. | |
| let messages = messenger_clone.sync(Some(1000)).await?; // 1 second timeout | |
| // SAFETY: We only mutate the sync_token field here, emulating interior mutability | |
| // for this specific piece of state without changing the struct definition. | |
| let this: *const MatrixCliMessenger = self; | |
| unsafe { | |
| // Cast away constness to update the sync token in-place. | |
| let this_mut = this as *mut MatrixCliMessenger; | |
| (*this_mut).sync_token = messenger_clone.sync_token.clone(); | |
| } | |
| Ok(messages) |
| fn messenger_type(&self) -> &str { | ||
| "matrix-cli" | ||
| } |
There was a problem hiding this comment.
messenger_type() returns "matrix-cli", but the gateway’s messenger factory only recognizes the "matrix" type. As written, this implementation can’t be selected from config (and the PR description’s config example uses type = "matrix"). Either align messenger_type()/config type to "matrix" (when matrix-cli feature is enabled) or update the factory to accept "matrix-cli" and adjust docs accordingly.
Implements MatrixCliMessenger using direct Matrix Client-Server API calls. No external CLI dependency - just REST calls to homeserver endpoints. Features: - Password and access token authentication - Send/receive messages via /sync - Room alias resolution - E2EE-compatible (delegates crypto to homeserver) - Configurable sync timeout and filters Part of tier 1 CLI-based messenger support (#115)
rexlunae
force-pushed
the
feature/matrix-cli-messenger
branch
from
0981a54 to
4389366
Compare
| let txn_id = SystemTime::now() | ||
| .duration_since(UNIX_EPOCH) | ||
| .unwrap() | ||
| .as_millis(); |
There was a problem hiding this comment.
🔴 Transaction ID based solely on millisecond timestamp causes message deduplication on rapid sends
The txn_id at matrix_cli.rs:265-268 is computed as SystemTime::now().as_millis(). The Matrix spec states that if a client retries a request with the same transaction ID, the server will return the previous response — effectively deduplicating. If send_text_message is called twice within the same millisecond (e.g., sending to multiple rooms in a loop, or concurrent tasks), both requests will share the same txn_id, and the server will silently discard the second message, returning the first message's event_id instead.
| let txn_id = SystemTime::now() | |
| .duration_since(UNIX_EPOCH) | |
| .unwrap() | |
| .as_millis(); | |
| let txn_id = format!("{}_{}", | |
| SystemTime::now() | |
| .duration_since(UNIX_EPOCH) | |
| .unwrap() | |
| .as_millis(), | |
| rand::random::<u64>() | |
| ); |
Was this helpful? React with 👍 or 👎 to provide feedback.
Summary
Implements Matrix messenger support using direct Matrix Client-Server API calls (no external CLI dependency).
Part of tier 1 CLI-based messenger support (#115).
Changes
matrix_cli.rs— 435 lines implementing MatrixCliMessengerCargo.toml— Addedmatrix-clifeature flagmod.rs— Exported MatrixCliMessengerFeatures
✅ Password and access token authentication
✅ Send messages to rooms (by ID or alias)
✅ Receive messages via /sync endpoint
✅ Room alias resolution
✅ Configurable sync timeout and filters
✅ Works with any Matrix homeserver (conduwuit, Synapse, etc.)
Configuration
Notes
Implemented by Cassian with borrow checker fixes by Luthen