Padreug
24acbe6674
- nostr/ vendors NostrEvent + the nostrclient WebSocket bridge from the events extension, retagged [TASKS] / subscription-id "tasks-*". - nostr_publisher builds kind 31922 with the `event-type: task` tag (per aiolabs/webapp#25 — disambiguates from kind-31922 activities on shared relays), kind 31925 with task-status / occurrence / completed_at, and kind 5 deletions for both. - nostr_hooks bridges task/completion mutations to the publisher and persists the resulting nostr_event_id back onto the local row. - nostr_sync subscribes to {31922, 31925, 5/#k} and filters 31922 client-side on `event-type: task` because most relays don't index custom single-letter tags. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""
|
|
Bidirectional Nostr client for the tasks extension.
|
|
|
|
Connects to the nostrclient extension's internal WebSocket to publish
|
|
and subscribe to NIP-52 calendar events (kind 31922) and the
|
|
task-status RSVP variant (kind 31925). Mirrors events/nostr/nostr_client.py
|
|
with a TASKS log prefix and subscription-id namespace.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
from asyncio import Queue
|
|
from collections import OrderedDict
|
|
|
|
from lnbits.helpers import encrypt_internal_message, urlsafe_short_hash
|
|
from lnbits.settings import settings
|
|
from loguru import logger
|
|
from websocket import WebSocketApp
|
|
|
|
from .event import NostrEvent
|
|
|
|
MAX_SEEN_EVENTS = 500
|
|
|
|
|
|
class NostrClient:
|
|
def __init__(self):
|
|
self.receive_event_queue: Queue = Queue()
|
|
self.send_req_queue: Queue = Queue()
|
|
self.ws: WebSocketApp | None = None
|
|
self.subscription_id = "tasks-" + urlsafe_short_hash()[:32]
|
|
self.running = False
|
|
self._seen_events: OrderedDict[str, None] = OrderedDict()
|
|
|
|
@property
|
|
def is_websocket_connected(self):
|
|
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("[TASKS] Connecting to nostrclient WebSocket...")
|
|
|
|
def on_open(_):
|
|
logger.info("[TASKS] Connected to nostrclient WebSocket")
|
|
|
|
def on_message(_, message):
|
|
try:
|
|
self.receive_event_queue.put_nowait(message)
|
|
except Exception as e:
|
|
logger.error(f"[TASKS] Failed to queue message: {e}")
|
|
|
|
def on_error(_, error):
|
|
logger.warning(f"[TASKS] WebSocket error: {error}")
|
|
|
|
def on_close(_, status_code, message):
|
|
logger.warning(f"[TASKS] 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"[TASKS] 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):
|
|
await self.send_req_queue.put(["EVENT", e.dict()])
|
|
|
|
async def subscribe(self, filters: list[dict]):
|
|
self.subscription_id = "tasks-" + urlsafe_short_hash()[:32]
|
|
await self.send_req_queue.put(["REQ", self.subscription_id, *filters])
|
|
logger.info(
|
|
f"[TASKS] Subscribed (sub: {self.subscription_id[:20]}...)"
|
|
)
|
|
|
|
async def unsubscribe(self):
|
|
await self.send_req_queue.put(["CLOSE", self.subscription_id])
|
|
|
|
async def stop(self):
|
|
await self.unsubscribe()
|
|
self.running = False
|
|
await asyncio.sleep(2)
|
|
if self.ws:
|
|
try:
|
|
self.ws.close()
|
|
except Exception:
|
|
pass
|
|
self.ws = None
|