Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -61,9 +61,8 @@ public async Task<IChannel> GetChannelAsync()
}

// Create new channel
var channel = await _channelProvider.GetChannelAsync() ?? throw new InvalidOperationException("Channel cannot be null.");
await EnsureQueueDeclaredAsync(channel);
return channel;
await EnsureQueueDeclaredAsync();
return await _channelProvider.GetChannelAsync() ?? throw new InvalidOperationException("Channel cannot be null.");
}
catch
{
Expand Down Expand Up @@ -180,7 +179,7 @@ public void ReturnChannel(IChannel channel)
}
}

private async Task EnsureQueueDeclaredAsync(IChannel channel)
private async Task EnsureQueueDeclaredAsync()
{
if (_queueDeclared) return;

Expand All @@ -189,7 +188,7 @@ private async Task EnsureQueueDeclaredAsync(IChannel channel)
{
if (_queueDeclared) return;

await DeclareQueueAsync(channel);
await DeclareQueueAsync();
_queueDeclared = true;
}
catch (Exception ex)
Expand All @@ -203,84 +202,141 @@ private async Task EnsureQueueDeclaredAsync(IChannel channel)
}
}

private async Task DeclareQueueAsync(IChannel channel)
// Every declaration runs on its own short-lived channel: a 406 PRECONDITION_FAILED closes the
// channel it happened on, so it must never be a channel handed back to a publisher or consumer,
// nor one reused for the next declaration.
private async Task DeclareQueueAsync()
{
var dlxName = $"{_queueName}.dlx";
var dlqName = $"{_queueName}{QueueingConstants.DeadletterAddition}";

// Main topology (DLX, DLQ, main queue) is required for publishing and consuming.
try
{
var dlxName = $"{_queueName}.dlx";
var dlqName = $"{_queueName}{QueueingConstants.DeadletterAddition}";
var retryExchange = $"{_queueName}{QueueingConstants.RetryAddition}";

// Ensure DLX exchange exists
await channel.ExchangeDeclareAsync(dlxName, ExchangeType.Direct, durable: true, autoDelete: false, arguments: null);
await WithDeclarationChannelAsync(async channel =>
{
await channel.ExchangeDeclareAsync(dlxName, ExchangeType.Direct, durable: true, autoDelete: false, arguments: null);

// Ensure DLQ exists and is bound to DLX
await channel.QueueDeclareAsync(dlqName, durable: true, exclusive: false, autoDelete: false,
arguments: new Dictionary<string, object?>
{
{ "x-queue-type", "quorum" },
{ "x-overflow", "reject-publish" }
});
await channel.QueueBindAsync(dlqName, dlxName, dlqName, arguments: null);
await channel.QueueDeclareAsync(dlqName, durable: true, exclusive: false, autoDelete: false,
arguments: new Dictionary<string, object?>
{
{ "x-queue-type", "quorum" },
{ "x-overflow", "reject-publish" }
});
await channel.QueueBindAsync(dlqName, dlxName, dlqName, arguments: null);

await channel.ExchangeDeclareAsync(retryExchange, ExchangeType.Direct, durable: true, autoDelete: false, arguments: null);
foreach (var retryDelay in QueueingConstants.RetryDelays)
{
var retryQueue = $"{retryExchange}.{GetRetryDelayKey(retryDelay)}";
await channel.QueueDeclareAsync(
retryQueue,
_queueName,
durable: true,
exclusive: false,
autoDelete: false,
arguments: new Dictionary<string, object?>
{
{ "x-queue-type", "quorum" },
{ "x-message-ttl", checked((int)retryDelay.TotalMilliseconds) },
{ "x-dead-letter-exchange", $"{_queueName}.exchange" },
{ "x-dead-letter-routing-key", _queueName },
{ "x-overflow", "reject-publish" },
{ "x-dead-letter-strategy", "at-least-once" }
{ "x-dead-letter-exchange", dlxName },
{ "x-dead-letter-routing-key", dlqName },
{ "x-dead-letter-strategy", "at-least-once" },
{ "x-delivery-limit", 10 }
});
await channel.QueueBindAsync(retryQueue, retryExchange, retryQueue, arguments: null);
}
});
}
catch (global::RabbitMQ.Client.Exceptions.OperationInterruptedException ex) when (IsInequivalentArgs(ex))
{
_logger.LogWarning(
ex,
"Queue {QueueName} exists with incompatible config. Using existing queue in compatibility mode.",
_queueName);
}

// Declare main queue with DLX args
await channel.QueueDeclareAsync(
_queueName,
durable: true,
exclusive: false,
autoDelete: false,
arguments: new Dictionary<string, object?>
{
{ "x-queue-type", "quorum" },
{ "x-overflow", "reject-publish" },
{ "x-dead-letter-exchange", dlxName },
{ "x-dead-letter-routing-key", dlqName },
{ "x-dead-letter-strategy", "at-least-once" },
{ "x-delivery-limit", 10 }
});
await WithDeclarationChannelAsync(BindToExchangeAsync);

await BindToExchangeAsync(channel);
// Retry topology is best-effort: a mismatch here must not block the main publish/consume path.
await DeclareRetryQueuesAsync();
}

