buzz/mobile/lib/shared/relay/relay_session.dart

import 'dart:async';
import 'dart:convert';
import 'dart:math';

import 'package:http/http.dart' as http;
import 'package:nostr/nostr.dart' as nostr;
import 'package:pointycastle/digests/sha256.dart';
import 'package:uuid/uuid.dart';

import 'package:flutter/foundation.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';

import '../../features/age_gate/age_signal_provider.dart';
import '../auth/auth.dart';
import '../hosted/hosted_relay_auth.dart';
import '../hosted/builderlab_account.dart';
import 'nostr_filters.dart';
import 'nostr_models.dart';
import 'relay_client.dart';
import 'relay_closed_policy.dart';
import 'relay_http_query_client.dart';
import 'relay_provider.dart';
import 'relay_query_response.dart';
import 'relay_rate_limit_gate.dart';
import 'relay_reconnect_policy.dart';
import 'relay_session_types.dart';
import 'relay_socket.dart';
import 'package:buzz/shared/crypto/event_signing.dart';

export 'relay_session_types.dart';

part 'relay_session_auth.dart';
part 'relay_session_subscriptions.dart';

class RelaySessionNotifier extends Notifier<SessionState> {
  RelaySessionNotifier({
    http.Client? httpClient,
    http.Client Function()? httpClientFactory,
    RelaySocketFactory socketFactory = RelaySocket.new,
    DateTime Function()? now,
    RelayRateLimitGate? rateLimitGate,
    RelayTimerFactory retryTimerFactory = Timer.new,
    Future<void> Function(Duration) replayDelay = Future.delayed,
    RelayTimerFactory reconnectTimerFactory = Timer.new,
    Random? random,
    RelayQueryDecoder queryDecoder = decodeRelayQueryResponse,
  }) : _httpQueryClient = RelayHttpQueryClient(
         client: httpClient,
         clientFactory: httpClientFactory,
       ),
       _socketFactory = socketFactory,
       _now = now ?? DateTime.now,
       _rateLimitGate = rateLimitGate ?? RelayRateLimitGate(),
       _retryTimerFactory = retryTimerFactory,
       _replayDelay = replayDelay,
       _reconnectTimerFactory = reconnectTimerFactory,
       _random = random ?? Random(),
       _queryDecoder = queryDecoder;

  final RelayHttpQueryClient _httpQueryClient;
  final RelaySocketFactory _socketFactory;
  final DateTime Function() _now;
  final RelayRateLimitGate _rateLimitGate;
  final RelayTimerFactory _retryTimerFactory;
  final Future<void> Function(Duration) _replayDelay;
  final RelayTimerFactory _reconnectTimerFactory;
  final Random _random;
  final RelayQueryDecoder _queryDecoder;

  static const _baseReconnectDelayMs = relayReconnectBaseDelayMs;
  static const _maxReconnectDelayMs = relayReconnectMaxDelayMs;
  static const _eventBatchMs = 16;
  static const _replayBatchSize = 8;
  static const _replayInterBatchDelay = Duration(milliseconds: 50);
  static const _maxRecentDeliveryKeys = 5000;
  static const _backgroundGraceDuration = Duration(seconds: 5);

  RelaySocket? _socket;
  final Map<String, _HistorySubscription> _historySubscriptions = {};
  final Map<String, _LiveSubscription> _liveSubscriptions = {};
  final Map<String, _ClosedRetry> _pendingClosedRetries = {};
  final Map<String, _PendingEvent> _pendingEvents = {};
  final List<_BufferedEvent> _eventBuffer = [];
  final Set<String> _recentDeliveryKeys = {};
  Timer? _reconnectTimer;
  Timer? _flushTimer;
  Timer? _backgroundGraceTimer;
  DateTime? _backgroundedAt;
  DateTime? _connectedAt;
  DateTime? _lastInboundAt;
  int _reconnectDelayMs = _baseReconnectDelayMs;
  int _subIdCounter = 0;
  bool _disposed = false;
  bool _ageRestricted = false;
  bool _paused = false;
  bool _hasConnectedOnce = false;
  int _connectionGeneration = 0;
  int _contextGeneration = 0;
  final Map<Object, String> _visibleChannelsByOwner = {};
  final Map<Object, Future<void> Function()> _beforePauseCallbacks = {};
  bool _socketConnected = false;
  bool _closedRetryReplayScheduled = false;
  String? _relayUrl;

  /// The relay origin this session serves. Switching community rebuilds the
  /// session in place, so work begun for another community compares this
  /// before reading from or publishing through it.
  String? get relayUrl => _relayUrl;

  @visibleForTesting
  set debugRelayUrl(String? value) => _relayUrl = value;

