using MongoDB.Driver; using MongoDB.Entities; using PrivaPub.Domain.Social; using PrivaPub.Federation.Actors; using PrivaPub.Federation.Inbox.Handlers; using PrivaPub.Federation.Outbox; using PrivaPub.Federation.Rendering; using PrivaPub.Infrastructure.Jobs; using PrivaPub.Models.Federation; using PrivaPub.Models.Jobs; using PrivaPub.Models.Post; using PrivaPub.Models.Social; using PrivaPub.Models.User; using PrivaPub.StaticServices; using System.Text.Json.Nodes; using PostEntity = PrivaPub.Models.Post.Post; namespace PrivaPub.Domain.Statuses { public sealed record PollDraft(IReadOnlyList Options, int ExpiresIn, bool Multiple, bool HideTotals); public interface IPollService { string Invalid(PollDraft draft); PostPoll Create(PollDraft draft); Task Scheduled(PostEntity post, CancellationToken token); Task Vote(LocalActor me, PostEntity post, IReadOnlyList choices, CancellationToken token); Task Receive(PostEntity post, ForeignAvatar voter, string choice, string activityId, CancellationToken token); Task> OwnVotes(string postId, string avatarId, CancellationToken token); } public class PollService : IPollService { public const int MaxOptions = 4; public const int MaxOptionLength = 50; public const int MinExpiration = 300; public const int MaxExpiration = 2_629_746; static readonly TimeSpan RefreshEvery = TimeSpan.FromMinutes(3); readonly DbEntities _dbEntities; readonly ILocalActorService _localActors; readonly IDeliveryService _delivery; readonly IJobQueue _queue; public PollService(DbEntities dbEntities, ILocalActorService localActors, IDeliveryService delivery, IJobQueue queue) { _dbEntities = dbEntities; _localActors = localActors; _delivery = delivery; _queue = queue; } public string Invalid(PollDraft draft) { var options = draft.Options.Select(o => o?.Trim()).ToList(); if (options.Count is < 2 or > MaxOptions) return $"Validation failed: A poll needs between 2 and {MaxOptions} options"; if (options.Any(o => string.IsNullOrEmpty(o) || o.Length > MaxOptionLength)) return $"Validation failed: Poll options must be between 1 and {MaxOptionLength} characters"; if (options.Distinct(StringComparer.Ordinal).Count() != options.Count) return "Validation failed: Poll options must be unique"; if (draft.ExpiresIn is < MinExpiration or > MaxExpiration) return "Validation failed: The poll's duration is out of range"; return default; } public PostPoll Create(PollDraft draft) => new() { Options = draft.Options.Select(o => new PollOption { Title = o.Trim() }).ToList(), Multiple = draft.Multiple, HideTotals = draft.HideTotals, ExpiresAt = DateTime.UtcNow.AddSeconds(draft.ExpiresIn), VotersCount = 0 }; public async Task Scheduled(PostEntity post, CancellationToken token) { if (post.Poll?.ExpiresAt is not { } ends) return; await _queue.EnqueueMany(new[] { new Job { Kind = JobKind.PollClose, Payload = post.ID, Host = LocalHost, DedupeKey = $"poll-close|{post.ID}", RunAt = ends } }, token); } public async Task Vote(LocalActor me, PostEntity post, IReadOnlyList choices, CancellationToken token) { var poll = post.Poll; if (poll == default) return StatusOutcome.Fail(StatusCodes.Status404NotFound, "Record not found"); if (Ended(poll)) return StatusOutcome.Fail(StatusCodes.Status422UnprocessableEntity, "Validation failed: The poll has already ended"); if (!post.IsFederatedCopy && post.GroupUserId == me.Id) return StatusOutcome.Fail(StatusCodes.Status422UnprocessableEntity, "Validation failed: You cannot vote in your own poll"); var picked = choices.Distinct().ToList(); if (picked.Count == 0 || picked.Any(c => c < 0 || c >= poll.Options.Count) || !poll.Multiple && picked.Count > 1) return StatusOutcome.Fail(StatusCodes.Status422UnprocessableEntity, "Validation failed: The choices are invalid"); if (await DB.Default.Find().Match(v => v.PostId == post.ID && v.ActorURI == me.Uri).ExecuteAnyAsync(token)) return StatusOutcome.Fail(StatusCodes.Status422UnprocessableEntity, "Validation failed: You have already voted on this poll"); foreach (var choice in picked) await DB.Default.SaveAsync(new PollVote { PostId = post.ID, VoterAccountId = me.Id, ActorURI = me.Uri, IsLocalVoter = true, Choice = choice, ActivityURI = me.Uri + $"#votes/{post.ID}/{choice}" }, token); await Count(post, picked, newVoter: true, token); if (post.IsFederatedCopy) await SendVotes(me, post, picked, token); else await Refreshed(post, token); await Scheduled(post, token); return new StatusOutcome(await _dbEntities.Posts.MatchID(post.ID).ExecuteFirstAsync(token)); } public async Task Receive(PostEntity post, ForeignAvatar voter, string choice, string activityId, CancellationToken token) { var poll = post.Poll; if (poll == default || post.IsFederatedCopy || Ended(poll) || !LikeHandler.MaySee(post, voter)) return; var index = poll.Options.FindIndex(o => o.Title == choice); if (index < 0) return; var earlier = await DB.Default.Find().Match(v => v.PostId == post.ID && v.ActorURI == voter.ActorURI).ExecuteAsync(token); if (earlier.Count > 0 && (!poll.Multiple || earlier.Any(v => v.Choice == index))) return; try { await DB.Default.SaveAsync(new PollVote { PostId = post.ID, VoterAccountId = voter.ID, ActorURI = voter.ActorURI, Choice = index, ActivityURI = activityId }, token); } catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey) { return; } await Count(post, new[] { index }, newVoter: earlier.Count == 0, token); await Refreshed(post, token); } public async Task> OwnVotes(string postId, string avatarId, CancellationToken token) => avatarId == default ? new HashSet() : (await DB.Default.Find().Match(v => v.PostId == postId && v.VoterAccountId == avatarId && v.IsLocalVoter).ExecuteAsync(token)) .Select(v => v.Choice).ToHashSet(); public static bool Ended(PostPoll poll) => poll.ClosedAt.HasValue || poll.ExpiresAt <= DateTime.UtcNow; async Task Count(PostEntity post, IReadOnlyList choices, bool newVoter, CancellationToken token) { var update = DB.Default.Update().MatchID(post.ID); foreach (var choice in choices) update = update.Modify(b => b.Inc($"{nameof(PostEntity.Poll)}.{nameof(PostPoll.Options)}.{choice}.{nameof(PollOption.Votes)}", 1)); if (newVoter && post.Poll.VotersCount.HasValue) update = update.Modify(b => b.Inc($"{nameof(PostEntity.Poll)}.{nameof(PostPoll.VotersCount)}", 1)); await update.ExecuteAsync(token); } async Task SendVotes(LocalActor me, PostEntity post, IReadOnlyList choices, CancellationToken token) { var owner = await _dbEntities.ForeignAvatars.Match(f => f.ActorURI == post.ActorURI).ExecuteFirstAsync(token); if (string.IsNullOrEmpty(owner?.InboxURL)) return; foreach (var choice in choices) { var voteId = me.Uri + $"#votes/{post.ID}/{choice}"; await _delivery.Enqueue(me, new[] { owner.InboxURL }, new JsonObject { ["@context"] = ActivityPubRenderer.ActivityStreams, ["id"] = voteId + "/activity", ["type"] = "Create", ["actor"] = me.Uri, ["to"] = new JsonArray(owner.ActorURI), ["object"] = new JsonObject { ["id"] = voteId, ["type"] = "Note", ["name"] = post.Poll.Options[choice].Title, ["attributedTo"] = me.Uri, ["to"] = new JsonArray(owner.ActorURI), ["inReplyTo"] = post.ObjectURI } }, token); } } async Task Refreshed(PostEntity post, CancellationToken token) { var bucket = DateTime.UtcNow.Ticks / RefreshEvery.Ticks; await _queue.EnqueueMany(new[] { new Job { Kind = JobKind.PollRefresh, Payload = post.ID, Host = LocalHost, DedupeKey = $"poll-refresh|{post.ID}|{bucket}", RunAt = new DateTime((bucket + 1) * RefreshEvery.Ticks, DateTimeKind.Utc) } }, token); } string LocalHost => new Uri(_localActors.BaseAddress).Host; } public class PollRefreshJob : IJobHandler { readonly DbEntities _dbEntities; readonly ILocalActorService _localActors; readonly IOutboxPublisher _outbox; public PollRefreshJob(DbEntities dbEntities, ILocalActorService localActors, IOutboxPublisher outbox) { _dbEntities = dbEntities; _localActors = localActors; _outbox = outbox; } public virtual JobKind Kind => JobKind.PollRefresh; public int Concurrency => 1; public int MaxAttempts => 3; public int PerHostLimit => 1; public virtual async Task Handle(Job job, CancellationToken token) { var post = await _dbEntities.Posts.MatchID(job.Payload).ExecuteFirstAsync(token); if (post?.Poll == default || post.IsFederatedCopy || post.DeletedAt.HasValue) return JobOutcome.Done; var author = await _localActors.FindById(LocalActorKind.Person, post.GroupUserId, token); if (author == default) return JobOutcome.Done; await Closing(post, author, token); if (!post.IsLocalOnly) await _outbox.PublishUpdate(author, post, $"poll-{DateTime.UtcNow.Ticks}", token); return JobOutcome.Done; } protected virtual Task Closing(PostEntity post, LocalActor author, CancellationToken token) => Task.CompletedTask; } public class PollCloseJob : PollRefreshJob { readonly DbEntities _posts; public PollCloseJob(DbEntities dbEntities, ILocalActorService localActors, IOutboxPublisher outbox) : base(dbEntities, localActors, outbox) { _posts = dbEntities; } public override JobKind Kind => JobKind.PollClose; public override async Task Handle(Job job, CancellationToken token) { var post = await _posts.Posts.MatchID(job.Payload).ExecuteFirstAsync(token); if (post?.Poll == default || post.DeletedAt.HasValue) return JobOutcome.Done; if (!post.IsFederatedCopy) return await base.Handle(job, token); await Notify(post, post.AuthorAccountId, post.ActorURI, token); return JobOutcome.Done; } protected override async Task Closing(PostEntity post, LocalActor author, CancellationToken token) { post.Poll.ClosedAt ??= DateTime.UtcNow; await DB.Default.Update().MatchID(post.ID).Modify(p => p.Poll.ClosedAt, post.Poll.ClosedAt).ExecuteAsync(token); await Notify(post, author.Id, author.Uri, token); } public static async Task Notify(PostEntity post, string authorAccountId, string authorUri, CancellationToken token) { var voters = await DB.Default.Find().Match(v => v.PostId == post.ID && v.IsLocalVoter).ExecuteAsync(token); foreach (var voter in voters.Select(v => v.VoterAccountId).Distinct()) await Notifications.Add(voter, NotificationType.Poll, authorAccountId, authorUri, post.ID, token); if (!post.IsFederatedCopy) await Notifications.Add(post.GroupUserId, NotificationType.Poll, post.GroupUserId, authorUri, post.ID, token); } } }