indexer: FromOrTo tx filter (MystenLabs#12251)

## Description 

This PR added a tx filter
```
    /// Query txs that have a given address as sender or recipient.
    FromOrToAddress { addr: SuiAddress },
```
such that on Explorer etc. the address page could fetch "related txs"
efficiently.

## Test Plan 

Local run network + indexer, query tx with new filter.

---
If your changes are not user-facing and not a breaking change, you can
skip the following section. Otherwise, please indicate what changed, and
then add to the Release Notes section as highlighted during the release
process.

### Type of Change (Check all that apply)

- [ ] protocol change
- [x] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

crates/sui-indexer/migrations/2023-03-05-165748_transaction_index/up.sql

@@ -22,6 +22,7 @@ CREATE TABLE recipients (
     recipient                   address         NOT NULL
 );
 CREATE INDEX recipients_transaction_digest ON recipients (transaction_digest);
+CREATE INDEX recipients_sender ON recipients (sender);
 CREATE INDEX recipients_recipient ON recipients (recipient);
 
 CREATE TABLE input_objects (

crates/sui-indexer/src/apis/indexer_api.rs

@@ -186,7 +186,7 @@ impl<S: IndexerStore> IndexerApi<S> {
                     .get_recipient_sequence_by_digest(cursor_str, is_descending)
                     .await?;
                 self.state
-                    .get_transaction_page_by_sender_recipient_address(
+                    .get_transaction_page_by_recipient_address(
                         /* from */ None,
                         recipient_address,
                         recipient_seq_number,
@@ -201,7 +201,7 @@ impl<S: IndexerStore> IndexerApi<S> {
                     .get_recipient_sequence_by_digest(cursor_str, is_descending)
                     .await?;
                 self.state
-                    .get_transaction_page_by_sender_recipient_address(
+                    .get_transaction_page_by_recipient_address(
                         Some(from),
                         to,
                         recipient_seq_number,
@@ -210,6 +210,15 @@ impl<S: IndexerStore> IndexerApi<S> {
                     )
                     .await
             }
+            Some(TransactionFilter::FromOrToAddress { addr }) => {
+                let start_sequence = self
+                    .state
+                    .get_recipient_sequence_by_digest(cursor_str, is_descending)
+                    .await?;
+                self.state
+                    .get_transaction_page_by_address(addr, start_sequence, limit, is_descending)
+                    .await
+            }
             Some(TransactionFilter::TransactionKind(tx_kind_name)) => {
                 let indexer_seq_number = self
                     .state

crates/sui-indexer/src/store/indexer_store.rs

@@ -132,7 +132,7 @@ pub trait IndexerStore {
         is_descending: bool,
     ) -> Result<Vec<Transaction>, IndexerError>;
 
-    async fn get_transaction_page_by_sender_recipient_address(
+    async fn get_transaction_page_by_recipient_address(
         &self,
         sender_address: Option<SuiAddress>,
         recipient_address: SuiAddress,
@@ -141,6 +141,15 @@ pub trait IndexerStore {
         is_descending: bool,
     ) -> Result<Vec<Transaction>, IndexerError>;
 
+    // `address` can be either sender or recipient address of the transaction
+    async fn get_transaction_page_by_address(
+        &self,
+        address: SuiAddress,
+        start_sequence: Option<i64>,
+        limit: usize,
+        is_descending: bool,
+    ) -> Result<Vec<Transaction>, IndexerError>;
+
     async fn get_transaction_page_by_input_object(
         &self,
         object_id: ObjectID,

crates/sui-indexer/src/store/pg_indexer_store.rs

@@ -1047,7 +1047,7 @@ impl IndexerStore for PgIndexerStore {
         self.multi_get_transactions_by_digests(&tx_digests).await
     }
 
-    async fn get_transaction_page_by_sender_recipient_address(
+    async fn get_transaction_page_by_recipient_address(
         &self,
         from: Option<SuiAddress>,
         to: SuiAddress,
@@ -1088,6 +1088,43 @@ impl IndexerStore for PgIndexerStore {
         self.multi_get_transactions_by_digests(&tx_digests).await
     }
 
+    async fn get_transaction_page_by_address(
+        &self,
+        address: SuiAddress,
+        start_sequence: Option<i64>,
+        limit: usize,
+        is_descending: bool,
+    ) -> Result<Vec<Transaction>, IndexerError> {
+        let sql_query = format!(
+            "SELECT transaction_digest as digest_name FROM (
+                SELECT transaction_digest, max(id) AS max_id
+                FROM recipients
+                WHERE recipient = '{}' OR sender = '{}'
+                {} GROUP BY transaction_digest
+                ORDER BY max_id {} LIMIT {}
+            ) AS t",
+            address,
+            address,
+            if let Some(start_sequence) = start_sequence {
+                if is_descending {
+                    format!("AND id < {}", start_sequence)
+                } else {
+                    format!("AND id > {}", start_sequence)
+                }
+            } else {
+                "".to_string()
+            },
+            if is_descending { "DESC" } else { "ASC" },
+            limit
+        );
+        let tx_digests: Vec<String> = read_only_blocking!(&self.blocking_cp, |conn| diesel::sql_query(sql_query).load(conn))
+                .context(&format!("Failed reading transaction digests by address {address} with start_sequence {start_sequence:?} and limit {limit}"))?
+                .into_iter()
+                .map(|table: TempDigestTable| table.digest_name)
+                .collect();
+        self.multi_get_transactions_by_digests(&tx_digests).await
+    }
+
     async fn get_network_metrics(&self) -> Result<NetworkMetrics, IndexerError> {
         get_network_metrics_cached(&self.blocking_cp).await
     }

crates/sui-json-rpc-types/src/sui_transaction.rs

@@ -1667,6 +1667,8 @@ pub enum TransactionFilter {
     ToAddress(SuiAddress),
     /// Query by sender and recipient address.
     FromAndToAddress { from: SuiAddress, to: SuiAddress },
+    /// Query txs that have a given address as sender or recipient.
+    FromOrToAddress { addr: SuiAddress },
     /// Query by transaction kind
     TransactionKind(String),
 }
@@ -1707,6 +1709,7 @@ impl Filter<EffectsWithInput> for TransactionFilter {
             TransactionFilter::TransactionKind(kind) => item.input.kind().to_string() == *kind,
             // these filters are not supported, rpc will reject these filters on subscription
             TransactionFilter::Checkpoint(_) => false,
+            TransactionFilter::FromOrToAddress { addr: _ } => false,
         }
     }
 }

crates/sui-open-rpc/spec/openrpc.json

@@ -9983,6 +9983,27 @@
             },
             "additionalProperties": false
           },
+          {
+            "description": "Query txs that have a given address as sender or recipient.",
+            "type": "object",
+            "required": [
+              "FromOrToAddress"
+            ],
+            "properties": {
+              "FromOrToAddress": {
+                "type": "object",
+                "required": [
+                  "addr"
+                ],
+                "properties": {
+                  "addr": {
+                    "$ref": "#/components/schemas/SuiAddress"
+                  }
+                }
+              }
+            },
+            "additionalProperties": false
+          },
           {
             "description": "Query by transaction kind",
             "type": "object",