  @override
  SessionState build() {
    final config = ref.watch(relayConfigProvider);
    _relayUrl = config.baseUrl;
    final authState = ref.watch(authProvider);
    _ageRestricted = ref.watch(ageSignalProvider) == AgeSignalState.restricted;

    // Reset disposed flag — build() may re-run on the same Notifier instance
    // after a provider dependency changes (e.g. auth completing).
    _disposed = false;

    ref.onDispose(_dispose);

    // Auto-connect when authenticated and we have a signing key (NIP-42 AUTH).
    final isAuthenticated = authState.value?.status == AuthStatus.authenticated;
    if (!_ageRestricted && isAuthenticated && config.nsec != null) {
      // Schedule connection after build completes.
      Future.microtask(() => _connect(config));
    }

    return const SessionState(status: SessionStatus.disconnected);
  }

  /// Execute a one-shot query via the relay's HTTP bridge (`POST /query`).
  Future<List<NostrEvent>> queryRelay(
    List<NostrFilter> filters, {
    Duration timeout = const Duration(seconds: 8),
  }) => _postForEvents(
    '/query',
    filters.map((filter) => filter.toJson()).toList(),
    timeout: timeout,
  );

  /// Finds people by profile name, for typeahead surfaces.
  ///
  /// NIP-50 search over kind:0 matches the whole profile JSON, so text in
  /// avatar data URLs, `about` or `nip05` can fill every result page and hide
  /// a real name match. The relay's `POST /profiles/search` applies the same
  /// prefix rule to just `name` and `display_name` and puts exact and prefix
  /// name matches first.
  ///
  /// A relay without that endpoint answers 404 or 405. The search then uses
  /// NIP-50 ([NostrFilters.searchUsers]), and so do later searches on that
  /// relay for [profileSearchRetry]; after that the endpoint is tried again.
  /// A query or limit the endpoint does not accept also uses NIP-50.
  Future<List<NostrEvent>> searchProfiles(
    String query, {
    int limit = 50,
    Duration timeout = const Duration(seconds: 8),
  }) async {
    final text = query.trim();
    final relay = ref.read(relayConfigProvider).baseUrl;
    final fallback = _profileSearchFallback;
    if (text.isNotEmpty &&
        text.runes.length <= _profileSearchQueryMax &&
        limit >= 1 &&
        limit <= _profileSearchLimitMax &&
        (fallback == null ||
            fallback.relay != relay ||
            !_now().isBefore(fallback.until))) {
      try {
        final events = await _postForEvents('/profiles/search', {
          'query': text,
          'limit': limit,
        }, timeout: timeout);
        return [
          for (final event in events)
            if (event.kind == 0) event,
        ];
      } on RelayException catch (error) {
        if (error.statusCode != 404 && error.statusCode != 405) rethrow;
        _profileSearchFallback = (
          relay: relay,
          until: _now().add(profileSearchRetry),
        );
      }
    }
    return queryRelay([
      NostrFilters.searchUsers(text, limit: limit),
    ], timeout: timeout);
  }

  /// How long a relay's 404 or 405 keeps [searchProfiles] on NIP-50.
  static const profileSearchRetry = Duration(minutes: 5);
  static const _profileSearchQueryMax = 100;
  static const _profileSearchLimitMax = 100;
  ({String relay, DateTime until})? _profileSearchFallback;

  /// POSTs [payload] as JSON to relay [path] with NIP-98 auth and decodes
  /// the event array it answers.
  Future<List<NostrEvent>> _postForEvents(
    String path,
    Object payload, {
    required Duration timeout,
  }) async {
    if (_disposed || _ageRestricted) {
      throw StateError('Relay session is unavailable');
    }
    final generation = _contextGeneration;
    final config = ref.read(relayConfigProvider);
    final url = Uri.parse(config.baseUrl).resolve(path).toString();
    final bodyBytes = utf8.encode(jsonEncode(payload));
    // Reuse the session transport on success. A timeout rotates immediately
    // for new queries, then closes the retired client after its peers finish.
    final response = await _httpQueryClient.post(
      Uri.parse(url),
      headers: {
        'Authorization': buildNip98AuthHeader(
          method: 'POST',
          url: url,
          bodyBytes: bodyBytes,
          nsec: config.nsec,
          pubkey: ref.read(myPubkeyProvider),
        ),
        'Content-Type': 'application/json',
      },
      body: bodyBytes,
      timeout: timeout,
      decorate: (client, active) => HostedRelayHttpClient(
        ref.read(hostedRelayAuthProvider),
        config,
        inner: client,
        canSend: () =>
            active() && !_disposed && generation == _contextGeneration,
      ),
    );
    bool retired() =>
        _disposed || _ageRestricted || generation != _contextGeneration;
    if (retired()) {
      throw StateError('Relay query belongs to a retired session');
    }
    if (response.statusCode < 200 || response.statusCode >= 300) {
      final body = relayResponseText(
        response.bodyBytes,
        response.headers['content-type'],
      );
      _activateRateLimitGateFromHttpError(body);
      throw RelayException(response.statusCode, body);
    }
    // Decoding runs off the UI isolate, so the session can retire meanwhile.
    // A retired session reports that, not whatever its bytes decoded to.
    final List<NostrEvent> events;
    try {
      events = await _queryDecoder(response);
    } catch (_) {
      if (retired()) {
        throw StateError('Relay query belongs to a retired session');
      }
      rethrow;
    }
    if (retired()) {
      throw StateError('Relay query belongs to a retired session');
    }
    return events;
  }

