//! What the event loop does with a post, a finished turn, and the time between: routing, starting //! turns, typing, and catching up after a gap. use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; use crate::deliver::{LOOP_DOWN, deliver}; use crate::mm::{Post, typing}; use crate::serve::{Gateway, Stop}; use crate::sessions::{BUSY, Batch, Ignored, Pushed, Route}; use crate::state::InFlight; use crate::ws::WsError; use crate::ws::conn::Ws; /// Now, in Mattermost's milliseconds. fn now_ms() -> i64 { // `let ms = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0);` // then `i64::try_from(ms).unwrap_or(i64::MAX)`. todo!() } impl Gateway { /// Post in a thread; a failure is logged, not fatal. pub(super) fn post(&self, channel: &str, root: &str, text: &str) { // `if let Err(e) = self.client.create_post(channel, root, text)`: log // "gatewayd: cannot post in (thread ): ", with `(self.log)(&line)`. // Nothing else: posting records nothing in the state. todo!() } /// Is this channel one whose posts `gatewayd` keeps track of? fn tracked(&self, channel: &str, channel_type: &str) -> bool { // `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. `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 from : 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> { // 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.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> { // `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 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. Written for you: it is the glue. pub(super) fn catch_up(&mut self) -> Result<(), Stop> { 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 : " and returns Ok(()). // 3. When `found.full`, log "gatewayd: : 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!() } }