SocialPub/PrivaPub/Domain/Statuses/PollService.cs

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<string> 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<StatusOutcome> Vote(LocalActor me, PostEntity post, IReadOnlyList<int> choices, CancellationToken token);
		Task Receive(PostEntity post, ForeignAvatar voter, string choice, string activityId, CancellationToken token);
		Task<HashSet<int>> 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<StatusOutcome> Vote(LocalActor me, PostEntity post, IReadOnlyList<int> 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<PollVote>().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<PollVote>().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<HashSet<int>> OwnVotes(string postId, string avatarId, CancellationToken token) =>
			avatarId == default
				? new HashSet<int>()
				: (await DB.Default.Find<PollVote>().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<int> choices, bool newVoter, CancellationToken token)
		{
			var update = DB.Default.Update<PostEntity>().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<int> 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<JobOutcome> 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.Publish(author, post, Update(post, author), token);
			return JobOutcome.Done;
		}

		protected virtual Task Closing(PostEntity post, LocalActor author, CancellationToken token) => Task.CompletedTask;

		static JsonObject Update(PostEntity post, LocalActor author)
		{
			var question = post.Visibility == PostVisibility.Direct
				? ActivityPubRenderer.DirectNote(post, author, Array.Empty<(string, string)>(), post.ContextURI)
				: ActivityPubRenderer.Note(post, author, default, post.InReplyToURI);
			question["to"] = new JsonArray(post.To.Select(t => (JsonNode)t).ToArray());
			question["cc"] = new JsonArray(post.Cc.Select(c => (JsonNode)c).ToArray());
			return new JsonObject
			{
				["@context"] = ActivityPubRenderer.Context(),
				["id"] = author.ActivityUri($"update-{post.ID}-poll-{DateTime.UtcNow.Ticks}"),
				["type"] = "Update",
				["actor"] = author.Uri,
				["to"] = question["to"]!.DeepClone(),
				["cc"] = question["cc"]!.DeepClone(),
				["object"] = question
			};
		}
	}

	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<JobOutcome> 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<PostEntity>().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<PollVote>().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);
		}
	}
}