  void _activateRateLimitGateFromHttpError(String body) {
    final dynamic decoded;
    try {
      decoded = jsonDecode(body);
    } on FormatException {
      return;
    }
    if (decoded is! Map<String, dynamic>) return;
    final message = decoded['error'];
    if (message is! String ||
        classifyRelayClosed(message) != RelayClosedClass.rateLimited) {
      return;
    }
    _rateLimitGate.activate(parseRateLimitRetrySeconds(message));
  }

  /// Fetch historical events matching [filter]. Sends REQ, collects events
  /// until EOSE, then resolves. One-shot subscription.
  Future<List<NostrEvent>> fetchHistory(
    NostrFilter filter, {
    Duration timeout = const Duration(seconds: 8),
  }) async {
    if (_rateLimitGate.isActive) await _rateLimitGate.wait();
    if (_disposed) throw StateError('Relay session is disposed');
    final subId = _nextSubId('h');
    final completer = Completer<List<NostrEvent>>();

    final timer = Timer(timeout, () {
      final sub = _historySubscriptions.remove(subId);
      if (sub != null && !sub.completer.isCompleted) {
        sub.completer.completeError(
          TimeoutException('Relay history request timed out after $timeout'),
        );
      }
      if (sub?.heldFilter == null) _sendClose(subId);
    });

    final subscription = _HistorySubscription(
      completer: completer,
      timeout: timer,
    );
    _historySubscriptions[subId] = subscription;

    // The relay rejects a REQ sent before NIP-42 AUTH completes. Hold it for
    // the authenticated connection instead of paying a rejection and retry.
    if (_socketConnected) {
      _sendReq(subId, filter);
    } else {
      subscription.heldFilter = filter;
    }
    return completer.future;
  }

  Future<void Function()> subscribe(
    NostrFilter filter,
    void Function(NostrEvent) onEvent, {
    void Function(String message)? onClosed,
  }) => _subscribe(filter, onEvent, onClosed: onClosed);

  /// Subscribe to a live stream and observe its recovery lifecycle.
  ///
  /// The returned future completes after initial EOSE (or the existing
  /// fallback timeout) and yields a cleanup callback. [onStatusChanged] emits
  /// [RelaySubscriptionStatus.ready] at each EOSE after buffered replay events
  /// have been delivered, and [RelaySubscriptionStatus.retrying] immediately
  /// when a retryable or rate-limited CLOSED begins backoff. Terminal CLOSED
  /// invokes [onClosed] and removes the subscription instead of retrying it.
  /// Calling the cleanup callback cancels pending retries and sends CLOSE.
  Future<void Function()> subscribeWithStatus(
    NostrFilter filter,
    void Function(NostrEvent) onEvent, {
    void Function(String message)? onClosed,
    required void Function(RelaySubscriptionStatus status) onStatusChanged,
  }) => _subscribe(
    filter,
    onEvent,
    onClosed: onClosed,
    onStatusChanged: onStatusChanged,
  );

  Future<void Function()> _subscribe(
    NostrFilter filter,
    void Function(NostrEvent) onEvent, {
    void Function(String message)? onClosed,
    void Function(RelaySubscriptionStatus status)? onStatusChanged,
  }) async {
    if (_disposed) throw StateError('Relay session is disposed');
    final subId = _nextSubId('l');
    final readyCompleter = Completer<void>();

    final subscription = _LiveSubscription(
      filter: filter,
      onEvent: onEvent,
      onClosed: onClosed,
      onStatusChanged: onStatusChanged,
      readyCompleter: readyCompleter,
    );
    _liveSubscriptions[subId] = subscription;

    // Before AUTH the relay closes it as auth-required, which is terminal.
    // The replay on connect sends it instead.
    if (_socketConnected) _sendLiveReq(subId, subscription);

    try {
      await readyCompleter.future.timeout(
        const Duration(milliseconds: 500),
        onTimeout: () {},
      );
    } catch (_) {
      _liveSubscriptions.remove(subId);
      _recentDeliveryKeys.removeWhere((key) => key.startsWith('$subId:'));
      rethrow;
    }
    final liveSub = _liveSubscriptions[subId];
    if (liveSub != null && liveSub.readyCompleter == readyCompleter) {
      liveSub.readyCompleter = null;
    }

    return () => _unsubscribe(subId);
  }

  Future<NostrEvent> publish(
    NostrEvent event, {
    Duration timeout = const Duration(seconds: 8),
  }) async {
    final generation = _connectionGeneration;
    if (_rateLimitGate.isActive) await _rateLimitGate.wait();
    if (!_isActiveConnection(generation) || !_socketConnected) {
      throw StateError('Relay session is not connected');
    }

    final completer = Completer<NostrEvent>();

    final timer = Timer(timeout, () {
      final pending = _pendingEvents.remove(event.id);
      if (pending != null && !pending.completer.isCompleted) {
        pending.completer.completeError(
          TimeoutException(
            'Event ${event.id} not acknowledged within $timeout',
          ),
        );
      }
    });

    _pendingEvents[event.id] = _PendingEvent(
      completer: completer,
      timeout: timer,
    );

    _socket?.send(['EVENT', event.toJson()]);
    return completer.future;
  }

