netget/client/matrix/
mod.rs

1//! Matrix client (client-server API v3). Logs in with a password, then one task follows
2//! `/sync` and raises what other people did (messages, invites, membership), while the
3//! session task performs the handler's actions, one HTTP/1.1 connection per request.
4pub mod actions;
5use crate::client::llm_budget::call_llm_for_client;
6use crate::llm::actions::client_trait::{Client, ClientActionResult};
7use crate::logging::emit::Log;
8use crate::protocol::{ConnectContext, Event};
9use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
10use crate::state::{AccessLogOwner, ClientStatus};
11pub use actions::MatrixClientProtocol;
12use anyhow::{bail, ensure, Context, Result};
13use bytes::Bytes;
14use http_body_util::{BodyExt, Full, Limited};
15use hyper::{
16    client::conn::http1,
17    header::{AUTHORIZATION, CONNECTION, CONTENT_LENGTH, CONTENT_TYPE, HOST},
18    Request, Uri,
19};
20use hyper_util::rt::TokioIo;
21use serde_json::{json, Value};
22use std::time::Duration;
23use tokio::sync::mpsc;
24
25/// Largest answer read (a first `/sync` on a busy account is the big one).
26pub const MAX_ANSWER_BYTES: usize = 4 * 1024 * 1024;
27/// One ordinary request, connect to last byte.
28pub const IO_TIMEOUT: Duration = Duration::from_secs(15);
29/// How long each `/sync` asks the server to wait for something new.
30pub const SYNC_WAIT: Duration = Duration::from_secs(25);
31/// After a failed `/sync`, wait this long before the next.
32pub const SYNC_BACKOFF: Duration = Duration::from_secs(2);
33/// A handler chain (answer → event → answer …) stops after this many follow-ups.
34pub const MAX_FOLLOWUP_DEPTH: usize = 8;
35
36#[derive(Clone)]
37struct Origin {
38    authority: String,
39    connect_addr: String,
40}
41
42fn origin(value: &str) -> Result<Origin> {
43    let value = if value.starts_with("http://") {
44        value.to_string()
45    } else {
46        format!("http://{value}")
47    };
48    let uri: Uri = value.parse().context("invalid homeserver address")?;
49    ensure!(
50        matches!(uri.path(), "" | "/") && uri.query().is_none(),
51        "give the homeserver as host:port or http://host:port, without a path"
52    );
53    let authority = uri.authority().context("homeserver host required")?;
54    Ok(Origin {
55        authority: authority.as_str().to_string(),
56        connect_addr: format!(
57            "{}:{}",
58            authority.host(),
59            authority.port_u16().unwrap_or(8008)
60        ),
61    })
62}
63
64struct Answer {
65    status: u16,
66    body: Value,
67}
68
69/// One request under `/_matrix/client/v3/`.
70async fn exchange(
71    origin: &Origin,
72    token: Option<&str>,
73    method: &str,
74    path: &str,
75    body: Option<&Value>,
76    deadline: Duration,
77) -> Result<Answer> {
78    tokio::time::timeout(deadline, async {
79        let socket = tokio::net::TcpStream::connect(&origin.connect_addr).await?;
80        let (mut sender, connection) = http1::Builder::new()
81            .max_headers(64)
82            .max_buf_size(64 * 1024)
83            .handshake(TokioIo::new(socket))
84            .await?;
85        let payload = body.map(|b| b.to_string()).unwrap_or_default();
86        let mut request = Request::builder()
87            .method(method)
88            .uri(format!("/_matrix/client/v3/{path}"))
89            .header(HOST, &origin.authority)
90            .header(CONNECTION, "close")
91            .header(CONTENT_LENGTH, payload.len());
92        if body.is_some() {
93            request = request.header(CONTENT_TYPE, "application/json");
94        }
95        if let Some(t) = token {
96            request = request.header(AUTHORIZATION, format!("Bearer {t}"));
97        }
98        let request = request.body(Full::new(Bytes::from(payload)))?;
99        let exchange = async {
100            let response = sender.send_request(request).await?;
101            let (parts, body) = response.into_parts();
102            let bytes = Limited::new(body, MAX_ANSWER_BYTES)
103                .collect()
104                .await
105                .map_err(|_| anyhow::anyhow!("response body exceeds 4 MiB or is incomplete"))?
106                .to_bytes();
107            Ok::<_, anyhow::Error>(Answer {
108                status: parts.status.as_u16(),
109                body: serde_json::from_slice(&bytes).unwrap_or(Value::Null),
110            })
111        };
112        tokio::pin!(connection);
113        tokio::pin!(exchange);
114        tokio::select! {
115            r = &mut exchange => r,
116            r = &mut connection => {
117                r.context("HTTP connection failed")?;
118                exchange.await
119            }
120        }
121    })
122    .await
123    .context("Matrix request deadline exceeded")?
124}
125
126fn matrix_error(answer: &Answer) -> (String, String) {
127    (
128        answer.body["errcode"]
129            .as_str()
130            .unwrap_or("M_UNKNOWN")
131            .to_string(),
132        crate::utils::truncate::truncate_for_log(
133            answer.body["error"].as_str().unwrap_or_default(),
134            512,
135        )
136        .to_string(),
137    )
138}
139
140struct Login {
141    user_id: String,
142    device_id: String,
143    token: String,
144}
145
146async fn login(origin: &Origin, ctx: &ConnectContext) -> Result<Login> {
147    let params = ctx
148        .startup_params
149        .as_ref()
150        .context("startup_params user and password are required")?;
151    let user = params.get_string("user")?;
152    let password = params.get_string("password")?;
153    let device = params.get_optional_string("device_id")?;
154    let mut body = json!({"type": "m.login.password", "identifier": {"type": "m.id.user", "user": user},
155                          "password": password, "initial_device_display_name": "NetGet"});
156    if let Some(d) = device {
157        body["device_id"] = json!(d);
158    }
159    let answer = exchange(origin, None, "POST", "login", Some(&body), IO_TIMEOUT).await?;
160    if answer.status != 200 {
161        let (code, error) = matrix_error(&answer);
162        bail!("login refused: {} {code} {error}", answer.status);
163    }
164    let field = |k: &str| -> Result<String> {
165        answer.body[k]
166            .as_str()
167            .map(str::to_string)
168            .with_context(|| format!("login answer has no {k}"))
169    };
170    Ok(Login {
171        user_id: field("user_id")?,
172        device_id: field("device_id")?,
173        token: field("access_token")?,
174    })
175}
176
177/// The events one `/sync` answer raises for the handler; `first` skips the timeline (what
178/// happened before login is not news).
179fn sync_events(body: &Value, me: &str, first: bool) -> Vec<Event> {
180    let mut out = Vec::new();
181    if let Some(invites) = body["rooms"]["invite"].as_object() {
182        for (room, data) in invites {
183            let state = data["invite_state"]["events"]
184                .as_array()
185                .cloned()
186                .unwrap_or_default();
187            let sender = state
188                .iter()
189                .find(|e| e["type"] == "m.room.member" && e["state_key"] == me)
190                .and_then(|e| e["sender"].as_str())
191                .map(str::to_string);
192            let name = state
193                .iter()
194                .find(|e| e["type"] == "m.room.name")
195                .and_then(|e| e["content"]["name"].as_str())
196                .map(str::to_string);
197            out.push(Event::new(
198                &actions::INVITE_EVENT,
199                json!({"room_id": room, "sender": sender, "room_name": name}),
200            ));
201        }
202    }
203    if first {
204        return out;
205    }
206    if let Some(joined) = body["rooms"]["join"].as_object() {
207        for (room, data) in joined {
208            for e in data["timeline"]["events"].as_array().into_iter().flatten() {
209                let kind = e["type"].as_str().unwrap_or_default();
210                if kind == "m.room.member" {
211                    let user = e["state_key"].as_str().unwrap_or_default();
212                    if user != me {
213                        out.push(Event::new(
214                            &actions::MEMBER_EVENT,
215                            json!({"room_id": room, "user_id": user,
216                                   "membership": e["content"]["membership"]}),
217                        ));
218                    }
219                    continue;
220                }
221                if e.get("state_key").is_some() || e["sender"] == me {
222                    continue;
223                }
224                out.push(Event::new(
225                    &actions::MESSAGE_EVENT,
226                    json!({"room_id": room, "sender": e["sender"], "event_type": kind,
227                           "content": e["content"], "event_id": e["event_id"]}),
228                ));
229            }
230        }
231    }
232    out
233}
234
235pub async fn connect(ctx: ConnectContext) -> Result<std::net::SocketAddr> {
236    let origin = origin(&ctx.remote_addr)?;
237    let login = login(&origin, &ctx).await?;
238    let first = exchange(
239        &origin,
240        Some(&login.token),
241        "GET",
242        "sync?timeout=0",
243        None,
244        IO_TIMEOUT,
245    )
246    .await?;
247    ensure!(
248        first.status == 200,
249        "first /sync failed: {} {:?}",
250        first.status,
251        matrix_error(&first)
252    );
253    let mut since = first.body["next_batch"]
254        .as_str()
255        .context("/sync answer has no next_batch")?
256        .to_string();
257    let joined: Vec<String> = first.body["rooms"]["join"]
258        .as_object()
259        .map(|m| m.keys().cloned().collect())
260        .unwrap_or_default();
261    Log::new(Some(&ctx.status_tx)).info(format!(
262        "Matrix client logged in as {} ({}) at {}",
263        login.user_id, login.device_id, origin.authority
264    ));
265    ctx.state
266        .update_client_status(ctx.client_id, ClientStatus::Connected)
267        .await;
268    let external =
269        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
270    let (internal_tx, internal_rx) = mpsc::channel::<(Value, usize)>(32);
271    let (event_tx, mut event_rx) = mpsc::channel::<(Event, usize)>(64);
272    event_tx.try_send((
273        Event::new(
274            &actions::CONNECTED_EVENT,
275            json!({"user_id": login.user_id, "device_id": login.device_id, "joined_rooms": joined}),
276        ),
277        0,
278    ))?;
279    for e in sync_events(&first.body, &login.user_id, true) {
280        event_tx.try_send((e, 0))?;
281    }
282
283    let events_ctx = ctx.clone();
284    let dispatcher = tokio::spawn(async move {
285        while let Some((event, depth)) = event_rx.recv().await {
286            events_ctx
287                .state
288                .record_access_log(
289                    AccessLogOwner::Client(events_ctx.client_id.as_u32()),
290                    "Matrix",
291                    None,
292                    event.id(),
293                    event.data.clone(),
294                    vec![],
295                )
296                .await;
297            let instruction = events_ctx
298                .state
299                .get_instruction_for_client(events_ctx.client_id)
300                .await
301                .unwrap_or_default();
302            let memory = events_ctx
303                .state
304                .get_memory_for_client(events_ctx.client_id)
305                .await
306                .unwrap_or_default();
307            match call_llm_for_client(
308                &events_ctx.llm_client,
309                &events_ctx.state,
310                events_ctx.client_id.to_string(),
311                &instruction,
312                &memory,
313                Some(&event),
314                &MatrixClientProtocol,
315                &events_ctx.status_tx,
316            )
317            .await
318            {
319                Ok(result) => {
320                    if let Some(memory) = result.memory_updates {
321                        events_ctx
322                            .state
323                            .set_memory_for_client(events_ctx.client_id, memory)
324                            .await;
325                    }
326                    for action in result.actions {
327                        if internal_tx.send((action, depth + 1)).await.is_err() {
328                            return;
329                        }
330                    }
331                }
332                Err(e) => Log::new(Some(&events_ctx.status_tx))
333                    .warn(format!("Matrix client handler: {e}")),
334            }
335        }
336    });
337    let dispatcher_abort = dispatcher.abort_handle();
338    ctx.state
339        .register_client_task(ctx.client_id, dispatcher)
340        .await;
341
342    let sync_ctx = ctx.clone();
343    let sync_origin = origin.clone();
344    let sync_token = login.token.clone();
345    let me = login.user_id.clone();
346    let sync_events_tx = event_tx.clone();
347    let syncer = tokio::spawn(async move {
348        let log = Log::new(Some(&sync_ctx.status_tx));
349        loop {
350            let path = format!(
351                "sync?timeout={}&since={}",
352                SYNC_WAIT.as_millis(),
353                urlencoding::encode(&since)
354            );
355            match exchange(
356                &sync_origin,
357                Some(&sync_token),
358                "GET",
359                &path,
360                None,
361                SYNC_WAIT + IO_TIMEOUT,
362            )
363            .await
364            {
365                Ok(a) if a.status == 200 => {
366                    if let Some(next) = a.body["next_batch"].as_str() {
367                        since = next.to_string();
368                    }
369                    for e in sync_events(&a.body, &me, false) {
370                        if sync_events_tx.send((e, 0)).await.is_err() {
371                            return;
372                        }
373                    }
374                }
375                Ok(a) => {
376                    let (code, error) = matrix_error(&a);
377                    log.warn(format!("Matrix /sync answered {} {code} {error}", a.status));
378                    if a.status == 401 {
379                        return;
380                    }
381                    tokio::time::sleep(SYNC_BACKOFF).await;
382                }
383                Err(e) => {
384                    log.warn(format!("Matrix /sync failed: {e:#}"));
385                    tokio::time::sleep(SYNC_BACKOFF).await;
386                }
387            }
388        }
389    });
390    let syncer_abort = syncer.abort_handle();
391    ctx.state.register_client_task(ctx.client_id, syncer).await;
392
393    let session_ctx = ctx.clone();
394    let task = tokio::spawn(async move {
395        let result = session(
396            &session_ctx,
397            &origin,
398            &login,
399            external,
400            internal_rx,
401            event_tx,
402        )
403        .await;
404        syncer_abort.abort();
405        dispatcher_abort.abort();
406        let _ = exchange(
407            &origin,
408            Some(&login.token),
409            "POST",
410            "logout",
411            Some(&json!({})),
412            IO_TIMEOUT,
413        )
414        .await;
415        let status = match result {
416            Ok(()) => ClientStatus::Disconnected,
417            Err(e) => {
418                Log::new(Some(&session_ctx.status_tx)).warn(format!("Matrix client ended: {e}"));
419                ClientStatus::Error(e.to_string())
420            }
421        };
422        session_ctx
423            .state
424            .update_client_status(session_ctx.client_id, status)
425            .await;
426        session_ctx
427            .state
428            .remove_client_handle(session_ctx.client_id)
429            .await;
430        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
431    });
432    ctx.state.register_client_task(ctx.client_id, task).await;
433    Ok("0.0.0.0:0".parse()?)
434}
435
436async fn session(
437    ctx: &ConnectContext,
438    origin: &Origin,
439    login: &Login,
440    mut external: mpsc::Receiver<ClientCommand>,
441    mut internal: mpsc::Receiver<(Value, usize)>,
442    events: mpsc::Sender<(Event, usize)>,
443) -> Result<()> {
444    let log = Log::new(Some(&ctx.status_tx));
445    let txn_prefix = format!("ng{}", crate::server::matrix::hub::random_id(8));
446    let mut txn = 0u64;
447    loop {
448        let (action, depth, mut injected) = tokio::select! {
449            command = external.recv() => match command {
450                Some(c) => (c.action.clone(), 0, Some(c)),
451                None => return Ok(()),
452            },
453            action = internal.recv() => match action {
454                Some((a, depth)) => (a, depth, None),
455                None => return Ok(()),
456            },
457        };
458        let reply = |injected: &mut Option<ClientCommand>, outcome: ClientSendOutcome| {
459            if let Some(command) = injected.take() {
460                crate::client::command_support::reply(command, Ok(outcome));
461            }
462        };
463        match MatrixClientProtocol.execute_action(action.clone()) {
464            Ok(ClientActionResult::Disconnect) => {
465                reply(&mut injected, ClientSendOutcome::Disconnected);
466                return Ok(());
467            }
468            Ok(_) => {}
469            Err(e) => {
470                log.warn(format!("Matrix client action refused: {e:#}"));
471                reply(
472                    &mut injected,
473                    ClientSendOutcome::Rejected {
474                        error: e.to_string(),
475                    },
476                );
477                continue;
478            }
479        }
480        if depth > MAX_FOLLOWUP_DEPTH {
481            log.warn(format!(
482                "Matrix client: handler chain stopped after {MAX_FOLLOWUP_DEPTH} follow-ups"
483            ));
484            continue;
485        }
486        if injected.is_some() {
487            ctx.state
488                .record_access_log(
489                    AccessLogOwner::Client(ctx.client_id.as_u32()),
490                    "Matrix",
491                    None,
492                    "injected_action",
493                    action.clone(),
494                    vec![],
495                )
496                .await;
497        }
498        txn += 1;
499        let (method, path, body) = actions::request(&action, &format!("{txn_prefix}.{txn}"))?;
500        let operation = action["type"].as_str().unwrap_or_default().to_string();
501        match exchange(
502            origin,
503            Some(&login.token),
504            method,
505            &path,
506            body.as_ref(),
507            IO_TIMEOUT,
508        )
509        .await
510        {
511            Ok(answer) => {
512                let ok = answer.status == 200;
513                let mut data = json!({"operation": operation, "status": answer.status});
514                if ok {
515                    data["result"] = match operation.as_str() {
516                        "matrix_messages" => json!({"events": answer.body["chunk"]}),
517                        _ => answer.body.clone(),
518                    };
519                } else {
520                    let (code, error) = matrix_error(&answer);
521                    data["result"] = json!({});
522                    data["errcode"] = json!(code);
523                    data["error"] = json!(error);
524                }
525                reply(
526                    &mut injected,
527                    ClientSendOutcome::Executed {
528                        detail: data.to_string(),
529                    },
530                );
531                // A successful send needs no answer from the handler; everything else does.
532                if ok && operation == "matrix_send" {
533                    ctx.state
534                        .record_access_log(
535                            AccessLogOwner::Client(ctx.client_id.as_u32()),
536                            "Matrix",
537                            None,
538                            "matrix_sent",
539                            data,
540                            vec![],
541                        )
542                        .await;
543                    continue;
544                }
545                events
546                    .try_send((Event::new(&actions::RESPONSE_EVENT, data), depth))
547                    .context("Matrix event queue full; consumer stalled")?;
548            }
549            Err(e) => {
550                log.warn(format!("Matrix request failed: {e:#}"));
551                reply(
552                    &mut injected,
553                    ClientSendOutcome::Rejected {
554                        error: format!("{e:#}"),
555                    },
556                );
557            }
558        }
559    }
560}
netget/client/matrix/mod.rs

netget/client/matrix/
mod.rs
