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

# Description of change

This PR is the first iteration attempt to provide streaming support for
the `iota-indexer` as a library componente. Currently it supports
streaming of `StoredEvents` and `StoredTransactions`. It also supports
filters.

## Links to any relevant issues

fixes #8315 

## How the change has been tested
created a separated binary which used the `iota-indexer` as a library,
used `testnet` network as source of checkpoints for synchronization,
subscribed to incoming events, applied different filter to assert that
they were respected.

- [x] Basic tests (linting, compilation, formatting, unit/integration
tests)
- [x] Patch-specific tests (correctness, functionality coverage)

### Infrastructure QA (only required for crates that are maintained by
@iotaledger/infrastructure)

- [ ] Synchronization of the indexer from genesis for a network
including migration objects.
- [x] Restart of indexer synchronization locally without resetting the
database.
- [ ] Restart of indexer synchronization on a production-like database.
- [ ] Deployment of services using Docker.
- [x] Verification of API backward compatibility.

### Release Notes
- [x] Indexer: Add streaming support for `events` and `transactions`.

Author: sergiupopescu199

Cargo.lock

@@ -28,13 +28,15 @@ prometheus.workspace = true
 rand = { workspace = true, optional = true }
 secrecy = "0.8.0"
 serde.workspace = true
+serde_json.workspace = true
 serde_with.workspace = true
 strum.workspace = true
 strum_macros.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"] }
 toml.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

@@ -44,6 +44,7 @@ pub mod pruning;
 pub mod read;
 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

@@ -455,7 +455,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!(

crates/iota-indexer/src/models/transactions.rs

@@ -870,6 +870,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,