feat: event arg filtering (#233)

* fix(Runner): PollingRunner did not pass through the topics

* fix: allow specifying filter params for event logs

* docs(Example): add demo of event log filtering to demo example

* docs: document event arg filtering

* refactor: use utility method from ethpm-types

---------

Co-authored-by: Mike Shultz <shultzm@gmail.com>

Author: fubuloubu

bots/example.py

@@ -6,6 +6,7 @@
 from ape import chain
 from ape.api import BlockAPI
 from ape.types import ContractLog
+from ape.utils import ZERO_ADDRESS
 from ape_tokens import tokens  # type: ignore[import]
 from taskiq import Context, TaskiqDepends, TaskiqState
 
@@ -82,6 +83,11 @@ def exec_event1(log):
     return {"amount": log.amount}
 
 
+@bot.on_(USDC.Transfer, sender=ZERO_ADDRESS)
+async def handle_mints(log):
+    assert log.sender == ZERO_ADDRESS
+
+
 @bot.on_(YFI.Approval)
 # Any handler function can be async too
 async def exec_event2(log: ContractLog):

docs/userguides/development.md

@@ -73,7 +73,7 @@ If you follow these suggestions, your Silverback deployments will be easy to use
 
 Creating a Silverback Bot is easy, to do so initialize the `silverback.SilverbackBot` class:
 
-```py
+```python
 from silverback import SilverbackBot
 
 bot = SilverbackBot()
@@ -91,7 +91,7 @@ This method lets us specify which event will trigger the execution of our handle
 
 To add a block handler, you will do the following:
 
-```py
+```python
 from ape import chain
 
 @bot.on_(chain.blocks)
@@ -107,7 +107,7 @@ Any errors you raise during this function will get captured by the client, and r
 
 Similarly to blocks, you can handle events emitted by a contract by adding an event handler:
 
-```
+```python
 from ape import Contract
 
 TOKEN = Contract(<your token address here>)
@@ -121,13 +121,32 @@ Inside of `handle_token_transfer_events` you can define any logic that you want
 Again, you can return any serializable data structure from this function and that will be stored in the results database as a trackable metric for the execution of this handler.
 Any errors you raise during this function will get captured by the client, and recorded as a failure to handle this `transfer` event log.
 
+### Event Log Filters
+
+You can also filter event logs by event parameters.
+For example, if you want to handle only `Transfer` events that represent a burn (a transfer to the zero address):
+
+```python
+@bot.on_(USDC.Transfer, to="0x0000000000000000000000000000000000000000")
+def handle_burn(log):
+    return {"burned": log.value}
+```
+
+In case an event parameter has a name that is an illegal keyword, we can also support a dictionary syntax:
+
+```python
+@bot.on_(USDC.Transfer, filter_args={"from":"0x0000000000000000000000000000000000000000"})
+def handle_burn(log):
+    return {"burned": log.value}
+```
+
 ## Cron Tasks
 
 You may also want to run some tasks according to a schedule, either for efficiency reasons or just that the task is not related to any chain-driven events.
 You can do that with the `@cron` task decorator.
 
 ```python
-@app.cron("* */1 * * *")
+@bot.cron("* */1 * * *")
 def every_hour():
     ...
 ```
@@ -140,7 +159,7 @@ For more information see [the linux handbook section on the crontab syntax](http
 
 If you have heavier resources you want to load during startup, or want to initialize things like database connections, you can add a worker startup function like so:
 
-```py
+```python
 @bot.on_worker_startup()
 def handle_on_worker_startup(state):
     # Connect to DB, set initial state, etc
@@ -164,7 +183,7 @@ The `state` variable is also useful as this can be made available to each handle
 
 To access the state from a handler, you must annotate `context` as a dependency like so:
 
-```py
+```python
 from typing import Annotated
 from taskiq import Context, TaskiqDepends
 
@@ -178,7 +197,7 @@ def block_handler(block, context: Annotated[Context, TaskiqDepends()]):
 
 You can also add an bot startup and shutdown handler that will be **executed once upon every bot startup**. This may be useful for things like processing historical events since the bot was shutdown or other one-time actions to perform at startup.
 
-```py
+```python
 @bot.on_startup()
 def handle_on_startup(startup_state):
     # Process missed events, etc
@@ -205,7 +224,7 @@ For example, you might want to pre-populate a large dataframe into state on star
 and then use that data to determine a signal under which you want trigger transactions to commit back to the chain.
 Such an bot might look like this:
 
-```py
+```python
 @bot.on_startup()
 def create_table(startup_state):
     df = contract.MyEvent.query(..., start_block=startup_state.last_block_processed)
@@ -279,7 +298,7 @@ $ silverback run my_bot --network :sepolia --account acct-name
 It's important to note that signers are optional, if not configured in the bot then `bot.signer` will be `None`.
 You can use this in your bot to enable a "test execution" mode, something like this:
 
-```py
+```python
 # Compute some metric that might lead to creating a transaction
 if bot.signer:
     # Execute a transaction via `sender=bot.signer`

silverback/main.py

@@ -11,9 +11,11 @@
 from ape.contracts import ContractEvent, ContractEventWrapper, ContractInstance
 from ape.logging import logger
 from ape.managers.chain import BlockContainer
-from ape.types import ContractLog
+from ape.types import AddressType, ContractLog
 from ape.utils import ManagerAccessMixin
+from eth_typing import HexStr
 from eth_utils import keccak, to_hex
+from ethpm_types.abi import encode_topic_value
 from packaging.version import Version
 from pydantic import BaseModel
 from taskiq import AsyncTaskiqDecoratedTask, TaskiqEvents
@@ -348,11 +350,33 @@ async def fork_handler(*args, **kwargs):
 
         return fork_handler
 
+    def _convert_arg_to_hexstr(self, arg_value: Any, arg_type: str) -> HexStr | list[HexStr] | None:
+        python_type: Any
+        if "int" in arg_type:
+            python_type = int
+        elif "bytes" in arg_type:
+            python_type = bytes
+        elif arg_type == "address":
+            python_type = AddressType
+        elif arg_type == "string":
+            python_type = str
+        else:
+            raise ValueError(f"Unable to support ABI Type '{arg_type}'.")
+
+        if isinstance(arg_value, list):
+            arg_value = [self.conversion_manager.convert(v, python_type) for v in arg_value]
+
+        else:
+            arg_value = self.conversion_manager.convert(arg_value, python_type)
+
+        return encode_topic_value(arg_type, arg_value)  # type: ignore[return-value]
+
     def broker_task_decorator(
         self,
         task_type: TaskType,
         container: BlockContainer | ContractEvent | ContractEventWrapper | None = None,
         cron_schedule: str | None = None,
+        filter_args: dict[str, Any] | None = None,
     ) -> Callable[[Callable], AsyncTaskiqDecoratedTask]:
         """
         Dynamically create a new broker task that handles tasks of ``task_type``.
@@ -411,17 +435,36 @@ def add_taskiq_task(
             elif task_type is TaskType.EVENT_LOG:
                 assert container is not None and isinstance(container, ContractEvent)
                 # NOTE: allows broad capture filters (matching multiple addresses)
-                if contract_address := getattr(container.contract, "address", None):
-                    labels["address"] = contract_address
+                if contract := getattr(container, "contract", None):
+                    labels["address"] = contract.address
+
                 labels["event"] = container.abi.signature
-                labels["topics"] = encode_topics_to_string(
-                    [
-                        # Topic 0: event_id
-                        to_hex(keccak(text=container.abi.selector)),
-                        # Topic 1-4: event args ([..., ...] represent OR)
-                        # TODO: Add filter args
-                    ]
-                )
+
+                topics: list[list[HexStr] | HexStr | None] = [
+                    # Topic 0: event_id
+                    to_hex(keccak(text=container.abi.selector))
+                ]
+
+                # Topic 1-3: event args ([..., ...] represent OR)
+                if filter_args:
+                    for arg in container.abi.inputs:
+                        if not arg.indexed:
+                            break  # Inputs should be ordered indexed first
+
+                        if arg_value := filter_args.pop(arg.name, None):
+                            topics.append(self._convert_arg_to_hexstr(arg_value, arg.type))
+
+                        else:
+                            # Skip this indexed argument (`None` is wildcard match)
+                            topics.append(None)
+                            # NOTE: Will clean up extra Nones in `encode_topics_to_string`
+
+                    if unmatched_args := "', '".join(filter_args):
+                        raise InvalidContainerTypeError(
+                            f"Args are not available for filtering: '{unmatched_args}'."
+                        )
+
+                labels["topics"] = encode_topics_to_string(topics)
 
                 handler = self._ensure_log(container, handler)
 
@@ -518,6 +561,8 @@ def on_(
         # TODO: possibly remove these
         new_block_timeout: int | None = None,
         start_block: int | None = None,
+        filter_args: dict[str, Any] | None = None,
+        **filter_kwargs: dict[str, Any],
     ):
         """
         Create task to handle events created by the `container` trigger.
@@ -565,7 +610,14 @@ def on_(
                 else:
                     self.poll_settings[key] = {"start_block": start_block}
 
-            return self.broker_task_decorator(TaskType.EVENT_LOG, container=container)
+            if filter_args:
+                filter_kwargs.update(filter_args)
+
+            return self.broker_task_decorator(
+                TaskType.EVENT_LOG,
+                container=container,
+                filter_args=filter_kwargs,
+            )
 
         # TODO: Support account transaction polling
         # TODO: Support mempool polling?

silverback/runner.py

@@ -444,8 +444,9 @@ async def _block_task(self, task_data: TaskData):
     async def _event_task(self, task_data: TaskData):
         contract_address = task_data.labels.get("address")
         event = EventABI.from_signature(task_data.labels["event"])
+        topics = decode_topics_from_string(task_data.labels.get("topics", "")) or None
         async for log in async_wrap_iter(
             # NOTE: No start block because we should begin polling from head
-            self.provider.poll_logs(address=contract_address, events=[event])
+            self.provider.poll_logs(address=contract_address, events=[event], topics=topics)
         ):
             self._runtime_task_group.create_task(self.run_task(task_data, log))