Fix withdrawals

bridge_indexer/dipdup.yaml

@@ -79,19 +79,19 @@ indexes:
     callback: tezos.on_head
 
   tezos_rollup_cement:
-#    kind: tezos.operations
-#    datasources:
-#      - tzkt
-#    types:
-#      - sr_cement
-#    contracts:
-#      - tezos_smart_rollup
-#    handlers:
-#      - callback: tezos.on_cement_commitment
-#        pattern:
-#          - type: sr_cement
-#            destination: tezos_smart_rollup
-#
+    kind: tezos.operations
+    datasources:
+      - tzkt
+    types:
+      - sr_cement
+    contracts:
+      - tezos_smart_rollup
+    handlers:
+      - callback: tezos.on_cement_commitment
+        pattern:
+          - type: sr_cement
+            destination: tezos_smart_rollup
+
 
   tezos_deposit_operations:
     kind: tezos.operations
@@ -128,6 +128,57 @@ indexes:
         from_: etherlink_rollup_kernel
 
 
+  etherlink_withdrawal_events:
+    kind: evm.events
+    datasources:
+      - etherlink_subsquid
+      - etherlink_node
+    handlers:
+      - callback: etherlink.on_withdraw
+        contract: etherlink_rollup_kernel
+        name: Withdrawal
+
+  tezos_withdrawal_operations:
+    kind: tezos.operations
+    datasources:
+      - tzkt
+    types:
+      - sr_execute
+    contracts:
+      - tezos_smart_rollup
+    handlers:
+      - callback: tezos.on_rollup_execute
+        pattern:
+          - type: sr_execute
+            destination: tezos_smart_rollup
+
+
+  etherlink_token_balance_update_events:
+    kind: evm.events
+    datasources:
+      - etherlink_subsquid
+      - etherlink_node
+    handlers:
+      - callback: etherlink.on_transfer
+        contract: l2_tzbtc_token
+        name: Transfer
+      - callback: etherlink.on_transfer
+        contract: l2_sirs_token
+        name: Transfer
+      - callback: etherlink.on_transfer
+        contract: l2_usdt_token
+        name: Transfer
+
+
+hooks:
+  bridge_matcher:
+    callback: bridge_matcher
+jobs:
+  bridge_matcher:
+    hook: bridge_matcher
+    daemon: True
+
+
 advanced:
   reindex:
     config_modified: ignore

bridge_indexer/handlers/batch.py

@@ -1,20 +1,16 @@
 import logging
-from typing import Any
 
-from dipdup.config import HandlerConfig
-from dipdup.context import HandlerContext
-from dipdup.index import Index
+logger = logging.getLogger('bridge_indexer.handlers.batch')
 
-from bridge_indexer.handlers.bridge_matcher import BridgeMatcher
 
-logger = logging.getLogger('bridge_indexer.handlers.batch')
+from collections.abc import Iterable
 
+from dipdup.context import HandlerContext
+from dipdup.index import MatchedHandler
 
 async def batch(
     ctx: HandlerContext,
-    handlers: tuple[tuple[Index[Any, Any, Any], HandlerConfig, Any]],
+    handlers: Iterable[MatchedHandler],
 ) -> None:
-    for index, handler, data in handlers:
-        await index._call_matched_handler(handler, data)
-
-    await BridgeMatcher.check_pending_transactions()
+    for handler in handlers:
+        await ctx.fire_matched_handler(handler)

bridge_indexer/handlers/bridge_matcher.py

@@ -1,5 +1,7 @@
+import threading
 from datetime import timedelta
 
+from bridge_indexer.handlers.bridge_matcher_locks import BridgeMatcherLocks
 from bridge_indexer.models import BridgeDepositOperation
 from bridge_indexer.models import BridgeOperation
 from bridge_indexer.models import BridgeOperationStatus
@@ -8,45 +10,22 @@
 from bridge_indexer.models import EtherlinkDepositOperation
 from bridge_indexer.models import EtherlinkWithdrawOperation
 from bridge_indexer.models import RollupInboxMessage
+from bridge_indexer.models import RollupOutboxMessage
 from bridge_indexer.models import TezosDepositOperation
 from bridge_indexer.models import TezosWithdrawOperation
 
 LAYERS_TIMESTAMP_GAP_MAX = timedelta(seconds=20 * 7)
 
 
 class BridgeMatcher:
-    _pending_tezos_deposits: bool = False
-    _pending_etherlink_withdrawals: bool = False
-    _pending_etherlink_deposits: bool = False
-    _pending_etherlink_xtz_deposits: bool = False
-    _pending_tezos_withdrawals: bool = False
-
-    @classmethod
-    def set_pending_tezos_deposits(cls):
-        cls._pending_tezos_deposits = True
-
-    @classmethod
-    def set_pending_etherlink_withdrawals(cls):
-        cls._pending_etherlink_withdrawals = True
-
-    @classmethod
-    def set_pending_etherlink_deposits(cls):
-        cls._pending_etherlink_deposits = True
-
-    @classmethod
-    def set_pending_etherlink_xtz_deposits(cls):
-        cls._pending_etherlink_xtz_deposits = True
-
-    @classmethod
-    def set_pending_tezos_withdrawals(cls):
-        cls._pending_tezos_withdrawals = True
+    matcher_lock = threading.Lock()
 
     @classmethod
     async def check_pending_tezos_deposits(cls):
-        if not cls._pending_tezos_deposits:
+        if not BridgeMatcherLocks.pending_tezos_deposits:
             return
         else:
-            cls._pending_tezos_deposits = False
+            BridgeMatcherLocks.pending_tezos_deposits = False
 
         qs = TezosDepositOperation.filter(bridge_deposits=None)
         async for l1_deposit in qs:
@@ -64,6 +43,11 @@ async def check_pending_tezos_deposits(cls):
 
     @classmethod
     async def check_pending_inbox(cls):
