Рефакторинг нотификаций

Joinrpg.sln

@@ -128,6 +128,12 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "JoinRpg.Web.Claims", "src\J
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JoinRpg.Dal.CommonEfCore", "src\JoinRpg.Dal.CommonEfCore\JoinRpg.Dal.CommonEfCore.csproj", "{A2E3883C-6DFB-4523-9F85-3F0FEC7CD79F}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JoinRpg.Services.Notifications", "src\JoinRpg.Services.Notifications\JoinRpg.Services.Notifications.csproj", "{14708BF4-BFC4-4059-A15F-F2F3803C8D4E}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JoinRpg.Services.Notifications.Tests", "src\JoinRpg.Services.Notifications.Tests\JoinRpg.Services.Notifications.Tests.csproj", "{579CCBF4-BD18-468A-ADFC-BED19EB13224}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JoinRpg.Dal.Notifications", "src\JoinRpg.Dal.Notifications\JoinRpg.Dal.Notifications.csproj", "{6CD906B0-F1BA-467F-9866-D36FAD7F542B}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -330,6 +336,18 @@ Global
 		{A2E3883C-6DFB-4523-9F85-3F0FEC7CD79F}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{A2E3883C-6DFB-4523-9F85-3F0FEC7CD79F}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{A2E3883C-6DFB-4523-9F85-3F0FEC7CD79F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{14708BF4-BFC4-4059-A15F-F2F3803C8D4E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{14708BF4-BFC4-4059-A15F-F2F3803C8D4E}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{14708BF4-BFC4-4059-A15F-F2F3803C8D4E}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{14708BF4-BFC4-4059-A15F-F2F3803C8D4E}.Release|Any CPU.Build.0 = Release|Any CPU
+		{579CCBF4-BD18-468A-ADFC-BED19EB13224}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{579CCBF4-BD18-468A-ADFC-BED19EB13224}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{579CCBF4-BD18-468A-ADFC-BED19EB13224}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{579CCBF4-BD18-468A-ADFC-BED19EB13224}.Release|Any CPU.Build.0 = Release|Any CPU
+		{6CD906B0-F1BA-467F-9866-D36FAD7F542B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{6CD906B0-F1BA-467F-9866-D36FAD7F542B}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{6CD906B0-F1BA-467F-9866-D36FAD7F542B}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{6CD906B0-F1BA-467F-9866-D36FAD7F542B}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -375,6 +393,9 @@ Global
 		{340E5B6E-2C23-43B6-869D-D9BF807E69D3} = {B8A1A61E-CD98-49F5-98BA-30B0869576B1}
 		{971478A6-DD20-47DA-8B9C-DFC7E19A9240} = {B3BB8906-62C8-44A0-822C-566A144C05AE}
 		{A2E3883C-6DFB-4523-9F85-3F0FEC7CD79F} = {B8A1A61E-CD98-49F5-98BA-30B0869576B1}
+		{14708BF4-BFC4-4059-A15F-F2F3803C8D4E} = {4D578A76-DAE7-463A-9ABE-268E4D8488DA}
+		{579CCBF4-BD18-468A-ADFC-BED19EB13224} = {8BB208B6-13CD-4942-B3AC-037BDF858727}
+		{6CD906B0-F1BA-467F-9866-D36FAD7F542B} = {B8A1A61E-CD98-49F5-98BA-30B0869576B1}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {1A5E4F92-A0F4-4350-92C7-361C88D58588}

docs/adr003-notifications.md

@@ -0,0 +1,19 @@
+# Notifications
+
+Сейчас система выглядит так:
+1. Доменный сервис вызывает EmailService, в котором описаны все возможные email, передает ему бизнес-сущности DataModel
+1. Тот собирает все нужные данные, подписки и формирует шаблон mailgun, в который в качестве переменных передаются все нужные поля, которые изменяются для каждого получателя, например имя, измененные поля (т.к. не все видят)
+1. Дальше через Common.EmailSending → Mailgun_csharp письмо уходит в mailgun, там шаблонизируется, ставится в очередь и отправляется получателям.
+
+## Проблемы
+
+1. Шаблон накладывается на стороне mailgun, соответственно нам нужно самостоятельно реализовать шаблонизацию, чтобы отправлять все в телегу
+1. Нам нужно отказаться от mailgun (иностранный сервис, а аналог например в Яндексе, не предусматривают шаблонизацию, просто позволяет поставить одно письмо в очередь)
+
+## Решение
+1. Создаем сервис Notifications. Он принимает на вход сообщение-шаблон + id пользователей, кому отослать + поля для них. Шаблонизирует, загружает настройки отправления в , и записывает в БД записи (1 ряд = 1 сообщение 1 пользователю по 1 каналу связи).
+1. Пока что реализуем один канал - inbox пользователя, на нем отрабатываем все, а также старую отсылку через mailgun (по прежнему на стороне сервиса)
+1. Переключаем все на новый интерфес (Notifications) вместо Common.EmailSending. Убеждаемся, что новая шаблонизация прилично работает.
+1. Добавляем джобу, которая реализует отсылку через телегу, добавляем телегу в список каналов, делаем у пользователя настройку через что отправлять (через телегу, через email, оба способа). Привязка аккаунта к телеге и получение разрешения на отправку сообщений от бота уже реализовано
+1. Переключаем отправку email вместо шаблонизации на стороне mailgun на нашу шаблонизацию.
+1. Меняем отсылку email вместо mailgun на яндексовский сервис без шаблонизации.

src/JoinRpg.Dal.Impl/Repositories/UserInfoRepository.cs

@@ -95,4 +95,26 @@ Task<UserAvatar> IUserRepository.LoadAvatar(AvatarIdentification userAvatarId)
 
         return res.Select(a => (new UserIdentification(a.UserId), new AvatarIdentification(a.UserAvatarId))).ToArray();
     }
+
+    async Task<UserNotificationInfoDto[]> IUserRepository.GetUsersNotificationInfo(UserIdentification[] userIds)
+    {
+        int[] ids = [.. userIds.Select(u => u.Value)];
+        var users = await ctx.Set<User>()
+            .Include(u => u.Auth)
+            .Include(u => u.ExternalLogins)
+            .Include(u => u.Extra)
+            .Where(u => ids.Contains(u.UserId))
+            .ToArrayAsync();
+
+        // TODO Here we need to load only required fields 
+        return [..
+            users.Select(u => new UserNotificationInfoDto
+            {
+                UserId = new(u.UserId),
+                DisplayName = u.ExtractDisplayName(),
+                Email = new Email(u.Email),
+                TelegramId = u.TryGetTelegramId(),
+            })
+            ];
+    }
 }

src/JoinRpg.Dal.Impl/Repositories/UserTransformationExtensions.cs

@@ -18,4 +18,19 @@ public static UserDisplayName ExtractDisplayName(this User user)
     {
         return UserDisplayName.Create(user.ExtractFullName(), new Email(user.Email));
     }
+
+    public static UserExternalLogin? TryGetExternalLoginByProviderId(this User user, string providerId)
+    {
+        return user.ExternalLogins.SingleOrDefault(l => l.Provider.Equals(providerId, StringComparison.InvariantCultureIgnoreCase));
+    }
+
+    public static TelegramId? TryGetTelegramId(this User user)
+    {
+        var elogin = user.TryGetExternalLoginByProviderId("telegram");
+        if (elogin == null)
+        {
+            return null;
+        }
+        return new TelegramId(long.Parse(elogin.Key), PrefferedName.FromOptional(user.Extra?.Telegram));
+    }
 }

