restaurant/nostr/nostr_client.py

"""
Bidirectional Nostr client for the restaurant extension.

Connects to the nostrclient extension's internal WebSocket to publish
menu listings (NIP-99) and subscribe to incoming order DMs (eventually
NIP-17). Pattern lifted from the events extension's NostrClient, kept
local so the two extensions can diverge as needed.
"""

import asyncio
import json
from asyncio import Queue
from collections import OrderedDict
from typing import Optional

from loguru import logger
from websocket import WebSocketApp

from lnbits.helpers import encrypt_internal_message, urlsafe_short_hash
from lnbits.settings import settings

from .event import NostrEvent

MAX_SEEN_EVENTS = 1000


class NostrClient:
    def __init__(self):
        self.receive_event_queue: Queue = Queue()
        self.send_req_queue: Queue = Queue()
        self.ws: Optional[WebSocketApp] = None
        self.subscription_id = "restaurant-" + urlsafe_short_hash()[:32]
        self.running = False
        self._seen_events: OrderedDict[str, None] = OrderedDict()

    @property
    def is_websocket_connected(self) -> bool:
        if not self.ws:
            return False
        return self.ws.keep_running

    async def connect(self) -> WebSocketApp:
        relay_endpoint = encrypt_internal_message("relay", urlsafe=True)
        ws_url = (
            f"ws://localhost:{settings.port}"
            f"/nostrclient/api/v1/{relay_endpoint}"
        )

        logger.info("[RESTAURANT] Connecting to nostrclient WebSocket...")

        def on_open(_):
            logger.info("[RESTAURANT] Connected to nostrclient WebSocket")

        def on_message(_, message):
            try:
                self.receive_event_queue.put_nowait(message)
            except Exception as e:
                logger.error(f"[RESTAURANT] Failed to queue message: {e}")

        def on_error(_, error):
            logger.warning(f"[RESTAURANT] WebSocket error: {error}")

        def on_close(_, status_code, message):
            logger.warning(
                f"[RESTAURANT] WebSocket closed: {status_code} {message}"
            )
            self.receive_event_queue.put_nowait(ValueError("WebSocket closed"))

        ws = WebSocketApp(
            ws_url,
            on_message=on_message,
            on_open=on_open,
            on_close=on_close,
            on_error=on_error,
        )

        from threading import Thread

        wst = Thread(target=ws.run_forever)
        wst.daemon = True
        wst.start()
        return ws

    async def run_forever(self):
        self.running = True
        while self.running:
            try:
                if not self.is_websocket_connected:
                    self.ws = await self.connect()
                    await asyncio.sleep(5)

                req = await self.send_req_queue.get()
                assert self.ws
                self.ws.send(json.dumps(req))
            except Exception as ex:
                logger.warning(f"[RESTAURANT] NostrClient error: {ex}")
                await asyncio.sleep(60)

    def is_duplicate_event(self, event_id: str) -> bool:
        if event_id in self._seen_events:
            return True
        self._seen_events[event_id] = None
        if len(self._seen_events) > MAX_SEEN_EVENTS:
            self._seen_events.popitem(last=False)
        return False

    async def get_event(self):
        value = await self.receive_event_queue.get()
        if isinstance(value, ValueError):
            raise value
        return value

    async def publish_nostr_event(self, e: NostrEvent) -> None:
        await self.send_req_queue.put(["EVENT", e.dict()])

    async def subscribe(self, filters: list[dict]) -> None:
        self.subscription_id = "restaurant-" + urlsafe_short_hash()[:32]
        await self.send_req_queue.put(
            ["REQ", self.subscription_id] + filters
        )
        logger.info(
            f"[RESTAURANT] Subscribed (sub: {self.subscription_id[:20]}...)"
        )

    async def unsubscribe(self) -> None:
        await self.send_req_queue.put(["CLOSE", self.subscription_id])

    async def stop(self) -> None:
        await self.unsubscribe()
        self.running = False
        await asyncio.sleep(2)
        if self.ws:
            try:
                self.ws.close()
            except Exception:
                pass
            self.ws = None