Feature/aave_v1

.gitignore

@@ -51,3 +51,5 @@ sync_record
 alembic.ini
 !indexer/modules/custom/hemera_ens/abi/*.json
 !indexer/modules/custom/cyber_id/abi/*.json
+!indexer/modules/custom/aave_v2/abi/*.json
+

cli/stream.py

@@ -132,6 +132,15 @@ def wrapper(*args, **kwargs):
     help="End block",
     envvar="END_BLOCK",
 )
+@click.option(
+    "-nb",
+    "--noncontinuous-blocks",
+    default=None,
+    show_default=True,
+    type=str,
+    help="a list of noncontinuous blocks. eg: 1,11,122",
+    envvar="",
+)
 @click.option(
     "--retry-from-record",
     default=True,
@@ -343,6 +352,7 @@ def stream(
     db_version,
     start_block,
     end_block,
+    noncontinuous_blocks,
     entity_types,
     output_types,
     source_types,
@@ -467,11 +477,21 @@ def stream(
         process_size=process_size,
         process_time_out=process_time_out,
     )
-
-    controller.action(
-        start_block=start_block,
-        end_block=end_block,
-        block_batch_size=block_batch_size,
-        period_seconds=period_seconds,
-        pid_file=pid_file,
-    )
+    if noncontinuous_blocks:
+        block_ids = sorted(set(map(int, noncontinuous_blocks.split(","))))
+        for block_id in block_ids:
+            controller.action(
+                start_block=block_id,
+                end_block=block_id,
+                block_batch_size=block_batch_size,
+                period_seconds=period_seconds,
+                pid_file=pid_file,
+            )
+    else:
+        controller.action(
+            start_block=start_block,
+            end_block=end_block,
+            block_batch_size=block_batch_size,
+            period_seconds=period_seconds,
+            pid_file=pid_file,
+        )

common/utils/abi_code_utils.py

@@ -5,7 +5,10 @@
 Author  : xuzh
 Project : hemera_indexer
 """
+import json
 import logging
+import os
+from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union, cast
 
 import eth_abi
@@ -63,6 +66,9 @@ def get_abi(self) -> ABIEvent:
         """
         return self._event_abi
 
+    def get_name(self) -> str:
+        return self.get_abi().get("name")
+
     def get_signature(self) -> str:
         """
         Returns the signature of the event.
@@ -100,6 +106,77 @@ def decode_log_ignore_indexed(self, log) -> Optional[Dict[str, Any]]:
         return decode_log_ignore_indexed(self._event_abi, log)
 
 
+class AbiReader:
+    """A class to read ABI (Application Binary Interface) files.
+
+    This class handles reading ABI files from a specified directory and provides
+    methods to access event ABIs and entire contract ABIs.
+
+    Attributes:
+        call_path (str): The path from where the class is being called
+        relative_path (str): Relative path to the ABI directory
+        abi_map (Dict[str, list]): Mapping of contract addresses to their ABIs
+    """
+
+    def __init__(self, call_path: str, relative_path: str = "abi") -> None:
+        """Initialize the AbiReader with paths and load ABIs.
+
+        Args:
+            call_path: Path from where the class is being called
+            relative_path: Relative path to the ABI directory
+
+        Raises:
+            FileNotFoundError: If the ABI directory doesn't exist
+        """
+        self.call_path = Path(call_path)
+        self.relative_path = relative_path
+        self.abi_map = self._read_abi()
+
+    def _get_absolute_path(self) -> Path:
+        current_dir = self.call_path.parent
+        return current_dir / self.relative_path
+
+    def _read_abi(self) -> Dict[str, list]:
+        abi_map = {}
+        abs_path = self._get_absolute_path()
+
+        if not abs_path.exists():
+            raise FileNotFoundError(f"ABI directory not found: {abs_path}")
+
+        try:
+            for file_path in abs_path.glob("*.json"):
+                try:
+                    abi_data = json.loads(file_path.read_text())
+                    address = abi_data["address"].lower()
+                    abi = abi_data["abi"]
+                    if not isinstance(abi, list):
+                        continue
+                    abi_map[address] = abi
+                except json.JSONDecodeError:
+                    print(f"Invalid JSON in file: {file_path}")
+                except KeyError as e:
+                    print(f"Missing required field {e} in file: {file_path}")
+
+        except Exception as e:
+            print(f"Error reading ABI files: {e}")
+            raise
+
+        return abi_map
+
+    def get_event_abi(self, address: str, event_name: str) -> Optional[Dict[str, Any]]:
+        address = address.lower()
+        if address not in self.abi_map:
+            print(f"No ABI found for address: {address}")
+            return None
+        return next(
+            (abi for abi in self.abi_map[address] if abi.get("type") == "event" and abi.get("name") == event_name), None
+        )
+
+    def get_entire_abi(self, address: str) -> Optional[list]:
+        address = address.lower()
+        return self.abi_map.get(address)
+
+
 def decode_log_ignore_indexed(
     fn_abi: ABIEvent,
     log,

config/indexer-config.yaml

@@ -38,3 +38,13 @@ eigen_layer_job:
     address: "0x858646372cc42e1a627fce94aa7a7033e7cf075a"
   DELEGATION:
     address: "0x39053d51b77dc0d36036fc1fcc8cb819df8ef37a"
+
+export_aave_v2_job:
+  POOL_CONFIGURE: "0x311bb771e4f8952e6da169b425e7e92d6ac45756"
+  POOL_V2: "0x7d2768de32b0b80b7a3454c06bdac94a69ddc7a9"
+
+export_aave_v1_job:
+  POOL_V1: "0x398ec7346dcd622edc5ae82352f02be94c62d119"
+  POOL_V1_CORE: "0x3dfd23A6c5E8BbcFc9581d2E864a68feb6a076d3"
+  INIT_PROXY: "0x4965f6fa20fe9728decf5165016fc338a5a85abf"
+  POOL: "0x7937d4799803fbbe595ed57278bc4ca21f3bffcb"

enumeration/entity_type.py

@@ -13,6 +13,20 @@
 from indexer.domain.token_transfer import ERC20TokenTransfer, ERC721TokenTransfer, ERC1155TokenTransfer
 from indexer.domain.trace import Trace
 from indexer.domain.transaction import Transaction
+from indexer.modules.custom.aave_v1.domains.aave_v1_domain import (
+    AaveV1AddressCurrentD,
+    AaveV1BorrowD,
+    AaveV1CallRecordsD,
+    AaveV1DepositD,
+    AaveV1FlashLoanD,
+    AaveV1LiquidationAddressCurrentD,
+    AaveV1LiquidationCallD,
+    AaveV1RepayD,
+    AaveV1ReserveD,
+    AaveV1ReserveDataCurrentD,
+    AaveV1ReserveDataD,
+    AaveV1WithdrawD,
+)
 from indexer.modules.custom.address_index.domain import *
 from indexer.modules.custom.address_index.domain.address_contract_operation import AddressContractOperation
 from indexer.modules.custom.address_index.domain.address_internal_transaction import AddressInternalTransaction
@@ -77,6 +91,8 @@ class EntityType(IntFlag):
 
     KARAK = 1 << 11
 
+    AAVEV1 = 1 << 12
+
     EIGEN_LAYER = 1 << 13
 
     UNISWAP_V2 = 1 << 14
@@ -213,3 +229,17 @@ def generate_output_types(entity_types):
         yield UniswapV2SwapEvent
         yield UniswapV2Erc20TotalSupply
         yield UniswapV2Erc20CurrentTotalSupply
+
+    if entity_types & EntityType.AAVEV1:
+        yield AaveV1AddressCurrentD
+        yield AaveV1BorrowD
+        yield AaveV1CallRecordsD
+        yield AaveV1DepositD
+        yield AaveV1FlashLoanD
+        yield AaveV1LiquidationAddressCurrentD
+        yield AaveV1LiquidationCallD
+        yield AaveV1RepayD
+        yield AaveV1ReserveD
+        yield AaveV1WithdrawD
+        yield AaveV1ReserveDataD
+        yield AaveV1ReserveDataCurrentD

indexer/controller/stream_controller.py

@@ -23,7 +23,7 @@ def __init__(
         sync_recorder: BaseRecorder,
         job_scheduler: JobScheduler,
         limit_reader: LimitReader,
-        max_retries=5,
+        max_retries=1,
         retry_from_record=False,
         delay=0,
         process_numbers=1,

indexer/modules/custom/aave_v1/__init__.py

@@ -0,0 +1,2 @@
+"""The Aave protocol v1 was released in January 2020 and eventually superseded by Aave protocol v2 in December on the same year
+"""