Services/Federation and Controllers/ServerToServer become
Federation/{Actors,Signing,Inbox,Outbox,Rendering,Controllers}, the first step
of the roadmap's layout. No type, route or behaviour changes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012CzABvBkbcFqoHdmi8b9WB
This commit is contained in:
1 parent
31f6015591
commit
034b792801
14 files changed
+40
-20
No files matched your search
@@ -0,0 +1,180 @@
|
||||
using MongoDB.Entities;
|
||||
|
||||
using PrivaPub.Models.Federation;
|
||||
using PrivaPub.StaticServices;
|
||||
|
||||
using System.Net;
|
||||
using System.Net.Http.Headers;
|
||||
using System.Text;
|
||||
using System.Text.Json.Nodes;
|
||||
using PrivaPub.Federation.Actors;
|
||||
using PrivaPub.Federation.Signing;
|
||||
|
||||
namespace PrivaPub.Federation.Outbox
|
||||
{
|
||||
public interface IDeliveryService
|
||||
{
|
||||
Task Enqueue(LocalActor signer, IEnumerable<string> inboxes, JsonObject activity, CancellationToken token);
|
||||
Task EnqueueToFollowers(LocalActor signer, JsonObject activity, CancellationToken token, IEnumerable<string> extraInboxes = default);
|
||||
Task<IReadOnlyList<string>> FollowerInboxes(LocalActor actor, CancellationToken token);
|
||||
}
|
||||
|
||||
public class DeliveryService : IDeliveryService
|
||||
{
|
||||
readonly DbEntities _dbEntities;
|
||||
|
||||
public DeliveryService(DbEntities dbEntities)
|
||||
{
|
||||
_dbEntities = dbEntities;
|
||||
}
|
||||
|
||||
public async Task Enqueue(LocalActor signer, IEnumerable<string> inboxes, JsonObject activity, CancellationToken token)
|
||||
{
|
||||
var body = activity.ToJsonString();
|
||||
var deliveries = inboxes
|
||||
.Where(i => !string.IsNullOrEmpty(i) && !i.StartsWith(signer.BaseAddress + "/", StringComparison.OrdinalIgnoreCase))
|
||||
.Distinct(StringComparer.Ordinal)
|
||||
.Select(inbox => new Delivery
|
||||
{
|
||||
SignerId = signer.Id,
|
||||
SignerKind = signer.Kind,
|
||||
InboxURL = inbox,
|
||||
Body = body
|
||||
})
|
||||
.ToList();
|
||||
if (deliveries.Count > 0)
|
||||
await DB.Default.SaveAsync(deliveries, token);
|
||||
}
|
||||
|
||||
public async Task EnqueueToFollowers(LocalActor signer, JsonObject activity, CancellationToken token, IEnumerable<string> extraInboxes = default)
|
||||
{
|
||||
var inboxes = (await FollowerInboxes(signer, token)).Concat(extraInboxes ?? Enumerable.Empty<string>());
|
||||
await Enqueue(signer, inboxes, activity, token);
|
||||
}
|
||||
|
||||
public async Task<IReadOnlyList<string>> FollowerInboxes(LocalActor actor, CancellationToken token)
|
||||
{
|
||||
var followers = await _dbEntities.Followers
|
||||
.Match(f => f.LocalActorId == actor.Id && f.LocalActorKind == actor.Kind && f.IsAccepted)
|
||||
.ExecuteAsync(token);
|
||||
return followers.Select(f => string.IsNullOrEmpty(f.SharedInboxURL) ? f.InboxURL : f.SharedInboxURL)
|
||||
.Distinct(StringComparer.Ordinal)
|
||||
.ToList();
|
||||
}
|
||||
}
|
||||
|
||||
public class DeliveryWorker : BackgroundService
|
||||
{
|
||||
const int MaxAttempts = 8;
|
||||
static readonly TimeSpan Poll = TimeSpan.FromSeconds(3);
|
||||
|
||||
readonly IServiceProvider _services;
|
||||
readonly IHttpClientFactory _httpClientFactory;
|
||||
readonly ILogger<DeliveryWorker> _logger;
|
||||
|
||||
public DeliveryWorker(IServiceProvider services, IHttpClientFactory httpClientFactory, ILogger<DeliveryWorker> logger)
|
||||
{
|
||||
_services = services;
|
||||
_httpClientFactory = httpClientFactory;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await DeliverDue(stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
return;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "{Worker} pass failed", nameof(DeliveryWorker));
|
||||
}
|
||||
await Task.Delay(Poll, stoppingToken);
|
||||
}
|
||||
}
|
||||
|
||||
async Task DeliverDue(CancellationToken token)
|
||||
{
|
||||
using var scope = _services.CreateScope();
|
||||
var dbEntities = scope.ServiceProvider.GetRequiredService<DbEntities>();
|
||||
var actors = scope.ServiceProvider.GetRequiredService<ILocalActorService>();
|
||||
var now = DateTime.UtcNow;
|
||||
var due = await dbEntities.Deliveries
|
||||
.Match(d => !d.DeliveredAt.HasValue && !d.AbandonedAt.HasValue && d.NextAttemptAt <= now)
|
||||
.Sort(d => d.NextAttemptAt, Order.Ascending)
|
||||
.Limit(20)
|
||||
.ExecuteAsync(token);
|
||||
|
||||
foreach (var delivery in due)
|
||||
{
|
||||
var signer = await actors.FindById(delivery.SignerKind, delivery.SignerId, token);
|
||||
if (signer == default)
|
||||
{
|
||||
delivery.AbandonedAt = DateTime.UtcNow;
|
||||
delivery.LastError = "the signing actor no longer exists";
|
||||
await DB.Default.SaveAsync(delivery, token);
|
||||
continue;
|
||||
}
|
||||
await Deliver(delivery, signer, token);
|
||||
await DB.Default.SaveAsync(delivery, token);
|
||||
}
|
||||
}
|
||||
|
||||
async Task Deliver(Delivery delivery, LocalActor signer, CancellationToken token)
|
||||
{
|
||||
delivery.Attempts++;
|
||||
try
|
||||
{
|
||||
if (!Uri.TryCreate(delivery.InboxURL, UriKind.Absolute, out var inbox) || !RemoteActorService.IsFetchable(inbox))
|
||||
{
|
||||
delivery.AbandonedAt = DateTime.UtcNow;
|
||||
delivery.LastError = "not a deliverable inbox";
|
||||
return;
|
||||
}
|
||||
|
||||
var body = Encoding.UTF8.GetBytes(delivery.Body);
|
||||
using var request = new HttpRequestMessage(HttpMethod.Post, inbox)
|
||||
{
|
||||
Content = new ByteArrayContent(body)
|
||||
};
|
||||
request.Content.Headers.ContentType = MediaTypeHeaderValue.Parse(RemoteActorService.ActivityJson);
|
||||
HttpSignatures.Sign(request, signer, body);
|
||||
|
||||
using var response = await _httpClientFactory.CreateClient(RemoteActorService.HttpClientName).SendAsync(request, token);
|
||||
if (response.IsSuccessStatusCode)
|
||||
{
|
||||
delivery.DeliveredAt = DateTime.UtcNow;
|
||||
delivery.LastError = default;
|
||||
return;
|
||||
}
|
||||
|
||||
delivery.LastError = $"{(int)response.StatusCode} {response.ReasonPhrase}";
|
||||
if (response.StatusCode is HttpStatusCode.Gone or HttpStatusCode.NotFound or HttpStatusCode.BadRequest or HttpStatusCode.Forbidden)
|
||||
{
|
||||
delivery.AbandonedAt = DateTime.UtcNow;
|
||||
_logger.LogWarning("Delivery {Id} to {Inbox} abandoned: {Status}", delivery.ID, delivery.InboxURL, delivery.LastError);
|
||||
return;
|
||||
}
|
||||
}
|
||||
catch (Exception ex) when (ex is HttpRequestException or TaskCanceledException && !token.IsCancellationRequested)
|
||||
{
|
||||
delivery.LastError = ex.Message;
|
||||
}
|
||||
|
||||
if (delivery.Attempts >= MaxAttempts)
|
||||
{
|
||||
delivery.AbandonedAt = DateTime.UtcNow;
|
||||
_logger.LogWarning("Delivery {Id} to {Inbox} abandoned after {Attempts} attempts: {Error}",
|
||||
delivery.ID, delivery.InboxURL, delivery.Attempts, delivery.LastError);
|
||||
return;
|
||||
}
|
||||
delivery.NextAttemptAt = DateTime.UtcNow.AddMinutes(Math.Pow(2, delivery.Attempts));
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user