Introduction

In today's blockchain ecosystem, querying transaction data across multiple chains has become an essential task for developers seeking to analyze trends, monitor transactions, and build cross-chain applications. Hypermanager enters as a solution to the challenges inherent in cross-chain querying. Built on Hypersync—a client that operates atop Envio's Hyperindex—Hypermanager enables seamless, efficient orchestration of cross-chain log queries across EVM networks.

Hypersync provides many powerful features, including querying transactions, blocks, logs, and traces and writing results directly to Polars DataFrames, decoding logs, and saving data as parquet files. However, re-using Hypersync queries across different applications can be tedious. Developers often need to repeat the same boilerplate code to fetch blocks, logs, and transactions, leading to redundancy and inefficiency.

Hypermanager addresses these challenges by serving as a query orchestration tool that enables reusable queries across various applications and multiple chains. By centralizing configurations and reducing boilerplate code, Hypermanager streamlines the process of cross-chain data analysis, offering developers a more efficient and scalable solution. This revised version maintains the core information while providing a bit more context and a smoother transition to the next section about Hypermanager's specific benefits.

Ease of Use

Hypermanager serves as a powerful query orchestration tool, enabling reusable queries across various applications and multiple chains. By reducing repetitive coding and centralizing configuration, it streamlines workflows and saves developers significant time and effort.

Developers new to Hypermanager will find the learning curve manageable, especially those already familiar with blockchain queries. The tool abstracts away the complexities of the Hypersync client, allowing users to focus on their data analysis tasks rather than worrying about low-level implementation details.

Hypermanager's intuitive API, comprehensive documentation, and practical examples accelerate the onboarding process for new users. The significant reduction in boilerplate code not only enhances efficiency but also lowers the entry barrier for developers venturing into blockchain data analysis.

The Power of Hypermanager: Efficiency in Action

The following example illustrates the contrast between using raw Hypersync code and Hypermanager to query blocks and transactions. Without Hypermanager, developers must continuously redefine transaction and block columns within each application, leading to verbose and redundant code.

import asyncio
import hypersync
import polars as pl
import time

from hypersync import ColumnMapping, DataType, TransactionField, BlockField, TraceField, TransactionSelection


async def historical_blocks_txs_sync():
    client = hypersync.HypersyncClient(
        hypersync.ClientConfig(url='http://eth.hypersync.xyz')
    )

    to_block: int = 20035000
    from_block: int = 20000000

    # # add +/-1 to the block range because the query is exclusive to the block number
    query = hypersync.Query(
        from_block=from_block-1,
        to_block=to_block+1,
        include_all_blocks=True,
        transactions=[TransactionSelection()],
        field_selection=hypersync.FieldSelection(
            block=[e.value for e in BlockField],
            transaction=[e.value for e in TransactionField],
        )
    )
    config = hypersync.StreamConfig(
        hex_output=hypersync.HexOutput.PREFIXED,
        column_mapping=ColumnMapping(
            transaction={
                TransactionField.GAS_USED: DataType.FLOAT64,
                TransactionField.MAX_FEE_PER_BLOB_GAS: DataType.FLOAT64,
                TransactionField.MAX_PRIORITY_FEE_PER_GAS: DataType.FLOAT64,
                TransactionField.GAS_PRICE: DataType.FLOAT64,
                TransactionField.CUMULATIVE_GAS_USED: DataType.FLOAT64,
                TransactionField.EFFECTIVE_GAS_PRICE: DataType.FLOAT64,
                TransactionField.NONCE: DataType.INT64,
                TransactionField.GAS: DataType.FLOAT64,
                TransactionField.MAX_FEE_PER_GAS: DataType.FLOAT64,
                TransactionField.MAX_FEE_PER_BLOB_GAS: DataType.FLOAT64,
                TransactionField.VALUE: DataType.FLOAT64,
            },
            block={
                BlockField.GAS_LIMIT: DataType.FLOAT64,
                BlockField.GAS_USED: DataType.FLOAT64,
                BlockField.SIZE: DataType.FLOAT64,
                BlockField.BLOB_GAS_USED: DataType.FLOAT64,
                BlockField.EXCESS_BLOB_GAS: DataType.FLOAT64,
                BlockField.BASE_FEE_PER_GAS: DataType.FLOAT64,
                BlockField.TIMESTAMP: DataType.INT64,
            }
        )
    )
	
	# save data into parquet files in "data" folder
    return await client.collect_parquet("data", query, config)

