1"""Arkiv Python SDK."""
  2
  3import asyncio
  4import base64
  5import logging
  6import logging.config
  7import typing
  8from collections.abc import (
  9    AsyncGenerator,
 10    Callable,
 11    Coroutine,
 12    Sequence,
 13)
 14from typing import (
 15    TYPE_CHECKING,
 16    Any,
 17    TypeAlias,
 18    cast,
 19)
 20
 21from eth_typing import ChecksumAddress, HexStr
 22from web3 import AsyncHTTPProvider, AsyncWeb3, WebSocketProvider
 23from web3.contract import AsyncContract
 24from web3.exceptions import ProviderConnectionError, Web3RPCError, Web3ValueError
 25from web3.method import Method, default_root_munger
 26from web3.middleware import SignAndSendRawMiddlewareBuilder
 27from web3.types import LogReceipt, RPCEndpoint, TxParams, TxReceipt, Wei
 28from web3.utils.subscriptions import (
 29    LogsSubscription,
 30    LogsSubscriptionContext,
 31)
 32
 33from .constants import (
 34    ARKIV_ABI,
 35    STORAGE_ADDRESS,
 36)
 37from .types import (
 38    Address,
 39    Annotation,
 40    ArkivCreate,
 41    ArkivDelete,
 42    ArkivExtend,
 43    ArkivTransaction,
 44    ArkivTransactionReceipt,
 45    ArkivUpdate,
 46    CreateEntityReturnType,
 47    EntityKey,
 48    EntityMetadata,
 49    ExtendEntityReturnType,
 50    GenericBytes,
 51    QueryEntitiesResult,
 52    UpdateEntityReturnType,
 53    WatchLogsHandle,
 54)
 55from .utils import rlp_encode_transaction
 56from .wallet import (
 57    WalletError,
 58    decrypt_wallet,
 59)
 60
 61__all__: Sequence[str] = [
 62    # Exports from .types
 63    "Address",
 64    "Annotation",
 65    "CreateEntityReturnType",
 66    "EntityKey",
 67    "EntityMetadata",
 68    "ExtendEntityReturnType",
 69    "GenericBytes",
 70    "ArkivCreate",
 71    "ArkivDelete",
 72    "ArkivExtend",
 73    "ArkivTransaction",
 74    "ArkivTransactionReceipt",
 75    "ArkivUpdate",
 76    "QueryEntitiesResult",
 77    "UpdateEntityReturnType",
 78    "WatchLogsHandle",
 79    # Exports from .constants
 80    "ARKIV_ABI",
 81    "STORAGE_ADDRESS",
 82    # Exports from .wallet
 83    "decrypt_wallet",
 84    "WalletError",
 85    # Exports from this file
 86    "ArkivClient",
 87    # Re-exports
 88    "Wei",
 89]
 90
 91if TYPE_CHECKING:
 92    HTTPClient: TypeAlias = AsyncWeb3[AsyncHTTPProvider]
 93    WSClient: TypeAlias = AsyncWeb3[WebSocketProvider]
 94else:
 95    HTTPClient: TypeAlias = AsyncWeb3
 96    WSClient: TypeAlias = AsyncWeb3
 97
 98
 99logger = logging.getLogger(__name__)
