feat: add the Kafka consumer for Studio's indexing requests

dipper-producer/Cargo.toml

@@ -20,6 +20,9 @@ tracing.workspace = true
 uuid = { workspace = true, features = ["v7"] }
 webpki-roots = "1.0.8"
 
+[dev-dependencies]
+serde_json.workspace = true
+
 # Protobuf code generation dependencies
 # Run with: RUSTFLAGS="--cfg gen_event_proto" cargo check -p dipper-producer
 [target.'cfg(gen_event_proto)'.build-dependencies]

dipper-producer/README.md

@@ -1,22 +1,22 @@
 # dipper-producer
 
-This crate contains protobuf definitions for dipper indexer agreement event streaming.
-The generated Rust bindings are committed to the repository and only need to be regenerated when the `.proto` file changes.
+This crate contains the protobuf definitions and Kafka plumbing for dipper event streaming: the producer side for the agreement lifecycle events the dipper emits, and the consumer side for the subgraph indexing request events Studio emits. The generated Rust bindings are committed to the repository and only need to be regenerated when a `.proto` file changes.
 
 ## Protobuf Generation
 
-The build script uses a configuration flag `gen_event_proto` that enables protobuf code generation via `prost-build`. When enabled, the build script compiles `proto/indexing-agreement-events.proto` into Rust types under `src/proto/`.
+The build script uses a configuration flag `gen_event_proto` that enables protobuf code generation via `prost-build`. When enabled, the build script compiles the schemas under `proto/` into Rust types under `src/proto/`:
+
+- `proto/indexing-agreement-events.proto`, owned by this repo, generates `src/proto/dipper.subgraph.indexing.agreement.events.v1.rs`.
+- `proto/subgraph-indexing-request-events.proto`, vendored from the subgraph-studio repo (`packages/shared/src/helpers/dips/proto/SubgraphIndexingRequest.proto`), generates `src/proto/studio.subgraph.indexing.requests.events.v1.rs`. When Studio's copy changes, re-vendor it here and regenerate.
 
 To regenerate protobuf bindings, run:
 
 ```bash
-just gen-indexing-agreement-events-proto
+just gen-event-protos
 ```
 
 Or using the full `cargo` command:
 
 ```bash
 RUSTFLAGS="--cfg gen_event_proto" cargo check -p dipper-producer
 ```
-
-This will regenerate `src/proto/dipper.subgraph.indexing.agreement.events.v1.rs` from `proto/indexing-agreement-events.proto`.

dipper-producer/build.rs

@@ -7,7 +7,13 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         config.out_dir("src/proto");
         config.protoc_arg("--experimental_allow_proto3_optional");
 
-        config.compile_protos(&["proto/indexing-agreement-events.proto"], &["proto/"])?;
+        config.compile_protos(
+            &[
+                "proto/indexing-agreement-events.proto",
+                "proto/subgraph-indexing-request-events.proto",
+            ],
+            &["proto/"],
+        )?;
 
         // Instruct cargo to rerun this build script if any of the proto files change
         println!("cargo:rerun-if-changed=proto");

dipper-producer/proto/subgraph-indexing-request-events.proto

