M3: what each handler did with an activity
Arrival carries a verdict that handlers set with one line before an existing return (Arrival.Drop, Reject, Accept, About), so no handler signature changes. InboxProcessor records it as an 'in' event, with the time taken, the wait since the inbox accepted it, the attempt, the audience (from the post's visibility, or else the activity's addressing), the local actor kind, the object's age for updates, deletes and reactions, and the FEP features of a delivered object. It also records types nothing handles (unknown-type), actors that cannot be loaded (deferred) and handler failures. Drop reasons: fetch-failed, unparseable, misattributed, duplicate, deleted, not-addressed, not-followed, not-public, not-visible, not-deleted, unknown-object, unknown-recipient, cross-origin, unsupported. Accepted sub-reasons: stored, poll-vote, edit, refresh, actor-refresh, actor-delete, removed, tombstone-only, auto-accepted, pending, follow-answer, quote-answer, quote-granted, undone, reaction, reported. Rejected: blocked, ignored, quote-refused. A circle's traffic is private and kindless, and a stranger posting into one is just not-addressed, so the event store cannot reveal a circle. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ELjqpznMFMNrJoJUj6K5p2
This commit is contained in:
1 parent
4fa53f63bd
commit
ac7cbd136b
20 files changed
+544
-26
No files matched your search
@@ -2,6 +2,7 @@ using MongoDB.Driver;
|
||||
using MongoDB.Entities;
|
||||
|
||||
using PrivaPub.Federation.Actors;
|
||||
using PrivaPub.Federation.Inbox;
|
||||
using PrivaPub.Federation.Inbox.Handlers;
|
||||
using PrivaPub.Federation.Objects;
|
||||
using PrivaPub.Federation.Outbox;
|
||||
@@ -43,7 +44,10 @@ namespace PrivaPub.Domain.Social
|
||||
public async Task Receive(PostEntity post, ForeignAvatar actor, string content, JsonNode tags, string activityId, CancellationToken token)
|
||||
{
|
||||
if (!LikeHandler.MaySee(post, actor) || Normalise(content, ObjectShapes.Emojis(tags)) is not { } emoji)
|
||||
{
|
||||
Arrival.Drop(LikeHandler.MaySee(post, actor) ? "unparseable" : "not-visible");
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
await DB.Default.SaveAsync(new Reaction
|
||||
@@ -53,8 +57,10 @@ namespace PrivaPub.Domain.Social
|
||||
}
|
||||
catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
|
||||
{
|
||||
Arrival.Drop("duplicate");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("reaction");
|
||||
if (!post.IsFederatedCopy)
|
||||
await Notifications.Add(post.GroupUserId, NotificationType.Reaction, actor.ID, actor.ActorURI, post.ID, token, emoji.Name, emoji.Url);
|
||||
}
|
||||
|
||||
@@ -107,11 +107,18 @@ namespace PrivaPub.Domain.Statuses
|
||||
var quotingUri = Id(instrument);
|
||||
if (objectUri == default || quotingUri == default || !Origin.Same(quotingUri, actor.ActorURI)
|
||||
|| instrument is JsonObject embedded && Id(embedded["attributedTo"]) is { } by && by != actor.ActorURI)
|
||||
{
|
||||
Arrival.Drop("misattributed");
|
||||
return;
|
||||
}
|
||||
var quoted = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && !p.IsFederatedCopy && !p.DeletedAt.HasValue).ExecuteFirstAsync(token);
|
||||
var author = quoted == default ? default : await _localActors.FindById(LocalActorKind.Person, quoted.GroupUserId, token);
|
||||
if (author == default || string.IsNullOrEmpty(actor.InboxURL))
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
Arrival.About("Note", quoted.Visibility, quoted.CreationDate);
|
||||
var answer = new JsonObject
|
||||
{
|
||||
["@context"] = ActivityPubRenderer.Context(),
|
||||
@@ -121,12 +128,14 @@ namespace PrivaPub.Domain.Statuses
|
||||
};
|
||||
if (await MayQuote(quoted, actor.ActorURI, token))
|
||||
{
|
||||
Arrival.Accept("quote-granted");
|
||||
answer["type"] = "Accept";
|
||||
answer["result"] = await Grant(author, quoted, quotingUri, actor.ActorURI, token);
|
||||
answer["id"] = author.ActivityUri($"accept-quote-{Guid.NewGuid():N}");
|
||||
}
|
||||
else
|
||||
{
|
||||
Arrival.Reject("quote-refused");
|
||||
answer["type"] = "Reject";
|
||||
answer["id"] = author.ActivityUri($"reject-quote-{Guid.NewGuid():N}");
|
||||
}
|
||||
|
||||
@@ -1,3 +1,8 @@
|
||||
using PrivaPub.Federation.Actors;
|
||||
using PrivaPub.Infrastructure.Statistics;
|
||||
using PrivaPub.Models.Federation;
|
||||
using PrivaPub.Models.Post;
|
||||
|
||||
namespace PrivaPub.Federation.Inbox
|
||||
{
|
||||
public sealed record Arrival(string ActivityId, string ActivityType, string ActorURI, string Inbox, string KeyId, string Algorithm,
|
||||
@@ -10,5 +15,64 @@ namespace PrivaPub.Federation.Inbox
|
||||
get => current.Value;
|
||||
set => current.Value = value;
|
||||
}
|
||||
|
||||
public ArrivalVerdict Verdict { get; } = new();
|
||||
|
||||
public static void Drop(string reason) => Current?.Verdict.Drop(reason);
|
||||
|
||||
public static void Reject(string reason) => Current?.Verdict.Reject(reason);
|
||||
|
||||
public static void Accept(string reason) => Current?.Verdict.Accept(reason);
|
||||
|
||||
public static void About(string objectType = default, PostVisibility? visibility = default, DateTime? created = default, LocalActor local = default) =>
|
||||
Current?.Verdict.About(objectType, visibility, created, local);
|
||||
}
|
||||
|
||||
public sealed class ArrivalVerdict
|
||||
{
|
||||
public string Outcome { get; private set; }
|
||||
public string Reason { get; private set; }
|
||||
public string ObjectType { get; private set; }
|
||||
public string Audience { get; private set; }
|
||||
public string LocalKind { get; private set; }
|
||||
public long? AgeSeconds { get; private set; }
|
||||
|
||||
public void Drop(string reason) => Decide(Interactions.Dropped, reason);
|
||||
|
||||
public void Reject(string reason) => Decide(Interactions.Rejected, reason);
|
||||
|
||||
public void Accept(string reason) => Decide(Interactions.Accepted, reason);
|
||||
|
||||
void Decide(string outcome, string reason)
|
||||
{
|
||||
if (Outcome != default)
|
||||
return;
|
||||
Outcome = outcome;
|
||||
Reason = reason;
|
||||
}
|
||||
|
||||
public void About(string objectType, PostVisibility? visibility, DateTime? created, LocalActor local)
|
||||
{
|
||||
ObjectType = objectType ?? ObjectType;
|
||||
if (visibility is { } known)
|
||||
Audience = AudienceOf(known);
|
||||
if (created is { } at)
|
||||
AgeSeconds = Math.Max(0, (long)(DateTime.UtcNow - at).TotalSeconds);
|
||||
if (local != default)
|
||||
LocalKind = local.Kind switch
|
||||
{
|
||||
LocalActorKind.Person => "person",
|
||||
LocalActorKind.Group when !local.IsCircle => "group",
|
||||
LocalActorKind.Application => "application",
|
||||
_ => default
|
||||
};
|
||||
}
|
||||
|
||||
public static string AudienceOf(PostVisibility visibility) => visibility switch
|
||||
{
|
||||
PostVisibility.Public => Interactions.Public,
|
||||
PostVisibility.Unlisted => Interactions.Unlisted,
|
||||
_ => Interactions.Private
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -30,10 +30,18 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
public async Task Handle(JsonNode activity, ForeignAvatar actor, CancellationToken token)
|
||||
{
|
||||
if (await _quotes.Answered(activity, actor, accepted: Type == "Accept", token))
|
||||
{
|
||||
Arrival.Accept("quote-answer");
|
||||
return;
|
||||
}
|
||||
var following = await FindFollowing(activity["object"], actor, _dbEntities, _localActors, token);
|
||||
if (following == default)
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
Arrival.About("Follow");
|
||||
Arrival.Accept("follow-answer");
|
||||
await Apply(following, token);
|
||||
}
|
||||
|
||||
|
||||
@@ -50,19 +50,37 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
var objectUri = Id(inner);
|
||||
var announceId = Id(activity);
|
||||
if (objectUri == default || announceId == default)
|
||||
{
|
||||
Arrival.Drop("unparseable");
|
||||
return;
|
||||
}
|
||||
if (await _dbEntities.Posts.Match(p => p.ObjectURI == announceId).ExecuteAnyAsync(token))
|
||||
{
|
||||
Arrival.Drop("duplicate");
|
||||
return;
|
||||
}
|
||||
|
||||
var isLocal = objectUri.StartsWith(_localActors.BaseAddress + "/", StringComparison.OrdinalIgnoreCase);
|
||||
var original = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && !p.DeletedAt.HasValue && p.ReblogOfPostId == null).ExecuteFirstAsync(token);
|
||||
var followed = await _dbEntities.Followings.Match(f => f.TargetActorURI == actor.ActorURI && f.State == FollowState.Accepted).ExecuteAnyAsync(token);
|
||||
if (original is not { IsFederatedCopy: false } && !followed)
|
||||
{
|
||||
Arrival.Drop("not-followed");
|
||||
return;
|
||||
}
|
||||
if (original == default && !isLocal)
|
||||
original = await _remotePosts.StoreContext(objectUri, 0, token);
|
||||
if (original == default || original.Visibility is not (PostVisibility.Public or PostVisibility.Unlisted))
|
||||
if (original == default)
|
||||
{
|
||||
Arrival.Drop("fetch-failed");
|
||||
return;
|
||||
}
|
||||
Arrival.About(original.ObjectType ?? "Note", original.Visibility, original.CreationDate);
|
||||
if (original.Visibility is not (PostVisibility.Public or PostVisibility.Unlisted))
|
||||
{
|
||||
Arrival.Drop("not-public");
|
||||
return;
|
||||
}
|
||||
|
||||
var to = Strings(activity["to"]);
|
||||
var cc = Strings(activity["cc"]);
|
||||
@@ -85,15 +103,21 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
UpdateDate = published
|
||||
};
|
||||
if (reblog.Visibility == PostVisibility.Direct)
|
||||
{
|
||||
Arrival.Drop("not-public");
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
await DB.Default.SaveAsync(reblog, token);
|
||||
}
|
||||
catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
|
||||
{
|
||||
Arrival.Drop("duplicate");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("stored");
|
||||
Arrival.About(local: original.IsFederatedCopy ? default : await _localActors.FindById(Models.Federation.LocalActorKind.Person, original.GroupUserId, token));
|
||||
|
||||
await DB.Default.Update<PostEntity>().MatchID(original.ID).Modify(b => b.Inc(p => p.ReblogsCount, 1)).ExecuteAsync(token);
|
||||
if (!original.IsFederatedCopy)
|
||||
@@ -103,18 +127,30 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
|
||||
async Task GroupActivity(JsonNode inner, ForeignAvatar group, CancellationToken token)
|
||||
{
|
||||
Arrival.About(Value(inner, "type"));
|
||||
if (group.AvatarType != AvatarType.Group
|
||||
|| !await _dbEntities.Followings.Match(f => f.TargetActorURI == group.ActorURI && f.State == FollowState.Accepted).ExecuteAnyAsync(token))
|
||||
{
|
||||
Arrival.Drop("not-followed");
|
||||
return;
|
||||
}
|
||||
var objectUri = Id(inner["object"]);
|
||||
switch (Value(inner, "type"))
|
||||
{
|
||||
case "Create" when objectUri != default:
|
||||
if (await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri).ExecuteAnyAsync(token))
|
||||
{
|
||||
Arrival.Drop("duplicate");
|
||||
return;
|
||||
}
|
||||
var post = await _remotePosts.StoreContext(objectUri, 0, token);
|
||||
if (post == default)
|
||||
{
|
||||
Arrival.Drop("fetch-failed");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("stored");
|
||||
Arrival.About(visibility: post.Visibility);
|
||||
await DB.Default.Update<PostEntity>().MatchID(post.ID).Modify(p => p.AudienceURI, group.ActorURI).ExecuteAsync(token);
|
||||
post.AudienceURI = group.ActorURI;
|
||||
await _fanout.Distribute(post, token);
|
||||
@@ -123,12 +159,20 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
var stored = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && p.AudienceURI == group.ActorURI && !p.DeletedAt.HasValue)
|
||||
.ExecuteFirstAsync(token);
|
||||
if (stored == default)
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
Arrival.About(visibility: stored.Visibility, created: stored.CreationDate);
|
||||
using (var fetched = await _remoteActors.FetchObject(objectUri, token))
|
||||
{
|
||||
var note = fetched == default ? default : NoteParser.Parse(System.Text.Json.Nodes.JsonNode.Parse(fetched.Root.GetRawText()));
|
||||
if (note == default || note.AttributedTo != stored.ActorURI)
|
||||
{
|
||||
Arrival.Drop(note == default ? "fetch-failed" : "misattributed");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("edit");
|
||||
stored.Revisions.Add(new PostRevision
|
||||
{
|
||||
Title = stored.Title,
|
||||
@@ -151,13 +195,24 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
case "Delete" when objectUri != default:
|
||||
var deleted = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && p.AudienceURI == group.ActorURI).ExecuteFirstAsync(token);
|
||||
if (deleted == default)
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
Arrival.About(visibility: deleted.Visibility, created: deleted.CreationDate);
|
||||
using (var check = await _remoteActors.FetchObject(objectUri, token))
|
||||
if (check != default && Value(System.Text.Json.Nodes.JsonNode.Parse(check.Root.GetRawText()), "type") != "Tombstone")
|
||||
{
|
||||
Arrival.Drop("not-deleted");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("removed");
|
||||
await DB.Default.DeleteAsync<PostEntity>(deleted.ID);
|
||||
await DB.Default.DeleteAsync<TimelineEntry>(e => e.PostId == deleted.ID);
|
||||
break;
|
||||
default:
|
||||
Arrival.Drop("unsupported");
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -68,26 +68,42 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
{
|
||||
using var fetched = await _remoteActors.FetchObject(Id(node), token);
|
||||
node = fetched == default ? default : JsonNode.Parse(fetched.Root.GetRawText());
|
||||
if (node == default)
|
||||
Arrival.Drop("fetch-failed");
|
||||
}
|
||||
var note = NoteParser.Parse(node);
|
||||
if (note == default || note.AttributedTo != author.ActorURI)
|
||||
{
|
||||
Arrival.Drop(note == default ? "unparseable" : "misattributed");
|
||||
return;
|
||||
}
|
||||
Arrival.About(note.Type);
|
||||
if (note.Title != default && note.InReplyTo != default && ObjectShapes.Text(note.ContentHtml, 1) == default)
|
||||
{
|
||||
var question = await _dbEntities.Posts.Match(p => p.ObjectURI == note.InReplyTo && p.Poll != null).ExecuteFirstAsync(token);
|
||||
if (question != default)
|
||||
{
|
||||
await _polls.Receive(question, author, note.Title, Id(activity), token);
|
||||
Arrival.Accept("poll-vote");
|
||||
Arrival.About(visibility: question.Visibility);
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (await _dbEntities.Posts.Match(p => p.ObjectURI == note.Id).ExecuteAnyAsync(token)
|
||||
|| await DB.Default.Find<DeletedObject>().Match(d => d.ObjectURI == note.Id).ExecuteAnyAsync(token))
|
||||
if (await _dbEntities.Posts.Match(p => p.ObjectURI == note.Id).ExecuteAnyAsync(token))
|
||||
{
|
||||
Arrival.Drop("duplicate");
|
||||
return;
|
||||
}
|
||||
if (await DB.Default.Find<DeletedObject>().Match(d => d.ObjectURI == note.Id).ExecuteAnyAsync(token))
|
||||
{
|
||||
Arrival.Drop("deleted");
|
||||
return;
|
||||
}
|
||||
|
||||
var to = note.To.Concat(Strings(activity["to"])).Distinct(StringComparer.Ordinal).ToList();
|
||||
var cc = note.Cc.Concat(Strings(activity["cc"])).Distinct(StringComparer.Ordinal).ToList();
|
||||
var visibility = Addressing.Classify(to, cc, author.FollowersURL);
|
||||
Arrival.About(visibility: visibility);
|
||||
|
||||
var addressed = to.Concat(cc)
|
||||
.Concat(Addresses(activity))
|
||||
@@ -111,9 +127,13 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
var circle = localTargets.FirstOrDefault(t => t is { Kind: LocalActorKind.Group, IsCircle: true });
|
||||
if (circle != default)
|
||||
{
|
||||
if (!await IsCircleMember(circle, author.ActorURI, token))
|
||||
return;
|
||||
visibility = PostVisibility.Circle;
|
||||
Arrival.About(visibility: visibility);
|
||||
if (!await IsCircleMember(circle, author.ActorURI, token))
|
||||
{
|
||||
Arrival.Drop("not-addressed");
|
||||
return;
|
||||
}
|
||||
}
|
||||
var group = circle ?? (visibility is PostVisibility.Public or PostVisibility.Unlisted
|
||||
? localTargets.FirstOrDefault(t => t is { Kind: LocalActorKind.Group, IsCircle: false })
|
||||
@@ -126,7 +146,10 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
.Match(f => f.TargetActorURI == author.ActorURI && f.State == FollowState.Accepted)
|
||||
.ExecuteAnyAsync(token);
|
||||
if (visibility == PostVisibility.Direct ? persons.Count == 0 : group == default && persons.Count == 0 && !repliesToLocal && !followed)
|
||||
{
|
||||
Arrival.Drop("not-addressed");
|
||||
return;
|
||||
}
|
||||
if (parent == default && visibility != PostVisibility.Direct)
|
||||
parent = await _remotePosts.Parent(note.InReplyTo, 1, token);
|
||||
|
||||
@@ -150,8 +173,11 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
}
|
||||
catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
|
||||
{
|
||||
Arrival.Drop("duplicate");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("stored");
|
||||
Arrival.About(local: group ?? persons.FirstOrDefault());
|
||||
await _records.Record(note, post, ObjectPath.Delivered, refetched, token);
|
||||
await _previews.Wanted(post, token);
|
||||
await _quotes.Resolve(post, note, token);
|
||||
|
||||
@@ -49,10 +49,15 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
|
||||
var objectUri = Id(delete["object"]);
|
||||
if (!Origin.Same(objectUri, actor.ActorURI))
|
||||
{
|
||||
Arrival.Drop("cross-origin");
|
||||
return;
|
||||
}
|
||||
|
||||
if (objectUri == actor.ActorURI)
|
||||
{
|
||||
Arrival.Accept("actor-delete");
|
||||
Arrival.About(actor.AvatarType.ToString(), created: actor.Published);
|
||||
actor.DeletionAt = DateTime.UtcNow;
|
||||
actor.AccountState = AvatarAccountState.Deleted;
|
||||
await DB.Default.SaveAsync(actor, token);
|
||||
@@ -76,9 +81,12 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
var post = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && p.ActorURI == actor.ActorURI).ExecuteFirstAsync(token);
|
||||
if (post == default)
|
||||
{
|
||||
Arrival.Accept("tombstone-only");
|
||||
await _quotes.Revoke(objectUri, token);
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("removed");
|
||||
Arrival.About(post.ObjectType, post.Visibility, post.CreationDate);
|
||||
await DB.Default.DeleteAsync<PostEntity>(post.ID);
|
||||
await DB.Default.DeleteAsync<ObjectRecord>(r => r.PostId == post.ID || r.ObjectURI == objectUri);
|
||||
await DB.Default.DeleteAsync<TimelineEntry>(e => e.PostId == post.ID || e.ReblogOfPostId == post.ID);
|
||||
|
||||
@@ -31,7 +31,11 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
? default
|
||||
: await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && !p.DeletedAt.HasValue && p.ReblogOfPostId == null).ExecuteFirstAsync(token);
|
||||
if (post == default || !LikeHandler.MaySee(post, actor))
|
||||
{
|
||||
Arrival.Drop(post == default ? "unknown-object" : "not-visible");
|
||||
return;
|
||||
}
|
||||
Arrival.About(post.ObjectType ?? "Note", post.Visibility, post.CreationDate);
|
||||
|
||||
try
|
||||
{
|
||||
@@ -39,8 +43,10 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
}
|
||||
catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
|
||||
{
|
||||
Arrival.Drop("duplicate");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("stored");
|
||||
await DB.Default.Update<PostEntity>().MatchID(post.ID).Modify(b => b.Inc(p => p.DownvotesCount, 1)).ExecuteAsync(token);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,8 +27,13 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
var post = objectUri == default
|
||||
? default
|
||||
: await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && !p.DeletedAt.HasValue && p.ReblogOfPostId == null).ExecuteFirstAsync(token);
|
||||
if (post != default)
|
||||
await _reactions.Receive(post, actor, Value(activity, "content"), activity["tag"], Id(activity), token);
|
||||
if (post == default)
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
Arrival.About(post.ObjectType ?? "Note", post.Visibility, post.CreationDate);
|
||||
await _reactions.Receive(post, actor, Value(activity, "content"), activity["tag"], Id(activity), token);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -50,7 +50,11 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
if (target == default && posts.Count > 0)
|
||||
target = await _localActors.FindById(LocalActorKind.Person, posts[0].GroupUserId, token);
|
||||
if (target == default)
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("reported");
|
||||
|
||||
await DB.Default.SaveAsync(new Report
|
||||
{
|
||||
|
||||
@@ -45,7 +45,10 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
|
||||
var target = await _localActors.FindByUri(Id(follow["object"]), token);
|
||||
if (target is not { IsFederated: true } || target.Kind == LocalActorKind.Application)
|
||||
{
|
||||
Arrival.Drop("unknown-recipient");
|
||||
return;
|
||||
}
|
||||
if (target.Kind == LocalActorKind.Person
|
||||
&& await DB.Default.Find<Models.Social.Block>().Match(b => b.AvatarId == target.Id && b.TargetActorURI == follower.ActorURI).ExecuteAnyAsync(token))
|
||||
{
|
||||
@@ -58,6 +61,7 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
["object"] = follow.DeepClone()
|
||||
};
|
||||
await _delivery.Enqueue(target, new[] { follower.InboxURL }, reject, token);
|
||||
Arrival.Reject("blocked");
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -75,6 +79,7 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
record.FollowActivityURI = Id(follow);
|
||||
record.IsAccepted = existing?.IsAccepted == true || !target.ManuallyApprovesFollowers;
|
||||
await DB.Default.SaveAsync(record, token);
|
||||
Arrival.Accept(record.IsAccepted ? "auto-accepted" : "pending");
|
||||
|
||||
if (target.Kind == LocalActorKind.Person)
|
||||
await Notifications.Add(target.Id, record.IsAccepted ? NotificationType.Follow : NotificationType.FollowRequest,
|
||||
|
||||
@@ -30,11 +30,19 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
{
|
||||
var objectUri = Id(activity["object"]);
|
||||
if (objectUri == default || !objectUri.StartsWith(_localActors.BaseAddress + "/", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
var post = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && !p.IsFederatedCopy && !p.DeletedAt.HasValue).ExecuteFirstAsync(token);
|
||||
var owner = post == default ? default : await _localActors.FindById(LocalActorKind.Person, post.GroupUserId, token);
|
||||
if (owner == default || string.IsNullOrEmpty(actor.InboxURL))
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
Arrival.About(post.ObjectType ?? "Note", post.Visibility, post.CreationDate);
|
||||
Arrival.Reject("ignored");
|
||||
await _delivery.Enqueue(owner, new[] { actor.InboxURL }, new JsonObject
|
||||
{
|
||||
["@context"] = ActivityPubRenderer.ActivityStreams,
|
||||
|
||||
@@ -35,7 +35,11 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
? default
|
||||
: await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && !p.DeletedAt.HasValue && p.ReblogOfPostId == null).ExecuteFirstAsync(token);
|
||||
if (post == default || !MaySee(post, actor))
|
||||
{
|
||||
Arrival.Drop(post == default ? "unknown-object" : "not-visible");
|
||||
return;
|
||||
}
|
||||
Arrival.About(post.ObjectType ?? "Note", post.Visibility, post.CreationDate);
|
||||
var content = Value(activity, "_misskey_reaction") ?? Value(activity, "content");
|
||||
if (!Reactions.IsHeart(content))
|
||||
{
|
||||
@@ -49,8 +53,10 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
}
|
||||
catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
|
||||
{
|
||||
Arrival.Drop("duplicate");
|
||||
return;
|
||||
}
|
||||
Arrival.Accept("stored");
|
||||
await DB.Default.Update<PostEntity>().MatchID(post.ID).Modify(b => b.Inc(p => p.FavouritesCount, 1)).ExecuteAsync(token);
|
||||
if (!post.IsFederatedCopy)
|
||||
await Notifications.Add(post.GroupUserId, NotificationType.Favourite, actor.ID, actor.ActorURI, post.ID, token);
|
||||
|
||||
@@ -37,20 +37,26 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
var innerType = inner is JsonObject ? Value(inner, "type") : default;
|
||||
var innerId = Id(inner);
|
||||
|
||||
Arrival.About(innerType);
|
||||
var undone = false;
|
||||
if (innerType is null or "Follow")
|
||||
await UndoFollow(inner, innerId, actor, token);
|
||||
undone |= await UndoFollow(inner, innerId, actor, token);
|
||||
if (innerType is null or "Like")
|
||||
await UndoLike(inner, innerId, actor, token);
|
||||
undone |= await UndoLike(inner, innerId, actor, token);
|
||||
if (innerType is null or "Announce")
|
||||
await UndoAnnounce(innerId, actor, token);
|
||||
undone |= await UndoAnnounce(innerId, actor, token);
|
||||
if (innerType is null or "Dislike")
|
||||
await UndoDislike(inner, innerId, actor, token);
|
||||
undone |= await UndoDislike(inner, innerId, actor, token);
|
||||
if (innerType is null or "Like" or "EmojiReact")
|
||||
await _reactions.Withdraw(actor, innerId, inner is JsonObject ? Id(inner["object"]) : default,
|
||||
undone |= await _reactions.Withdraw(actor, innerId, inner is JsonObject ? Id(inner["object"]) : default,
|
||||
inner is JsonObject ? Value(inner, "_misskey_reaction") ?? Value(inner, "content") : default, token);
|
||||
if (undone)
|
||||
Arrival.Accept("undone");
|
||||
else
|
||||
Arrival.Drop("unknown-object");
|
||||
}
|
||||
|
||||
async Task UndoDislike(JsonNode inner, string innerId, ForeignAvatar actor, CancellationToken token)
|
||||
async Task<bool> UndoDislike(JsonNode inner, string innerId, ForeignAvatar actor, CancellationToken token)
|
||||
{
|
||||
var downvote = innerId == default
|
||||
? default
|
||||
@@ -63,16 +69,18 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
: await DB.Default.Find<Downvote>().Match(d => d.PostId == post.ID && d.ActorURI == actor.ActorURI).ExecuteFirstAsync(token);
|
||||
}
|
||||
if (downvote == default)
|
||||
return;
|
||||
return false;
|
||||
await DB.Default.DeleteAsync<Downvote>(downvote.ID);
|
||||
await DB.Default.Update<PostEntity>().MatchID(downvote.PostId).Modify(b => b.Inc(p => p.DownvotesCount, -1)).ExecuteAsync(token);
|
||||
return true;
|
||||
}
|
||||
|
||||
async Task UndoFollow(JsonNode inner, string innerId, ForeignAvatar actor, CancellationToken token)
|
||||
async Task<bool> UndoFollow(JsonNode inner, string innerId, ForeignAvatar actor, CancellationToken token)
|
||||
{
|
||||
var followers = await _dbEntities.Followers.Match(f => f.ActorURI == actor.ActorURI).ExecuteAsync(token);
|
||||
var targetUri = inner is JsonObject ? Id(inner["object"]) : default;
|
||||
var target = targetUri == default ? default : await _localActors.FindByUri(targetUri, token);
|
||||
var undone = false;
|
||||
foreach (var follower in followers)
|
||||
{
|
||||
var matchesActivity = innerId != default && follower.FollowActivityURI == innerId;
|
||||
@@ -83,10 +91,12 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
await DB.Default.DeleteAsync<Follower>(follower.ID);
|
||||
if (follower.LocalActorKind == LocalActorKind.Group)
|
||||
await RemoveForeignMember(follower.LocalActorId, actor.ActorURI, token);
|
||||
undone = true;
|
||||
}
|
||||
return undone;
|
||||
}
|
||||
|
||||
async Task UndoLike(JsonNode inner, string innerId, ForeignAvatar actor, CancellationToken token)
|
||||
async Task<bool> UndoLike(JsonNode inner, string innerId, ForeignAvatar actor, CancellationToken token)
|
||||
{
|
||||
var favourite = innerId == default
|
||||
? default
|
||||
@@ -99,23 +109,25 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
: await _dbEntities.Favourites.Match(f => f.PostId == post.ID && f.ActorURI == actor.ActorURI).ExecuteFirstAsync(token);
|
||||
}
|
||||
if (favourite == default)
|
||||
return;
|
||||
return false;
|
||||
await DB.Default.DeleteAsync<Favourite>(favourite.ID);
|
||||
await DB.Default.Update<PostEntity>().MatchID(favourite.PostId).Modify(b => b.Inc(p => p.FavouritesCount, -1)).ExecuteAsync(token);
|
||||
return true;
|
||||
}
|
||||
|
||||
async Task UndoAnnounce(string innerId, ForeignAvatar actor, CancellationToken token)
|
||||
async Task<bool> UndoAnnounce(string innerId, ForeignAvatar actor, CancellationToken token)
|
||||
{
|
||||
if (innerId == default)
|
||||
return;
|
||||
return false;
|
||||
var reblog = await _dbEntities.Posts
|
||||
.Match(p => p.ObjectURI == innerId && p.ActorURI == actor.ActorURI && p.ReblogOfPostId != null)
|
||||
.ExecuteFirstAsync(token);
|
||||
if (reblog == default)
|
||||
return;
|
||||
return false;
|
||||
await DB.Default.DeleteAsync<PostEntity>(reblog.ID);
|
||||
await DB.Default.DeleteAsync<TimelineEntry>(e => e.PostId == reblog.ID);
|
||||
await DB.Default.Update<PostEntity>().MatchID(reblog.ReblogOfPostId).Modify(b => b.Inc(p => p.ReblogsCount, -1)).ExecuteAsync(token);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -46,17 +46,29 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
if (Id(inner) == actor.ActorURI)
|
||||
{
|
||||
await _remoteActors.GetActor(actor.ActorURI, refresh: true, token);
|
||||
Arrival.Accept("actor-refresh");
|
||||
Arrival.About(actor.AvatarType.ToString());
|
||||
return;
|
||||
}
|
||||
if (inner is not JsonObject || !Origin.Same(Id(inner), actor.ActorURI))
|
||||
{
|
||||
Arrival.Drop("cross-origin");
|
||||
return;
|
||||
}
|
||||
var note = NoteParser.Parse(inner);
|
||||
if (note == default || note.AttributedTo != actor.ActorURI)
|
||||
{
|
||||
Arrival.Drop(note == default ? "unparseable" : "misattributed");
|
||||
return;
|
||||
}
|
||||
|
||||
var post = await _dbEntities.Posts.Match(p => p.ObjectURI == note.Id && p.ActorURI == actor.ActorURI).ExecuteFirstAsync(token);
|
||||
if (post == default || post.DeletedAt.HasValue)
|
||||
{
|
||||
Arrival.Drop("unknown-object");
|
||||
return;
|
||||
}
|
||||
Arrival.About(note.Type, post.Visibility, post.CreationDate);
|
||||
|
||||
post.Poll = note.Poll ?? post.Poll;
|
||||
post.QuotePolicy = note.QuotePolicy;
|
||||
@@ -65,12 +77,14 @@ namespace PrivaPub.Federation.Inbox.Handlers
|
||||
post.Event = note.Event ?? post.Event;
|
||||
if (!IsEdit(note, post))
|
||||
{
|
||||
Arrival.Accept("refresh");
|
||||
await DB.Default.SaveAsync(post, token);
|
||||
await _records.Revise(note, Id(activity), token);
|
||||
await Requote(post, note, token);
|
||||
return;
|
||||
}
|
||||
|
||||
Arrival.Accept("edit");
|
||||
post.Revisions.Add(new PostRevision
|
||||
{
|
||||
Title = post.Title,
|
||||
|
||||
@@ -1,8 +1,12 @@
|
||||
using PrivaPub.Federation.Actors;
|
||||
using PrivaPub.Federation.Objects;
|
||||
using PrivaPub.Infrastructure.Jobs;
|
||||
using PrivaPub.Infrastructure.Statistics;
|
||||
using PrivaPub.Models.Jobs;
|
||||
using PrivaPub.Models.Statistics;
|
||||
using PrivaPub.Models.User;
|
||||
|
||||
using System.Diagnostics;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Nodes;
|
||||
|
||||
@@ -21,12 +25,15 @@ namespace PrivaPub.Federation.Inbox
|
||||
readonly IRemoteActorService _remoteActors;
|
||||
readonly IReadOnlyDictionary<string, IActivityHandler> _handlers;
|
||||
readonly ILogger<InboxProcessor> _logger;
|
||||
readonly IInteractionLedger _ledger;
|
||||
|
||||
public InboxProcessor(IRemoteActorService remoteActors, IEnumerable<IActivityHandler> handlers, ILogger<InboxProcessor> logger)
|
||||
public InboxProcessor(IRemoteActorService remoteActors, IEnumerable<IActivityHandler> handlers, ILogger<InboxProcessor> logger,
|
||||
IInteractionLedger ledger = default)
|
||||
{
|
||||
_remoteActors = remoteActors;
|
||||
_handlers = handlers.ToDictionary(h => h.Type, StringComparer.Ordinal);
|
||||
_logger = logger;
|
||||
_ledger = ledger;
|
||||
}
|
||||
|
||||
public JobKind Kind => JobKind.ProcessInbox;
|
||||
@@ -36,28 +43,86 @@ namespace PrivaPub.Federation.Inbox
|
||||
|
||||
public async Task<JobOutcome> Handle(Job job, CancellationToken token)
|
||||
{
|
||||
var started = Stopwatch.GetTimestamp();
|
||||
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))
|
||||
{
|
||||
Record(job, payload, activity, type, started, new ArrivalVerdict(), Interactions.Dropped, "unknown-type");
|
||||
return JobOutcome.Done;
|
||||
}
|
||||
|
||||
var actor = await _remoteActors.GetActor(payload.ActorURI, refresh: false, token);
|
||||
if (actor == default)
|
||||
{
|
||||
Record(job, payload, activity, type, started, new ArrivalVerdict(), Interactions.Deferred, "actor-unavailable");
|
||||
return JobOutcome.Retry("the actor could not be loaded");
|
||||
}
|
||||
|
||||
Arrival.Current = new Arrival(Id(activity), type, actor.ActorURI, payload.Inbox, payload.KeyId, payload.Algorithm,
|
||||
var arrival = new Arrival(Id(activity), type, actor.ActorURI, payload.Inbox, payload.KeyId, payload.Algorithm,
|
||||
payload.SignedHeaders ?? Array.Empty<string>(), payload.ReceivedAt ?? job.CreatedAt, activity["@context"]?.ToJsonString());
|
||||
Arrival.Current = arrival;
|
||||
try
|
||||
{
|
||||
await handler.Handle(activity, actor, token);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
Record(job, payload, activity, type, started, arrival.Verdict, Interactions.Failed, ex.GetType().Name.ToLowerInvariant());
|
||||
throw;
|
||||
}
|
||||
finally
|
||||
{
|
||||
Arrival.Current = default;
|
||||
}
|
||||
Record(job, payload, activity, type, started, arrival.Verdict, default, default);
|
||||
_logger.LogInformation("Processed {Type} {Id} from {Actor}", type, Id(activity), actor.ActorURI);
|
||||
return JobOutcome.Done;
|
||||
}
|
||||
|
||||
void Record(Job job, InboxPayload payload, JsonNode activity, string type, long started, ArrivalVerdict verdict, string outcome, string reason)
|
||||
{
|
||||
if (_ledger == default || payload == default)
|
||||
return;
|
||||
var embedded = activity?["object"] as JsonObject;
|
||||
_ledger.Record(new InteractionEvent
|
||||
{
|
||||
Channel = Interactions.In,
|
||||
Host = Interactions.HostOf(payload.ActorURI),
|
||||
Activity = type,
|
||||
Object = verdict.ObjectType ?? Value(embedded, "type"),
|
||||
Outcome = outcome ?? verdict.Outcome ?? Interactions.Accepted,
|
||||
Reason = reason ?? verdict.Reason,
|
||||
LatencyMs = (int)Stopwatch.GetElapsedTime(started).TotalMilliseconds,
|
||||
WaitMs = (int)Math.Min(int.MaxValue, Math.Max(0, (DateTime.UtcNow - (payload.ReceivedAt ?? job.CreatedAt)).TotalMilliseconds)),
|
||||
Attempt = job.Attempts,
|
||||
Audience = verdict.Audience ?? Audience(activity),
|
||||
LocalKind = verdict.LocalKind,
|
||||
AgeSeconds = verdict.AgeSeconds,
|
||||
Inbox = payload.Inbox,
|
||||
Signature = payload.Algorithm == default ? default : "cavage:" + payload.Algorithm,
|
||||
Features = type is "Create" or "Update" && embedded != default ? ObjectFeatures.Detect(embedded) : default
|
||||
}, payload.ActorURI);
|
||||
}
|
||||
|
||||
static string Audience(JsonNode activity)
|
||||
{
|
||||
var source = activity?["to"] != default || activity?["cc"] != default ? activity : activity?["object"] as JsonObject;
|
||||
var to = Addresses(source?["to"]);
|
||||
var cc = Addresses(source?["cc"]);
|
||||
if (to.Any(Addressing.IsPublic))
|
||||
return Interactions.Public;
|
||||
if (cc.Any(Addressing.IsPublic))
|
||||
return Interactions.Unlisted;
|
||||
return to.Count + cc.Count == 0 ? Interactions.None : Interactions.Private;
|
||||
}
|
||||
|
||||
static List<string> Addresses(JsonNode node) => node switch
|
||||
{
|
||||
JsonArray array => array.Select(Id).Where(id => id != default).ToList(),
|
||||
JsonNode single when Id(single) is { } id => new List<string> { id },
|
||||
_ => new List<string>()
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -53,7 +53,7 @@ namespace PrivaPub.Infrastructure.Statistics
|
||||
e.LocalKind = e.Audience is Public or Unlisted && e.LocalKind != default && LocalKinds.Contains(e.LocalKind) ? e.LocalKind : default;
|
||||
e.Inbox = e.Inbox != default && Inboxes.Contains(e.Inbox) ? e.Inbox : default;
|
||||
e.Signature = e.Signature == default ? default : SignatureName().IsMatch(e.Signature) ? e.Signature : Other;
|
||||
e.Features = e.Features?.Select(Word).Where(f => f != default).Distinct().Take(32).ToList();
|
||||
e.Features = e.Features?.Select(Feature).Where(f => f != default).Distinct().Take(32).ToList();
|
||||
if (e.Features?.Count == 0)
|
||||
e.Features = default;
|
||||
return e;
|
||||
@@ -86,6 +86,9 @@ namespace PrivaPub.Infrastructure.Statistics
|
||||
static string Word(string value) =>
|
||||
value == default ? default : KebabWord().IsMatch(value) ? value : Other;
|
||||
|
||||
static string Feature(string value) =>
|
||||
value == default ? default : FeatureName().IsMatch(value) ? value : Other;
|
||||
|
||||
[GeneratedRegex("^[A-Za-z][A-Za-z0-9_-]{0,31}$")]
|
||||
private static partial Regex TypeWord();
|
||||
|
||||
@@ -95,6 +98,9 @@ namespace PrivaPub.Infrastructure.Statistics
|
||||
[GeneratedRegex("^[a-z0-9][a-z0-9:-]{0,39}$")]
|
||||
private static partial Regex SignatureName();
|
||||
|
||||
[GeneratedRegex("^[a-z0-9][a-z0-9:/+.-]{0,63}$")]
|
||||
private static partial Regex FeatureName();
|
||||
|
||||
[GeneratedRegex("^[a-z0-9]([a-z0-9.-]*[a-z0-9])?$")]
|
||||
private static partial Regex HostName();
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user