feat: add a cutoff date for aggregated topic processing (#116)

* feat: add dry run mode to skip contract calls

Signed-off-by: Tomás Migone <tomas@edgeandnode.com>

* ci: formatter was sad

Signed-off-by: Tomás Migone <tomas@edgeandnode.com>

* fix: revert previous hacky fix to ignore pre-horizon data

Signed-off-by: Tomás Migone <tomas@edgeandnode.com>

* feat: add cutoff date for aggregated topic

Signed-off-by: Tomás Migone <tomas@edgeandnode.com>

---------

Signed-off-by: Tomás Migone <tomas@edgeandnode.com>

Author: tmigone

src/config.rs

@@ -47,4 +47,7 @@ pub struct Kafka {
     pub config: BTreeMap<String, String>,
     pub realtime_topic: String,
     pub aggregated_topic: Option<String>,
+    /// Cutoff timestamp (unix milliseconds) for aggregated topic data.
+    /// Aggregated records older than this are ignored.
+    pub aggregated_cutoff_timestamp: Option<i64>,
 }

src/kafka.rs

@@ -66,14 +66,22 @@ mod receipts {
                     .with_context(|| anyhow!("missing payload at {partition} {offset}"))?;
                 let msg = IndexerFeesHourlyProtobuf::decode(payload)?;
                 latest_aggregated_timestamp = latest_aggregated_timestamp.max(msg.timestamp);
+                if let Some(cutoff) = config.aggregated_cutoff_timestamp {
+                    if msg.timestamp < cutoff {
+                        continue;
+                    }
+                }
                 for aggregation in &msg.aggregations {
                     if !signers.contains(&Address::from_slice(&aggregation.signer)) {
                         continue;
                     }
-                    // Aggregated topic doesn't include allocation, skip these entries
-                    // as we can't determine if they're from legacy allocations.
-                    // The realtime topic will provide allocation-level data.
-                    let _ = (msg.timestamp, &aggregation.receiver, aggregation.fee_grt);
+                    let update = Update {
+                        timestamp: DateTime::from_timestamp_millis(msg.timestamp)
+                            .context("timestamp out of range")?,
+                        indexer: Address::from_slice(&aggregation.receiver),
+                        fee: (aggregation.fee_grt * 1e18) as u128,
+                    };
+                    db.send(update).await.unwrap();
                 }
 
                 if latest_aggregated_offsets.get(&partition).unwrap() == &offset {
@@ -133,9 +141,6 @@ mod receipts {
         /// 20 bytes (address)
         #[prost(bytes, tag = "1")]
         indexer: Vec<u8>,
-        /// 20 bytes (address)
-        #[prost(bytes, tag = "3")]
-        allocation: Vec<u8>,
         #[prost(double, tag = "6")]
         fee_grt: f64,
     }
@@ -177,7 +182,7 @@ mod receipts {
                 for indexer_query in payload.indexer_queries {
                     let update = Update {
                         timestamp,
-                        allocation: Address::from_slice(&indexer_query.allocation),
+                        indexer: Address::from_slice(&indexer_query.indexer),
                         fee: (indexer_query.fee_grt * 1e18) as u128,
                     };
                     let _ = db.send(update).await;
@@ -189,11 +194,12 @@ mod receipts {
 
     pub struct Update {
         pub timestamp: DateTime<Utc>,
-        pub allocation: Address,
+        pub indexer: Address,
         pub fee: u128,
     }
 
     pub struct DB {
+        // indexer debts, aggregated per hour
         data: BTreeMap<Address, BTreeMap<i64, u128>>,
         window: Duration,
         tx: watch::Sender<BTreeMap<Address, u128>>,
@@ -239,7 +245,7 @@ mod receipts {
             }
             let entry = self
                 .data
-                .entry(update.allocation)
+                .entry(update.indexer)
                 .or_default()
                 .entry(hourly_timestamp(update.timestamp))
                 .or_default();
@@ -257,7 +263,7 @@ mod receipts {
         fn snapshot(&self) -> BTreeMap<Address, u128> {
             self.data
                 .iter()
-                .map(|(allocation, entries)| (*allocation, entries.values().sum()))
+                .map(|(indexer, entries)| (*indexer, entries.values().sum()))
                 .collect()
         }
     }

src/main.rs

@@ -149,24 +149,20 @@ async fn main() -> anyhow::Result<()> {
         tracing::debug!(receivers = receivers.len());
 
         let mut indexer_ravs: BTreeMap<Address, u128> = Default::default();
-        let mut indexer_receipts: BTreeMap<Address, u128> = Default::default();
         {
             let allocation_ravs = ravs.borrow();
-            let allocation_receipts = receipts.borrow();
-            for allocation in &allocations {
+            for allocation in allocations {
                 if let Some(value) = allocation_ravs.get(&allocation.id) {
                     *indexer_ravs.entry(allocation.indexer).or_default() += *value;
                 }
-                if let Some(value) = allocation_receipts.get(&allocation.id) {
-                    *indexer_receipts.entry(allocation.indexer).or_default() += *value;
-                }
             }
         }
 
         let mut debts: BTreeMap<Address, u128> = Default::default();
         {
+            let receipts = receipts.borrow();
             for receiver in &receivers {
-                let receipts = *indexer_receipts.get(receiver).unwrap_or(&0);
+                let receipts = *receipts.get(receiver).unwrap_or(&0);
                 let ravs = *indexer_ravs.get(receiver).unwrap_or(&0);
                 debts.insert(*receiver, u128::max(receipts, ravs));
                 tracing::info!(