@@ -0,0 +1,88 @@
+// Subgraph Indexing Requests Events Protocol Buffer Schema
+//
+// Vendored copy of the schema owned by Subgraph Studio:
+//   repo:   edgeandnode/subgraph-studio
+//   path:   packages/shared/src/helpers/dips/proto/SubgraphIndexingRequest.proto
+//   commit: a8140613
+//
+// Studio produces these messages on a Redpanda topic; the dipper consumes them
+// and turns each propose event into a set_indexing_target_candidates call.
+// When Studio's copy changes, re-vendor it here and regenerate the bindings
+// (see the crate README).
+//
+// Local divergence from the vendored source, kept intentionally:
+//   - SubgraphIndexingRequestPropose.indexed_network_caip2id (field 2) is the
+//     network the subgraph indexes, which the dipper needs to key the indexing
+//     request and Studio has at every call site. It is a backward-compatible
+//     proto3 addition agreed as the preferred fix for that gap; drop this note
+//     once Studio's copy carries the field.
+//
+// Event flow:
+// 1. subgraph.indexing.request.propose - Emitted when the developer sends a request to have their Subgraph indexed
+// 2. subgraph.indexing.agreements.terminate - Emitted when the developer sends a request to terminate active Indexing agreements
+//
+// Partition key format: {the_graph_network_caip2id}/{subgraph_deployment_qm_hash}/{request/terminate}
+//
+//  - the_graph_network_caip2id -> CAIP2 ID of The Graph network where the Subgraph is published
+//  - subgraph_deployment_qm_hash -> Qm hash of the Subgraph deployment requesting to be indexed
+//  Example: eip155:42161/QmTXzATwNfgGVukV1fX2T6xw9f6LAYRVWpsdXyRWzUR2H9/request
+
+syntax = "proto3";
+
+package studio.subgraph.indexing.requests.events.v1;
+
+// SubgraphIndexingRequestEvent is the envelope that wraps all the Subgraph Indexing Request event types.
+//
+// Provides:
+//  - event_id -> unique identifier (uuid)
+//  - event_type -> event type discrimination
+//  - event_version -> versioning for schema evolution awareness to the consumer
+//  - timestamp -> when the event occurred, useful for ordering/debugging
+//  - subgraph_deployment_qm_hash -> Qm hash of the Subgraph deployment
+//  - the_graph_network_caip2id -> CAIP-2 chain id of The Graph protocol network the Subgraph Deployment is published to (e.g. eip155:42161)
+//  - payload -> the event payload, determined by the event_type
+message SubgraphIndexingRequestEvent {
+  // Unique event identifier.
+  // Format: UUID v7 (time-ordered) for natural chronological sorting.
+  // Example: "01912345-6789-7abc-def0-123456789abc"
+  string event_id = 1;
+
+  // Event type discriminator for routing and filtering.
+  // Values: "subgraph.indexing.request.propose", "subgraph.indexing.agreements.terminate"
+  string event_type = 2;
+
+  // Schema version for forward compatibility.
+  // Consumers should handle unknown fields gracefully.
+  // Current version: "1.0"
+  string event_version = 3;
+
+  // Event timestamp in RFC 3339 format.
+  // Example: "2024-01-15T10:30:00.123Z"
+  string timestamp = 4;
+
+  // Qm hash of the Subgraph deployment with a submitted indexing agreement event.
+  string subgraph_deployment_qm_hash = 5;
+
+  // CAIP-2 chain id of The Graph protocol network where the Subgraph was published to.
+  // Format: "eip155:{chain_id}". Examples: "eip155:42161" (arbitrum), "eip155:421614" (arbitrum-sepolia)
+  string the_graph_network_caip2id = 6;
+
+  // Event payload - exactly one of the specific event types.
+  // Use event_type field to determine which payload is present.
+  oneof payload {
+    SubgraphIndexingRequestPropose subgraph_indexing_request_propose = 7;
+  }
+}
+
+// SubgraphIndexingRequestPropose is emitted when the Subgraph developer initiates a request to have the Subgraph indexed
+//
+// Event type: subgraph.indexing.request.propose
+message SubgraphIndexingRequestPropose {
+  // Number of requested Indexing agreements to find
+  int32 indexing_agreements_requested = 1;
+
+  // CAIP-2 chain id of the network the Subgraph indexes (its data source),
+  // distinct from the_graph_network_caip2id on the envelope.
+  // Format: "eip155:{chain_id}". Example: "eip155:1" for a subgraph indexing Ethereum mainnet.
+  string indexed_network_caip2id = 2;
+}

dipper-producer/src/events/subgraph_indexing_agreements_events_emitter.rs

@@ -711,6 +711,7 @@ mod tests {
             sasl_password: None,
             tls_enabled: false,
             tls_ca_cert_path: None,
+            connect_timeout_secs: 60,
         }
     }
 

dipper-producer/src/kafka.rs

@@ -1,27 +1,11 @@
-//! Kafka client for subgraph indexing agreement event streaming.
-//!
-//! This module provides a Kafka producer for emitting subgraph indexing agreement lifecycle events
-//! to a kafka topic
-//!
-//! Events are encoded using Protocol Buffers for compact, schema-enforced messages.
-//!
-//! # Example
-//!
-//! ```ignore
-//! use dipper_producer::kafka::{KafkaConfig, KafkaProducer};
-//!
-//! let config = KafkaConfig {
-//!     brokers: vec!["localhost:9092".to_string()],
-//!     topic: "dipper.subgraph.indexing.agreement.events".to_string(),
-//!     partitions: 16,
-//! };
-//!
-//! let producer = KafkaProducer::new(&config).await?;
-//!
-//! // Send an event with partition key and protobuf payload
-//! producer.send("QmT329Bej8AwSLahmgnmi6fdYkj3rorYAcCes45gDv9aJ4", &encoded_event).await?;
-//! ```
+//! Kafka clients for dipper event streaming: a producer for the agreement
+//! lifecycle events the dipper emits and a consumer for the indexing request
+//! events Studio emits. Events are Protocol Buffers encoded.
 
+mod connection;
+mod consumer;
 mod producer;
 
+pub use connection::ConnectionError;
+pub use consumer::{ConsumerError, KafkaConsumer, KafkaConsumerConfig};
 pub use producer::{Error, KafkaConfig, KafkaProducer};