  void sendRaw(List<dynamic> payload) {
    _socket?.send(payload);
  }

  @visibleForTesting
  void debugHandleMessage(List<dynamic> data) => _handleMessage(data);

  @visibleForTesting
  void debugFlushEventBuffer() => _flushEventBuffer();

  @visibleForTesting
  Future<void> debugHandleConnected() =>
      _handleConnected(_connectionGeneration);

  @visibleForTesting
  Future<void> debugReplayLiveSubscriptions() =>
      _replayLiveSubscriptions(_connectionGeneration);

  @visibleForTesting
  void debugDispose() => _dispose();

  @visibleForTesting
  void debugSupersedeConnection() => _connectionGeneration++;

  @visibleForTesting
  void debugHandleDisconnected([Object? error]) {
    _socketConnected = false;
    _handleDisconnected(_connectionGeneration, error);
  }

  @visibleForTesting
  void debugResetClosedRetriesForDisconnect() {
    _socketConnected = false;
    _resetAllClosedRetries();
  }

  @visibleForTesting
  void debugSetSessionStatus(SessionStatus status) {
    _socketConnected = status == SessionStatus.connected;
  }

  @visibleForTesting
  void debugPauseNow() => _pauseNow();

  @visibleForTesting
  void debugHandleSocketMessageForTest(List<dynamic> data) =>
      _handleMessage(data);

  @visibleForTesting
  void debugAttachSocketForTest(RelaySocket socket) {
    _socket?.dispose();
    _socket = socket;
    _socketConnected = true;
  }

  /// Registers a visible channel and returns an owner-scoped release callback.
  /// The most recently registered owner is prioritized during reconnect replay.
  void Function() registerVisibleChannel(String channelId) {
    final owner = Object();
    _visibleChannelsByOwner[owner] = channelId;
    return () => _visibleChannelsByOwner.remove(owner);
  }

  /// Force a reconnect (e.g., returning from background).
  Future<void> reconnect() async {
    _socketConnected = false;
    await _socket?.disconnect();
    _reconnectDelayMs = _baseReconnectDelayMs;
    final config = ref.read(relayConfigProvider);
    await _connect(config);
  }

  /// Registers work that must settle before the background grace disconnect.
  void Function() registerBeforePause(Future<void> Function() callback) {
    final owner = Object();
    _beforePauseCallbacks[owner] = callback;
    return () => _beforePauseCallbacks.remove(owner);
  }

  /// Called by the app lifecycle provider when the app goes to background.
  void onAppPaused() {
    _backgroundedAt = _now();
    _backgroundGraceTimer?.cancel();
    _backgroundGraceTimer = Timer(_backgroundGraceDuration, () {
      unawaited(_pauseAfterCallbacks());
    });
  }

  Future<void> _pauseAfterCallbacks() async {
    final callbacks = _beforePauseCallbacks.values.toList();
    try {
      await Future.wait(callbacks.map((callback) => callback()));
    } catch (error) {
      debugPrint('Background cleanup failed: $error');
    }
    if (_backgroundedAt != null) _pauseNow();
  }

  void _pauseNow() {
    _paused = true;
    _socketConnected = false;
    _reconnectTimer?.cancel();
    _cancelAllHistory(Exception('App moved to background'));
    _rejectAllPending(Exception('App moved to background'));
    _socket?.disconnect();
    state = const SessionState(status: SessionStatus.disconnected);
  }

  /// Called by the app lifecycle provider when the app returns to foreground.
  void onAppResumed() {
    _paused = false;
    final backgroundedAt = _backgroundedAt;
    _backgroundedAt = null;
    _backgroundGraceTimer?.cancel();
    _backgroundGraceTimer = null;

    final backgroundedLongEnoughToRequireReconnect =
        backgroundedAt != null &&
        _now().difference(backgroundedAt) >= _backgroundGraceDuration;
    if (!backgroundedLongEnoughToRequireReconnect &&
        state.status == SessionStatus.connected) {
      return;
    }

    // Cancel any in-flight reconnect backoff timer so we reconnect immediately
    // instead of waiting for the (possibly large) exponential delay.
    _reconnectTimer?.cancel();
    _reconnectDelayMs = _baseReconnectDelayMs;
    final config = ref.read(relayConfigProvider);
    _connect(config);
  }

