implement donations

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/Chat/TwitchEventSubChat.cs

@@ -14,6 +14,7 @@
 using TPP.Core.Overlay;
 using TPP.Core.Overlay.Events;
 using TPP.Core.Utils;
+using TPP.Model;
 using TPP.Persistence;
 using TPP.Twitch.EventSub;
 using TPP.Twitch.EventSub.Notifications;
@@ -507,7 +508,7 @@ private async Task ChannelSubscribeReceived(ChannelSubscribe channelSubscribe)
         await _overlayConnection.Send(new NewSubscriber
         {
             User = subscriptionInfo.Subscriber,
-            Emotes = subscriptionInfo.Emotes.Select(EmoteInfo.FromOccurence).ToImmutableList(),
+            Emotes = subscriptionInfo.Emotes.Select(e => EmoteInfo.FromIdAndCode(e.Id, e.Code)).ToImmutableList(),
             SubMessage = subscriptionInfo.Message,
             ShareSub = true,
         }, CancellationToken.None);
@@ -545,7 +546,7 @@ private async Task ChannelSubscriptionMessageReceived(ChannelSubscriptionMessage
         await _overlayConnection.Send(new NewSubscriber
         {
             User = subscriptionInfo.Subscriber,
-            Emotes = subscriptionInfo.Emotes.Select(EmoteInfo.FromOccurence).ToImmutableList(),
+            Emotes = subscriptionInfo.Emotes.Select(e => EmoteInfo.FromIdAndCode(e.Id, e.Code)).ToImmutableList(),
             SubMessage = subscriptionInfo.Message,
             ShareSub = true,
         }, CancellationToken.None);

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/Commands/Definitions/DonationCommands.cs

@@ -0,0 +1,43 @@
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using NodaTime;
+using TPP.Model;
+using TPP.Persistence;
+
+namespace TPP.Core.Commands.Definitions;
+
+public class DonationCommands(
+    IDonationRepo donationRepo,
+    IClock clock
+) : ICommandCollection
+{
+    public IEnumerable<Command> Commands =>
+    [
+        new("donationrecords", DonationRecords)
+        {
+            Aliases = ["tiprecords"],
+            Description =
+                "Shows the current donation records. " +
+                "Check !support for how to tip, and get bonus token for every record you break!"
+        }
+    ];
+
+    public async Task<CommandResult> DonationRecords(CommandContext context)
+    {
+        SortedDictionary<DonationRecordBreakType, Donation> recordBreaks =
+            await donationRepo.GetRecordDonations(clock.GetCurrentInstant());
+
+        IEnumerable<string> recordStrings = DonationRecordBreaks.Types
+            .Select(recordBreakType =>
+            {
+                Donation? donation = recordBreaks.GetValueOrDefault(recordBreakType);
+                string x = donation == null
+                    ? "no donation"
+                    : "$" + (donation.Cents / 100f).ToString("F2") + " by " + donation.UserName;
+                return recordBreakType.Name + ": " + x;
+            });
+        string response = "Current donation records: " + string.Join(", ", recordStrings);
+        return new CommandResult { Response = response };
+    }
+}

TPP.Core/Configuration/BaseConfig.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Immutable;
 using System.ComponentModel;
 using NodaTime;
@@ -50,6 +51,14 @@ public sealed class BaseConfig : ConfigBase, IRootConfig
     public Duration AdvertisePollsInterval { get; init; } = Duration.FromHours(1);
 
     public ImmutableHashSet<TppFeatures> DisabledFeatures { get; init; } = ImmutableHashSet<TppFeatures>.Empty;
+
+    [Description("Donation handling via Streamlabs")]
+    public StreamlabsConfig StreamlabsConfig { get; init; } = new();
+
+    /// How many cents donated give you 1 token. Smaller number = more tokens per money
+    public int CentsPerToken { get; init; } = 50;
+    /// How many cents donated total before a user gets the donor badge
+    public int DonorBadgeCents { get; init; } = 20000;
 }
 
 /// <summary>
@@ -79,3 +88,10 @@ public sealed class DiscordLoggingConfig : ConfigBase
     public string WebhookToken { get; init; } = "";
     public LogEventLevel MinLogLevel { get; init; } = LogEventLevel.Warning;
 }
