From 18fddd03d92ad9bbeb14d1f4a60944d440f2ed50 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Wed, 23 Sep 2026 21:04:00 -0700 Subject: [PATCH] gatewayd: mm, Mattermost's events and REST calls, typed Implemented-By: OpenCode session (model recorded in docs/implementer-log.md) --- crates/gatewayd/src/lib.rs | 1 + crates/gatewayd/src/mm/mod.rs | 208 ++++++++++++++++ crates/gatewayd/src/mm/rest.rs | 194 +++++++++++++++ crates/gatewayd/tests/mm_json.rs | 173 +++++++++++++ crates/gatewayd/tests/mm_rest.rs | 242 +++++++++++++++++++ crates/gatewayd/tests/support/http_server.rs | 93 +++++++ docs/implementer-log.md | 1 + 7 files changed, 912 insertions(+) create mode 100644 crates/gatewayd/src/mm/mod.rs create mode 100644 crates/gatewayd/src/mm/rest.rs create mode 100644 crates/gatewayd/tests/mm_json.rs create mode 100644 crates/gatewayd/tests/mm_rest.rs create mode 100644 crates/gatewayd/tests/support/http_server.rs diff --git a/crates/gatewayd/src/lib.rs b/crates/gatewayd/src/lib.rs index 8877e28..f74fc8a 100644 --- a/crates/gatewayd/src/lib.rs +++ b/crates/gatewayd/src/lib.rs @@ -3,6 +3,7 @@ pub mod config; pub mod http; +pub mod mm; pub mod net; pub mod secrets; pub mod ws; diff --git a/crates/gatewayd/src/mm/mod.rs b/crates/gatewayd/src/mm/mod.rs new file mode 100644 index 0000000..e59309c --- /dev/null +++ b/crates/gatewayd/src/mm/mod.rs @@ -0,0 +1,208 @@ +//! Mattermost's JSON, typed: the posts and users we read, the events that arrive over the +//! WebSocket, and the requests we send over it. Mattermost's JSON is not ours: unknown fields are +//! ignored, but every id we keep must be a valid id. + +pub mod rest; + +use std::time::Duration; + +use serde::Deserialize; + +use crate::config::valid_id; + +/// At most this many changed posts come back from one `posts?since` call (Mattermost v11.11.0, +/// `SqlPostStore::GetPostsSince`); a full answer may have left some out. +pub const SINCE_LIMIT: usize = 1000; + +#[derive(Debug)] +pub enum MmError { + /// No answer: connecting, TLS, or the HTTP exchange failed. + Net(String), + /// 401 or 403: the token is refused. + Auth(u16), + /// 429, still, after waiting as asked. + RateLimited(Duration), + /// Any other status that is not 2xx, with the start of the body. + Status(u16, String), + /// The answer is not the JSON we expect. + Json(String), +} + +impl std::fmt::Display for MmError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + MmError::Net(why) => write!(f, "{why}"), + MmError::Auth(status) => write!(f, "Mattermost refused the token ({status})"), + MmError::RateLimited(wait) => { + write!(f, "rate limited; asked to wait {} s", wait.as_secs()) + } + // Quoted: the body is the server's text, and must not forge a log line. + MmError::Status(status, body) => write!(f, "status {status}: {body:?}"), + MmError::Json(why) => write!(f, "unexpected JSON: {why}"), + } + } +} + +impl std::error::Error for MmError {} + +/// This bot, from `GET /users/me`. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +pub struct Me { + pub id: String, + pub username: String, +} + +/// The fields of a post that `gatewayd` uses. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +pub struct Post { + pub id: String, + pub user_id: String, + pub channel_id: String, + #[serde(default)] + pub root_id: String, + #[serde(default)] + pub message: String, + pub create_at: i64, + #[serde(default)] + pub delete_at: i64, + /// Empty for a message a user wrote; anything else is a system message. + #[serde(default, rename = "type")] + pub kind: String, +} + +impl Post { + /// Every id is a Mattermost id (the root may be empty): they end up in session ids and paths. + pub(crate) fn check(self) -> Result { + // id, user_id and channel_id must be `valid_id`, and root_id empty or `valid_id`; otherwise + // Json("a post with an invalid id: "). + if !valid_id(&self.id) || !valid_id(&self.user_id) || !valid_id(&self.channel_id) { + return Err(MmError::Json(format!( + "a post with an invalid id: {:?}", + self.id + ))); + } + if !self.root_id.is_empty() && !valid_id(&self.root_id) { + return Err(MmError::Json(format!( + "a post with an invalid id: {:?}", + self.root_id + ))); + } + Ok(self) + } +} + +/// An event from the WebSocket. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Event { + Hello, + /// A new post, and the type of its channel: `D` direct, `G` group, `O` open, `P` private. + Posted { + post: Post, + channel_type: String, + }, + /// Any other event, or a reply to one of our requests, by its name (empty for a reply). + Other(String), +} + +#[derive(Deserialize)] +struct RawEvent { + #[serde(default)] + event: String, + #[serde(default)] + data: serde_json::Map, +} + +pub(crate) fn json(bytes: &[u8]) -> Result { + // `serde_json::from_slice`; its error becomes Json(e.to_string()). + serde_json::from_slice(bytes).map_err(|e| MmError::Json(e.to_string())) +} + +/// One WebSocket text message as an event. `posted` carries the post as a JSON **string**. +pub fn parse_event(text: &str) -> Result { + // Parse into RawEvent. "hello" -> Hello. "posted" -> data.post must be a JSON *string*; parse + // that string as a Post and `check` it (a missing or non-string post is Json). channel_type is + // data.channel_type when it is a string, else "". Any other event name -> Other(name) (a reply + // has no event: Other("")). + let raw: RawEvent = json(text.as_bytes())?; + match raw.event.as_str() { + "hello" => Ok(Event::Hello), + "posted" => { + let post = raw + .data + .get("post") + .and_then(|p| p.as_str()) + .ok_or_else(|| { + MmError::Json("a posted event carries no post string".to_string()) + })?; + let post: Post = json(post.as_bytes())?; + let channel_type = raw + .data + .get("channel_type") + .and_then(|c| c.as_str()) + .unwrap_or("") + .to_string(); + Ok(Event::Posted { + post: post.check()?, + channel_type, + }) + } + other => Ok(Event::Other(other.to_string())), + } +} + +/// The WebSocket request that shows this bot as typing in a thread (`parent` is the root). +pub fn typing(seq: u64, channel: &str, parent: &str) -> String { + // serde_json::json!({"action": "user_typing", "seq": seq, "data": {"channel_id": channel, + // "parent_id": parent}}) as a string. + serde_json::json!({ + "action": "user_typing", + "seq": seq, + "data": {"channel_id": channel, "parent_id": parent} + }) + .to_string() +} + +/// What a `posts?since` call gave back. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Since { + /// Posts created after the time, not deleted, oldest first. + pub posts: Vec, + /// Mattermost's limit was reached: some posts may be missing. + pub full: bool, +} + +#[derive(Deserialize)] +struct PostList { + #[serde(default)] + order: Vec, + #[serde(default)] + posts: std::collections::HashMap, +} + +/// The body of `GET /channels/{id}/posts?since=`. Only ids in `order` changed after +/// `since`; `posts` also holds the roots of their threads, which may be older. Edited and deleted +/// posts come back too: only new posts count. +pub fn since_list(body: &[u8], since: i64) -> Result { + // Parse a PostList. full = order.len() >= SINCE_LIMIT. For each id in `order` (never the keys + // of `posts`), skip it if it is not in `posts`; keep the post if create_at > since and + // delete_at == 0, after `check`. Sort by (create_at, id), remove repeated ids. + let list: PostList = json(body)?; + let full = list.order.len() >= SINCE_LIMIT; + let mut seen = std::collections::HashSet::new(); + let mut posts = Vec::new(); + for id in &list.order { + let post = match list.posts.get(id) { + Some(p) => p.clone(), + None => continue, + }; + let post = post.check()?; + if post.create_at <= since || post.delete_at != 0 { + continue; + } + if seen.insert(post.id.clone()) { + posts.push(post); + } + } + posts.sort_by(|a, b| a.create_at.cmp(&b.create_at).then_with(|| a.id.cmp(&b.id))); + Ok(Since { posts, full }) +} diff --git a/crates/gatewayd/src/mm/rest.rs b/crates/gatewayd/src/mm/rest.rs new file mode 100644 index 0000000..08261b3 --- /dev/null +++ b/crates/gatewayd/src/mm/rest.rs @@ -0,0 +1,194 @@ +//! Mattermost's REST calls, one connection per request, with the token in the `Authorization` +//! header. A 429 waits as asked; a 5xx is tried twice more; a 401 or 403 is `Auth` at once. + +use std::time::{Duration, SystemTime}; + +use serde::Deserialize; + +use crate::config::valid_id; +use crate::http::{self, Response, rate_limit_wait}; +use crate::mm::{Me, MmError, Post, Since, json, since_list}; +use crate::net::Connector; +use crate::secrets::Secret; +use crate::ws::conn::host_header; + +/// The pause before trying a call again after a 5xx. +pub const RETRY_5XX: Duration = Duration::from_millis(500); +/// How many times a call is tried again after a 5xx, and waits after a 429. +pub const RETRIES: u32 = 2; +/// How much of an error body is kept for the message. +const BODY_KEPT: usize = 200; + +pub struct Client { + connector: Connector, + token: Secret, + timeout: Duration, +} + +#[derive(Deserialize)] +struct Channel { + id: String, +} + +impl Client { + /// `timeout` bounds connecting and each read: no bytes for that long is an error. + pub fn new(connector: Connector, token: Secret, timeout: Duration) -> Client { + // Store the three. + Client { + connector, + token, + timeout, + } + } + + pub fn connector(&self) -> &Connector { + // The Connector. + &self.connector + } + + pub fn token(&self) -> &Secret { + // The Secret. + &self.token + } + + fn once(&self, method: &str, path: &str, body: Option<&[u8]>) -> Result { + // 1. `self.connector.connect(self.timeout)`, then `set_read_timeout(Some(self.timeout))`. + let mut stream = self + .connector + .connect(self.timeout) + .map_err(|e| MmError::Net(format!("{method} {path}: {e}")))?; + stream + .set_read_timeout(Some(self.timeout)) + .map_err(|e| MmError::Net(format!("{method} {path}: {e}")))?; + // 2. Headers: Authorization "Bearer " (the only use of `expose`), Accept + // "application/json", and Content-Type "application/json" when there is a body. Host is + // `host_header(self.connector.server())`. + let auth = format!("Bearer {}", self.token.expose()); + let mut headers: Vec<(&str, &str)> = vec![ + ("Authorization", auth.as_str()), + ("Accept", "application/json"), + ]; + if body.is_some() { + headers.push(("Content-Type", "application/json")); + } + // 3. `http::request`. Every error on the way is Net(" : "). + http::request( + &mut stream, + method, + &host_header(self.connector.server()), + path, + &headers, + body, + ) + .map_err(|e| MmError::Net(format!("{method} {path}: {e}"))) + } + + /// One call, tried again as the module comment says; the body of a 2xx answer. + fn call(&self, method: &str, path: &str, body: Option<&[u8]>) -> Result, MmError> { + // Loop over `once`: 2xx -> the body. 401 or 403 -> Auth(status) at once. 429 -> after + // RETRIES waits already, RateLimited(wait); else sleep `rate_limit_wait(&head, + // SystemTime::now())` and try again. 5xx -> up to RETRIES more tries, sleeping RETRY_5XX + // before each. Anything else -> Status(status, the first BODY_KEPT characters of the body, + // lossy UTF-8). + let mut tries: u32 = 0; + loop { + let answer = self.once(method, path, body)?; + let status = answer.head.status; + if (200..300).contains(&status) { + return Ok(answer.body); + } + match status { + 401 | 403 => return Err(MmError::Auth(status)), + 429 => { + let wait = rate_limit_wait(&answer.head, SystemTime::now()); + if tries < RETRIES { + tries += 1; + std::thread::sleep(wait); + continue; + } + return Err(MmError::RateLimited(wait)); + } + _ if status >= 500 => { + if tries < RETRIES { + tries += 1; + std::thread::sleep(RETRY_5XX); + continue; + } + return Err(status_error(status, &answer.body)); + } + _ => return Err(status_error(status, &answer.body)), + } + } + } + + /// `GET /api/v4/users/me`: who this token is. + pub fn me(&self) -> Result { + // GET /api/v4/users/me into Me; an id that is not `valid_id` or an empty username is Json. + let body = self.call("GET", "/api/v4/users/me", None)?; + let me: Me = json(&body)?; + if !valid_id(&me.id) || me.username.is_empty() { + return Err(MmError::Json( + "the user me has no id or no username".to_string(), + )); + } + Ok(me) + } + + /// `POST /api/v4/posts`: a post in `channel`, in the thread of `root` (empty: top level). + pub fn create_post(&self, channel: &str, root: &str, message: &str) -> Result { + // POST /api/v4/posts with {"channel_id", "root_id", "message"}; the answer is a Post, + // `check`ed. + let payload = serde_json::json!({ + "channel_id": channel, + "root_id": root, + "message": message + }) + .to_string(); + let body = self.call("POST", "/api/v4/posts", Some(payload.as_bytes()))?; + let post: Post = json(&body)?; + post.check() + } + + /// `GET /api/v4/channels/{channel}/posts?since=`: the posts created after `since`. + pub fn posts_since(&self, channel: &str, since: i64) -> Result { + // A channel that is not `valid_id` is Json, and nothing is sent. Otherwise GET + // /api/v4/channels//posts?since=, through `since_list`. + if !valid_id(channel) { + return Err(MmError::Json( + "a channel path must be a Mattermost id".to_string(), + )); + } + let body = self.call( + "GET", + &format!("/api/v4/channels/{channel}/posts?since={since}"), + None, + )?; + since_list(&body, since) + } + + /// `POST /api/v4/channels/direct`: the id of the direct channel between two users. + pub fn direct_channel(&self, a: &str, b: &str) -> Result { + // POST /api/v4/channels/direct with the JSON array [a, b]; the answer has an "id", which + // must be `valid_id` (else Json). + let payload = serde_json::to_vec(&serde_json::json!([a, b])) + .map_err(|e| MmError::Json(e.to_string()))?; + let body = self.call("POST", "/api/v4/channels/direct", Some(payload.as_slice()))?; + let channel: Channel = json(&body)?; + if !valid_id(&channel.id) { + return Err(MmError::Json("the direct channel has no id".to_string())); + } + Ok(channel.id) + } +} + +fn status_error(status: u16, body: &[u8]) -> MmError { + // Any non-2xx, non-4xx-that-isn't-401/403/429/5xx: the status and the first `BODY_KEPT` + // characters, lossy UTF-8. + MmError::Status( + status, + String::from_utf8_lossy(body) + .chars() + .take(BODY_KEPT) + .collect(), + ) +} diff --git a/crates/gatewayd/tests/mm_json.rs b/crates/gatewayd/tests/mm_json.rs new file mode 100644 index 0000000..2857859 --- /dev/null +++ b/crates/gatewayd/tests/mm_json.rs @@ -0,0 +1,173 @@ +//! Mattermost's JSON: events as the server sends them (v11.11.0 shapes), the typing request, and +//! the `posts?since` list, whose edited, deleted and root-only posts must not be answered (M4a spec, +//! sections 7 and 9). Do not edit. + +use gatewayd::mm::{Event, MmError, SINCE_LIMIT, parse_event, since_list, typing}; +use serde_json::json; + +const KYLE: &str = "k0000000000000000000000000"; +const DM: &str = "d0000000000000000000000000"; + +fn post(id: &str, create_at: i64) -> serde_json::Value { + json!({ + "id": id, "create_at": create_at, "update_at": create_at, "edit_at": 0, "delete_at": 0, + "is_pinned": false, "user_id": KYLE, "channel_id": DM, "root_id": "", "original_id": "", + "message": "hello", "type": "", "props": {"from_bot": "true"}, "hashtags": "", + "pending_post_id": "", "reply_count": 0, "metadata": {} + }) +} + +fn id(n: u32) -> String { + format!("p{n:025}") +} + +fn posted(post: &serde_json::Value, channel_type: &str) -> String { + json!({ + "event": "posted", + "data": { + "channel_display_name": "@kyle", "channel_name": "x__y", "channel_type": channel_type, + "post": post.to_string(), "sender_name": "@kyle", "set_online": true, "team_id": "" + }, + "broadcast": {"omit_users": null, "user_id": "", "channel_id": DM, "team_id": ""}, + "seq": 3 + }) + .to_string() +} + +#[test] +fn hello_replies_and_other_events() { + let hello = + json!({"event": "hello", "data": {"server_version": "11.11.0"}, "broadcast": {}, "seq": 0}); + assert_eq!(parse_event(&hello.to_string()).unwrap(), Event::Hello); + assert_eq!( + parse_event(r#"{"status":"OK","seq_reply":1}"#).unwrap(), + Event::Other(String::new()) + ); + let typing = json!({"event": "typing", "data": {"parent_id": ""}, "broadcast": {}, "seq": 4}); + assert_eq!( + parse_event(&typing.to_string()).unwrap(), + Event::Other("typing".to_string()) + ); +} + +#[test] +fn a_posted_event_carries_the_post_as_a_string() { + let mut p = post(&id(1), 1_758_650_000_123); + p["root_id"] = json!(id(0)); + p["message"] = json!("line one\nline two"); + let Event::Posted { post, channel_type } = parse_event(&posted(&p, "D")).unwrap() else { + panic!("not a post") + }; + assert_eq!(channel_type, "D"); + assert_eq!( + (post.id.as_str(), post.root_id.as_str()), + (id(1).as_str(), id(0).as_str()) + ); + assert_eq!( + (post.user_id.as_str(), post.channel_id.as_str()), + (KYLE, DM) + ); + assert_eq!( + (post.message.as_str(), post.create_at, post.kind.as_str()), + ("line one\nline two", 1_758_650_000_123, "") + ); + let mut s = self::post(&id(2), 5); + s["type"] = json!("system_join_channel"); + let Event::Posted { post, .. } = parse_event(&posted(&s, "O")).unwrap() else { + panic!() + }; + assert_eq!(post.kind, "system_join_channel"); +} + +#[test] +fn bad_posts_are_errors_not_panics() { + let p = post(&id(1), 5); + let as_object = + json!({"event": "posted", "data": {"post": p, "channel_type": "D"}}).to_string(); + let mut cases = vec![ + as_object, + "not json".to_string(), + "[1,2]".to_string(), + "{\"event\":5}".to_string(), + ]; + for (field, value) in [ + ("id", json!("../../etc")), + ("user_id", json!("")), + ("channel_id", json!("A0000000000000000000000000")), + ("root_id", json!("short")), + ("create_at", json!("soon")), + ] { + let mut bad = post(&id(1), 5); + bad[field] = value; + cases.push(posted(&bad, "D")); + } + let mut missing = post(&id(1), 5); + missing.as_object_mut().unwrap().remove("create_at"); + cases.push(posted(&missing, "D")); + for case in cases { + assert!( + matches!(parse_event(&case), Err(MmError::Json(_))), + "{case}" + ); + } +} + +#[test] +fn the_typing_request() { + let got: serde_json::Value = serde_json::from_str(&typing(7, DM, &id(0))).unwrap(); + assert_eq!( + got, + json!({"action": "user_typing", "seq": 7, "data": {"channel_id": DM, "parent_id": id(0)}}) + ); +} + +#[test] +fn since_keeps_new_posts_in_order_and_nothing_else() { + let since = 1000; + let mut edited = post(&id(1), 900); + edited["update_at"] = json!(1500); + let mut deleted = post(&id(2), 1100); + deleted["delete_at"] = json!(1200); + let root_only = post(&id(3), 10); + let later = post(&id(4), 1300); + let earlier = post(&id(5), 1200); + let at_since = post(&id(6), 1000); + let mut posts = serde_json::Map::new(); + for p in [&edited, &deleted, &root_only, &later, &earlier, &at_since] { + posts.insert(p["id"].as_str().unwrap().to_string(), p.clone()); + } + let order = json!([id(4), id(2), id(1), id(5), id(9), id(6)]); + let body = json!({"order": order, "posts": posts, "next_post_id": "", "prev_post_id": "", "has_next": false}); + let got = since_list(body.to_string().as_bytes(), since).unwrap(); + let ids: Vec<&str> = got.posts.iter().map(|p| p.id.as_str()).collect(); + assert_eq!(ids, [id(5), id(4)]); + assert!(!got.full); +} + +#[test] +fn a_full_since_answer_says_so() { + let mut posts = serde_json::Map::new(); + let mut order = Vec::new(); + for n in 0..SINCE_LIMIT { + let n = u32::try_from(n).unwrap(); + posts.insert(id(n), post(&id(n), 2000 + i64::from(n))); + order.push(id(n)); + } + let body = json!({"order": order, "posts": posts}); + let got = since_list(body.to_string().as_bytes(), 1000).unwrap(); + assert!(got.full); + assert_eq!(got.posts.len(), SINCE_LIMIT); + let empty = since_list(br#"{"order":[],"posts":{}}"#, 0).unwrap(); + assert!(empty.posts.is_empty() && !empty.full); +} + +#[test] +fn a_server_error_body_is_quoted_in_messages() { + let e = MmError::Status(500, "boom\ngatewayd: a forged line".to_string()); + let text = e.to_string(); + assert!(!text.contains('\n'), "{text}"); + assert_eq!( + MmError::Auth(401).to_string(), + "Mattermost refused the token (401)" + ); +} diff --git a/crates/gatewayd/tests/mm_rest.rs b/crates/gatewayd/tests/mm_rest.rs new file mode 100644 index 0000000..e2a937b --- /dev/null +++ b/crates/gatewayd/tests/mm_rest.rs @@ -0,0 +1,242 @@ +//! Mattermost's REST calls against a scripted server: the requests on the wire, and what a 401, +//! 403, 429, 5xx or 404 does (M4a spec, sections 5 and 9). Do not edit. + +#[path = "support/http_server.rs"] +mod http_server; +#[path = "support/tls_server.rs"] +mod tls_server; + +use std::net::SocketAddr; +use std::sync::mpsc::Receiver; +use std::time::{Duration, Instant}; + +use gatewayd::config::ServerUrl; +use gatewayd::mm::MmError; +use gatewayd::mm::rest::Client; +use gatewayd::net::Connector; +use gatewayd::secrets::Secret; +use http_server::{Request, reply, serve_http}; +use serde_json::json; +use tls_server::{fixture, server_config}; + +const BOT: &str = "b0000000000000000000000000"; +const KYLE: &str = "k0000000000000000000000000"; +const DM: &str = "d0000000000000000000000000"; +const POST: &str = "p0000000000000000000000001"; + +fn client(addr: SocketAddr, tls: bool) -> Client { + let url = ServerUrl { + tls, + host: "localhost".to_string(), + port: addr.port(), + }; + let ca = tls.then(|| fixture("test-ca.pem")); + let connector = Connector::new(url, ca.as_deref()).unwrap(); + Client::new( + connector, + Secret::new("TOKEN".to_string()), + Duration::from_secs(5), + ) +} + +fn me_body() -> String { + json!({"id": BOT, "username": "boxmaker-straylight", "roles": "system_user", "is_bot": true}) + .to_string() +} + +fn requests(rx: &Receiver) -> Vec { + rx.try_iter().collect() +} + +#[test] +fn me_plain_and_over_tls() { + for tls in [false, true] { + let config = tls.then(|| server_config("server")); + let (addr, rx) = serve_http(config, |_, _| reply(200, "", &me_body())); + let me = client(addr, tls).me().unwrap(); + assert_eq!( + (me.id.as_str(), me.username.as_str()), + (BOT, "boxmaker-straylight") + ); + let r = rx.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!( + (r.method.as_str(), r.path.as_str()), + ("GET", "/api/v4/users/me") + ); + assert!( + r.head.contains("Authorization: Bearer TOKEN\r\n"), + "{}", + r.head + ); + assert!( + r.head + .contains(&format!("Host: localhost:{}\r\n", addr.port())), + "{}", + r.head + ); + } +} + +#[test] +fn create_post_sends_the_thread_and_reads_the_post_back() { + let (addr, rx) = serve_http(None, |_, r| { + let mut p = r.json(); + p["id"] = json!(POST); + p["user_id"] = json!(BOT); + p["create_at"] = json!(5); + reply(201, "", &p.to_string()) + }); + let post = client(addr, false) + .create_post(DM, KYLE, "an answer\nin two lines") + .unwrap(); + assert_eq!((post.id.as_str(), post.root_id.as_str()), (POST, KYLE)); + let r = rx.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!( + (r.method.as_str(), r.path.as_str()), + ("POST", "/api/v4/posts") + ); + assert!( + r.head.contains("Content-Type: application/json\r\n"), + "{}", + r.head + ); + assert_eq!( + r.json(), + json!({"channel_id": DM, "root_id": KYLE, "message": "an answer\nin two lines"}) + ); +} + +#[test] +fn posts_since_and_the_direct_channel() { + let (addr, rx) = serve_http(None, |_, r| { + if r.path.contains("/posts?since=") { + reply(200, "", r#"{"order":[],"posts":{}}"#) + } else { + reply(201, "", &json!({"id": DM, "type": "D"}).to_string()) + } + }); + let c = client(addr, false); + assert!( + c.posts_since(DM, 1_758_650_000_000) + .unwrap() + .posts + .is_empty() + ); + assert_eq!(c.direct_channel(BOT, KYLE).unwrap(), DM); + let rs = requests(&rx); + assert_eq!( + rs[0].path, + format!("/api/v4/channels/{DM}/posts?since=1758650000000") + ); + assert_eq!( + (rs[1].method.as_str(), rs[1].path.as_str()), + ("POST", "/api/v4/channels/direct") + ); + assert_eq!(rs[1].json(), json!([BOT, KYLE])); +} + +#[test] +fn a_channel_that_is_not_an_id_is_never_sent() { + let (addr, rx) = serve_http(None, |_, _| reply(200, "", r#"{"order":[],"posts":{}}"#)); + let err = client(addr, false) + .posts_since("../users/me?x=", 0) + .unwrap_err(); + assert!(matches!(err, MmError::Json(_)), "{err}"); + std::thread::sleep(Duration::from_millis(100)); + assert!(requests(&rx).is_empty()); +} + +#[test] +fn a_refused_token_stops_at_once() { + for status in [401, 403] { + let (addr, rx) = serve_http(None, move |_, _| { + reply( + status, + "", + r#"{"id":"api.context.session_expired.app_error"}"#, + ) + }); + let err = client(addr, false).me().unwrap_err(); + assert!(matches!(err, MmError::Auth(s) if s == status), "{err}"); + std::thread::sleep(Duration::from_millis(100)); + assert_eq!(requests(&rx).len(), 1, "no retry on {status}"); + } +} + +#[test] +fn a_server_error_is_tried_twice_more() { + let (addr, rx) = serve_http(None, |n, _| { + if n < 2 { + reply(503, "", "{}") + } else { + reply(200, "", &me_body()) + } + }); + assert!(client(addr, false).me().is_ok()); + assert_eq!(requests(&rx).len(), 3); + + let (addr, rx) = serve_http(None, |_, _| reply(502, "", "{}")); + let err = client(addr, false).me().unwrap_err(); + assert!(matches!(err, MmError::Status(502, _)), "{err}"); + std::thread::sleep(Duration::from_millis(100)); + assert_eq!(requests(&rx).len(), 3); +} + +#[test] +fn a_rate_limit_is_waited_out() { + let (addr, rx) = serve_http(None, |n, _| { + if n == 0 { + reply(429, "X-Ratelimit-Reset: 1\r\n", "{}") + } else { + reply(200, "", &me_body()) + } + }); + let started = Instant::now(); + assert!(client(addr, false).me().is_ok()); + assert!( + started.elapsed() >= Duration::from_millis(900), + "{:?}", + started.elapsed() + ); + assert_eq!(requests(&rx).len(), 2); + + let (addr, _rx) = serve_http(None, |_, _| reply(429, "X-Ratelimit-Reset: 1\r\n", "{}")); + assert!(matches!( + client(addr, false).me(), + Err(MmError::RateLimited(_)) + )); +} + +#[test] +fn other_statuses_keep_the_start_of_the_body() { + let long = format!("{{\"message\":\"{}\"}}", "x".repeat(1000)); + let (addr, _rx) = serve_http(None, move |_, _| reply(404, "", &long)); + let Err(MmError::Status(404, body)) = client(addr, false).me() else { + panic!("not a 404") + }; + assert_eq!(body.chars().count(), 200); +} + +#[test] +fn answers_that_are_not_what_we_expect() { + for body in [ + "not json", + r#"{"id":"short","username":"x"}"#, + r#"{"id":"b0000000000000000000000000","username":""}"#, + ] { + let (addr, _rx) = serve_http(None, move |_, _| reply(200, "", body)); + assert!( + matches!(client(addr, false).me(), Err(MmError::Json(_))), + "{body}" + ); + } +} + +#[test] +fn nobody_listening_is_a_network_error() { + let port = std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap(); + assert!(matches!(client(port, false).me(), Err(MmError::Net(_)))); +} diff --git a/crates/gatewayd/tests/support/http_server.rs b/crates/gatewayd/tests/support/http_server.rs new file mode 100644 index 0000000..8dc803a --- /dev/null +++ b/crates/gatewayd/tests/support/http_server.rs @@ -0,0 +1,93 @@ +//! A scripted HTTP server for tests: it records each request and answers with what the test's +//! function returns for it. Built on `tls_server`. Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::io::{Read, Write}; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; + +use rustls::ServerConfig; + +use crate::tls_server::serve; + +#[derive(Debug, Clone)] +pub struct Request { + pub method: String, + pub path: String, + /// The whole head, for tests that look for a header. + pub head: String, + pub body: Vec, +} + +impl Request { + pub fn json(&self) -> serde_json::Value { + serde_json::from_slice(&self.body).unwrap() + } +} + +/// A response with a JSON body. +pub fn reply(status: u16, extra_headers: &str, body: &str) -> String { + format!( + "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\n{extra_headers}Content-Length: {}\r\n\r\n{body}", + body.len() + ) +} + +fn read_request(conn: &mut dyn Read) -> Option { + let mut head = Vec::new(); + let mut byte = [0u8; 1]; + while !head.ends_with(b"\r\n\r\n") { + if conn.read(&mut byte).ok()? == 0 { + return None; + } + head.push(byte[0]); + } + let head = String::from_utf8(head).ok()?; + let mut first = head.split_whitespace(); + let method = first.next()?.to_string(); + let path = first.next()?.to_string(); + let length = head + .lines() + .find_map(|l| { + l.to_ascii_lowercase() + .strip_prefix("content-length:") + .map(|v| v.trim().parse().ok()) + }) + .flatten() + .unwrap_or(0); + let mut body = vec![0u8; length]; + conn.read_exact(&mut body).ok()?; + Some(Request { + method, + path, + head, + body, + }) +} + +/// Serve requests: `answer(n, &request)` gives the response to the n-th request (from 0). Every +/// request is sent on the returned channel. +pub fn serve_http( + tls: Option>, + answer: F, +) -> (SocketAddr, mpsc::Receiver) +where + F: Fn(usize, &Request) -> String + Send + Sync + 'static, +{ + let (tx, rx) = mpsc::channel(); + let tx = Mutex::new(tx); + let count = AtomicUsize::new(0); + let addr = serve(tls, move |mut conn| { + let Some(request) = read_request(&mut conn) else { + return; + }; + let n = count.fetch_add(1, Ordering::SeqCst); + let response = answer(n, &request); + let _ = tx.lock().unwrap().send(request); + let _ = conn.write_all(response.as_bytes()); + let _ = conn.flush(); + }); + (addr, rx) +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 140de28..d792ac2 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone. | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| M4a/10-gatewayd-mm | 2026-09-23 | done | 1 | pass | none | Copied `tests/mm_json.rs`, `tests/mm_rest.rs` and `tests/support/http_server.rs`, added the `src/mm/mod.rs` and `src/mm/rest.rs` skeletons to `crates/gatewayd/src/mm/` and `pub mod mm;` to `lib.rs` (before `secrets`, alphabetical). Filled `mod.rs`: `Post::check` requires `id`/`user_id`/`channel_id` `valid_id` and `root_id` empty-or-`valid_id`, else `Json` quoting the offending id with `{:?}`; `json` is `serde_json::from_slice` mapped to `Json(e.to_string())`; `parse_event` matches `hello`/`posted`/other — `posted` takes `data.post` as a JSON *string* (an object or missing is `Json`), parses it, `check`s it, and reads `data.channel_type` (else ""), any other name (or an empty-name reply) is `Other(name)`; `typing` is `serde_json::json!` compacted; `since_list` walks `order` only (skipping ids not in `posts`, keeping `create_at > since && delete_at == 0` after `check`, deduping, then sorting by `(create_at, id)`), `full` when `order.len() >= SINCE_LIMIT`. Filled `rest.rs`: `Client::new` stores the three fields; `once` connects within `timeout`, sets the read timeout, sends `Authorization: Bearer ` (the only `expose`), `Accept`/`Content-Type` headers and `host_header`, mapping every error to `Net(" : ")`; `call` loops `once` — 2xx returns the body, 401/403 `Auth`, 429 waits `rate_limit_wait` up to `RETRIES` then `RateLimited`, 5xx retried up to `RETRIES` times sleeping `RETRY_5XX`, else `Status` with the first `BODY_KEPT` lossy-UTF-8 chars via a `status_error` helper; `me`/`create_post`/`posts_since`/`direct_channel` build the four calls, `posts_since` and `me`/`direct_channel` reject non-`valid_id` ids as `Json` before sending. All 7 `mm_json` and 10 `mm_rest` tests pass (the latter ~3 s on two deliberate rate-limit waits); `make gate` prints `gate: ok` first run. | ? | | M4a/09-gatewayd-ws-conn | 2026-09-23 | done | 1 | pass | none | Filled the eight functions in the copied `crates/gatewayd/src/ws/conn.rs` skeleton (the written `poll` was the glue). `open`: `connector.connect(dead_after)` mapped to `Handshake(e.to_string())`, then `handshake` with `host_header(connector.server())`, a Ws with a new `Decoder` and `last_heard`/`last_ping` both `now`. `send`: `read_exact` 4 mask bytes from `random`, then `encode(opcode, payload, mask)` written and flushed. `send_text`: `send(TEXT, text.as_bytes())`. `take_messages`: loop `next_message`, `Text` returns, `Ping` answered with `send(PONG, &payload)`, `Pong` ignored, `Close` replies the code as 2 big-endian bytes (empty when none) via a best-effort `send(CLOSE, ...)` (the peer may be gone) and returns `Closed`. `keep_alive`: `now.duration_since(last_heard) >= dead_after` is `Dead`, else `now.duration_since(last_ping) >= ping_every` pings and stamps `last_ping`. `read_timeout`: the least of next-ping, next-dead and until-left (each `saturating_duration_since`), then `.max(1ms)`. `read_some`: `set_read_timeout`, a 16 KiB buffer, `Ok(0)` -> `Closed`, `Ok(n)` feeds `buf.get(..n).unwrap_or_default()` and stamps `last_heard`, `WouldBlock`/`TimedOut`/`Interrupted` -> `Ok(())`, any other `Err` -> `Io`. `close`: best-effort `send(CLOSE, &1000u16.to_be_bytes())`. `host_header`: host alone when the port is the scheme default (443 for tls, 80 otherwise) else `host:port`. All 10 tests in `tests/ws_conn.rs` pass five runs under a second; `make gate` prints `gate: ok` first run. | ? | | M4a/08-gatewayd-ws-frames | 2026-09-23 | done | 1 | pass | none | Copied `tests/ws_frame.rs` and the `src/ws/frame.rs` skeleton, added `pub mod frame;` (before `handshake`, alphabetical). Filled the seven functions the comments specified verbatim: `check_first_bytes` (reserved bits `b0 & 0x70`, mask `b1 & 0x80`, opcode `matches!(b0 & 0x0F, CONTINUATION | TEXT | CLOSE | PING | PONG)`); `check_control` (not fin, then >125); `length` (match on `short`: 0..=125, 126 reading `buf.get(2..4)` into `u16::from_be_bytes` with `len < 126` refused, 127 reading `buf.get(2..10)` into `u64::from_be_bytes` with top-bit and `<= 0xFFFF` refused, `usize::try_from(len).unwrap_or(usize::MAX)`); `check_data` (TEXT while partial, CONTINUATION while none, `payload_len > MAX_MESSAGE.saturating_sub(so_far)`); `data_frame` (empty Vec for TEXT else `partial.take().unwrap_or_default()`, append, defer when not fin, `String::from_utf8` at fin); `close` (slice-pattern match, `u16::from_be_bytes` code, UTF-8 reason); `encode` (FIN+opcode, three length branches with mask bit, XOR with `mask.iter().cycle()`). All string literals got `.to_string()` for the `Protocol(String)` variant, matching http.rs. `length`'s match needed a defensive `_` arm (`128..=u8::MAX` unreachable since `short = b1 & 0x7F`) so the codec stays exhaustive without a panic. All 7 tests in `tests/ws_frame.rs` pass including the 300-seed property test against the naive decoder; `make gate` prints `gate: ok` first run. | ? | | M4a/07-gatewayd-ws-handshake | 2026-09-23 | done | 1 | pass | none | Filled the copied `crates/gatewayd/src/ws/handshake.rs` skeleton. `base64`: 3-byte chunks to 4 chars over ALPHABET with `=` padding, reading each byte via `first`/`get(..).copied().unwrap_or(0)` (no indexing) and masking to 0..=63 before the alphabet index; `accept_for`: `base64(sha1(key + GUID))` building `key+GUID` into one Vec; `new_key`: `read_exact` 16 bytes (too few is an io error) then base64; `check_response` in the task's order — status 101 (`"status "`), `Upgrade` == `websocket` (ASCII case-insensitive), `Connection` with a comma-split token == `upgrade` (case-insensitive), then `Sec-WebSocket-Accept` exactly == `accept_for(key)`; `handshake`: `new_key`, write `request_text` + flush, `read_head` (mapped to `Handshake`), `check_response`, reading nothing past the head. All 7 tests in `tests/ws_handshake.rs` pass including the RFC 6455 accept vector and the first-frame-left-unread handshake; `make gate` prints `gate: ok` first run. | ? |