Merge pull request #193 from Spherre-Labs/feat/implement-indexer

Implement apibara indexer

Author: codebestia

backend/spherre/indexer/config.py

@@ -0,0 +1,8 @@
+import os
+
+NETWORK = "starknet-sepolia"
+
+SPHERRE_CONTRACT_ADDRESS = ""
+SERVER_URL = ""
+MONGO_URL = ""
+DNA_TOKEN = os.environ.get("DNA_TOKEN")

backend/spherre/indexer/main.py

@@ -0,0 +1,42 @@
+import asyncio
+from functools import wraps
+
+import click
+
+from spherre.indexer.config import DNA_TOKEN, MONGO_URL, SERVER_URL
+from spherre.indexer.service.indexers import run_main_indexer
+
+
+def async_command(f):
+    @wraps(f)
+    def wrapper(*args, **kwargs):
+        return asyncio.run(f(*args, **kwargs))
+
+    return wrapper
+
+
+@click.group()
+def cli():
+    pass
+
+
+@cli.command()
+@click.option("--server-url", default=None, help="Apibara stream url.")
+@click.option("--mongo-url", default=None, help="MongoDB url.")
+@click.option("--restart", is_flag=True, help="Restart indexing from the beginning.")
+@async_command
+async def start(server_url, mongo_url, restart):
+    """Start the indexer."""
+    if server_url is None:
+        server_url = SERVER_URL
+    if mongo_url is None:
+        mongo_url = MONGO_URL
+
+    print("Starting Apibara indexer...")
+    print(f"Server url: {server_url}")
+    await run_main_indexer(
+        restart=restart,
+        server_url=server_url,
+        mongo_url=mongo_url,
+        dna_token=DNA_TOKEN,
+    )

backend/spherre/indexer/service/account_creation.py

@@ -0,0 +1,23 @@
+from typing import Callable, Dict
+
+from spherre.indexer.service.types import (
+    AccountCreationEvent,
+    BaseEventModel,
+    EventEnum,
+)
+
+
+class EventHandlers:
+    @classmethod
+    def handle_account_creation(cls, data: AccountCreationEvent):
+        pass
+
+    @classmethod
+    def handle_token_transafer(cls):
+        pass
+
+
+DATA_HANDLERS: Dict[EventEnum, Callable[[BaseEventModel], None]] = {
+    EventEnum.ACCOUNT_CREATION: EventHandlers.handle_account_creation,
+    EventEnum.TOKEN_TRANSFER: EventHandlers.handle_token_transafer,
+}

backend/spherre/indexer/service/event_handlers.py