+
+public sealed class StreamlabsConfig : ConfigBase
+{
+    public bool Enabled { get; init; } = false;
+    public string AccessToken { get; init; } = "";
+    public TimeSpan PollingInterval { get; init; } = TimeSpan.FromMinutes(1);
+}

TPP.Core/Configuration/ConnectionConfig.cs

@@ -69,6 +69,7 @@ public enum SuppressionType { Whisper, Message, Command }
         public CaseInsensitiveImmutableHashSet SuppressionOverrides { get; init; } = new([]);
 
         public Duration? GetChattersInterval { get; init; } = Duration.FromMinutes(5);
+        public Duration? GetEmotesInterval { get; init; } = Duration.FromHours(1);
     }
 
     public sealed class Simulation : ConnectionConfig

TPP.Core/DonationHandler.cs

@@ -0,0 +1,189 @@
+using System;
+using System.Collections.Generic;
+using System.Collections.Immutable;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using NodaTime;
+using TPP.Core.Chat;
+using TPP.Core.Overlay;
+using TPP.Core.Overlay.Events;
+using TPP.Core.Streamlabs;
+using TPP.Model;
+using TPP.Persistence;
+using static TPP.Common.Utils.StringExtensions;
+
+namespace TPP.Core;
+
+public class DonationHandler(
+    ILogger<DonationHandler> logger,
+    IDonationRepo donationRepo,
+    IUserRepo userRepo,
+    IBank<User> tokensBank,
+    IMessageSender messageSender,
+    OverlayConnection overlayConnection,
+    IChattersSnapshotsRepo chattersSnapshotsRepo,
+    TwitchEmotesLookup? twitchEmotesLookup,
+    int centsPerToken,
+    int donorBadgeCents)
+{
+    public record NewDonation(
+        long Id,
+        Instant CreatedAt,
+        string Username,
+        decimal Amount,
+        string Currency,
+        string? Message)
+    {
+        public static NewDonation FromStreamlabs(StreamlabsClient.Donation donation) => new(
+            Id: donation.DonationId,
+            CreatedAt: donation.CreatedAt,
+            Username: donation.Name,
+            Amount: donation.Amount,
+            Currency: donation.Currency,
+            Message: donation.Message
+        );
+    }
+
+    public async Task Process(NewDonation donation)
+    {
+        if (await donationRepo.FindDonation(donation.Id) is { } existingDonation)
+        {
+            logger.LogDebug("Skipping donation because it already exists in the database. " +
+                            "Donation: {Donation}, existing donation: {DbDonation}", donation, existingDonation);
+            return;
+        }
+        // We only support processing USD.
+        // Streamlabs API uses apostrophes to indicate that it may have converted the amount to USD from its original currency.
+        if (donation.Currency is not ("'USD'" or "USD"))
+        {
+            logger.LogError("Skipping donation because its amount must be in USD. Donation: {Donation}", donation);
+            return;
+        }
+        int cents = (int)Math.Round(donation.Amount * 100, 0);
+
+        User? donor = await userRepo.FindBySimpleName(donation.Username.ToLower());
+        if (donor is null)
+            // Warn, but store it in the DB anyway
+            logger.LogWarning("Could not find a user for donor: {Donor}", donation.Username);
+
+        logger.LogInformation("New donation: {Donation}", donation);
+
+        await donationRepo.InsertDonation(
+            donationId: donation.Id,
+            createdAt: donation.CreatedAt,
+            userName: donation.Username,
+            userId: donor?.Id,
+            cents: cents,
+            message: donation.Message
+        );
+
+        var recordBreaks = await GetRecordBreaks(donation.Id, donation.CreatedAt);
+        DonationTokens tokens = GetDonationTokens(cents, recordBreaks);
+        if (donor != null)
+        {
+            await UpdateHasDonationBadge(donor);
+            await GivenTokensToDonorAndNotifyThem(donor, donation.Id, tokens);
+        }
+        await RandomlyDistributeTokens(donation.CreatedAt, donation.Id, donation.Username, tokens.Total());
+        await overlayConnection.Send(new NewDonationEvent
+        {
+            Emotes = twitchEmotesLookup?.FindEmotesInText(donation.Message ?? "") ?? [],
+            RecordDonations = new Dictionary<int, List<string>>
+            {
+                [cents] = recordBreaks.Select(recordBreak => recordBreak.Name).ToList()
+            },
+            Donation = new NewDonationEvent.DonationInfo
+            {
+                Username = donation.Username,
+                Cents = cents,
+                Message = donation.Message
+            }
+        }, CancellationToken.None);
+    }
+
+    private async Task UpdateHasDonationBadge(User user)
+    {
+        HashSet<string> userIds = [user.Id];
+        bool hasCentsRequired = (await donationRepo.GetCentsPerUser(donorBadgeCents, userIds)).ContainsKey(user.Id);
+        logger.LogDebug("User {Username} should have donor badge now? {HasDonorBadge}", user.Name, hasCentsRequired);
+        await userRepo.SetHasDonorBadge(user, hasCentsRequired);
+    }
+
+    record DonationTokens(int Base, int Bonus)
+    {
+        public int Total() => Base + Bonus;
+    }
+
+    /// Assumes the donation has already been persisted.
+    private async Task<ImmutableSortedSet<DonationRecordBreakType>> GetRecordBreaks(long donationId, Instant createdAt)
+    {
+        return (await donationRepo.GetRecordDonations(createdAt))
+            .Where(kvp => kvp.Value.DonationId == donationId)
+            .Select(kvp => kvp.Key)
+            .ToImmutableSortedSet();
+    }
+
+    /// Calculated a donation's reward tokens, which consists of some base tokens per cents,
+    /// plus bonus tokens obtained from donation record breaks.
+    private DonationTokens GetDonationTokens(int cents, ISet<DonationRecordBreakType> recordBreaks)
+    {
+        int baseTokens = cents / centsPerToken;
+        int bonusTokens = recordBreaks.Sum(recordBreakType => recordBreakType.TokenWinning);
+
+        return new DonationTokens(baseTokens, bonusTokens);
+    }
+
+    private async Task GivenTokensToDonorAndNotifyThem(User user, long donationId, DonationTokens tokens)
+    {
+        var additionalData = new Dictionary<string, object?>
+        {
+            ["donation"] = donationId,
+            ["donation_base_tokens"] = tokens.Base,
+            ["donation_bonus_tokens"] = tokens.Bonus,
+        };
+        var transaction = new Transaction<User>(user, tokens.Total(), TransactionType.DonationTokens, additionalData);
+        await tokensBank.PerformTransaction(transaction);
+
+        string message = tokens.Bonus > 0
+            ? $"You got T{tokens.Base} + T{tokens.Bonus} from record breaks for your donation!"
+            : $"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

@@ -0,0 +1,49 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using TPP.Core.Streamlabs;
+using TPP.Model;
+using TPP.Persistence;
+
+namespace TPP.Core;
+
+public sealed class DonationsWorker(
+    ILoggerFactory loggerFactory,
+    TimeSpan pollingInterval,
+    StreamlabsClient streamlabsClient,
+    IDonationRepo donationRepo,
+    DonationHandler donationHandler
+) : IWithLifecycle
+{
+    private readonly ILogger<ChattersWorker> _logger = loggerFactory.CreateLogger<ChattersWorker>();
+
+    public async Task Start(CancellationToken cancellationToken)
+    {
+        try { await Task.Delay(pollingInterval, cancellationToken); }
+        catch (OperationCanceledException) { return; }
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            try
+            {
+                Donation? mostRecentDonation = await donationRepo.GetMostRecentDonation();
+                _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));
+                foreach (var donation in donations.OrderBy(d => d.CreatedAt)) // process in chronological order
+                    await donationHandler.Process(DonationHandler.NewDonation.FromStreamlabs(donation));
+            }
+            catch (Exception e)
+            {
+                _logger.LogError(e, "Failed polling for new donations");
+            }
+
+            try { await Task.Delay(pollingInterval, cancellationToken); }
+            catch (OperationCanceledException) { break; }
+        }
+    }
+}