100"""@private"""
101
102
103class ArkivHttpClient(HTTPClient):
104    """Subclass of AsyncWeb3 with added Arkiv methods."""
105
106    def __init__(self, rpc_url: str):
107        super().__init__(AsyncHTTPProvider(rpc_url, request_kwargs={"timeout": 60}))
108
109        self.eth.attach_methods(
110            {
111                "get_storage_value": Method(
112                    json_rpc_method=RPCEndpoint("golembase_getStorageValue"),
113                    mungers=[default_root_munger],
114                ),
115                "get_entity_metadata": Method(
116                    json_rpc_method=RPCEndpoint("golembase_getEntityMetaData"),
117                    mungers=[default_root_munger],
118                ),
119                "get_entities_to_expire_at_block": Method(
120                    json_rpc_method=RPCEndpoint("golembase_getEntitiesToExpireAtBlock"),
121                    mungers=[default_root_munger],
122                ),
123                "get_entity_count": Method(
124                    json_rpc_method=RPCEndpoint("golembase_getEntityCount"),
125                    mungers=[default_root_munger],
126                ),
127                "get_all_entity_keys": Method(
128                    json_rpc_method=RPCEndpoint("golembase_getAllEntityKeys"),
129                    mungers=[default_root_munger],
130                ),
131                "get_entities_of_owner": Method(
132                    json_rpc_method=RPCEndpoint("golembase_getEntitiesOfOwner"),
133                    mungers=[default_root_munger],
134                ),
135                "query_entities": Method(
136                    json_rpc_method=RPCEndpoint("golembase_queryEntities"),
137                    mungers=[default_root_munger],
138                ),
139            }
140        )
141
142    async def get_storage_value(self, entity_key: EntityKey) -> bytes:
143        """Get the storage value stored in the given entity."""
144        return base64.b64decode(
145            await self.eth.get_storage_value(  # type: ignore[attr-defined]
146                entity_key.as_hex_string()
147            )
148        )
149
150    async def get_entity_metadata(self, entity_key: EntityKey) -> EntityMetadata:
151        """Get the metadata of the given entity."""
152        metadata = await self.eth.get_entity_metadata(  # type: ignore[attr-defined]
153            entity_key.as_hex_string()
154        )
155
156        return EntityMetadata(
157            entity_key=entity_key,
158            owner=Address(GenericBytes.from_hex_string(metadata.owner)),
159            expires_at_block=metadata.expiresAtBlock,
160            string_annotations=list(
161                map(
162                    lambda ann: Annotation(key=ann["key"], value=ann["value"]),
163                    metadata.stringAnnotations or list(),
164                )
165            ),
166            numeric_annotations=list(
167                map(
168                    lambda ann: Annotation(key=ann["key"], value=ann["value"]),
169                    metadata.numericAnnotations or list(),
170                )
171            ),
172        )
173
174    async def get_entities_to_expire_at_block(
175        self, block_number: int
176    ) -> Sequence[EntityKey]:
177        """Get all entities that will expire at the given block."""
178        return list(
179            map(
180                lambda e: EntityKey(GenericBytes.from_hex_string(e)),
181                await self.eth.get_entities_to_expire_at_block(  # type: ignore[attr-defined]
182                    block_number
183                ),
184            )
185        )
186
187    async def get_entity_count(self) -> int:
188        """Get the total entity count in Arkiv."""
189        return cast(int, await self.eth.get_entity_count())  # type: ignore[attr-defined]
190
191    async def get_all_entity_keys(self) -> Sequence[EntityKey]:
192        """Get all entity keys in Arkiv."""
193        return list(
194            map(
195                lambda e: EntityKey(GenericBytes.from_hex_string(e)),
196                await self.eth.get_all_entity_keys(),  # type: ignore[attr-defined]
197            )
198        )
199
200    async def get_entities_of_owner(
201        self, owner: ChecksumAddress
202    ) -> Sequence[EntityKey]:
203        """Get all the entities owned by the given address."""
204        return list(
205            map(
206                lambda e: EntityKey(GenericBytes.from_hex_string(e)),
207                await self.eth.get_entities_of_owner(owner),  # type: ignore[attr-defined]
208            )
209        )
210
211    async def query_entities(self, query: str) -> Sequence[QueryEntitiesResult]:
212        """Get all entities that satisfy the given Arkiv query."""
213        return list(
214            map(
215                lambda result: QueryEntitiesResult(
216                    entity_key=result.key, storage_value=base64.b64decode(result.value)
217                ),
218                await self.eth.query_entities(query),  # type: ignore[attr-defined]
219            )
220        )
221
222
223class ArkivROClient:
224    _http_client: ArkivHttpClient
225    _ws_client: WSClient
226    _arkiv_contract: AsyncContract
227    _background_tasks: set[asyncio.Task[None]]
228
229    @staticmethod
230    async def create_ro_client(rpc_url: str, ws_url: str) -> "ArkivROClient":
231        """
232        Create an `ArkivClient` instance.
233
234        This is the preferred method to create an instance.
235        """
236        return ArkivROClient(rpc_url, await ArkivROClient._create_ws_client(ws_url))
237
238    @staticmethod
239    async def _create_ws_client(ws_url: str) -> "AsyncWeb3[WebSocketProvider]":
240        ws_client: WSClient = await AsyncWeb3(WebSocketProvider(ws_url))
241        return ws_client
242
243    def __init__(self, rpc_url: str, ws_client: WSClient) -> None:
244        """Initialise the ArkivClient instance."""
245        self._http_client = ArkivHttpClient(rpc_url)
246        self._ws_client = ws_client
247
248        # Keep references to async tasks we created
249        self._background_tasks = set()
250
251        def is_connected(
252            client: HTTPClient,
253        ) -> Callable[[bool], Coroutine[Any, Any, bool]]:
254            async def inner(show_traceback: bool) -> bool:
255                try:
256                    logger.debug("Calling eth_blockNumber to test connectivity...")
257                    await client.eth.get_block_number()
258                    return True
259                except (OSError, ProviderConnectionError) as e:
260                    logger.debug(
261                        "Problem connecting to provider", exc_info=show_traceback
262                    )
263                    if show_traceback:
264                        raise ProviderConnectionError(
265                            "Problem connecting to provider"
266                        ) from e
267                    return False
268
269            return inner
270
271        # The default is_connected method calls web3_clientVersion, but the web3
272        # API is not enabled on all our nodes, so let's monkey patch this to call
273        # eth_getBlockNumber instead.
274        # The method on the provider is usually not called directly, instead you
275        # can call the eponymous method on the client, which will delegate to the
276        # provider.
277        object.__setattr__(
278            self.http_client().provider,
279            "is_connected",
280            is_connected(self.http_client()),
281        )
282
283        # Allow caching of certain methods to improve performance
284        self.http_client().provider.cache_allowed_requests = True
285
286        # https://github.com/pylint-dev/pylint/issues/3162
287        # pylint: disable=no-member
288        self.arkiv_contract = self.http_client().eth.contract(
289            address=STORAGE_ADDRESS.as_address(),
290            abi=ARKIV_ABI,
291        )
292        for event in self.arkiv_contract.all_events():
293            logger.debug(
294                "Registered event %s with hash %s", event.signature, event.topic
295            )
296
297    def http_client(self) -> ArkivHttpClient:
298        """Get the underlying web3 http client."""
299        return self._http_client
300
301    def ws_client(self) -> WSClient:
302        """Get the underlying web3 websocket client."""
303        return self._ws_client
304
305    async def is_connected(self) -> bool:
306        """Check whether the client's underlying http client is connected."""
307        return cast(bool, await self.http_client().is_connected())  # type: ignore[redundant-cast]
308
309    async def disconnect(self) -> None:
310        """
311        Disconnect this client.
312
313        this method disconnects both the underlying http and ws clients and
314        unsubscribes from all subscriptions.
315        """
316        await self.http_client().provider.disconnect()
317        await self.ws_client().subscription_manager.unsubscribe_all()
318        await self.ws_client().provider.disconnect()
319
320    async def get_storage_value(self, entity_key: EntityKey) -> bytes:
321        """Get the storage value stored in the given entity."""
322        return await self.http_client().get_storage_value(entity_key)
323
324    async def get_entity_metadata(self, entity_key: EntityKey) -> EntityMetadata:
325        """Get the metadata of the given entity."""
326        return await self.http_client().get_entity_metadata(entity_key)
327
328    async def get_entities_to_expire_at_block(
329        self, block_number: int
330    ) -> Sequence[EntityKey]:
331        """Get all entities that will expire at the given block."""
332        return await self.http_client().get_entities_to_expire_at_block(block_number)
333
334    async def get_entity_count(self) -> int:
335        """Get the total entity count in Arkiv."""
336        return await self.http_client().get_entity_count()
337
338    async def get_all_entity_keys(self) -> Sequence[EntityKey]:
339        """Get all entity keys in Arkiv."""
340        return await self.http_client().get_all_entity_keys()
341
342    async def get_entities_of_owner(
343        self, owner: ChecksumAddress
344    ) -> Sequence[EntityKey]:
345        """Get all the entities owned by the given address."""
346        return await self.http_client().get_entities_of_owner(owner)
347
348    async def query_entities(self, query: str) -> Sequence[QueryEntitiesResult]:
349        """Get all entities that satisfy the given Arkiv query."""
350        return await self.http_client().query_entities(query)
351
352    async def watch_logs(
353        self,
354        *,
355        label: str,
356        create_callback: Callable[[CreateEntityReturnType], None] | None = None,
357        update_callback: Callable[[UpdateEntityReturnType], None] | None = None,
358        delete_callback: Callable[[EntityKey], None] | None = None,
359        extend_callback: Callable[[ExtendEntityReturnType], None] | None = None,
360    ) -> WatchLogsHandle:
361        """
362        Subscribe to events on Arkiv.
363
364        You can pass in four different callbacks, and the right one will
365        be invoked for every create, update, delete, and extend operation.
366        """
367
368        async def log_handler(
369            handler_context: LogsSubscriptionContext,
370        ) -> None:
371            # We only use this handler for log receipts
372            # TypeDicts cannot be checked at runtime
373            log_receipt = typing.cast(LogReceipt, handler_context.result)
374            logger.debug("New log: %s", log_receipt)
375            res = await self._process_arkiv_log_receipt(log_receipt)
376
377            if res:
378                if create_callback:
379                    for create in res.creates:
380                        create_callback(create)
381                if update_callback:
382                    for update in res.updates:
383                        update_callback(update)
384                if delete_callback:
385                    for key in res.deletes:
386                        delete_callback(key)
387                if extend_callback:
388                    for extension in res.extensions:
389                        extend_callback(extension)
390
391        def create_subscription(topic: HexStr) -> LogsSubscription:
392            return LogsSubscription(
393                label=f"Arkiv subscription to topic {topic} with label {label}",
394                address=self.arkiv_contract.address,
395                topics=[topic],
396                handler=log_handler,
397                # optional `handler_context` args to help parse a response
398                handler_context={},
399            )
400
401        event_names = []
402        if create_callback:
403            event_names.append("GolemBaseStorageEntityCreated")
404        if update_callback:
405            event_names.append("GolemBaseStorageEntityUpdated")
406        if delete_callback:
407            event_names.append("GolemBaseStorageEntityDeleted")
408        if extend_callback:
409            event_names.append("GolemBaseStorageEntityBTLExtended")
410
411        events = list(
412            map(
413                lambda event_name: create_subscription(
414                    self.arkiv_contract.get_event_by_name(event_name).topic
415                ),
416                event_names,
417            )
418        )
419        subscription_ids = await self._ws_client.subscription_manager.subscribe(
420            events,
421        )
422        logger.info("Sub ID: %s", subscription_ids)
423
424        # Start a subscription loop in case there is none running
425        await self._start_subscription_loop()
426
427        async def unsubscribe() -> None:
428            await self._ws_client.subscription_manager.unsubscribe(subscription_ids)
429
430        return WatchLogsHandle(_unsubscribe=unsubscribe)
431
432    async def _start_subscription_loop(self) -> None:
433        """Create a long running task to handle subscriptions."""
434        # The loop will finish when there are no subscriptions left, so this method
435        # gets called every time a subscription is created, and we'll check
436        # whether we need to make a new task or whether one is already running.
437        if not self._background_tasks:
438            # Start the asyncio event loop
439            task = asyncio.create_task(
440                self.ws_client().subscription_manager.handle_subscriptions()
441            )
442            self._background_tasks.add(task)
443
444            def task_done(task: asyncio.Task[None]) -> None:
445                logger.info("Subscription background task done, removing...")
446                self._background_tasks.discard(task)
447
448            task.add_done_callback(task_done)
449
450    async def _process_arkiv_log_receipt(
451        self,
452        log_receipt: LogReceipt,
453    ) -> ArkivTransactionReceipt | None:
454        # Read the first entry of the topics array,
455        # which is the hash of the event signature, identifying the event
456        topic = AsyncWeb3.to_hex(log_receipt["topics"][0])
457        try:
458            # Look up the corresponding event
459            # If there is no such event in the ABI, it probably needs to be added
460            event = self.arkiv_contract.get_event_by_topic(topic)
461
462            # Use the event to process the whole log
463            event_data = event.process_log(log_receipt)
464        except Web3ValueError:
465            return None
466
467        creates: list[CreateEntityReturnType] = []
468        updates: list[UpdateEntityReturnType] = []
469        deletes: list[EntityKey] = []
470        extensions: list[ExtendEntityReturnType] = []
471
472        match event_data["event"]:
473            case "GolemBaseStorageEntityCreated":
474                creates.append(
475                    CreateEntityReturnType(
476                        expiration_block=event_data["args"]["expirationBlock"],
477                        entity_key=EntityKey(
478                            GenericBytes(
479                                event_data["args"]["entityKey"].to_bytes(32, "big")
480                            )
481                        ),
482                    )
483                )
484            case "GolemBaseStorageEntityUpdated":
485                updates.append(
486                    UpdateEntityReturnType(
487                        expiration_block=event_data["args"]["expirationBlock"],
488                        entity_key=EntityKey(
489                            GenericBytes(
490                                event_data["args"]["entityKey"].to_bytes(32, "big")
491                            )
492                        ),
493                    )
494                )
495            case "GolemBaseStorageEntityDeleted":
496                deletes.append(
497                    EntityKey(
498                        GenericBytes(
499                            event_data["args"]["entityKey"].to_bytes(32, "big")
500                        ),
501                    )
502                )
503            case "GolemBaseStorageEntityBTLExtended":
504                extensions.append(
505                    ExtendEntityReturnType(
506                        old_expiration_block=event_data["args"]["oldExpirationBlock"],
507                        new_expiration_block=event_data["args"]["newExpirationBlock"],
508                        entity_key=EntityKey(
509                            GenericBytes(
510                                event_data["args"]["entityKey"].to_bytes(32, "big")
511                            )
512                        ),
513                    )
514                )
515
516        return ArkivTransactionReceipt(
517            creates=creates,
518            updates=updates,
519            deletes=deletes,
520            extensions=extensions,
521        )
522
523    async def _process_arkiv_receipt(
524        self, receipt: TxReceipt
525    ) -> ArkivTransactionReceipt:
526        # There doesn't seem to be a method for this in the web3 lib.
527        # The only option in the lib is to iterate over the events in the ABI
528        # and call process_receipt on each of them to try and decode the logs.
529        # This is inefficient though compared to reading the actual topic signature
530        # and immediately selecting the right event from the ABI, which is what
531        # we do here.
532        async def process_receipt(
533            receipt: TxReceipt,
534        ) -> AsyncGenerator[ArkivTransactionReceipt, None]:
535            for log in receipt["logs"]:
536                processed = await self._process_arkiv_log_receipt(log)
537                if processed:
538                    yield processed
539
540        creates: list[CreateEntityReturnType] = []
541        updates: list[UpdateEntityReturnType] = []
542        deletes: list[EntityKey] = []
543        extensions: list[ExtendEntityReturnType] = []
544
545        async for res in process_receipt(receipt):
546            creates.extend(res.creates)
547            updates.extend(res.updates)
548            deletes.extend(res.deletes)
549            extensions.extend(res.extensions)
550
551        return ArkivTransactionReceipt(
552            creates=creates,
553            updates=updates,
554            deletes=deletes,
555            extensions=extensions,
556        )
557
558
559class ArkivClient(ArkivROClient):
560    """
561    The Arkiv client used to interact with Arkiv.
562
563    Many useful methods are implemented directly on this type, while more
564    generic ethereum methods can be accessed through the underlying
565    web3 client that you can access with the
566    `ArkivClient.http_client()`
567    method.
568    """
569
570    @staticmethod
571    async def create_rw_client(
572        rpc_url: str, ws_url: str, private_key: bytes
573    ) -> "ArkivClient":
574        """
575        Create a read-write Arkiv client.
576
577        This is the preferred method to create an instance.
578        """
579        return ArkivClient(
580            rpc_url, await ArkivROClient._create_ws_client(ws_url), private_key
581        )
582
583    @staticmethod
584    async def create(rpc_url: str, ws_url: str, private_key: bytes) -> "ArkivClient":
585        """
586        Create a read-write Arkiv client.
587
588        This method is deprecated in favour of `ArkivClient.create_rw_client()`.
589        """
590        return await ArkivClient.create_rw_client(rpc_url, ws_url, private_key)
591
592    def __init__(self, rpc_url: str, ws_client: WSClient, private_key: bytes) -> None:
593        """Initialise the ArkivClient instance."""
594        super().__init__(rpc_url, ws_client)
595
596        # Set up the ethereum account
597        self.account = self.http_client().eth.account.from_key(private_key)
598        # Inject a middleware that will sign transactions with the account that
599        # we created
600        self.http_client().middleware_onion.inject(
601            # pylint doesn't detect nested @curry annotations properly...
602            # pylint: disable=no-value-for-parameter
603            SignAndSendRawMiddlewareBuilder.build(self.account),
604            layer=0,
605        )
606        # Set the account as the default, so we don't need to specify the from field
607        # every time
608        self.http_client().eth.default_account = self.account.address
609        logger.debug("Using account: %s", self.account.address)
610
611    def get_account_address(self) -> ChecksumAddress:
612        """Get the address associated with the private key of this client."""
613        return cast(ChecksumAddress, self.account.address)
614
615    async def create_entities(
616        self,
617        creates: Sequence[ArkivCreate],
618        *,
619        gas: int | None = None,
620        maxFeePerGas: Wei | None = None,
621        maxPriorityFeePerGas: Wei | None = None,
622    ) -> Sequence[CreateEntityReturnType]:
623        """Create entities in Arkiv."""
624        return (
625            await self.send_transaction(
626                creates=creates,
627                gas=gas,
628                maxFeePerGas=maxFeePerGas,
629                maxPriorityFeePerGas=maxPriorityFeePerGas,
630            )
631        ).creates
632
633    async def update_entities(
634        self,
635        updates: Sequence[ArkivUpdate],
636        *,
637        gas: int | None = None,
638        maxFeePerGas: Wei | None = None,
639        maxPriorityFeePerGas: Wei | None = None,
640    ) -> Sequence[UpdateEntityReturnType]:
641        """Update entities in Arkiv."""
642        return (
643            await self.send_transaction(
644                updates=updates,
645                gas=gas,
646                maxFeePerGas=maxFeePerGas,
647                maxPriorityFeePerGas=maxPriorityFeePerGas,
648            )
649        ).updates
650
651    async def delete_entities(
652        self,
653        deletes: Sequence[ArkivDelete],
654        *,
655        gas: int | None = None,
656        maxFeePerGas: Wei | None = None,
657        maxPriorityFeePerGas: Wei | None = None,
658    ) -> Sequence[EntityKey]:
659        """Delete entities from Arkiv."""
660        return (
661            await self.send_transaction(
662                deletes=deletes,
663                gas=gas,
664                maxFeePerGas=maxFeePerGas,
665                maxPriorityFeePerGas=maxPriorityFeePerGas,
666            )
667        ).deletes
668
669    async def extend_entities(
670        self,
671        extensions: Sequence[ArkivExtend],
672        *,
673        gas: int | None = None,
674        maxFeePerGas: Wei | None = None,
675        maxPriorityFeePerGas: Wei | None = None,
676    ) -> Sequence[ExtendEntityReturnType]:
677        """Extend the BTL of entities in Arkiv."""
678        return (
679            await self.send_transaction(
680                extensions=extensions,
681                gas=gas,
682                maxFeePerGas=maxFeePerGas,
683                maxPriorityFeePerGas=maxPriorityFeePerGas,
684            )
685        ).extensions
686
687    async def send_transaction(
688        self,
689        *,
690        creates: Sequence[ArkivCreate] | None = None,
691        updates: Sequence[ArkivUpdate] | None = None,
692        deletes: Sequence[ArkivDelete] | None = None,
693        extensions: Sequence[ArkivExtend] | None = None,
694        gas: int | None = None,
695        maxFeePerGas: Wei | None = None,
696        maxPriorityFeePerGas: Wei | None = None,
697    ) -> ArkivTransactionReceipt:
698        """
699        Send a generic transaction to Arkiv.
700
701        This transaction can contain multiple create, update, delete and
702        extend operations.
703        """
704        tx = ArkivTransaction(
705            creates=creates,
706            updates=updates,
707            deletes=deletes,
708            extensions=extensions,
709            gas=gas,
710            maxFeePerGas=maxFeePerGas,
711            maxPriorityFeePerGas=maxPriorityFeePerGas,
712        )
713        return await self._send_arkiv_transaction(tx)
714
715    async def _send_arkiv_transaction(
716        self, tx: ArkivTransaction
717    ) -> ArkivTransactionReceipt:
718        txData: TxParams = {
719            # https://github.com/pylint-dev/pylint/issues/3162
720            # pylint: disable=no-member
721            "to": STORAGE_ADDRESS.as_address(),
722            "value": AsyncWeb3.to_wei(0, "ether"),
723            "data": rlp_encode_transaction(tx),
724        }
725
726        if tx.gas:
727            txData |= {"gas": tx.gas}
728        if tx.maxFeePerGas:
729            txData |= {"maxFeePerGas": tx.maxFeePerGas}
730        if tx.maxPriorityFeePerGas:
731            txData |= {"maxPriorityFeePerGas": tx.maxPriorityFeePerGas}
732
733        txhash = await self.http_client().eth.send_transaction(txData)
734        receipt = await self.http_client().eth.wait_for_transaction_receipt(txhash)
735
736        # If we get a receipt and the transaction was failed, we run the same
737        # transaction with eth_call, which will simulate it and get us back the
738        # error that was reported by geth.
739        # Otherwise the error is not actually present in the receipt and so we
740        # don't have something useful to present to the user.
741        # This only happens when the gas price was explicitly provided, since
742        # otherwise there will be a call to eth_estimateGas, which will fail with
743        # the same error message that we would get here (and so we'll never actually
744        # get to submitting the transaction).
745        # The status in the receipt is either 0x0 for failed or 0x1 for success.
746        if not int(receipt["status"]):
747            # This call will lead to an exception, but that's OK, what we want
748            # is to raise a useful exception to the user with an error message.
749            try:
750                await self.http_client().eth.call(txData)
751            except Web3RPCError as e:
752                if e.rpc_response:
753                    error = e.rpc_response["error"]["message"]
754                    raise Exception(
755                        f"Error while processing transaction: {error}"
756                    ) from e
757                else:
758                    raise e
759
760        return await self._process_arkiv_receipt(receipt)