kyle/boxmaker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

M5c-1 plan: task 03, gatewayd sets the speaker from the burst's one author

Browse Source 
A batch carries the Mattermost user id of each message; the turn's speaker
is the username of the one user who wrote all of it (Client::username,
kept in a map on the event loop), and None for two or more people or a
failed lookup. The turn's text is unchanged.

Checks: reference replayed from the task text on a fresh ref-02 worktree
with its own build dir, byte-identical; make gate ok with 1148 tests
(1138 + 10); /tmp unchanged around both gate runs; 21 mutations, all
caught; new tests stressed 40 times under load, no failure; every before
block found once at ref 02; no chat-template tags.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
kyle
2026-09-28 00:30:24 -07:00
co-authored by Claude Opus 5.5
parent 4e823a76e6
commit 5465946d98
8 changed files with 2330 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/M5c1/03-gatewayd-speaker.md
+613
View File
@@ -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<String>`, 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 <user id>: <the MmError>; 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/<id>` 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<String>`, 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<String, MmError>` (unchanged) |
| `Option::map` | `speaker.map(str::to_string)`: `Option<&str>` to `Option<String>` |
| `Option::as_deref` | `speaker.as_deref()`: `Option<String>` to `Option<&str>` |
| `Batch::sender` | `pub fn sender(&self) -> Option<&str>` (new, S4) |
| `Gateway::speaker` | `fn speaker(&mut self, batch: &Batch) -> Option<String>` (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<String>,
}
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<String>,
}
```
become
```rust
struct Pending {
thread: Thread,
waiting: Vec<String>,
/// The author of each message in `waiting`, in the same order.
senders: Vec<String>,
}
```
**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<String, String>,
}
```
**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), &notice);
```
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),
&notice,
);
```
**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<String> {
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.
docs/plans/M5c1/files/crates/gatewayd/tests/deliver.rs
+316
View File
@@ -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<Vec<(String, String, String)>>,
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<String> {
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<String> {
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()),
&notice,
);
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::<Vec<_>>(),
[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::<Vec<_>>()
);
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)]
);
}
docs/plans/M5c1/files/crates/gatewayd/tests/deliver_links.rs
+198
View File
@@ -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<Vec<String>>,
}
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<String> {
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()),
&notice,
);
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]);
}
docs/plans/M5c1/files/crates/gatewayd/tests/deliver_notice.rs
+103
View File
@@ -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<Vec<String>>);
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<String>, 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"]);
}
docs/plans/M5c1/files/crates/gatewayd/tests/deliver_speaker.rs
+121
View File
@@ -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<Vec<String>>);
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<fake_loop::Reply>,
) -> (Vec<Turn>, Vec<String>) {
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()),
&notice,
);
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<fake_loop::Reply> {
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"]);
}
docs/plans/M5c1/files/crates/gatewayd/tests/serve_speaker.rs
+188
View File
@@ -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::<Vec<_>>(),
[&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());
}
docs/plans/M5c1/files/crates/gatewayd/tests/sessions.rs
+401
View File
@@ -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<String> = 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");
}
docs/plans/M5c1/files/crates/gatewayd/tests/support/fake_mm.rs
+390
View File
@@ -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/<id>/posts?since=`, by channel.
since: HashMap<String, Value>,
/// 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/<id>` (not `users/me`) answers 404.
refuse_users: bool,
}
pub struct FakeMm {
pub addr: SocketAddr,
inner: Arc<Mutex<Inner>>,
sockets: Mutex<mpsc::Receiver<WsPeer>>,
}
/// One WebSocket connection from `gatewayd`.
pub struct WsPeer {
writer: TcpStream,
/// The text of every text frame `gatewayd` sends.
pub texts: mpsc::Receiver<String>,
}
/// 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<u8> {
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<String> {
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<String>) {
let mut exact = |n: usize| -> Option<Vec<u8>> {
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<u8> = 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<Value> = posts.iter().map(|p| p["id"].clone()).collect();
let map: serde_json::Map<String, Value> = 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<Inner>, ws_tx: &mpsc::Sender<WsPeer>) {
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::<usize>().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<Inner>, 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<String> = 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::<Value>(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"})),
}
}