using Microsoft.Extensions.Options; using MongoDB.Entities; using PrivaPub.Federation.Actors; using PrivaPub.Federation.Objects; using PrivaPub.Federation.Outbox; using PrivaPub.Federation.Rendering; using PrivaPub.Infrastructure.Http; using PrivaPub.Models.Federation; using PrivaPub.Models.User; using System.Text.Json.Nodes; using static PrivaPub.Federation.Objects.ActivityJson; namespace PrivaPub.Federation.Relays { public interface IRelays { /// The relays Federation:Relays names, as it names them. IReadOnlyList Configured { get; } /// The relays named by the configuration or by the server's settings, each once. IReadOnlyList Named { get; } Task Reconcile(CancellationToken token); Task Answered(JsonNode activity, ForeignAvatar actor, bool accepted, CancellationToken token); Task Passes(string actorUri, CancellationToken token); Task> Inboxes(CancellationToken token); } // Relays (Activity-Relay, aode-relay, pub-relay): the instance actor follows Public at each relay the configuration // (Federation:Relays) or the server's settings (ServerSettings.Relays, from the administrator's page) name, as Mastodon // subscribes, and takes what they pass on: public posts, forwarded as their authors sent them (read again from their // origin, as any forwarded post) or announced by the relay. A persona's public posts go to the relays that accepted us // (owner decision 2026-10-06), as Mastodon sends them; nothing else of a persona's does. public class Relays : IRelays { static readonly TimeSpan AskAgain = TimeSpan.FromDays(1); readonly IOptions _options; readonly IRemoteActorService _remoteActors; readonly ILocalActorService _localActors; readonly IDeliveryService _delivery; readonly Domain.Instance.IInstanceSettings _settings; readonly SemaphoreSlim _reconciling = new(1, 1);//the hourly round and the administrator's page, one at a time volatile HashSet _accepted; public Relays(IOptions options, IRemoteActorService remoteActors, ILocalActorService localActors, IDeliveryService delivery, Domain.Instance.IInstanceSettings settings) { _options = options; _remoteActors = remoteActors; _localActors = localActors; _delivery = delivery; _settings = settings; } public IReadOnlyList Configured => Clean(_options.Value.Relays); public IReadOnlyList Named => Clean(Configured.Concat(_settings.Current.Relays ?? new())); static List Clean(IEnumerable addresses) => (addresses ?? Array.Empty()).Where(r => !string.IsNullOrWhiteSpace(r)).Select(r => r.Trim()).Distinct(StringComparer.Ordinal).ToList(); // subscribes to the relays the configuration and the settings name (again a day after an unanswered or refused // request), and unsubscribes from those they no longer name: with an Undo, unless the same relay is still named // under another address (an Undo would end that subscription too) public async Task Reconcile(CancellationToken token) { await _reconciling.WaitAsync(token); try { await ReconcileNow(token); } finally { _reconciling.Release(); } } async Task ReconcileNow(CancellationToken token) { var configured = Named; var known = await DB.Default.Find().ExecuteAsync(token); if (configured.Count == 0 && known.Count == 0) return; var instance = await _localActors.GetInstanceActor(token); var kept = known.Where(k => configured.Contains(k.Configured)).Select(k => k.ActorURI).Where(a => a != default).ToHashSet(StringComparer.Ordinal); foreach (var gone in known.Where(k => !configured.Contains(k.Configured))) { if (gone.State != RelayState.Rejected && !string.IsNullOrEmpty(gone.InboxURL) && !kept.Contains(gone.ActorURI)) await _delivery.Enqueue(instance, new[] { gone.InboxURL }, Undo(instance, gone.FollowActivityURI), token); await DB.Default.DeleteAsync(gone.ID); } _accepted = default; foreach (var address in configured) { var subscription = known.FirstOrDefault(k => k.Configured == address); if (subscription is { State: RelayState.Accepted } || subscription != default && DateTime.UtcNow - subscription.RequestedAt < AskAgain) continue; var relay = await Resolve(address, token); if (relay == default || string.IsNullOrEmpty(relay.InboxURL)) continue; subscription ??= new RelaySubscription { Configured = address }; subscription.ActorURI = relay.ActorURI; subscription.InboxURL = relay.InboxURL; subscription.FollowActivityURI = instance.ActivityUri($"relay-{Guid.NewGuid():N}"); subscription.State = RelayState.Pending; subscription.RequestedAt = DateTime.UtcNow; subscription.AnsweredAt = default; await DB.Default.SaveAsync(subscription, token); await _delivery.Enqueue(instance, new[] { relay.InboxURL }, Follow(instance, subscription.FollowActivityURI), token); } _accepted = default; } // the relay's Accept or Reject of the subscription, by the Follow it answers public async Task Answered(JsonNode activity, ForeignAvatar actor, bool accepted, CancellationToken token) { var followId = Id(activity["object"]); if (followId == default) return false; var subscription = await DB.Default.Find().Match(s => s.FollowActivityURI == followId && s.ActorURI == actor.ActorURI) .ExecuteFirstAsync(token); if (subscription == default) return false; await DB.Default.Update().MatchID(subscription.ID) .Modify(s => s.State, accepted ? RelayState.Accepted : RelayState.Rejected) .Modify(s => s.AnsweredAt, DateTime.UtcNow) .ExecuteAsync(token); _accepted = default; return true; } // whether the actor is a relay that accepted our subscription public async Task Passes(string actorUri, CancellationToken token) { if (string.IsNullOrEmpty(actorUri)) return false; var accepted = _accepted ??= (await DB.Default.Find().Match(s => s.State == RelayState.Accepted).ExecuteAsync(token)) .Select(s => s.ActorURI).ToHashSet(StringComparer.Ordinal); return accepted.Contains(actorUri); } // the inboxes of the relays that accepted our subscription, where a persona's public posts also go public async Task> Inboxes(CancellationToken token) => (await DB.Default.Find().Match(s => s.State == RelayState.Accepted).ExecuteAsync(token)) .Select(s => s.InboxURL).Where(i => !string.IsNullOrEmpty(i)).Distinct(StringComparer.Ordinal).ToList(); // the relay's actor, from its address or, for a relay named by its inbox, from the /actor beside it async Task Resolve(string address, CancellationToken token) { var actor = await _remoteActors.GetActor(address, refresh: false, token); if (actor != default || !address.EndsWith("/inbox", StringComparison.Ordinal)) return actor; return await _remoteActors.GetActor(address[..^"/inbox".Length] + "/actor", refresh: false, token); } static JsonObject Follow(LocalActor instance, string id) => new() { ["@context"] = ActivityPubRenderer.Context(), ["id"] = id, ["type"] = "Follow", ["actor"] = instance.Uri, ["object"] = Addressing.Public }; static JsonObject Undo(LocalActor instance, string followId) => new() { ["@context"] = ActivityPubRenderer.Context(), ["id"] = instance.ActivityUri($"relay-undo-{Guid.NewGuid():N}"), ["type"] = "Undo", ["actor"] = instance.Uri, ["object"] = new JsonObject { ["id"] = followId, ["type"] = "Follow", ["actor"] = instance.Uri, ["object"] = Addressing.Public } }; } // subscribes to the relays named (configuration and settings) a minute after start, then every six hours public sealed class RelaySubscriber(IServiceScopeFactory scopes, ILogger logger) : BackgroundService { static readonly TimeSpan Every = TimeSpan.FromHours(6); protected override async Task ExecuteAsync(CancellationToken token) { await Task.Delay(TimeSpan.FromMinutes(1), token); using var timer = new PeriodicTimer(Every); do { try { using var scope = scopes.CreateScope(); await scope.ServiceProvider.GetRequiredService().Reconcile(token); } catch (Exception ex) when (ex is not OperationCanceledException) { logger.LogWarning(ex, "Relay subscriptions could not be brought up to date"); } } while (await timer.WaitForNextTickAsync(token)); } } }