Efficiency Gains with Hypermanager:

By using Hypermanager, the same query can be completed in about 10 lines of code, reducing boilerplate by 6x and greatly improving readability:

import asyncio
import polars as pl

from hypermanager.manager import HyperManager

hypersync_client: str = "https://eth.hypersync.xyz"

async def get_events():
    manager = HyperManager(url=hypersync_client)

    df: pl.DataFrame = await manager.get_blocks(
        from_block=20000000, to_block=20035000, save_data=True
    )

if __name__ == "__main__":
    asyncio.run(get_events())

With Hypermanager, developers no longer need to manually define columns or repeatedly initialize data maps. The tool handles all standard configurations, allowing users to focus more on data analysis and customizing the data pipeline than setting all up.

Streamlined Event Decoding

When using Hypersync directly, developers must define and decode each event signature individually. Below is an example where each event signature is defined in Hypersync:

import hypersync
import asyncio


# Define events (matching the structure from the second code block)
event_1 = (
    "V3FundsDeposited(address inputToken,address outputToken,uint256 inputAmount,"
    "uint256 outputAmount,uint256 indexed destinationChainId,uint32 indexed depositId,"
    "uint32 quoteTimestamp,uint32 fillDeadline,uint32 exclusivityDeadline,"
    "address indexed depositor,address recipient,address exclusiveRelayer,bytes message)"
)
event_2 = (
    "RequestedSpeedUpV3Deposit(uint256 updatedOutputAmount,uint32 indexed depositId,"
    "address indexed depositor,address updatedRecipient,bytes updatedMessage,"
    "bytes depositorSignature)"
)
event_3 = (
    "FilledV3Relay(address inputToken,address outputToken,uint256 inputAmount,"
    "uint256 outputAmount,uint256 repaymentChainId,uint256 indexed originChainId,"
    "uint32 indexed depositId,uint32 fillDeadline,uint32 exclusivityDeadline,"
    "address exclusiveRelayer,address indexed relayer,address depositor,"
    "address recipient,bytes message,V3RelayExecutionEventInfo relayExecutionInfo)"
)
event_4 = (
    "RequestedV3SlowFill(address inputToken,address outputToken,uint256 inputAmount,"
    "uint256 outputAmount,uint256 indexed originChainId,uint32 indexed depositId,"
    "uint32 fillDeadline,uint32 exclusivityDeadline,address exclusiveRelayer,"
    "address depositor,address recipient,bytes message)"
)

# List of events
events_set = [event_1, event_2, event_3, event_4]

# Hypersync client configuration
hypersync_client_url = "https://base.hypersync.xyz"
contract = "0x09aea4b2242abC8bb4BB78D537A67a245A7bEC64"


async def main():
    # Create Hypersync client using the provided base URL
    client = hypersync.HypersyncClient(hypersync.ClientConfig(url=hypersync_client_url))

    # Lowercase the smart contract address
    smart_contract = contract.lower()

    # Iterate over the list of events
    for event in events_set:
        try:
            # Convert event signature to topic0 (hash of the event signature)
            topic = hypersync.signature_to_topic0(event)

            # Prepare the query to fetch logs for the event
            query = hypersync.preset_query_logs_of_event(
                smart_contract, topic, from_block=19_500_000
            )

            print("Running the query for event...")

            # Run the query and fetch the logs
            res = await client.get(query)

            # Check if logs were returned
            if len(res.data.logs) == 0:
                print("No logs found for event")
                continue

            # Process the logs (print the log count)
            print(f"Query returned {len(res.data.logs)} logs for event")

        except Exception as e:
            print(f"Error querying event, error: {e}")
            continue  # Continue to the next event


# Run the asynchronous main function
asyncio.run(main())

