guild member join and remove

bulbbot.Gateway/src/events/event_handler.rs

@@ -73,7 +73,9 @@ impl EventHandler for Handler {
     }
 
     // Dispatched when a user joins a guild.
-    async fn guild_member_addition(&self, _ctx: Context, _new_member: Member) {}
+    async fn guild_member_addition(&self, ctx: Context, new_member: Member) {
+        self.handle_guild_member_addition(ctx, new_member).await;
+    }
 
     /// Dispatched when a member is updated (e.g their nickname is updated).
     async fn guild_member_update(
@@ -88,11 +90,12 @@ impl EventHandler for Handler {
     // Dispatched when a user's membership ends by leaving, getting kicked, or being banned.
     async fn guild_member_removal(
         &self,
-        _ctx: Context,
-        _guild_id: GuildId,
-        _user: User,
+        ctx: Context,
+        guild_id: GuildId,
+        user: User,
         _member_data_if_available: Option<Member>,
     ) {
+        self.handle_guild_member_removal(ctx, guild_id, user).await;
     }
 
     // Dispatched when a channel is created.

bulbbot.Gateway/src/events/handler/guild_member_addition.rs

@@ -0,0 +1,82 @@
+use crate::events::event_handler::Handler;
+use crate::events::models::event::Event;
+use crate::manger_container_structs::RabbitMQMangerContainer;
+use crate::rabbit_mq::RabbitMqInjector;
+use lapin::types::FieldTable;
+use lapin::{options::BasicPublishOptions, BasicProperties};
+use opentelemetry::global::ObjectSafeSpan;
+use opentelemetry::trace::{SpanKind, Status, TraceContextExt};
+use opentelemetry::KeyValue;
+use opentelemetry::{global, trace::Tracer};
+use serde::{Deserialize, Serialize};
+use serenity::all::Member;
+use serenity::prelude::Context;
+use tracing::debug;
+
+#[derive(Serialize, Deserialize)]
+pub struct GuildMemberAdditionEvent {
+    pub event: Event,
+    pub shard_id: u32,
+    pub timestamp: u64,
+    pub content: Member,
+}
+
+impl Handler {
+    pub async fn handle_guild_member_addition(&self, ctx: Context, member: Member) {
+        let tracer = global::tracer(String::new());
+
+        let mut span = tracer
+            .span_builder("guild_member_addition")
+            .with_kind(SpanKind::Producer)
+            .start(&tracer);
+
+        span.set_attribute(KeyValue::new("guild_id", member.guild_id.to_string()));
+        span.set_attribute(KeyValue::new("user_id", member.user.id.to_string()));
+        span.set_attribute(KeyValue::new("shard_id", ctx.shard_id.0.to_string()));
+
+        let cx = opentelemetry::Context::current_with_span(span);
+
+        let mut headers = FieldTable::default();
+        global::get_text_map_propagator(|propagator| {
+            propagator.inject_context(&cx, &mut RabbitMqInjector(&mut headers))
+        });
+
+        let data = ctx.clone();
+        let data_read = data.data.read().await;
+
+        let channel = data_read
+            .get::<RabbitMQMangerContainer>()
+            .expect("[EVENT/GUILD_MEMBER_ADDITION] failed to get the Rabbit MQ Channel");
+
+        let event = GuildMemberAdditionEvent {
+            event: Event::GuildMemberAddition,
+            shard_id: ctx.shard_id.0,
+            timestamp: Handler::get_unix_time(),
+            content: member,
+        };
+        let serialized = serde_json::to_string(&event)
+            .expect("[EVENT/GUILD_MEMBER_ADDITION] failed to serialize event");
+
+        let payload = serialized.as_bytes();
+
+        let confirm = channel
+            .basic_publish(
+                "",
+                "bulbbot.gateway",
+                BasicPublishOptions::default(),
+                payload,
+                BasicProperties::default().with_headers(headers),
+            )
+            .await
+            .expect("[EVENT/GUILD_MEMBER_ADDITION] failed to publish to channel")
+            .await
+            .expect(
+                "[EVENT/GUILD_MEMBER_ADDITION] failed to get confirmation message from channel",
+            );
+
+        debug!("Rabbit MQ channel publish return message: {:#?}", confirm);
+
+        cx.span().set_status(Status::Ok);
+        cx.span().end();
+    }
+}

bulbbot.Gateway/src/events/handler/guild_member_removal.rs

@@ -0,0 +1,86 @@
+use crate::events::event_handler::Handler;
+use crate::events::models::event::Event;
+use crate::manger_container_structs::RabbitMQMangerContainer;
+use crate::rabbit_mq::RabbitMqInjector;
+use lapin::types::FieldTable;
+use lapin::{options::BasicPublishOptions, BasicProperties};
+use opentelemetry::global::ObjectSafeSpan;
+use opentelemetry::trace::{SpanKind, Status, TraceContextExt};
+use opentelemetry::KeyValue;
+use opentelemetry::{global, trace::Tracer};
+use serde::{Deserialize, Serialize};
+use serenity::all::{GuildId, User};
+use serenity::prelude::Context;
+use tracing::debug;
+
+#[derive(Serialize, Deserialize)]
+pub struct GuildMemberRemovalEventContent {
+    pub guild_id: GuildId,
+    pub user: User,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct GuildMemberRemovalEvent {
+    pub event: Event,
+    pub shard_id: u32,
+    pub timestamp: u64,
+    pub content: GuildMemberRemovalEventContent,
+}
+
+impl Handler {
+    pub async fn handle_guild_member_removal(&self, ctx: Context, guild_id: GuildId, user: User) {
+        let tracer = global::tracer(String::new());
+
+        let mut span = tracer
+            .span_builder("guild_member_removal")
+            .with_kind(SpanKind::Producer)
+            .start(&tracer);
+
+        span.set_attribute(KeyValue::new("guild_id", guild_id.to_string()));
+        span.set_attribute(KeyValue::new("user_id", user.id.to_string()));
+        span.set_attribute(KeyValue::new("shard_id", ctx.shard_id.0.to_string()));
+
+        let cx = opentelemetry::Context::current_with_span(span);
+
+        let mut headers = FieldTable::default();
+        global::get_text_map_propagator(|propagator| {
+            propagator.inject_context(&cx, &mut RabbitMqInjector(&mut headers))
+        });
+
+        let data = ctx.clone();
+        let data_read = data.data.read().await;
+
+        let channel = data_read
+            .get::<RabbitMQMangerContainer>()
+            .expect("[EVENT/GUILD_MEMBER_REMOVAL] failed to get the Rabbit MQ Channel");
+
+        let event = GuildMemberRemovalEvent {
+            event: Event::GuildMemberAddition,
+            shard_id: ctx.shard_id.0,
+            timestamp: Handler::get_unix_time(),
+            content: GuildMemberRemovalEventContent { guild_id, user },
+        };
+        let serialized = serde_json::to_string(&event)
+            .expect("[EVENT/GUILD_MEMBER_REMOVAL] failed to serialize event");
+
+        let payload = serialized.as_bytes();
+
+        let confirm = channel
+            .basic_publish(
+                "",
+                "bulbbot.gateway",
+                BasicPublishOptions::default(),
+                payload,
+                BasicProperties::default().with_headers(headers),
+            )
+            .await
+            .expect("[EVENT/GUILD_MEMBER_REMOVAL] failed to publish to channel")
+            .await
+            .expect("[EVENT/GUILD_MEMBER_REMOVAL] failed to get confirmation message from channel");
+
+        debug!("Rabbit MQ channel publish return message: {:#?}", confirm);
+
+        cx.span().set_status(Status::Ok);
+        cx.span().end();
+    }
+}

bulbbot.Gateway/src/events/handler/mod.rs

@@ -1,4 +1,6 @@
 pub mod event_utills;
+pub mod guild_member_addition;
+pub mod guild_member_removal;
 pub mod message;
 pub mod message_delete;
 pub mod message_update;

bulbbot.GatewayHandler/src/events/guild_member_addition.rs

@@ -0,0 +1,52 @@
+use crate::injector::ReqwestInjector;
+use crate::{handler::Handler, models::event_type::EventType};
+use opentelemetry::{global, Context};
+use opentelemetry::{global::BoxedSpan, trace::Span};
+use serde::{Deserialize, Serialize};
+use serenity::all::Member;
+use tracing::{debug, error};
+
+#[derive(Serialize, Deserialize)]
+pub struct GuildMemberAdditionEvent {
+    pub event: EventType,
+    pub shard_id: u32,
+    pub timestamp: u64,
+    pub content: Member,
+}
+
+impl Handler {
+    pub async fn handle_guild_member_addition_event(
+        &self,
+        event_data: &str,
+        span: &mut BoxedSpan,
+        cx: &Context,
+    ) -> bool {
+        let message: GuildMemberAdditionEvent =
+            serde_json::from_str(event_data).expect("Failed to parse data as message delete event");
+
+        let url = format!("{}/guilds/member/add", self.get_url(&message.event));
+        span.add_event(format!("Sending post request to {}", url), Vec::new());
+
+        let mut request = self.client.post(&url).json(&message).build().unwrap();
+        global::get_text_map_propagator(|propagator| {
+            propagator.inject_context(
+                cx,
+                &mut ReqwestInjector {
+                    headers: request.headers_mut(),
+                },
+            )
+        });
+        let response = self.client.execute(request).await;
+
+        match response {
+            Ok(_) => {
+                debug!("Sucessful request to {}", url);
+                true
+            }
+            Err(err) => {
+                error!("Request to {} errored with {:#?}", url, err);
+                false
+            }
+        }
+    }
+}

bulbbot.GatewayHandler/src/events/guild_member_removal.rs

@@ -0,0 +1,58 @@
+use crate::injector::ReqwestInjector;
+use crate::{handler::Handler, models::event_type::EventType};
+use opentelemetry::{global, Context};
+use opentelemetry::{global::BoxedSpan, trace::Span};
+use serde::{Deserialize, Serialize};
+use serenity::all::{GuildId, User};
+use tracing::{debug, error};
+
+#[derive(Serialize, Deserialize)]
+pub struct GuildMemberRemovalEventContent {
+    pub guild_id: GuildId,
+    pub user: User,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct GuildMemberRemovalEvent {
+    pub event: EventType,
+    pub shard_id: u32,
+    pub timestamp: u64,
+    pub content: GuildMemberRemovalEventContent,
+}
+
+impl Handler {
+    pub async fn handle_guild_member_removal_event(
+        &self,
+        event_data: &str,
+        span: &mut BoxedSpan,
+        cx: &Context,
+    ) -> bool {
+        let message: GuildMemberRemovalEvent =
+            serde_json::from_str(event_data).expect("Failed to parse data as message delete event");
+
+        let url = format!("{}/guilds/member/remove", self.get_url(&message.event));
+        span.add_event(format!("Sending post request to {}", url), Vec::new());
+
+        let mut request = self.client.post(&url).json(&message).build().unwrap();
+        global::get_text_map_propagator(|propagator| {
+            propagator.inject_context(
+                cx,
+                &mut ReqwestInjector {
+                    headers: request.headers_mut(),
+                },
+            )
+        });
+        let response = self.client.execute(request).await;
+
+        match response {
+            Ok(_) => {
+                debug!("Sucessful request to {}", url);
+                true
+            }
+            Err(err) => {
+                error!("Request to {} errored with {:#?}", url, err);
+                false
+            }
+        }
+    }
+}

bulbbot.GatewayHandler/src/events/mod.rs

@@ -1,3 +1,5 @@
+pub mod guild_member_addition;
+pub mod guild_member_removal;
 pub mod mesage_update;
 pub mod message;
 pub mod message_delete;

bulbbot.GatewayHandler/src/handler.rs

@@ -27,10 +27,12 @@ impl Handler {
 
     pub fn get_url(&self, event_type: &EventType) -> String {
         match event_type {
-            EventType::Message => String::from("http://localhost:3521/api"),
-            EventType::MessageUpdate => String::from("http://localhost:3521/api"),
-            EventType::MessageDelete => String::from("http://localhost:3521/api"),
-            _ => String::new(),
+            EventType::Message | EventType::MessageUpdate | EventType::MessageDelete => {
+                String::from("http://localhost:3521/api")
+            }
+            EventType::GuildMemberAddition | EventType::GuildMemberRemoval => {
+                String::from("http://localhost:4614/api")
+            }
         }
     }
 
@@ -57,12 +59,14 @@ impl Handler {
                 self.handle_mesage_update_event(event_data, &mut span, cx)
                     .await
             }
-            _ => false, /*
-                        EventType::MessageUpdate => todo!(),
-                        EventType::MessageDelete => todo!(),
-                        EventType::GuildMemberAddition => todo!(),
-                        EventType::GuildMemberRemoval => todo!(),
-                         */
+            EventType::GuildMemberAddition => {
+                self.handle_guild_member_addition_event(event_data, &mut span, cx)
+                    .await
+            }
+            EventType::GuildMemberRemoval => {
+                self.handle_guild_member_removal_event(event_data, &mut span, cx)
+                    .await
+            }
         };
 
         span.add_event("Handled the event", Vec::new());