M4a tasks 14 and 15: glue written, calls tabled, after task 14's session wrote nothing

Task 14's session read the crate to learn the APIs and planned `connect` in prose until it was
cut off. `connect`, `Gateway::new` and `catch_up` are now written; the task lists every call with
its signature and makes the copy and the failing test the first actions. Task 15's comment is
rewritten one step per item. Both checked to fail, then pass when filled from their comments.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

docs/plans/M4a/files/crates/gatewayd/src/main.rs

@@ -25,13 +25,20 @@ fn main() -> ExitCode {
 }
 
 fn serve(path: &Path) -> ExitCode {
-    // Each failure prints one line (plus its pointer) and returns ExitCode::from(1):
-    // 1. `Config::load`: "gatewayd: <error>\n<START_FAILED>". 2. `token_source`: "gatewayd: <path>:
-    //    <why>\n<START_FAILED>".
-    // 3. `secrets::load(MATTERMOST_TOKEN, &source, &|k| std::env::var_os(k))`: "gatewayd: <error>"
-    //    (it carries its pointer). Print the warning, if any, as it is. 4. Create the state file
-    //    directory, recursive, 0700: "gatewayd: cannot prepare <dir>: <e>\n<START_FAILED>". 5.
-    //    `run` with Tuning::default(), a log that prints each line to standard error, and a stop
-    //    flag that is never set; print the Stop it returns.
+    // Each failure prints its message with `eprintln!` and returns `ExitCode::from(1)`.
+    // 1. `let config = match Config::load(path) { ... }`: an Err(e) prints
+    //    "gatewayd: {e}\n{START_FAILED}".
+    // 2. `let source = match config.token_source() { ... }`: an Err(why) prints
+    //    "gatewayd: {}: {why}\n{START_FAILED}" with `path.display()`.
+    // 3. `let loaded = match secrets::load(MATTERMOST_TOKEN, &source, &|k| std::env::var_os(k))`:
+    //    an Err(e) prints "gatewayd: {e}" (the error carries its own pointer).
+    //    Then `if let Some(warning) = &loaded.warning { eprintln!("{warning}"); }`.
+    // 4. The state file's directory:
+    //    `let dir = config.state_path().parent().map(Path::to_path_buf).unwrap_or_default();`
+    //    `std::fs::DirBuilder::new().recursive(true).mode(0o700).create(&dir)`: an Err(e) prints
+    //    "gatewayd: cannot prepare {}: {e}\n{START_FAILED}" with `dir.display()`.
+    // 5. `let log: gatewayd::serve::Log = Arc::new(|line: &str| eprintln!("{line}"));`
+    //    `let stop = run(config, loaded.secret, Tuning::default(), log, &AtomicBool::new(false));`
+    //    `eprintln!("{stop}");` and `ExitCode::from(1)`.
     todo!()
 }

docs/plans/M4a/files/crates/gatewayd/src/serve/handle.rs

