Home timelines, mention notifications, and threads that fetch their parents

Fanout writes a TimelineEntry for every persona a post should reach: the
author, local followers of a local author, local followers of a remote one,
the members of a direct conversation or a circle. Mastodon's home rules
apply when it is written: a reply shows only to followers of both sides
(or to the one replied to), a reblog only where reblogs are wanted. A
mention of a local persona becomes a Mention notification.

Inbound posts are now also kept when a persona follows their author.

RemotePosts holds what CreateHandler and backfill share: building a Post
from a note, and fetching a public parent the first time a reply to it
arrives, so its author is known (a reply to an unknown or non-public
parent stays out of home timelines). Each fetched ancestor queues a
FetchAncestors job for the next one, up to ten deep.

/clientapi/timeline/home and /clientapi/notifications (with
/notifications/read) page by max_id for the persona's own root only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012CzABvBkbcFqoHdmi8b9WB

PrivaPub/Federation/Inbox/Handlers/CreateHandler.cs

@@ -1,6 +1,7 @@
 using MongoDB.Driver;
 using MongoDB.Entities;
 
+using PrivaPub.Domain.Timelines;
 using PrivaPub.Federation.Actors;
 using PrivaPub.Federation.Moderation;
 using PrivaPub.Federation.Objects;
@@ -10,6 +11,7 @@ using PrivaPub.Federation.Rendering;
 using PrivaPub.Models.Federation;
 using PrivaPub.Models.Group;
 using PrivaPub.Models.Post;
+using PrivaPub.Models.Social;
 using PrivaPub.Models.User;
 using PrivaPub.StaticServices;
 
@@ -29,10 +31,14 @@ namespace PrivaPub.Federation.Inbox.Handlers
 		readonly IRemoteActorService _remoteActors;
 		readonly IDeliveryService _delivery;
 		readonly IDomainBlocks _domainBlocks;
+		readonly IFanout _fanout;
+		readonly IRemotePosts _remotePosts;
 
 		public CreateHandler(DbEntities dbEntities, ILocalActorService localActors, IRemoteActorService remoteActors, IDeliveryService delivery,
-			IDomainBlocks domainBlocks)
+			IDomainBlocks domainBlocks, IFanout fanout, IRemotePosts remotePosts)
 		{
+			_fanout = fanout;
+			_remotePosts = remotePosts;
 			_dbEntities = dbEntities;
 			_localActors = localActors;
 			_remoteActors = remoteActors;
@@ -86,21 +92,13 @@ namespace PrivaPub.Federation.Inbox.Handlers
 				group = default;
 
 			var repliesToLocal = parent is { IsFederatedCopy: false } && visibility != PostVisibility.Direct;
-			if (visibility == PostVisibility.Direct ? persons.Count == 0 : group == default && persons.Count == 0 && !repliesToLocal)
+			var followed = visibility != PostVisibility.Direct && await _dbEntities.Followings
+				.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)
 				return;
-
-			var mentions = new List<PostMention>();
-			foreach (var mention in note.Mentions)
-			{
-				var local = await _localActors.FindByUri(mention.ActorURI, token);
-				mentions.Add(new PostMention
-				{
-					ActorURI = mention.ActorURI,
-					Handle = mention.Handle,
-					IsLocal = local != default,
-					AccountId = local?.Id
-				});
-			}
+			if (parent == default && visibility != PostVisibility.Direct)
+				parent = await _remotePosts.Parent(note.InReplyTo, 1, token);
 
 			DmGroup conversation = default;
 			if (visibility == PostVisibility.Direct)
@@ -112,39 +110,10 @@ namespace PrivaPub.Federation.Inbox.Handlers
 				conversation = await FindOrCreateConversation(participants, Origin.Same(note.Context, author.ActorURI) ? note.Context : default, token);
 			}
 
