feat: expose the outcome of a request on each relay

packages/ndk/lib/domain_layer/entities/relay_request_outcome.dart

@@ -0,0 +1,52 @@
+/// What a relay did with a request.
+///
+/// Declared in collapse precedence order: one request can run on several
+/// connections to the same relay, an anonymous one handed over to an
+/// authenticated one, and the url reports the first of these its connections
+/// reached.
+enum RelayRequestOutcomeType {
+  /// the relay has not ended the request, it may still send events
+  pending,
+
+  /// the relay sent an EOSE, it has given everything it stored
+  eose,
+
+  /// the relay ended the request itself with a CLOSED
+  closed,
+
+  /// the connection went away before the relay ended the request
+  disconnected,
+
+  /// the request ran into its timeout before the relay ended it
+  timedOut,
+
+  /// the request never reached the relay
+  notSent,
+}
+
+/// What a request ended with on a single relay
+class RelayRequestOutcome {
+  /// what the relay did with the request
+  final RelayRequestOutcomeType type;
+
+  /// why, when there is a reason to give: the message of a CLOSED, or what
+  /// kept the request from being sent
+  final String? message;
+
+  /// creates a new [RelayRequestOutcome]
+  const RelayRequestOutcome(this.type, {this.message});
+
+  @override
+  String toString() => message == null ? type.name : '${type.name}: $message';
+
+  @override
+  bool operator ==(Object other) =>
+      identical(this, other) ||
+      other is RelayRequestOutcome &&
+          runtimeType == other.runtimeType &&
+          type == other.type &&
+          message == other.message;
+
+  @override
+  int get hashCode => Object.hash(type, message);
+}

packages/ndk/lib/domain_layer/entities/request_response.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 
 import 'nip_01_event.dart';
+import 'relay_request_outcome.dart';
 
 // coverage:ignore-start
 
@@ -15,12 +16,43 @@ class NdkResponse {
   /// as they arrive from the nostr request.
   final Stream<Nip01Event> stream;
 
+  final Map<String, RelayRequestOutcome> Function() _relayOutcomes;
+
+  final Future<Map<String, RelayRequestOutcome>> _relayOutcomesDone;
+
   /// A future that resolves to a list of all [Nip01Event] objects
   /// once the request is complete (EOSE rcv).
   Future<List<Nip01Event>> get future => stream.toList();
 
+  /// What the request ended with on each relay it was sent to, as it stands
+  /// now, keyed by relay url.
+  ///
+  /// Reading it tells an exhausted relay from a silent one: a relay that has
+  /// not answered yet is [RelayRequestOutcomeType.pending], which is what a
+  /// live subscription shows for as long as it runs.
+  Map<String, RelayRequestOutcome> get relayOutcomes => _relayOutcomes();
+
+  /// [relayOutcomes] once the request is over, either because every relay
+  /// ended it or because it ran into its timeout.
+  ///
+  /// A subscription is only over once it is closed, so this resolves on
+  /// `closeSubscription` for one.
+  Future<Map<String, RelayRequestOutcome>> get relayOutcomesDone =>
+      _relayOutcomesDone;
+
   /// Creates a new [NdkResponse] instance.
-  NdkResponse(this.requestId, this.stream);
+  NdkResponse(
+    this.requestId,
+    this.stream, {
+    Map<String, RelayRequestOutcome> Function()? relayOutcomes,
+    Future<Map<String, RelayRequestOutcome>>? relayOutcomesDone,
+  }) : _relayOutcomes = relayOutcomes ?? _noOutcomes,
+       _relayOutcomesDone =
+           relayOutcomesDone ??
+           Future.value(const <String, RelayRequestOutcome>{});
+
+  static Map<String, RelayRequestOutcome> _noOutcomes() =>
+      const <String, RelayRequestOutcome>{};
 }
 
 // coverage:ignore-end

packages/ndk/lib/domain_layer/entities/request_state.dart

@@ -7,6 +7,7 @@ import 'filter.dart';
 import 'ndk_request.dart';
 import 'nip_01_event.dart';
 import 'relay_connection_key.dart';