private async Task DeclareRetryQueuesAsync()
{
var retryExchange = $"{_queueName}{QueueingConstants.RetryAddition}";

try
{
await WithDeclarationChannelAsync(channel =>
channel.ExchangeDeclareAsync(retryExchange, ExchangeType.Direct, durable: true, autoDelete: false, arguments: null));
}
catch (global::RabbitMQ.Client.Exceptions.OperationInterruptedException ex)
catch (Exception ex)
{
if (ex.ShutdownReason?.ReplyCode == 406 &&
ex.ShutdownReason.ReplyText.Contains("inequivalent arg"))
_logger.LogError(ex, "Failed to declare retry exchange {RetryExchange}; failed messages cannot be scheduled for retry.", retryExchange);
return;
}

foreach (var retryDelay in QueueingConstants.RetryDelays)
{
var retryQueue = $"{retryExchange}.{GetRetryDelayKey(retryDelay)}";
try
{
await WithDeclarationChannelAsync(async channel =>
{
await channel.QueueDeclareAsync(
retryQueue,
durable: true,
exclusive: false,
autoDelete: false,
arguments: new Dictionary<string, object?>
{
{ "x-queue-type", "quorum" },
{ "x-message-ttl", checked((int)retryDelay.TotalMilliseconds) },
{ "x-dead-letter-exchange", $"{_queueName}.exchange" },
{ "x-dead-letter-routing-key", _queueName },
{ "x-overflow", "reject-publish" },
{ "x-dead-letter-strategy", "at-least-once" }
});
});
}
catch (global::RabbitMQ.Client.Exceptions.OperationInterruptedException ex) when (IsInequivalentArgs(ex))
{
_logger.LogWarning(
ex,
"Queue {QueueName} exists with incompatible config. Using existing queue in compatibility mode.",
_queueName);
"Retry queue {RetryQueue} exists with incompatible config; using it as-is. Delete the queue to have it redeclared with current arguments.",
retryQueue);
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to declare retry queue {RetryQueue}.", retryQueue);
continue;
}

await BindToExchangeAsync(channel);
try
{
await WithDeclarationChannelAsync(channel =>
channel.QueueBindAsync(retryQueue, retryExchange, retryQueue, arguments: null));
}
else
catch (Exception ex)
{
throw;
_logger.LogError(ex, "Failed to bind retry queue {RetryQueue} to {RetryExchange}.", retryQueue, retryExchange);
}
}
}

private async Task WithDeclarationChannelAsync(Func<IChannel, Task> declare)
{
var channel = await _channelProvider.GetChannelAsync()
?? throw new InvalidOperationException($"Unable to obtain a channel to declare topology for queue {_queueName}.");
try
{
await declare(channel);
}
finally
{
// The underlying provider disposes the channel instead of pooling it if a 406 closed it.
_channelProvider.ReturnChannel(channel);
}
}

private static bool IsInequivalentArgs(global::RabbitMQ.Client.Exceptions.OperationInterruptedException ex) =>
ex.ShutdownReason?.ReplyCode == 406 &&
ex.ShutdownReason.ReplyText.Contains("inequivalent arg");

private static string GetRetryDelayKey(TimeSpan delay) =>
$"{(int)delay.TotalSeconds}s";

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,10 @@ public virtual async Task<List<EntityChangeWithUsername>> GetEntityChangeByTypeW
})));

