discord worker

.idea/locationoverflow.iml

@@ -5,6 +5,7 @@
       <sourceFolder url="file://$MODULE_DIR$/api/src" isTestSource="false" />
       <sourceFolder url="file://$MODULE_DIR$/azalea-worker/src" isTestSource="false" />
       <sourceFolder url="file://$MODULE_DIR$/common/src" isTestSource="false" />
+      <sourceFolder url="file://$MODULE_DIR$/discord-worker/src" isTestSource="false" />
       <sourceFolder url="file://$MODULE_DIR$/websocket-worker/src" isTestSource="false" />
       <excludeFolder url="file://$MODULE_DIR$/target" />
     </content>

Cargo.lock

@@ -2450,6 +2450,26 @@ dependencies = [
  "crypto-common",
 ]
 
+[[package]]
+name = "discord-worker"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "chrono",
+ "common",
+ "log 0.4.19",
+ "reqwest",
+ "serde",
+ "serde_json",
+ "serenity",
+ "simple_logger",
+ "tokio",
+ "tokio-threadpool",
+ "toml",
+ "url 2.4.0",
+ "websocket",
+]
+
 [[package]]
 name = "dispatch"
 version = "0.2.0"

Cargo.toml

@@ -3,5 +3,6 @@ members = [
     "api",
     "common",
     "websocket-worker",
-    "azalea-worker"
+    "azalea-worker",
+    "discord-worker"
 ]

discord-worker/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "discord-worker"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+log = "0.4"
+simple_logger = "4.1"
+serde = { version = "1", features = ["derive"] }
+toml = "0.7"
+url = { version = "2.3", features = ["serde"] }
+tokio = { version = "1", features = ["full"] }
+reqwest = { version = "0.11", features = ["json"] }
+serde_json = "1"
+async-trait = "0.1"
+websocket = { version = "0.26", features = ["async"], no-default-features = true }
+common = { path = "../common" }
+serenity = "0.11"
+tokio-threadpool = "0.1"
+chrono = "0.4"

discord-worker/config.toml

@@ -0,0 +1,5 @@
+api_status_url = "https://data-api.locationoverflow.coredoes.dev/status"
+api_token = "rw-minecraft-b1d648ab-498a-4499-b5a7-64aedcd7c836"
+discord_token = "NzM3MzIzNDgzMTkyNzU0MjY3.GbGBcv.BchYB4l196mToItLuHwpWKKbc2L3LedBzUuyyM"
+relay_channels = [1130340095606673458]
+prefix = "c!"

discord-worker/src/config.rs

@@ -0,0 +1,11 @@
+use serde::{Deserialize, Serialize};
+use url::Url;
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Config {
+    pub api_status_url: Url,
+    pub api_token: String,
+    pub discord_token: String,
+    pub prefix: String,
+    pub relay_channels: Vec<u64>
+}

discord-worker/src/main.rs