+import 'relay_request_outcome.dart';
 
 /// Single relay request state
 class RelayRequestState {
@@ -19,6 +20,12 @@ class RelayRequestState {
   bool receivedEOSE = false;
   bool receivedClosed = false;
 
+  /// message the relay sent with its CLOSED, null when it sent none
+  String? closedMessage;
+
+  /// set when the connection was gone before the relay ended the request
+  bool connectionGone = false;
+
   /// set while this connection authenticates to satisfy the request: the relay
   /// closed it, but it is on its way back and must not count as finished
   bool retryingAuth = false;
@@ -71,6 +78,13 @@ class RequestState {
   /// timeout duration, closes all streams
   Duration? timeoutDuration;
 
+  /// set when the request ended on its timeout instead of on the relays
+  bool timedOut = false;
+
+  /// request this one was merged into by the concurrency check, when its stream
+  /// got replaced by an identical request already in flight
+  RequestState? servedBy;
+
   /// called when timeout is triggered
   Function(RequestState)? onTimeout;
 
@@ -105,6 +119,7 @@ class RequestState {
   void _startTimeout(Duration duration) {
     _timeoutStartedAt = DateTime.now();
     _timeout = Timer(duration, () {
+      timedOut = true;
       onTimeout?.call(this);
       close();
     });
@@ -137,6 +152,49 @@ class RequestState {
         !element.retryingAuth,
   );
 
+  /// What the request ended with on each relay it was sent to, as it stands now
+  ///
+  /// Keyed by relay url: several connections to one relay collapse into the
+  /// outcome that comes first in [RelayRequestOutcomeType].
+  Map<String, RelayRequestOutcome> get relayOutcomes {
+    final served = servedBy;
+    if (served != null) {
+      return served.relayOutcomes;
+    }
+
+    final outcomes = <String, RelayRequestOutcome>{};
+    for (final request in requests.values) {
+      final outcome = _outcomeOf(request);
+      final current = outcomes[request.url];
+      if (current == null || outcome.type.index < current.type.index) {
+        outcomes[request.url] = outcome;
+      }
+    }
+    return outcomes;
+  }
+
+  RelayRequestOutcome _outcomeOf(RelayRequestState request) {
+    if (request.retryingAuth) {
+      return const RelayRequestOutcome(RelayRequestOutcomeType.pending);
+    }
+    if (request.receivedEOSE) {
+      return const RelayRequestOutcome(RelayRequestOutcomeType.eose);
+    }
+    if (request.receivedClosed) {
+      return RelayRequestOutcome(
+        RelayRequestOutcomeType.closed,
+        message: request.closedMessage,
+      );
+    }
+    if (request.connectionGone) {
+      return const RelayRequestOutcome(RelayRequestOutcomeType.disconnected);
+    }
+    if (timedOut) {
+      return const RelayRequestOutcome(RelayRequestOutcomeType.timedOut);
+    }
+    return const RelayRequestOutcome(RelayRequestOutcomeType.pending);
+  }
+
   /// Adds single relay request to the state
   void addRequest(RelayConnectionKey key, List<Filter> filters) {
     if (!requests.containsKey(key)) {

packages/ndk/lib/domain_layer/usecases/relay_manager.dart

@@ -1235,7 +1235,7 @@ class RelayManager<T> {
 
     // Check if this is an auth-required CLOSED message
     if (message != null && message.startsWith("auth-required")) {
-      _handleClosedAuthRequired(id, relayConnectivity);
+      _handleClosedAuthRequired(id, relayConnectivity, message);
       return;
     }
 
@@ -1247,7 +1247,7 @@ class RelayManager<T> {
       );
       RelayRequestState? request = state.requests[relayConnectivity.key];
       if (request != null) {
-        _endRequestOnRelay(relayConnectivity, id, request);
+        _endRequestOnRelay(relayConnectivity, id, request, message);
       }
 
       _checkNetworkClose(state);
@@ -1263,6 +1263,7 @@ class RelayManager<T> {
     RelayRequestState request,
   ) {
     request.receivedClosed = false;
+    request.closedMessage = null;
     request.retryingAuth = false;
     send(
       relayConnectivity,
@@ -1276,8 +1277,10 @@ class RelayManager<T> {
     RelayConnectivity relayConnectivity,
     String reqId,
     RelayRequestState request,
+    String? message,
   ) {
     request.receivedClosed = true;
+    request.closedMessage = message;
     relayConnectivity.stats.openRequestIds.remove(reqId);
   }
 
@@ -1295,6 +1298,7 @@ class RelayManager<T> {
   void _handleClosedAuthRequired(
     String reqId,
     RelayConnectivity relayConnectivity,
+    String message,
   ) {
     final state = globalState.inFlightRequests[reqId];
     if (state == null) {
@@ -1314,7 +1318,7 @@ class RelayManager<T> {
     }
 
     // whatever we do next, the relay just closed this one on this connection
-    _endRequestOnRelay(relayConnectivity, reqId, request);
+    _endRequestOnRelay(relayConnectivity, reqId, request, message);
 
     if (!key.isAnonymous) {
       if (_authenticatedConnections.contains(key)) {
@@ -1376,6 +1380,8 @@ class RelayManager<T> {
       }
       if (bound == null) {
         retry.receivedClosed = true;
+        retry.closedMessage =
+            "auth-required: no authenticated connection could be opened";
         _checkNetworkClose(state);
         return;
       }
@@ -1533,6 +1539,12 @@ class RelayManager<T> {
     );
 
     if (didAllRelaysFinish) {
+      for (final key in myNotConnectedRelays) {
+        final request = state.requests[key]!;
+        if (!request.receivedEOSE && !request.receivedClosed) {
+          request.connectionGone = true;
+        }
+      }
       state.networkController.close();
       updateRelayConnectivity();
     }

packages/ndk/lib/domain_layer/usecases/relay_sets_engine.dart

@@ -333,7 +333,12 @@ class RelaySetsEngine implements NetworkEngine {
       });
     }
 
-    return NdkResponse(state.id, state.stream);
+    return NdkResponse(
+      state.id,
+      state.stream,
+      relayOutcomes: () => state.relayOutcomes,
+      relayOutcomesDone: state.controller.done.then((_) => state.relayOutcomes),
+    );
   }
 
   @override

packages/ndk/lib/domain_layer/usecases/requests/concurrency_check.dart

@@ -30,8 +30,10 @@ class ConcurrencyCheck {
 
     // add already running stream to duplicate request
     // When original stream ends, close the duplicate's controller
+    final served = _globalState.inFlightRequests[hash]!;
+    requestState.servedBy = served;
     requestState.controller
-        .addStream(_globalState.inFlightRequests[hash]!.stream)
+        .addStream(served.stream)
         .then((_) => requestState.controller.close());
 
     return true;

packages/ndk/lib/domain_layer/usecases/requests/requests.dart

@@ -14,6 +14,7 @@ import '../../entities/ndk_request.dart';
 import '../../entities/nip_01_event.dart';
 import '../../entities/relay_connectivity.dart';
 import '../../entities/relay_set.dart';
+import '../../entities/relay_request_outcome.dart';
 import '../../entities/request_response.dart';
 import '../../entities/request_state.dart';
 import '../../repositories/event_verifier.dart';
@@ -302,7 +303,12 @@ class Requests {
   NdkResponse requestNostrEvent(NdkRequest request) {
     final state = RequestState(request);
 
-    final response = NdkResponse(state.id, state.stream);
+    final response = NdkResponse(
+      state.id,
+      state.stream,
+      relayOutcomes: () => state.relayOutcomes,
+      relayOutcomesDone: state.controller.done.then((_) => state.relayOutcomes),
+    );
 
     final concurrency = ConcurrencyCheck(_globalState);
 
@@ -409,6 +415,11 @@ class Requests {
     final aggregatedController = ReplaySubject<Nip01Event>();
     final seenEventIds = <String>{};
 
+    // a relay is paginated by its own sequence of requests, so what it ended
+    // with is what its last page ended with
+    final relayOutcomes = <String, RelayRequestOutcome>{};
+    final relayOutcomesDone = Completer<Map<String, RelayRequestOutcome>>();
+
     Future<void> paginate() async {
       final since = filter.since;
 
@@ -432,6 +443,7 @@ class Requests {
       );
 
       final initialEvents = await initialResponse.future;
+      relayOutcomes.addAll(initialResponse.relayOutcomes);
 
       // Emit initial events and discover relays
       final relayState = <String, _RelayPaginationState>{};
@@ -506,7 +518,9 @@ class Requests {
             ),
           );
 
-          return MapEntry(relay, await response.future);
+          final pageEvents = await response.future;
+          relayOutcomes.addAll(response.relayOutcomes);
+          return MapEntry(relay, pageEvents);
         });
 
         final results = await Future.wait(futures);
@@ -545,9 +559,16 @@ class Requests {
     }
 
     // Start pagination asynchronously
-    paginate();
+    paginate().whenComplete(
+      () => relayOutcomesDone.complete(Map.of(relayOutcomes)),
+    );
 
-    return NdkResponse(requestId, aggregatedController.stream);
+    return NdkResponse(
+      requestId,
+      aggregatedController.stream,
+      relayOutcomes: () => Map.of(relayOutcomes),
+      relayOutcomesDone: relayOutcomesDone.future,
+    );
   }
 
   /// Records fetched ranges for each relay that received EOSE

packages/ndk/lib/entities.dart

@@ -26,6 +26,7 @@ export 'domain_layer/entities/relay.dart';
 export 'domain_layer/entities/relay_connection_key.dart';
 export 'domain_layer/entities/relay_connectivity.dart';
 export 'domain_layer/entities/relay_info.dart';
+export 'domain_layer/entities/relay_request_outcome.dart';
 export 'domain_layer/entities/relay_set.dart';
 export 'domain_layer/entities/relay_stats.dart';
 export 'domain_layer/entities/request_response.dart';

packages/ndk/lib/ndk.dart

@@ -27,6 +27,7 @@ export 'domain_layer/entities/contact_list.dart';
 export 'domain_layer/entities/read_write.dart';
 export 'domain_layer/entities/relay.dart';
 export 'domain_layer/entities/relay_connection_key.dart';
+export 'domain_layer/entities/relay_request_outcome.dart';
 export 'domain_layer/entities/relay_set.dart';
 export 'domain_layer/entities/metadata.dart';
 export 'domain_layer/entities/event_filter.dart';