refactor(notification-outbox): align outbox repository with cancellationToken convention
This commit is contained in:
@@ -27,22 +27,22 @@ public class NotificationOutboxRepository : INotificationOutboxRepository
|
||||
_context = context ?? throw new ArgumentNullException(nameof(context));
|
||||
}
|
||||
|
||||
public async Task<bool> InsertIfNotExistsAsync(Notification n, CancellationToken ct = default)
|
||||
public async Task<bool> InsertIfNotExistsAsync(Notification n, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var exists = await _context.Notifications
|
||||
.AnyAsync(x => x.NotificationId == n.NotificationId, ct);
|
||||
.AnyAsync(x => x.NotificationId == n.NotificationId, cancellationToken);
|
||||
if (exists)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
await _context.Notifications.AddAsync(n, ct);
|
||||
await _context.SaveChangesAsync(ct);
|
||||
await _context.Notifications.AddAsync(n, cancellationToken);
|
||||
await _context.SaveChangesAsync(cancellationToken);
|
||||
return true;
|
||||
}
|
||||
|
||||
public async Task<IReadOnlyList<Notification>> GetDueAsync(
|
||||
DateTimeOffset now, int batchSize, CancellationToken ct = default)
|
||||
DateTimeOffset now, int batchSize, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return await _context.Notifications
|
||||
.Where(n => n.Status == NotificationStatus.Pending
|
||||
@@ -51,20 +51,20 @@ public class NotificationOutboxRepository : INotificationOutboxRepository
|
||||
&& n.NextAttemptAt <= now))
|
||||
.OrderBy(n => n.CreatedAt)
|
||||
.Take(batchSize)
|
||||
.ToListAsync(ct);
|
||||
.ToListAsync(cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<Notification?> GetByIdAsync(string notificationId, CancellationToken ct = default)
|
||||
=> await _context.Notifications.FindAsync(new object[] { notificationId }, ct);
|
||||
public async Task<Notification?> GetByIdAsync(string notificationId, CancellationToken cancellationToken = default)
|
||||
=> await _context.Notifications.FindAsync(new object[] { notificationId }, cancellationToken);
|
||||
|
||||
public async Task UpdateAsync(Notification n, CancellationToken ct = default)
|
||||
public async Task UpdateAsync(Notification n, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_context.Notifications.Update(n);
|
||||
await _context.SaveChangesAsync(ct);
|
||||
await _context.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<(IReadOnlyList<Notification> Rows, int TotalCount)> QueryAsync(
|
||||
NotificationOutboxFilter filter, int pageNumber, int pageSize, CancellationToken ct = default)
|
||||
NotificationOutboxFilter filter, int pageNumber, int pageSize, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var query = _context.Notifications.AsQueryable();
|
||||
|
||||
@@ -110,45 +110,45 @@ public class NotificationOutboxRepository : INotificationOutboxRepository
|
||||
query = query.Where(n => n.CreatedAt <= to);
|
||||
}
|
||||
|
||||
var totalCount = await query.CountAsync(ct);
|
||||
var totalCount = await query.CountAsync(cancellationToken);
|
||||
|
||||
var rows = await query
|
||||
.OrderByDescending(n => n.CreatedAt)
|
||||
.Skip((pageNumber - 1) * pageSize)
|
||||
.Take(pageSize)
|
||||
.ToListAsync(ct);
|
||||
.ToListAsync(cancellationToken);
|
||||
|
||||
return (rows, totalCount);
|
||||
}
|
||||
|
||||
public async Task<int> DeleteTerminalOlderThanAsync(DateTimeOffset cutoff, CancellationToken ct = default)
|
||||
public async Task<int> DeleteTerminalOlderThanAsync(DateTimeOffset cutoff, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return await _context.Notifications
|
||||
.Where(n => TerminalStatuses.Contains(n.Status) && n.CreatedAt < cutoff)
|
||||
.ExecuteDeleteAsync(ct);
|
||||
.ExecuteDeleteAsync(cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<NotificationKpiSnapshot> ComputeKpisAsync(
|
||||
DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince, CancellationToken ct = default)
|
||||
DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
|
||||
var queueDepth = await _context.Notifications
|
||||
.CountAsync(n => n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying, ct);
|
||||
|| n.Status == NotificationStatus.Retrying, cancellationToken);
|
||||
|
||||
var stuckCount = await _context.Notifications
|
||||
.CountAsync(n => (n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying)
|
||||
&& n.CreatedAt < stuckCutoff, ct);
|
||||
&& n.CreatedAt < stuckCutoff, cancellationToken);
|
||||
|
||||
var parkedCount = await _context.Notifications
|
||||
.CountAsync(n => n.Status == NotificationStatus.Parked, ct);
|
||||
.CountAsync(n => n.Status == NotificationStatus.Parked, cancellationToken);
|
||||
|
||||
var deliveredLastInterval = await _context.Notifications
|
||||
.CountAsync(n => n.Status == NotificationStatus.Delivered
|
||||
&& n.DeliveredAt != null
|
||||
&& n.DeliveredAt >= deliveredSince, ct);
|
||||
&& n.DeliveredAt >= deliveredSince, cancellationToken);
|
||||
|
||||
// Oldest non-terminal CreatedAt. The DateTimeOffset value converter makes a SQL
|
||||
// Min aggregate awkward, so order ascending and take the first instead.
|
||||
@@ -157,12 +157,12 @@ public class NotificationOutboxRepository : INotificationOutboxRepository
|
||||
|| n.Status == NotificationStatus.Retrying);
|
||||
|
||||
TimeSpan? oldestPendingAge = null;
|
||||
if (await nonTerminal.AnyAsync(ct))
|
||||
if (await nonTerminal.AnyAsync(cancellationToken))
|
||||
{
|
||||
var oldestCreatedAt = await nonTerminal
|
||||
.OrderBy(n => n.CreatedAt)
|
||||
.Select(n => n.CreatedAt)
|
||||
.FirstAsync(ct);
|
||||
.FirstAsync(cancellationToken);
|
||||
oldestPendingAge = now - oldestCreatedAt;
|
||||
}
|
||||
|
||||
@@ -174,6 +174,6 @@ public class NotificationOutboxRepository : INotificationOutboxRepository
|
||||
OldestPendingAge: oldestPendingAge);
|
||||
}
|
||||
|
||||
public async Task<int> SaveChangesAsync(CancellationToken ct = default)
|
||||
=> await _context.SaveChangesAsync(ct);
|
||||
public async Task<int> SaveChangesAsync(CancellationToken cancellationToken = default)
|
||||
=> await _context.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user