enable nats jetstream deduplication (#2)

Author: ayazabbas

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-stream"
-version = "0.1.0"
+version = "0.1.1"
 edition = "2021"
 
 [lib]

Cargo.toml

@@ -1,5 +1,6 @@
 use anyhow::Result;
 use async_nats::jetstream::{self};
+use async_nats::HeaderMap;
 use clap::Parser;
 use config::Config;
 use pyth_sdk_solana::state::{load_price_account, PriceStatus, PythnetPriceAccount};
@@ -160,13 +161,26 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
 
             let message = serde_json::to_string(&price_update)?;
 
+            // Create a unique message ID
+            let message_id = format!(
+                "{}:{}",
+                price_update.price_feed.id, price_update.price_feed.price.publish_time
+            );
+
+            // Create headers with the Nats-Msg-Id
+            let mut headers = HeaderMap::new();
+            headers.insert("Nats-Msg-Id", message_id.as_str());
+
             let jetstream_clone = jetstream.clone();
             task::spawn(async move {
                 match jetstream_clone
-                    .publish("pyth.price.updates", message.into())
+                    .publish_with_headers("pyth.price.updates", headers, message.into())
                     .await
                 {
-                    Ok(_) => debug!("Published price update to JetStream"),
+                    Ok(_) => debug!(
+                        "Published price update to JetStream with ID: {}",
+                        message_id
+                    ),
                     Err(e) => warn!("Failed to publish price update to JetStream: {}", e),
                 }
             });

src/bin/pyth_reader.rs

@@ -14,6 +14,8 @@ pub async fn setup_jetstream(nats_client: &async_nats::Client) -> Result<jetstre
         duplicate_window: Duration::from_secs(60),
         discard: stream::DiscardPolicy::New,
         max_messages_per_subject: 100,
+        allow_direct: true,
+        allow_rollup: true,
         ..Default::default()
     };