From fd3e4cdc6b58ed0cea1ad9d09f86fcc119507454 Mon Sep 17 00:00:00 2001
From: Theo Butler <theodusbutler@gmail.com>
Date: Thu, 12 Dec 2024 11:57:10 -0500
Subject: [PATCH] WIP

---
 src/config.rs                    |  2 +-
 src/main.rs                      |  3 +-
 src/network/indexer_blocklist.rs | 99 +++++++++++++++++++++++++++++---
 3 files changed, 95 insertions(+), 9 deletions(-)

diff --git a/src/config.rs b/src/config.rs
index f967eb89..a7aefef3 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -140,7 +140,7 @@ pub enum ExchangeRateProvider {
 /// Kafka configuration.
 ///
 /// See [`Config`]'s [`kafka`](struct.Config.html#structfield.kafka).
-#[derive(Deserialize)]
+#[derive(Clone, Deserialize)]
 pub struct KafkaConfig(BTreeMap<String, String>);
 
 impl Default for KafkaConfig {
diff --git a/src/main.rs b/src/main.rs
index 06eac060..640afc6a 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -109,7 +109,8 @@ async fn main() {
         }
         None => Default::default(),
     };
-    let indexer_blocklist = indexer_blocklist::Blocklist::spawn(conf.blocklist);
+    let indexer_blocklist =
+        indexer_blocklist::Blocklist::spawn(conf.blocklist, conf.kafka.clone().into());
     let mut network = network::service::spawn(
         http_client.clone(),
         network_subgraph_client,
diff --git a/src/network/indexer_blocklist.rs b/src/network/indexer_blocklist.rs
index 5f6dbdce..f88dc3b2 100644
--- a/src/network/indexer_blocklist.rs
+++ b/src/network/indexer_blocklist.rs
@@ -1,5 +1,7 @@
 use std::collections::{HashMap, HashSet};
 
+use futures::StreamExt;
+use rdkafka::Message;
 use thegraph_core::{alloy::primitives::Address, DeploymentId, ProofOfIndexing};
 use tokio::sync::watch;
 
@@ -13,7 +15,7 @@ pub struct Blocklist {
 }
 
 impl Blocklist {
-    pub fn spawn(init: Vec<BlocklistEntry>) -> Self {
+    pub fn spawn(init: Vec<BlocklistEntry>, kafka_config: rdkafka::ClientConfig) -> Self {
         let (blocklist_tx, blocklist_rx) = watch::channel(Default::default());
         let (poi_tx, poi_rx) = watch::channel(Default::default());
         let (indexer_tx, indexer_rx) = watch::channel(Default::default());
@@ -26,7 +28,7 @@ impl Blocklist {
             actor.add_entry(entry);
         }
         tokio::spawn(async move {
-            actor.run().await;
+            actor.run(kafka_config.create().unwrap()).await;
         });
         Self {
             blocklist: blocklist_rx,
@@ -37,14 +39,34 @@ impl Blocklist {
 }
 
 struct Actor {
-    pub blocklist: watch::Sender<Vec<BlocklistEntry>>,
-    pub poi: watch::Sender<HashMap<DeploymentId, Vec<(u64, ProofOfIndexing)>>>,
-    pub indexer: watch::Sender<HashMap<Address, HashSet<DeploymentId>>>,
+    blocklist: watch::Sender<Vec<BlocklistEntry>>,
+    poi: watch::Sender<HashMap<DeploymentId, Vec<(u64, ProofOfIndexing)>>>,
+    indexer: watch::Sender<HashMap<Address, HashSet<DeploymentId>>>,
 }
 
 impl Actor {
-    async fn run(&mut self) {
-        todo!();
+    async fn run(&mut self, consumer: rdkafka::consumer::StreamConsumer) {
+        let mut stream = consumer.stream();
+        while let Some(msg) = stream.next().await {
+            let msg = match msg {
+                Ok(msg) => msg,
+                Err(blocklist_recv_error) => {
+                    tracing::error!(%blocklist_recv_error);
+                    continue;
+                }
+            };
+            todo!("handle null (remove entry)");
+            let entry: BlocklistEntry = match msg.payload().map(serde_json::from_slice) {
+                Some(Ok(entry)) => entry,
+                Some(Err(blocklist_deserialize_err)) => {
+                    tracing::error!(%blocklist_deserialize_err);
+                    continue;
+                }
+                None => continue,
+            };
+            self.add_entry(entry);
+        }
+        tracing::error!("blocklist consumer stopped");
     }
 
     fn add_entry(&mut self, entry: BlocklistEntry) {
@@ -75,4 +97,67 @@ impl Actor {
         self.blocklist
             .send_modify(move |blocklist| blocklist.push(entry));
     }
+
+    fn remove_entry(&mut self, entry: &BlocklistEntry) {
+        match entry {
+            BlocklistEntry::Poi {
+                deployment,
+                block,
+                public_poi,
+                ..
+            } => {
+                self.poi.send_modify(|blocklist| {
+                    if let Some(entry) = blocklist.get_mut(deployment) {
+                        entry.retain(|value| &(*block, (*public_poi).into()) != value);
+                    }
+                });
+            }
+            BlocklistEntry::Other {
+                deployment,
+                indexer,
+                ..
+            } => {
+                self.indexer.send_modify(|blocklist| {
+                    if let Some(entry) = blocklist.get_mut(indexer) {
+                        entry.remove(deployment);
+                    }
+                });
+            }
+        };
+        fn matching(a: &BlocklistEntry, b: &BlocklistEntry) -> bool {
+            match (a, b) {
+                (
+                    BlocklistEntry::Poi {
+                        deployment,
+                        public_poi,
+                        block,
+                        info: _,
+                    },
+                    BlocklistEntry::Poi {
+                        deployment: deployment_,
+                        public_poi: public_poi_,
+                        block: block_,
+                        info: _,
+                    },
+                ) => {
+                    (deployment == deployment_) && (public_poi == public_poi_) && (block == block_)
+                }
+                (
+                    BlocklistEntry::Other {
+                        indexer,
+                        deployment,
+                        info: _,
+                    },
+                    BlocklistEntry::Other {
+                        indexer: indexer_,
+                        deployment: deployment_,
+                        info: _,
+                    },
+                ) => (indexer == indexer_) && (deployment == deployment_),
+                _ => false,
+            }
+        }
+        self.blocklist
+            .send_modify(|blocklist| blocklist.retain(|value| !matching(entry, value)));
+    }
 }