feat(omnievent): repeat chain_id and contract address in event occurrence (#34)

Author: azixus

omnievent/proto/events.proto

@@ -115,6 +115,12 @@ message EventOccurrence {
 
   // Event data (decoded parameters)
   repeated EventData event_data = 4;
+
+  // Chain id
+  uint64 chain_id = 5;
+
+  // Address of the contract
+  bytes address = 6;
 }
 
 // Block information

omnievent/sql/schema.sql

@@ -20,3 +20,18 @@ CREATE TABLE IF NOT EXISTS event_occurrences (
     FOREIGN KEY (event_id) REFERENCES registered_events(id) ON DELETE CASCADE
 );
 
+-- View that combines event_occurrences with registered_events data
+CREATE VIEW event_occurrences_with_context AS
+SELECT
+    occurrence.id,
+    occurrence.event_id,
+    occurrence.block_number,
+    occurrence.block_hash,
+    occurrence.block_timestamp,
+    occurrence.raw_log_json,
+    occurrence.fields_json,
+    event.chain_id,
+    event.address,
+    event.event_name
+FROM event_occurrences occurrence
+    INNER JOIN registered_events event ON occurrence.event_id = event.id;

omnievent/src/event_manager.rs

@@ -14,6 +14,7 @@ use crate::event_manager::listener::{
 };
 use crate::proto_types::EventOccurrenceFilter;
 use crate::types::{EventFieldData, EventId, EventOccurrence, ParsedRegisterNewEventRequest};
+use alloy::primitives::Address;
 use alloy::rpc::types::Log;
 use futures_util::stream::SelectAll;
 use std::collections::HashMap;
@@ -29,6 +30,8 @@ const BROADCAST_STREAM_CAPACITY: usize = 64;
 #[derive(Clone, Debug)]
 pub(crate) struct DecodedEvent {
     event_id: EventId,
+    chain_id: u64,
+    address: Address,
     data: Vec<EventFieldData>,
     log: Log,
 }

omnievent/src/event_manager/db/sql/sqlite.rs

@@ -2,6 +2,7 @@
 
 use crate::event_manager::db::EventsDatabase;
 use crate::types::{BlockInfo, EventId, EventOccurrence, RegisteredEvent};
+use alloy::primitives::Address;
 use chrono::{DateTime, Utc};
 use sqlx::{FromRow, QueryBuilder, Row, Sqlite, SqlitePool};
 use std::str::FromStr;
@@ -169,7 +170,7 @@ impl EventsDatabase for SqliteEventDatabase {
         };
 
         let mut query_builder: QueryBuilder<Sqlite> =
-            QueryBuilder::new("SELECT * FROM event_occurrences WHERE event_id IN (");
+            QueryBuilder::new("SELECT * FROM event_occurrences_with_context WHERE event_id IN (");
         let mut separated = query_builder.separated(", ");
         for id in event_ids {
             separated.push_bind(Uuid::from(id));
@@ -195,12 +196,23 @@ impl From<(sqlx::Error, &'static str)> for SqliteEventDatabaseError {
 impl<'r> FromRow<'r, sqlx::sqlite::SqliteRow> for EventOccurrence {
     fn from_row(row: &'r sqlx::sqlite::SqliteRow) -> Result<Self, sqlx::Error> {
         let event_id: Uuid = row.try_get("event_id")?;
+        let address: Vec<u8> = row.try_get("address")?;
+        let chain_id_str: String = row.try_get("chain_id")?;
         let block_number_str: String = row.try_get("block_number")?;
         let block_hash: Vec<u8> = row.try_get("block_hash")?;
         let block_timestamp: DateTime<Utc> = row.try_get("block_timestamp")?;
         let raw_log_json: String = row.try_get("raw_log_json")?;
         let fields_json: String = row.try_get("fields_json")?;
 
+        let address =
+            Address::try_from(address.as_slice()).map_err(|e| sqlx::Error::ColumnDecode {
+                index: "address".to_owned(),
+                source: Box::new(e),
+            })?;
+        let chain_id = u64::from_str(&chain_id_str).map_err(|e| sqlx::Error::ColumnDecode {
+            index: "chain_id".to_owned(),
+            source: Box::new(e),
+        })?;
         let block_number =
             u64::from_str(&block_number_str).map_err(|e| sqlx::Error::ColumnDecode {
                 index: "block_number".to_owned(),
@@ -218,6 +230,8 @@ impl<'r> FromRow<'r, sqlx::sqlite::SqliteRow> for EventOccurrence {
 
         Ok(Self {
             event_id: event_id.into(),
+            address,
+            chain_id,
             block_info: BlockInfo {
                 number: block_number,
                 hash: block_hash.into(),
@@ -285,6 +299,8 @@ mod tests {
         let res = db
             .store_event_occurrence(EventOccurrence {
                 event_id: EventId::new(b"invalid event id"),
+                address: Default::default(),
+                chain_id: 0,
                 data: vec![],
                 raw_log: LogData::empty(),
                 block_info: BlockInfo {
@@ -325,6 +341,8 @@ mod tests {
         let res = db
             .store_event_occurrence(EventOccurrence {
                 event_id,
+                address: Default::default(),
+                chain_id: 0,
                 data: vec![],
                 raw_log: LogData::empty(),
                 block_info: BlockInfo {
@@ -364,6 +382,8 @@ mod tests {
 
         let occurrence = EventOccurrence {
             event_id,
+            address: Default::default(),
+            chain_id: 0,
             data: vec![],
             raw_log: LogData::empty(),
             block_info: BlockInfo {

omnievent/src/event_manager/events_occurrence.rs

@@ -121,6 +121,8 @@ fn event_occurrence_from_decoded_event(decoded_event: DecodedEvent) -> EventOccu
 
     EventOccurrence {
         event_id: decoded_event.event_id,
+        chain_id: decoded_event.chain_id,
+        address: decoded_event.address,
         raw_log: decoded_event.log.data().to_owned(),
         data: decoded_event.data,
         block_info: BlockInfo {

omnievent/src/event_manager/listener.rs

@@ -194,6 +194,8 @@ impl EventListener {
                     // Send decoded event through channel
                     if sender.send(DecodedEvent {
                         event_id,
+                        address: event.address,
+                        chain_id: event.chain_id,
                         data: decoded_fields,
                         log,
                     }).await.is_err() {

omnievent/src/types.rs

@@ -258,6 +258,8 @@ impl From<BlockInfo> for proto_types::BlockInfo {
 #[derive(Clone, Debug, PartialEq)]
 pub struct EventOccurrence {
     pub event_id: EventId,
+    pub chain_id: u64,
+    pub address: Address,
     pub block_info: BlockInfo,
     pub raw_log: LogData,
     pub data: Vec<EventFieldData>,
@@ -277,6 +279,8 @@ impl From<EventOccurrence> for proto_types::EventOccurrence {
 
         Self {
             event_uuid: event.event_id.into(),
+            address: event.address.to_vec().into(),
+            chain_id: event.chain_id,
             event_data: data,
             raw_log_data: Some(event.raw_log.data.into()),
             block_info: Some(event.block_info.into()),