dynamically update relays and remove redundant nostr query

Author: f321x

electrum/simple_config.py

@@ -816,6 +816,7 @@ def __setattr__(self, name, value):
     SWAPSERVER_URL = ConfigVar('swapserver_url', default='', type_=str)
     TEST_SWAPSERVER_REFUND = ConfigVar('test_swapserver_refund', default=False, type_=bool)
     SWAPSERVER_NPUB = ConfigVar('swapserver_npub', default=None, type_=str)
+    LAST_SWAPSERVER_RELAYS = ConfigVar('last_swapserver_relays', default=None, type_=str)
     SWAPSERVER_POW_TARGET = ConfigVar('swapserver_pow_target', default=30, type_=int)
 
     # nostr

electrum/submarine_swaps.py

@@ -202,6 +202,7 @@ def __init__(self, *, wallet: 'Abstract_Wallet', lnworker: 'LNWallet'):
                 self.prepayments[swap.prepay_hash] = bytes.fromhex(k)
         self.is_server = False # overriden by swapserver plugin if enabled
         self.is_initialized = asyncio.Event()
+        self.pairs_updated = asyncio.Event()
 
     def start_network(self, network: 'Network'):
         assert network
@@ -922,7 +923,15 @@ def update_pairs(self, pairs):
         self.percentage = pairs.percentage
         self._min_amount = pairs.min_amount
         self._max_amount = pairs.max_amount
-        self.is_initialized.set()
+        self.trigger_pairs_updated()
+
+    def trigger_pairs_updated(self):
+        def trigger():
+            self.is_initialized.set()
+            self.pairs_updated.set()
+            self.pairs_updated.clear()
+        loop = get_asyncio_loop()
+        loop.call_soon_threadsafe(trigger)
 
     def get_max_amount(self) -> int:
         """in satoshis"""
@@ -1065,7 +1074,7 @@ def create_claim_txin(
             *,
             txin: PartialTxInput,
             swap: SwapData,
-    ) -> PartialTransaction:
+    ) -> Tuple[PartialTxInput, Optional[int]]:
         if swap.is_reverse:  # successful reverse swap
             locktime = None
             # preimage will be set in sign_tx
@@ -1280,11 +1289,10 @@ def __init__(self, config, sm, keypair):
         self.private_key = keypair.privkey
         self.nostr_private_key = to_nip19('nsec', keypair.privkey.hex())
         self.nostr_pubkey = keypair.pubkey.hex()[2:]
-        self.dm_replies = defaultdict(asyncio.Future)  # type: Dict[bytes, asyncio.Future]
+        self.dm_replies = defaultdict(asyncio.Future)  # type: Dict[str, asyncio.Future]
         self.ssl_context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH, cafile=ca_path)
         self.relay_manager = None
         self.taskgroup = OldTaskGroup()
-        self.server_relays = None
 
     def __enter__(self):
         asyncio.run_coroutine_threadsafe(self.main_loop(), self.network.asyncio_loop)
@@ -1311,8 +1319,8 @@ async def main_loop(self):
         else:
             tasks = [
                 self.check_direct_messages(),
-                self.receive_offers(),
                 self.get_pairs(),
+                self.update_relays()
             ]
         try:
             async with self.taskgroup as group:
@@ -1333,12 +1341,16 @@ async def stop(self):
 
     @property
     def relays(self):
-        return self.network.config.NOSTR_RELAYS.split(',')
+        our_relays = self.config.NOSTR_RELAYS.split(',') if self.config.NOSTR_RELAYS else []
+        if self.sm.is_server:
+            return our_relays
+        last_swapserver_relays = self.config.LAST_SWAPSERVER_RELAYS.split(',') if self.config.LAST_SWAPSERVER_RELAYS else []
+        return list(set(our_relays + last_swapserver_relays))
 
     def get_relay_manager(self):
         assert get_running_loop() == get_asyncio_loop(), f"this must be run on the asyncio thread!"
         if not self.relay_manager:
-            if self.network.proxy:
+            if self.network.proxy and self.network.proxy.enabled:
                 proxy = make_aiohttp_proxy_connector(self.network.proxy, self.ssl_context)
             else:
                 proxy: Optional['ProxyConnector'] = None
@@ -1396,7 +1408,7 @@ async def publish_offer(self, sm):
             private_key=self.nostr_private_key)
         self.logger.info(f"published offer {event_id}")
 