Something like the below. Plus the transactions and blocks decoding too.

       decoded_log={
                "inputAmount": DataType.FLOAT64,
                "outputAmount": DataType.FLOAT64,
                "quoteTimestamp": DataType.INT64,
                "fillDeadline": DataType.UINT64,
                "exclusivityDeadline": DataType.INT64,
                "destinationChainId": DataType.UINT64,
                "depositId": DataType.UINT64,
            },

This setup requires manual event signature definitions and individual data type configurations for each event, increasing development time and complexity.

Hypermanager’s Simplified Decoding Process:

Hypermanager centralizes event decoding, enabling developers to query events across multiple chains without needing to redefine them. This simplifies cross-chain queries, as the configuration only needs to be written once. The following Hypermanager example demonstrates querying multiple events with pre-loaded decoded values:

import asyncio
import os
import polars as pl
from hypermanager.events import EventConfig
from hypermanager.manager import HyperManager
from hypermanager.protocols.across import (
    client_config,
    across_config,
)

async def get_events():
    # Iterate through the different HyperSync clients and their associated SpokePool addresses
    for client, spoke_pool_address in client_config.items():
        print(f"Querying events for {client.name}...")
        print(f"SpokePool Address: {spoke_pool_address.value}")

        # **Modified Loop**: Iterate over the values of the across_config dictionary
        for base_event_config in across_config.values():
            try:
                manager = HyperManager(url=client.client)

                # Query events using the event configuration and return the result as a Polars DataFrame
                df: pl.DataFrame = await manager.execute_event_query(
                    base_event_config,
                    tx_data=True,
                    block_range=10_000,  # query the most recent 10,000 blocks from each chain
                )

                # Check if any events were found
                if df.is_empty():
                    print(
                        f"No events found for {base_event_config.name} on {client.name}, continuing..."
                    )
                    continue  # Move to the next event if none found


                print(f"Events found for {base_event_config.name} on {client.name}:")
                print(df.shape)  


            except Exception as e:
                print(f"Error querying {base_event_config.name} on {
                      client.name}: {e}")


if __name__ == "__main__":
    asyncio.run(get_events())

With Hypermanager, events, transactions, and blocks can all be queried with decoded values already pre-loaded. This eliminates the need for repetitive event definitions and custom data type decoding across applications.

Centralized Configuration and Reusability

One of Hypermanager’s core strengths is its centralized data decoding setup. Once configured, the event decoding information is reusable across queries Hypermanager not only reduces setup time but also makes cross-chain querying more efficient and scalable.

In summary, Hypermanager’s orchestration abilities streamline the complexity of cross-chain log queries by centralizing configurations, reducing boilerplate, and enabling developers to quickly reuse queries across applications. With Hypermanager, the barrier to efficient cross-chain analysis is lowered, making it a vital tool for anyone working with blockchain data.

Hypermanager's centralized log repository is a collaborative effort, and developers are encouraged to contribute new logs to expand its coverage and utility. By submitting pull requests with new log configurations, developers can help grow the repository, benefiting the entire community.

Hyperemanager + DuckDB Example:

To demonstrate the practicality and ease of integrating Hypermanager into a real-world data pipeline, I developed a backend MVP focused on key log events from the mev-commit chain. This proof of concept showcases the seamless integration of Hypermanager, Polars, and DuckDB, highlighting the efficiency and ergonomics of building a data pipeline for querying and storing sparse events from the mev-commit chain.

The data pipeline architecture consists of the following components:

1. Hypermanager

2. Polars

3. DuckDB

By leveraging Hypermanager, the data pipeline benefits from simplified cross-chain querying and reduced boilerplate code. Hypermanager's centralized event decoding allows for seamless retrieval of log events from the mev-commit chain without the need for repetitive event definitions.

The retrieved log data is then processed using Polars, taking advantage of its high-performance DataFrame operations. Polars' integration with DuckDB enables efficient storage and querying of the processed data within the embedded OLAP database. To streamline the deployment and testing process, I dockerized the data pipeline using Docker Compose. This allows for easy setup and execution of the pipeline with a single command: docker-compose up --build db.