presence indicator

2 files changed, 117 insertions, 10 deletions

diff --git a/src/listener.rs b/src/listener.rs
index ca69466..31c134d 100644
--- a/src/listener.rs
+++ b/src/listener.rs
@@ -1,7 +1,7 @@
 use tokio::sync::mpsc::Sender;
 use twilight_http::Client;
 use twilight_model::{channel::Message, id::{Id, marker::{ChannelMarker, MessageMarker, UserMarker}}, user::User};
-use twilight_gateway::{error::ReceiveMessageError, Intents, Shard, ShardId};
+use twilight_gateway::{error::ReceiveMessageError, Intents, MessageSender, Shard, ShardId};
 use twilight_model::util::Timestamp;
 
 pub struct Listener {
@@ -21,14 +21,18 @@ impl Listener {
         let intents = Intents::GUILD_MEMBERS | Intents::GUILD_PRESENCES | Intents::GUILD_MESSAGES | Intents::MESSAGE_CONTENT;
 
         let mut shard = Shard::new(ShardId::ONE, self.config.discord_token.clone(), intents);
-        let mut client = Client::new(self.config.discord_token.clone());
-        
+        let client = Client::new(self.config.discord_token.clone());
+
         loop {
             match shard.next_event().await {
                 Ok(event) => {
                     match event {
                         twilight_gateway::Event::Ready(client) => {
                             println!("Bot started for {}#{}", client.user.name, client.user.discriminator);
+                            channel.send(ClientEvent::Ready {
+                                client_name: self.config.name.clone(),
+                                send_channel: shard.sender(),
+                            }).await;
                         },
 
                         twilight_gateway::Event::MessageCreate(message_create) => {
@@ -96,6 +100,13 @@ impl Listener {
 }
 
 pub enum ClientEvent {
+    Ready {
+        client_name: String,
+        send_channel: MessageSender,
+    },
+    AutoproxyTimeout {
+        last_message: Timestamp,
+    },
     Message {
         event_time: Timestamp,
         message: Message,
diff --git a/src/system.rs b/src/system.rs
index e3882ad..69d8b0e 100644
--- a/src/system.rs
+++ b/src/system.rs
@@ -1,9 +1,10 @@
-use std::{collections::HashMap, num::NonZeroUsize, str::FromStr};
+use std::{collections::HashMap, num::NonZeroUsize, str::FromStr, time::Duration};
 
-use tokio::sync::mpsc::channel;
+use tokio::{sync::mpsc::{channel, Receiver, Sender}, time::{Sleep,sleep}};
 use futures::future::join_all;
+use twilight_gateway::MessageSender;
 use twilight_http::Client;
-use twilight_model::{channel::{message::MessageType, Message}, id::{Id, marker::{ChannelMarker, MessageMarker, UserMarker}}};
+use twilight_model::{channel::{message::MessageType, Message}, gateway::{payload::outgoing::{update_presence::UpdatePresencePayload, UpdatePresence}, presence::Status, OpCode}, id::{Id, marker::{ChannelMarker, MessageMarker, UserMarker}}};
 use twilight_model::util::Timestamp;
 use twilight_model::http::attachment::Attachment;
 
@@ -15,6 +16,9 @@ pub struct System {
     pub message_dedup_cache: lru::LruCache<Id<MessageMarker>, Timestamp>,
     pub clients: HashMap<MemberName, Client>,
     pub latch_state: Option<(Member, Timestamp)>,
+    pub channel: (Sender<ClientEvent>, Receiver<ClientEvent>),
+    pub autoproxy_timeout: Option<Timestamp>,
+    pub gateway_channels: HashMap<MemberName, MessageSender>,
 }
 
 impl System {
@@ -25,14 +29,15 @@ impl System {
             message_dedup_cache: lru::LruCache::new(NonZeroUsize::new(100).unwrap()),
             clients: HashMap::new(),
             latch_state: None,
+            autoproxy_timeout: None,
+            channel: channel::<ClientEvent>(100),
+            gateway_channels: HashMap::new(),
         }
     }
 
     pub async fn start_clients(&mut self) {
         println!("Starting clients for system {}", self.name);
 
-        let (tx, mut rx) = channel::<ClientEvent>(100);
-
         let reference_user_id : Id<UserMarker> = Id::from_str(self.config.reference_user_id.as_str())
             .expect(format!("Invalid user ID: {}", self.config.reference_user_id).as_str());
 
@@ -40,7 +45,7 @@ impl System {
             let client = twilight_http::Client::new(member.discord_token.clone());
             self.clients.insert(member.name.clone(), client);
 
-            let tx = tx.clone();
+            let tx = self.channel.0.clone();
             let member = member.clone();
             tokio::spawn(async move {
                 let mut listener = Listener::new(member, reference_user_id);
@@ -49,8 +54,11 @@ impl System {
         }
 
         loop {
-            match rx.recv().await {
+            match self.channel.1.recv().await {
                 Some(event) => match event {
+                    ClientEvent::Ready { client_name, send_channel } => {
+                        self.gateway_channels.insert(client_name, send_channel);
+                    },
                     ClientEvent::Message { event_time, message } => {
                         if self.is_new_message(message.id, event_time) {
                             self.handle_message(message, event_time).await;
@@ -60,6 +68,14 @@ impl System {
                         println!("Client ran into an error for system {}", self.name);
                         return
                     },
+                    ClientEvent::AutoproxyTimeout { last_message } => {
+                        if let Some(current_last_message) = self.autoproxy_timeout {
+                            if current_last_message == last_message {
+                                self.latch_state = None;
+                                self.update_presence().await;
+                            }
+                        }
+                    },
                 },
                 None => {
                     return
@@ -108,6 +124,7 @@ impl System {
         if let Some((member, matched_content)) = match_prefix {
             self.proxy_message(&message, member, matched_content).await;
             self.update_autoproxy_state_after_message(member.clone(), timestamp);
+            self.update_presence().await;
             return
         }
 
@@ -126,6 +143,7 @@ impl System {
                         if time_since_last <= (*timeout_seconds).into() {
                             self.proxy_message(&message, &member, message.content.as_str()).await;
                             self.latch_state = Some((member.clone(), timestamp));
+                            self.update_presence().await;
                         }
                     }
                 },
@@ -178,6 +196,84 @@ impl System {
             Some(AutoproxyConfig::Member { name }) => (),
             Some(AutoproxyConfig::Latch { scope, timeout_seconds, presence_indicator }) => {
                 self.latch_state = Some((member.clone(), timestamp));
+                self.autoproxy_timeout = Some(timestamp);
+
+                let tx = self.channel.0.clone();
+                let last_message = timestamp.clone();
+                let timeout_seconds = timeout_seconds.clone();
+
+                tokio::spawn(async move {
+                    sleep(Duration::from_secs(timeout_seconds.into())).await;
+                    tx.send(ClientEvent::AutoproxyTimeout { last_message })
+                        .await.expect("Channel has closed");
+                });
+            }
+        }
+    }
+
+    async fn update_presence(&mut self) {
+        match &self.config.autoproxy {
+            None => (),
+            Some(AutoproxyConfig::Member { name }) => {
+                for member in &self.config.members {
+                    let gateway_channel = self.gateway_channels.get(&member.name).expect("No gateway shard for member");
+
+                    println!("Updating {} to {}", member.name, if member.name == *name { "online" } else { "offline" });
+
+                    gateway_channel.command(&UpdatePresence {
+                        d: UpdatePresencePayload {
+                            activities: Vec::new(),
+                            afk: false,
+                            since: None,
+                            status: if member.name == *name {
+                                Status::Online
+                            } else {
+                                Status::Invisible
+                            },
+                        },
+                        op: OpCode::PresenceUpdate,
+                    }).expect("Could not send command to gateway");
+                }
+            },
+            Some(AutoproxyConfig::Latch { scope, timeout_seconds, presence_indicator }) => {
+                if let Some((member, last_timestamp)) = &self.latch_state {
+                    let name = &member.name;
+                    for member in &self.config.members {
+                        let gateway_channel = self.gateway_channels.get(&member.name).expect("No gateway shard for member");
+
+                        println!("Updating {} to {}", member.name, if member.name == *name { "online" } else { "offline" });
+
+                        gateway_channel.command(&UpdatePresence {
+                            d: UpdatePresencePayload {
+                                activities: Vec::new(),
+                                afk: false,
+                                since: None,
+                                status: if member.name == *name {
+                                    Status::Online
+                                } else {
+                                    Status::Invisible
+                                },
+                            },
+                            op: OpCode::PresenceUpdate,
+                        }).expect("Could not send command to gateway");
+                    }
+                } else {
+                    for member in &self.config.members {
+                        let gateway_channel = self.gateway_channels.get(&member.name).expect("No gateway shard for member");
+
+                        println!("Updating {} to offline", member.name);
+
+                        gateway_channel.command(&UpdatePresence {
+                            d: UpdatePresencePayload {
+                                activities: Vec::new(),
+                                afk: false,
+                                since: None,
+                                status: Status::Invisible,
+                            },
+                            op: OpCode::PresenceUpdate,
+                        }).expect("Could not send command to gateway");
+                    }
+                }
             }
         }
     }