@@ -0,0 +1,137 @@
+from apibara.indexer import IndexerRunner, IndexerRunnerConfiguration, Info
+from apibara.indexer.indexer import IndexerConfiguration
+from apibara.protocol.proto.stream_pb2 import Cursor, DataFinality
+from apibara.starknet import EventFilter, Filter, StarkNetIndexer, felt
+from apibara.starknet.cursor import starknet_cursor
+from apibara.starknet.proto.starknet_pb2 import Block
+from loguru import logger
+
+from spherre.indexer.config import NETWORK, SPHERRE_CONTRACT_ADDRESS
+from spherre.indexer.service.event_handlers import DATA_HANDLERS
+from spherre.indexer.service.types import (
+    EVENT_SELECTORS,
+    EventEnum,
+)
+from spherre.indexer.service.utils import DATA_TRANSFORMERS
+
+
+class SpherreMainIndexer(StarkNetIndexer):
+    def start_account_indexer(self, address: str):
+        # create a new filter for the account address
+        account_address = felt.from_hex(address)
+        filter = (
+            EventFilter()
+            .with_from_address(account_address)
+            .with_keys([EVENT_SELECTORS.values()])
+        )
+        # add the filter to the indexer
+        self.update_filter(filter)
+
+    def indexer_id(self) -> str:
+        return "spherre"
+
+    def initial_configuration(self) -> Filter:
+        # Return initial configuration of the indexer.
+        address = felt.from_hex(SPHERRE_CONTRACT_ADDRESS)
+        return IndexerConfiguration(
+            filter=Filter().add_event(
+                EventFilter()
+                .with_from_address(address)
+                .with_keys([EVENT_SELECTORS[EventEnum.ACCOUNT_CREATION]])
+            ),
+            starting_cursor=starknet_cursor(10_000),
+            finality=DataFinality.DATA_STATUS_ACCEPTED,
+        )
+
+    async def handle_data(self, info: Info, data: Block):
+        # Handle one block of data
+        for event_with_tx in data.events:
+            tx_hash = felt.to_hex(event_with_tx.transaction.meta.hash)
+            event = event_with_tx.event
+            event_address = felt.to_hex(event.from_address)
+            logger.info("Transaction Hash:", tx_hash)
+
+            if event_address != SPHERRE_CONTRACT_ADDRESS:
+                event_type = event.keys
+                event_enum = EVENT_SELECTORS.inverse[event_type]
+                transformed_data = DATA_TRANSFORMERS[event_enum](event)
+                if transformed_data:
+                    DATA_HANDLERS[event_enum](transformed_data)
+                    logger.info(
+                        (
+                            f"Event with type '{event_type}' from address"
+                            f"'{event_address}' handled"
+                        )
+                    )
+                else:
+                    logger.error(
+                        (
+                            f"Failed to handle event of type '{event_type}'"
+                            f"from address '{event_address}'"
+                        )
+                    )
+            else:
+                event_type = event.keys
+                if event_type == EVENT_SELECTORS[EventEnum.ACCOUNT_CREATION]:
+                    transformed_data = DATA_TRANSFORMERS[EventEnum.ACCOUNT_CREATION](
+                        event
+                    )
+                    if transformed_data:
+                        DATA_HANDLERS[EventEnum.ACCOUNT_CREATION](transformed_data)
+                        self.start_account_indexer(transformed_data.account_address)
+                        logger.info(
+                            (
+                                "Account created for address "
+                                f"'{transformed_data.account_address}'"
+                            )
+                        )
+                    else:
+                        account_address = felt.to_hex(event.data[0])
+                        logger.error(
+                            (
+                                "Failed to handle account creation event"
+                                f"of address '{account_address}'"
+                            )
+                        )
+
+    async def handle_invalidate(self, _info: Info, _cursor: Cursor):
+        raise ValueError("data must be finalized")
+
+
+def run_in_thread(self):
+    pass
+
+
+async def run_indexer(
+    server_url=None,
+    mongo_url=None,
+    restart=None,
+    dna_token=None,
+    indexer_cls=None,
+    indexer_args=None,
+):
+    runner = IndexerRunner(
+        config=IndexerRunnerConfiguration(
+            stream_url=server_url,
+            storage_url=mongo_url,
+            token=dna_token,
+        ),
+        reset_state=restart,
+    )
+
+    await runner.run(indexer_cls(indexer_args), ctx={"network": NETWORK})
+
+
+async def run_main_indexer(
+    server_url=None, mongo_url=None, restart=None, dna_token=None
+):
+    runner = IndexerRunner(
+        config=IndexerRunnerConfiguration(
+            stream_url=server_url,
+            storage_url=mongo_url,
+            token=dna_token,
+        ),
+        reset_state=restart,
+    )
+
+    await runner.run(SpherreMainIndexer(), ctx={"network": NETWORK})

backend/spherre/indexer/service/indexers.py

