diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..981d3f0 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,59 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +Nostrclient is an LNbits extension that acts as an always-on Nostr relay multiplexer. Multiple Nostr clients connect to a single WebSocket endpoint, which fans out requests to multiple configured Nostr relays and aggregates/deduplicates responses. It rewrites subscription IDs per-client to prevent conflicts. + +## Build & Development Commands + +All commands use `uv` as the Python package manager: + +```bash +make format # Format: prettier + black + ruff --fix +make check # All checks: mypy, pyright, black --check, ruff check, prettier --check +make test # Run pytest (DEBUG=true PYTHONUNBUFFERED=1 uv run pytest) +make mypy # Type check (excludes nostr/ directory) +make pre-commit # Run pre-commit hooks on all files +``` + +Individual checks: `make checkblack`, `make checkruff`, `make checkprettier`. + +CI runs lint then pytest with `LNBITS_BACKEND_WALLET_CLASS=FakeWallet`. + +## Architecture + +**Request flow:** Nostr Clients → WebSocket → NostrRouter → RelayManager → Nostr Relays + +Key components: + +- **NostrRouter** (`router.py`) — One per client WebSocket connection. Rewrites subscription IDs (original → hashed → original) to isolate clients. Two async tasks: `_client_to_nostr` (forward requests) and `_nostr_to_client` (deliver aggregated responses). + +- **NostrClient** (`nostr/client/client.py`) — Singleton orchestrator. Owns the RelayManager. Polls MessagePool and dispatches events via callbacks to routers. + +- **RelayManager** (`nostr/relay_manager.py`) — Manages connections to multiple relays. Caches subscriptions so new relays receive existing subscriptions. Runs health checks via `check_and_restart_relays()`. + +- **Relay** (`nostr/relay.py`) — Individual relay WebSocket connection with retry/backoff, ping latency tracking, and error counting. + +- **MessagePool** (`nostr/message_pool.py`) — Thread-safe event aggregation with deduplication by event ID across all relays. + +**Hybrid threading model:** Relay connections use threads (via `RelayManager.open_connections()`); client communication uses asyncio. The bridge is in `tasks.py` where `subscribe_events()` runs in a thread executor. + +**Lifecycle** (`__init__.py`): `nostrclient_start()` spawns three background tasks (init relays, subscribe events, check relays). `nostrclient_stop()` cancels tasks, stops routers, closes the client. + +## API Endpoints (views_api.py) + +- REST endpoints under `/api/v1/` for relay CRUD and config (admin-authenticated) +- WebSocket endpoints: `/api/v1/{ws_id}` (private, encrypted ID) and `/api/v1/relay` (public, if enabled) + +## Database + +Three migrations in `migrations.py`: relays table, config table (JSON `extra` field), config owner scoping. CRUD in `crud.py` uses LNbits database abstraction. + +## Code Quality Notes + +- **mypy excludes `nostr/*`** — this is a custom Nostr protocol implementation, not a third-party package +- **Ruff rules:** F, E, W, I, A, C, N, UP, RUF, B +- **Frontend:** Vue.js + Quasar via LNbits base templates (`templates/nostrclient/index.html`) +- Pub key helpers in `helpers.py` normalize between hex and bech32 (npub1) formats diff --git a/README.md b/README.md index 2006af6..70593a8 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,6 @@ flowchart LR 3. **Fan-Out** - Subscription requests are sent to all configured relays 4. **Aggregation** - Events from all relays are collected and deduplicated 5. **Response** - Events are sent back to the client with the original subscription ID -6. **Publish Acknowledgement** - Every `EVENT` a client publishes gets exactly one `["OK", , , ]` reply (NIP-01): `true` as soon as any relay accepts it, `false` once every relay has rejected it, the wait times out (10 s), or no relay is connected. A `REQ` while no relay is configured at all is answered with `CLOSED` ## Configuration diff --git a/config.json b/config.json index 048a5a7..1f58e7b 100644 --- a/config.json +++ b/config.json @@ -1,7 +1,7 @@ { "name": "Nostr Client", "short_description": "Nostr relay multiplexer", - "version": "1.2.0-aio.3", + "version": "1.1.0", "tile": "/nostrclient/static/images/nostr-bitcoin.png", "contributors": ["calle", "motorina0", "dni"], "min_lnbits_version": "1.4.0", diff --git a/nostr/client/client.py b/nostr/client/client.py index 0412c1b..d6fb5c8 100644 --- a/nostr/client/client.py +++ b/nostr/client/client.py @@ -39,13 +39,11 @@ class NostrClient: callback_events_func=None, callback_notices_func=None, callback_eosenotices_func=None, - callback_command_results_func=None, ): while self.running: self._check_events(callback_events_func) self._check_notices(callback_notices_func) self._check_eos_notices(callback_eosenotices_func) - self._check_command_results(callback_command_results_func) await asyncio.sleep(0.2) @@ -75,12 +73,3 @@ class NostrClient: callback_eosenotices_func(event_msg) except Exception as e: logger.debug(e) - - def _check_command_results(self, callback_command_results_func=None): - try: - while self.relay_manager.message_pool.has_command_results(): - result_msg = self.relay_manager.message_pool.get_command_result() - if callback_command_results_func: - callback_command_results_func(result_msg) - except Exception as e: - logger.debug(e) diff --git a/nostr/message_pool.py b/nostr/message_pool.py index 30c1469..a3e6c5f 100644 --- a/nostr/message_pool.py +++ b/nostr/message_pool.py @@ -27,22 +27,11 @@ class EndOfStoredEventsMessage: self.url = url -class CommandResultMessage: - """An `["OK", , , ]` reply from one relay.""" - - def __init__(self, event_id: str, accepted: bool, message: str, url: str) -> None: - self.event_id = event_id - self.accepted = accepted - self.message = message - self.url = url - - class MessagePool: def __init__(self) -> None: self.events: Queue[EventMessage] = Queue() self.notices: Queue[NoticeMessage] = Queue() self.eose_notices: Queue[EndOfStoredEventsMessage] = Queue() - self.command_results: Queue[CommandResultMessage] = Queue() self._unique_events: set = set() self.lock: Lock = Lock() @@ -58,9 +47,6 @@ class MessagePool: def get_eose_notice(self): return self.eose_notices.get() - def get_command_result(self): - return self.command_results.get() - def has_events(self): return self.events.qsize() > 0 @@ -70,9 +56,6 @@ class MessagePool: def has_eose_notices(self): return self.eose_notices.qsize() > 0 - def has_command_results(self): - return self.command_results.qsize() > 0 - def _process_message(self, message: str, url: str): message_json = json.loads(message) message_type = message_json[0] @@ -92,13 +75,6 @@ class MessagePool: self.notices.put(NoticeMessage(message_json[1], url)) elif message_type == RelayMessageType.END_OF_STORED_EVENTS: self.eose_notices.put(EndOfStoredEventsMessage(message_json[1], url)) - elif message_type == RelayMessageType.COMMAND_RESULT: - message = message_json[3] if len(message_json) > 3 else "" - self.command_results.put( - CommandResultMessage( - message_json[1], bool(message_json[2]), str(message), url - ) - ) def _accept_event(self, event_message: EventMessage): """ diff --git a/nostr/relay.py b/nostr/relay.py index d762963..bac1be7 100644 --- a/nostr/relay.py +++ b/nostr/relay.py @@ -1,6 +1,9 @@ +from __future__ import annotations + import asyncio import json import time +from collections.abc import Callable from queue import Queue from loguru import logger @@ -28,6 +31,7 @@ class Relay: self.num_subscriptions: int = 0 self.queue: Queue = Queue() + self.on_connect: Callable[[Relay], None] | None = None def connect(self): self.ws = WebSocketApp( @@ -97,6 +101,11 @@ class Relay: logger.info(f"[Relay: {self.url}] Connected.") self.connected = True self.shutdown = False + if self.on_connect: + try: + self.on_connect(self) + except Exception as e: + logger.warning(f"[Relay: {self.url}] on_connect callback error: {e}") def _on_close(self, _, status_code, message): logger.warning( diff --git a/nostr/relay_manager.py b/nostr/relay_manager.py index 2aa27c5..2b90216 100644 --- a/nostr/relay_manager.py +++ b/nostr/relay_manager.py @@ -1,6 +1,8 @@ import asyncio +import json import threading import time +from collections import deque from typing import List from loguru import logger @@ -9,6 +11,8 @@ from .message_pool import MessagePool, NoticeMessage from .relay import Relay from .subscription import Subscription +PENDING_EVENTS_MAX = 100 + class RelayManager: def __init__(self) -> None: @@ -18,6 +22,8 @@ class RelayManager: self.message_pool = MessagePool() self._cached_subscriptions: dict[str, Subscription] = {} self._subscriptions_lock = threading.Lock() + self._pending_events: deque[str] = deque(maxlen=PENDING_EVENTS_MAX) + self._pending_events_lock = threading.Lock() def add_relay(self, url: str) -> Relay: if url in list(self.relays.keys()): @@ -25,6 +31,7 @@ class RelayManager: return self.relays[url] relay = Relay(url, self.message_pool) + relay.on_connect = self._on_relay_connect self.relays[url] = relay self._open_connection(relay) @@ -100,8 +107,19 @@ class RelayManager: relay.close() def publish_message(self, message: str): - for relay in self.relays.values(): - relay.publish(message) + connected_relays = [ + r for r in self.relays.values() if r.connected and not r.shutdown + ] + if connected_relays: + for relay in self.relays.values(): + relay.publish(message) + else: + with self._pending_events_lock: + self._pending_events.append(message) + logger.warning( + f"No connected relays. Queued outgoing event " + f"({len(self._pending_events)}/{PENDING_EVENTS_MAX})." + ) def handle_notice(self, notice: NoticeMessage): relay = next((r for r in self.relays.values() if r.url == notice.url)) @@ -137,7 +155,47 @@ class RelayManager: logger.info(f"Restarting connection to relay '{relay.url}'") + self._drain_relay_queue(relay) self.remove_relay(relay.url) new_relay = self.add_relay(relay.url) new_relay.error_counter = relay.error_counter new_relay.error_list = relay.error_list + + def _drain_relay_queue(self, relay: Relay): + """Move pending EVENT messages from a dead relay's queue to the + manager's pending queue so they can be resent on reconnection.""" + drained = 0 + while not relay.queue.empty(): + try: + message = relay.queue.get_nowait() + data = json.loads(message) + if data[0] == "EVENT": + with self._pending_events_lock: + self._pending_events.append(message) + drained += 1 + except Exception: + break + if drained: + logger.info(f"Drained {drained} pending event(s) from relay '{relay.url}'.") + + def _on_relay_connect(self, _relay: Relay): + self._flush_pending_events() + + def _flush_pending_events(self): + with self._pending_events_lock: + if not self._pending_events: + return + connected_relays = [ + r for r in self.relays.values() if r.connected and not r.shutdown + ] + if not connected_relays: + return + count = len(self._pending_events) + while self._pending_events: + message = self._pending_events.popleft() + for relay in connected_relays: + relay.publish(message) + logger.info( + f"Flushed {count} pending event(s) to " + f"{len(connected_relays)} relay(s)." + ) diff --git a/router.py b/router.py index 8446512..a7054e9 100644 --- a/router.py +++ b/router.py @@ -1,6 +1,5 @@ import asyncio import json -import time from typing import ClassVar from fastapi import WebSocket, WebSocketDisconnect @@ -10,30 +9,11 @@ from loguru import logger from .nostr.client.client import NostrClient # from . import nostr_client -from .nostr.message_pool import ( - CommandResultMessage, - EndOfStoredEventsMessage, - EventMessage, - NoticeMessage, -) +from .nostr.message_pool import EndOfStoredEventsMessage, EventMessage, NoticeMessage nostr_client: NostrClient = NostrClient() all_routers: list["NostrRouter"] = [] -# How long to wait for relays to answer an EVENT before replying `OK false`. -PUBLISH_TIMEOUT_SECONDS = 10 - - -class PendingPublish: - """An EVENT a client sent that still awaits its `OK` reply.""" - - def __init__(self, event_id: str, relay_count: int) -> None: - self.event_id = event_id - # relays connected at publish time; once this many answered we stop - # waiting for an acceptance - self.relay_count = relay_count - self.sent_at = time.time() - class NostrRouter: received_subscription_events: ClassVar[dict[str, list[EventMessage]]] = {} @@ -41,14 +21,12 @@ class NostrRouter: received_subscription_eosenotices: ClassVar[dict[str, EndOfStoredEventsMessage]] = ( {} ) - received_command_results: ClassVar[dict[str, list[CommandResultMessage]]] = {} def __init__(self, websocket: WebSocket): self.connected: bool = True self.websocket: WebSocket = websocket self.tasks: list[asyncio.Task] = [] self.original_subscription_ids: dict[str, str] = {} - self.pending_publishes: dict[str, PendingPublish] = {} @property def subscriptions(self) -> list[str]: @@ -62,9 +40,6 @@ class NostrRouter: async def stop(self): nostr_client.relay_manager.close_subscriptions(self.subscriptions) self.connected = False - for event_id in self.pending_publishes: - NostrRouter.received_command_results.pop(event_id, None) - self.pending_publishes.clear() for t in self.tasks: try: @@ -99,7 +74,6 @@ class NostrRouter: while self.connected: try: await self._handle_subscriptions() - await self._handle_command_results() self._handle_notices() except Exception as e: logger.debug(f"Failed to handle response for client: '{e!s}'.") @@ -146,42 +120,6 @@ class NostrRouter: f"[NOSTRCLIENT] Error in _handle_received_subscription_events: {e}" ) - async def _handle_command_results(self): - """ - Reply exactly one `["OK", , , ]` per EVENT a - client published (NIP-01). The EVENT was fanned out to every relay, so - several OKs can come back for one id: the client gets `true` as soon - as any relay accepts, and `false` once every relay has rejected it or - the wait times out. - """ - for event_id in list(self.pending_publishes.keys()): - pending = self.pending_publishes[event_id] - results = NostrRouter.received_command_results.get(event_id, []) - accepted = next((r for r in results if r.accepted), None) - if accepted: - await self._send_ok(event_id, True, accepted.message) - elif len(results) >= pending.relay_count: - await self._send_ok(event_id, False, results[-1].message) - elif time.time() - pending.sent_at > PUBLISH_TIMEOUT_SECONDS: - message = ( - results[-1].message - if results - else "error: timed out waiting for relays" - ) - await self._send_ok(event_id, False, message) - else: - continue - self.pending_publishes.pop(event_id, None) - NostrRouter.received_command_results.pop(event_id, None) - - async def _send_ok(self, event_id: str, accepted: bool, message: str): - try: - await self.websocket.send_text( - json.dumps(["OK", event_id, accepted, message]) - ) - except Exception as e: - logger.debug(f"Failed to send OK for '{event_id}': {e}") - def _handle_notices(self): while len(NostrRouter.received_subscription_notices): my_event = NostrRouter.received_subscription_notices.pop(0) @@ -195,7 +133,7 @@ class NostrRouter: assert len(json_data), "Bad JSON array" if json_data[0] == "REQ": - await self._handle_client_req(json_data) + self._handle_client_req(json_data) return if json_data[0] == "CLOSE": @@ -203,49 +141,18 @@ class NostrRouter: return if json_data[0] == "EVENT": - await self._handle_client_event(json_data, json_str) + nostr_client.relay_manager.publish_message(json_str) return - async def _handle_client_event(self, json_data, json_str): - event = json_data[1] if len(json_data) > 1 else None - event_id = event.get("id") if isinstance(event, dict) else None - if not event_id: - logger.debug("Ignoring EVENT without an id.") - return - - connected = [ - r for r in nostr_client.relay_manager.relays.values() if r.connected - ] - if not connected: - await self._send_ok(event_id, False, "error: no relays connected") - return - - self.pending_publishes[event_id] = PendingPublish(event_id, len(connected)) - nostr_client.relay_manager.publish_message(json_str) - - async def _handle_client_req(self, json_data): + def _handle_client_req(self, json_data): subscription_id = json_data[1] logger.info(f"New subscription: '{subscription_id}'") - if not nostr_client.relay_manager.relays: - # NIP-01: nothing will ever serve this subscription, so refuse it. - # Only when no relay is configured at all: a configured relay that - # is merely reconnecting gets the cached subscription replayed. - await self._send_closed(subscription_id, "error: no relay connections") - return subscription_id_rewritten = urlsafe_short_hash() self.original_subscription_ids[subscription_id_rewritten] = subscription_id filters = json_data[2:] nostr_client.relay_manager.add_subscription(subscription_id_rewritten, filters) - async def _send_closed(self, subscription_id: str, message: str): - try: - await self.websocket.send_text( - json.dumps(["CLOSED", subscription_id, message]) - ) - except Exception as e: - logger.debug(f"Failed to send CLOSED for '{subscription_id}': {e}") - def _handle_client_close(self, subscription_id): subscription_id_rewritten = next( ( diff --git a/tasks.py b/tasks.py index e2abdcb..2a76765 100644 --- a/tasks.py +++ b/tasks.py @@ -4,13 +4,8 @@ import threading from loguru import logger from .crud import get_relays -from .nostr.message_pool import ( - CommandResultMessage, - EndOfStoredEventsMessage, - EventMessage, - NoticeMessage, -) -from .router import NostrRouter, all_routers, nostr_client +from .nostr.message_pool import EndOfStoredEventsMessage, EventMessage, NoticeMessage +from .router import NostrRouter, nostr_client async def init_relays(): @@ -60,24 +55,12 @@ async def subscribe_events(): NostrRouter.received_subscription_eosenotices[sub_id] = event_message - def callback_command_results(result_message: CommandResultMessage): - event_id = result_message.event_id - # Only keep OKs some client is still waiting on; the rest would - # accumulate forever (events published by other relay users, or - # results arriving after the client already got its reply). - if not any(event_id in r.pending_publishes for r in all_routers): - return - NostrRouter.received_command_results.setdefault(event_id, []).append( - result_message - ) - def wrap_async_subscribe(): asyncio.run( nostr_client.subscribe( callback_events, callback_notices, callback_eose_notices, - callback_command_results, ) ) diff --git a/tests/test_router_ok.py b/tests/test_router_ok.py deleted file mode 100644 index e13301a..0000000 --- a/tests/test_router_ok.py +++ /dev/null @@ -1,167 +0,0 @@ -import json -import time - -import pytest - -from .. import router as router_module -from ..nostr.message_pool import CommandResultMessage -from ..router import PUBLISH_TIMEOUT_SECONDS, NostrRouter - - -class FakeWebSocket: - def __init__(self): - self.sent: list[list] = [] - - async def send_text(self, text: str): - self.sent.append(json.loads(text)) - - -class FakeRelay: - def __init__(self, connected: bool): - self.connected = connected - - -class FakeRelayManager: - def __init__(self, relays: dict[str, FakeRelay]): - self.relays = relays - self.published: list[str] = [] - self.subscriptions: list[str] = [] - - def publish_message(self, message: str): - self.published.append(message) - - def add_subscription(self, subscription_id: str, filters): - self.subscriptions.append(subscription_id) - - def close_subscriptions(self, subscriptions): - pass - - -EVENT_ID = "ab" * 32 -EVENT_MSG = json.dumps(["EVENT", {"id": EVENT_ID, "kind": 1, "content": "hi"}]) - - -def _router(monkeypatch, relays: dict[str, FakeRelay]): - manager = FakeRelayManager(relays) - monkeypatch.setattr(router_module.nostr_client, "relay_manager", manager) - NostrRouter.received_command_results.clear() - ws = FakeWebSocket() - return NostrRouter(ws), ws, manager # type: ignore[arg-type] - - -def _ok_from(url: str, accepted: bool, message: str = ""): - NostrRouter.received_command_results.setdefault(EVENT_ID, []).append( - CommandResultMessage(EVENT_ID, accepted, message, url) - ) - - -@pytest.mark.asyncio -async def test_no_connected_relays_replies_ok_false_immediately(monkeypatch): - router, ws, manager = _router(monkeypatch, {"wss://a": FakeRelay(False)}) - - await router._handle_client_to_nostr(EVENT_MSG) - - assert manager.published == [] - assert ws.sent == [["OK", EVENT_ID, False, "error: no relays connected"]] - assert router.pending_publishes == {} - - -@pytest.mark.asyncio -async def test_event_without_id_is_ignored(monkeypatch): - router, ws, manager = _router(monkeypatch, {"wss://a": FakeRelay(True)}) - - await router._handle_client_to_nostr(json.dumps(["EVENT", {"kind": 1}])) - - assert manager.published == [] - assert ws.sent == [] - - -@pytest.mark.asyncio -async def test_any_accepting_relay_yields_ok_true(monkeypatch): - relays = {"wss://a": FakeRelay(True), "wss://b": FakeRelay(True)} - router, ws, manager = _router(monkeypatch, relays) - - await router._handle_client_to_nostr(EVENT_MSG) - assert manager.published == [EVENT_MSG] - assert EVENT_ID in router.pending_publishes - - # nothing answered yet: no OK - await router._handle_command_results() - assert ws.sent == [] - - _ok_from("wss://a", False, "blocked: kind not allowed") - await router._handle_command_results() - assert ws.sent == [] # one rejection out of two relays: keep waiting - - _ok_from("wss://b", True, "") - await router._handle_command_results() - assert ws.sent == [["OK", EVENT_ID, True, ""]] - assert router.pending_publishes == {} - assert EVENT_ID not in NostrRouter.received_command_results - - # a late OK must not produce a second reply - _ok_from("wss://a", True, "") - await router._handle_command_results() - assert len(ws.sent) == 1 - - -@pytest.mark.asyncio -async def test_all_relays_rejecting_yields_ok_false(monkeypatch): - relays = {"wss://a": FakeRelay(True), "wss://b": FakeRelay(True)} - router, ws, _ = _router(monkeypatch, relays) - - await router._handle_client_to_nostr(EVENT_MSG) - _ok_from("wss://a", False, "invalid: bad sig") - _ok_from("wss://b", False, "blocked: kind not allowed") - await router._handle_command_results() - - assert ws.sent == [["OK", EVENT_ID, False, "blocked: kind not allowed"]] - assert router.pending_publishes == {} - - -@pytest.mark.asyncio -async def test_timeout_yields_ok_false(monkeypatch): - router, ws, _ = _router(monkeypatch, {"wss://a": FakeRelay(True)}) - - await router._handle_client_to_nostr(EVENT_MSG) - router.pending_publishes[EVENT_ID].sent_at = ( - time.time() - PUBLISH_TIMEOUT_SECONDS - 1 - ) - await router._handle_command_results() - - assert ws.sent == [["OK", EVENT_ID, False, "error: timed out waiting for relays"]] - assert router.pending_publishes == {} - - -@pytest.mark.asyncio -async def test_stop_drops_pending_publishes(monkeypatch): - router, _, _ = _router(monkeypatch, {"wss://a": FakeRelay(True)}) - - await router._handle_client_to_nostr(EVENT_MSG) - _ok_from("wss://a", True) - await router.stop() - - assert router.pending_publishes == {} - assert NostrRouter.received_command_results == {} - - -@pytest.mark.asyncio -async def test_req_without_configured_relays_is_closed(monkeypatch): - router, ws, manager = _router(monkeypatch, {}) - - await router._handle_client_to_nostr(json.dumps(["REQ", "sub1", {"kinds": [1]}])) - - assert ws.sent == [["CLOSED", "sub1", "error: no relay connections"]] - assert router.subscriptions == [] - assert manager.subscriptions == [] - - -@pytest.mark.asyncio -async def test_req_with_configured_but_disconnected_relay_is_kept(monkeypatch): - router, ws, manager = _router(monkeypatch, {"wss://a": FakeRelay(False)}) - - await router._handle_client_to_nostr(json.dumps(["REQ", "sub1", {"kinds": [1]}])) - - assert ws.sent == [] - assert len(router.subscriptions) == 1 - assert len(manager.subscriptions) == 1