streamAddressableMetrics method

Stream<Nip85AddressableMetrics> streamAddressableMetrics(String eventAddress, { Set<Nip85Metric>? metrics, List<Nip85TrustedProvider>? providers, })

Stream addressable event metrics from trusted providers

Implementation

Stream<Nip85AddressableMetrics> streamAddressableMetrics(
  String eventAddress, {
  Set<Nip85Metric>? metrics,
  List<Nip85TrustedProvider>? providers,
}) {
  final effectiveProviders = _filterProviders(
    providers ?? _defaultProviders,
    kind: Nip85Kind.addressable,
    metrics: metrics,
  );

  final controller = StreamController<Nip85AddressableMetrics>();

  if (effectiveProviders.isEmpty) {
    controller.close();
    return controller.stream;
  }

  final providersByRelay = <String, List<Nip85TrustedProvider>>{};
  for (final provider in effectiveProviders) {
    providersByRelay.putIfAbsent(provider.relay, () => []).add(provider);
  }

  final subscriptionIds = <String>[];

  for (final entry in providersByRelay.entries) {
    final relay = entry.key;
    final relayProviders = entry.value;
    final providerPubkeys = relayProviders.map((p) => p.pubkey).toList();

    final response = _requests.subscription(
      filter: Filter(
        kinds: [Nip85Kind.addressable],
        authors: providerPubkeys,
        dTags: [eventAddress],
      ),
      explicitRelays: [relay],
    );

    subscriptionIds.add(response.requestId);

    response.stream.listen(
      (event) {
        final parsed = _parseAddressableMetricsEvent(event, metrics);
        if (parsed != null) {
          controller.add(parsed);
        }
      },
      onError: (e) {},
    );
  }

  controller.onCancel = () async {
    for (final id in subscriptionIds) {
      await _requests.closeSubscription(id);
    }
  };

  return controller.stream;
}