feat: add owner changed watching for (Async)Arkiv, cleanup event watching

Author: Matthias Zimmermann

README.md

@@ -4,14 +4,6 @@ Arkiv is a permissioned storage system for decentralized apps, supporting flexib
 
 The Arkiv SDK is the official Python library for interacting with Arkiv networks. It offers a type-safe, developer-friendly API for managing entities, querying data, subscribing to events, and offchain verification—ideal for both rapid prototyping and production use.
 
-## TODO AsyncArkiv
-
-- check for duplicate code with Arkiv
-- refactor
-- re-check testing
-- commit
-- move to ArkivModule for async usage
-
 ## Architecture
 
 Principles:

src/arkiv/events.py

@@ -5,14 +5,17 @@
 import logging
 import threading
 import time
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING
 
 from web3._utils.filters import LogFilter
 from web3.contract import Contract
-from web3.types import EventData, LogReceipt
+from web3.types import LogReceipt
+
+from arkiv.utils import get_tx_hash, to_event
 
 from .events_base import EventFilterBase
 from .types import (
+    ChangeOwnerCallback,
     CreateCallback,
     DeleteCallback,
     EventType,
@@ -26,7 +29,13 @@
 logger = logging.getLogger(__name__)
 
 # Union of all sync callback types
-SyncCallback = CreateCallback | UpdateCallback | ExtendCallback | DeleteCallback
+SyncCallback = (
+    CreateCallback
+    | UpdateCallback
+    | ExtendCallback
+    | DeleteCallback
+    | ChangeOwnerCallback
+)
 
 
 class EventFilter(EventFilterBase[SyncCallback]):
@@ -78,36 +87,41 @@ def start(self) -> None:
         logger.info(f"Starting event filter for {self.event_type}")
 
         # Create the Web3 filter using base class helper
-        self._filter = self._create_filter()
+        # Note: base class returns LogFilter | AsyncLogFilter, but for sync we only use LogFilter
+        filter_result = self._create_filter()
+        assert isinstance(filter_result, LogFilter), (
+            "Expected LogFilter for sync client"
+        )
+        self._filter = filter_result
 
         # Start polling thread
         self._running = True
         self._thread = threading.Thread(target=self._poll_loop, daemon=True)
         self._thread.start()
 
-        logger.info(f"Event filter for {self.event_type} started")
+        logger.info(f"Event filter for '{self.event_type}' started")
 
     def stop(self) -> None:
         """
         Stop polling for events.
         """
         if not self._running:
-            logger.warning(f"Filter for {self.event_type} is not running")
+            logger.warning(f"Filter for '{self.event_type}' is not running")
             return
 
-        logger.info(f"Stopping event filter for {self.event_type}")
+        logger.info(f"Stopping event filter for '{self.event_type}'")
         self._running = False
 
         # Wait for thread to finish
         if self._thread:
             self._thread.join(timeout=5.0)
             self._thread = None
 
-        logger.info(f"Event filter for {self.event_type} stopped")
+        logger.info(f"Event filter for '{self.event_type}' stopped")
 
     def uninstall(self) -> None:
         """Uninstall the filter and cleanup resources."""
-        logger.info(f"Uninstalling event filter for {self.event_type}")
+        logger.info(f"Uninstalling event filter for '{self.event_type}'")
 
         # Stop polling if running
         if self._running:
@@ -126,13 +140,10 @@ def _poll_loop(self) -> None:
             try:
                 # Get new entries from filter
                 if self._filter:
-                    new_entries: list[LogReceipt] = self._filter.get_new_entries()
-
-                    for entry in new_entries:
+                    logs: list[LogReceipt] = self._filter.get_new_entries()
+                    for log in logs:
                         try:
-                            # LogFilter from contract event has log_entry_formatter that
-                            # converts LogReceipt to EventData, but type system shows LogReceipt
-                            self._process_event(cast(EventData, entry))
+                            self._process_log(log)
                         except Exception as e:
                             logger.error(f"Error processing event: {e}", exc_info=True)
 
@@ -145,18 +156,20 @@ def _poll_loop(self) -> None:
 
         logger.debug(f"Poll loop ended for {self.event_type}")
 
-    def _process_event(self, event_data: EventData) -> None:
+    def _process_log(self, log: LogReceipt) -> None:
         """
-        Process a single event and trigger sync callback.
+        Process a single log receipt and trigger sync callback.
 
-        Args:
-            event_data: Event data from Web3 filter
+        Only processes logs from the contract address we're monitoring.
+        Logs from other contracts are silently skipped.
         """
-        # Use base class to parse event data
-        event, tx_hash = self._parse_event_data(event_data)
-
-        # Trigger sync callback with error handling
         try:
+            # to_event handles both raw logs and already-processed EventData
+            event = to_event(self.contract, log)
+            tx_hash = get_tx_hash(log)
+
+            logger.info(f"Starting callback for hash: {tx_hash} and event: {event}")
             self.callback(event, tx_hash)  # type: ignore[arg-type]
+
         except Exception as e:
             logger.error(f"Error in callback: {e}", exc_info=True)

src/arkiv/events_async.py

@@ -4,14 +4,17 @@
 
 import asyncio
 import logging
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any
 
 from web3._utils.filters import LogFilter
 from web3.contract import Contract
-from web3.types import EventData, LogReceipt
+from web3.types import LogReceipt
+
+from arkiv.utils import get_tx_hash, to_event
 
 from .events_base import EventFilterBase
 from .types import (
+    AsyncChangeOwnerCallback,
     AsyncCreateCallback,
     AsyncDeleteCallback,
     AsyncExtendCallback,
@@ -30,6 +33,7 @@
     | AsyncUpdateCallback
     | AsyncExtendCallback
     | AsyncDeleteCallback
+    | AsyncChangeOwnerCallback
 )
 
 
@@ -70,7 +74,7 @@ def __init__(
         self._task: asyncio.Task[None] | None = None
         self._filter: LogFilter | None = None
 
-    async def _create_filter(self) -> Any:
+    async def _create_filter(self) -> Any:  # type: ignore[override]
         """
         Create a Web3 contract event filter for async HTTP polling.
 
@@ -83,7 +87,10 @@ async def _create_filter(self) -> Any:
         """
         event_name = self._get_contract_event_name()
         contract_event = self.contract.events[event_name]
-        return await contract_event.create_filter(from_block=self.from_block)
+        return await contract_event.create_filter(
+            from_block=self.from_block,
+            address=self.contract.address,
+        )
 
     async def start(self) -> None:
         """
@@ -147,15 +154,10 @@ async def _poll_loop(self) -> None:
             try:
                 # Get new entries from filter
                 if self._filter:
-                    # For async providers, get_new_entries() returns a coroutine
-                    # Type system doesn't reflect this, so we need to ignore the type error
-                    new_entries: list[LogReceipt] = await self._filter.get_new_entries()  # type: ignore[misc]
-
-                    for entry in new_entries:
+                    logs: list[LogReceipt] = await self._filter.get_new_entries()  # type: ignore[misc]
+                    for log in logs:
                         try:
-                            # LogFilter from contract event has log_entry_formatter that
-                            # converts LogReceipt to EventData, but type system shows LogReceipt
-                            await self._process_event(cast(EventData, entry))
+                            await self._process_log(log)
                         except Exception as e:
                             logger.error(f"Error processing event: {e}", exc_info=True)
 
@@ -171,18 +173,30 @@ async def _poll_loop(self) -> None:
 
         logger.debug(f"Async poll loop ended for {self.event_type}")
 
-    async def _process_event(self, event_data: EventData) -> None:
+    async def _process_log(self, log: LogReceipt) -> None:
         """
         Process a single event and trigger async callback.
 
-        Args:
-            event_data: Event data from Web3 filter
+        Only processes logs from the contract address we're monitoring.
+        Logs from other contracts are silently skipped.
         """
-        # Use base class to parse event data
-        event, tx_hash = self._parse_event_data(event_data)
-
-        # Trigger async callback with error handling
         try:
+            # Defensive check: Only process logs from our contract
+            log_address = log.get("address")
+            if log_address and log_address.lower() != self.contract.address.lower():
+                logger.debug(
+                    f"Skipping log from different contract: {log_address} "
+                    f"(expected {self.contract.address})"
+                )
+                return
+
+            # to_event handles both raw logs and already-processed EventData
+            event = to_event(self.contract, log)
+            tx_hash = get_tx_hash(log)
+
+            logger.info(f"Starting callback for hash: {tx_hash} and event: {event}")
+
             await self.callback(event, tx_hash)  # type: ignore[arg-type]
+
         except Exception as e:
             logger.error(f"Error in async callback: {e}", exc_info=True)

src/arkiv/events_base.py

@@ -6,22 +6,19 @@
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, Any, Generic, TypeVar
 
-from eth_typing import HexStr
-from web3.types import EventData
-
 from .contract import EVENTS
 from .types import (
     CreateEvent,
     DeleteEvent,
     EventType,
     ExtendEvent,
-    TxHash,
     UpdateEvent,
 )
-from .utils import to_entity_key
 
 if TYPE_CHECKING:
+    from web3._utils.filters import AsyncLogFilter, LogFilter
     from web3.contract import Contract
+    from web3.contract.contract import ContractEvent
 
 logger = logging.getLogger(__name__)
 
@@ -85,7 +82,7 @@ def _get_contract_event_name(self) -> str:
         Get the Web3 contract event name for this event type.
 
         Returns:
-            Contract event name (e.g., "GolemBaseStorageEntityCreated")
+            Contract event name (e.g., "ArkivEntityCreated")
 
         Raises:
             NotImplementedError: If event type is not supported
@@ -96,11 +93,15 @@ def _get_contract_event_name(self) -> str:
             )
         return EVENTS[self.event_type]
 
-    def _create_filter(self) -> Any:
+    def _create_filter(self) -> LogFilter | AsyncLogFilter:
         """
         Create a Web3 contract event filter for HTTP polling.
 
         This method creates a LogFilter using the contract's create_filter method.
+        The filter is automatically configured to:
+        - Only receive events matching the specific event signature (topic[0])
+        - Only receive events from this contract's address
+
         Subclasses that use different subscription mechanisms (e.g., WebSocket)
         should override this method to return an appropriate subscription handle.
 
@@ -112,83 +113,21 @@ def _create_filter(self) -> Any:
             override to create subscription handles via provider-specific APIs.
         """
         event_name = self._get_contract_event_name()
-        contract_event = self.contract.events[event_name]
-        return contract_event.create_filter(from_block=self.from_block)
-
-    def _extract_tx_hash(self, event_data: EventData) -> TxHash:
-        """
-        Extract and normalize transaction hash from event data.
-
-        Args:
-            event_data: Event data from Web3 filter
-
-        Returns:
-            Transaction hash with 0x prefix
-        """
-        tx_hash_hex = event_data["transactionHash"].hex()
-        if not tx_hash_hex.startswith("0x"):
-            tx_hash_hex = f"0x{tx_hash_hex}"
-        return TxHash(HexStr(tx_hash_hex))
-
-    # TODO (1) check/match against utils.py::to_receipt
-    # TODO (2) update to new log/event definition in contract.py
-    def _parse_event_data(self, event_data: EventData) -> tuple[EventObject, TxHash]:
-        """
-        Parse event data and create appropriate event object.
-
-        This method contains the shared logic for parsing events from Web3.
-        It does NOT trigger the callback - that's done by subclasses to allow
-        for sync vs async callback invocation.
-
-        Args:
-            event_data: Event data from Web3 filter
-
-        Returns:
-            Tuple of (event_object, tx_hash)
-
-        Raises:
-            ValueError: If event_type is unknown
-        """
-        logger.info(f"Parsing event: {event_data}")
-
-        # Extract common data
-        entity_key = to_entity_key(event_data["args"]["entityKey"])
-        tx_hash = self._extract_tx_hash(event_data)
-
-        # Create event object based on type
-        event: EventObject
-        if self.event_type == "created":
-            event = CreateEvent(
-                entity_key=entity_key,
-                owner_address=event_data["args"]["ownerAddress"],
-                expiration_block=event_data["args"]["expirationBlock"],
-                cost=event_data["args"]["cost"],
-            )
-        elif self.event_type == "updated":
-            event = UpdateEvent(
-                entity_key=entity_key,
-                owner_address=event_data["args"]["ownerAddress"],
-                old_expiration_block=event_data["args"]["oldExpirationBlock"],
-                new_expiration_block=event_data["args"]["newExpirationBlock"],
-                cost=event_data["args"]["cost"],
-            )
-        elif self.event_type == "extended":
-            event = ExtendEvent(
-                entity_key=entity_key,
-                owner_address=event_data["args"]["ownerAddress"],
-                old_expiration_block=event_data["args"]["oldExpirationBlock"],
-                new_expiration_block=event_data["args"]["newExpirationBlock"],
-                cost=event_data["args"]["cost"],
-            )
-        elif self.event_type == "deleted":
-            event = DeleteEvent(
-                entity_key=entity_key,
-                owner_address=event_data["args"]["ownerAddress"],
-            )
-        else:
-            raise ValueError(f"Unknown event type: {self.event_type}")
-
-        return event, tx_hash
+        contract_event: ContractEvent = self.contract.events[event_name]
+
+        # Create filter with explicit address filtering
+        # The ContractEvent.create_filter automatically sets the event signature
+        # as topic[0], and we explicitly filter by contract address
+        filter: LogFilter | AsyncLogFilter = contract_event.create_filter(
+            from_block=self.from_block,
+            address=self.contract.address,
+        )
+
+        logger.info(
+            f"Created filter for event {event_name} from block {self.from_block} "
+            f"at address {self.contract.address}: {filter}"
+        )
+        return filter
 
     # Abstract methods that subclasses must implement
     @abstractmethod