diff --git a/docs/plans/M5c1/03-gatewayd-speaker.md b/docs/plans/M5c1/03-gatewayd-speaker.md new file mode 100644 index 0000000..12d4164 --- /dev/null +++ b/docs/plans/M5c1/03-gatewayd-speaker.md @@ -0,0 +1,613 @@ +# M5c-1 task 03: `gatewayd` sets a turn's speaker + +**Branch:** `m5c1` (run `git switch m5c1`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `gatewayd: a turn's speaker, the username of the one person who wrote its burst` +**Log row:** the Task cell is `M5c1/03-gatewayd-speaker`. + +## Goal + +Task 01 gave `Turn` a field `speaker: Option`, and task 02 made `loopd` log it for +`memoryd`, which will send a turn to Honcho only when it has a speaker (M5c spec, sections 4 and +11). `gatewayd` still sends `speaker: None` on every turn. This task sets it. + +`gatewayd` joins the messages that arrive while a turn runs into one turn, a **burst**: their +texts joined with a blank line. The speaker is the **Mattermost username** of the person who +wrote the turn, and only when one person wrote all of it: + +| The turn's messages | `Turn.speaker` | +|---|---| +| one message, from Kyle (username `kyle`) | `Some("kyle")` | +| a burst, every message from Kyle | `Some("kyle")` | +| a burst from Eve and Kyle | `None`: two people | +| Kyle's messages, but Mattermost does not answer the username lookup | `None`, and a log line | + +`None` is the safe side: a turn without a speaker is never sent to Honcho (the owner's rule: no +speaker, or incomplete speaker information, is not synced). **Never** put anything else there: +not the user id, not the first message's author, not a name from an earlier turn. + +1. **Each message knows its author.** `sessions::Message` gains `user`, the post's `user_id`, + set by `Router::route`. +2. **Each batch knows the author of each of its messages.** `Batch` gains `senders`, one user + id per message, in order; `Queues` keeps them beside the waiting texts. `Batch::sender` gives + the one user who wrote them all, or `None`. +3. **The username.** `gatewayd` already looks usernames up for approvals: `Client::username`, + `GET /api/v4/users/{user}`. The event loop now keeps each name it found in a map, so a user is + looked up once; a failure is logged and not kept. +4. **The turn.** `deliver` gets the speaker as a new argument and puts it on the `Turn`. + **The turn's text does not change**: the join stays exactly as it is, and the speaker never + goes into the text. + +## Every place that changes + +This is the complete list. + +| Place | What happens there | +|---|---| +| `crates/gatewayd/src/sessions.rs`, `Message` and `Router::route` | new field `user`, set on both `Route::Queue` paths (edits S1 to S3) | +| `crates/gatewayd/src/sessions.rs`, `Batch`, `Pending`, `Queues` | new `senders`, `Batch::sender` (edits S4 to S8) | +| `crates/gatewayd/src/deliver.rs`, `one_turn`, `run_turn`, `deliver` | a new parameter `speaker: Option<&str>`, put on the `Turn` (edits D1 to D5) | +| `crates/gatewayd/src/serve/mod.rs`, `Gateway` | new field `usernames` (edits G1, G2) | +| `crates/gatewayd/src/serve/handle.rs`, `start` | new method `speaker`; passes it to `deliver` (edits H1, H2) | +| `crates/gatewayd/src/serve/approvals.rs` | **no change**: it keeps its own lookups | +| `crates/gatewayd/tests/sessions_mention.rs` | reads `text` only; no change | +| tests that build a `Batch` or a `Message`, or call `deliver` | new versions are given | + +## Files + +- Copy, from `docs/plans/M5c1/files/` (the same path under it): + - new: `crates/gatewayd/tests/deliver_speaker.rs`, `crates/gatewayd/tests/serve_speaker.rs` + - **new versions** of (they replace the files in the tree): `crates/gatewayd/tests/sessions.rs`, + `deliver.rs`, `deliver_links.rs`, `deliver_notice.rs` and + `crates/gatewayd/tests/support/fake_mm.rs` (it gains `refuse_users`). +- Modify: `crates/gatewayd/src/sessions.rs`, `crates/gatewayd/src/deliver.rs`, + `crates/gatewayd/src/serve/mod.rs`, `crates/gatewayd/src/serve/handle.rs`, + `docs/implementer-log.md`. +- Do not touch: any other source file (`serve/approvals.rs` and `mm/rest.rs` in particular), any + file of another crate, any `Cargo.toml`. + +## Facts you need + +- **Where the author is known.** A Mattermost post carries `user_id`, a 26-character id, and + `Router::route` sees the whole `Post`. `route` makes a `Route::Queue(Message { .. })` in two + places: for a message that starts with `!!`, and for any other message. Both get `user`. +- **Where the texts are joined.** `Queues::push` starts a turn at once for an idle session: the + batch holds that one message, so its `senders` is that message's author alone. While a turn + runs, each new message waits in `Pending::waiting`; its author goes into `Pending::senders`, + at the same place. A message refused by a full queue is in neither. `Queues::finish` makes the + next batch from all that waited, and `std::mem::take` leaves `senders` empty for the batch + after it, as `waiting.clear()` does for the texts. +- **The username.** `Client::username` (in `mm/rest.rs`, unchanged) sends + `GET /api/v4/users/{user}` and checks what comes back: the id must be a Mattermost id and the + username 1 to 64 characters of `a-z`, `0-9`, `.`, `-` and `_`; anything else is an error. Pass + the name on **as it comes**: `memoryd` turns it into a Honcho peer name in a later task (a `.` + becomes `_` there, not here). +- **The map.** `Gateway` belongs to the event loop's thread, and `start` runs on it, so a plain + `HashMap` needs no lock. The lookup is a REST call on that thread before the turn's own thread + starts, as the approvals' `place` does; with the map it is made once per user. A failed lookup + is not stored, so the next turn of that user looks again. +- **The log line** for a failed lookup is exactly this, and holds no text of the message: + + ```text + gatewayd: no username for : ; the turn has no speaker + ``` + + With the fake Mattermost refusing the lookup, the test expects this one line (`MmError`'s + `Display` quotes the body): + + ```text + gatewayd: no username for k0000000000000000000000000: status 404: "{\"message\":\"not found\"}"; the turn has no speaker + ``` +- **The fake Mattermost** answers `GET /api/v4/users/` with the username `kyle` for Kyle's id + and `user-` plus the id's first four characters for anyone else (`user-e000` for Eve). The new + `refuse_users(true)` makes it answer 404 instead; `users/me` is not affected. +- **The speaker never reaches the model.** `loopd` logs `Turn.speaker` on the turn's `User` + record and builds the request from the text alone (task 02). So the text `gatewayd` sends is + the batch's `text` as before, with or without a speaker. +- The calls the edits use: + + | Call | Signature, and what it gives | + |---|---| + | `slice::first` | `self.senders.first() -> Option<&String>` | + | `Iterator::all` | `self.senders.iter().all(\|user\| user == first) -> bool` | + | `std::mem::take` | `std::mem::take(&mut pending.senders) -> Vec`, leaving an empty `Vec` | + | `HashMap::get`, `insert` | `self.usernames.get(user) -> Option<&String>`; `insert(String, String)` | + | `Client::username` | `pub fn username(&self, user: &str) -> Result` (unchanged) | + | `Option::map` | `speaker.map(str::to_string)`: `Option<&str>` to `Option` | + | `Option::as_deref` | `speaker.as_deref()`: `Option` to `Option<&str>` | + | `Batch::sender` | `pub fn sender(&self) -> Option<&str>` (new, S4) | + | `Gateway::speaker` | `fn speaker(&mut self, batch: &Batch) -> Option` (new, H2) | + | `run_turn` | `pub fn run_turn(socket: &Path, batch: &Batch, speaker: Option<&str>, on_event: &mut dyn FnMut(&TurnEvent)) -> Outcome` (D3) | + | `deliver` | `pub fn deliver(poster: &dyn Poster, socket: &Path, batch: &Batch, speaker: Option<&str>, log: &dyn Fn(&str), notice: &Notice)` (D4) | + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. + +### `crates/gatewayd/src/sessions.rs` + +**S1.** In `struct Message`, the lines + +```rust + /// A thread in a channel or group message that this Boxmaker now takes part in. + pub joins_thread: bool, +} +``` + +become + +```rust + /// A thread in a channel or group message that this Boxmaker now takes part in. + pub joins_thread: bool, + /// Who wrote it: the post's `user_id`, a Mattermost user id. + pub user: String, +} +``` + +**S2.** In `Router::route`, the lines + +```rust + text: format!("!{rest}"), + joins_thread: channel_type != "D", + }); +``` + +become + +```rust + text: format!("!{rest}"), + joins_thread: channel_type != "D", + user: post.user_id.clone(), + }); +``` + +**S3.** At the end of `Router::route`, the lines + +```rust + text: message.to_string(), + joins_thread: channel_type != "D", + }) +``` + +become + +```rust + text: message.to_string(), + joins_thread: channel_type != "D", + user: post.user_id.clone(), + }) +``` + +**S4.** The lines + +```rust +pub struct Batch { + pub session: SessionId, + pub thread: Thread, + pub resume: bool, + pub text: String, +} +``` + +become + +```rust +pub struct Batch { + pub session: SessionId, + pub thread: Thread, + pub resume: bool, + pub text: String, + /// Who wrote each message of `text`, in order: Mattermost user ids. + pub senders: Vec, +} + +impl Batch { + /// The one user who wrote every message of the batch. `None` when two or more people did, or + /// when it names nobody: a burst joined from several people has no speaker (M5c spec, + /// section 4). + pub fn sender(&self) -> Option<&str> { + let first = self.senders.first()?; + if self.senders.iter().all(|user| user == first) { + Some(first.as_str()) + } else { + None + } + } +} +``` + +**S5.** The lines + +```rust +struct Pending { + thread: Thread, + waiting: Vec, +} +``` + +become + +```rust +struct Pending { + thread: Thread, + waiting: Vec, + /// The author of each message in `waiting`, in the same order. + senders: Vec, +} +``` + +**S6.** In `Queues::push`, the lines + +```rust + pending.waiting.push(message.text.clone()); + return Pushed::Waiting; +``` + +become + +```rust + pending.waiting.push(message.text.clone()); + pending.senders.push(message.user.clone()); + return Pushed::Waiting; +``` + +**S7.** In `Queues::push`, the lines + +```rust + text: message.text.clone(), + }; + self.running.insert( + message.session, + Pending { + thread: message.thread.clone(), + waiting: Vec::new(), + }, + ); +``` + +become + +```rust + text: message.text.clone(), + senders: vec![message.user.clone()], + }; + self.running.insert( + message.session, + Pending { + thread: message.thread.clone(), + waiting: Vec::new(), + senders: Vec::new(), + }, + ); +``` + +**S8.** In `Queues::finish`, the lines + +```rust + text: pending.waiting.join("\n\n"), + }; +``` + +become + +```rust + text: pending.waiting.join("\n\n"), + senders: std::mem::take(&mut pending.senders), + }; +``` + +### `crates/gatewayd/src/deliver.rs` + +**D1.** The lines + +```rust + batch: &Batch, + resume: bool, + on_event: &mut dyn FnMut(&TurnEvent), +) -> Outcome { +``` + +become + +```rust + batch: &Batch, + resume: bool, + speaker: Option<&str>, + on_event: &mut dyn FnMut(&TurnEvent), +) -> Outcome { +``` + +**D2.** In `one_turn`, the lines + +```rust + // Set in M5c-1 task 03. Until then no turn has a speaker, and none is synced. + speaker: None, +``` + +become + +```rust + // On the turn only, never in its text: `memoryd` reads it from the log, the model never. + speaker: speaker.map(str::to_string), +``` + +**D3.** The lines + +```rust +/// A turn for a batch. A reply in a thread `loopd` does not know creates the session, as +/// `bxctl chat --session` does. +pub fn run_turn(socket: &Path, batch: &Batch, on_event: &mut dyn FnMut(&TurnEvent)) -> Outcome { + // `one_turn` with batch.resume. When that is Refused with NoSuchSession and batch.resume was + // true, one more `one_turn` with resume false. Otherwise the first outcome. + match one_turn(socket, batch, batch.resume, on_event) { + Outcome::Refused(err) if err.code == ErrorCode::NoSuchSession && batch.resume => { + one_turn(socket, batch, false, on_event) + } +``` + +become + +```rust +/// A turn for a batch, spoken by `speaker` (a Mattermost username) when one person wrote all of +/// it. A reply in a thread `loopd` does not know creates the session, as `bxctl chat --session` +/// does. +pub fn run_turn( + socket: &Path, + batch: &Batch, + speaker: Option<&str>, + on_event: &mut dyn FnMut(&TurnEvent), +) -> Outcome { + // `one_turn` with batch.resume. When that is Refused with NoSuchSession and batch.resume was + // true, one more `one_turn` with resume false. Otherwise the first outcome. + match one_turn(socket, batch, batch.resume, speaker, on_event) { + Outcome::Refused(err) if err.code == ErrorCode::NoSuchSession && batch.resume => { + one_turn(socket, batch, false, speaker, on_event) + } +``` + +**D4.** The lines + +```rust +/// Run a batch's turn and post what comes of it in its thread. A post that fails is logged. +pub fn deliver( + poster: &dyn Poster, + socket: &Path, + batch: &Batch, + log: &dyn Fn(&str), +``` + +become + +```rust +/// Run a batch's turn, spoken by `speaker`, and post what comes of it in its thread. A post that +/// fails is logged. +pub fn deliver( + poster: &dyn Poster, + socket: &Path, + batch: &Batch, + speaker: Option<&str>, + log: &dyn Fn(&str), +``` + +**D5.** In `deliver`, the line + +```rust + let outcome = run_turn(socket, batch, &mut |event| { +``` + +becomes + +```rust + let outcome = run_turn(socket, batch, speaker, &mut |event| { +``` + +### `crates/gatewayd/src/serve/mod.rs` + +**G1.** At the end of `struct Gateway`, the lines + +```rust + /// How long a turn waits for its approval to come through `gateway.sock` before the notice. + notice: Duration, +} +``` + +become + +```rust + /// How long a turn waits for its approval to come through `gateway.sock` before the notice. + notice: Duration, + /// Mattermost usernames by user id, each looked up once (M5c spec, section 11). + usernames: HashMap, +} +``` + +**G2.** In `Gateway::new`, the lines + +```rust + askers: HashMap::new(), + notice, + } +``` + +become + +```rust + askers: HashMap::new(), + notice, + usernames: HashMap::new(), + } +``` + +`HashMap` is imported at the top of the file already. + +### `crates/gatewayd/src/serve/handle.rs` + +**H1.** In `start`, the lines + +```rust + let (session, thread) = (batch.session.clone(), batch.thread.clone()); + let spawned = std::thread::Builder::new().spawn(move || { + let posted = |id: u64| registry.was_posted(id); + let notice = Notice { + posted: &posted, + wait, + }; + deliver(client.as_ref(), &socket, &batch, &|line| log(line), ¬ice); +``` + +become + +```rust + let (session, thread) = (batch.session.clone(), batch.thread.clone()); + let speaker = self.speaker(&batch); + let spawned = std::thread::Builder::new().spawn(move || { + let posted = |id: u64| registry.was_posted(id); + let notice = Notice { + posted: &posted, + wait, + }; + deliver( + client.as_ref(), + &socket, + &batch, + speaker.as_deref(), + &|line| log(line), + ¬ice, + ); +``` + +**H2.** The lines + +```rust + /// Turns that ended: out of flight, who asked forgotten, and the next batch of each session + /// started. Never blocks. +``` + +become + +```rust + /// The turn's speaker: the username of the one person who wrote all of the batch, as + /// Mattermost gives it (M5c spec, section 11). `None` when two or more people did, or when + /// the name cannot be found: never the user id, never a guess. A name found is kept; a + /// failure is logged and not kept, so the next turn looks again. + fn speaker(&mut self, batch: &Batch) -> Option { + let user = batch.sender()?; + if let Some(name) = self.usernames.get(user) { + return Some(name.clone()); + } + match self.client.username(user) { + Ok(name) => { + self.usernames.insert(user.to_string(), name.clone()); + Some(name) + } + Err(e) => { + (self.log)(&format!( + "gatewayd: no username for {user}: {e}; the turn has no speaker" + )); + None + } + } + } + + /// Turns that ended: out of flight, who asked forgotten, and the next batch of each session + /// started. Never blocks. +``` + +`Batch` is imported at the top of the file already. + +## Steps + +- [ ] **1. Copy the given files.** From the repository root, seven commands: + + ```sh + cp docs/plans/M5c1/files/crates/gatewayd/tests/deliver_speaker.rs crates/gatewayd/tests/ + cp docs/plans/M5c1/files/crates/gatewayd/tests/serve_speaker.rs crates/gatewayd/tests/ + cp docs/plans/M5c1/files/crates/gatewayd/tests/sessions.rs crates/gatewayd/tests/ + cp docs/plans/M5c1/files/crates/gatewayd/tests/deliver.rs crates/gatewayd/tests/ + cp docs/plans/M5c1/files/crates/gatewayd/tests/deliver_links.rs crates/gatewayd/tests/ + cp docs/plans/M5c1/files/crates/gatewayd/tests/deliver_notice.rs crates/gatewayd/tests/ + cp docs/plans/M5c1/files/crates/gatewayd/tests/support/fake_mm.rs \ + crates/gatewayd/tests/support/ + ``` + + Then `git status --short` lists 5 modified files and 2 new ones, all under + `crates/gatewayd/tests/`. +- [ ] **2. See them fail.** Two commands: + - `cargo build --tests --keep-going -p gatewayd`: five test targets do not compile: + `deliver`, `deliver_links`, `deliver_notice`, `deliver_speaker` and `sessions`. Every error is + about a name this task adds or changes: `E0560` (``struct `Batch` has no field named + `senders` `` or ``struct `gatewayd::sessions::Message` has no field named `user` ``), `E0609` + (``no field `user` on type `gatewayd::sessions::Message` `` or ``no field `senders` on type + `&Batch` ``), `E0599` (``no method named `sender` found for struct `Batch` ``), `E0061` + (``this function takes 5 arguments but 6 arguments were supplied``, for `deliver`) or, once, + `E0282` (``type annotations needed``, in `deliver.rs`, on the closure passed to `deliver`). + - `cargo test -p gatewayd --test serve_speaker`: it compiles, and its four tests fail. Three + fail on an assertion that finds `None` where a speaker was expected + (`one_persons_message_has_their_username`, `a_username_is_looked_up_once`, + `bursts_have_a_speaker_only_when_one_person_wrote_them`); the fourth, + `a_name_that_cannot_be_found_leaves_the_turn_without_a_speaker`, with + `no log line with "no username for"`. +- [ ] **3. `sessions.rs`.** Make edits S1 to S8. `cargo check -p gatewayd`: no error. Then + `cargo test -p gatewayd --test sessions`: 11 passed. +- [ ] **4. `deliver.rs`.** Make edits D1 to D5. `cargo check -p gatewayd` now fails in + `serve/handle.rs` only, at the call of `deliver`: `E0061` (``this function takes 6 arguments + but 5 arguments were supplied``) and `E0277` (``expected an `Fn(&str)` closure, found + `Notice<'_>` ``). The notes under them also point into `deliver.rs`, at `deliver`'s + definition; that is where the call is compared, not a second error. +- [ ] **5. `serve`.** Make edits G1, G2, H1 and H2. `cargo check -p gatewayd`: no error and no + warning. +- [ ] **6. See them pass.** `cargo test -p gatewayd --test deliver_speaker --test serve_speaker`: + 3 and 4 passed. Then `cargo test -p gatewayd`: every suite passes. The new tests: + + | Test | What it checks | + |---|---| + | `a_queued_message_names_its_author` (`sessions.rs`) | S2, S3: `user` on both kinds of queued message | + | `a_batch_carries_the_author_of_each_message` (`sessions.rs`) | S4 to S8: one author, two authors, and nothing carried to the next batch | + | `a_batch_with_no_authors_has_no_sender` (`sessions.rs`) | S4: no authors, and a last one that differs | + | `the_speaker_goes_on_the_turn_and_not_into_its_text` | D1, D2, D4, D5: the speaker on the `Turn`, the text unchanged | + | `without_a_speaker_the_turn_has_none` | D2: `None` stays `None` | + | `the_turn_sent_again_for_an_unknown_session_keeps_the_speaker` | D3: both turns carry it | + | `one_persons_message_has_their_username` | H1, H2: `kyle`, not the id | + | `a_username_is_looked_up_once` | G1, H2: one `GET /api/v4/users/…` for two turns | + | `bursts_have_a_speaker_only_when_one_person_wrote_them` | end to end: `kyle`, then `None`, then `user-e000` | + | `a_name_that_cannot_be_found_leaves_the_turn_without_a_speaker` | H2: `None`, the log line, and a later lookup | + + `a_full_queue_drops_the_message` in `sessions.rs` also checks that a refused message's author + is not kept. +- [ ] **7. Check.** Three commands: + - `grep -rn 'speaker: None' crates/gatewayd/src` prints nothing: the placeholder is gone. + - `git diff --stat -- crates/gatewayd/src` lists exactly `deliver.rs`, `serve/handle.rs`, + `serve/mod.rs` and `sessions.rs`. + - `git diff --stat -- crates/bxcore crates/loopd crates/memoryd crates/brokerd crates/bxctl` + prints nothing. +- [ ] **8. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1148 tests: 1138 before this task, plus 10 new ones (three + in `sessions.rs`, three in `deliver_speaker.rs`, four in `serve_speaker.rs`; the new versions + of the other files keep their counts). +- [ ] **9. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is + `M5c1/03-gatewayd-speaker`. `git status --short` lists the four sources, the five replaced test + files and your log as modified, and `deliver_speaker.rs` and `serve_speaker.rs` as new. Stage + with `git add crates/gatewayd docs/implementer-log.md Cargo.lock`, then commit as `AGENTS.md` + says. `Cargo.lock` does not change in this task; the `git add` is harmless. + +## Done when + +- `make gate` prints `gate: ok` with 1148 tests. +- `git diff HEAD~1 --stat` lists only the four source files of "Modify", the seven copied files + and `docs/implementer-log.md`. + +## Stop and report if + +- A "before" block of any edit is not in its file exactly as shown, or is there more than once. +- Step 2 shows a different result: other test targets, an error of another kind, or other + failures in `serve_speaker`. +- Step 4's `cargo check` fails anywhere other than the call of `deliver` in `serve/handle.rs`, or + with other errors. +- After step 3, 5 or 6, a test fails. Do not change a test: stop, and give the failing tests' + names and their messages. +- The fix seems to need a change anywhere else: in `serve/approvals.rs`, `mm/rest.rs`, another + crate, a change to how the texts are joined, or a speaker when the name cannot be found. It + does not; stop and say why you think so. +- A test in `inferproxy`, `toolkit` or `memoryd` fails in the gate: check once whether it fails + without your change, then stop and report both results. +- The gate passes with a count other than 1148. +- A comment in an edit does not fit the code under it. Quote it and stop. diff --git a/docs/plans/M5c1/files/crates/gatewayd/tests/deliver.rs b/docs/plans/M5c1/files/crates/gatewayd/tests/deliver.rs new file mode 100644 index 0000000..b4baf9a --- /dev/null +++ b/docs/plans/M5c1/files/crates/gatewayd/tests/deliver.rs @@ -0,0 +1,316 @@ +//! A turn on `loop.sock` and its answer in the thread, against a fake `loopd`: the answer, long +//! answers, approvals, errors, an unknown session, and a loop that is not there (M4a spec, section +//! 8; M4b spec, section 7, for the notice). Do not edit. + +#[path = "support/fake_loop.rs"] +mod fake_loop; +#[path = "support/tmp.rs"] +mod tmp; + +use std::sync::Mutex; +use std::time::Duration; + +use bxcore::{DataClass, ErrorCode, SessionId, Timestamp, TurnEvent}; +use fake_loop::{Reply, done, error, event, serve_loop}; +use gatewayd::deliver::{EMPTY_ANSWER, LOOP_DOWN, MAX_POST, Notice, Poster, deliver, split_answer}; +use gatewayd::mm::MmError; +use gatewayd::sessions::{Batch, Thread}; +use tmp::TempDir; + +const DM: &str = "d0000000000000000000000000"; +const ROOT: &str = "r0000000000000000000000000"; +const KYLE: &str = "k0000000000000000000000000"; + +#[derive(Default)] +struct Record { + posts: Mutex>, + fail: bool, +} + +impl Poster for Record { + fn post(&self, channel: &str, root: &str, text: &str) -> Result<(), MmError> { + if self.fail { + return Err(MmError::Status(500, "down".to_string())); + } + self.posts + .lock() + .unwrap() + .push((channel.to_string(), root.to_string(), text.to_string())); + Ok(()) + } +} + +impl Record { + fn texts(&self) -> Vec { + let posts = self.posts.lock().unwrap(); + assert!( + posts.iter().all(|(c, r, _)| c == DM && r == ROOT), + "every post in the thread" + ); + posts.iter().map(|(_, _, t)| t.clone()).collect() + } +} + +fn batch(resume: bool, text: &str) -> Batch { + Batch { + session: SessionId::new(&format!("mm-{ROOT}")).unwrap(), + thread: Thread { + channel: DM.to_string(), + root: ROOT.to_string(), + }, + resume, + text: text.to_string(), + senders: vec![KYLE.to_string()], + } +} + +fn run(dir: &TempDir, poster: &Record, batch: &Batch) -> Vec { + let log = Mutex::new(Vec::new()); + let never = |_: u64| false; + let notice = Notice { + posted: &never, + wait: Duration::ZERO, + }; + deliver( + poster, + &dir.path().join("loop.sock"), + batch, + None, + &|line| log.lock().unwrap().push(line.to_string()), + ¬ice, + ); + log.into_inner().unwrap() +} + +#[test] +fn the_answer_is_posted_and_nothing_else() { + let dir = TempDir::new("deliver-answer"); + let turns = serve_loop(&dir.path().join("loop.sock"), |_, _| { + vec![ + event(TurnEvent::Progress { + total: 10, + cache: 0, + processed: 10, + }), + event(TurnEvent::Reasoning { + text: "private thoughts".to_string(), + }), + event(TurnEvent::ToolCallStarted { + name: "read_file".to_string(), + }), + event(TurnEvent::ToolResult { + name: "read_file".to_string(), + class: DataClass::Private, + truncated: false, + }), + event(TurnEvent::Content { + text: "The ans".to_string(), + }), + done("The answer."), + ] + }); + let poster = Record::default(); + let log = run(&dir, &poster, &batch(true, "one\n\ntwo")); + assert_eq!(poster.texts(), ["The answer."]); + assert!(log.is_empty(), "{log:?}"); + let turn = turns.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!( + (turn.session.as_str(), turn.content.as_str(), turn.resume), + (format!("mm-{ROOT}").as_str(), "one\n\ntwo", true) + ); +} + +#[test] +fn an_approval_is_announced_once_before_the_answer() { + let dir = TempDir::new("deliver-approval"); + let expires = Timestamp::from_unix_millis(1_758_650_000_000).unwrap(); + let _turns = serve_loop(&dir.path().join("loop.sock"), move |_, _| { + vec![ + event(TurnEvent::ApprovalPending { + approval: 42, + tool: "shell".to_string(), + expires, + }), + done("done"), + ] + }); + let poster = Record::default(); + run(&dir, &poster, &batch(false, "go")); + assert_eq!( + poster.texts(), + [ + "waiting for approval 42: brokerd has not sent it here; answer it with bxctl approvals", + "done" + ] + ); +} + +#[test] +fn errors_are_posted_with_their_code() { + let dir = TempDir::new("deliver-error"); + let _turns = serve_loop(&dir.path().join("loop.sock"), |_, _| { + vec![error( + ErrorCode::Inference, + "the model server failed\nsee docs/runbook.md#loopd-selftest-failed", + )] + }); + let poster = Record::default(); + run(&dir, &poster, &batch(false, "go")); + assert_eq!( + poster.texts(), + ["Error: inference: the model server failed\nsee docs/runbook.md#loopd-selftest-failed"] + ); +} + +#[test] +fn a_reply_in_a_thread_loopd_does_not_know_creates_the_session() { + let dir = TempDir::new("deliver-unknown"); + let turns = serve_loop(&dir.path().join("loop.sock"), |n, _| { + if n == 0 { + vec![error(ErrorCode::NoSuchSession, "no such session")] + } else { + vec![done("hello")] + } + }); + let poster = Record::default(); + run(&dir, &poster, &batch(true, "hi")); + assert_eq!(poster.texts(), ["hello"]); + let first = turns.recv_timeout(Duration::from_secs(5)).unwrap(); + let second = turns.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!( + (first.resume, second.resume, second.content.as_str()), + (true, false, "hi") + ); +} + +#[test] +fn a_new_session_is_not_retried() { + let dir = TempDir::new("deliver-noretry"); + let turns = serve_loop(&dir.path().join("loop.sock"), |_, _| { + vec![error(ErrorCode::NoSuchSession, "odd")] + }); + let poster = Record::default(); + run(&dir, &poster, &batch(false, "hi")); + assert_eq!(poster.texts(), ["Error: no_such_session: odd"]); + assert!(turns.recv_timeout(Duration::from_secs(5)).is_ok()); + assert!( + turns.recv_timeout(Duration::from_millis(200)).is_err(), + "one turn only" + ); +} + +#[test] +fn a_loop_that_is_not_there_or_goes_away() { + let dir = TempDir::new("deliver-down"); + let poster = Record::default(); + let log = run(&dir, &poster, &batch(false, "hi")); + assert_eq!(poster.texts(), [LOOP_DOWN]); + assert_eq!(log.len(), 1, "{log:?}"); + assert!( + log[0].starts_with(&format!("gatewayd: mm-{ROOT}: cannot connect to ")), + "{log:?}" + ); + + for (n, replies) in [ + vec![ + event(TurnEvent::Content { + text: "x".to_string(), + }), + Reply::Close, + ], + vec![Reply::Bytes(b"\x00\x00\x00\x05{bad}".to_vec())], + vec![ + Reply::Frame(bxcore::Envelope { + v: 1, + id: 2, + r#final: true, + msg: bxcore::Message::Ok(bxcore::Empty {}), + }), + done("late"), + ], + ] + .into_iter() + .enumerate() + { + let dir = TempDir::new(&format!("deliver-early-{n}")); + let replies = Mutex::new(Some(replies)); + let _turns = serve_loop(&dir.path().join("loop.sock"), move |_, _| { + replies.lock().unwrap().take().unwrap_or_default() + }); + let poster = Record::default(); + let log = run(&dir, &poster, &batch(false, "hi")); + assert_eq!(poster.texts(), [LOOP_DOWN], "case {n}"); + assert_eq!(log.len(), 1, "case {n}: {log:?}"); + } +} + +#[test] +fn a_post_that_fails_is_logged() { + let dir = TempDir::new("deliver-postfail"); + let _turns = serve_loop(&dir.path().join("loop.sock"), |_, _| vec![done("lost")]); + let poster = Record { + fail: true, + ..Record::default() + }; + let log = run(&dir, &poster, &batch(false, "hi")); + assert_eq!( + log, + [format!( + "gatewayd: cannot post in {DM} (thread {ROOT}): status 500: \"down\"" + )] + ); +} + +#[test] +fn long_answers_are_split_at_newlines() { + let short = "a".repeat(MAX_POST); + assert_eq!(split_answer(&short), [short.as_str()]); + let over = "a".repeat(MAX_POST + 1); + assert_eq!(split_answer(&over), ["a".repeat(MAX_POST), "a".to_string()]); + let lines = format!( + "{}\n{}\n{}", + "a".repeat(10_000), + "b".repeat(5_000), + "c".repeat(2_000) + ); + assert_eq!( + split_answer(&lines), + [ + format!("{}\n{}", "a".repeat(10_000), "b".repeat(5_000)), + "c".repeat(2_000) + ] + ); + let wide = "é".repeat(MAX_POST + 5); + let parts = split_answer(&wide); + assert_eq!( + parts.iter().map(|p| p.chars().count()).collect::>(), + [MAX_POST, 5], + "characters, not bytes" + ); + let leading = format!("\n{}", "x".repeat(MAX_POST + 1)); + let parts = split_answer(&leading); + assert!( + parts + .iter() + .all(|p| !p.is_empty() && p.chars().count() <= MAX_POST), + "{:?}", + parts.iter().map(|p| p.len()).collect::>() + ); + assert_eq!(parts.concat(), leading, "a hard cut drops nothing"); + assert_eq!(split_answer(""), [EMPTY_ANSWER]); + assert_eq!(split_answer(" \n "), [EMPTY_ANSWER]); +} + +#[test] +fn a_long_answer_is_posted_in_order() { + let dir = TempDir::new("deliver-long"); + let answer = format!("{}\n{}", "a".repeat(MAX_POST - 1), "b".repeat(MAX_POST)); + let sent = answer.clone(); + let _turns = serve_loop(&dir.path().join("loop.sock"), move |_, _| vec![done(&sent)]); + let poster = Record::default(); + run(&dir, &poster, &batch(false, "hi")); + assert_eq!( + poster.texts(), + ["a".repeat(MAX_POST - 1), "b".repeat(MAX_POST)] + ); +} diff --git a/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_links.rs b/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_links.rs new file mode 100644 index 0000000..09bcfc0 --- /dev/null +++ b/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_links.rs @@ -0,0 +1,198 @@ +//! A reply's images and links as `deliver` posts them, against a fake `loopd` whose `TurnDone` +//! says whether the session has read untrusted content; and `answer_posts` on replies long enough +//! to be split (M5b spec, section 9). Do not edit. + +#[path = "support/fake_loop.rs"] +mod fake_loop; +#[path = "support/tmp.rs"] +mod tmp; + +use std::sync::Mutex; +use std::time::Duration; + +use bxcore::{Envelope, Message, PROTOCOL_VERSION, SessionId, TurnDone, Usage}; +use fake_loop::{Reply, done, serve_loop}; +use gatewayd::deliver::{EMPTY_ANSWER, MAX_POST, Notice, Poster, answer_posts, deliver}; +use gatewayd::links::NOTICE; +use gatewayd::mm::MmError; +use gatewayd::sessions::{Batch, Thread}; +use tmp::TempDir; + +const DM: &str = "d0000000000000000000000000"; +const ROOT: &str = "r0000000000000000000000000"; + +#[derive(Default)] +struct Record { + posts: Mutex>, +} + +impl Poster for Record { + fn post(&self, channel: &str, root: &str, text: &str) -> Result<(), MmError> { + assert_eq!((channel, root), (DM, ROOT), "every post in the thread"); + self.posts.lock().unwrap().push(text.to_string()); + Ok(()) + } +} + +/// The frame that ends a turn of a session that has read untrusted content. +fn done_untrusted(content: &str) -> Reply { + let msg = Message::TurnDone(TurnDone { + content: content.to_string(), + usage: Usage { + cache_n: 0, + prompt_n: 1, + predicted_n: 1, + reasoning_tokens: 0, + thinking_capped: false, + }, + untrusted: true, + }); + Reply::Frame(Envelope { + v: PROTOCOL_VERSION, + id: 1, + r#final: true, + msg, + }) +} + +/// One turn against a fake `loopd` that ends it with `reply`; the posts, in order. +fn posts(tag: &str, reply: fn(&str) -> Reply, content: &'static str) -> Vec { + let dir = TempDir::new(tag); + let socket = dir.path().join("loop.sock"); + let turns = serve_loop(&socket, move |_, _| vec![reply(content)]); + let batch = Batch { + session: SessionId::new(&format!("mm-{ROOT}")).unwrap(), + thread: Thread { + channel: DM.to_string(), + root: ROOT.to_string(), + }, + resume: false, + text: "hi".to_string(), + senders: vec!["k0000000000000000000000000".to_string()], + }; + let never = |_: u64| false; + let notice = Notice { + posted: &never, + wait: Duration::ZERO, + }; + let record = Record::default(); + let log = Mutex::new(Vec::new()); + deliver( + &record, + &socket, + &batch, + None, + &|line| log.lock().unwrap().push(line.to_string()), + ¬ice, + ); + turns.recv_timeout(Duration::from_secs(5)).unwrap(); + assert!(log.lock().unwrap().is_empty(), "{:?}", log.lock().unwrap()); + record.posts.into_inner().unwrap() +} + +const LINKED: &str = "See [the docs](https://docs.example/p) and https://b.example/x."; + +#[test] +fn an_untrusted_reply_posts_its_links_as_code_and_ends_with_the_notice() { + assert_eq!( + posts("links-untrusted", done_untrusted, LINKED), + [format!( + "See `[the docs](https://docs.example/p)` and `https://b.example/x`.\n\n{NOTICE}" + )] + ); +} + +#[test] +fn a_clean_reply_keeps_its_links() { + assert_eq!(posts("links-clean", done, LINKED), [LINKED]); +} + +#[test] +fn an_image_that_carries_data_is_posted_as_text_in_a_clean_reply() { + assert_eq!( + posts( + "links-image", + done, + "Summary done. ![s](https://evil.example/p.png?d=c2VjcmV0)" + ), + ["Summary done. `![s](https://evil.example/p.png?d=c2VjcmV0)`"] + ); +} + +#[test] +fn an_untrusted_reply_with_only_an_image_has_no_notice() { + assert_eq!( + posts( + "links-only-image", + done_untrusted, + "![s](https://e.example/a)" + ), + ["`![s](https://e.example/a)`"] + ); +} + +#[test] +fn code_in_an_untrusted_reply_is_left_and_needs_no_notice() { + let text = "Run `curl https://example.com/x`:\n```\nhttps://example.com/y\n```"; + assert_eq!(posts("links-code", done_untrusted, text), [text]); +} + +#[test] +fn a_long_untrusted_reply_has_one_notice_at_its_end() { + let content = format!( + "https://a.example/1\n{}\nhttps://b.example/2", + "x".repeat(MAX_POST - 10) + ); + let posts = answer_posts(&content, true); + assert_eq!( + posts, + [ + "`https://a.example/1`".to_string(), + "x".repeat(MAX_POST - 10), + format!("`https://b.example/2`\n\n{NOTICE}"), + ] + ); +} + +#[test] +fn a_part_that_starts_inside_a_fence_is_disarmed_again() { + let content = format!( + "```\n{}\n{}\n![i](https://e.example/a)\n```", + "y".repeat(10_000), + "z".repeat(10_000) + ); + assert_eq!( + answer_posts(&content, false), + [ + format!("```\n{}", "y".repeat(10_000)), + format!( + "{}\n````![i](https://e.example/a)````\n```", + "z".repeat(10_000) + ), + ] + ); +} + +#[test] +fn a_link_found_only_after_the_split_still_brings_the_notice() { + let content = format!( + "```\n{}\n{}\nhttps://e.example/a\n```", + "y".repeat(10_000), + "z".repeat(10_000) + ); + assert_eq!( + answer_posts(&content, true), + [ + format!("```\n{}", "y".repeat(10_000)), + format!( + "{}\n````https://e.example/a````\n```\n\n{NOTICE}", + "z".repeat(10_000) + ), + ] + ); +} + +#[test] +fn an_empty_answer_is_posted_as_before() { + assert_eq!(answer_posts(" \n", true), [EMPTY_ANSWER]); +} diff --git a/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_notice.rs b/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_notice.rs new file mode 100644 index 0000000..181d4d7 --- /dev/null +++ b/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_notice.rs @@ -0,0 +1,103 @@ +//! The notice for an approval that `brokerd` did not bring: posted only when it has not arrived +//! within the wait (M4b spec, section 7). Do not edit. + +#[path = "support/fake_loop.rs"] +mod fake_loop; +#[path = "support/tmp.rs"] +mod tmp; + +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use bxcore::{SessionId, Timestamp, TurnEvent}; +use fake_loop::{done, event, serve_loop}; +use gatewayd::deliver::{Notice, Poster, deliver}; +use gatewayd::mm::MmError; +use gatewayd::sessions::{Batch, Thread}; +use tmp::TempDir; + +const DM: &str = "d0000000000000000000000000"; +const ROOT: &str = "r0000000000000000000000000"; +const NOTICE: &str = + "waiting for approval 42: brokerd has not sent it here; answer it with bxctl approvals"; + +#[derive(Default)] +struct Record(Mutex>); + +impl Poster for Record { + fn post(&self, _: &str, _: &str, text: &str) -> Result<(), MmError> { + self.0.lock().unwrap().push(text.to_string()); + Ok(()) + } +} + +/// A turn whose loop says approval 42 is pending, then answers; what was posted, and how long it took. +fn turn(tag: &str, posted: &dyn Fn(u64) -> bool, wait: Duration) -> (Vec, Duration) { + let dir = TempDir::new(tag); + let socket = dir.path().join("loop.sock"); + let expires = Timestamp::parse("2026-09-24T08:20:00.000Z").unwrap(); + let _turns = serve_loop(&socket, move |_, _| { + vec![ + event(TurnEvent::ApprovalPending { + approval: 42, + tool: "read_file".to_string(), + expires, + }), + done("done"), + ] + }); + let batch = Batch { + session: SessionId::new(&format!("mm-{ROOT}")).unwrap(), + thread: Thread { + channel: DM.to_string(), + root: ROOT.to_string(), + }, + resume: false, + text: "go".to_string(), + senders: vec!["k0000000000000000000000000".to_string()], + }; + let poster = Record::default(); + let started = Instant::now(); + deliver( + &poster, + &socket, + &batch, + None, + &|_| {}, + &Notice { posted, wait }, + ); + let took = started.elapsed(); + (poster.0.into_inner().unwrap(), took) +} + +#[test] +fn brought_already_no_notice_and_no_wait() { + let (posts, took) = turn("notice-there", &|id| id == 42, Duration::from_secs(5)); + assert_eq!(posts, ["done"]); + assert!(took < Duration::from_secs(1), "{took:?}"); +} + +#[test] +fn brought_during_the_wait_no_notice() { + let started = Instant::now(); + let later = move |id: u64| id == 42 && started.elapsed() > Duration::from_millis(300); + let (posts, took) = turn("notice-later", &later, Duration::from_secs(5)); + assert_eq!(posts, ["done"]); + assert!(took < Duration::from_secs(2), "{took:?}"); +} + +#[test] +fn never_brought_the_notice_after_the_wait() { + let (posts, took) = turn("notice-never", &|_| false, Duration::from_millis(400)); + assert_eq!(posts, [NOTICE, "done"]); + assert!( + took >= Duration::from_millis(400) && took < Duration::from_secs(3), + "{took:?}" + ); +} + +#[test] +fn another_approval_brought_does_not_count() { + let (posts, _) = turn("notice-other", &|id| id == 7, Duration::from_millis(200)); + assert_eq!(posts, [NOTICE, "done"]); +} diff --git a/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_speaker.rs b/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_speaker.rs new file mode 100644 index 0000000..9b4ff2c --- /dev/null +++ b/docs/plans/M5c1/files/crates/gatewayd/tests/deliver_speaker.rs @@ -0,0 +1,121 @@ +//! The speaker `deliver` is given goes on the turn it sends to `loopd`, and nowhere else: not into +//! the turn's text (M5c spec, section 11). Do not edit. + +#[path = "support/fake_loop.rs"] +mod fake_loop; +#[path = "support/tmp.rs"] +mod tmp; + +use std::sync::Mutex; +use std::time::Duration; + +use bxcore::{ErrorCode, SessionId, Turn}; +use fake_loop::{done, error, serve_loop}; +use gatewayd::deliver::{Notice, Poster, deliver}; +use gatewayd::mm::MmError; +use gatewayd::sessions::{Batch, Thread}; +use tmp::TempDir; + +const DM: &str = "d0000000000000000000000000"; +const ROOT: &str = "r0000000000000000000000000"; +const KYLE: &str = "k0000000000000000000000000"; +const WAIT: Duration = Duration::from_secs(5); + +#[derive(Default)] +struct Record(Mutex>); + +impl Poster for Record { + fn post(&self, _: &str, _: &str, text: &str) -> Result<(), MmError> { + self.0.lock().unwrap().push(text.to_string()); + Ok(()) + } +} + +/// One `deliver` of "one\n\ntwo" with `speaker`, against a fake `loopd` whose n-th turn gets +/// `script(n)`; the turns it received and what was posted. +fn deliver_with( + tag: &str, + speaker: Option<&str>, + script: fn(usize) -> Vec, +) -> (Vec, Vec) { + let dir = TempDir::new(tag); + let socket = dir.path().join("loop.sock"); + let turns = serve_loop(&socket, move |n, _| script(n)); + let batch = Batch { + session: SessionId::new(&format!("mm-{ROOT}")).unwrap(), + thread: Thread { + channel: DM.to_string(), + root: ROOT.to_string(), + }, + resume: true, + text: "one\n\ntwo".to_string(), + senders: vec![KYLE.to_string(), KYLE.to_string()], + }; + let never = |_: u64| false; + let notice = Notice { + posted: &never, + wait: Duration::ZERO, + }; + let poster = Record::default(); + let log = Mutex::new(Vec::new()); + deliver( + &poster, + &socket, + &batch, + speaker, + &|line| log.lock().unwrap().push(line.to_string()), + ¬ice, + ); + assert!(log.lock().unwrap().is_empty(), "{:?}", log.lock().unwrap()); + // The fake reports each turn before it answers, so every turn is in by now. + let mut got = vec![turns.recv_timeout(WAIT).unwrap()]; + while let Ok(turn) = turns.try_recv() { + got.push(turn); + } + (got, poster.0.into_inner().unwrap()) +} + +fn answer(_: usize) -> Vec { + vec![done("fine")] +} + +#[test] +fn the_speaker_goes_on_the_turn_and_not_into_its_text() { + let (turns, posts) = deliver_with("speaker-some", Some("first.last"), answer); + assert_eq!(turns.len(), 1); + let turn = &turns[0]; + assert_eq!(turn.speaker.as_deref(), Some("first.last")); + assert_eq!( + (turn.content.as_str(), turn.resume), + ("one\n\ntwo", true), + "the text is the batch's, unchanged" + ); + assert_eq!(posts, ["fine"]); +} + +#[test] +fn without_a_speaker_the_turn_has_none() { + let (turns, posts) = deliver_with("speaker-none", None, answer); + assert_eq!(turns.len(), 1); + assert_eq!(turns[0].speaker, None, "never the user id"); + assert_eq!(turns[0].content, "one\n\ntwo"); + assert_eq!(posts, ["fine"]); +} + +#[test] +fn the_turn_sent_again_for_an_unknown_session_keeps_the_speaker() { + let unknown_first = |n: usize| { + if n == 0 { + vec![error(ErrorCode::NoSuchSession, "no such session")] + } else { + vec![done("hello")] + } + }; + let (turns, posts) = deliver_with("speaker-again", Some("kyle"), unknown_first); + let seen: Vec<(bool, Option<&str>)> = turns + .iter() + .map(|t| (t.resume, t.speaker.as_deref())) + .collect(); + assert_eq!(seen, [(true, Some("kyle")), (false, Some("kyle"))]); + assert_eq!(posts, ["hello"]); +} diff --git a/docs/plans/M5c1/files/crates/gatewayd/tests/serve_speaker.rs b/docs/plans/M5c1/files/crates/gatewayd/tests/serve_speaker.rs new file mode 100644 index 0000000..e1dfce9 --- /dev/null +++ b/docs/plans/M5c1/files/crates/gatewayd/tests/serve_speaker.rs @@ -0,0 +1,188 @@ +//! `gatewayd` end to end: each turn's speaker is the Mattermost username of the one person who +//! wrote every message of its burst, and there is none when two people did or the name cannot be +//! found (M5c spec, sections 4 and 11). Do not edit. + +#[path = "support/fake_loop.rs"] +mod fake_loop; +#[path = "support/fake_mm.rs"] +mod fake_mm; +#[path = "support/gateway.rs"] +mod gateway; +#[path = "support/tmp.rs"] +mod tmp; + +use std::sync::{Mutex, mpsc}; +use std::time::{Duration, Instant}; + +use fake_loop::{done, serve_loop}; +use fake_mm::{DM, EVE, FakeMm, KYLE, id, post}; +use gateway::{SHARED, WAIT, loop_dir, read_state, start, up}; +use gatewayd::config::Config; +use tmp::TempDir; + +/// Kyle and Eve both allowed, in the shared channel. +fn two_people(home: &TempDir, url: &str) -> Config { + let text = format!( + r#" +[mattermost] +url = "{url}" +[secrets.mattermost_token] +env = "NOT_READ_BY_RUN" +[allow] +users = ["{KYLE}", "{EVE}"] +channels = ["{SHARED}"] +[paths] +home = "{}" +[limits] +typing_every_ms = 100 +"#, + home.path().display() + ); + Config::parse(&text).unwrap() +} + +/// Wait until `gatewayd` has handled `post`: its id is in the state file's `recent`. The post is +/// queued by then, since the event loop finishes a post before it looks at ended turns. +fn wait_handled(home: &TempDir, post: &str) { + let until = Instant::now() + WAIT; + loop { + let state = read_state(home); + if state["recent"] + .as_array() + .is_some_and(|r| r.iter().any(|p| p == post)) + { + return; + } + assert!(Instant::now() < until, "{post} not handled: {state}"); + std::thread::sleep(Duration::from_millis(10)); + } +} + +fn lookups(mm: &FakeMm, user: &str) -> usize { + let path = format!("/api/v4/users/{user}"); + mm.calls() + .iter() + .filter(|(m, p)| m == "GET" && *p == path) + .count() +} + +#[test] +fn one_persons_message_has_their_username() { + let home = TempDir::new("speaker-one"); + let (mm, _running, turns, mut ws) = up(&home, Duration::ZERO); + ws.posted(&post(&id('p', 1), KYLE, DM, "", "hello there", 5), "D"); + let turn = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (turn.content.as_str(), turn.speaker.as_deref()), + ("hello there", Some("kyle")) + ); + assert_eq!(mm.wait_posts(1, WAIT).len(), 1); +} + +#[test] +fn a_username_is_looked_up_once() { + let home = TempDir::new("speaker-cache"); + let (mm, _running, turns, mut ws) = up(&home, Duration::ZERO); + ws.posted(&post(&id('p', 1), KYLE, DM, "", "first", 5), "D"); + ws.posted(&post(&id('p', 2), KYLE, DM, "", "second", 6), "D"); + let speakers = [ + turns.recv_timeout(WAIT).unwrap().speaker, + turns.recv_timeout(WAIT).unwrap().speaker, + ]; + assert_eq!( + speakers, + [Some("kyle".to_string()), Some("kyle".to_string())] + ); + assert_eq!(lookups(&mm, KYLE), 1, "{:?}", mm.calls()); +} + +#[test] +fn bursts_have_a_speaker_only_when_one_person_wrote_them() { + let home = TempDir::new("speaker-burst"); + loop_dir(&home); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let release = Mutex::new(release_rx); + let turns = serve_loop(&home.path().join("run/loop/loop.sock"), move |n, turn| { + if n <= 1 { + let _ = release.lock().unwrap().recv_timeout(WAIT); + } + vec![done(&format!("answer to {}", turn.content))] + }); + let mm = FakeMm::start(); + let _running = start(two_people(&home, &mm.url())); + let mut ws = mm.next_ws(WAIT); + let p1 = id('p', 1); + ws.posted( + &post(&p1, KYLE, SHARED, "", "@boxmaker-straylight start", 5), + "O", + ); + let first = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (first.content.as_str(), first.speaker.as_deref()), + ("start", Some("kyle")) + ); + + // Eve and Kyle while the first turn runs: one turn, from two people. + ws.posted(&post(&id('p', 2), EVE, SHARED, &p1, "two", 6), "O"); + ws.posted(&post(&id('p', 3), KYLE, SHARED, &p1, "three", 7), "O"); + wait_handled(&home, &id('p', 3)); + release_tx.send(()).unwrap(); + let second = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (second.content.as_str(), second.resume), + ("two\n\nthree", true), + "the text is joined as before" + ); + assert_eq!(second.speaker, None, "two people: no speaker"); + + // Eve alone while the second runs: her name, and nothing left over from the burst before. + ws.posted(&post(&id('p', 4), EVE, SHARED, &p1, "four", 8), "O"); + wait_handled(&home, &id('p', 4)); + release_tx.send(()).unwrap(); + let third = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (third.content.as_str(), third.speaker.as_deref()), + ("four", Some("user-e000")) + ); + assert_eq!(mm.wait_posts(3, WAIT).len(), 3); +} + +#[test] +fn a_name_that_cannot_be_found_leaves_the_turn_without_a_speaker() { + let home = TempDir::new("speaker-missing"); + let (mm, running, turns, mut ws) = up(&home, Duration::ZERO); + mm.refuse_users(true); + ws.posted(&post(&id('p', 1), KYLE, DM, "", "hello", 5), "D"); + let turn = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (turn.content.as_str(), turn.speaker), + ("hello", None), + "never the user id, never a guess" + ); + assert_eq!( + mm.wait_posts(1, WAIT).len(), + 1, + "the turn is answered all the same" + ); + let line = format!( + "gatewayd: no username for {KYLE}: status 404: \"{{\\\"message\\\":\\\"not found\\\"}}\"; \ + the turn has no speaker" + ); + let log = running.wait_log("no username for"); + assert_eq!( + log.iter() + .filter(|l| l.contains("no username")) + .collect::>(), + [&line] + ); + + // A failure is not remembered: the next turn looks again. + mm.refuse_users(false); + ws.posted(&post(&id('p', 2), KYLE, DM, "", "again", 6), "D"); + let again = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (again.content.as_str(), again.speaker.as_deref()), + ("again", Some("kyle")) + ); + assert_eq!(lookups(&mm, KYLE), 2, "{:?}", mm.calls()); +} diff --git a/docs/plans/M5c1/files/crates/gatewayd/tests/sessions.rs b/docs/plans/M5c1/files/crates/gatewayd/tests/sessions.rs new file mode 100644 index 0000000..3868b7b --- /dev/null +++ b/docs/plans/M5c1/files/crates/gatewayd/tests/sessions.rs @@ -0,0 +1,401 @@ +//! Which posts become turns, in which session, and how messages wait for a running turn (M4a spec, +//! section 7, including its table of examples), with the author of each (M5c spec, section 11). +//! Do not edit. + +use bxcore::SessionId; +use gatewayd::mm::Post; +use gatewayd::sessions::{ + ApprovalCommand, BUSY, Batch, Ignored, Message, Pushed, Queues, Route, Router, Thread, + UNKNOWN_COMMAND, named, +}; + +const BOT: &str = "b0000000000000000000000000"; +const KYLE: &str = "k0000000000000000000000000"; +const EVE: &str = "e0000000000000000000000000"; +const DM: &str = "d0000000000000000000000000"; +const SHARED: &str = "c0000000000000000000000000"; +const OTHER: &str = "o0000000000000000000000000"; +const ROOT: &str = "r0000000000000000000000000"; +const POST: &str = "p0000000000000000000000000"; + +fn router() -> Router { + Router::new( + BOT, + "boxmaker-straylight", + &[KYLE.to_string()], + &[SHARED.to_string()], + ) +} + +fn post(channel: &str, root: &str, message: &str) -> Post { + Post { + id: POST.to_string(), + user_id: KYLE.to_string(), + channel_id: channel.to_string(), + root_id: root.to_string(), + message: message.to_string(), + create_at: 5, + delete_at: 0, + kind: String::new(), + } +} + +fn known(root: &str) -> bool { + root == ROOT +} + +fn unknown(_: &str) -> bool { + false +} + +fn queued(route: Route) -> Message { + match route { + Route::Queue(m) => m, + other => panic!("not queued: {other:?}"), + } +} + +#[test] +fn the_examples_in_the_spec() { + let r = router(); + let cases = [ + ("", "@boxmaker-straylight summarise the audit log", true), + (ROOT, "and the older files?", true), + (ROOT, "@hermes what do you think?", false), + ("", "@boxmaker-straylight @hermes compare notes", true), + ("", "@boxmaker-straylightx hello", false), + ("", "@channel standup in five", false), + ]; + for (root, message, yes) in cases { + let got = r.route(&post(SHARED, root, message), "O", &known); + assert_eq!(matches!(got, Route::Queue(_)), yes, "{message}: {got:?}"); + if !yes { + assert_eq!(got, Route::Ignore(Ignored::NotForUs), "{message}"); + } + } +} + +#[test] +fn naming() { + assert_eq!(named("@Boxmaker-Straylight, look"), ["boxmaker-straylight"]); + assert_eq!(named("ask @boxmaker-straylight."), ["boxmaker-straylight"]); + assert_eq!(named("@a.b_c-d... and @e"), ["a.b_c-d", "e"]); + assert_eq!(named("@ alone, @@x, trailing @"), ["x"]); + assert_eq!(named("ünïcødé @ʙob @bob"), ["bob"]); + assert!(named("no names here").is_empty()); + let r = router(); + for message in [ + "hi @BOXMAKER-STRAYLIGHT", + "@boxmaker-straylight.", + "(@boxmaker-straylight)", + ] { + assert!( + matches!( + r.route(&post(SHARED, "", message), "P", &unknown), + Route::Queue(_) + ), + "{message}" + ); + } +} + +#[test] +fn replies_in_our_thread_that_name_everyone_are_still_ours() { + let r = router(); + for message in ["@here any news?", "@all done", "thanks @channel"] { + assert!( + matches!( + r.route(&post(SHARED, ROOT, message), "O", &known), + Route::Queue(_) + ), + "{message}" + ); + } + let got = r.route( + &post( + SHARED, + "q0000000000000000000000000", + "a reply in someone else's thread", + ), + "O", + &known, + ); + assert_eq!(got, Route::Ignore(Ignored::NotForUs)); +} + +#[test] +fn who_and_where() { + let r = router(); + let mut own = post(DM, "", "hello"); + own.user_id = BOT.to_string(); + assert_eq!(r.route(&own, "D", &unknown), Route::Ignore(Ignored::Own)); + let mut system = post(DM, "", "joined"); + system.kind = "system_join_channel".to_string(); + assert_eq!( + r.route(&system, "D", &unknown), + Route::Ignore(Ignored::System) + ); + let mut stranger = post(DM, "", "@boxmaker-straylight hello"); + stranger.user_id = EVE.to_string(); + assert_eq!( + r.route(&stranger, "D", &unknown), + Route::Ignore(Ignored::NotAllowed) + ); + let mut stranger_cmd = post(DM, "", "!approve 1"); + stranger_cmd.user_id = EVE.to_string(); + assert_eq!( + r.route(&stranger_cmd, "D", &unknown), + Route::Ignore(Ignored::NotAllowed) + ); + let naming = "@boxmaker-straylight hello"; + assert_eq!( + r.route(&post(OTHER, "", naming), "O", &unknown), + Route::Ignore(Ignored::NotForUs) + ); + assert_eq!( + r.route(&post(SHARED, "", naming), "X", &unknown), + Route::Ignore(Ignored::NotForUs) + ); + assert!(matches!( + r.route(&post(SHARED, "", naming), "G", &unknown), + Route::Queue(_) + )); + assert!(matches!( + r.route(&post(DM, "", "no name needed"), "D", &unknown), + Route::Queue(_) + )); +} + +#[test] +fn sessions_and_threads() { + let r = router(); + let top = queued(r.route(&post(DM, "", "hello"), "D", &unknown)); + assert_eq!(top.session.as_str(), format!("mm-{POST}")); + assert_eq!( + top.thread, + Thread { + channel: DM.to_string(), + root: POST.to_string() + } + ); + assert!(!top.resume && !top.joins_thread); + assert_eq!(top.text, "hello"); + assert_eq!(top.user, KYLE); + let reply = queued(r.route(&post(DM, ROOT, "more"), "D", &unknown)); + assert_eq!(reply.session.as_str(), format!("mm-{ROOT}")); + assert_eq!(reply.thread.root, ROOT); + assert!(reply.resume); + let channel = queued(r.route(&post(SHARED, "", "@boxmaker-straylight hi"), "O", &unknown)); + assert!(channel.joins_thread && !channel.resume); + assert_eq!(channel.text, "hi", "the leading mention is stripped (M4b)"); +} + +#[test] +fn commands() { + let r = router(); + let thread = Thread { + channel: DM.to_string(), + root: ROOT.to_string(), + }; + for (message, command) in [ + ("!approve 42", ApprovalCommand::Approve(42)), + ( + "!deny 42 not now", + ApprovalCommand::Deny(42, Some("not now".to_string())), + ), + ("!deny", ApprovalCommand::Malformed), + ( + "@boxmaker-straylight !approve 7", + ApprovalCommand::Approve(7), + ), + ] { + assert_eq!( + r.route(&post(DM, ROOT, message), "D", &unknown), + Route::Approval { + thread: thread.clone(), + command, + user: KYLE.to_string(), + post: POST.to_string(), + at: 5, + }, + "{message}" + ); + } + for message in ["!approved", "!help", "!"] { + assert_eq!( + r.route(&post(DM, ROOT, message), "D", &unknown), + Route::Reply { + thread: thread.clone(), + text: UNKNOWN_COMMAND.to_string(), + }, + "{message}" + ); + } + assert_eq!( + queued(r.route(&post(DM, ROOT, "!!approve is a word"), "D", &unknown)).text, + "!approve is a word" + ); + assert_eq!( + queued(r.route(&post(DM, ROOT, "!!"), "D", &unknown)).text, + "!" + ); + assert_eq!( + queued(r.route(&post(DM, ROOT, " !help"), "D", &unknown)).text, + " !help" + ); +} + +fn message(root: &str, resume: bool, text: &str) -> Message { + message_by(KYLE, root, resume, text) +} + +fn message_by(user: &str, root: &str, resume: bool, text: &str) -> Message { + Message { + session: SessionId::new(&format!("mm-{root}")).unwrap(), + thread: Thread { + channel: DM.to_string(), + root: root.to_string(), + }, + resume, + text: text.to_string(), + joins_thread: false, + user: user.to_string(), + } +} + +fn senders(batch: &Batch) -> Vec<&str> { + batch.senders.iter().map(String::as_str).collect() +} + +#[test] +fn messages_wait_for_the_running_turn_and_go_together() { + let mut q = Queues::new(3); + let first = message(ROOT, false, "one"); + let Pushed::Start(batch) = q.push(first.clone()) else { + panic!("not started") + }; + assert_eq!( + batch, + Batch { + session: first.session.clone(), + thread: first.thread.clone(), + resume: false, + text: "one".to_string(), + senders: vec![KYLE.to_string()], + } + ); + assert_eq!(q.push(message(ROOT, true, "two")), Pushed::Waiting); + assert_eq!(q.push(message(ROOT, true, "three\nlines")), Pushed::Waiting); + let other = message(POST, false, "elsewhere"); + assert!( + matches!(q.push(other.clone()), Pushed::Start(_)), + "another session starts at once" + ); + assert_eq!(q.running(), 2); + let mut roots: Vec = q.threads().into_iter().map(|t| t.root).collect(); + roots.sort(); + assert_eq!(roots, [POST, ROOT]); + let next = q.finish(&first.session).unwrap(); + assert_eq!( + (next.text.as_str(), next.resume), + ("two\n\nthree\nlines", true) + ); + assert_eq!(q.finish(&first.session), None); + assert_eq!(q.finish(&other.session), None); + assert_eq!(q.running(), 0); + assert!( + matches!(q.push(message(ROOT, true, "later")), Pushed::Start(_)), + "idle again" + ); +} + +#[test] +fn a_full_queue_drops_the_message() { + let mut q = Queues::new(2); + let m = message(ROOT, false, "run"); + assert!(matches!(q.push(m.clone()), Pushed::Start(_))); + assert_eq!(q.push(message(ROOT, true, "a")), Pushed::Waiting); + assert_eq!(q.push(message(ROOT, true, "b")), Pushed::Waiting); + assert_eq!( + q.push(message_by(EVE, ROOT, true, "c")), + Pushed::Full(m.thread.clone()) + ); + let next = q.finish(&m.session).unwrap(); + assert_eq!(next.text, "a\n\nb"); + assert_eq!( + senders(&next), + [KYLE, KYLE], + "the dropped message's author is not kept" + ); + assert_eq!(next.sender(), Some(KYLE)); + assert!(!BUSY.is_empty()); + assert_eq!(q.finish(&SessionId::new("mm-never").unwrap()), None); +} + +#[test] +fn a_queued_message_names_its_author() { + let r = Router::new( + BOT, + "boxmaker-straylight", + &[KYLE.to_string(), EVE.to_string()], + &[SHARED.to_string()], + ); + for (text, expected) in [ + ("hello", "hello"), + ("!!approve is a word", "!approve is a word"), + ] { + let mut from_eve = post(DM, ROOT, text); + from_eve.user_id = EVE.to_string(); + let m = queued(r.route(&from_eve, "D", &unknown)); + assert_eq!( + (m.user.as_str(), m.text.as_str()), + (EVE, expected), + "{text}" + ); + } +} + +#[test] +fn a_batch_carries_the_author_of_each_message() { + let mut q = Queues::new(5); + let Pushed::Start(first) = q.push(message(ROOT, false, "one")) else { + panic!("not started") + }; + assert_eq!((senders(&first), first.sender()), (vec![KYLE], Some(KYLE))); + assert_eq!(q.push(message_by(EVE, ROOT, true, "two")), Pushed::Waiting); + assert_eq!(q.push(message(ROOT, true, "three")), Pushed::Waiting); + let session = SessionId::new(&format!("mm-{ROOT}")).unwrap(); + let mixed = q.finish(&session).unwrap(); + assert_eq!(mixed.text, "two\n\nthree", "the text is joined as before"); + assert_eq!(senders(&mixed), [EVE, KYLE]); + assert_eq!(mixed.sender(), None, "two people: no one sender"); + assert_eq!(q.push(message_by(EVE, ROOT, true, "four")), Pushed::Waiting); + assert_eq!(q.push(message_by(EVE, ROOT, true, "five")), Pushed::Waiting); + let one = q.finish(&session).unwrap(); + assert_eq!( + (senders(&one), one.sender()), + (vec![EVE, EVE], Some(EVE)), + "each batch has only its own messages' authors" + ); + assert_eq!(q.finish(&session), None); +} + +#[test] +fn a_batch_with_no_authors_has_no_sender() { + let batch = Batch { + session: SessionId::new(&format!("mm-{ROOT}")).unwrap(), + thread: Thread { + channel: DM.to_string(), + root: ROOT.to_string(), + }, + resume: false, + text: "orphan".to_string(), + senders: Vec::new(), + }; + assert_eq!(batch.sender(), None); + let three = Batch { + senders: vec![KYLE.to_string(), KYLE.to_string(), EVE.to_string()], + ..batch + }; + assert_eq!(three.sender(), None, "the last one differs"); +} diff --git a/docs/plans/M5c1/files/crates/gatewayd/tests/support/fake_mm.rs b/docs/plans/M5c1/files/crates/gatewayd/tests/support/fake_mm.rs new file mode 100644 index 0000000..bf05d97 --- /dev/null +++ b/docs/plans/M5c1/files/crates/gatewayd/tests/support/fake_mm.rs @@ -0,0 +1,390 @@ +//! A fake Mattermost on 127.0.0.1, plain TCP: the four REST calls `gatewayd` makes, and the +//! WebSocket, whose events the test sends and whose requests it reads. Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream}; +use std::sync::{Arc, Mutex, mpsc}; +use std::time::{Duration, Instant}; + +use gatewayd::ws::handshake::accept_for; +use serde_json::{Value, json}; + +pub const BOT: &str = "b0000000000000000000000000"; +pub const BOT_NAME: &str = "boxmaker-straylight"; +pub const KYLE: &str = "k0000000000000000000000000"; +pub const EVE: &str = "e0000000000000000000000000"; +/// The direct channel between the bot and Kyle, and between the bot and anyone else. +pub const DM: &str = "d0000000000000000000000000"; +pub const EVE_DM: &str = "f0000000000000000000000000"; + +#[derive(Default)] +struct Inner { + /// Status for `users/me`: 200 unless a test sets another. + me_status: u16, + /// The body for `channels//posts?since=`, by channel. + since: HashMap, + /// Every post made: channel, root, message. + posts: Vec<(String, String, String)>, + /// Every REST call: method and path. + calls: Vec<(String, String)>, + /// Every edit: the post and its new text. + edits: Vec<(String, String)>, + /// When set, `POST /posts` is refused with 400. + refuse_posts: bool, + /// When set, `GET /users/` (not `users/me`) answers 404. + refuse_users: bool, +} + +pub struct FakeMm { + pub addr: SocketAddr, + inner: Arc>, + sockets: Mutex>, +} + +/// One WebSocket connection from `gatewayd`. +pub struct WsPeer { + writer: TcpStream, + /// The text of every text frame `gatewayd` sends. + pub texts: mpsc::Receiver, +} + +/// A post as Mattermost sends it. +pub fn post( + id: &str, + user: &str, + channel: &str, + root: &str, + message: &str, + create_at: i64, +) -> Value { + json!({ + "id": id, "create_at": create_at, "update_at": create_at, "delete_at": 0, "user_id": user, + "channel_id": channel, "root_id": root, "message": message, "type": "", "props": {} + }) +} + +pub fn id(prefix: char, n: u32) -> String { + format!("{prefix}{n:025}") +} + +fn frame(opcode: u8, payload: &[u8]) -> Vec { + let mut out = vec![0x80 | opcode]; + match payload.len() { + n if n < 126 => out.push(n as u8), + n => { + out.push(126); + out.extend_from_slice(&(n as u16).to_be_bytes()); + } + } + out.extend_from_slice(payload); + out +} + +impl WsPeer { + pub fn event(&mut self, value: &Value) { + let _ = self + .writer + .write_all(&frame(0x1, value.to_string().as_bytes())); + } + + pub fn posted(&mut self, post: &Value, channel_type: &str) { + let data = json!({"post": post.to_string(), "channel_type": channel_type, "team_id": ""}); + self.event(&json!({"event": "posted", "data": data, "broadcast": {}, "seq": 1})); + } + + /// End the connection without a close frame. + pub fn drop_connection(self) { + let _ = self.writer.shutdown(Shutdown::Both); + } + + /// The `user_typing` requests received within `wait`, as (channel, parent). + pub fn typing_within(&self, wait: Duration) -> Vec<(String, String)> { + let until = Instant::now() + wait; + let mut got = Vec::new(); + while let Ok(text) = self + .texts + .recv_timeout(until.saturating_duration_since(Instant::now())) + { + let v: Value = serde_json::from_str(&text).unwrap(); + if v["action"] == "user_typing" { + let data = &v["data"]; + got.push(( + data["channel_id"].as_str().unwrap().to_string(), + data["parent_id"].as_str().unwrap().to_string(), + )); + } + } + got + } +} + +fn read_head(stream: &mut TcpStream) -> Option { + let mut head = Vec::new(); + let mut byte = [0u8; 1]; + while !head.ends_with(b"\r\n\r\n") { + if stream.read(&mut byte).ok()? == 0 { + return None; + } + head.push(byte[0]); + } + String::from_utf8(head).ok() +} + +/// Unmask the client's frames and send each text on `tx`, until the connection ends. +fn read_frames(mut stream: TcpStream, tx: mpsc::Sender) { + let mut exact = |n: usize| -> Option> { + let mut buf = vec![0u8; n]; + stream.read_exact(&mut buf).ok().map(|()| buf) + }; + loop { + let Some(head) = exact(2) else { return }; + let len = match head[1] & 0x7F { + 126 => u16::from_be_bytes(exact(2).unwrap().try_into().unwrap()) as usize, + 127 => return, + n => n as usize, + }; + let Some(mask) = exact(4) else { return }; + let Some(raw) = exact(len) else { return }; + let payload: Vec = raw + .iter() + .zip(mask.iter().cycle()) + .map(|(b, m)| b ^ m) + .collect(); + if head[0] & 0x0F == 0x1 { + let _ = tx.send(String::from_utf8(payload).unwrap()); + } + } +} + +impl FakeMm { + pub fn start() -> FakeMm { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let inner = Arc::new(Mutex::new(Inner { + me_status: 200, + ..Inner::default() + })); + let (ws_tx, ws_rx) = mpsc::channel(); + let shared = Arc::clone(&inner); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(stream) = stream else { continue }; + let (inner, ws_tx) = (Arc::clone(&shared), ws_tx.clone()); + std::thread::spawn(move || connection(stream, &inner, &ws_tx)); + } + }); + FakeMm { + addr, + inner, + sockets: Mutex::new(ws_rx), + } + } + + pub fn url(&self) -> String { + format!("http://127.0.0.1:{}", self.addr.port()) + } + + pub fn refuse_token(&self) { + self.inner.lock().unwrap().me_status = 401; + } + + pub fn set_since(&self, channel: &str, posts: &[Value]) { + let order: Vec = posts.iter().map(|p| p["id"].clone()).collect(); + let map: serde_json::Map = posts + .iter() + .map(|p| (p["id"].as_str().unwrap().to_string(), p.clone())) + .collect(); + self.inner + .lock() + .unwrap() + .since + .insert(channel.to_string(), json!({"order": order, "posts": map})); + } + + /// The next WebSocket `gatewayd` opens, after its hello. + pub fn next_ws(&self, wait: Duration) -> WsPeer { + self.sockets + .lock() + .unwrap() + .recv_timeout(wait) + .expect("no WebSocket connection") + } + + pub fn posts(&self) -> Vec<(String, String, String)> { + self.inner.lock().unwrap().posts.clone() + } + + /// Wait until at least `n` posts were made, for at most `wait`. + pub fn wait_posts(&self, n: usize, wait: Duration) -> Vec<(String, String, String)> { + let until = Instant::now() + wait; + while self.posts().len() < n && Instant::now() < until { + std::thread::sleep(Duration::from_millis(10)); + } + self.posts() + } + + pub fn calls(&self) -> Vec<(String, String)> { + self.inner.lock().unwrap().calls.clone() + } + + /// The id of the n-th post made (from 0). + pub fn post_id(n: usize) -> String { + id('x', u32::try_from(n).unwrap()) + } + + pub fn edits(&self) -> Vec<(String, String)> { + self.inner.lock().unwrap().edits.clone() + } + + /// Wait until at least `n` edits were made, for at most `wait`. + pub fn wait_edits(&self, n: usize, wait: Duration) -> Vec<(String, String)> { + let until = Instant::now() + wait; + while self.edits().len() < n && Instant::now() < until { + std::thread::sleep(Duration::from_millis(10)); + } + self.edits() + } + + pub fn refuse_posts(&self, refuse: bool) { + self.inner.lock().unwrap().refuse_posts = refuse; + } + + pub fn refuse_users(&self, refuse: bool) { + self.inner.lock().unwrap().refuse_users = refuse; + } +} + +fn connection(mut stream: TcpStream, inner: &Mutex, ws_tx: &mpsc::Sender) { + let Some(head) = read_head(&mut stream) else { + return; + }; + let mut words = head.split_whitespace(); + let (method, path) = ( + words.next().unwrap_or("").to_string(), + words.next().unwrap_or("").to_string(), + ); + if path == "/api/v4/websocket" { + let key = head + .lines() + .find_map(|l| l.strip_prefix("Sec-WebSocket-Key: ")) + .unwrap_or("") + .trim() + .to_string(); + let reply = format!( + "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {}\r\n\r\n", + accept_for(&key) + ); + let _ = stream.write_all(reply.as_bytes()); + let _ = stream.write_all(&frame( + 0x1, + br#"{"event":"hello","data":{},"broadcast":{},"seq":0}"#, + )); + let (tx, texts) = mpsc::channel(); + let reader = stream.try_clone().unwrap(); + std::thread::spawn(move || read_frames(reader, tx)); + let _ = ws_tx.send(WsPeer { + writer: stream, + texts, + }); + return; + } + let length = head + .lines() + .find_map(|l| { + l.to_ascii_lowercase() + .strip_prefix("content-length:") + .map(|v| v.trim().parse::().unwrap_or(0)) + }) + .unwrap_or(0); + let mut body = vec![0u8; length]; + let _ = stream.read_exact(&mut body); + let (status, answer) = rest(inner, &method, &path, &body); + let text = answer.to_string(); + let reply = format!( + "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{text}", + text.len() + ); + let _ = stream.write_all(reply.as_bytes()); +} + +fn rest(inner: &Mutex, method: &str, path: &str, body: &[u8]) -> (u16, Value) { + let mut inner = inner.lock().unwrap(); + inner.calls.push((method.to_string(), path.to_string())); + match (method, path) { + ("GET", "/api/v4/users/me") if inner.me_status == 200 => { + (200, json!({"id": BOT, "username": BOT_NAME})) + } + ("GET", "/api/v4/users/me") => ( + inner.me_status, + json!({"id": "api.context.session_expired.app_error"}), + ), + ("POST", "/api/v4/channels/direct") => { + let users: Vec = serde_json::from_slice(body).unwrap(); + let channel = if users.iter().any(|u| u == KYLE) { + DM + } else { + EVE_DM + }; + (201, json!({"id": channel, "type": "D"})) + } + ("POST", "/api/v4/posts") if inner.refuse_posts => { + (400, json!({"id": "api.post.create_post.refused"})) + } + ("POST", "/api/v4/posts") => { + let p: Value = serde_json::from_slice(body).unwrap(); + let n = u32::try_from(inner.posts.len()).unwrap(); + let (channel, root, message) = ( + p["channel_id"].as_str().unwrap(), + p["root_id"].as_str().unwrap(), + p["message"].as_str().unwrap(), + ); + inner + .posts + .push((channel.to_string(), root.to_string(), message.to_string())); + (201, post(&id('x', n), BOT, channel, root, message, 1)) + } + ("GET", p) if p.contains("/posts?since=") => { + let channel = p + .trim_start_matches("/api/v4/channels/") + .split('/') + .next() + .unwrap_or(""); + ( + 200, + inner + .since + .get(channel) + .cloned() + .unwrap_or(json!({"order": [], "posts": {}})), + ) + } + ("PUT", p) if p.starts_with("/api/v4/posts/") && p.ends_with("/patch") => { + let post_id = p + .trim_start_matches("/api/v4/posts/") + .trim_end_matches("/patch") + .to_string(); + let message = serde_json::from_slice::(body).unwrap()["message"] + .as_str() + .unwrap() + .to_string(); + inner.edits.push((post_id.clone(), message.clone())); + (200, post(&post_id, BOT, DM, "", &message, 1)) + } + ("GET", p) if p.starts_with("/api/v4/users/") && inner.refuse_users => { + (404, json!({"message": "not found"})) + } + ("GET", p) if p.starts_with("/api/v4/users/") => { + let user = p.trim_start_matches("/api/v4/users/"); + let name = if user == KYLE { + "kyle".to_string() + } else { + format!("user-{}", &user[..4]) + }; + (200, json!({"id": user, "username": name})) + } + _ => (404, json!({"message": "not found"})), + } +}