basically everything is done uwu

services/api/src/main.rs

@@ -19,6 +19,7 @@ struct Factions {
     tr: u32,
     nc: u32,
     vs: u32,
+    ns: u32,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -95,10 +96,11 @@ async fn get_world_pop(world_id: String) -> WorldPopulation {
         .unwrap()
         .as_secs();
 
-    let (vs, nc, tr): (u32, u32, u32) = redis::pipe()
-        .zcount(format!("{}/{}", world_id, 1), filter_timestamp, "+inf")
-        .zcount(format!("{}/{}", world_id, 2), filter_timestamp, "+inf")
-        .zcount(format!("{}/{}", world_id, 3), filter_timestamp, "+inf")
+    let (vs, nc, tr, ns): (u32, u32, u32, u32) = redis::pipe()
+        .zcount(format!("wp:{}/{}", world_id, 1), filter_timestamp, "+inf")
+        .zcount(format!("wp:{}/{}", world_id, 2), filter_timestamp, "+inf")
+        .zcount(format!("wp:{}/{}", world_id, 3), filter_timestamp, "+inf")
+        .zcount(format!("wp:{}/{}", world_id, 4), filter_timestamp, "+inf")
         .query(&mut con)
         .unwrap();
 
@@ -107,7 +109,7 @@ async fn get_world_pop(world_id: String) -> WorldPopulation {
     let response = WorldPopulation {
         world_id: world_id.parse().unwrap(),
         total,
-        factions: Factions { tr, nc, vs },
+        factions: Factions { tr, nc, vs, ns },
     };
 
     response

services/tasks/Cargo.toml

@@ -6,3 +6,5 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+redis = "0.22.1"
+once_cell = "1.16.0"

services/tasks/src/main.rs

@@ -1,3 +1,48 @@
-fn main() {
-    println!("Hello, world!");
+use once_cell::sync::Lazy;
+use redis::Commands;
+use std::env::args;
+use std::ops::Sub;
+use std::time::{Duration, SystemTime};
+
+pub static REDIS_CLIENT: Lazy<redis::Client> = Lazy::new(|| {
+    redis::Client::open(std::env::var("REDIS_ADDR").unwrap_or("redis://localhost:6379".to_string()))
+        .unwrap()
+});
+
+fn cmd_prune() {
+    println!("Pruning old data...");
+    let mut con = REDIS_CLIENT.get_connection().unwrap();
+
+    let prune_after = SystemTime::now()
+        .sub(Duration::from_secs(60))
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap()
+        .as_secs();
+
+    let keys: Vec<String> = con.keys("wp:*").unwrap();
+    for key in keys {
+        println!("-> Pruning {}", key);
+        let removed_items: u64 = con.zrembyscore(key, 0, prune_after).unwrap();
+        println!("==> Removed {} items", removed_items);
+    }
+}
+
+fn cmd_help() {
+    println!("Usage: {} [command]", args().nth(0).unwrap());
+    println!("Commands:");
+    println!("  help - Show this help message");
+    println!("  prune - Remove stale data from Redis");
+}
+
+fn main() {
+    let command = args().nth(1).unwrap_or("help".to_string());
+
+    match command.as_str() {
+        "help" => cmd_help(),
+        "prune" => cmd_prune(),
+        _ => {
+            println!("Unknown command: {}", command);
+            cmd_help();
+        }
+    }
 }

services/websocket/Cargo.toml

@@ -6,3 +6,12 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+redis = "0.22.1"
+once_cell = "1.16.0"
+tokio-tungstenite = { version = "0.17.2", features=["native-tls"] }
+serde_json = "1.0.88"
+serde = { version = "1.0.147", features = ["derive"] }
+tokio = { version = "1.22.0" }
+url = "2.3.1"
+futures-util = "0.3.25"
+futures = "0.3.25"

services/websocket/src/main.rs

@@ -1,3 +1,122 @@
-fn main() {
-    println!("Hello, world!");
+use futures::{pin_mut, FutureExt};
+use futures_util::StreamExt;
+use once_cell::sync::Lazy;
+use redis::Commands;
+use serde::Deserialize;
+use serde_json::json;
+use std::{env, time::SystemTime};
+use tokio_tungstenite::{connect_async, tungstenite::Message};
+
+pub static REDIS_CLIENT: Lazy<redis::Client> = Lazy::new(|| {
+    redis::Client::open(std::env::var("REDIS_ADDR").unwrap_or("redis://localhost:6379".to_string()))
+        .unwrap()
+});
+
+async fn send_init(tx: futures::channel::mpsc::UnboundedSender<Message>) {
+    let worlds_raw = env::var("WORLDS").unwrap_or_default();
+    if worlds_raw == "" {
+        println!("WORLDS not set");
+        return;
+    }
+    let worlds: Vec<&str> = worlds_raw.split(',').collect();
+
+    // Send setup message
+    let setup_msg = json!({
+        "action": "subscribe",
+        "worlds": worlds,
+        "eventNames": ["Death", "VehicleDestroy"],
+        "characters": ["all"],
+        "logicalAndCharactersWithWorlds": true,
+        "service": "event",
+    });
+
+    tx.unbounded_send(Message::text(setup_msg.to_string()))
+        .unwrap();
+
+    println!("Sent setup message");
+}
+
+fn process_event(event: &Event) {
+    let mut con = REDIS_CLIENT.get_connection().unwrap();
+
+    let timestamp = SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap()
+        .as_secs();
+
+    let key: String = format!("wp:{}/{}", event.world_id, event.team_id);
+    con.zadd::<String, u64, String, ()>(key, event.character_id.clone(), timestamp)
+        .unwrap();
+
+    if event.attacker_character_id != "" {
+        let key = format!("wp:{}/{}", event.world_id, event.attacker_team_id);
+        con.zadd::<String, u64, String, ()>(key, event.attacker_character_id.clone(), timestamp)
+            .unwrap();
+    }
+}
+
+#[derive(Deserialize, Debug, Clone)]
+struct Event {
+    event_name: String,
+    world_id: String,
+    character_id: String,
+    attacker_character_id: String,
+    attacker_team_id: String,
+    team_id: String,
+}
+
+#[derive(Deserialize, Debug, Clone)]
+struct Payload {
+    payload: Event,
+}
+
+#[tokio::main]
+async fn main() {
+    let addr = env::var("WS_ADDR").unwrap_or_default();
+    if addr == "" {
+        println!("WS_ADDR not set");
+        return;
+    }
+    let url = url::Url::parse(&addr).unwrap();
+    let (tx, rx) = futures::channel::mpsc::unbounded();
+    let (ws_stream, _) = connect_async(url).await.expect("Failed to connect");
+    let (write, read) = ws_stream.split();
+
+    let fused_writer = rx.map(Ok).forward(write).fuse();
+    let fused_reader = read
+        .for_each(|msg| async move {
+            let body = &msg.unwrap().to_string();
+            let data: Payload = serde_json::from_str(body).unwrap_or(Payload {
+                payload: Event {
+                    event_name: "".to_string(),
+                    world_id: "".to_string(),
+                    character_id: "".to_string(),
+                    attacker_character_id: "".to_string(),
+                    attacker_team_id: "".to_string(),
+                    team_id: "".to_string(),
+                },
+            });
+
+            if data.payload.event_name == "" {
+                return;
+            }
+
+            process_event(&data.payload);
+        })
+        .fuse();
+
+    pin_mut!(fused_writer, fused_reader);
+
+    let init = tokio::spawn(send_init(tx.clone()));
+
+    futures::select! {
+        _ = fused_reader => {
+            println!("Reader finished");
+        }
+        _ = fused_writer => {
+            println!("Writer finished");
+        }
+    }
+
+    init.await.unwrap();
 }