src/JoinRpg.Dal.Notifications/GlobalUsings.cs

@@ -0,0 +1,5 @@
+global using System.ComponentModel.DataAnnotations;
+global using JoinRpg.Data.Write.Interfaces.Notifications;
+global using JoinRpg.PrimitiveTypes.Notifications;
+global using Microsoft.EntityFrameworkCore;
+global using Microsoft.Extensions.Logging;

src/JoinRpg.Dal.Notifications/JoinRpg.Dal.Notifications.csproj

@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net9.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+
+  <ItemGroup>
+    <ProjectReference Include="..\JoinRpg.Dal.CommonEfCore\JoinRpg.Dal.CommonEfCore.csproj" />
+    <ProjectReference Include="..\JoinRpg.Data.Write.Interfaces\JoinRpg.Data.Write.Interfaces.csproj" />
+  </ItemGroup>
+
+</Project>

src/JoinRpg.Dal.Notifications/NotificationMessage.cs

@@ -0,0 +1,18 @@
+namespace JoinRpg.Dal.Notifications;
+
+[Index(nameof(RecepientUserId))]
+[Index(nameof(InitiatorUserId))]
+internal class NotificationMessage
+{
+    public int NotificationMessageId { get; set; }
+    [MaxLength(1024)]
+    public required string Header { get; set; }
+    public required string Body { get; set; }
+    public required int InitiatorUserId { get; set; }
+    [MaxLength(1024)]
+    public required string InitiatorAddress { get; set; }
+
+    public required int RecepientUserId { get; set; }
+
+    public virtual HashSet<NotificationMessageChannel> NotificationMessageChannels { get; set; } = [];
+}

