implement randomly distributed donation tokens

TPP.Common/Utils/StringExtensions.cs

@@ -0,0 +1,9 @@
+namespace TPP.Common.Utils;
+
+public static class StringExtensions
+{
+    public static string Genitive(this string self) =>
+        self.Length > 0 && self[^1] == 's'
+            ? self + "'"
+            : self + "'s";
+}

TPP.Core/ChattersWorker.cs

@@ -18,7 +18,8 @@ public sealed class ChattersWorker(
     IClock clock,
     TwitchApi twitchApi,
     IChattersSnapshotsRepo chattersSnapshotsRepo,
-    ConnectionConfig.Twitch chatConfig
+    ConnectionConfig.Twitch chatConfig,
+    IUserRepo userRepo
 ) : IWithLifecycle
 {
     private readonly ILogger<ChattersWorker> _logger = loggerFactory.CreateLogger<ChattersWorker>();
@@ -38,6 +39,14 @@ public async Task Start(CancellationToken cancellationToken)
 
                 ImmutableList<string> chatterNames = chatters.Select(c => c.UserLogin).ToImmutableList();
                 ImmutableList<string> chatterIds = chatters.Select(c => c.UserId).ToImmutableList();
+
+                // Record all yet unknown users. Makes other code that retrieves users via chatters easier,
+                // because that code can then rely on all users from the chatters snapshot actually existing in the DB.
+                HashSet<string> knownIds = (await userRepo.FindByIds(chatterIds)).Select(u => u.Id).ToHashSet();
+                HashSet<string> unknownIds = chatterIds.Except(knownIds).ToHashSet();
+                foreach (Chatter newUser in chatters.Where(ch => unknownIds.Contains(ch.UserId)))
+                    await userRepo.RecordUser(new UserInfo(newUser.UserId, newUser.UserName, newUser.UserLogin));
+
                 await chattersSnapshotsRepo.LogChattersSnapshot(
                     chatterNames, chatterIds, chatConfig.Channel, clock.GetCurrentInstant());
             }
@@ -66,7 +75,8 @@ private async Task<List<Chatter>> GetChatters(CancellationToken cancellationToke
             chatters.AddRange(getChattersResponse.Data);
             nextCursor = getChattersResponse.Pagination?.Cursor;
         } while (nextCursor != null);
-        _logger.LogDebug("Retrieved {NumChatters} chatters", chatters.Count);
+        _logger.LogDebug("Retrieved {NumChatters} chatters: {ChatterNames}",
+            chatters.Count, string.Join(", ", chatters.Select(c => c.UserLogin)));
         return chatters;
     }
 }

TPP.Core/DonationHandler.cs

@@ -12,6 +12,7 @@
 using TPP.Core.Streamlabs;
 using TPP.Model;
 using TPP.Persistence;
+using static TPP.Common.Utils.StringExtensions;
 
 namespace TPP.Core;
 
