user-ingest: Push matchdata_salts to rmq stream

user-ingest/Cargo.toml

@@ -14,3 +14,4 @@ reqwest = "0.12.8"
 tempfile = "3.13.0"
 futures = "0.3.31"
 lapin = "2.5.0"
+serde_json = "1.0.128"

user-ingest/src/download.rs

@@ -48,7 +48,7 @@ pub async fn process_data(salts: &Salts, data_type: DataType) -> Result<(), Proc
 
     download_to_file(salts, data_type, &local_path).await?;
     s3::upload_to_s3(&local_path, &s3_path).await?;
-    rmq::add_to_queue(&s3_path).await?;
+    rmq::add_to_queue("db_ingest_queue", &s3_path).await?;
     info!("Uploaded {}", s3_path);
 
     local_file.close().map_err(ProcessError::Io)

user-ingest/src/main.rs

@@ -48,6 +48,13 @@ async fn main() -> Result<(), io::Error> {
         let semaphore = Arc::new(semaphore);
         let mut tasks = FuturesUnordered::new();
         while let Some(salts) = salts_channel_receiver.recv().await {
+            let serialized_salts = serde_json::to_string(&salts);
+            if let Ok(serialized_salts) = serialized_salts {
+                match rmq::add_to_queue("matchdata_salts", &serialized_salts).await {
+                    Ok(_) => debug!("Sent salts to queue"),
+                    Err(e) => error!("Failed to send salts to queue: {:?}", e),
+                }
+            }
             let permit = semaphore.clone().acquire_owned().await.unwrap();
 
             debug!("Received metadata download task: {:?}", salts);

user-ingest/src/models.rs

@@ -1,4 +1,4 @@
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use std::fmt::Display;
 use std::io;
 
@@ -10,7 +10,7 @@ pub enum ProcessError {
     RmqError(lapin::Error),
 }
 
-#[derive(Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Salts {
     pub cluster_id: u32,
     pub match_id: u64,

user-ingest/src/rmq.rs

@@ -13,18 +13,16 @@ static RABBITMQ_PASS: LazyLock<String> =
 static RABBITMQ_HOST: LazyLock<String> = LazyLock::new(|| std::env::var("RABBITMQ_HOST").unwrap());
 static RABBITMQ_PORT: LazyLock<String> = LazyLock::new(|| std::env::var("RABBITMQ_PORT").unwrap());
 
-const RABBITMQ_QUEUE: &str = "db_ingest_queue";
-
 static RABBITMQ_CONNECTION: OnceCell<Connection> = OnceCell::const_new();
 static RABBITMQ_CHANNEL: OnceCell<Channel> = OnceCell::const_new();
 
-pub async fn add_to_queue(body: &str) -> Result<(), ProcessError> {
+pub async fn add_to_queue(routing_key: &str, body: &str) -> Result<(), ProcessError> {
     let rmq_channel = get_rmq_channel().await?;
-    info!("Sending message {} to queue: {}", body, RABBITMQ_QUEUE);
+    info!("Sending message {} to queue: {}", body, routing_key);
     rmq_channel
         .basic_publish(
             "",
-            RABBITMQ_QUEUE,
+            routing_key,
             BasicPublishOptions::default(),
             body.as_bytes(),
             BasicProperties::default(),