src/JoinRpg.Dal.Notifications/NotificationMessageChannel.cs

@@ -0,0 +1,16 @@
+namespace JoinRpg.Dal.Notifications;
+
+[Index(nameof(Channel), nameof(NotificationMessageChannelId), IsUnique = true)]
+[Index(nameof(NotificationMessageStatus), nameof(Channel), nameof(NotificationMessageChannelId))]
+internal class NotificationMessageChannel
+{
+    public int NotificationMessageChannelId { get; set; }
+    public required NotificationMessage NotificationMessage { get; set; }
+
+    public int NotificationMessageId { get; set; }
+
+    public required NotificationChannel Channel { get; set; }
+    [MaxLength(1024)]
+    public required string ChannelSpecificValue { get; set; }
+    public required NotificationMessageStatus NotificationMessageStatus { get; set; }
+}

src/JoinRpg.Dal.Notifications/NotificationMessageStatus.cs

@@ -0,0 +1,9 @@
+namespace JoinRpg.Dal.Notifications;
+
+public enum NotificationMessageStatus
+{
+    Queued = 0,
+    Sending = 1,
+    Sent = 2,
+    Failed = 3,
+}

src/JoinRpg.Dal.Notifications/NotificationsDataDbContext.cs

@@ -0,0 +1,12 @@
+using JoinRpg.Dal.CommonEfCore;
+
+namespace JoinRpg.Dal.Notifications;
+
+public class NotificationsDataDbContext(DbContextOptions<NotificationsDataDbContext> options) : JoinPostgreSqlEfContextBase(options)
+{
+    internal DbSet<NotificationMessage> Notifications { get; set; } = null!;
+
+    internal DbSet<NotificationMessageChannel> NotificationMessageChannels { get; set; } = null!;
+
+
+}

src/JoinRpg.Dal.Notifications/NotificationsRepository.cs