+        if not BridgeMatcherLocks.pending_inbox:
+            return
+        else:
+            BridgeMatcherLocks.pending_inbox = False
+
         qs = BridgeDepositOperation.filter(
             inbox_message=None,
         ).order_by(
@@ -86,10 +70,10 @@ async def check_pending_inbox(cls):
 
     @classmethod
     async def check_pending_etherlink_deposits(cls):
-        if not cls._pending_etherlink_deposits:
+        if not BridgeMatcherLocks.pending_etherlink_deposits:
             return
         else:
-            cls._pending_etherlink_deposits = False
+            BridgeMatcherLocks.pending_etherlink_deposits = False
 
         qs = EtherlinkDepositOperation.filter(
             bridge_deposits=None,
@@ -128,10 +112,10 @@ async def check_pending_etherlink_deposits(cls):
 
     @classmethod
     async def check_pending_etherlink_xtz_deposits(cls):
-        if not cls._pending_etherlink_xtz_deposits:
+        if not BridgeMatcherLocks.pending_etherlink_xtz_deposits:
             return
         else:
-            cls._pending_etherlink_xtz_deposits = False
+            BridgeMatcherLocks.pending_etherlink_xtz_deposits = False
 
         qs = EtherlinkDepositOperation.filter(
             bridge_deposits=None,
@@ -171,18 +155,67 @@ async def check_pending_etherlink_xtz_deposits(cls):
             bridge_operation.status = BridgeOperationStatus.finished
             await bridge_operation.save()
 
+    @classmethod
+    async def check_pending_etherlink_withdrawals(cls):
+        if not BridgeMatcherLocks.pending_etherlink_withdrawals:
+            return
+        else:
+            BridgeMatcherLocks.pending_etherlink_withdrawals = False
+
+        qs = EtherlinkWithdrawOperation.filter(bridge_withdrawals=None)
+        async for l2_withdrawal in qs:
+            bridge_withdrawal = await BridgeWithdrawOperation.create(l2_transaction=l2_withdrawal)
+            await BridgeOperation.create(
+                id=bridge_withdrawal.id,
+                type=BridgeOperationType.withdrawal,
+                l1_account=l2_withdrawal.l1_account,
+                l2_account=l2_withdrawal.l2_account,
+                created_at=l2_withdrawal.timestamp,
+                updated_at=l2_withdrawal.timestamp,
+                status=BridgeOperationStatus.created,
+            )
+
+    @classmethod
+    async def check_pending_outbox(cls):
+        if not BridgeMatcherLocks.pending_outbox:
+            return
+        else:
+            BridgeMatcherLocks.pending_outbox = False
+
+        qs = BridgeWithdrawOperation.filter(
+            outbox_message=None,
+        ).order_by(
+            'l2_transaction__level', 'l2_transaction__transaction_index', 'l2_transaction__log_index',
+        ).prefetch_related('l2_transaction')
+        async for bridge_withdrawal in qs:
+            bridge_withdrawal: BridgeWithdrawOperation
+            outbox_message = await RollupOutboxMessage.filter(
+                parameters_hash=bridge_withdrawal.l2_transaction.parameters_hash,
+                # created_at__gte=bridge_withdrawal.l2_transaction.timestamp,
+                # created_at__gte=datetime.fromtimestamp(bridge_withdrawal.l2_transaction.timestamp, tz=timezone.utc),
+            ).order_by('level', 'index').first()
+
+            if outbox_message:
+                bridge_withdrawal.outbox_message = outbox_message
+                await bridge_withdrawal.save()
+                bridge_withdrawal.l2_transaction.parameters_hash = None
+                await bridge_withdrawal.l2_transaction.save()
+                outbox_message.parameters_hash = None
+                await outbox_message.save()
+
     @classmethod
     async def check_pending_tezos_withdrawals(cls):
-        if not cls._pending_tezos_withdrawals:
+        if not BridgeMatcherLocks.pending_tezos_withdrawals:
             return
         else:
-            cls._pending_tezos_withdrawals = False
+            BridgeMatcherLocks.pending_tezos_withdrawals = False
 
-        qs = TezosWithdrawOperation.filter(bridge_withdrawals__isnull=True).order_by('level')
+        qs = TezosWithdrawOperation.filter(bridge_withdrawals=None).order_by('level')
         async for l1_withdrawal in qs:
+            l1_withdrawal: TezosWithdrawOperation
             bridge_withdrawal = await BridgeWithdrawOperation.filter(
                 l1_transaction=None,
-                l2_transaction__outbox_message_id=l1_withdrawal.outbox_message_id,
+                outbox_message_id=l1_withdrawal.outbox_message_id,
             ).first()
 
             if not bridge_withdrawal:
@@ -197,12 +230,3 @@ async def check_pending_tezos_withdrawals(cls):
             bridge_operation.updated_at = l1_withdrawal.timestamp
             bridge_operation.status = BridgeOperationStatus.finished
             await bridge_operation.save()
-
-    @staticmethod
-    async def check_pending_transactions():
-        await BridgeMatcher.check_pending_tezos_deposits()
-        await BridgeMatcher.check_pending_etherlink_withdrawals()
-
-        await BridgeMatcher.check_pending_etherlink_deposits()
-        await BridgeMatcher.check_pending_etherlink_xtz_deposits()
-        await BridgeMatcher.check_pending_tezos_withdrawals()

bridge_indexer/handlers/bridge_matcher_locks.py

@@ -0,0 +1,36 @@
+class BridgeMatcherLocks:
+    pending_tezos_deposits: bool = False
+    pending_etherlink_withdrawals: bool = False
+    pending_etherlink_deposits: bool = False
+    pending_etherlink_xtz_deposits: bool = False
+    pending_tezos_withdrawals: bool = False
+    pending_inbox: bool = False
+    pending_outbox: bool = False
+
+    @classmethod
+    def set_pending_tezos_deposits(cls):
+        BridgeMatcherLocks.pending_tezos_deposits = True
+
+    @classmethod
+    def set_pending_etherlink_withdrawals(cls):
+        BridgeMatcherLocks.pending_etherlink_withdrawals = True
+
+    @classmethod
+    def set_pending_etherlink_deposits(cls):
+        BridgeMatcherLocks.pending_etherlink_deposits = True
+
+    @classmethod
+    def set_pending_etherlink_xtz_deposits(cls):
+        BridgeMatcherLocks.pending_etherlink_xtz_deposits = True
+
+    @classmethod
+    def set_pending_tezos_withdrawals(cls):
+        BridgeMatcherLocks.pending_tezos_withdrawals = True
+
+    @classmethod
+    def set_pending_inbox(cls):
+        BridgeMatcherLocks.pending_inbox = True
+
+    @classmethod
+    def set_pending_outbox(cls):
+        BridgeMatcherLocks.pending_outbox = True

bridge_indexer/handlers/etherlink/on_deposit.py

@@ -4,7 +4,7 @@
 from dipdup.context import HandlerContext
 from dipdup.models.evm import EvmEvent
 
-from bridge_indexer.handlers.bridge_matcher import BridgeMatcher
+from bridge_indexer.handlers.bridge_matcher_locks import BridgeMatcherLocks
 from bridge_indexer.models import EtherlinkDepositOperation
 from bridge_indexer.models import EtherlinkToken
 from bridge_indexer.models import TezosTicket
@@ -82,5 +82,4 @@ async def on_deposit(
 
     ctx.logger.info(f'Etherlink Deposit Event registered: {deposit.id}')
 
-    BridgeMatcher.set_pending_etherlink_deposits()
-    await BridgeMatcher.check_pending_transactions()
+    BridgeMatcherLocks.set_pending_etherlink_deposits()