SocialPub/PrivaPub/Domain/Social/FeedFollows.cs

using MongoDB.Driver;
using MongoDB.Entities;

using PrivaPub.Federation.Actors;
using PrivaPub.Infrastructure.Jobs;
using PrivaPub.Models.Jobs;
using PrivaPub.Models.Social;
using PrivaPub.Models.User;
using PrivaPub.StaticServices;

namespace PrivaPub.Domain.Social
{
	public interface IFeedFollows
	{
		Task<bool> Subscribe(LocalActor me, ForeignAvatar feed, bool on, CancellationToken token);
		Task<bool> IsSubscribed(string avatarId, string feedActorUri, CancellationToken token);
		Task Reconcile(CancellationToken token);
	}

	// Feeds read through the server (owner decision 2026-10-07): a persona following a feed is kept here alone, and the
	// server's reader (LocalActorService.ReaderUserName) follows every community in the feeds anyone reads, as Lemmy's own
	// follower of multi-communities does, and stops following one no feed holds any more. What it follows arrives as any
	// community's posts do; nobody's home shows them, the feed's timeline does.
	public class FeedFollows : IFeedFollows
	{
		public const int MaxCommunities = 500;
		static readonly TimeSpan Stale = TimeSpan.FromDays(1);

		readonly DbEntities _dbEntities;
		readonly ILocalActorService _localActors;
		readonly IRemoteActorService _remoteActors;
		readonly IFollowService _follows;
		readonly IJobQueue _jobs;

		public FeedFollows(DbEntities dbEntities, ILocalActorService localActors, IRemoteActorService remoteActors, IFollowService follows, IJobQueue jobs)
		{
			_dbEntities = dbEntities;
			_localActors = localActors;
			_remoteActors = remoteActors;
			_follows = follows;
			_jobs = jobs;
		}

		public static Task<bool> Sync(IJobQueue jobs, CancellationToken token) =>
			jobs.Enqueue(JobKind.SyncFeeds, string.Empty, default, $"feeds|{Guid.NewGuid():N}", token);

		public async Task<bool> Subscribe(LocalActor me, ForeignAvatar feed, bool on, CancellationToken token)
		{
			if (feed is not { AvatarType: AvatarType.Feed })
				return false;
			if (on)
				try
				{
					await DB.Default.SaveAsync(new FeedSubscription { AvatarId = me.Id, FeedActorURI = feed.ActorURI, FeedAccountId = feed.ID }, token);
				}
				catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
				{
					return true;
				}
			else if ((await DB.Default.DeleteAsync<FeedSubscription>(s => s.AvatarId == me.Id && s.FeedActorURI == feed.ActorURI)).DeletedCount == 0)
				return true;
			await Sync(_jobs, token);
			return true;
		}

		public Task<bool> IsSubscribed(string avatarId, string feedActorUri, CancellationToken token) =>
			DB.Default.Find<FeedSubscription>().Match(s => s.AvatarId == avatarId && s.FeedActorURI == feedActorUri).ExecuteAnyAsync(token);

		public async Task Reconcile(CancellationToken token)
		{
			var read = (await DB.Default.Find<FeedSubscription, string>().Project(s => s.FeedActorURI).ExecuteAsync(token)).Distinct().ToList();
			var feeds = await _dbEntities.ForeignAvatars.Match(a => read.Contains(a.ActorURI) && a.AvatarType == AvatarType.Feed && !a.DeletionAt.HasValue)
				.ExecuteAsync(token);
			// a feed's communities are as its server listed them a day ago at most (AccountCountsJob reads them, then asks
			// for this again)
			foreach (var stale in feeds.Where(f => !(DateTime.UtcNow - f.CountedAt < Stale)))
				if (Uri.TryCreate(stale.ActorURI, UriKind.Absolute, out var uri))
					await _jobs.Enqueue(JobKind.CountAccount, stale.ActorURI, uri.Host.ToLowerInvariant(), AccountCountsJob.DedupeKey(stale.ActorURI, DateTime.UtcNow), token);

			var ours = _localActors.BaseAddress + "/";
			var wanted = feeds.SelectMany(f => f.FeedCommunities).Where(c => !c.StartsWith(ours, StringComparison.Ordinal))
				.Distinct(StringComparer.Ordinal).Take(MaxCommunities).ToHashSet(StringComparer.Ordinal);
			var reader = await _localActors.GetReaderActor(token);
			var following = await _dbEntities.Followings.Match(f => f.AvatarId == reader.Id).ExecuteAsync(token);
			foreach (var gone in following.Where(f => !wanted.Contains(f.TargetActorURI)))
				await _follows.UnfollowAs(reader, gone.TargetActorURI, token);
			var followed = following.Select(f => f.TargetActorURI).ToHashSet(StringComparer.Ordinal);
			foreach (var community in wanted.Where(c => !followed.Contains(c)))
				// only a community: a feed lists nothing else, and the reader follows nothing else
				if (await _remoteActors.GetActor(community, refresh: false, token) is { AvatarType: AvatarType.Group })
					await _follows.FollowAs(reader, community, showReblogs: false, token);
		}
	}

	public class FeedSyncJob(IServiceScopeFactory scopes) : IJobHandler
	{
		public JobKind Kind => JobKind.SyncFeeds;
		public int Concurrency => 1;
		public int MaxAttempts => 2;
		public int PerHostLimit => 1;

		public async Task<JobOutcome> Handle(Job job, CancellationToken token)
		{
			using var scope = scopes.CreateScope();
			await scope.ServiceProvider.GetRequiredService<IFeedFollows>().Reconcile(token);
			return JobOutcome.Done;
		}
	}

	// brings the reader's follows up to date with the feeds read here, a minute after start and then every six hours
	public sealed class FeedReader(IServiceScopeFactory scopes, ILogger<FeedReader> logger) : BackgroundService
	{
		static readonly TimeSpan Every = TimeSpan.FromHours(6);

		protected override async Task ExecuteAsync(CancellationToken token)
		{
			await Task.Delay(TimeSpan.FromMinutes(1), token);
			using var timer = new PeriodicTimer(Every);
			do
			{
				try
				{
					using var scope = scopes.CreateScope();
					await FeedFollows.Sync(scope.ServiceProvider.GetRequiredService<IJobQueue>(), token);
				}
				catch (Exception ex) when (ex is not OperationCanceledException)
				{
					logger.LogWarning(ex, "Feeds could not be brought up to date");
				}
			}
			while (await timer.WaitForNextTickAsync(token));
		}
	}
}