var totalCount = await query.LongCountAsync(GetCancellationToken(cancellationToken));
var sortedQuery = ApplySorting(query, sorting);
var userOrder = await GetUserOrderAsync(sorting, cancellationToken);
var sortedQuery = userOrder != null
? ApplyUserSorting(query, userOrder, sorting!)
: ApplySorting(query, sorting);
var items = await sortedQuery
.Skip(skipCount)
.Take(maxResultCount)
Expand All @@ -145,6 +148,45 @@ public virtual async Task<List<EntityChangeWithUsername>> GetEntityChangeByTypeW
return (totalCount, items);
}

private static bool IsUserSorting(string? sorting, out bool bySurname, out bool descending)
{
var parts = (sorting ?? string.Empty).Trim().ToLowerInvariant().Split(' ', StringSplitOptions.RemoveEmptyEntries);
var field = parts.Length > 0 ? parts[0] : string.Empty;
bySurname = field == "usersurname";
descending = parts.Length > 1 && parts[1] == "desc";
return field is "userfirstname" or "usersurname";
}

// Identity users live in a different DbContext, so users are ranked in memory and the rank is translated to array_position.
private async Task<Guid[]?> GetUserOrderAsync(string? sorting, CancellationToken cancellationToken)
{
if (!IsUserSorting(sorting, out var bySurname, out _))
{
return null;
}

var users = await identityUserRepository.GetListAsync(
sorting: null, maxResultCount: int.MaxValue, skipCount: 0, includeDetails: false,
cancellationToken: GetCancellationToken(cancellationToken));

var ordered = bySurname
? users.OrderBy(u => u.Surname ?? string.Empty, StringComparer.OrdinalIgnoreCase).ThenBy(u => u.Name ?? string.Empty, StringComparer.OrdinalIgnoreCase)
: users.OrderBy(u => u.Name ?? string.Empty, StringComparer.OrdinalIgnoreCase).ThenBy(u => u.Surname ?? string.Empty, StringComparer.OrdinalIgnoreCase);

return ordered.Select(u => u.Id).ToArray();
}

private static IOrderedQueryable<AuditLogEntityChange> ApplyUserSorting(
IQueryable<AuditLogEntityChange> query,
Guid[] userOrder,
string sorting)
{
IsUserSorting(sorting, out _, out var descending);
return descending
? query.OrderByDescending(x => x.UserId == null ? -1 : Array.IndexOf(userOrder, x.UserId.Value)).ThenByDescending(x => x.ChangeTime)
: query.OrderBy(x => x.UserId == null ? -1 : Array.IndexOf(userOrder, x.UserId.Value)).ThenByDescending(x => x.ChangeTime);
}

private static IOrderedQueryable<AuditLogEntityChange> ApplySorting(
IQueryable<AuditLogEntityChange> query,
string? sorting)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ public async Task<ActionResult<ResolvedRecipientsDto>> GetResolvedTemplateRecipi

if (string.Equals(identifier, "AdditionalContact", StringComparison.OrdinalIgnoreCase))
{
emailAddresses.UnionWith(await _scheduledNotificationHelper.GetAdditionalContactEmailsAsync(applicationId.Value));
emailAddresses.UnionWith(await scheduledNotificationHelper.GetAdditionalContactEmailsAsync(applicationId.Value));
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,12 +41,56 @@
padding-right: 0;
flex-grow: 1;
flex-shrink: 1;
flex-basis: 0;
flex-basis: auto;
min-width: 0;
width: auto;
max-height: none;
}

/* The Endpoints stylesheet bundled on this page forces #AbpContentToolbar visible;
keep it tight so the admin sections sit directly beneath it. */
#AbpContentToolbar {
padding: 0;
margin-top: 10px;
}

#reconciliation-report-div,
#audit-logs-div,
#audit-log-settings-div {
margin-top: 0;
}

#audit-logs-div .reconciliation-filter-inputs {
align-items: center;
}

#audit-logs-div .reconciliation-filter-inputs > div:has(> #audit-search) {
display: flex;
align-items: center;
align-self: center;
width: 250px;
height: auto;
margin: 0 10px;
}

#audit-logs-div .tbl-search-label {
display: none;
}

#audit-search,
#audit-entity-type {
height: 38px;
box-sizing: border-box;
}

#audit-search {
display: block;
width: 100%;
padding: 6px 12px;
margin-top: 0;
align-self: flex-end;
}

#audit-logs-div .reconciliation-filter-buttons {
align-items: flex-end !important;
margin-left: auto;
Expand Down
Loading
Loading