feat(indexer): add streaming support for events and transactions

Author: sergiupopescu199

Cargo.lock

@@ -28,11 +28,13 @@ prometheus.workspace = true
 rand = { workspace = true, optional = true }
 secrecy = "0.8.0"
 serde.workspace = true
+serde_json.workspace = true
 serde_with.workspace = true
 tap.workspace = true
 tempfile.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["full"] }
+tokio-postgres = "0.7.13"
 tokio-stream.workspace = true
 tokio-util = { workspace = true, features = ["rt"] }
 tracing.workspace = true

crates/iota-indexer/Cargo.toml

@@ -0,0 +1,2 @@
+DROP TRIGGER IF EXISTS checkpoint_committed_trigger ON checkpoints;
+DROP FUNCTION IF EXISTS notify_checkpoint_committed();

crates/iota-indexer/migrations/pg/2025-09-10-071522_events_and_transactions_triggers/down.sql

@@ -0,0 +1,21 @@
+CREATE OR REPLACE FUNCTION notify_checkpoint_committed()
+RETURNS TRIGGER AS $$
+BEGIN
+    -- Send notification with just the range - let client query events
+    PERFORM pg_notify('checkpoint_committed',
+        json_build_object(
+            'checkpoint_sequence_number', NEW.sequence_number,
+            'min_tx_sequence_number', NEW.min_tx_sequence_number,
+            'max_tx_sequence_number', NEW.max_tx_sequence_number
+        )::text
+    );
+
+    RETURN NULL;
+END;
+$$ LANGUAGE plpgsql;
+
+
+CREATE TRIGGER checkpoint_committed_trigger
+    AFTER INSERT ON checkpoints
+    FOR EACH ROW
+    EXECUTE FUNCTION notify_checkpoint_committed();

crates/iota-indexer/migrations/pg/2025-09-10-071522_events_and_transactions_triggers/up.sql

@@ -869,6 +869,20 @@ impl IndexerReader {
         run_query!(&self.pool, |conn| query.load::<StoredTransaction>(conn))
     }
 
+    pub(crate) fn multi_get_transactions_by_sequence_numbers_range(
+        &self,
+        min_seq: i64,
+        max_seq: i64,
+    ) -> Result<Vec<StoredTransaction>, IndexerError> {
+        use crate::schema::transactions::dsl as txdsl;
+        let query = txdsl::transactions
+            .filter(txdsl::tx_sequence_number.ge(min_seq))
+            .filter(txdsl::tx_sequence_number.le(max_seq))
+            .order(txdsl::tx_sequence_number.asc())
+            .into_boxed();
+        run_query!(&self.pool, |conn| query.load::<StoredTransaction>(conn))
+    }
+
     pub async fn get_owned_objects_in_blocking_task(
         &self,
         address: IotaAddress,

crates/iota-indexer/src/indexer_reader.rs

@@ -44,6 +44,7 @@ pub mod processors;
 pub(crate) mod rolling;
 pub mod schema;
 pub mod store;
+pub mod stream;
 pub mod system_package_task;
 pub mod test_utils;
 pub mod types;

crates/iota-indexer/src/lib.rs

@@ -431,7 +431,7 @@ impl StoredTransaction {
         })
     }
 
-    fn try_into_sender_signed_data(&self) -> IndexerResult<SenderSignedData> {
+    pub(crate) fn try_into_sender_signed_data(&self) -> IndexerResult<SenderSignedData> {
         let sender_signed_data: SenderSignedData =
             bcs::from_bytes(&self.raw_transaction).map_err(|e| {
                 IndexerError::PersistentStorageDataCorruption(format!(