feat(events producer): subgraph indexing agreement event emitter to send events to kafka

dipper-producer/Cargo.toml

@@ -5,13 +5,17 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
+chrono = "0.4.45"
+fs-err = "3.3.0"
 prost = "0.14.4"
 rskafka = { version = "0.6.0", features = ["transport-tls"] }
 rustls = "0.23.41"
 rustls-pemfile = "2.2.0"
 thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true
+uuid = { workspace = true, features = ["v7"] }
+webpki-roots = "1.0.8"
 
 # Protobuf code generation dependencies
 # Run with: RUSTFLAGS="--cfg gen_event_proto" cargo check -p dipper-producer

dipper-producer/proto/indexing-agreement-events.proto

@@ -14,7 +14,7 @@
 //  5. subgraph.indexing.agreement.n_indexers_unavailable - Emitted when selection returns fewer candidates than requested
 //  6. subgraph.indexing.agreement.terminated - Emitted if either the Subgraph developer, or Indexer cancel an indexing agreement on a Subgraph.
 //
-// Partition key format: {subgraph_deployment_qm_hash}
+// Partition key format: {the_graph_network}/{subgraph_deployment_qm_hash}
 
 syntax = "proto3";
 

dipper-producer/src/events.rs

@@ -0,0 +1,66 @@
+//! Subgraph Indexing agreement event streaming module.
+//!
+//! This module provides infrastructure for emitting Subgraph Indexing agreement lifecycle events:
+//! - subgraph.indexing.agreement.request.receive
+//! - subgraph.indexing.agreement.proposed
+//! - subgraph.indexing.agreement.accepted
+//! - subgraph.indexing.agreement.request.expired
+//! - subgraph.indexing.agreement.n_indexers_unavailable
+//! - subgraph.indexing.agreement.terminated
+//!
+//! # Example - disabled
+//!
+//! ```ignore
+//! use dipper_producer::events::SubgraphIndexingAgreementsEventsEmitter;
+//!
+//! SubgraphIndexingAgreementsEventsEmitter::disabled();
+//! ```
+//!
+//! # Example - enabled with Kafka config
+//!
+//! ```ignore
+//! use dipper_producer::{
+//!     events::SubgraphIndexingAgreementsEventsEmitter,
+//!     kafka::{KafkaConfig, KafkaProducer, proto}
+//! };
+//!
+//! let config = KafkaConfig {
+//!     brokers: vec!["localhost:9092".to_string()],
+//!     topic: "dipper.subgraph.indexing.agreement.events".to_string(),
+//!     partitions: 16,
+//! };
+//!
+//! let producer = match tokio::time::timeout(
+//!     std::time::Duration::from_secs(30),
+//!     KafkaProducer::new(&config),
+//! )
+//! .await
+//! {
+//!     Ok(Ok(producer)) => producer,
+//!     Ok(Err(err)) => {
+//!         tracing::warn!(error = %err, "failed to initialize KafkaProducer, events disabled");
+//!         return std::sync::Arc::new(SubgraphIndexingAgreementsEventsEmitter::disabled());
+//!     },
+//!     Err(elapsed) => {
+//!         tracing::warn!(error = %elapsed, "failed to initialize KafkaProducer, events disabled");
+//!         return std::sync::Arc::new(SubgraphIndexingAgreementsEventsEmitter::disabled());
+//!     }
+//! };
+//!
+//! let emitter = std::sync::Arc::new(SubgraphIndexingAgreementsEventsEmitter::enabled(
+//!     std::sync::Arc::new(producer),
+//!     16,
+//! ));
+//!
+//! emitter.produce_subgraph_indexing_agreement_request_received(
+//!     "QmTXzATwNfgGVukV1fX2T6xw9f6LAYRVWpsdXyRWzUR2H9".to_string(),
+//!     "arbitrum".to_string(),
+//!     proto::SubgraphIndexingAgreementRequestReceived {
+//!         agreements_requested: 2
+//!     }
+//! );
+//! ```
+
+mod subgraph_indexing_agreements_events_emitter;
+
+pub use subgraph_indexing_agreements_events_emitter::SubgraphIndexingAgreementsEventsEmitter;