-    async def send_direct_message(self, pubkey: str, relays, content: str) -> str:
+    async def send_direct_message(self, pubkey: str, content: str) -> str:
         event_id = await aionostr._add_event(
             self.relay_manager,
             kind=self.NOSTR_DM,
@@ -1407,27 +1419,29 @@ async def send_direct_message(self, pubkey: str, relays, content: str) -> str:
 
     @log_exceptions
     async def send_request_to_server(self, method: str, request_data: dict) -> dict:
+        self.logger.debug(f"swapserver req: method: {method} relays: {self.relays}")
         request_data['method'] = method
-        request_data['relays'] = self.config.NOSTR_RELAYS
         server_pubkey = self.config.SWAPSERVER_NPUB
-        event_id = await self.send_direct_message(server_pubkey, self.server_relays, json.dumps(request_data))
+        event_id = await self.send_direct_message(server_pubkey, json.dumps(request_data))
         response = await self.dm_replies[event_id]
         return response
 
-    async def receive_offers(self):
+    async def get_pairs(self):
         await self.is_connected.wait()
         query = {
             "kinds": [self.USER_STATUS_NIP38],
             "limit":10,
             "#d": [f"electrum-swapserver-{self.NOSTR_EVENT_VERSION}"],
             "#r": [f"net:{constants.net.NET_NAME}"],
-            "since": int(time.time()) - self.OFFER_UPDATE_INTERVAL_SEC
+            "since": int(time.time()) - 60 * 60,
+            "until": int(time.time()) + 60 * 60,
         }
         async for event in self.relay_manager.get_events(query, single_event=False, only_stored=False):
             try:
                 content = json.loads(event.content)
                 tags = {k: v for k, v in event.tags}
             except Exception as e:
+                self.logger.debug(f"failed to parse event: {e}")
                 continue
             if tags.get('d') != f"electrum-swapserver-{self.NOSTR_EVENT_VERSION}":
                 continue
@@ -1436,8 +1450,9 @@ async def receive_offers(self):
             # check if this is the most recent event for this pubkey
             pubkey = event.pubkey
             ts = self._offers.get(pubkey, {}).get('timestamp', 0)
-            if event.created_at <= ts:
-                #print('skipping old event', pubkey[0:10], event.id)
+            if (event.created_at <= ts
+                    or event.created_at > time.time() + 60 * 60
+                    or event.created_at < time.time() - 60 * 60):
                 continue
             try:
                 pow_bits = get_nostr_ann_pow_amount(
@@ -1452,40 +1467,28 @@ async def receive_offers(self):
             content['pow_bits'] = pow_bits
             content['pubkey'] = pubkey
             content['timestamp'] = event.created_at
+            server_relays = content['relays'].split(',') if 'relays' in content else []
+            content['relays'] = ','.join(server_relays[:10])  # limit to 10 relays
             self._offers[pubkey] = content
+            if self.config.SWAPSERVER_NPUB == pubkey:
+                pairs = self._parse_offer(content)
+                self.sm.update_pairs(pairs)
             # mirror event to other relays
-            server_relays = content['relays'].split(',') if 'relays' in content else []
             await self.taskgroup.spawn(self.rebroadcast_event(event, server_relays))
 
-    async def get_pairs(self):
-        if not self.config.SWAPSERVER_NPUB:
-            return
-        query = {
-            "kinds": [self.USER_STATUS_NIP38],
-            "authors": [self.config.SWAPSERVER_NPUB],
-            "#d": [f"electrum-swapserver-{self.NOSTR_EVENT_VERSION}"],
-            "#r": [f"net:{constants.net.NET_NAME}"],
-            "since": int(time.time()) - self.OFFER_UPDATE_INTERVAL_SEC,
-            "limit": 1
-        }
-        async for event in self.relay_manager.get_events(query, single_event=True, only_stored=False):
-            try:
-                content = json.loads(event.content)
-                tags = {k: v for k, v in event.tags}
-            except Exception:
-                continue
-            if tags.get('d') != f"electrum-swapserver-{self.NOSTR_EVENT_VERSION}":
-                continue
-            if tags.get('r') != f"net:{constants.net.NET_NAME}":
-                continue
-            # check if this is the most recent event for this pubkey
-            pubkey = event.pubkey
-            content['pubkey'] = pubkey
-            content['timestamp'] = event.created_at
-            self.logger.info(f'received offer from {age(event.created_at)}')
-            pairs = self._parse_offer(content)
-            self.sm.update_pairs(pairs)
-            self.server_relays = content['relays'].split(',')
+    async def update_relays(self):
+        """
+        Update the relays when update_pairs is called.
+        This ensures we try to connect to the same relays as the ones announced by the swap server.
+        """
+        while True:
+            previous_relays = self.config.LAST_SWAPSERVER_RELAYS
+            await self.sm.pairs_updated.wait()
+            # assign the latest known swapserver relays to the config
+            self.config.LAST_SWAPSERVER_RELAYS = self._offers[self.config.SWAPSERVER_NPUB]['relays']
+            if self.config.LAST_SWAPSERVER_RELAYS != previous_relays:
+                self.logger.debug(f"updating relays")
+                await self.relay_manager.update_relays(self.relays)
 
     async def rebroadcast_event(self, event: Event, server_relays: Sequence[str]):
         """If the relays of the origin server are different from our relays we rebroadcast the
@@ -1528,8 +1531,7 @@ async def handle_request(self, request):
         method = request.pop('method')
         event_id = request.pop('event_id')
         event_pubkey = request.pop('event_pubkey')
-        self.logger.info(f'handle_request: id={event_id} {method} {request}')
-        relays = request.pop('relays').split(',')
+        print(f'handle_request: id={event_id} {method} {request}')
         if method == 'addswapinvoice':
             r = self.sm.server_add_swap_invoice(request)
         elif method == 'createswap':
@@ -1540,4 +1542,4 @@ async def handle_request(self, request):
             raise Exception(method)
         r['reply_to'] = event_id
         self.logger.info(f'sending response id={event_id}')
-        await self.send_direct_message(event_pubkey, relays, json.dumps(r))
+        await self.send_direct_message(event_pubkey, json.dumps(r))