@@ -0,0 +1,127 @@
+using System.Diagnostics.Metrics;
+
+namespace JoinRpg.Dal.Notifications;
+
+internal class NotificationsRepository : INotificationRepository
+{
+    private readonly Counter<int> lostRaceCounter;
+    private readonly Counter<int> successRaceCounter;
+    private readonly NotificationsDataDbContext dbContext;
+    private readonly ILogger<NotificationsRepository> logger;
+
+    public NotificationsRepository(
+        NotificationsDataDbContext dbContext,
+        ILogger<NotificationsRepository> logger,
+        IMeterFactory meterFactory)
+    {
+        this.dbContext = dbContext;
+        this.logger = logger;
+        var meter = meterFactory.Create("JoinRpg.Dal.Notifications.Repository");
+        lostRaceCounter = meter.CreateCounter<int>("joinRpg.dal.notifications.repository.notifications_select_lost_races");
+        successRaceCounter = meter.CreateCounter<int>("joinRpg.dal.notifications.repository.notifications_select_success_races");
+    }
+
+    async Task INotificationRepository.InsertNotifications(NotificationMessageDto[] notifications)
+    {
+        foreach (var x in notifications)
+        {
+            var message = new NotificationMessage()
+            {
+                Body = x.Body.Contents!,
+                Header = x.Header,
+                InitiatorAddress = x.InitiatorAddress,
+                InitiatorUserId = x.Initiator.Value,
+                RecepientUserId = x.Recepient.Value,
+                NotificationMessageChannels = [
+                    .. x.Channels.Select(c => new NotificationMessageChannel()
+                    {
+                        Channel = c.Channel,
+                        ChannelSpecificValue = c.ChannelSpecificValue,
+                        NotificationMessageStatus = NotificationMessageStatus.Queued,
+                        NotificationMessage = null!,
+                    })
+                    ]
+            };
+            _ = dbContext.Notifications.Add(message);
+        }
+        _ = await dbContext.SaveChangesAsync();
+
+    }
+    async Task INotificationRepository.MarkSendFailed(NotificationId id, NotificationChannel channel)
+    {
+        if (!await TrySetStatus(id.Value, channel, from: NotificationMessageStatus.Sending, to: NotificationMessageStatus.Failed))
+        {
+            logger.LogWarning("Notification {notificationId} for channel {channel} failed to set status to failed", id, channel);
+        }
+    }
+
+    async Task INotificationRepository.MarkSendSuccess(NotificationId id, NotificationChannel channel)
+    {
+        if (!await TrySetStatus(id.Value, channel, from: NotificationMessageStatus.Sending, to: NotificationMessageStatus.Sent))
+        {
+            logger.LogWarning("Notification {notificationId} for channel {channel} failed to set status to success", id, channel);
+        }
+    }
+    async Task<(NotificationId Id, NotificationMessageDto Message)?> INotificationRepository.SelectNextNotificationForSending(NotificationChannel channel)
+    {
+        var tryCount = 0;
+        while (tryCount < 5)
+        {
+            var candidate =
+                await dbContext.NotificationMessageChannels
+                .Include(c => c.NotificationMessage)
+                .ThenInclude(m => m.NotificationMessageChannels)
+                .ForChannelAndStatus(channel, NotificationMessageStatus.Queued)
+                .FirstOrDefaultAsync();
+
+            if (candidate is null)
+            {
+                return null;
+            }
+
+            if (await TrySetStatus(candidate.NotificationMessageId, channel, from: NotificationMessageStatus.Queued, to: NotificationMessageStatus.Sending))
+            {
+                successRaceCounter.Add(1);
+                return (new(candidate.NotificationMessageId), CreateNotificationMessageDto(candidate.NotificationMessage));
+            }
+
+            // lost race
+
+            logger.LogDebug("Lost race when try to acquire candidate for sending!");
+            lostRaceCounter.Add(1);
+            await Task.Delay(Random.Shared.Next(100 * tryCount));
+            tryCount++;
+        }
+        logger.LogWarning("Constantly losing race...");
+        return null;
+    }
+
+    private static NotificationMessageDto CreateNotificationMessageDto(NotificationMessage candidate)
+    {
+
+        return new NotificationMessageDto
+        {
+            Body = new DataModel.MarkdownString(candidate.Body),
+            Channels = [.. candidate.NotificationMessageChannels.Select(c => new NotificationChannelDto(c.Channel, c.ChannelSpecificValue))],
+            Header = candidate.Header,
+            Initiator = new(candidate.InitiatorUserId),
+            InitiatorAddress = new(candidate.InitiatorAddress),
+            Recepient = new(candidate.RecepientUserId),
+        };
+    }
+
+    private async Task<bool> TrySetStatus(int id, NotificationChannel channel, NotificationMessageStatus from, NotificationMessageStatus to)
+    {
+        var totalRows = await dbContext
+            .NotificationMessageChannels
+            .Where(n => n.NotificationMessageId == id)
+            .ForChannelAndStatus(channel, from)
+            .ExecuteUpdateAsync(ch => ch.SetProperty(x => x.NotificationMessageStatus, to));
+        return totalRows switch
+        {
+            0 => false,
+            1 => true,
+            _ => throw new InvalidOperationException("Unexpected — too many rows updated")
+        };
+    }
+}

