Apibara Direct Node Access (DNA)

To receive a continuous stream of DnaMessage objects, follow these steps:

1. Use StreamDataRequestBuilder::new() to initialize a request.
2. Configure the request using .with_starting_cursor(start_block) to define where the stream begins.
3. Add filters using .add_filter(). Filters can be constructed using starknet::FilterBuilder and starknet::EventFilterBuilder to target specific contract addresses, event keys, or transactions.
4. Call client.stream_data(request).await? to obtain a stream.
5. Iterate over the stream using .try_next().await? to process each DnaMessage.

let usdc_address = starknet::FieldElement::from_hex(
    "0x053c91253bc9682c04929ca02ed00b3e423f6710d2ee7e0d5ebb06f3ecf368a8",
)?;
let transfer_hash = starknet::FieldElement::from_hex(
    "0x99cd8bde557814842a3121e8ddfd433a539b8c9f14bf31ebf108d12e6196e9",
)?;
let request = StreamDataRequestBuilder::new()
    .with_starting_cursor(start_block)
    .add_filter(
        starknet::FilterBuilder::new()
            .add_event(
                starknet::EventFilterBuilder::single_contract(usdc_address)
                    .with_keys(true, vec![Some(transfer_hash)])
                    .with_transaction()
                    .build(),
            )
            .build(),
    )
    .build();

let mut stream = client.stream_data(request).await?;

while let Some(message) = stream.try_next().await? {
    // Process the message here
}

When ingesting Starknet blocks, the system processes a StateDiff and produces a BlockIngestionResult. This result is composed of three primary data categories, each containing a body (the raw data), an index (for efficient lookups), and a join fragment:

1. Storage Diffs: Contains storage_diffs. It is indexed by contract_address using INDEX_STORAGE_DIFF_BY_CONTRACT_ADDRESS.
2. Contract Changes: Contains contract_changes (including DeclaredClass, ReplacedClass, and DeployedContract). It is indexed by change type using INDEX_CONTRACT_CHANGE_BY_TYPE.
3. Nonce Updates: Contains nonces. It is indexed by contract_address using INDEX_NONCE_UPDATE_BY_CONTRACT_ADDRESS.

Each category uses specific fragment IDs and names (e.g., STORAGE_DIFF_FRAGMENT_ID, CONTRACT_CHANGE_FRAGMENT_ID, NONCE_UPDATE_FRAGMENT_ID) to organize the data into BodyFragment, IndexFragment, and JoinFragment structures.

The Starknet SDK uses a hierarchical builder pattern to construct complex data filters. The FilterBuilder acts as the root, allowing you to compose multiple specialized filters into a single request.

To build a filter, you typically:

1. Initialize a FilterBuilder using FilterBuilder::new().
2. Use specialized builders (like TransactionFilterBuilder or EventFilterBuilder) to create specific filter types.
3. Add these specialized filters to the root builder using methods like add_transaction or add_event.
4. Call .build() on the root builder to produce the final Filter object.

This composition allows you to request transactions, events, messages to L1, storage diffs, contract changes, and nonce updates in a single unified stream.

let filter = FilterBuilder::new()
    .add_transaction(TransactionFilterBuilder::only_invoke_v3().build())
    .add_event(EventFilterBuilder::single_contract(contract_address).build())
    .build();