indexer: support TransactionFilter::AffectedObject (#19614)

## Description

Add an `AffectedObject` filter on `TransactionFilter` and implement it
in `IndexerReader` and in the subscription system.

This filter is not implemented on fullnode indices, and in a follow-up
PR, support for `InputObject` and `ChangedObject` will be removed from
the PG-backed implementation of JSON-RPC.

It was difficult to replace `InputObject`/`ChangedObject` support on
fullnode with `AffectedObject` because we don't have (and don't want to
add) the necessary indices, and the current filters are referred to in
certain tests.

## Test plan

Manually tested with a local network:

```
sui$ cargo run --bin sui -- start --with-indexer --force-regenesis
```

```
sui$ curl -LX POST  "http://localhost:9124" \
        --header 'Content-Type: application/json' \
        --data '{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "suix_queryTransactionBlocks",
  "params": [
    { "filter": { "AffectedObject":"0x2" } },
    null,
    50,
    true
  ]
}' | jq '.result.data.[].digest'
```

## Stack

- #19474 

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [x] Indexer: Adds support for filtering transactions by their affected
object
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

crates/sui-indexer/src/indexer_reader.rs

@@ -739,6 +739,13 @@ impl IndexerReader {
  format!("object_id = '\\x{}'::bytea", object_id),
  )
  }
+ Some(TransactionFilter::AffectedObject(object_id)) => {
+ let object_id = Hex::encode(object_id.to_vec());
+ (
+ "tx_affected_objects".into(),
+ format!("affected = '\\x{}'::bytea", object_id),
+ )
+ }
  Some(TransactionFilter::FromAddress(from_address)) => {
  let from_address = Hex::encode(from_address.to_vec());
  (

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

@@ -2361,6 +2361,8 @@ pub enum TransactionFilter {
  InputObject(ObjectID),
  /// Query by changed object, including created, mutated and unwrapped objects.
  ChangedObject(ObjectID),
+ /// Query for transactions that touch this object.
+ AffectedObject(ObjectID),
  /// Query by sender address.
  FromAddress(SuiAddress),
  /// Query by recipient address.
@@ -2390,6 +2392,18 @@ impl Filter<EffectsWithInput> for TransactionFilter {
  .mutated()
  .iter()
  .any(|oref: &OwnedObjectRef| &oref.reference.object_id == o),
+ TransactionFilter::AffectedObject(o) => item
+ .effects
+ .created()
+ .iter()
+ .chain(item.effects.mutated().iter())
+ .chain(item.effects.unwrapped().iter())
+ .map(|oref: &OwnedObjectRef| &oref.reference)
+ .chain(item.effects.shared_objects().iter())
+ .chain(item.effects.deleted().iter())
+ .chain(item.effects.unwrapped_then_deleted().iter())
+ .chain(item.effects.wrapped().iter())
+ .any(|oref| &oref.object_id == o),
  TransactionFilter::FromAddress(a) => &item.input.sender() == a,
  TransactionFilter::ToAddress(a) => {
  let mutated: &[OwnedObjectRef] = item.effects.mutated();

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

@@ -10862,6 +10862,19 @@
  },
  "additionalProperties": false
  },
+ {
+ "description": "Query for transactions that touch this object.",
+ "type": "object",
+ "required": [
+ "AffectedObject"
+ ],
+ "properties": {
+ "AffectedObject": {
+ "$ref": "#/components/schemas/ObjectID"
+ }
+ },
+ "additionalProperties": false
+ },
  {
  "description": "Query by sender address.",
  "type": "object",