chore: snapshot event emitter branch before rebase check

Cargo.toml

@@ -7,6 +7,7 @@ members = [
     "dipper-iisa",
     "dipper-pgmq",
     "dipper-pgregistry",
+    "dipper-producer",
     "dipper-rpc",
 ]
 

bin/dipper-service/Cargo.toml

@@ -12,6 +12,7 @@ dipper-core = { version = "0.1.0", path = "../../dipper-core" }
 dipper-iisa = { path = "../../dipper-iisa" }
 dipper-pgmq = { path = "../../dipper-pgmq" }
 dipper-pgregistry = { path = "../../dipper-pgregistry" }
+dipper-producer = { path = "../../dipper-producer" }
 dipper-rpc = { version = "0.1.0", path = "../../dipper-rpc", features = ["admin-rpc", "indexer-rpc"] }
 futures-lite = { version = "2.4.0", default-features = false }
 graph-networks-registry.workspace = true

bin/dipper-service/src/admin_rpc_server/context.rs

@@ -1,6 +1,7 @@
 use std::{collections::BTreeSet, sync::Arc};
 
 use dipper_core::state::FromState;
+use dipper_producer::events::SubgraphIndexingAgreementEventsProducer;
 use thegraph_core::alloy::primitives::Address;
 
 use super::handlers::{IndexingAgreementsCtx, IndexingRequestsCtx};
@@ -23,6 +24,10 @@ pub struct Ctx<R, W> {
 
     /// The message queue worker
     pub worker: W,
+
+    /// Subgraph Indexing Agreements Events emitter for sending on the topic, if configured and enabled
+    pub subgraph_indexing_agreements_events_emitter:
+        Arc<dyn SubgraphIndexingAgreementEventsProducer>,
 }
 
 impl<R, W> FromState<Ctx<R, W>> for IndexingRequestsCtx<R, W>
@@ -37,6 +42,9 @@ where
             registry: ctx.registry.clone(),
             worker: ctx.worker.clone(),
             max_candidates: ctx.max_candidates,
+            subgraph_indexing_agreements_events_emitter: ctx
+                .subgraph_indexing_agreements_events_emitter
+                .clone(),
         }
     }
 }

bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs

@@ -2,6 +2,7 @@ use std::{collections::BTreeSet, sync::Arc};
 
 use async_trait::async_trait;
 use dipper_core::{ids::IndexingRequestId, state::FromState};