  Future<void> _connect(RelayConfig config, {bool renewed = false}) async {
    if (_disposed || _ageRestricted) return;

    // Whatever connection came before ends here, capped at its last frame.
    _markLiveSubscriptionsClosed();
    _connectedAt = null;
    final generation = ++_connectionGeneration;
    // A replacement socket is unauthenticated until its AUTH completes.
    _socketConnected = false;
    state = SessionState(
      status: _hasConnectedOnce
          ? SessionStatus.reconnecting
          : SessionStatus.connecting,
      reconnectAttempt: state.reconnectAttempt,
    );

    _socket?.dispose();
    late final RelaySocket socket;
    socket = _socketFactory(
      wsUrl: config.wsUrl,
      nsec: config.nsec,
      onMessage: (message) {
        if (generation == _connectionGeneration) _handleMessage(message);
      },
      onConnected: () => _handleConnected(generation),
      onDisconnected: (error) {
        if (_disposed || generation != _connectionGeneration) return;
        final auth = ref.read(hostedRelayAuthProvider);
        if (auth.protects(config) && relayUpgradeRefused(error)) {
          auth.rejected(
            config,
            socket.headers['Nostr-Federated-Identity'],
            terminal: renewed,
          );
          if (!renewed) {
            unawaited(_connect(config, renewed: true));
            return;
          }
          error = AccountFailure.denied;
        }
        _handleDisconnected(generation, error, closeCode: socket.closeCode);
      },
    );
    _socket = socket;

    try {
      final auth = ref.read(hostedRelayAuthProvider);
      if (auth.protects(config)) socket.headers = await auth.headers(config);
      if (_disposed || generation != _connectionGeneration) return;
      await socket.connect();
    } on AccountFailure catch (error) {
      _handleDisconnected(generation, error);
    }
  }

  Future<void> _handleConnected(int generation) async {
    if (_disposed || generation != _connectionGeneration) return;
    _socketConnected = true;
    ref.read(hostedRelayAuthProvider).connected(ref.read(relayConfigProvider));
    _hasConnectedOnce = true;
    // The backoff is not reset here: a client bounced repeatedly must keep
    // backing off. [_handleDisconnected] resets it after a stable connection.
    _connectedAt = _now();
    _lastInboundAt = _connectedAt;
    state = const SessionState(status: SessionStatus.connected);
    await _replayLiveSubscriptions(generation);
  }

  void _handleDisconnected(int generation, Object? error, {int? closeCode}) {
    if (_disposed || generation != _connectionGeneration) return;
    _socketConnected = false;
    final connectedAt = _connectedAt;
    _connectedAt = null;
    if (connectedAt != null &&
        isRelayConnectionStable(_now().difference(connectedAt))) {
      _reconnectDelayMs = _baseReconnectDelayMs;
    }
    _cancelAllHistory(error);
    _rejectAllPending(error);
    _resetAllClosedRetries();
    _eventBuffer.clear();
    _flushTimer?.cancel();
    _flushTimer = null;
    if (error is RelayAuthRejectedException ||
        (error is AccountFailure && error != AccountFailure.unavailable)) {
      _reconnectTimer?.cancel();
      state = const SessionState(status: SessionStatus.disconnected);
      return;
    }
    _scheduleReconnect(closeCode);
  }

  void _scheduleReconnect(int? closeCode) {
    if (_disposed || _paused) return;
    final attempt = state.reconnectAttempt + 1;
    state = SessionState(
      status: SessionStatus.reconnecting,
      reconnectAttempt: attempt,
    );

    _reconnectTimer?.cancel();
    final delay = relayReconnectDelay(
      backoffMs: _reconnectDelayMs,
      closeCode: closeCode,
      jitterUnit: _random.nextDouble(),
      spreadUnit: _random.nextDouble(),
    );
    _reconnectTimer = _reconnectTimerFactory(delay, () {
      _reconnectDelayMs = min(_reconnectDelayMs * 2, _maxReconnectDelayMs);
      final config = ref.read(relayConfigProvider);
      _connect(config);
    });
  }

  /// Records when each subscription that was live on the closing connection
  /// stopped being live, so the next replay can resume from there.
  ///
  /// Liveness ends at the last frame received, not at the close: after a
  /// device sleep the loss is only noticed on wake, and nothing arrived in
  /// between.
  void _markLiveSubscriptionsClosed() {
    final now = _now();
    final inbound = _lastInboundAt;
    final end = inbound == null || inbound.isAfter(now) ? now : inbound;
    final closedAt = end.millisecondsSinceEpoch ~/ 1000;
    for (final subscription in _liveSubscriptions.values) {
      if (!subscription.liveOnConnection) continue;
      subscription.liveOnConnection = false;
      subscription.liveUntil = closedAt;
    }
  }

  /// Replay all live subscriptions after a reconnect, with a time skew to
  /// catch events that occurred during the disconnect.
  Future<void> _replayLiveSubscriptions(int generation) async {
    if (_rateLimitGate.isActive) await _rateLimitGate.wait();
    if (!_isActiveConnection(generation)) return;
    _sendHeldHistory();

    final entries = _liveSubscriptions.entries.toList();
    final visibleChannelId = _visibleChannelsByOwner.isEmpty
        ? null
        : _visibleChannelsByOwner.values.last;
    if (visibleChannelId != null) {
      entries.sort((left, right) {
        final leftVisible =
            left.value.filter.tags['#h']?.contains(visibleChannelId) ?? false;
        final rightVisible =
            right.value.filter.tags['#h']?.contains(visibleChannelId) ?? false;
        if (leftVisible == rightVisible) return 0;
        return leftVisible ? -1 : 1;
      });
    }

    await _sendReplayBatches(entries, generation);
  }

