hrynco/hrynco-notification-service
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(push): add service-owned firebase channels and delivery

Browse Source 
Add push contracts, provider delivery and channel diagnostics. Link email channels to services and document migration and validation.

Ref: IT-1074
This commit is contained in:
Anatolii Hrynchuk
2026-10-24 12:31:35 +03:00
parent b30dba93e8
commit 1631786ce2
105 changed files with 3838 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HrynCo.NotificationService.DAL.EF/Repositories/EmailChannelRepository.cs
+9 -8
View File
@@ -15,14 +15,14 @@ internal sealed class EmailChannelRepository : NotificationBaseRepository<EmailC
public async Task<IReadOnlyList<EmailChannel>> GetAllAsync(CancellationToken ct = default)
{
var entities = await EfRepository.Get().ToListAsync(ct);
var entities = await EfRepository.Get().AsNoTracking().Include(x => x.Service).ToListAsync(ct);
return entities.Select(MapToDomain).ToList();
}
public async Task<IReadOnlyList<EmailChannel>> GetByServiceAsync(string serviceName, CancellationToken ct = default)
public async Task<IReadOnlyList<EmailChannel>> GetByServiceAsync(Guid serviceId, CancellationToken ct = default)
{
var entities = await EfRepository.Get(x => x.ServiceName == serviceName)
var entities = await EfRepository.Get(x => x.ServiceId == serviceId).Include(x => x.Service)
.OrderBy(x => x.Priority)
.ToListAsync(ct);
@@ -32,8 +32,8 @@ internal sealed class EmailChannelRepository : NotificationBaseRepository<EmailC
public async Task<IReadOnlyList<ChannelWithUsage>> GetAllWithUsageSummaryAsync(
DateOnly today, CancellationToken ct = default)
{
var rows = await EfRepository.Get()
.OrderBy(c => c.ServiceName)
var rows = await EfRepository.Get().AsNoTracking().Include(c => c.Service)
.OrderBy(c => c.Service.Name)
.ThenBy(c => c.Priority)
.Select(c => new
{
@@ -54,7 +54,7 @@ internal sealed class EmailChannelRepository : NotificationBaseRepository<EmailC
public async Task<EmailChannel?> GetByIdAsync(Guid id, CancellationToken ct = default)
{
EmailChannelEntity? entity = await EfRepository.Get().AsNoTracking().SingleOrDefaultAsync(c => c.Id == id, ct);
EmailChannelEntity? entity = await EfRepository.Get().AsNoTracking().Include(c => c.Service).SingleOrDefaultAsync(c => c.Id == id, ct);
return entity is null ? null : MapToDomain(entity);
}
@@ -79,7 +79,8 @@ internal sealed class EmailChannelRepository : NotificationBaseRepository<EmailC
return new EmailChannel
{
Id = e.Id,
ServiceName = e.ServiceName,
ServiceId = e.ServiceId,
ServiceName = e.Service.Name,
Priority = e.Priority,
EmailChannelType = e.EmailChannelType,
Settings = DeserializeSettings(e.EmailChannelType, e.SettingsJson),
@@ -97,7 +98,7 @@ internal sealed class EmailChannelRepository : NotificationBaseRepository<EmailC
return new EmailChannelEntity
{
Id = p.Id,
ServiceName = p.ServiceName,
ServiceId = p.ServiceId,
Priority = p.Priority,
EmailChannelType = p.EmailChannelType,
SettingsJson = JsonSerializer.Serialize(p.Settings, p.Settings.GetType()),
HrynCo.NotificationService.DAL.EF/Repositories/NotificationServiceRepository.cs
+9 -2
View File
@@ -38,8 +38,7 @@ internal sealed class NotificationServiceRepository(NotificationDbContext db, Te
return;
if (service.Revision != revision)
throw new EmailTemplateWriteException("The service revision changed. Read it again before deleting.");
if (await db.TemplateDefinitions.AnyAsync(d => d.ServiceId == id, ct))
throw new EmailTemplateWriteException("Remove every template definition before deleting this service, including definitions without translations.");
await EnsureServiceIsEmptyAsync(id, ct);
db.ServiceLanguages.RemoveRange(service.Languages);
db.Services.Remove(service);
removed = true;
@@ -47,6 +46,14 @@ internal sealed class NotificationServiceRepository(NotificationDbContext db, Te
return removed;
}
private async Task EnsureServiceIsEmptyAsync(Guid id, CancellationToken ct)
{
if (await db.TemplateDefinitions.AnyAsync(d => d.ServiceId == id, ct))
throw new EmailTemplateWriteException("Remove every template definition before deleting this service, including definitions without translations.");
if (await db.Providers.AnyAsync(c => c.ServiceId == id, ct) || await db.PushChannels.AnyAsync(c => c.ServiceId == id, ct))
throw new EmailTemplateWriteException("Reassign or remove every notification channel before deleting this service.");
}
public Task SaveAsync(NotificationServiceInfo model, bool isNew, CancellationToken ct = default) => mutations.ExecuteAsync(async () =>
{
Validate(model);
HrynCo.NotificationService.DAL.EF/Repositories/PushChannelRepository.cs
+60
View File
@@ -0,0 +1,60 @@
using HrynCo.NotificationService.DAL.Abstract.PushChannels;
using HrynCo.NotificationService.DAL.EF.Entities;
using Microsoft.EntityFrameworkCore;
namespace HrynCo.NotificationService.DAL.EF.Repositories;
internal sealed class PushChannelRepository(NotificationDbContext db, TemplateMutationRunner mutations) : IPushChannelRepository
{
public async Task<IReadOnlyList<PushChannel>> GetAllAsync(CancellationToken ct = default) =>
await ReadModels().OrderBy(c => c.ServiceName).ToListAsync(ct);
public Task<PushChannel?> GetByIdAsync(Guid id, CancellationToken ct = default) =>
ReadModels().SingleOrDefaultAsync(c => c.Id == id, ct);
public Task<PushChannel?> GetActiveForServiceAsync(Guid serviceId, CancellationToken ct = default) =>
ReadModels().SingleOrDefaultAsync(c => c.ServiceId == serviceId && c.IsActive, ct);
public Task<string?> GetCredentialsAsync(Guid id, CancellationToken ct = default) =>
db.PushChannels.AsNoTracking().Where(c => c.Id == id).Select(c => c.CredentialsJson).SingleOrDefaultAsync(ct);
public Task SaveAsync(PushChannel channel, string credentialsJson, bool isNew, CancellationToken ct = default) =>
mutations.ExecuteAsync(async () =>
{
await ValidateServiceAsync(channel, ct);
var entity = await LoadForSaveAsync(channel.Id, isNew, ct);
entity.ServiceId = channel.ServiceId;
entity.FirebaseProjectId = channel.FirebaseProjectId;
entity.CredentialsJson = credentialsJson;
entity.IsActive = channel.IsActive;
entity.Updated = isNew ? null : DateTimeOffset.UtcNow;
channel.Id = entity.Id;
}, ct);
private async Task ValidateServiceAsync(PushChannel channel, CancellationToken ct)
{
if (!await db.Services.AnyAsync(s => s.Id == channel.ServiceId, ct))
throw new PushChannelWriteException("Select an existing service.", false);
if (await db.PushChannels.AnyAsync(c => c.ServiceId == channel.ServiceId && c.Id != channel.Id, ct))
throw new PushChannelWriteException("This service already has a push channel. Edit that channel instead.");
}
private async Task<PushChannelEntity> LoadForSaveAsync(Guid id, bool isNew, CancellationToken ct)
{
if (!isNew)
return await db.PushChannels.SingleOrDefaultAsync(c => c.Id == id, ct)
?? throw new PushChannelWriteException("The push channel no longer exists. Reload the list.");
var entity = new PushChannelEntity
{
Id = id, FirebaseProjectId = string.Empty, CredentialsJson = "{}", Created = DateTimeOffset.UtcNow
};
db.PushChannels.Add(entity);
return entity;
}
private IQueryable<PushChannel> ReadModels() => db.PushChannels.AsNoTracking().Select(c => new PushChannel
{
Id = c.Id, ServiceId = c.ServiceId, ServiceName = c.Service.Name,
FirebaseProjectId = c.FirebaseProjectId, IsActive = c.IsActive, Created = c.Created, Updated = c.Updated
});
}
HrynCo.NotificationService.DAL.EF/Repositories/PushDeliveryRepository.cs
+36
View File
@@ -0,0 +1,36 @@
namespace HrynCo.NotificationService.DAL.EF.Repositories;
using HrynCo.NotificationService.DAL.Abstract.Push;
using HrynCo.NotificationService.DAL.EF.Entities;
using Microsoft.EntityFrameworkCore;
public sealed class PushDeliveryRepository(NotificationDbContext db) : IPushDeliveryRepository
{
public async Task<PushDeliveryOutcome> DeliverOnceAsync(Guid messageId, string serviceName,
Func<CancellationToken, Task<PushDeliveryOutcome>> deliver, CancellationToken cancellationToken)
{
// Thirty days exceeds the maximum payload lifetime; old replays cannot send expired data.
var cutoff = DateTimeOffset.UtcNow.AddDays(-30);
await db.Set<PushDeliveryEntity>().Where(x => x.CompletedAt < cutoff).ExecuteDeleteAsync(cancellationToken);
await using var transaction = await db.Database.BeginTransactionAsync(cancellationToken);
var identity = $"{serviceName}:{messageId:D}";
await db.Database.ExecuteSqlInterpolatedAsync(
$"SELECT pg_advisory_xact_lock(hashtextextended({identity}, 1074))", cancellationToken);
var previous = await db.Set<PushDeliveryEntity>().AsNoTracking()
.SingleOrDefaultAsync(x => x.MessageId == messageId && x.ServiceName == serviceName, cancellationToken);
if (previous is not null)
return new PushDeliveryOutcome(previous.Error, previous.InvalidRegistration);
// One bounded provider call while holding this message's lock prevents concurrent sends.
// A crash after FCM accepts still requires client-side message-id deduplication.
var outcome = await deliver(cancellationToken);
db.Add(new PushDeliveryEntity
{
MessageId = messageId, ServiceName = serviceName, Error = outcome.Error,
InvalidRegistration = outcome.InvalidRegistration, CompletedAt = DateTimeOffset.UtcNow
});
await db.SaveChangesAsync(cancellationToken);
await transaction.CommitAsync(cancellationToken);
return outcome;
}
}