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 Subscribe(LocalActor me, ForeignAvatar feed, bool on, CancellationToken token); Task 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 Sync(IJobQueue jobs, CancellationToken token) => jobs.Enqueue(JobKind.SyncFeeds, string.Empty, default, $"feeds|{Guid.NewGuid():N}", token); public async Task 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(s => s.AvatarId == me.Id && s.FeedActorURI == feed.ActorURI)).DeletedCount == 0) return true; await Sync(_jobs, token); return true; } public Task IsSubscribed(string avatarId, string feedActorUri, CancellationToken token) => DB.Default.Find().Match(s => s.AvatarId == avatarId && s.FeedActorURI == feedActorUri).ExecuteAnyAsync(token); public async Task Reconcile(CancellationToken token) { var read = (await DB.Default.Find().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 Handle(Job job, CancellationToken token) { using var scope = scopes.CreateScope(); await scope.ServiceProvider.GetRequiredService().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 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(), 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)); } } }