  Future<void> _replayPendingClosedRetries(int generation) async {
    if (!_isActiveConnection(generation)) return;
    final entries = _pendingClosedRetries.entries
        .where((entry) => entry.value.generation == generation)
        .map(
          (entry) => MapEntry<String, _LiveSubscription>(
            entry.key,
            entry.value.subscription,
          ),
        )
        .toList();
    await _sendReplayBatches(entries, generation, pendingClosedRetries: true);
  }

  Future<void> _sendReplayBatches(
    List<MapEntry<String, _LiveSubscription>> entries,
    int generation, {
    bool pendingClosedRetries = false,
  }) async {
    for (var i = 0; i < entries.length; i += _replayBatchSize) {
      if (_rateLimitGate.isActive) await _rateLimitGate.wait();
      if (!_isActiveConnection(generation)) return;
      final batch = entries.sublist(
        i,
        min(i + _replayBatchSize, entries.length),
      );
      for (final entry in batch) {
        if (_liveSubscriptions[entry.key] != entry.value) continue;
        if (pendingClosedRetries) {
          final pendingRetry = _pendingClosedRetries[entry.key];
          if (pendingRetry?.subscription != entry.value ||
              pendingRetry?.generation != generation) {
            continue;
          }
          _pendingClosedRetries.remove(entry.key);
        }
        _sendLiveReq(entry.key, entry.value);
      }
      if (i + _replayBatchSize < entries.length) {
        await _replayDelay(_replayInterBatchDelay);
      }
    }
  }

  bool _isActiveConnection(int generation) =>
      !_disposed && generation == _connectionGeneration;

  NostrFilter _replayFilter(_LiveSubscription subscription) {
    final filter = subscription.filter;
    final since = relayResumeSince(
      originalSince: filter.since,
      until: filter.until,
      lastSeenCreatedAt: subscription.lastSeenCreatedAt,
      liveUntil: subscription.liveUntil,
    );
    return since == null || since == filter.since
        ? filter
        : filter.copyWithSince(since);
  }

  void _handleMessage(List<dynamic> data) {
    _lastInboundAt = _now();
    if (data.isEmpty) return;
    final type = data[0] as String;

    switch (type) {
      case 'EVENT':
        _handleEvent(data);
      case 'EOSE':
        _handleEose(data);
      case 'CLOSED':
        _handleClosed(data);
      case 'OK':
        _handleOk(data);
    }
  }

  void _handleEvent(List<dynamic> data) {
    if (data.length < 3) return;
    final subId = data[1] as String;
    final eventJson = data[2] as Map<String, dynamic>;
    final event = NostrEvent.fromJson(eventJson);

    // History subscriptions accumulate immediately.
    final historySub = _historySubscriptions[subId];
    if (historySub != null) {
      historySub.events.add(event);
      return;
    }

    // Live subscriptions get batched. An EVENT proves the stream is active,
    // but not that a retry replay is complete; only EOSE is that boundary.
    final liveSub = _liveSubscriptions[subId];
    if (liveSub != null) {
      _resetClosedRetry(liveSub);
      // Track last seen timestamp for reconnect replay.
      if (liveSub.lastSeenCreatedAt == null ||
          event.createdAt > liveSub.lastSeenCreatedAt!) {
        liveSub.lastSeenCreatedAt = event.createdAt;
      }
      _eventBuffer.add(_BufferedEvent(subId, event));
      _scheduleFlush();
    }
  }

  void _handleEose(List<dynamic> data) {
    if (data.length < 2) return;
    final subId = data[1] as String;

    // History subscription: resolve with collected events.
    final historySub = _historySubscriptions.remove(subId);
    if (historySub != null) {
      historySub.timeout.cancel();
      if (!historySub.completer.isCompleted) {
        historySub.completer.complete(historySub.events);
      }
      _sendClose(subId);
      return;
    }

    // Live subscription: flush replay callbacks before signaling ready. This
    // ordering matters for retry replays, whose original ready completer has
    // already been released.
    final liveSub = _liveSubscriptions[subId];
    if (liveSub != null) {
      // A second REQ on this socket makes its EOSE ambiguous: it may belong
      // to the REQ it replaced, so it cannot vouch for the replacement.
      liveSub.liveOnConnection =
          liveSub.ambiguousGeneration != _connectionGeneration;
      _resetClosedRetry(liveSub);
      _flushBufferedEventsNow();
      liveSub.onStatusChanged?.call(RelaySubscriptionStatus.ready);
    }
    if (liveSub != null &&
        liveSub.readyCompleter != null &&
        !liveSub.readyCompleter!.isCompleted) {
      liveSub.readyCompleter!.complete();
      liveSub.readyCompleter = null;
    }
  }