@@ -0,0 +1,186 @@
+use std::error::Error;
+use std::fs;
+use std::sync::Arc;
+use std::time::SystemTime;
+use chrono::DateTime;
+use log::{debug, error, info};
+use tokio_threadpool::ThreadPool;
+use url::Url;
+use websocket::{ClientBuilder, Message as WsMessage, OwnedMessage, WebSocketError, client::sync::Client as WsClient};
+use websocket::websocket_base::result::WebSocketResult;
+use common::message::{GatewayChatMessage, GatewayChatSource, GatewayPacketC2S, GatewayPacketS2C};
+use common::status::{DATA_API_VERSION, Status};
+use crate::config::Config;
+use serenity::async_trait;
+use serenity::builder::ParseValue;
+use serenity::prelude::*;
+use serenity::model::channel::{Message, Reaction, ReactionType};
+use serenity::framework::standard::macros::{command, group, hook};
+use serenity::framework::standard::{StandardFramework, CommandResult, Args};
+use serenity::http::Http;
+use serenity::model::prelude::Webhook;
+use websocket::stream::sync::NetworkStream;
+
+pub mod config;
+
+#[group]
+struct General;
+
+struct Handler;
+
+#[async_trait]
+impl EventHandler for Handler {}
+
+struct ConfigLock;
+
+impl TypeMapKey for ConfigLock {
+    type Value = Arc<RwLock<Config>>;
+}
+
+struct WsLock;
+
+impl TypeMapKey for WsLock {
+    type Value = Arc<Mutex<WsClient<Box<dyn NetworkStream + Send>>>>;
+}
+
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    simple_logger::init_with_env().unwrap();
+
+    info!("Loading config");
+
+    let mut args = std::env::args();
+    if args.len() != 2 {
+        eprintln!("usage: ./discord-worker <config>");
+        std::process::exit(1);
+    }
+
+    let file = args.nth(1).unwrap();
+
+    let config_str = fs::read_to_string(&file)?;
+
+    let config: Config = toml::from_str(&config_str)?;
+
+    info!("Config loaded from {}", file);
+
+    info!("Loading status from the API ({})...", config.api_status_url);
+
+    let status: Status = reqwest::get(config.api_status_url.clone()).await?.json().await?;
+
+    debug!("{:?}", status);
+
+    if status.data_api_version != DATA_API_VERSION {
+        error!("Data API is incompatible. This version of the websocket worker was compiled with Data API v{}, but your Data API is Data API v{}", DATA_API_VERSION, status.data_api_version);
+        std::process::exit(1);
+    }
+
+    info!("Connecting to gateway uri {}", status.gateway_url.to_string());
+
+    let mut client = ClientBuilder::new(status.gateway_url.clone().as_str())?.connect(None)?;
+
+    // send the authentication packet, and then start listen
+
+    let message = WsMessage::text(serde_json::to_string(&GatewayPacketC2S::Authenticate {
+        token: config.api_token.clone(),
+        request_write_perms: true,
+    })?);
+
+    client.send_message(&message)?;
+
+    for msg in client.incoming_messages() {
+        match msg {
+            Ok(msg) => {
+                match msg {
+                    OwnedMessage::Text(txt) => {
+                        // decode
+                        debug!("{}", txt);
+
+                        let packet: GatewayPacketS2C = serde_json::from_str(&txt)?;
+
+                        match packet {
+                            GatewayPacketS2C::AuthenticationAccepted { write_perms } => {
+                                info!("auth accepted by server");
+                                if !write_perms {
+                                    error!("Server did not grant us write permission to the gateway socket. This is required.");
+                                    error!("Check that you provided the proper token and the API has it read as write-allow.");
+                                    std::process::exit(1);
+                                }
+                                break;
+                            }
+                            GatewayPacketS2C::Disconnect { reason } => {
+                                info!("disconnected by server: {:?}", reason);
+                                return Err("Disconnected by server")?;
+                            }
+                            _ => ()
+                        }
+                    }
+                    OwnedMessage::Close(_) => {
+                        info!("closing connection");
+                        return Ok(());
+                    }
+                    _ => {
+                        debug!("ignoring unknown type");
+                    }
+                }
+            }
+            Err(e) => {
+                if matches!(e, WebSocketError::NoDataAvailable) {
+                    continue;
+                }
+                error!("rx error: {}", e);
+                return Err(e.into());
+            }
+        }
+    }
+
+    info!("Connected to gateway, starting the discord client");
+
+    let framework = StandardFramework::new().group(&GENERAL_GROUP).normal_message(normal_message);
+
+    let intents = GatewayIntents::non_privileged() | GatewayIntents::MESSAGE_CONTENT;
+    let mut d_client = Client::builder(config.discord_token.clone(), intents).event_handler(Handler).framework(framework).await?;
+
+    {
+        let mut data = d_client.data.write().await;
+        data.insert::<ConfigLock>(Arc::new(RwLock::new(config.clone())));
+        data.insert::<WsLock>(Arc::new(Mutex::new(client)));
+    }
+
+    if let Err(why) = d_client.start().await {
+        error!("Discord client error: {}", why);
+    }
+
+    Ok(())
+}
+
+#[hook]
+async fn normal_message(ctx: &Context, msg: &Message) {
+    let (config, ws) = {
+        let data_read = ctx.data.read().await;
+
+        (data_read.get::<ConfigLock>().unwrap().clone(), data_read.get::<WsLock>().unwrap().clone())
+    };
+
+    if !config.read().await.relay_channels.contains(msg.channel_id.as_u64()) { return; }
+
+    let mut content = msg.content.clone();
+
+    if !msg.attachments.is_empty() {
+        for attachment in &msg.attachments {
+            content += " ";
+            content += &attachment.url;
+        }
+    }
+
+    let message = WsMessage::text(serde_json::to_string(&GatewayPacketC2S::Relay {
+        msg: GatewayChatMessage {
+            source: GatewayChatSource::Discord,
+            username: format!("@{}", msg.author.name),
+            message: content,
+            timestamp: DateTime::from(SystemTime::now())
+        }
+    }).unwrap());
+
+    ws.lock().await.send_message(&message).unwrap();
+}