The inbox answers once it has verified, and processes from the job queue
InboxService is split the way the roadmap lays out Federation/Inbox:
- InboxReceiver reads and verifies the request exactly as before, runs
the checks that need no fetch (the activity id's origin, a Follow of a
missing or local-only actor, an Undo of someone else's activity, an
embedded object attributed to someone else), queues a ProcessInbox job
and answers 202. The job's dedupe key is the activity id, so a peer that
delivers the same activity twice is processed once.
- InboxProcessor (two at a time, eight attempts) loads the verified actor
and hands the activity to the handler for its type.
- Handlers/{Follow,Undo,Create,Delete,Update}Handler are the old methods,
unchanged except that they no longer produce status codes; the JSON
helpers live in Objects/ActivityJson and the group membership helpers in
Inbox/ForeignMembers.
A slow fetch of an object or a remote actor now delays the job, not the
sender's HTTP request.
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
d1a91c40c4
commit
796f06acf4
13 files changed
+755
-430
No files matched your search
@@ -0,0 +1,54 @@
|
||||
using PrivaPub.Federation.Actors;
|
||||
using PrivaPub.Infrastructure.Jobs;
|
||||
using PrivaPub.Models.Jobs;
|
||||
using PrivaPub.Models.User;
|
||||
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Nodes;
|
||||
|
||||
using static PrivaPub.Federation.Objects.ActivityJson;
|
||||
|
||||
namespace PrivaPub.Federation.Inbox
|
||||
{
|
||||
public interface IActivityHandler
|
||||
{
|
||||
string Type { get; }
|
||||
Task Handle(JsonNode activity, ForeignAvatar actor, CancellationToken token);
|
||||
}
|
||||
|
||||
public class InboxProcessor : IJobHandler
|
||||
{
|
||||
readonly IRemoteActorService _remoteActors;
|
||||
readonly IReadOnlyDictionary<string, IActivityHandler> _handlers;
|
||||
readonly ILogger<InboxProcessor> _logger;
|
||||
|
||||
public InboxProcessor(IRemoteActorService remoteActors, IEnumerable<IActivityHandler> handlers, ILogger<InboxProcessor> logger)
|
||||
{
|
||||
_remoteActors = remoteActors;
|
||||
_handlers = handlers.ToDictionary(h => h.Type, StringComparer.Ordinal);
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public JobKind Kind => JobKind.ProcessInbox;
|
||||
public int Concurrency => 2;
|
||||
public int MaxAttempts => 8;
|
||||
public int PerHostLimit => 2;
|
||||
|
||||
public async Task<JobOutcome> Handle(Job job, CancellationToken token)
|
||||
{
|
||||
var payload = JsonSerializer.Deserialize<InboxPayload>(job.Payload);
|
||||
var activity = payload == default ? default : JsonNode.Parse(payload.Activity);
|
||||
var type = Value(activity, "type");
|
||||
if (type == default || !_handlers.TryGetValue(type, out var handler))
|
||||
return JobOutcome.Done;
|
||||
|
||||
var actor = await _remoteActors.GetActor(payload.ActorURI, refresh: false, token);
|
||||
if (actor == default)
|
||||
return JobOutcome.Retry("the actor could not be loaded");
|
||||
|
||||
await handler.Handle(activity, actor, token);
|
||||
_logger.LogInformation("Processed {Type} {Id} from {Actor}", type, Id(activity), actor.ActorURI);
|
||||
return JobOutcome.Done;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user