refresh topic before resubscribing

ntfy/bot.py

@@ -152,13 +152,17 @@ class NtfyBot(Plugin):
             if task.done() and not task.cancelled():
                 exc = task.exception()
                 self.log.exception(
-                    "Subscription task errored, resubscribing", exc_info=exc
-                )
-                self.tasks[topic.id] = self.loop.create_task(asyncio.sleep(10.0))
-                self.tasks[topic.id].add_done_callback(
-                    lambda _: self.subscribe_to_topic(topic)
+                    "Subscription task errored, resubscribing in 10 seconds", exc_info=exc
                 )
 
+                async def resub(topic: Topic):
+                    # sleep to make sure we don't have a busy loop
+                    await asyncio.sleep(10.0)
+                    # refresh topic to make sure we don't use stale data
+                    topic = await self.db.get_topic_by_id(topic.id)
+                    self.subscribe_to_topic(topic)
+                self.tasks[topic.id] = self.loop.create_task(resub(topic))
+
         self.log.info(f"Subscribing to {topic.server}/{topic.topic}")
         url = f"{topic.server}/{topic.topic}/json"
         if not url.startswith(("http://", "https://")):

ntfy/db.py

@@ -147,6 +147,14 @@ class DB:
         """
         return Topic.from_row(await self.db.fetchrow(query, server, topic))
 
+    async def get_topic_by_id(self, id: int) -> Topic | None:
+        query = """
+        SELECT id, server, topic, last_event_id
+        FROM topics
+        WHERE id = $1
+        """
+        return Topic.from_row(await self.db.fetchrow(query, id))
+
     async def get_subscription(self, topic_id: int, room_id: RoomID) -> Tuple[Subscription | None, Topic | None]:
         query = """
         SELECT id, server, topic, last_event_id, topic_id, room_id