swarm/lib/src/nostr_publish_actor.dart

import 'dart:async';
import 'dart:convert';
import 'package:actors/actors.dart';
import 'package:retry/retry.dart';
import 'relay_connection.dart';

final class RetryConfig {
  final int maxAttempts;
  final Duration baseDelay;
  final Duration maxDelay;
  final bool jitter;
  const RetryConfig({
    this.maxAttempts = 3,
    this.baseDelay = const Duration(seconds: 1),
    this.maxDelay = const Duration(seconds: 32),
    this.jitter = true,
  });
}

sealed class RelayOutcome {
  const RelayOutcome();
}

final class RelayConfirmed extends RelayOutcome {
  const RelayConfirmed();
}

final class RelayRejected extends RelayOutcome {
  final String reason;
  const RelayRejected(this.reason);
}

final class RelayFailed extends RelayOutcome {
  const RelayFailed();
}

sealed class PublishMessage {
  const PublishMessage();
}

final class PublishBatch extends PublishMessage {
  final List<String> events;
  const PublishBatch(this.events);
}

sealed class PublishResult {
  const PublishResult();
}

final class BatchPublished extends PublishResult {
  final Map<String, RelayOutcome> outcomes;
  const BatchPublished(this.outcomes);
}

final class _RelayRejectedException implements Exception {
  final String reason;
  const _RelayRejectedException(this.reason);
}

class NostrPublishActor with Handler<PublishMessage, PublishResult> {
  final List<String> _relayUrls;
  final RetryConfig _retryConfig;
  final Duration _batchTimeout;
  final RelayConnection Function(String url) _connectionFactory;

  NostrPublishActor({
    required List<String> relayUrls,
    RetryConfig retryConfig = const RetryConfig(),
    Duration batchTimeout = const Duration(seconds: 10),
    RelayConnection Function(String url) connectionFactory =
        WebSocketRelayConnection.new,
  })  : _relayUrls = relayUrls,
        _retryConfig = retryConfig,
        _batchTimeout = batchTimeout,
        _connectionFactory = connectionFactory;

  @override
  Future<PublishResult> handle(PublishMessage message) => switch (message) {
        PublishBatch(:final events) => _publishBatch(events),
      };

  Future<PublishResult> _publishBatch(List<String> events) async {
    final ids = events
        .map((e) => (jsonDecode(e) as Map<String, dynamic>)['id'] as String)
        .toList();
    final outcomes = await Future.wait(
      _relayUrls.map((url) => _publishToRelay(url, events, ids)),
    );
    return BatchPublished(Map.fromIterables(_relayUrls, outcomes));
  }

  Future<RelayOutcome> _publishToRelay(
    String url,
    List<String> events,
    List<String> ids,
  ) async {
    try {
      await retry(
        () => _attemptPublish(url, events, ids),
        retryIf: (e) => e is! _RelayRejectedException,
        maxAttempts: _retryConfig.maxAttempts,
        delayFactor: _retryConfig.baseDelay,
        maxDelay: _retryConfig.maxDelay,
        randomizationFactor: _retryConfig.jitter ? 0.25 : 0.0,
      );
      return const RelayConfirmed();
    } on _RelayRejectedException catch (e) {
      return RelayRejected(e.reason);
    } catch (_) {
      return const RelayFailed();
    }
  }

  Future<void> _attemptPublish(
    String url,
    List<String> events,
    List<String> ids,
  ) async {
    final conn = _connectionFactory(url);
    final pending = {...ids};
    try {
      await conn.connect();
      for (final e in events) {
        await conn.send(jsonEncode(['EVENT', jsonDecode(e)]));
      }
      await for (final raw in conn.messages.timeout(_batchTimeout)) {
        try {
          final list = jsonDecode(raw) as List;
          if (list[0] != 'OK') continue;
          if (list[2] == false) {
            throw _RelayRejectedException(list[3] as String);
          }
          pending.remove(list[1] as String);
          if (pending.isEmpty) break;
        } catch (e) {
          if (e is _RelayRejectedException) rethrow;
          // malformed message, skip
        }
      }
    } finally {
      await conn.close();
    }
  }
}