@@ -0,0 +1,85 @@
+from enum import Enum
+from typing import Any, List, get_args, get_origin
+
+from apibara.starknet import felt
+from bidict import bidict
+from pydantic import BaseModel, ValidationInfo, field_validator
+
+
+class EventEnum(Enum):
+    ACCOUNT_CREATION = 1
+    TOKEN_TRANSFER = 2
+    TRANSACTION_PROPOSAL = 3
+    TRANSACTION_APPROVAL = 4
+    TRANSACTION_EXECUTION = 5
+
+
+EVENT_SELECTORS = bidict(
+    {
+        EventEnum.ACCOUNT_CREATION: felt.from_hex(
+            "0x347f3a34b919109f055acc8440a003ecda76b4c63c101bbc99b9d00296db557"
+        ),
+        EventEnum.TOKEN_TRANSFER: felt.from_hex(
+            "0x99cd8bde557814842a3121e8ddfd433a539b8c9f14bf31ebf108d12e6196e9"
+        ),
+        EventEnum.TRANSACTION_PROPOSAL: felt.from_hex(""),
+        EventEnum.TRANSACTION_APPROVAL: felt.from_hex(
+            "0x8fac5268bfb16142e4e1fce5f985e337db83f8b307525a0960480cd8473436"
+        ),
+        EventEnum.TRANSACTION_EXECUTION: felt.from_hex(
+            "0x1dcde06aabdbca2f80aa51392b345d7549d7757aa855f7e37f5d335ac8243b1"
+        ),
+    }
+)
+
+
+class BaseEventModel(BaseModel):
+    def __init__(self, *args, **kwargs):
+        if args and not kwargs:
+            # If only positional args provided, map to fields
+            field_names = list(self.model_fields.keys())
+            kwargs = dict(zip(field_names, args))
+        super().__init__(**kwargs)
+
+    @field_validator("*", mode="before")  # Apply to all fields
+    @classmethod
+    def restructure_data_before_parsing(cls, v: Any, info: ValidationInfo):
+        field_info = cls.model_fields[info.field_name]
+        field_type = field_info.annotation
+
+        print(f"\n{info.field_name}:")
+        print(f"  Annotation: {field_type}")
+
+        new_value = v
+
+        if isinstance(field_type, str):
+            new_value = felt.to_hex(v)
+            return new_value
+        elif isinstance(field_type, int):
+            new_value = felt.to_int(v)
+        elif get_origin(field_type):
+            args = get_args(field_type)
+            if isinstance(args, str):
+                new_value = [felt.to_hex(val) for val in v]
+            elif isinstance(args, int):
+                new_value = [felt.to_int(val) for val in v]
+
+        return new_value
+
+
+class AccountCreationEvent(BaseEventModel):
+    account_address: str
+    owner: str
+    name: str
+    description: str
+    members: List[str]
+    threshold: int
+    deployer: int
+    date_deployed: int
+
+
+class TokenTransferEvent(BaseEventModel):
+    account_address: str
+    from_address: str
+    to_address: str
+    amount: int

backend/spherre/indexer/service/types.py

@@ -0,0 +1,58 @@
+from typing import Callable, Dict
+
+from apibara.starknet import felt
+from apibara.starknet.proto.starknet_pb2 import Event
+from loguru import logger
+from pydantic import ValidationError
+
+from spherre.indexer.service.types import (
+    AccountCreationEvent,
+    BaseEventModel,
+    EventEnum,
+)
+
+
+class DataTransformer:
+    @classmethod
+    def transform_account_creation_event(cls, event: Event) -> AccountCreationEvent:
+        account_address = felt.to_hex(event.data[0])
+        owner = felt.to_hex(event.data[1])
+        name = felt.to_hex(event.data[2])
+        description = felt.to_hex(event.data[3])
+        members_array_length = felt.to_int(event.data[4])
+        members = []
+        for i in range(5, members_array_length + 1 + 4):
+            members.append(felt.to_hex(event.data[i]))
+        threshold = felt.to_int(event.data[i + 1])
+        deployer = felt.to_int(event.data[i + 2])
+        date_deployed = felt.to_int(event.data[i + 3])
+        print("Account Deployed")
+        print("Account Address:", account_address)
+        print("Owner:", owner)
+        print("Name:", name)
+        print("Description:", description)
+        print("Members:", members)
+        print("Threshold:", threshold)
+        print("Deployer:", deployer)
+        print("Date Deployed:", date_deployed)
+        try:
+            account_data = AccountCreationEvent(
+                account_address=account_address,
+                owner=owner,
+                name=name,
+                description=description,
+                members=members,
+                threshold=threshold,
+                deployer=deployer,
+                date_deployed=date_deployed,
+            )
+        except ValidationError as err:
+            logger.error("Error parsing account creation event data")
+            logger.error(err)
+            return None
+        return account_data
+
+
+DATA_TRANSFORMERS: Dict[EventEnum, Callable[[Event], BaseEventModel]] = {
+    EventEnum.ACCOUNT_CREATION: DataTransformer.transform_account_creation_event,
+}