fix: add some configs to ingestor

Author: EvolveArt

Diff for: .gitignore

@@ -1,7 +1,6 @@
 /target
 .idea
 .env
-db/
 infra/db/password.txt
 
 # Vscode settings

Diff for: pragma-ingestor/src/config.rs

@@ -9,6 +9,22 @@ pub(crate) struct Ingestor {
     /// Number of consumers to run
     #[arg(long, env = "NUM_CONSUMERS", default_value = "10")]
     pub(crate) num_consumers: usize,
+
+    /// Channel capacity for message queues
+    #[arg(long, env = "CHANNEL_CAPACITY", default_value = "1000000")]
+    pub(crate) channel_capacity: usize,
+
+    /// Publisher name for entries
+    #[arg(long, env = "PUBLISHER_NAME", default_value = "PRAGMA")]
+    pub(crate) publisher_name: String,
+
+    /// Kafka consumer group ID
+    #[arg(long, env = "KAFKA_GROUP_ID", default_value = "pragma-ingestor")]
+    pub(crate) kafka_group_id: String,
+
+    /// OpenTelemetry endpoint for telemetry data
+    #[arg(long, env = "OTEL_EXPORTER_OTLP_ENDPOINT")]
+    pub(crate) otel_endpoint: Option<String>,
 }
 
 pub(crate) fn load_configuration() -> Ingestor {

Diff for: pragma-ingestor/src/db.rs

@@ -0,0 +1,75 @@
+use deadpool_diesel::postgres::Pool;
+use pragma_entities::{
+    Entry, FundingRate, FutureEntry, InfraError, NewEntry, NewFundingRate, NewFutureEntry,
+};
+use tracing::debug;
+
+#[tracing::instrument(skip_all, fields(num_entries = new_entries.len()))]
+pub(crate) async fn insert_spot_entries(
+    pool: &Pool,
+    new_entries: Vec<NewEntry>,
+) -> Result<(), InfraError> {
+    let conn = pool.get().await.map_err(InfraError::DbPoolError)?;
+    conn.interact(move |conn| Entry::create_many(conn, new_entries))
+        .await
+        .map_err(InfraError::DbInteractionError)?
+        .map_err(InfraError::DbResultError)?;
+
+    Ok(())
+}
+
+#[tracing::instrument(skip_all, fields(num_entries = new_entries.len()))]
+pub(crate) async fn insert_future_entries(
+    pool: &Pool,
+    new_entries: Vec<NewFutureEntry>,
+) -> Result<(), InfraError> {
+    let conn = pool.get().await.map_err(InfraError::DbPoolError)?;
+
+    let new_entries = new_entries
+        .into_iter()
+        .map(|mut entry| {
+            if let Some(expiration_timestamp) = entry.expiration_timestamp {
+                if expiration_timestamp.and_utc().timestamp() == 0 {
+                    entry.expiration_timestamp = None;
+                }
+            }
+            entry
+        })
+        .collect::<Vec<_>>();
+
+    debug!(
+        "[PERP] {} new entries available",
+        new_entries
+            .iter()
+            .filter(|entry| entry.expiration_timestamp.is_none())
+            .count()
+    );
+
+    conn.interact(move |conn| FutureEntry::create_many(conn, new_entries))
+        .await
+        .map_err(InfraError::DbInteractionError)?
+        .map_err(InfraError::DbResultError)?;
+    Ok(())
+}
+
+#[tracing::instrument(skip_all, fields(num_entries = new_entries.len()))]
+pub(crate) async fn insert_funding_rate_entries(
+    pool: &Pool,
+    new_entries: Vec<NewFundingRate>,
+) -> Result<(), InfraError> {
+    let conn = pool.get().await.map_err(InfraError::DbPoolError)?;
+    let entries = conn
+        .interact(move |conn| FundingRate::create_many(conn, new_entries))
+        .await
+        .map_err(InfraError::DbInteractionError)?
+        .map_err(InfraError::DbResultError)?;
+
+    for entry in &entries {
+        debug!(
+            "new funding rate entry created {} - {}({}) - {}",
+            entry.source, entry.pair, entry.annualized_rate, entry.timestamp
+        );
+    }
+
+    Ok(())
+}

Diff for: pragma-ingestor/src/db/insert.rs

@@ -0,0 +1,2 @@
+pub(crate) mod insert;
+pub(crate) mod process;

Diff for: pragma-ingestor/src/db/mod.rs

@@ -0,0 +1,94 @@
+use deadpool_diesel::postgres::Pool;
+use pragma_entities::{NewEntry, NewFundingRate, NewFutureEntry};
+use tokio::sync::mpsc;
+
+use crate::db::insert::{insert_funding_rate_entries, insert_future_entries, insert_spot_entries};
+
+const BATCH_SIZE: usize = 100;
+
+#[tracing::instrument(skip(pool, rx))]
+pub async fn process_spot_entries(pool: Pool, mut rx: mpsc::Receiver<NewEntry>) {
+    let mut buffer = Vec::with_capacity(BATCH_SIZE);
+
+    loop {
+        tokio::select! {
+            Some(entry) = rx.recv() => {
+                buffer.push(entry);
+
+                if buffer.len() >= BATCH_SIZE {
+                    if let Err(e) = insert_spot_entries(&pool, std::mem::take(&mut buffer)).await {
+                        tracing::error!("❌ Failed to insert spot entries: {}", e);
+                    }
+                    buffer = Vec::with_capacity(BATCH_SIZE);
+                }
+            }
+            else => {
+                // Channel closed, flush remaining entries
+                if !buffer.is_empty() {
+                    if let Err(e) = insert_spot_entries(&pool, buffer).await {
+                        tracing::error!("❌ Failed to flush final spot entries: {}", e);
+                    }
+                }
+                break;
+            }
+        }
+    }
+}
+
+#[tracing::instrument(skip(pool, rx))]
+pub async fn process_future_entries(pool: Pool, mut rx: mpsc::Receiver<NewFutureEntry>) {
+    let mut buffer = Vec::with_capacity(BATCH_SIZE);
+
+    loop {
+        tokio::select! {
+            Some(entry) = rx.recv() => {
+                buffer.push(entry);
+
+                if buffer.len() >= BATCH_SIZE {
+                    if let Err(e) = insert_future_entries(&pool, std::mem::take(&mut buffer)).await {
+                        tracing::error!("❌ Failed to insert future entries: {}", e);
+                    }
+                    buffer = Vec::with_capacity(BATCH_SIZE);
+                }
+            }
+            else => {
+                // Channel closed, flush remaining entries
+                if !buffer.is_empty() {
+                    if let Err(e) = insert_future_entries(&pool, buffer).await {
+                        tracing::error!("❌ Failed to flush final future entries: {}", e);
+                    }
+                }
+                break;
+            }
+        }
+    }
+}
+
+#[tracing::instrument(skip(pool, rx))]
+pub async fn process_funding_rate_entries(pool: Pool, mut rx: mpsc::Receiver<NewFundingRate>) {
+    let mut buffer = Vec::with_capacity(BATCH_SIZE);
+
+    loop {
+        tokio::select! {
+            Some(entry) = rx.recv() => {
+                buffer.push(entry);
+
+                if buffer.len() >= BATCH_SIZE {
+                    if let Err(e) = insert_funding_rate_entries(&pool, std::mem::take(&mut buffer)).await {
+                        tracing::error!("❌ Failed to insert funding rate entries: {}", e);
+                    }
+                    buffer = Vec::with_capacity(BATCH_SIZE);
+                }
+            }
+            else => {
+                // Channel closed, flush remaining entries
+                if !buffer.is_empty() {
+                    if let Err(e) = insert_funding_rate_entries(&pool, buffer).await {
+                        tracing::error!("❌ Failed to flush final funding rate entries: {}", e);
+                    }
+                }
+                break;
+            }
+        }
+    }
+}