Merge pull request #643 from PR0M3TH3AN/codex/implement-observer-pattern-in-pubsub-module

Add event bus and sync progress events

src/seedpass/core/api.py

@@ -15,6 +15,7 @@ import json
 from pydantic import BaseModel
 
 from .manager import PasswordManager
+from .pubsub import bus
 
 
 class VaultExportRequest(BaseModel):
@@ -202,7 +203,9 @@ class SyncService:
         """Publish the vault to Nostr and return event info."""
 
         with self._lock:
+            bus.publish("sync_started")
             result = self._manager.sync_vault()
+            bus.publish("sync_finished", result)
         if not result:
             return None
         return SyncResponse(**result)

src/seedpass/core/manager.py

@@ -33,6 +33,7 @@ from .vault import Vault
 from .portable_backup import export_backup, import_backup
 from .totp import TotpManager
 from .entry_types import EntryType
+from .pubsub import bus
 from utils.key_derivation import (
     derive_key_from_parent_seed,
     derive_key_from_password,
@@ -1243,7 +1244,9 @@ class PasswordManager:
 
         def _worker() -> None:
             try:
-                asyncio.run(self.sync_vault_async(alt_summary=alt_summary))
+                bus.publish("sync_started")
+                result = asyncio.run(self.sync_vault_async(alt_summary=alt_summary))
+                bus.publish("sync_finished", result)
             except Exception as exc:
                 logging.error(f"Background vault sync failed: {exc}", exc_info=True)
 
@@ -1252,7 +1255,13 @@ class PasswordManager:
         except RuntimeError:
             threading.Thread(target=_worker, daemon=True).start()
         else:
-            asyncio.create_task(self.sync_vault_async(alt_summary=alt_summary))
+
+            async def _async_worker() -> None:
+                bus.publish("sync_started")
+                result = await self.sync_vault_async(alt_summary=alt_summary)
+                bus.publish("sync_finished", result)
+
+            asyncio.create_task(_async_worker())
 
     async def attempt_initial_sync_async(self) -> bool:
         """Attempt to download the initial vault snapshot from Nostr.

src/seedpass/core/pubsub.py

@@ -0,0 +1,27 @@
+from collections import defaultdict
+from typing import Callable, Dict, List, Any
+
+
+class PubSub:
+    """Simple in-process event bus using the observer pattern."""
+
+    def __init__(self) -> None:
+        self._subscribers: Dict[str, List[Callable[..., None]]] = defaultdict(list)
+
+    def subscribe(self, event: str, callback: Callable[..., None]) -> None:
+        """Register ``callback`` to be invoked when ``event`` is published."""
+        self._subscribers[event].append(callback)
+
+    def unsubscribe(self, event: str, callback: Callable[..., None]) -> None:
+        """Unregister ``callback`` from ``event`` notifications."""
+        if callback in self._subscribers.get(event, []):
+            self._subscribers[event].remove(callback)
+
+    def publish(self, event: str, *args: Any, **kwargs: Any) -> None:
+        """Notify all subscribers of ``event`` passing ``*args`` and ``**kwargs``."""
+        for callback in list(self._subscribers.get(event, [])):
+            callback(*args, **kwargs)
+
+
+# Global bus instance for convenience
+bus = PubSub()

src/tests/test_pubsub.py

@@ -0,0 +1,28 @@
+from seedpass.core.pubsub import PubSub
+
+
+def test_subscribe_and_publish():
+    bus = PubSub()
+    calls = []
+
+    def handler(arg):
+        calls.append(arg)
+
+    bus.subscribe("event", handler)
+    bus.publish("event", 123)
+
+    assert calls == [123]
+
+
+def test_unsubscribe():
+    bus = PubSub()
+    calls = []
+
+    def handler():
+        calls.append(True)
+
+    bus.subscribe("event", handler)
+    bus.unsubscribe("event", handler)
+    bus.publish("event")
+
+    assert calls == []