feat: add cloudflare queues to the streams (#314)

Author: joshstevens19

cli/Makefile

@@ -73,3 +73,5 @@ hyperwarp_codegen_typings:
 	RUSTFLAGS='-C target-cpu=native' cargo run --release --features jemalloc -- codegen --path $(CURDIR)/../examples/hyperevm typings
 hyperwarp_codegen_indexer:
 	RUSTFLAGS='-C target-cpu=native' cargo run --release --features jemalloc -- codegen --path $(CURDIR)/../examples/hyperevm indexer
+start_cloudflare_indexer:
+	RUSTFLAGS='-C target-cpu=native' RUST_BACKTRACE='full' cargo run --release --features jemalloc -- start --path $(CURDIR)/../examples/rindexer_cloudflare_queues indexer

core/src/manifest/stream.rs

@@ -160,6 +160,21 @@ pub struct KafkaStreamConfig {
     pub topics: Vec<KafkaStreamQueueConfig>,
 }
 
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct CloudflareQueuesStreamQueueConfig {
+    pub queue_id: String,
+    pub networks: Vec<String>,
+    #[serde(default)]
+    pub events: Vec<StreamEvent>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct CloudflareQueuesStreamConfig {
+    pub api_token: String,
+    pub account_id: String,
+    pub queues: Vec<CloudflareQueuesStreamQueueConfig>,
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct StreamsConfig {
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -176,6 +191,9 @@ pub struct StreamsConfig {
 
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub redis: Option<RedisStreamConfig>,
+
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub cloudflare_queues: Option<CloudflareQueuesStreamConfig>,
 }
 
 impl StreamsConfig {
@@ -199,6 +217,8 @@ impl StreamsConfig {
             path.push_str("kafka_");
         } else if self.redis.is_some() {
             path.push_str("redis_");
+        } else if self.cloudflare_queues.is_some() {
+            path.push_str("cloudflare_queues_");
         }
 
         path.trim_end_matches('_').to_string()

core/src/streams/clients.rs

@@ -14,13 +14,15 @@ use crate::{
     event::{filter_event_data_by_conditions, EventMessage},
     indexer::native_transfer::EVENT_NAME,
     manifest::stream::{
-        KafkaStreamConfig, KafkaStreamQueueConfig, RabbitMQStreamConfig, RabbitMQStreamQueueConfig,
-        RedisStreamConfig, RedisStreamStreamConfig, SNSStreamTopicConfig, StreamEvent,
-        StreamsConfig, WebhookStreamConfig,
+        CloudflareQueuesStreamConfig, CloudflareQueuesStreamQueueConfig, KafkaStreamConfig,
+        KafkaStreamQueueConfig, RabbitMQStreamConfig, RabbitMQStreamQueueConfig, RedisStreamConfig,
+        RedisStreamStreamConfig, SNSStreamTopicConfig, StreamEvent, StreamsConfig,
+        WebhookStreamConfig,
     },
     streams::{
         kafka::{Kafka, KafkaError},
-        RabbitMQ, RabbitMQError, Redis, RedisError, Webhook, WebhookError, SNS,
+        CloudflareQueues, CloudflareQueuesError, RabbitMQ, RabbitMQError, Redis, RedisError,
+        Webhook, WebhookError, SNS,
     },
 };
 
@@ -54,6 +56,9 @@ pub enum StreamError {
     #[error("Redis could not publish: {0}")]
     RedisCouldNotPublish(#[from] RedisError),
 
+    #[error("Cloudflare Queues could not publish: {0}")]
+    CloudflareQueuesCouldNotPublish(#[from] CloudflareQueuesError),
+
     #[error("Task failed: {0}")]
     JoinError(JoinError),
 }
@@ -82,17 +87,25 @@ pub struct RedisStream {
     client: Arc<Redis>,
 }
 
+#[derive(Debug)]
+pub struct CloudflareQueuesStream {
+    config: CloudflareQueuesStreamConfig,
+    client: Arc<CloudflareQueues>,
+}
+
 #[derive(Debug)]
 pub struct StreamsClients {
     sns: Option<SNSStream>,
     webhook: Option<WebhookStream>,
     rabbitmq: Option<RabbitMQStream>,
     kafka: Option<KafkaStream>,
     redis: Option<RedisStream>,
+    cloudflare_queues: Option<CloudflareQueuesStream>,
 }
 
 impl StreamsClients {
     pub async fn new(stream_config: StreamsConfig) -> Self {
+        #[allow(clippy::manual_map)]
         let sns = if let Some(config) = &stream_config.sns {
             Some(SNSStream {
                 config: config.topics.clone(),
@@ -102,11 +115,13 @@ impl StreamsClients {
             None
         };
 
+        #[allow(clippy::manual_map)]
         let webhook = stream_config.webhooks.as_ref().map(|config| WebhookStream {
             config: config.clone(),
             client: Arc::new(Webhook::new()),
         });
 
+        #[allow(clippy::manual_map)]
         let rabbitmq = if let Some(config) = stream_config.rabbitmq.as_ref() {
             Some(RabbitMQStream {
                 config: config.clone(),
@@ -116,6 +131,7 @@ impl StreamsClients {
             None
         };
 
+        #[allow(clippy::manual_map)]
         let kafka = if let Some(config) = stream_config.kafka.as_ref() {
             Some(KafkaStream {
                 config: config.clone(),
@@ -129,6 +145,7 @@ impl StreamsClients {
             None
         };
 
+        #[allow(clippy::manual_map)]
         let redis = if let Some(config) = stream_config.redis.as_ref() {
             Some(RedisStream {
                 config: config.clone(),
@@ -142,7 +159,20 @@ impl StreamsClients {
             None
         };
 
-        Self { sns, webhook, rabbitmq, kafka, redis }
+        #[allow(clippy::manual_map)]
+        let cloudflare_queues = if let Some(config) = stream_config.cloudflare_queues.as_ref() {
+            Some(CloudflareQueuesStream {
+                config: config.clone(),
+                client: Arc::new(CloudflareQueues::new(
+                    config.api_token.clone(),
+                    config.account_id.clone(),
+                )),
+            })
+        } else {
+            None
+        };
+
+        Self { sns, webhook, rabbitmq, kafka, redis, cloudflare_queues }
     }
 
     fn has_any_streams(&self) -> bool {
@@ -151,6 +181,7 @@ impl StreamsClients {
             || self.rabbitmq.is_some()
             || self.kafka.is_some()
             || self.redis.is_some()
+            || self.cloudflare_queues.is_some()
     }
 
     fn chunk_data(&self, data_array: &Vec<Value>) -> Vec<Vec<Value>> {
@@ -450,6 +481,39 @@ impl StreamsClients {
         tasks
     }
 
+    fn cloudflare_queues_stream_tasks(
+        &self,
+        config: &CloudflareQueuesStreamQueueConfig,
+        client: Arc<CloudflareQueues>,
+        id: &str,
+        event_message: &EventMessage,
+        chunks: Arc<Vec<Vec<Value>>>,
+    ) -> StreamPublishes {
+        let tasks: Vec<_> = chunks
+            .iter()
+            .enumerate()
+            .map(|(index, chunk)| {
+                let filtered_chunk: Vec<Value> = self.filter_chunk_event_data_by_conditions(
+                    &config.events,
+                    event_message,
+                    chunk,
+                );
+
+                let publish_message_id = self.generate_publish_message_id(id, index, &None);
+                let client = Arc::clone(&client);
+                let queue_id = config.queue_id.clone();
+                let publish_message =
+                    self.create_chunk_message_json(&config.events, event_message, &filtered_chunk);
+
+                task::spawn(async move {
+                    client.publish(&publish_message_id, &queue_id, &publish_message).await?;
+                    Ok(filtered_chunk.len())
+                })
+            })
+            .collect();
+        tasks
+    }
+
     pub async fn stream(
         &self,
         id: String,
@@ -549,6 +613,22 @@ impl StreamsClients {
                 }
             }
 
+            if let Some(cloudflare_queues) = &self.cloudflare_queues {
+                for config in &cloudflare_queues.config.queues {
+                    if config.events.iter().any(|e| e.event_name == event_message.event_name)
+                        && config.networks.contains(&event_message.network)
+                    {
+                        streams.push(self.cloudflare_queues_stream_tasks(
+                            config,
+                            Arc::clone(&cloudflare_queues.client),
+                            &id,
+                            event_message,
+                            Arc::clone(&chunks),
+                        ));
+                    }
+                }
+            }
+
             let mut streamed_total = 0;
 
             if index_event_in_order {

core/src/streams/cloudflare_queues.rs

@@ -0,0 +1,122 @@
+use reqwest::Client;
+use serde_json::Value;
+use thiserror::Error;
+
+use crate::streams::STREAM_MESSAGE_ID_KEY;
+
+#[derive(Error, Debug)]
+#[allow(clippy::enum_variant_names)]
+pub enum CloudflareQueuesError {
+    #[error("Request error: {0}")]
+    RequestError(#[from] reqwest::Error),
+
+    #[error("Cloudflare API error: {status} - {message}")]
+    ApiError { status: u16, message: String },
+
+    #[error("Serialization error: {0}")]
+    SerializationError(#[from] serde_json::Error),
+}
+
+#[derive(Debug, Clone)]
+pub struct CloudflareQueues {
+    client: Client,
+    api_token: String,
+    account_id: String,
+}
+
+impl CloudflareQueues {
+    pub fn new(api_token: String, account_id: String) -> Self {
+        Self { client: Client::new(), api_token, account_id }
+    }
+
+    pub async fn publish(
+        &self,
+        id: &str,
+        queue_id: &str,
+        message: &Value,
+    ) -> Result<(), CloudflareQueuesError> {
+        let url = format!(
+            "https://api.cloudflare.com/client/v4/accounts/{}/queues/{}/messages",
+            self.account_id, queue_id
+        );
+
+        let mut message_with_metadata = message.clone();
+        if let Value::Object(ref mut map) = message_with_metadata {
+            map.insert("message_id".to_string(), Value::String(id.to_string()));
+        }
+
+        let payload = serde_json::json!({
+            "body": message_with_metadata
+        });
+
+        let response = self
+            .client
+            .post(&url)
+            .header("Authorization", format!("Bearer {}", self.api_token))
+            .header("Content-Type", "application/json")
+            .header(STREAM_MESSAGE_ID_KEY, id)
+            .json(&payload)
+            .send()
+            .await?;
+
+        if response.status().is_success() {
+            Ok(())
+        } else {
+            let status = response.status().as_u16();
+            let error_text = response.text().await.unwrap_or_else(|_| "Unknown error".to_string());
+            Err(CloudflareQueuesError::ApiError { status, message: error_text })
+        }
+    }
+
+    #[allow(dead_code)]
+    pub async fn publish_batch(
+        &self,
+        messages: Vec<(String, Value)>,
+        queue_id: &str,
+    ) -> Result<(), CloudflareQueuesError> {
+        if messages.is_empty() {
+            return Ok(());
+        }
+
+        // Cloudflare Queues supports up to 100 messages per batch
+        const MAX_BATCH_SIZE: usize = 100;
+
+        for chunk in messages.chunks(MAX_BATCH_SIZE) {
+            let url = format!(
+                "https://api.cloudflare.com/client/v4/accounts/{}/queues/{}/messages/batch",
+                self.account_id, queue_id
+            );
+
+            let batch_messages: Vec<Value> = chunk
+                .iter()
+                .map(|(id, message)| {
+                    let mut message_with_metadata = message.clone();
+                    if let Value::Object(ref mut map) = message_with_metadata {
+                        map.insert("message_id".to_string(), Value::String(id.to_string()));
+                    }
+                    serde_json::json!({
+                        "body": message_with_metadata
+                    })
+                })
+                .collect();
+
+            let response = self
+                .client
+                .post(&url)
+                .header("Authorization", format!("Bearer {}", self.api_token))
+                .header("Content-Type", "application/json")
+                .json(&batch_messages)
+                .send()
+                .await?;
+
+            if !response.status().is_success() {
+                let status = response.status().as_u16();
+                let error_text =
+                    response.text().await.unwrap_or_else(|_| "Unknown error".to_string());
+                return Err(CloudflareQueuesError::ApiError { status, message: error_text });
+            }
+        }
+
+        Ok(())
+    }
+}

core/src/streams/mod.rs

@@ -15,4 +15,7 @@ mod redis;
 pub use clients::StreamsClients;
 pub use redis::{Redis, RedisError};
 
+mod cloudflare_queues;
+pub use cloudflare_queues::{CloudflareQueues, CloudflareQueuesError};
+
 pub const STREAM_MESSAGE_ID_KEY: &str = "x-rindexer-id";

core/src/types/core.rs

@@ -1,32 +1,5 @@
+use alloy::dyn_abi::DynSolValue;
 use alloy::json_abi::Param;
-use alloy::{
-    dyn_abi::DynSolValue,
-    primitives::{Address, BlockHash, TxHash, U256, U64},
-};
-use serde::{Deserialize, Serialize};
-
-/// Metadata inside a log
-#[allow(dead_code)]
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
-pub struct LogMeta {
-    /// Address from which this log originated
-    pub address: Address,
-
-    /// The block in which the log was emitted
-    pub block_number: U64,
-
-    /// The block hash in which the log was emitted
-    pub block_hash: BlockHash,
-
-    /// The transaction hash in which the log was emitted
-    pub transaction_hash: TxHash,
-
-    /// Transactions index position log was created from
-    pub transaction_index: U64,
-
-    /// Log index position in the block
-    pub log_index: U256,
-}
 
 /// Decoded log param.
 #[derive(Debug, PartialEq, Clone)]

documentation/docs/pages/docs/changelog.mdx

@@ -6,6 +6,7 @@
 ### Features
 -------------------------------------------------
 - feat: add health endpoint with comprehensive system status monitoring
+- feat: add cloudflare queues to the streams
 
 ### Bug fixes
 -------------------------------------------------