diff --git a/backend/package/yuxi/channels/adapters/nostr/adapter.py b/backend/package/yuxi/channels/adapters/nostr/adapter.py index cc7645fb..030998a6 100644 --- a/backend/package/yuxi/channels/adapters/nostr/adapter.py +++ b/backend/package/yuxi/channels/adapters/nostr/adapter.py @@ -1,16 +1,31 @@ from __future__ import annotations import asyncio +import json as _json from dataclasses import dataclass from typing import Any +import aiohttp + +from yuxi.channels.adapters.nostr.config import NostrConfig +from yuxi.channels.adapters.nostr.crypto import NostrCrypto, NostrCryptoError, normalize_pubkey +from yuxi.channels.adapters.nostr.guard import GuardPolicy, NostrGuard +from yuxi.channels.adapters.nostr.health import RelayHealthTracker +from yuxi.channels.adapters.nostr.metrics import NostrMetrics +from yuxi.channels.adapters.nostr.monitor import NostrMonitor +from yuxi.channels.adapters.nostr.probe import probe_relay +from yuxi.channels.adapters.nostr.relay_manager import RelayManager +from yuxi.channels.adapters.nostr.send import NostrSender +from yuxi.channels.adapters.nostr.send_cache import SendCache +from yuxi.channels.adapters.nostr.state_store import NostrStateStore +from yuxi.channels.adapters.nostr.thread_simulator import NostrThreadSimulator from yuxi.channels.base import BaseChannelAdapter from yuxi.channels.capabilities import ChannelCapabilities -from yuxi.channels.meta import ChannelMeta from yuxi.channels.exceptions import ( ChannelAuthenticationError, ChannelException, ) +from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( Attachment, ChannelIdentity, @@ -25,24 +40,9 @@ from yuxi.channels.models import ( MessageType, ) from yuxi.channels.registry import register_builtin_adapter -from yuxi.channels.adapters.nostr.config import NostrConfig -from yuxi.channels.adapters.nostr.crypto import NostrCrypto, NostrCryptoError -from yuxi.channels.adapters.nostr.relay_manager import RelayManager -from yuxi.channels.adapters.nostr.send import NostrSender -from yuxi.channels.adapters.nostr.monitor import NostrMonitor -from yuxi.channels.adapters.nostr.probe import probe_relay -from yuxi.channels.adapters.nostr.guard import NostrGuard, GuardPolicy -from yuxi.channels.adapters.nostr.health import RelayHealthTracker -from yuxi.channels.adapters.nostr.crypto import normalize_pubkey -from yuxi.channels.adapters.nostr.state_store import NostrStateStore -from yuxi.channels.adapters.nostr.metrics import NostrMetrics -from yuxi.channels.adapters.nostr.send_cache import SendCache from yuxi.utils.datetime_utils import utc_now_naive from yuxi.utils.logging_config import logger -import aiohttp -import json as _json - @dataclass class _AccountComponents: @@ -114,8 +114,10 @@ class NostrAdapter(BaseChannelAdapter): self._order_buffer: list[dict] = [] self._order_lock = asyncio.Lock() self._ordering_task: asyncio.Task | None = None + self._score_sync_task: asyncio.Task | None = None self._account_managers: dict[str, _AccountComponents] = {} self._active_account_id: str = "default" + self._thread_simulator: NostrThreadSimulator | None = None async def connect(self) -> None: self._status = ChannelStatus.CONNECTING @@ -143,6 +145,8 @@ class NostrAdapter(BaseChannelAdapter): await self._relay_manager.connect_all() + self._thread_simulator = NostrThreadSimulator(relay_pool=self._relay_manager, max_depth=10) + for components in self._account_managers.values(): if components.relay_manager is not self._relay_manager: await components.relay_manager.connect_all() @@ -163,6 +167,8 @@ class NostrAdapter(BaseChannelAdapter): if self._nostr_config.message_ordering: self._ordering_task = asyncio.create_task(self._run_order_buffer()) + self._score_sync_task = asyncio.create_task(self._run_score_sync()) + self._status = ChannelStatus.CONNECTED account_count = 1 + len(self._account_managers) logger.info(f"[Nostr] Channel connected. npub: {self._crypto.npub}, 账户数: {account_count}") @@ -192,7 +198,7 @@ class NostrAdapter(BaseChannelAdapter): guard_policy = self._build_guard_policy(cfg) self._guard = NostrGuard(self._crypto.pubkey_hex(), guard_policy) - self._restore_state() + await self._restore_state() async def _init_additional_account(self, account_id: str, cfg: NostrConfig) -> None: try: @@ -247,20 +253,28 @@ class NostrAdapter(BaseChannelAdapter): allow_from=cfg.allow_from, ) - def _restore_state(self) -> None: - saved_events = self._state_store.load_seen_events() + async def _restore_state(self) -> None: + saved_events = await self.state_get("seen_events", namespace="dedup") + if saved_events is None: + saved_events = self._state_store.load_seen_events() for event_id in saved_events: self._guard._seen.mark_seen(event_id) - saved_inflight = self._state_store.load_inflight() + + saved_inflight = await self.state_get("inflight", namespace="dedup") + if saved_inflight is None: + saved_inflight = self._state_store.load_inflight() for event_id in saved_inflight: self._guard._inflight.add(event_id) + if saved_events or saved_inflight: logger.info( "[Nostr] 从 state_store 恢复状态: seen=%s, inflight=%s", len(saved_events), len(saved_inflight), ) - last_ts = self._state_store.load_last_processed_at() + last_ts = await self.state_get("last_processed_at", namespace="connection") + if last_ts is None: + last_ts = self._state_store.load_last_processed_at() if last_ts and self._monitor._last_subscribe_ts == 0: self._monitor._last_subscribe_ts = last_ts @@ -296,16 +310,42 @@ class NostrAdapter(BaseChannelAdapter): if failed: logger.warning(f"[Nostr] NIP-42 AUTH 最终失败 {len(failed)} 个 Relay: {failed}") + total_urls = len(auth_urls) + success_count = total_urls - len(failed) + self._metrics.increment("nip42_auth.success", success_count) + self._metrics.increment("nip42_auth.failure", len(failed)) + self._metrics.set_gauge("nip42_auth.success_rate", success_count / max(total_urls, 1)) + + if failed and len(failed) == total_urls: + logger.warning("[Nostr] 所有 Relay NIP-42 AUTH 失败，适配器进入 degraded 状态") + self._status = ChannelStatus.CONNECTED + for url in failed: + self._health_tracker.record_failure(url) + async def disconnect(self) -> None: if self._ordering_task: self._ordering_task.cancel() self._ordering_task = None + if self._score_sync_task: + self._score_sync_task.cancel() + self._score_sync_task = None + if self._state_store: - self._state_store.save_last_processed_at(self._monitor._last_subscribe_ts if self._monitor else 0) + last_ts = self._monitor._last_subscribe_ts if self._monitor else 0 + self._state_store.save_last_processed_at(last_ts) + await self.state_set("last_processed_at", last_ts, namespace="connection") + self._state_store.save_send_cache(self._send_cache.to_dict_list()) + send_cache_data = self._send_cache.to_dict_list() + await self.state_set("send_cache", send_cache_data, namespace="cache", ttl_seconds=3600) + if self._guard: + seen_events = list(self._guard._seen._store.keys()) self._state_store.save_inflight(list(self._guard.inflight)) + inflight_data = list(self._guard.inflight) + await self.state_set("seen_events", seen_events, namespace="dedup", ttl_seconds=600) + await self.state_set("inflight", inflight_data, namespace="dedup", ttl_seconds=600) tasks = [] if self._monitor: tasks.append(self._monitor.stop()) @@ -390,20 +430,18 @@ class NostrAdapter(BaseChannelAdapter): if not self._sender: return DeliveryResult(success=False, error="Sender 未初始化") - target_pubkey = msg_id - if ":" in msg_id: - parts = msg_id.split(":") - target_pubkey = parts[-1] - return await self._sender.send_reaction(chat_id, target_pubkey, emoji) + target_pubkey = "" + if ":" in chat_id: + target_pubkey = chat_id.split(":")[-1] + return await self._sender.send_reaction(msg_id, target_pubkey, emoji) async def remove_reaction(self, chat_id: str, msg_id: str) -> DeliveryResult: if not self._sender: return DeliveryResult(success=False, error="Sender 未初始化") - target_pubkey = msg_id - if ":" in msg_id: - parts = msg_id.split(":") - target_pubkey = parts[-1] + target_pubkey = "" + if ":" in chat_id: + target_pubkey = chat_id.split(":")[-1] return await self._sender.remove_reaction(msg_id, target_pubkey) async def send_chat_action(self, chat_id: str, action: str) -> DeliveryResult: @@ -424,43 +462,38 @@ class NostrAdapter(BaseChannelAdapter): if streaming_mode == "off": if finished: - identity = self._build_stream_identity(chat_id, msg_id) - response = ChannelResponse(identity=identity, content=chunk) - return await self.send(response) + return await self._flush_stream_final(chat_id, msg_id, chunk, receiver_pubkey) return DeliveryResult(success=True, message_id=None) if streaming_mode == "block": if finished: - if msg_id and self._sender: - return await self._sender.send_edit( - original_event_id=msg_id, - new_content=chunk, - receiver_pubkey=receiver_pubkey, - ) - identity = self._build_stream_identity(chat_id, msg_id) - response = ChannelResponse(identity=identity, content=chunk) - return await self.send(response) + return await self._flush_stream_final(chat_id, msg_id, chunk, receiver_pubkey) return DeliveryResult(success=True, message_id=None) - # progress 模式：每块都发送编辑 if finished: - if msg_id and self._sender: - return await self._sender.send_edit( - original_event_id=msg_id, - new_content=chunk, - receiver_pubkey=receiver_pubkey, - ) - identity = self._build_stream_identity(chat_id, msg_id) - response = ChannelResponse(identity=identity, content=chunk) - return await self.send(response) + return await self._flush_stream_final(chat_id, msg_id, chunk, receiver_pubkey) + return await self._flush_stream_edit_or_send(chat_id, msg_id, chunk, receiver_pubkey) + async def _flush_stream_final(self, chat_id: str, msg_id: str, chunk: str, receiver_pubkey: str) -> DeliveryResult: if msg_id and self._sender: return await self._sender.send_edit( original_event_id=msg_id, new_content=chunk, receiver_pubkey=receiver_pubkey, ) + identity = self._build_stream_identity(chat_id, msg_id) + response = ChannelResponse(identity=identity, content=chunk) + return await self.send(response) + async def _flush_stream_edit_or_send( + self, chat_id: str, msg_id: str, chunk: str, receiver_pubkey: str + ) -> DeliveryResult: + if msg_id and self._sender: + return await self._sender.send_edit( + original_event_id=msg_id, + new_content=chunk, + receiver_pubkey=receiver_pubkey, + ) identity = self._build_stream_identity(chat_id, msg_id) response = ChannelResponse(identity=identity, content=chunk) return await self.send(response) @@ -807,22 +840,26 @@ class NostrAdapter(BaseChannelAdapter): try: decrypted = await self._crypto.decrypt_nip17(raw_event.get("content", "")) raw_event = {**raw_event, "content": decrypted, "kind": 1059} + self._metrics.record_decrypt_success("nip17") if self._guard: plaintext_reject = self._guard.check_plaintext_size(decrypted) if plaintext_reject: logger.debug("NIP-17 plaintext rejected: %s", plaintext_reject) raw_event["content"] = f"(plaintext too large: {len(decrypted)} bytes)" except Exception as e: - logger.debug( + self._metrics.record_decrypt_failure("nip17") + logger.warning( "NIP-17 解密失败: event=%s, sender=%s, error=%s", event_id[:8], raw_event.get("pubkey", "?")[:8], e, ) + return if kind == 4 and self._crypto: try: content = self._crypto.decrypt_nip04(raw_event.get("content", ""), raw_event.get("pubkey", "")) + self._metrics.record_decrypt_success("nip04") if self._guard: plaintext_reject = self._guard.check_plaintext_size(content) if plaintext_reject: @@ -830,15 +867,28 @@ class NostrAdapter(BaseChannelAdapter): content = f"(plaintext too large: {len(content)} bytes)" raw_event["content"] = content except Exception as e: - logger.debug( + self._metrics.record_decrypt_failure("nip04") + logger.warning( "NIP-04 解密失败: event=%s, sender=%s, error=%s", event_id[:8], raw_event.get("pubkey", "?")[:8], e, ) + return self._metrics.record_event_accepted() message = self.normalize_inbound(raw_event) + + if self._thread_simulator and kind not in (5, 7): + try: + thread_chain = await self._thread_simulator.build_thread_from_event(raw_event) + if len(thread_chain) > 1: + parent_ids = [e.get("id", "") for e in thread_chain[:-1]] + message.metadata = message.metadata or {} + message.metadata["thread_parent_ids"] = parent_ids + except Exception: + pass + await self._handle_message(message) finally: if self._guard and event_id: @@ -882,6 +932,16 @@ class NostrAdapter(BaseChannelAdapter): self._pending_tasks = [t for t in self._pending_tasks if not t.done()] self._pending_tasks.append(task) + async def _run_score_sync(self) -> None: + interval = 30.0 + try: + while True: + await asyncio.sleep(interval) + if self._relay_manager: + self._relay_manager.sync_scores_from_health_tracker(self._health_tracker) + except asyncio.CancelledError: + pass + def _resolve_state_dir(self) -> str: from pathlib import Path diff --git a/backend/package/yuxi/channels/adapters/nostr/config.py b/backend/package/yuxi/channels/adapters/nostr/config.py index c9685612..1d0d64c9 100644 --- a/backend/package/yuxi/channels/adapters/nostr/config.py +++ b/backend/package/yuxi/channels/adapters/nostr/config.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os from dataclasses import dataclass, field from typing import Any @@ -53,6 +54,7 @@ class NostrConfig: message_ordering: bool = False, message_ordering_window_ms: int = 500, multi_account_enabled: bool = False, + require_tls: bool = True, ): self.private_key = private_key or "" self.relays = relays or [ @@ -80,6 +82,7 @@ class NostrConfig: self.message_ordering = message_ordering self.message_ordering_window_ms = message_ordering_window_ms self.multi_account_enabled = multi_account_enabled + self.require_tls = require_tls self._validate() def _validate(self): @@ -104,6 +107,10 @@ class NostrConfig: for url in self.relays: if not url.startswith(("ws://", "wss://")): raise NostrConfigError(f"Relay URL 必须以 ws:// 或 wss:// 开头: {url}") + if self.require_tls and url.startswith("ws://"): + raise NostrConfigError( + f"Relay URL 不允许明文 ws:// (require_tls=True): {url}。请使用 wss:// 或将 require_tls 设为 False。" + ) @classmethod def from_dict(cls, config: dict[str, Any] | None) -> NostrConfig: @@ -157,6 +164,88 @@ class NostrConfig: message_ordering=default_account.get("message_ordering", False), message_ordering_window_ms=default_account.get("message_ordering_window_ms", 500), multi_account_enabled=default_account.get("multi_account_enabled", False), + require_tls=default_account.get("require_tls", True), + ) + + @classmethod + def from_env(cls, prefix: str = "NOSTR_") -> NostrConfig: + def _env(key: str, default: Any = None) -> Any: + return os.getenv(prefix + key, default) + + def _env_bool(key: str, default: bool = False) -> bool: + val = os.getenv(prefix + key) + if val is None: + return default + return val.lower() in ("1", "true", "yes", "on") + + def _env_int(key: str, default: int = 0) -> int: + val = os.getenv(prefix + key) + if val is None: + return default + try: + return int(val) + except ValueError: + return default + + def _env_float(key: str, default: float = 0.0) -> float: + val = os.getenv(prefix + key) + if val is None: + return default + try: + return float(val) + except ValueError: + return default + + def _env_list(key: str, default: list[str] | None = None) -> list[str] | None: + val = os.getenv(prefix + key) + if val is None: + return default + return [item.strip() for item in val.split(",") if item.strip()] + + relays = _env_list("RELAYS") or [ + "wss://relay.damus.io", + "wss://relay.primal.net", + "wss://relay.nostr.info", + "wss://nos.lol", + ] + + guard_policy = GuardPolicyConfig( + allowed_kinds=_env_list("GUARD_ALLOWED_KINDS") or [1, 4, 5, 7, 1059], + max_ciphertext_bytes=_env_int("GUARD_MAX_CIPHERTEXT_BYTES", 50_000), + max_plaintext_bytes=_env_int("GUARD_MAX_PLAINTEXT_BYTES", 10_000), + max_future_skew_sec=_env_int("GUARD_MAX_FUTURE_SKEW_SEC", 30), + rate_limit=RateLimitConfig( + window_ms=_env_int("RATE_LIMIT_WINDOW_MS", 10_000), + max_per_sender_per_window=_env_int("RATE_LIMIT_MAX_PER_SENDER", 20), + max_global_per_window=_env_int("RATE_LIMIT_MAX_GLOBAL", 200), + ), + ) + + allow_from = _env_list("ALLOW_FROM") + + return cls( + private_key=_env("PRIVATE_KEY", ""), + relays=relays, + dm_policy=_env("DM_POLICY", "pairing"), + nip17_enabled=_env_bool("NIP17_ENABLED", True), + streaming_mode=_env("STREAMING_MODE", "block"), + reconnect_interval_sec=_env_int("RECONNECT_INTERVAL_SEC", 5), + relay_timeout_sec=_env_int("RELAY_TIMEOUT_SEC", 30), + relay_degraded_threshold=_env_float("RELAY_DEGRADED_THRESHOLD", 0.5), + nip42_auth_enabled=_env_bool("NIP42_AUTH_ENABLED", False), + nip42_auth_urls=_env_list("NIP42_AUTH_URLS"), + guard_policy=guard_policy, + backfill_window_sec=_env_int("BACKFILL_WINDOW_SEC", 120), + allow_from=allow_from, + markdown_table_mode=_env("MARKDOWN_TABLE_MODE", "off"), + profile=None, + enabled=_env_bool("ENABLED", True), + name=_env("NAME", ""), + send_message_cache_size=_env_int("SEND_CACHE_SIZE", 100), + message_ordering=_env_bool("MESSAGE_ORDERING", False), + message_ordering_window_ms=_env_int("MESSAGE_ORDERING_WINDOW_MS", 500), + multi_account_enabled=_env_bool("MULTI_ACCOUNT_ENABLED", False), + require_tls=_env_bool("REQUIRE_TLS", True), ) def get_account_configs(self, raw_config: dict[str, Any] | None = None) -> dict[str, NostrConfig]: diff --git a/backend/package/yuxi/channels/adapters/nostr/guard.py b/backend/package/yuxi/channels/adapters/nostr/guard.py index e17bf754..01b078be 100644 --- a/backend/package/yuxi/channels/adapters/nostr/guard.py +++ b/backend/package/yuxi/channels/adapters/nostr/guard.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json as _json import time from collections import OrderedDict, defaultdict from dataclasses import dataclass, field @@ -7,8 +8,6 @@ from dataclasses import dataclass, field from yuxi.channels.adapters.nostr.crypto import normalize_pubkey from yuxi.utils.logging_config import logger -import json as _json - @dataclass class AuditRecord: diff --git a/backend/package/yuxi/channels/adapters/nostr/monitor.py b/backend/package/yuxi/channels/adapters/nostr/monitor.py index fa736412..57872444 100644 --- a/backend/package/yuxi/channels/adapters/nostr/monitor.py +++ b/backend/package/yuxi/channels/adapters/nostr/monitor.py @@ -55,7 +55,7 @@ class NostrMonitor: self._last_subscribe_ts = now - self._backfill_window_sec since = self._last_subscribe_ts self._last_subscribe_ts = now - filters = [{"kinds": [1, 4, 5, 7, 1059], "since": since}] + filters = [{"kinds": [1, 4, 5, 7, 1059], "since": since, "until": now + 60}] self._relay_manager.set_pubkey_filter([self._crypto.pubkey_hex()]) await self._relay_manager.subscribe(filters) self._subscription_active = True diff --git a/backend/package/yuxi/channels/adapters/nostr/profile.py b/backend/package/yuxi/channels/adapters/nostr/profile.py index 41bdc6d3..510dd438 100644 --- a/backend/package/yuxi/channels/adapters/nostr/profile.py +++ b/backend/package/yuxi/channels/adapters/nostr/profile.py @@ -2,17 +2,16 @@ from __future__ import annotations import asyncio import ipaddress +import json as _json import time from urllib.parse import urlparse from yuxi.channels.adapters.nostr.crypto import NostrCrypto -from yuxi.channels.adapters.nostr.relay_manager import RelayManager from yuxi.channels.adapters.nostr.models import NostrProfile +from yuxi.channels.adapters.nostr.relay_manager import RelayManager from yuxi.channels.adapters.nostr.state_store import NostrStateStore from yuxi.utils.logging_config import logger -import json as _json - def _validate_profile_url(url: str, field_name: str = "url") -> str | None: if not url: diff --git a/backend/package/yuxi/channels/adapters/nostr/profile_api.py b/backend/package/yuxi/channels/adapters/nostr/profile_api.py index d0a7deed..82436dc2 100644 --- a/backend/package/yuxi/channels/adapters/nostr/profile_api.py +++ b/backend/package/yuxi/channels/adapters/nostr/profile_api.py @@ -1,6 +1,5 @@ from __future__ import annotations - from fastapi import APIRouter, HTTPException from yuxi.channels.adapters.nostr.models import NostrProfile diff --git a/backend/package/yuxi/channels/adapters/nostr/relay_manager.py b/backend/package/yuxi/channels/adapters/nostr/relay_manager.py index 53b17c3c..abf1a878 100644 --- a/backend/package/yuxi/channels/adapters/nostr/relay_manager.py +++ b/backend/package/yuxi/channels/adapters/nostr/relay_manager.py @@ -296,9 +296,19 @@ class RelayManager: pass self._schedule_reconnect(url) - async def query(self, filters: list[dict], timeout: float = 10.0) -> list[dict]: + async def query( + self, filters: list[dict], timeout: float = 10.0, since: int | None = None, until: int | None = None + ) -> list[dict]: sub_id = "forcepilot_nostr_query" - req = json.dumps(["REQ", sub_id, *filters]) + effective_filters: list[dict] = [] + for f in filters: + f_copy = dict(f) + if since is not None: + f_copy["since"] = since + if until is not None: + f_copy["until"] = until + effective_filters.append(f_copy) + req = json.dumps(["REQ", sub_id, *effective_filters]) close_req = json.dumps(["CLOSE", sub_id]) active_ws = [(url, ws) for url, ws in self._connections.items() if ws and ws.open] @@ -329,6 +339,11 @@ class RelayManager: pass except Exception: pass + finally: + try: + await _ws.send(close_req) + except Exception: + pass return results receive_tasks = [_receive_from_relay(url, ws) for url, ws in active_ws] @@ -343,12 +358,6 @@ class RelayManager: seen.add(event_id) events.append(event) - for url, ws in active_ws: - try: - await ws.send(close_req) - except Exception: - pass - return events async def send_auth(self, raw_signed_auth_event: dict, urls: list[str] | None = None) -> dict[str, bool]: @@ -378,5 +387,15 @@ class RelayManager: def set_relay_scores(self, scores: dict[str, float]) -> None: self._relay_scores.update(scores) + def sync_scores_from_health_tracker(self, health_tracker, urls: list[str] | None = None) -> None: + target_urls = urls or list(self._relay_urls) + for url in target_urls: + score = health_tracker.get_score(url) + self._relay_scores[url] = score + def get_relay_score(self, url: str) -> float: return self._relay_scores.get(url, 0.5) + + async def query_event(self, event_id: str, timeout: float = 10.0) -> dict | None: + events = await self.query([{"ids": [event_id], "limit": 1}], timeout=timeout) + return events[0] if events else None diff --git a/backend/package/yuxi/channels/adapters/nostr/send.py b/backend/package/yuxi/channels/adapters/nostr/send.py index 893e757a..87cff852 100644 --- a/backend/package/yuxi/channels/adapters/nostr/send.py +++ b/backend/package/yuxi/channels/adapters/nostr/send.py @@ -1,13 +1,14 @@ from __future__ import annotations +import asyncio import json as _json import re import time -from yuxi.channels.models import ChannelResponse, DeliveryResult -from yuxi.channels.adapters.nostr.crypto import NostrCrypto, NostrCryptoError from yuxi.channels.adapters.nostr.config import NostrConfig +from yuxi.channels.adapters.nostr.crypto import NostrCrypto, NostrCryptoError from yuxi.channels.adapters.nostr.relay_manager import RelayManager +from yuxi.channels.models import ChannelResponse, DeliveryResult from yuxi.utils.logging_config import logger @@ -22,8 +23,9 @@ class NostrSender: async def send(self, response: ChannelResponse) -> DeliveryResult: try: chunks = self._chunk_text(response.content) + total = len(chunks) results = [] - for chunk in chunks: + for i, chunk in enumerate(chunks): chunk_response = ChannelResponse( identity=response.identity, content=chunk, @@ -31,8 +33,14 @@ class NostrSender: reply_to_message_id=response.reply_to_message_id, ) event = await self.format_outbound(chunk_response) + if total > 1: + tags = event.get("tags", []) + tags.append(["chunk", f"{i + 1}/{total}"]) + event["tags"] = tags success_count = await self._relay_manager.broadcast(event) results.append(success_count > 0) + if total > 1 and i < total - 1: + await asyncio.sleep(0.2) all_success = all(results) event = await self.format_outbound(response) event_id = event.get("id", "") diff --git a/backend/package/yuxi/channels/adapters/nostr/setup.py b/backend/package/yuxi/channels/adapters/nostr/setup.py index 993d4c01..550a7bd9 100644 --- a/backend/package/yuxi/channels/adapters/nostr/setup.py +++ b/backend/package/yuxi/channels/adapters/nostr/setup.py @@ -1,9 +1,10 @@ from __future__ import annotations +import asyncio + from yuxi.channels.adapters.nostr.crypto import NostrCrypto, NostrCryptoError from yuxi.channels.adapters.nostr.probe import probe_relay from yuxi.utils.logging_config import logger -import asyncio class NostrSetupAdapter: diff --git a/backend/package/yuxi/channels/adapters/nostr/setup_plugin.py b/backend/package/yuxi/channels/adapters/nostr/setup_plugin.py index 35a584c3..dfb242dc 100644 --- a/backend/package/yuxi/channels/adapters/nostr/setup_plugin.py +++ b/backend/package/yuxi/channels/adapters/nostr/setup_plugin.py @@ -2,9 +2,9 @@ from __future__ import annotations from typing import Any -from yuxi.channels.adapters.nostr.setup import NostrSetupAdapter -from yuxi.channels.adapters.nostr.crypto import NostrCrypto, NostrCryptoError from yuxi.channels.adapters.nostr.config import NostrConfig +from yuxi.channels.adapters.nostr.crypto import NostrCrypto, NostrCryptoError +from yuxi.channels.adapters.nostr.setup import NostrSetupAdapter class NostrSetupPlugin: