feat: various new omnievent features (#33)

* feat(omnievent): properly cleanup grpc stream

* feat(omnievent): historical event occurrences filtering

* refactor(omnievent): move events filtering to new module

* feat(omnievent): add time measurement to filter

* feat(omnievent): add rusty type for BlockInfo

* feat(sql): events database for sqlite implementation

* docs(omnievent): various doc comments

* tests(omnievent/sqlite): add various tests

* docs(omnievent): various docs update

* chore: update cargo.lock

* chore(omnievent): add .sqlx queries

* chore: mark in_memory module as public

* feat(omnievent): impl From<&RegisterNewEventRequest> for uuid::Uuid

* fix(omnievent/sql): use varchar(20) for u64 integers

* fix(omnievent/grpc): rename abi to abi_bytes

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

Co-authored-by: PM <3749956+CluEleSsUK@users.noreply.github.com>

* docs(omnievent): add readme

---------

Co-authored-by: PM <3749956+CluEleSsUK@users.noreply.github.com>

Author: azixus

Cargo.lock

@@ -59,9 +59,14 @@ prost-types = { version = "0.13" }
 serde = { version = "1.0", features = ["derive"], default-features = false }
 serde_json = "1.0"
 
+# storage
+sqlx = { version = "0.8" }
+
 # misc
 anyhow = "1.0"
 base64 = "0.22"
+bytes = "1.10"
+chrono = "0.4"
 clap = { version = "4.5", features = ["derive", "env"] }
 figment = { version = "0.10" }
 hex = "0.4"

Cargo.toml

@@ -3,8 +3,13 @@ name = "omnievent"
 version.workspace = true
 edition.workspace = true
 
+[features]
+sqlite = ["sql", "sqlx/sqlite", "sqlx/uuid", "sqlx/chrono"]
+sql = ["dep:sqlx"]
+timings = []
+
 [dependencies]
-alloy = { workspace = true, features = ["default", "provider-ws"] }
+alloy = { workspace = true, features = ["default", "provider-ws", "serde"] }
 superalloy.workspace = true
 
 # async
@@ -24,8 +29,15 @@ tonic = { version = "0.13" }
 # serde
 prost.workspace = true
 prost-types.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+
+# storage
+sqlx = { workspace = true, features = ["runtime-tokio"], optional = true }
 
 # misc
+bytes.workspace = true
+chrono.workspace = true
 itertools.workspace = true
 thiserror.workspace = true
 uuid = { version = "1.0", features = ["v5"] }

omnievent/.sqlx/query-1ed67d9efd3904696b7a687bdb094f8d50f9dbb9a05769880c3f7b8ce66a0285.json

@@ -0,0 +1,72 @@
+# omnievent - a multi chain event listener
+
+## gRPC
+
+We support various operations through gRPC: register events, obtain a stream of upcoming event occurrences, and fetch historical event occurrences.
+
+One easy way to interact with the server with a cli is to rely on [`grpcurl`](https://github.com/fullstorydev/grpcurl).
+
+### Register a new event
+To register an event, you need to provide a chain id, a contract address, a requested block safety level, the name of the event and its fields.
+
+This can be done with the following command:
+```bash
+> grpcurl -import-path ./proto -proto events.proto -plaintext -d '{"chain_id": 1337, "address": "IO7wOMg7eg81fUq8ZLj2OUJ9evY=", "event_name": "StringEmitted", "fields": [{"sol_type": "string", "indexed": false}], "block_safety": "BLOCK_SAFETY_LATEST" }' 127.0.0.1:8089 events.OmniEventService/RegisterEvent
+{
+  "uuid": "ijWGFy9LUq+s2fJASjY7VQ=="
+}
+```
+
+It returns a deterministic `uuid` v5 which is obtained from the protobuf encoding of the registration request.
+
+### Stream event occurrences
+Upcoming event occurrences can be streamed as followed by specifying the event identifier:
+```bash
+> grpcurl -import-path ./proto -proto events.proto -plaintext -d '{"event_uuids": ["ijWGFy9LUq+s2fJASjY7VQ=="]}' 127.0.0.1:8089 events.OmniEventService/StreamEvents                         
+{
+  "eventUuid": "ijWGFy9LUq+s2fJASjY7VQ==",
+  "blockInfo": {
+    "blockNumber": "2",
+    "blockHash": "Ecr/C4bYj5zC69VO1nEuptsgPEOlJpxQwY3St5nQ5p4=",
+    "timestamp": "2025-07-07T16:50:51Z"
+  },
+  "rawLogData": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADEhlbGxvIFdvcmxkIQAAAAAAAAAAAAAAAAAAAAAAAAAA",
+  "eventData": [
+    {
+      "solType": "string",
+      "stringValue": "Hello World!"
+    }
+  ]
+}
+```
+
+### Obtain historical event occurrences
+To obtain past event occurrences with filtering, the following command may be used:
+```bash
+> grpcurl -import-path ./proto -proto events.proto -plaintext -d '{"event_uuids": ["ijWGFy9LUq+s2fJASjY7VQ=="], "filter": {"data_filters": [{"data_index": 0, "string": {"exact_values": ["Hello World!"]}}]}}' 127.0.0.1:8089 events.OmniEventService/GetHistoricalEvents
+{
+  "occurrences": [
+    {
+      "eventUuid": "ijWGFy9LUq+s2fJASjY7VQ==",
+      "blockInfo": {
+        "blockNumber": "2",
+        "blockHash": "Ecr/C4bYj5zC69VO1nEuptsgPEOlJpxQwY3St5nQ5p4=",
+        "timestamp": "2025-07-07T16:50:51Z"
+      },
+      "rawLogData": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADEhlbGxvIFdvcmxkIQAAAAAAAAAAAAAAAAAAAAAAAAAA",
+      "eventData": [
+        {
+          "solType": "string",
+          "stringValue": "Hello World!"
+        }
+      ]
+    }
+  ]
+}
+```
+
+## Compilation
+Run the server: TODO
+
+Due to a compile-time verification of SQL queries, you will need to specify a database url as follows when compiling the code:  
+`DATABASE_URL=sqlite:///tmp/temp.db cargo build --all-features --all-targets`

omnievent/.sqlx/query-7d637919d9e2e9efe2772c5e52cc53caf6e666671341eb0a67235836ac8b48c3.json

@@ -72,10 +72,8 @@ message GetHistoricalEventsRequest {
   // Uuid uniquely representing the event
   repeated bytes event_uuids = 1;
 
-  // Block range (required for historical queries)
-  BlockRange block_range = 2;
-
-  // Add filters (chain id, address, custom tags, values, etc) in the future
+  // Filter the event occurrences.
+  EventOccurrenceFilter filter = 2;
 }
 
 // Response with historical events
@@ -167,12 +165,61 @@ enum BlockSafety {
   BLOCK_SAFETY_SAFE = 2;
 }
 
-// Block range for queries
-message BlockRange {
+// Filter for an event occurrence that requires all conditions to be met
+message EventOccurrenceFilter {
+  optional BlockFilter block_filter = 1;
+
+  repeated OccurrenceDataFilter data_filters = 2;
+}
+
+// A filter for blocks
+message BlockFilter {
   // Start block (inclusive)
-  uint64 from_block = 1;
+  optional uint64 from_block = 1;
 
-  // Optional end block (exclusive)
-  // if not set, relies on the event's block safety or latest.
+  // End block (exclusive)
   optional uint64 to_block = 2;
 }
+
+// Filters the data from an event occurrence
+message OccurrenceDataFilter {
+  uint32 data_index = 1;
+
+  // A filter depending on the data type.
+  oneof filter {
+    StringDataFilter string = 10;
+    UintDataFilter uint = 11;
+    IntDataFilter int = 12;
+    BoolDataFilter bool = 13;
+    BytesDataFilter bytes = 14;
+    AddressDataFilter address = 15;
+    // For other types
+    BytesDataFilter abi_bytes = 19;
+  }
+}
+
+message StringDataFilter {
+  repeated string exact_values = 1;
+}
+
+message IntDataFilter {
+  // Hex-encoded 256 bits signed integers
+  repeated string exact_hex_values = 1;
+}
+
+message UintDataFilter {
+  // Hex-encoded 256 bits unsigned integers
+  repeated string exact_hex_values = 1;
+}
+
+message BoolDataFilter {
+  bool exact_value = 1;
+}
+
+message BytesDataFilter {
+  repeated bytes exact_values = 1;
+}
+
+message AddressDataFilter {
+  repeated bytes exact_values = 1;
+}

omnievent/Cargo.toml

@@ -0,0 +1,22 @@
+-- Registered events table
+CREATE TABLE IF NOT EXISTS registered_events (
+    id UUID PRIMARY KEY NOT NULL,
+    chain_id VARCHAR(20) NOT NULL, -- can't completely store u64 in INTEGER. 20 digits for int repr.
+    address BLOB NOT NULL,
+    block_safety INTEGER NOT NULL,
+    event_name TEXT NOT NULL,
+    fields_json TEXT NOT NULL  -- json encoded fields
+);
+
+-- Event occurrences table
+CREATE TABLE IF NOT EXISTS event_occurrences (
+    id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+    event_id UUID NOT NULL,
+    block_number VARCHAR(20) NOT NULL, -- can't completely store u64 in INTEGER. 20 digits for int repr.
+    block_hash BYTES NOT NULL,
+    block_timestamp DATETIME NOT NULL,
+    raw_log_json TEXT NOT NULL,
+    fields_json TEXT NOT NULL,  -- json encoded fields
+    FOREIGN KEY (event_id) REFERENCES registered_events(id) ON DELETE CASCADE
+);
+

omnievent/README.md

@@ -3,6 +3,7 @@
 
 pub mod db;
 mod events_occurrence;
+mod filtering;
 pub(crate) mod listener;
 mod register;
 
@@ -11,6 +12,7 @@ use crate::event_manager::events_occurrence::HandleEventsOccurrenceTask;
 use crate::event_manager::listener::{
     EventListener, EventListenerHandle, EventReceiverHandleError,
 };
+use crate::proto_types::EventOccurrenceFilter;
 use crate::types::{EventFieldData, EventId, EventOccurrence, ParsedRegisterNewEventRequest};
 use alloy::rpc::types::Log;
 use futures_util::stream::SelectAll;
@@ -70,9 +72,14 @@ pub(crate) enum EventManagerError {
 
     #[error("database error")]
     Database(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
+
+    #[error("filter error")]
+    Filter(#[from] FilterError),
 }
 
-// export [`CreateStreamError`] since it's used in [`EventManagerError`]
+// export other event_manager's module errors
+use crate::event_manager::filtering::filter_occurrences;
+pub(crate) use filtering::FilterError;
 pub(crate) use register::CreateStreamError;
 
 impl<MP, DB> EventManager<MP, DB>
@@ -191,26 +198,85 @@ where
 
     pub(crate) async fn get_ethereum_multi_event_stream(
         &self,
-        events_id: impl IntoIterator<Item = EventId>,
+        event_ids: impl IntoIterator<Item = EventId>,
     ) -> Result<SelectAll<BroadcastStream<EventOccurrence>>, EventManagerError> {
         // TODO: n locks, not great, improve
         let streams = futures::future::try_join_all(
-            events_id
+            event_ids
                 .into_iter()
                 .map(|id| self.get_ethereum_event_stream(id)),
         )
         .await?;
         Ok(futures::stream::select_all(streams))
     }
 
+    pub(crate) async fn unregister_ethereum_multi_event_stream(
+        &self,
+        event_ids: impl IntoIterator<Item = EventId>,
+        stream: SelectAll<BroadcastStream<EventOccurrence>>, // take ownership to make sure that it's dropped properly
+    ) {
+        // Drop the stream (i.e., the receiver side of the broadcast channel)
+        drop(stream);
+
+        let mut active_events_map = self.active_events_map.write().await;
+        for event_id in event_ids {
+            let _span =
+                tracing::info_span!("unregister_ethereum_multi_event_stream", event_id = ?event_id)
+                    .entered();
+
+            let Some(entry) = active_events_map.get_mut(&event_id) else {
+                tracing::error!("Attempting to unregister a stream not in active_events_map");
+                continue;
+            };
+
+            let Some(outgoing_stream) = &entry.outgoing_stream else {
+                tracing::debug!("Attempting to unregister a stream that was already removed");
+                continue;
+            };
+
+            // If there are no more receivers, remove the outgoing stream
+            let rx_count = outgoing_stream.receiver_count();
+            if rx_count == 0 {
+                tracing::debug!("Removing broadcast channel");
+                entry.outgoing_stream = None;
+            } else {
+                tracing::debug!(
+                    rx_count,
+                    "Leaving broadcast channel open due to remaining receivers"
+                );
+            }
+        }
+    }
+
+    /// Get a vector of historical event occurrences, optionally with filtering enabled.
+    /// TODO: Currently, the function uses a very naive approach to filtering. It fetches every events
+    /// from the database, and then applies filters on top. This needs to be reworked once we have
+    /// more concrete usage, to know which filters are the most important so that they can be offloaded
+    /// to the database implementation.
     pub(crate) async fn get_historical_event_occurrences(
         &self,
         event_ids: impl IntoIterator<Item = EventId> + Send,
+        filter: Option<EventOccurrenceFilter>,
     ) -> Result<Vec<EventOccurrence>, EventManagerError> {
-        self.events_db
+        let mut occurrences = self
+            .events_db
             .get_event_occurrences(event_ids)
             .await
-            .map_err(|e| EventManagerError::Database(Box::new(e)))
+            .map_err(|e| EventManagerError::Database(Box::new(e)))?;
+
+        tracing::debug!(
+            n_occurrences = occurrences.len(),
+            "Obtained occurrences from database"
+        );
+
+        if let Some(filter) = filter {
+            occurrences = filter_occurrences(occurrences, filter).map_err(|e| {
+                tracing::info!(error = ?e, "Failed to apply filters to occurrences");
+                e
+            })?;
+        }
+
+        Ok(occurrences)
     }
 }
 

omnievent/proto/events.proto

@@ -2,7 +2,10 @@
 
 use crate::types::{EventId, EventOccurrence, RegisteredEvent};
 
-pub(crate) mod in_memory;
+pub mod in_memory;
+
+#[cfg(feature = "sql")]
+pub mod sql;
 
 pub trait EventsDatabase {
     type Error: std::error::Error + Send + Sync + 'static;