src/JoinRpg.Dal.Notifications/Queries.cs

@@ -0,0 +1,13 @@
+namespace JoinRpg.Dal.Notifications;
+
+internal static class Queries
+{
+    public static IQueryable<NotificationMessageChannel> ForChannelAndStatus(
+        this IQueryable<NotificationMessageChannel> query,
+        NotificationChannel channel,
+        NotificationMessageStatus status)
+        => query
+            .Where(n => n.Channel == channel)
+            .Where(n => n.NotificationMessageStatus == status);
+
+}

src/JoinRpg.Dal.Notifications/Registrations.cs

@@ -0,0 +1,14 @@
+using JoinRpg.Dal.CommonEfCore;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+
+namespace JoinRpg.Dal.Notifications;
+public static class Registrations
+{
+    public static void AddNotificationsDal(this IServiceCollection services, IConfiguration configuration, IHostEnvironment environment)
+    {
+        services.AddJoinEfCoreDbContext<NotificationsDataDbContext>(configuration, environment, "Notifications");
+        services.AddTransient<INotificationRepository, NotificationsRepository>();
+    }
+}

src/JoinRpg.Data.Interfaces/IUserRepository.cs

@@ -8,6 +8,8 @@ public interface IUserRepository
 {
     Task<User> GetById(int id);
 
+    Task<UserNotificationInfoDto[]> GetUsersNotificationInfo(UserIdentification[] userIds);
+
     Task<User> WithProfile(int userId);
     Task<User> GetWithSubscribe(int currentUserId);
     Task<User?> GetByEmail(string email);

src/JoinRpg.Data.Interfaces/UserNotificationInfoDto.cs

@@ -0,0 +1,11 @@
+using JoinRpg.PrimitiveTypes;
+
+namespace JoinRpg.Data.Interfaces;
+
+public class UserNotificationInfoDto
+{
+    public required UserIdentification UserId { get; set; }
+    public required UserDisplayName DisplayName { get; set; }
+    public required Email Email { get; set; }
+    public required TelegramId? TelegramId { get; set; }
+}

src/JoinRpg.Data.Write.Interfaces/JoinRpg.Data.Write.Interfaces.csproj

@@ -4,8 +4,6 @@
     <AssemblyTitle>JoinRpg.Data.Write.Interfaces</AssemblyTitle>
     <Product>JoinRpg.Data.Write.Interfaces</Product>
   </PropertyGroup>
-  <ItemGroup>
-  </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\JoinRpg.Data.Interfaces\JoinRpg.Data.Interfaces.csproj" />
   </ItemGroup>

src/JoinRpg.Data.Write.Interfaces/Notifications/INotificationRepository.cs

@@ -0,0 +1,10 @@
+using JoinRpg.PrimitiveTypes.Notifications;
+
+namespace JoinRpg.Data.Write.Interfaces.Notifications;
+public interface INotificationRepository
+{
+    Task InsertNotifications(NotificationMessageDto[] notifications);
+    Task<(NotificationId Id, NotificationMessageDto Message)?> SelectNextNotificationForSending(NotificationChannel channel);
+    Task MarkSendSuccess(NotificationId id, NotificationChannel channel);
+    Task MarkSendFailed(NotificationId id, NotificationChannel channel);
+}