+use dipper_producer::{events::SubgraphIndexingAgreementEventsProducer, proto};
 use dipper_rpc::admin::{
     SignedMessage,
     indexing_requests::{
@@ -31,6 +32,8 @@ pub struct Ctx<R, W> {
     pub registry: R,
     pub worker: W,
     pub max_candidates: usize,
+    pub subgraph_indexing_agreements_events_emitter:
+        Arc<dyn SubgraphIndexingAgreementEventsProducer>,
 }
 
 pub struct RpcServerImpl<R, W>(Ctx<R, W>);
@@ -130,6 +133,21 @@ where
                     num_candidates,
                     "Inserted new indexing request"
                 );
+
+                // A new request was received: emit the lifecycle event. Only the
+                // `Inserted` outcome is a genuinely new request; `Updated`/`Canceled`
+                // are later transitions in the lifecycle, not "request received".
+                // `the_graph_network` is the protocol network (signer chain id), not
+                // the deployment's data-source `chain_id`.
+                self.subgraph_indexing_agreements_events_emitter
+                    .produce_subgraph_indexing_agreement_request_received(
+                        deployment_id,
+                        self.signer.chain_id(),
+                        proto::SubgraphIndexingAgreementRequestReceived {
+                            agreements_requested: num_candidates as i32,
+                        },
+                    );
+
                 (Some(id), Some(num_candidates))
             }
             SetTargetOutcome::Updated {
@@ -230,3 +248,270 @@ fn into_indexing_request_status(status: IndexingRequestRecordStatus) -> Indexing
         IndexingRequestRecordStatus::Canceled => IndexingRequestStatus::Canceled,
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use async_trait::async_trait;
+    use dipper_core::ids::{IndexingAgreementId, IndexingRequestId};
+    use dipper_rpc::admin::indexing_requests::SetIndexingTargetCandidates;
+    use thegraph_core::{
+        DeploymentId,
+        alloy::{
+            primitives::{Address, ChainId},
+            signers::local::PrivateKeySigner,
+        },
+        deployment_id,
+        signed_message::sign,
+    };
+    use url::Url;
+
+    use super::*;
+    use crate::{
+        registry::Result as RegistryResult,
+        test_support::{CapturedEvent, CapturingEventsProducer},
+        worker::queue::JobId,
+    };
+
+    /// The protocol (signer) chain id. Deliberately different from the
+    /// request's deployment `chain_id` so the assertion that the emitted
+    /// event carries the *signer* chain id is meaningful.
+    const SIGNER_CHAIN_ID: ChainId = 42161;
+
+    /// The request's deployment chain id (a data-source chain), intentionally
+    /// distinct from [`SIGNER_CHAIN_ID`].
+    const DEPLOYMENT_CHAIN_ID: ChainId = 1;
+
+    /// A registry whose `set_indexing_target_candidates` returns a configured
+    /// outcome. All other trait methods are unused by the handler path under
+    /// test and therefore `unimplemented!()`.
+    #[derive(Clone)]
+    struct MockRegistry {
+        outcome: SetTargetOutcome,
+    }
+
+    #[async_trait]
+    impl IndexingRequestRegistry for MockRegistry {
+        async fn set_indexing_target_candidates(
+            &self,
+            _requested_by: Address,
+            _deployment_id: DeploymentId,
+            _deployment_chain_id: ChainId,
+            _num_candidates: usize,
+        ) -> RegistryResult<SetTargetOutcome> {
+            Ok(self.outcome.clone())
+        }
+
+        async fn get_all_indexing_requests(&self) -> RegistryResult<Vec<IndexingRequestRecord>> {
+            unimplemented!()
+        }
+
+        async fn get_indexing_request_by_id(
+            &self,
+            _id: &IndexingRequestId,
+        ) -> RegistryResult<Option<IndexingRequestRecord>> {
+            unimplemented!()
+        }
+
+        async fn get_indexing_requests_by_deployment_id(
+            &self,
+            _deployment_id: &DeploymentId,
+        ) -> RegistryResult<Vec<IndexingRequestRecord>> {
+            unimplemented!()
+        }
+
+        async fn get_open_indexing_requests_for_reassessment(
+            &self,
+            _min_age_seconds: i64,
+            _batch_size: i64,
+        ) -> RegistryResult<Vec<IndexingRequestRecord>> {
+            unimplemented!()
+        }
+    }
+
+    /// A worker that accepts `reassess_indexing_request` and returns a default
+    /// `JobId`. All other queue methods are unused by this path.
+    #[derive(Clone)]
+    struct MockWorker;
+
+    #[async_trait]
+    impl WorkerQueue for MockWorker {
+        async fn send_indexing_agreement_proposal(
+            &self,
+            _candidate_url: Url,
+            _agreement_id: IndexingAgreementId,
+            _indexing_request_id: IndexingRequestId,
+            _deployment_id: DeploymentId,
+            _deployment_chain_id: ChainId,
+            _priority: crate::worker::queue::JobPriority,
+        ) -> anyhow::Result<JobId> {
+            unimplemented!()
+        }
+
+        async fn reassess_indexing_request(
+            &self,
+            _indexing_request_id: IndexingRequestId,
+            _deployment_id: DeploymentId,
+            _deployment_chain_id: ChainId,
+            _num_candidates: usize,
+            _priority: crate::worker::queue::JobPriority,
+        ) -> anyhow::Result<JobId> {
+            Ok(JobId::default())
+        }
+
+        async fn cancel_rejected_agreement_on_chain(
+            &self,
+            _agreement_id: IndexingAgreementId,
+            _priority: crate::worker::queue::JobPriority,
+        ) -> anyhow::Result<JobId> {
+            unimplemented!()
+        }
+
+        async fn submit_offer(
+            &self,
+            _agreement_id: IndexingAgreementId,
+            _indexing_request_id: IndexingRequestId,
+            _indexer_url: Url,
+            _deployment_id: DeploymentId,
+            _deployment_chain_id: ChainId,
+            _priority: crate::worker::queue::JobPriority,
+        ) -> anyhow::Result<JobId> {
+            unimplemented!()
+        }
+    }
+
+    /// Build a signed `set_indexing_target_candidates` request using the
+    /// canonical `thegraph_core::signed_message::sign` helper and the admin
+    /// EIP-712 domain, returning both the signer (so its address can be
+    /// allowlisted) and the wrapped `SignedMessage`.
+    fn signed_request(
+        deployment_id: DeploymentId,
+        chain_id: ChainId,
+        num_candidates: Option<usize>,
+    ) -> (PrivateKeySigner, SignedMessage<SetIndexingTargetCandidates>) {
+        let signer = PrivateKeySigner::random();
+        let domain = dipper_rpc::admin::eip712_domain();
+        let message = SetIndexingTargetCandidates {
+            deployment_id,
+            chain_id,
+            num_candidates,
+        };
+        let inner = sign(&signer, &domain, message).expect("signing failed");
+        (signer, inner.into())
+    }
+
+    /// Assemble an `RpcServerImpl` whose signer's chain id is
+    /// [`SIGNER_CHAIN_ID`], whose allowlist contains `allowed`, and whose
+    /// registry returns `outcome`. Returns the server and the shared events
+    /// capture for assertions.
+    fn server(
+        allowed: Address,
+        outcome: SetTargetOutcome,
+    ) -> (
+        RpcServerImpl<MockRegistry, MockWorker>,
+        CapturingEventsProducer,
+    ) {
+        let domain = dipper_rpc::admin::eip712_domain();
+        // The signer's own address is irrelevant to recovery here (the handler
+        // recovers the *message* signer), so any address works.
+        let signer = Eip712Signer::new(SIGNER_CHAIN_ID, domain);
+
+        let events = CapturingEventsProducer::new();
+
+        let ctx = Ctx {
+            signer: Arc::new(signer),
+            gateway_operator_allowlist: Arc::new(BTreeSet::from([allowed])),
+            registry: MockRegistry { outcome },
+            worker: MockWorker,
+            max_candidates: 10,
+            subgraph_indexing_agreements_events_emitter: Arc::new(events.clone()),
+        };
+
+        (RpcServerImpl(ctx), events)
+    }
+
+    #[tokio::test]
+    async fn inserted_emits_single_request_received_with_signer_chain_id() {
+        let deployment = deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv");
+        let num_candidates = 7usize;
+
+        let (signer, req) = signed_request(deployment, DEPLOYMENT_CHAIN_ID, Some(num_candidates));
+        let (server, events) = server(
+            signer.address(),
+            SetTargetOutcome::Inserted {
+                id: IndexingRequestId::new(),
+            },
+        );
+
+        let result = server.set_indexing_target_candidates(req).await;
+        assert!(result.is_ok(), "handler returned an error: {result:?}");
+
+        let captured = events.events();
+        assert_eq!(captured.len(), 1, "expected exactly one emitted event");
+
+        match &captured[0] {
+            CapturedEvent::RequestReceived {
+                deployment: ev_deployment,
+                chain_id: ev_chain_id,
+                event,
+            } => {
+                assert_eq!(*ev_deployment, deployment, "deployment id mismatch");
+                assert_eq!(
+                    *ev_chain_id, SIGNER_CHAIN_ID,
+                    "event must carry the signer (protocol) chain id, not the deployment chain id"
+                );
+                assert_ne!(
+                    *ev_chain_id, DEPLOYMENT_CHAIN_ID,
+                    "event chain id must not be the request's deployment chain id"
+                );
+                assert_eq!(
+                    event.agreements_requested, num_candidates as i32,
+                    "agreements_requested mismatch"
+                );
+            }
+            other => panic!("expected RequestReceived, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn noop_emits_no_event() {
+        assert_no_event_for(SetTargetOutcome::NoOp {
+            id: IndexingRequestId::new(),
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn updated_emits_no_event() {
+        assert_no_event_for(SetTargetOutcome::Updated {
+            id: IndexingRequestId::new(),
+            new_num_candidates: 3,
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn canceled_emits_no_event() {
+        assert_no_event_for(SetTargetOutcome::Canceled {
+            id: IndexingRequestId::new(),
+        })
+        .await;
+    }
+
+    /// Drive the handler with `outcome` and assert no lifecycle event is
+    /// emitted. `Updated`/`Canceled`/`NoOp` are not "request received".
+    async fn assert_no_event_for(outcome: SetTargetOutcome) {
+        let deployment = deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv");
+
+        let (signer, req) = signed_request(deployment, DEPLOYMENT_CHAIN_ID, Some(5));
+        let (server, events) = server(signer.address(), outcome);
+
+        let result = server.set_indexing_target_candidates(req).await;
+        assert!(result.is_ok(), "handler returned an error: {result:?}");
+
+        assert!(
+            events.events().is_empty(),
+            "expected no emitted events, got {:?}",
+            events.events()
+        );
+    }
+}

bin/dipper-service/src/cancel_dispatch.rs

@@ -202,6 +202,7 @@ mod tests {
                     subgraph_deployment_id: deployment_id,
                     protocol_network: 1u64,
                     chain_id: 1u64,
+                    proposed_at: 0,
                 },
             },
             last_block_height: None,