@@ -22,6 +23,7 @@ public class DonationHandler(
     IBank<User> tokensBank,
     IMessageSender messageSender,
     OverlayConnection overlayConnection,
+    IChattersSnapshotsRepo chattersSnapshotsRepo,
     int donorBadgeCents)
 {
     public record NewDonation(
@@ -82,7 +84,7 @@ await donationRepo.InsertDonation(
             await UpdateHasDonationBadge(donor);
             await GivenTokensToDonorAndNotifyThem(donor, donation.Id, tokens);
         }
-        // TODO randomly distribute donation tokens
+        await RandomlyDistributeTokens(donation.CreatedAt, donation.Id, donation.Username, tokens.Total());
         await overlayConnection.Send(new NewDonationEvent
         {
             // We used to look up emotes using the internal Emote Service, but this small feature (emotes in donations)
@@ -151,4 +153,39 @@ private async Task GivenTokensToDonorAndNotifyThem(User user, long donationId, D
             : $"You got T{tokens.Base} for your donation!";
         await messageSender.SendWhisper(user, message);
     }
+
+    private async Task RandomlyDistributeTokens(Instant createdAt, long donationId, string donorName, int tokens)
+    {
+        ChattersSnapshot? snapshot = await chattersSnapshotsRepo.GetRecentChattersSnapshot(
+            from: createdAt.Minus(Duration.FromMinutes(10)),
+            to: createdAt);
+        IReadOnlyList<string> candidateIds = snapshot?.ChatterIds ?? [];
+        logger.LogDebug("Token distribution candidates before eligibility filter: {Candidates}", candidateIds);
+        List<User> eligibleUsers = await userRepo.FindByIdsEligibleForHandouts(candidateIds);
+        if (eligibleUsers.Count == 0)
+        {
+            logger.LogWarning("Aborting distribution of {NumTokens} random tokens due to lack of candidates", tokens);
+            return;
+        }
+
+        Random rng = new();
+        Dictionary<User, int> winners = Enumerable
+            .Range(0, tokens)
+            .Select(_ => eligibleUsers[rng.Next(eligibleUsers.Count)])
+            .GroupBy(user => user)
+            .ToDictionary(grp => grp.Key, grp => grp.Count());
+        logger.LogInformation("Some users won tokens from a random donation distribution: {UsersToTokens}",
+            string.Join(", ", winners.Select(kvp => $"{kvp.Key}: {kvp.Value}")));
+        foreach ((User recipient, int winnerTokens) in winners)
+            await GivenTokensToRandomRecipientAndNotifyThem(recipient, donorName, donationId, winnerTokens);
+    }
+
+    private async Task GivenTokensToRandomRecipientAndNotifyThem(
+        User recipient, string donorName, long donationId, int tokens)
+    {
+        var transaction = new Transaction<User>(recipient, tokens, TransactionType.DonationRandomlyDistributedTokens,
+            new Dictionary<string, object?> { ["donation"] = donationId });
+        await tokensBank.PerformTransaction(transaction);
+        await messageSender.SendWhisper(recipient, $"You won T{tokens} from {donorName.Genitive()} donation!");
+    }
 }

TPP.Core/DonationsWorker.cs

@@ -29,7 +29,8 @@ public async Task Start(CancellationToken cancellationToken)
             try
             {
                 Donation? mostRecentDonation = await donationRepo.GetMostRecentDonation();
-                _logger.LogDebug("Polling for new donations... most recent one is {Donation}", mostRecentDonation);
+                _logger.LogDebug("Polling for new donations... most recent one is {DonationId}",
+                    mostRecentDonation?.DonationId);
                 List<StreamlabsClient.Donation> donations =
                     await streamlabsClient.GetDonations(after: mostRecentDonation?.DonationId, currency: "USD");
                 _logger.LogDebug("Received new donations: {Donations}", string.Join(", ", donations));

TPP.Core/Modes/ModeBase.cs

@@ -153,7 +153,8 @@ public ModeBase(
             _chattersWorker = primaryChat == null
                 ? null
                 : new ChattersWorker(loggerFactory, clock,
-                    ((TwitchChat)_chats[primaryChat.Name]).TwitchApi, repos.ChattersSnapshotsRepo, primaryChat);
+                    ((TwitchChat)_chats[primaryChat.Name]).TwitchApi, repos.ChattersSnapshotsRepo, primaryChat,
+                    repos.UserRepo);
 
             StreamlabsConfig streamlabsConfig = baseConfig.StreamlabsConfig;
             if (streamlabsConfig.Enabled)
@@ -169,7 +170,7 @@ public ModeBase(
                         _logger.LogWarning("Multiple chats configured, using {Chat} for donation token whispers", chatName);
                     DonationHandler donationHandler = new(loggerFactory.CreateLogger<DonationHandler>(),
                         repos.DonationRepo, repos.UserRepo, repos.TokensBank, chat, overlayConnection,
-                        baseConfig.DonorBadgeCents);
+                        repos.ChattersSnapshotsRepo, baseConfig.DonorBadgeCents);
                     StreamlabsClient streamlabsClient = new(loggerFactory.CreateLogger<StreamlabsClient>(),
                         streamlabsConfig.AccessToken);
                     _donationsWorker = new DonationsWorker(loggerFactory, streamlabsConfig.PollingInterval,

TPP.Model/User.cs

@@ -76,6 +76,8 @@ public class User : PropertyEquatable<User>
 
     public Instant? TimeoutExpiration { get; init; }
     public bool Banned { get; init; }
+    public bool IsBot { get; init; }
+    public bool IsCaptchaSuspended { get; init; }
 
     public bool DonorBadge { get; init; }
 

TPP.Persistence.MongoDB/Repos/ChattersSnapshotsRepo.cs

@@ -1,8 +1,10 @@
 using System.Collections.Immutable;
+using System.Linq;
 using System.Threading.Tasks;
 using MongoDB.Bson.Serialization;
 using MongoDB.Bson.Serialization.IdGenerators;
 using MongoDB.Driver;
+using MongoDB.Driver.Linq;
 using NodaTime;
 using TPP.Model;
 using TPP.Persistence.MongoDB.Serializers;
@@ -54,4 +56,10 @@ public async Task<ChattersSnapshot> LogChattersSnapshot(
         await Collection.InsertOneAsync(item);
         return item;
     }
+
+    public async Task<ChattersSnapshot?> GetRecentChattersSnapshot(Instant from, Instant to) =>
+        await Collection.AsQueryable()
+            .Where(snapshot => snapshot.Timestamp >= from && snapshot.Timestamp <= to)
+            .OrderByDescending(snapshot => snapshot.Timestamp)
+            .FirstOrDefaultAsync();
 }

TPP.Persistence.MongoDB/Repos/UserRepo.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Collections.Immutable;
+using System.Linq;
 using System.Linq.Expressions;
 using System.Threading.Tasks;
 using MongoDB.Bson.Serialization;
@@ -59,6 +60,8 @@ static UserRepo()
                 cm.MapProperty(u => u.LoyaltyLeague).SetElementName("loyalty_tier");
                 cm.MapProperty(u => u.SubscriptionUpdatedAt).SetElementName("subscription_updated_at");
                 cm.MapProperty(u => u.Banned).SetElementName("banned");
+                cm.MapProperty(u => u.IsBot).SetElementName("is_bot");
+                cm.MapProperty(u => u.IsCaptchaSuspended).SetElementName("captcha_suspended");
                 cm.MapProperty(u => u.TimeoutExpiration).SetElementName("timeout_expiration");
                 cm.MapProperty(u => u.Roles).SetElementName("roles")
                     .SetDefaultValue(new HashSet<Role>());
@@ -194,6 +197,12 @@ public async Task<User> RecordUser(UserInfo userInfo)
         public async Task<User?> FindById(string userId) =>
             await Collection.Find(u => u.Id == userId).FirstOrDefaultAsync();
 
+        public async Task<List<User>> FindByIds(IReadOnlyList<string> userIds) =>
+            await Collection.Find(u => userIds.Contains(u.Id)).ToListAsync();
+
+        public async Task<List<User>> FindByIdsEligibleForHandouts(IReadOnlyList<string> userIds) =>
+            await Collection.Find(u => userIds.Contains(u.Id) && !u.Banned && !u.IsBot && !u.IsCaptchaSuspended).ToListAsync();
+
         public async Task<User?> FindByDisplayName(string displayName) =>
             await Collection.Find(u => u.TwitchDisplayName == displayName).SortByDescending(u => u.LastActiveAt).FirstOrDefaultAsync();
 

TPP.Persistence/IBank.cs

@@ -29,6 +29,7 @@ public static class TransactionType
         public const string DonationGive = "donation_give";
         public const string DonationReceive = "donation_recieve"; // typo kept for backwards-compatibility for now
         public const string DonationTokens = "donation_tokens";
+        public const string DonationRandomlyDistributedTokens = "donation_randomly_distributed_tokens";
         public const string Match = "match";
         public const string Subscription = "subscription";
         public const string SubscriptionGift = "subscription gift";

TPP.Persistence/IChattersSnapshotsRepo.cs

@@ -12,4 +12,6 @@ Task<ChattersSnapshot> LogChattersSnapshot(
         IImmutableList<string> chatterIds,
         string channel,
         Instant timestamp);
+
+    Task<ChattersSnapshot?> GetRecentChattersSnapshot(Instant from, Instant to);
 }

TPP.Persistence/IUserRepo.cs

@@ -10,6 +10,8 @@ public interface IUserRepo
     {
         public Task<User> RecordUser(UserInfo userInfo);
         public Task<User?> FindById(string userId);
+        public Task<List<User>> FindByIds(IReadOnlyList<string> userIds);
+        public Task<List<User>> FindByIdsEligibleForHandouts(IReadOnlyList<string> userIds);
         public Task<User?> FindBySimpleName(string simpleName);
         public Task<User?> FindByDisplayName(string displayName);
         public Task<List<User>> FindAllByPokeyenUnder(long yen);