  void _handleClosed(List<dynamic> data) {
    if (data.length < 2) return;
    final subId = data[1] as String;
    final message = data.length >= 3 && data[2] is String
        ? data[2] as String
        : 'subscription closed by relay';
    final closedClass = classifyRelayClosed(message);

    final historySub = _historySubscriptions.remove(subId);
    if (historySub != null) {
      if (closedClass == RelayClosedClass.rateLimited) {
        _rateLimitGate.activate(parseRateLimitRetrySeconds(message));
      }
      historySub.timeout.cancel();
      if (!historySub.completer.isCompleted) {
        historySub.completer.completeError(Exception(message));
      }
      return;
    }

    final liveSub = _liveSubscriptions[subId];
    if (liveSub == null) return;
    final readyCompleter = liveSub.readyCompleter;
    // The stream is interrupted until the retry's EOSE.
    liveSub.liveOnConnection = false;
    // A deadline is terminal for one-shot history, but a live subscription
    // retries it: same filter under backoff, which can keep retrying at the
    // cap. `since` is not advanced, so events missed while closed aren't
    // skipped.
    if (closedClass == RelayClosedClass.terminal &&
        !isRelayDeadlineError(message)) {
      if (readyCompleter != null && !readyCompleter.isCompleted) {
        readyCompleter.completeError(Exception(message));
      }
      liveSub.onClosed?.call(message);
      _removeLiveSubscription(subId, liveSub);
      return;
    }
    if (readyCompleter != null && !readyCompleter.isCompleted) {
      readyCompleter.complete();
      liveSub.readyCompleter = null;
    }
    liveSub.onStatusChanged?.call(RelaySubscriptionStatus.retrying);
    if (liveSub.closedRetryTimer != null) return;

    final attempt = liveSub.closedRetryAttempt;
    final backoffMs = attempt >= 5
        ? _maxReconnectDelayMs
        : _baseReconnectDelayMs * (1 << attempt);
    var delayMs = backoffMs;
    if (closedClass == RelayClosedClass.rateLimited) {
      final retrySeconds = parseRateLimitRetrySeconds(message);
      _rateLimitGate.activate(retrySeconds);
      final fallbackMs =
          (retrySeconds != null
              ? min(retrySeconds, RelayRateLimitGate.maxRetrySeconds)
              : RelayRateLimitGate.defaultRetrySeconds) *
          1000;
      delayMs = max(
        backoffMs,
        _rateLimitGate.remainingMs() == 0
            ? fallbackMs
            : _rateLimitGate.remainingMs(),
      );
    }

    liveSub.closedRetryAttempt = attempt + 1;
    final retryGeneration = _connectionGeneration;
    liveSub.closedRetryTimer = _retryTimerFactory(
      Duration(milliseconds: delayMs),
      () async {
        liveSub.closedRetryTimer = null;
        if (!_isActiveConnection(retryGeneration) ||
            _liveSubscriptions[subId] != liveSub) {
          return;
        }
        if (_rateLimitGate.isActive) await _rateLimitGate.wait();
        if (!_isActiveConnection(retryGeneration) ||
            _liveSubscriptions[subId] != liveSub ||
            !_socketConnected) {
          return;
        }
        _pendingClosedRetries[subId] = _ClosedRetry(
          subscription: liveSub,
          generation: retryGeneration,
        );
        _scheduleClosedRetryReplay(retryGeneration);
      },
    );
  }

  void _scheduleClosedRetryReplay(int generation) {
    if (_closedRetryReplayScheduled) return;
    _closedRetryReplayScheduled = true;
    scheduleMicrotask(() async {
      try {
        await _replayPendingClosedRetries(generation);
      } finally {
        _closedRetryReplayScheduled = false;
        _pendingClosedRetries.removeWhere(
          (_, retry) => retry.generation != _connectionGeneration,
        );
        if (_pendingClosedRetries.values.any(
          (retry) => retry.generation == _connectionGeneration,
        )) {
          _scheduleClosedRetryReplay(_connectionGeneration);
        }
      }
    });
  }

  void _handleOk(List<dynamic> data) {
    if (data.length < 3) return;
    final eventId = data[1] as String;
    final accepted = data[2] as bool;
    final message = data.length > 3 && data[3] is String
        ? data[3] as String
        : '';

    final pending = _pendingEvents.remove(eventId);
    if (pending == null) return;
    pending.timeout.cancel();

    if (accepted) {
      // We don't have the full event here; create a minimal placeholder.
      // Command kinds (e.g. 41010, 30620, 46020) return "response:{...}" in
      // the OK message — preserve it in `content` so callers can parse it.
      if (!pending.completer.isCompleted) {
        pending.completer.complete(
          NostrEvent(
            id: eventId,
            pubkey: '',
            createdAt: 0,
            kind: 0,
            tags: [],
            content: message,
            sig: '',
          ),
        );
      }
    } else {
      // Back-pressure now arrives here rather than as a NOTICE: the relay
      // rejects an over-quota EVENT on the OK channel so this pending publish
      // can be settled at all. Without arming the gate the send would fail
      // without ever backing off.
      if (message.startsWith('rate-limited:')) {
        _rateLimitGate.activate(parseRateLimitRetrySeconds(message));
      }
      if (!pending.completer.isCompleted) {
        pending.completer.completeError(
          Exception(message.isNotEmpty ? message : 'Event rejected'),
        );
      }
    }
  }

