Found by the first live run against GoToSocial: - A remote post's id came from its published second plus random bytes, so a reply arriving in the same second as the post it answers could sort under it, and a late arrival landed behind a client's since_id. A post published within the last hour now gets an id for its arrival; backfill keeps its published time. - NoteParser.Time returned default(DateTime) for a missing "updated", so every remote post was stored as edited in year one. Migration _006 clears the stored ones. - published now carries milliseconds, so peers that derive ids from it (GoToSocial, Mastodon) keep our posts in order within a second. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012CzABvBkbcFqoHdmi8b9WB
172 lines
6.5 KiB
C#
172 lines
6.5 KiB
C#
using MongoDB.Driver;
|
|
using MongoDB.Entities;
|
|
|
|
using PrivaPub.Domain.Social;
|
|
using PrivaPub.Domain.Timelines;
|
|
using PrivaPub.Federation.Actors;
|
|
using PrivaPub.Federation.Objects;
|
|
using PrivaPub.Infrastructure.Ids;
|
|
using PrivaPub.Models.Post;
|
|
using PrivaPub.Models.Social;
|
|
using PrivaPub.Models.User;
|
|
using PrivaPub.StaticServices;
|
|
|
|
using System.Globalization;
|
|
using System.Text.Json.Nodes;
|
|
|
|
using static PrivaPub.Federation.Objects.ActivityJson;
|
|
|
|
using PostEntity = PrivaPub.Models.Post.Post;
|
|
|
|
namespace PrivaPub.Federation.Inbox.Handlers
|
|
{
|
|
public class AnnounceHandler : IActivityHandler
|
|
{
|
|
readonly DbEntities _dbEntities;
|
|
readonly ILocalActorService _localActors;
|
|
readonly IRemotePosts _remotePosts;
|
|
readonly IFanout _fanout;
|
|
readonly IRemoteActorService _remoteActors;
|
|
|
|
public AnnounceHandler(DbEntities dbEntities, ILocalActorService localActors, IRemotePosts remotePosts, IFanout fanout, IRemoteActorService remoteActors)
|
|
{
|
|
_remoteActors = remoteActors;
|
|
_dbEntities = dbEntities;
|
|
_localActors = localActors;
|
|
_remotePosts = remotePosts;
|
|
_fanout = fanout;
|
|
}
|
|
|
|
public string Type => "Announce";
|
|
|
|
public async Task Handle(JsonNode activity, ForeignAvatar actor, CancellationToken token)
|
|
{
|
|
var inner = activity["object"];
|
|
if (inner is JsonObject && Value(inner, "type") is "Create" or "Update" or "Delete" or "Like" or "Dislike" or "Undo" or "Add" or "Remove" or "Block")
|
|
{
|
|
await GroupActivity(inner, actor, token);
|
|
return;
|
|
}
|
|
var objectUri = Id(inner);
|
|
var announceId = Id(activity);
|
|
if (objectUri == default || announceId == default)
|
|
return;
|
|
if (await _dbEntities.Posts.Match(p => p.ObjectURI == announceId).ExecuteAnyAsync(token))
|
|
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)
|
|
return;
|
|
if (original == default && !isLocal)
|
|
original = await _remotePosts.StoreContext(objectUri, 0, token);
|
|
if (original == default || original.Visibility is not (PostVisibility.Public or PostVisibility.Unlisted))
|
|
return;
|
|
|
|
var to = Strings(activity["to"]);
|
|
var cc = Strings(activity["cc"]);
|
|
var published = DateTimeOffset.TryParse(Value(activity, "published"), CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal, out var at)
|
|
? at.UtcDateTime
|
|
: DateTime.UtcNow;
|
|
var reblog = new PostEntity
|
|
{
|
|
ID = PrivacyIds.Arrived(published),
|
|
ObjectURI = announceId,
|
|
ActivityURI = announceId,
|
|
ActorURI = actor.ActorURI,
|
|
AuthorAccountId = actor.ID,
|
|
ReblogOfPostId = original.ID,
|
|
Visibility = Addressing.Classify(to, cc, actor.FollowersURL),
|
|
To = to,
|
|
Cc = cc,
|
|
IsFederatedCopy = true,
|
|
CreationDate = published,
|
|
UpdateDate = published
|
|
};
|
|
if (reblog.Visibility == PostVisibility.Direct)
|
|
return;
|
|
try
|
|
{
|
|
await DB.Default.SaveAsync(reblog, token);
|
|
}
|
|
catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
|
|
{
|
|
return;
|
|
}
|
|
|
|
await DB.Default.Update<PostEntity>().MatchID(original.ID).Modify(b => b.Inc(p => p.ReblogsCount, 1)).ExecuteAsync(token);
|
|
if (!original.IsFederatedCopy)
|
|
await Notifications.Add(original.GroupUserId, NotificationType.Reblog, actor.ID, actor.ActorURI, original.ID, token);
|
|
await _fanout.Distribute(reblog, token);
|
|
}
|
|
|
|
async Task GroupActivity(JsonNode inner, ForeignAvatar group, CancellationToken token)
|
|
{
|
|
if (group.AvatarType != AvatarType.Group
|
|
|| !await _dbEntities.Followings.Match(f => f.TargetActorURI == group.ActorURI && f.State == FollowState.Accepted).ExecuteAnyAsync(token))
|
|
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))
|
|
return;
|
|
var post = await _remotePosts.StoreContext(objectUri, 0, token);
|
|
if (post == default)
|
|
return;
|
|
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);
|
|
break;
|
|
case "Update" when objectUri != default:
|
|
var stored = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && p.AudienceURI == group.ActorURI && !p.DeletedAt.HasValue)
|
|
.ExecuteFirstAsync(token);
|
|
if (stored == default)
|
|
return;
|
|
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)
|
|
return;
|
|
stored.Revisions.Add(new PostRevision
|
|
{
|
|
Title = stored.Title,
|
|
SpoilerText = stored.SpoilerText,
|
|
ContentHtml = stored.ContentHtml,
|
|
HasContentWarning = stored.HasContentWarning,
|
|
EditedAt = stored.EditedAt ?? stored.CreationDate
|
|
});
|
|
stored.Title = note.Title;
|
|
stored.SpoilerText = note.SpoilerText;
|
|
stored.HasContentWarning = note.Sensitive;
|
|
stored.Text = note.ContentHtml;
|
|
stored.ContentHtml = note.ContentHtml;
|
|
stored.Tags = note.Tags.ToList();
|
|
stored.Media = note.Attachments.ToList();
|
|
stored.EditedAt = note.Updated ?? DateTime.UtcNow;
|
|
await DB.Default.SaveAsync(stored, token);
|
|
}
|
|
break;
|
|
case "Delete" when objectUri != default:
|
|
var deleted = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri && p.AudienceURI == group.ActorURI).ExecuteFirstAsync(token);
|
|
if (deleted == default)
|
|
return;
|
|
using (var check = await _remoteActors.FetchObject(objectUri, token))
|
|
if (check != default && Value(System.Text.Json.Nodes.JsonNode.Parse(check.Root.GetRawText()), "type") != "Tombstone")
|
|
return;
|
|
await DB.Default.DeleteAsync<PostEntity>(deleted.ID);
|
|
await DB.Default.DeleteAsync<TimelineEntry>(e => e.PostId == deleted.ID);
|
|
break;
|
|
}
|
|
}
|
|
|
|
static List<string> Strings(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>()
|
|
};
|
|
}
|
|
}
|