-			var post = new PostEntity
-			{
-				ID = PrivacyIds.At(note.Published),
-				ObjectURI = note.Id,
-				ActivityURI = Id(activity),
-				ActorURI = author.ActorURI,
-				AuthorAccountId = author.ID,
-				Visibility = visibility,
-				To = to,
-				Cc = cc,
-				Url = note.Url,
-				ContextURI = note.Context,
-				QuoteURI = note.QuoteUri,
-				GroupId = group?.Id,
-				ConversationId = conversation?.ID,
-				Title = note.Title,
-				SpoilerText = note.SpoilerText,
-				HasContentWarning = note.Sensitive,
-				Text = note.ContentHtml,
-				ContentHtml = note.ContentHtml,
-				ContentFormat = ContentFormat.Html,
-				Language = note.Language,
-				Mentions = mentions,
-				Tags = note.Tags.ToList(),
-				Media = _domainBlocks.Find(new Uri(author.ActorURI).Host)?.RejectMedia == true ? new() : note.Attachments.ToList(),
-				InReplyToURI = note.InReplyTo,
-				AnsweringToPostId = parent?.ID,
-				InReplyToAccountId = parent?.AuthorAccountId ?? parent?.GroupUserId,
-				IsFederatedCopy = true,
-				CreationDate = note.Published,
-				UpdateDate = note.Updated ?? note.Published,
-				EditedAt = note.Updated
-			};
+			var post = await _remotePosts.Build(note, author, visibility, to, cc, parent, token);
+			post.ActivityURI = Id(activity);
+			post.GroupId = group?.Id;
+			post.ConversationId = conversation?.ID;
 			try
 			{
 				await DB.Default.SaveAsync(post, token);
@@ -156,6 +125,7 @@ namespace PrivaPub.Federation.Inbox.Handlers
 
 			if (parent != default)
 				await DB.Default.Update<PostEntity>().MatchID(parent.ID).Modify(b => b.Inc(p => p.RepliesCount, 1)).ExecuteAsync(token);
+			await _fanout.Distribute(post, token);
 			if (conversation != default)
 				await DB.Default.Update<DmGroup>().MatchID(conversation.ID).Modify(g => g.UpdatedAt, DateTime.UtcNow).ExecuteAsync(token);
 			if (group != default)

PrivaPub/Federation/Inbox/RemotePosts.cs

@@ -0,0 +1,173 @@
+using MongoDB.Driver;
+using MongoDB.Entities;
+
+using PrivaPub.Federation.Actors;
+using PrivaPub.Federation.Moderation;
+using PrivaPub.Federation.Objects;
+using PrivaPub.Infrastructure.Ids;
+using PrivaPub.Infrastructure.Jobs;
+using PrivaPub.Models.Jobs;
+using PrivaPub.Models.Post;
+using PrivaPub.Models.User;
+using PrivaPub.StaticServices;
+
+using System.Text.Json;
+using System.Text.Json.Nodes;
+
+using PostEntity = PrivaPub.Models.Post.Post;
+
+namespace PrivaPub.Federation.Inbox
+{
+	public sealed record AncestorsPayload(string PostId, int Depth);
+
+	public interface IRemotePosts
+	{
+		Task<PostEntity> Build(NoteDocument note, ForeignAvatar author, PostVisibility visibility, IReadOnlyList<string> to, IReadOnlyList<string> cc,
+			PostEntity parent, CancellationToken token);
+		Task<PostEntity> Parent(string inReplyTo, int depth, CancellationToken token);
+		Task<PostEntity> StoreContext(string objectUri, int depth, CancellationToken token);
+	}
+
+	public class RemotePosts : IRemotePosts
+	{
+		public const int MaxDepth = 10;
+
+		readonly DbEntities _dbEntities;
+		readonly ILocalActorService _localActors;
+		readonly IRemoteActorService _remoteActors;
+		readonly IDomainBlocks _domainBlocks;
+		readonly IJobQueue _queue;
+
+		public RemotePosts(DbEntities dbEntities, ILocalActorService localActors, IRemoteActorService remoteActors, IDomainBlocks domainBlocks,
+			IJobQueue queue)
+		{
+			_dbEntities = dbEntities;
+			_localActors = localActors;
+			_remoteActors = remoteActors;
+			_domainBlocks = domainBlocks;
+			_queue = queue;
+		}
+
+		public async Task<PostEntity> Build(NoteDocument note, ForeignAvatar author, PostVisibility visibility, IReadOnlyList<string> to,
+			IReadOnlyList<string> cc, PostEntity parent, CancellationToken token)
+		{
+			var mentions = new List<PostMention>();
+			foreach (var mention in note.Mentions)
+			{
+				var local = await _localActors.FindByUri(mention.ActorURI, token);
+				mentions.Add(new PostMention { ActorURI = mention.ActorURI, Handle = mention.Handle, IsLocal = local != default, AccountId = local?.Id });
+			}
+
+			return new PostEntity
+			{
+				ID = PrivacyIds.At(note.Published),
+				ObjectURI = note.Id,
+				ActorURI = author.ActorURI,
+				AuthorAccountId = author.ID,
+				Visibility = visibility,
+				To = to.ToList(),
+				Cc = cc.ToList(),
+				Url = note.Url,
+				ContextURI = note.Context,
+				QuoteURI = note.QuoteUri,
+				Title = note.Title,
+				SpoilerText = note.SpoilerText,
+				HasContentWarning = note.Sensitive,
+				Text = note.ContentHtml,
+				ContentHtml = note.ContentHtml,
+				ContentFormat = ContentFormat.Html,
+				Language = note.Language,
+				Mentions = mentions,
+				Tags = note.Tags.ToList(),
+				Media = _domainBlocks.Find(new Uri(author.ActorURI).Host)?.RejectMedia == true ? new() : note.Attachments.ToList(),
+				InReplyToURI = note.InReplyTo,
+				AnsweringToPostId = parent?.ID,
+				InReplyToAccountId = parent?.AuthorAccountId ?? parent?.GroupUserId,
+				IsFederatedCopy = true,
+				CreationDate = note.Published,
+				UpdateDate = note.Updated ?? note.Published,
+				EditedAt = note.Updated
+			};
+		}
+
+		public async Task<PostEntity> Parent(string inReplyTo, int depth, CancellationToken token)
+		{
+			if (string.IsNullOrEmpty(inReplyTo))
+				return default;
+			var known = await _dbEntities.Posts.Match(p => p.ObjectURI == inReplyTo && !p.DeletedAt.HasValue).ExecuteFirstAsync(token);
+			if (known != default || inReplyTo.StartsWith(_localActors.BaseAddress + "/", StringComparison.OrdinalIgnoreCase))
+				return known;
+			return await StoreContext(inReplyTo, depth, token);
+		}
+
+		public async Task<PostEntity> StoreContext(string objectUri, int depth, CancellationToken token)
+		{
+			var existing = await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri).ExecuteFirstAsync(token);
+			if (existing != default || depth > MaxDepth)
+				return existing;
+
+			using var fetched = await _remoteActors.FetchObject(objectUri, token);
+			var note = fetched == default ? default : NoteParser.Parse(JsonNode.Parse(fetched.Root.GetRawText()));
+			if (note == default || !Origin.Same(note.Id, note.AttributedTo))
+				return default;
+			var visibility = Addressing.Classify(note.To, note.Cc, default);
+			if (visibility is not (PostVisibility.Public or PostVisibility.Unlisted))
+				return default;
+			var author = await _remoteActors.GetActor(note.AttributedTo, refresh: false, token);
+			if (author == default)
+				return default;
+
+			var grandparent = string.IsNullOrEmpty(note.InReplyTo)
+				? default
+				: await _dbEntities.Posts.Match(p => p.ObjectURI == note.InReplyTo && !p.DeletedAt.HasValue).ExecuteFirstAsync(token);
+			var post = await Build(note, author, visibility, note.To, note.Cc, grandparent, token);
+			try
+			{
+				await DB.Default.SaveAsync(post, token);
+			}
+			catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
+			{
+				return await _dbEntities.Posts.Match(p => p.ObjectURI == objectUri).ExecuteFirstAsync(token);
+			}
+
+			if (grandparent == default && !string.IsNullOrEmpty(note.InReplyTo) && depth < MaxDepth)
+				await _queue.Enqueue(JobKind.FetchAncestors, JsonSerializer.Serialize(new AncestorsPayload(post.ID, depth + 1)), new Uri(note.InReplyTo).Host,
+					"ancestors|" + post.ID, token);
+			return post;
+		}
+	}
+
+	public class AncestorsJobHandler : IJobHandler
+	{
+		readonly DbEntities _dbEntities;
+		readonly IRemotePosts _remotePosts;
+
+		public AncestorsJobHandler(DbEntities dbEntities, IRemotePosts remotePosts)
+		{
+			_dbEntities = dbEntities;
+			_remotePosts = remotePosts;
+		}
+
+		public JobKind Kind => JobKind.FetchAncestors;
+		public int Concurrency => 1;
+		public int MaxAttempts => 3;
+		public int PerHostLimit => 1;
+
+		public async Task<JobOutcome> Handle(Job job, CancellationToken token)
+		{
+			var payload = JsonSerializer.Deserialize<AncestorsPayload>(job.Payload);
+			var child = await _dbEntities.Posts.MatchID(payload.PostId).ExecuteFirstAsync(token);
+			if (child == default || string.IsNullOrEmpty(child.InReplyToURI) || !string.IsNullOrEmpty(child.AnsweringToPostId))
+				return JobOutcome.Done;
+
+			var parent = await _remotePosts.Parent(child.InReplyToURI, payload.Depth, token);
+			if (parent == default)
+				return JobOutcome.Done;
+			await DB.Default.Update<PostEntity>().MatchID(child.ID)
+				.Modify(p => p.AnsweringToPostId, parent.ID)
+				.Modify(p => p.InReplyToAccountId, parent.AuthorAccountId ?? parent.GroupUserId)
+				.ExecuteAsync(token);
+			return JobOutcome.Done;
+		}
+	}
+}