@@ -21,58 +21,101 @@ fn now_ms() -> i64 {
 impl Gateway {
     /// Post in a thread; a failure is logged, not fatal.
     pub(super) fn post(&self, channel: &str, root: &str, text: &str) {
-        // `self.client.create_post`; an error is logged, "gatewayd: cannot post in <channel>
-        // (thread <root>): <error>".
+        // `SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0)`, then
+        // `i64::try_from(ms).unwrap_or(i64::MAX)`.
         todo!()
     }
 
     /// Is this channel one whose posts `gatewayd` keeps track of?
     fn tracked(&self, channel: &str, channel_type: &str) -> bool {
-        // channel_type "D", or the channel is in allow.channels.
+        // `channel_type == "D" || self.config.allow.channels.iter().any(|c| c == channel)`.
         todo!()
     }
 
     /// One new post, live or caught up.
     pub(super) fn handle_post(&mut self, post: &Post, channel_type: &str) -> Result<(), Stop> {
-        // 1. Not tracked, or seen -> nothing. 2. `state.handled(...)?` before anything else. 3.
-        //    Route it: NotAllowed -> log "gatewayd: ignored post <id> from <user>: not allowed"
-        //    (never the message); other Ignore -> nothing; Reply -> post it; Queue -> join the
-        //    thread when joins_thread, then push: Start -> `start`, Waiting -> nothing, Full ->
-        //    post BUSY.
+        // 1. `if !self.tracked(&post.channel_id, channel_type) || self.state.seen(&post.id)`:
+        //    return Ok(()).
+        // 2. `self.state.handled(&post.id, &post.channel_id, post.create_at)?;` before anything
+        //    else: after a crash, a post is never answered twice.
+        // 3. Route it, with the state answering "is this thread ours?":
+        //    `let state = &self.state;`
+        //    `let route = self.router.route(post, channel_type, &|root| state.knows_thread(root));`
+        //    then `match route`:
+        //    - Route::Ignore(Ignored::NotAllowed): log
+        //      "gatewayd: ignored post <post.id> from <post.user_id>: not allowed"
+        //      (never the text).
+        //    - Route::Ignore(_): nothing.
+        //    - Route::Reply { thread, text }: `self.post(&thread.channel, &thread.root, &text)`.
+        //    - Route::Queue(message): if `message.joins_thread`,
+        //      `self.state.join_thread(&message.thread.root)?`; then
+        //      `match self.queues.push(message)`:
+        //      Pushed::Start(batch) -> `self.start(batch)?`; Pushed::Waiting -> nothing;
+        //      Pushed::Full(thread) -> `self.post(&thread.channel, &thread.root, BUSY)`.
+        // 4. Ok(()).
         todo!()
     }
 
     /// Record the turn as in flight and run it on its own thread.
     fn start(&mut self, batch: Batch) -> Result<(), Stop> {
-        // `state.start_turn` (session, channel, root). Spawn with std::thread::Builder: `deliver`
-        // with the client, the loop socket, the batch and the log, then send the session on
-        // done_tx. If spawning fails: log "gatewayd: cannot start a thread for <session>: <e>",
-        // post LOOP_DOWN in the thread, and send the session on done_tx.
+        // 1. `self.state.start_turn(InFlight { session: batch.session.as_str().to_string(),
+        //    channel: batch.thread.channel.clone(), root: batch.thread.root.clone() })?;`
+        // 2. Clone what the thread takes, before the `move`:
+        //    `let (client, socket, log, done) = (Arc::clone(&self.client),
+        //    self.loop_socket.clone(), Arc::clone(&self.log), self.done_tx.clone());`
+        //    `let (session, thread) = (batch.session.clone(), batch.thread.clone());`
+        // 3. `let spawned = std::thread::Builder::new().spawn(move || {
+        //    deliver(client.as_ref(), &socket, &batch, &|line| log(line));
+        //    let _ = done.send(batch.session); });`
+        // 4. `if let Err(e) = spawned`: log "gatewayd: cannot start a thread for <session>: <e>"
+        //    (`session.as_str()`), `self.post(&thread.channel, &thread.root, LOOP_DOWN)`, and
+        //    `let _ = self.done_tx.send(session);` so the session does not stay busy.
+        // 5. Ok(()).
         todo!()
     }
 
     /// Turns that ended: out of flight, and the next batch of each session started.
     pub(super) fn finished(&mut self) -> Result<(), Stop> {
-        // For each session on done_rx (try_recv, never blocking): `end_turn`, then `queues.finish`;
-        // a batch it returns is started.
+        // `while let Ok(session) = self.done_rx.try_recv() {` (never blocks)
+        //    `self.state.end_turn(session.as_str())?;`
+        //    `if let Some(batch) = self.queues.finish(&session) { self.start(batch)?; }` `}`
+        // Then Ok(()).
         todo!()
     }
 
     /// Show this bot as typing in every thread with a turn running.
     pub(super) fn typing(&mut self, ws: &mut Ws) -> Result<(), WsError> {
-        // For each of `queues.threads()`: seq += 1, send `typing(seq, channel, root)` as text.
+        // `for thread in self.queues.threads() { self.seq += 1;
+        //    ws.send_text(&typing(self.seq, &thread.channel, &thread.root))?; }` then Ok(()).
         todo!()
     }
 
     /// Answer what arrived while `gatewayd` was away: the direct channel with each allowed user,
-    /// and each allowed channel. A channel seen for the first time starts from now.
+    /// and each allowed channel. Written for you: it is the glue.
     pub(super) fn catch_up(&mut self) -> Result<(), Stop> {
-        // Channels: the direct channel with each allowed user ("D"; an error is logged "gatewayd:
-        // no direct channel with <user>: <e>" and skipped), then each allowed channel ("O"). For
-        // each: no mark -> `mark(channel, now_ms())` and skip; else `posts_since(channel, mark)`
-        // (an error is logged "gatewayd: cannot catch up <channel>: <e>" and skipped); when full,
-        // log "gatewayd: <channel>: too many posts to catch up; some may be missed"; `handle_post`
-        // each post in order.
+        let mut channels = Vec::new();
+        for user in self.config.allow.users.clone() {
+            match self.client.direct_channel(&self.me.id, &user) {
+                Ok(id) => channels.push((id, "D")),
+                Err(e) => (self.log)(&format!("gatewayd: no direct channel with {user}: {e}")),
+            }
+        }
+        channels.extend(self.config.allow.channels.iter().map(|c| (c.clone(), "O")));
+        for (channel, channel_type) in channels {
+            self.catch_up_channel(&channel, channel_type)?;
+        }
+        Ok(())
+    }
+
+    /// One channel's posts since its mark. A channel seen for the first time starts from now.
+    fn catch_up_channel(&mut self, channel: &str, channel_type: &str) -> Result<(), Stop> {
+        // 1. `let Some(since) = self.state.since(channel) else { ... }`: with no mark,
+        //    `self.state.mark(channel, now_ms())?` and return Ok(()).
+        // 2. `self.client.posts_since(channel, since)`: an Err(e) is logged
+        //    "gatewayd: cannot catch up <channel>: <e>" and returns Ok(()).
+        // 3. When `found.full`, log "gatewayd: <channel>: too many posts to catch up; some may be
+        //    missed".
+        // 4. `for post in &found.posts { self.handle_post(post, channel_type)?; }` and Ok(()).
         todo!()
     }
 }

docs/plans/M4a/files/crates/gatewayd/src/serve/mod.rs

@@ -60,7 +60,7 @@ impl std::fmt::Display for Stop {
 
 impl From<StateError> for Stop {
     fn from(e: StateError) -> Stop {
-        // Stop::State(e).
+        // `Stop::State(e)`.
         todo!()
     }
 }
@@ -107,14 +107,16 @@ pub(crate) struct Gateway {
 
 /// The wait before attempt `n` (from 0) after a loss.
 pub fn backoff(tuning: &Tuning, n: usize) -> Duration {
-    // tuning.backoff[n], or its last entry when n is past the end (30 s if the list is empty). No
-    // indexing.
+    // `let last = tuning.backoff.last().copied().unwrap_or(Duration::from_secs(30));` then
+    // `tuning.backoff.get(n).copied().unwrap_or(last)`.
     todo!()
 }
 
 /// Sleep for `d`, in short steps, unless `stop` is set.
 fn sleep_unless(stop: &AtomicBool, d: Duration) {
-    // Sleep in steps of at most 20 ms until `d` has passed or `stop` is set.
+    // `let until = Instant::now() + d;` then, while `!stop.load(Ordering::SeqCst)` and
+    // `Instant::now() < until`, sleep
+    // `Duration::from_millis(20).min(until.saturating_duration_since(Instant::now()))`.
     todo!()
 }
 
@@ -167,28 +169,69 @@ pub fn run(config: Config, token: Secret, tuning: Tuning, log: Log, stop: &Atomi
 
 /// Who we are, and a WebSocket that has said hello.
 fn connect(config: &Config, client: &Client, tuning: &Tuning) -> Result<(Me, Ws), MmError> {
-    // 1. `client.me()?`. 2. Timing from limits (ping_every_ms, dead_after_ms). 3. Open /dev/urandom
-    //    (an error is Net("/dev/urandom: <e>")). 4. `Ws::open(client.connector(),
-    //    client.token().expose(), timing, ..)`.
-    // 5. Poll with tuning.poll until a text that `parse_event`s to Hello, for at most dead_after.
-    //    Every WebSocket error, and no hello in time, is Net("websocket: <why>").
-    todo!()
+    let me = client.me()?;
+    let timing = Timing {
+        ping_every: Duration::from_millis(config.limits.ping_every_ms),
+        dead_after: Duration::from_millis(config.limits.dead_after_ms),
+    };
+    let random =
+        File::open("/dev/urandom").map_err(|e| MmError::Net(format!("/dev/urandom: {e}")))?;
+    let net = |e: String| MmError::Net(format!("websocket: {e}"));
+    let mut ws = Ws::open(
+        client.connector(),
+        client.token().expose(),
+        timing,
+        Box::new(random),
+    )
+    .map_err(|e| net(e.to_string()))?;
+    // The server says hello first; anything before it is skipped, and silence is an error.
+    let until = Instant::now() + timing.dead_after;
+    while Instant::now() < until {
+        if let Some(text) = ws.poll(tuning.poll).map_err(|e| net(e.to_string()))?
+            && matches!(parse_event(&text), Ok(Event::Hello))
+        {
+            return Ok((me, ws));
+        }
+    }
+    Err(net("no hello from the server".to_string()))
 }
 
 impl Gateway {
     fn new(config: Config, client: Arc<Client>, state: State, log: Log) -> Gateway {
-        // A done channel, a Router with empty ids (set on connect), an empty Me, loop_socket from
-        // config, Queues with limit limits.queue (usize::try_from), seq 0, restarted false.
-        todo!()
+        let (done_tx, done_rx) = mpsc::channel();
+        let limit = usize::try_from(config.limits.queue).unwrap_or(usize::MAX);
+        Gateway {
+            // Both are set on each connection, from `users/me`.
+            router: Router::new("", "", &[], &[]),
+            me: Me {
+                id: String::new(),
+                username: String::new(),
+            },
+            loop_socket: config.loop_socket(),
+            queues: Queues::new(limit),
+            config,
+            client,
+            state,
+            done_tx,
+            done_rx,
+            log,
+            seq: 0,
+            restarted: false,
+        }
     }
 
     /// A connection is up: say so, route as this user, and on the first one, answer the turns a
     /// restart cut off.
     fn connected(&mut self, me: Me) -> Result<(), Stop> {
-        // Log exactly "gatewayd: connected to <url> as <username>". A new Router from me and the
-        // allow lists; store me. The first time only (`restarted`): for each turn `take_in_flight`
-        // gives back, post INTERRUPTED in its channel and root. A later reconnect must not: those
-        // turns are still running.
+        // 1. Log exactly "gatewayd: connected to <url> as <username>"
+        //    (`self.config.mattermost.url`, `me.username`), with `(self.log)(&line)`.
+        // 2. `let allow = &self.config.allow;`
+        //    `self.router = Router::new(&me.id, &me.username, &allow.users, &allow.channels);`
+        //    then `self.me = me;`.
+        // 3. The first time only (`if !self.restarted { self.restarted = true; ... }`): for each
+        //    `turn` in `self.state.take_in_flight()?`, `self.post(&turn.channel, &turn.root,
+        //    INTERRUPTED)`. A later reconnect must not: those turns are still running.
+        // 4. Ok(()).
         todo!()
     }