  void _scheduleFlush() {
    _flushTimer ??= Timer(
      const Duration(milliseconds: _eventBatchMs),
      _flushEventBuffer,
    );
  }

  void _flushBufferedEventsNow() {
    _flushTimer?.cancel();
    _flushTimer = null;
    _flushEventBuffer();
  }

  void _flushEventBuffer() {
    _flushTimer = null;
    if (_eventBuffer.isEmpty) return;

    final batch = List<_BufferedEvent>.from(_eventBuffer);
    _eventBuffer.clear();

    for (final buffered in batch) {
      final sub = _liveSubscriptions[buffered.subId];
      if (sub == null) continue;

      // Deduplicate per subscription. The same relay event can legitimately
      // match multiple live subscriptions, e.g. the channel list unread listener
      // and the open channel message listener.
      final deliveryKey = '${buffered.subId}:${buffered.event.id}';
      if (_recentDeliveryKeys.contains(deliveryKey)) continue;

      // Cap the dedup set to prevent unbounded memory growth.
      if (_recentDeliveryKeys.length >= _maxRecentDeliveryKeys) {
        _recentDeliveryKeys.clear();
      }
      _recentDeliveryKeys.add(deliveryKey);

      sub.onEvent(buffered.event);
    }
  }

  String _nextSubId(String prefix) {
    _subIdCounter++;
    return '$prefix-$_subIdCounter';
  }

  void _sendLiveReq(String subId, _LiveSubscription subscription) {
    if (subscription.reqGeneration == _connectionGeneration) {
      subscription.ambiguousGeneration = _connectionGeneration;
    }
    subscription.reqGeneration = _connectionGeneration;
    _sendReq(subId, _replayFilter(subscription));
  }

  void _sendReq(String subId, NostrFilter filter) {
    _socket?.send(['REQ', subId, filter.toJson()]);
  }

  void _sendHeldHistory() {
    for (final MapEntry(key: subId, value: sub)
        in _historySubscriptions.entries) {
      final filter = sub.heldFilter;
      if (filter == null) continue;
      sub.heldFilter = null;
      _sendReq(subId, filter);
    }
  }

  void _sendClose(String subId) {
    _socket?.send(['CLOSE', subId]);
  }

  void _unsubscribe(String subId) {
    final subscription = _liveSubscriptions[subId];
    if (subscription != null) {
      _removeLiveSubscription(subId, subscription);
    }
    _sendClose(subId);
  }

  void _removeLiveSubscription(String subId, _LiveSubscription subscription) {
    if (_liveSubscriptions[subId] != subscription) return;
    _liveSubscriptions.remove(subId);
    _pendingClosedRetries.remove(subId);
    subscription.closedRetryTimer?.cancel();
    subscription.closedRetryTimer = null;
    _recentDeliveryKeys.removeWhere((key) => key.startsWith('$subId:'));
  }

  void _resetClosedRetry(_LiveSubscription subscription) {
    subscription.closedRetryAttempt = 0;
    subscription.closedRetryTimer?.cancel();
    subscription.closedRetryTimer = null;
  }

  void _cancelAllClosedRetries() {
    _pendingClosedRetries.clear();
    for (final subscription in _liveSubscriptions.values) {
      subscription.closedRetryTimer?.cancel();
      subscription.closedRetryTimer = null;
    }
  }

  void _resetAllClosedRetries() {
    _pendingClosedRetries.clear();
    for (final subscription in _liveSubscriptions.values) {
      _resetClosedRetry(subscription);
    }
  }

  void _cancelAllHistory(Object? error) {
    for (final entry in _historySubscriptions.values) {
      entry.timeout.cancel();
      if (!entry.completer.isCompleted) {
        entry.completer.completeError(error ?? Exception('Connection lost'));
      }
    }
    _historySubscriptions.clear();
  }

  void _rejectAllPending(Object? error) {
    for (final entry in _pendingEvents.values) {
      entry.timeout.cancel();
      if (!entry.completer.isCompleted) {
        entry.completer.completeError(error ?? Exception('Connection lost'));
      }
    }
    _pendingEvents.clear();
  }

  void _dispose() {
    _disposed = true;
    _contextGeneration++;
    _beforePauseCallbacks.clear();
    _connectionGeneration++;
    _reconnectTimer?.cancel();
    _flushTimer?.cancel();
    _backgroundGraceTimer?.cancel();
    _backgroundedAt = null;
    _cancelAllClosedRetries();
    _rateLimitGate.reset();
    _visibleChannelsByOwner.clear();
    _socketConnected = false;
    _cancelAllHistory(null);
    _rejectAllPending(null);
    final subscriptions = _liveSubscriptions.values.toList();
    _liveSubscriptions.clear();
    for (final subscription in subscriptions) {
      subscription.closedRetryTimer?.cancel();
      subscription.closedRetryTimer = null;
    }
    _recentDeliveryKeys.clear();
    _socket?.dispose();
    _socket = null;
    _httpQueryClient.close();
  }
}

final relaySessionProvider =
    NotifierProvider<RelaySessionNotifier, SessionState>(
      RelaySessionNotifier.new,
    );