Publish application layer as a NuGet

.github/workflows/alpha-release.yml

@@ -119,11 +119,16 @@ jobs:
           echo "version=$ALPHA_VERSION" >> $GITHUB_OUTPUT
           echo "Generated version: $ALPHA_VERSION"
 
-      - name: Pack
+      - name: Pack CLI
         env:
           VERSION: ${{ steps.version.outputs.version }}
         run: dotnet pack src/ServiceBusToolset.CLI -c Release --no-build -o ./artifacts -p:Version="$VERSION"
 
+      - name: Pack Application
+        env:
+          VERSION: ${{ steps.version.outputs.version }}
+        run: dotnet pack src/ServiceBusToolset.Application -c Release --no-build -o ./artifacts -p:Version="$VERSION"
+
       - name: Push to NuGet
         env:
           NUGET_API_KEY: ${{ secrets.NUGET_API_KEY }}
@@ -133,3 +138,4 @@ jobs:
             exit 0
           fi
           dotnet nuget push ./artifacts/*.nupkg --api-key "$NUGET_API_KEY" --source https://api.nuget.org/v3/index.json --skip-duplicate
+          dotnet nuget push ./artifacts/*.snupkg --api-key "$NUGET_API_KEY" --source https://api.nuget.org/v3/index.json --skip-duplicate

.github/workflows/stable-release.yml

@@ -63,11 +63,16 @@ jobs:
           echo "version=$VERSION" >> "$GITHUB_OUTPUT"
           echo "Releasing version: $VERSION"
 
-      - name: Pack
+      - name: Pack CLI
         env:
           VERSION: ${{ steps.version.outputs.version }}
         run: dotnet pack src/ServiceBusToolset.CLI -c Release --no-build -o ./artifacts -p:Version="$VERSION"
 
+      - name: Pack Application
+        env:
+          VERSION: ${{ steps.version.outputs.version }}
+        run: dotnet pack src/ServiceBusToolset.Application -c Release --no-build -o ./artifacts -p:Version="$VERSION"
+
       - name: Generate SBOM
         env:
           VERSION: ${{ steps.version.outputs.version }}
@@ -91,6 +96,7 @@ jobs:
             exit 1
           fi
           dotnet nuget push ./artifacts/*.nupkg --api-key "$NUGET_API_KEY" --source https://api.nuget.org/v3/index.json --skip-duplicate
+          dotnet nuget push ./artifacts/*.snupkg --api-key "$NUGET_API_KEY" --source https://api.nuget.org/v3/index.json --skip-duplicate
 
       - name: Upload package to GitHub Release
         uses: softprops/action-gh-release@153bb8e04406b158c6c84fc1615b65b24149a1fe # v2.6.1

src/ServiceBusToolset.Application/DeadLetters/DiagnoseDlq/DiagnoseBatchCommand.cs

@@ -0,0 +1,14 @@
+using Ardalis.Result;
+using Mediator;
+using ServiceBusToolset.Application.DeadLetters.DiagnoseDlq.Models;
+
+namespace ServiceBusToolset.Application.DeadLetters.DiagnoseDlq;
+
+public sealed record DiagnoseBatchCommand(string AppInsightsResourceId,
+                                          IReadOnlyList<OperationInfo> Operations) : ICommand<Result<IReadOnlyList<DiagnosticResult>>>;
+
+public sealed record OperationInfo(string OperationId,
+                                   DateTimeOffset EnqueuedTime,
+                                   string? MessageId,
+                                   string? Subject,
+                                   string? DeadLetterReason);

src/ServiceBusToolset.Application/DeadLetters/DiagnoseDlq/DiagnoseBatchCommandHandler.cs

@@ -0,0 +1,49 @@
+using Ardalis.Result;
+using Mediator;
+using ServiceBusToolset.Application.DeadLetters.DiagnoseDlq.Common.AppInsights;
+using ServiceBusToolset.Application.DeadLetters.DiagnoseDlq.Models;
+
+namespace ServiceBusToolset.Application.DeadLetters.DiagnoseDlq;
+
+public sealed class DiagnoseBatchCommandHandler(IAppInsightsService appInsightsService)
+    : ICommandHandler<DiagnoseBatchCommand, Result<IReadOnlyList<DiagnosticResult>>>
+{
+    public async ValueTask<Result<IReadOnlyList<DiagnosticResult>>> Handle(
+        DiagnoseBatchCommand command,
+        CancellationToken cancellationToken)
+    {
+        if (command.Operations.Count == 0)
+        {
+            return Result.Success<IReadOnlyList<DiagnosticResult>>([]);
+        }
+
+        appInsightsService.Initialize(command.AppInsightsResourceId);
+
+        var operations = command.Operations
+                                .Select(op => (op.OperationId, op.EnqueuedTime))
+                                .ToList();
+
+        var diagnosticResults = await appInsightsService.DiagnoseBatchAsync(operations,
+                                                                            null,
+                                                                            cancellationToken);
+
+        var results = new List<DiagnosticResult>();
+        // Deduplicate by OperationId — take the first occurrence if duplicates exist
+        var operationsById = command.Operations
+                                    .DistinctBy(op => op.OperationId)
+                                    .ToDictionary(op => op.OperationId);
+
+        foreach (var (operationId, result) in diagnosticResults)
+        {
+            if (operationsById.TryGetValue(operationId, out var operation))
+            {
+                result.MessageId = operation.MessageId;
+                result.Subject = operation.Subject;
+                result.DeadLetterReason = operation.DeadLetterReason;
+                results.Add(result);
+            }
+        }
+
+        return Result.Success<IReadOnlyList<DiagnosticResult>>(results);
+    }
+}

src/ServiceBusToolset.Application/DeadLetters/PeekDlq/PeekDlqBatchCommand.cs

@@ -0,0 +1,11 @@
+using Ardalis.Result;
+using Mediator;
+using ServiceBusToolset.Application.Common.ServiceBus.Models;
+
+namespace ServiceBusToolset.Application.DeadLetters.PeekDlq;
+
+public sealed record PeekDlqBatchCommand(string FullyQualifiedNamespace,
+                                         EntityTarget Target,
+                                         int BatchSize = 500,
+                                         long? FromSequenceNumber = null,
+                                         long? KnownDeadLetterCount = null) : ICommand<Result<PeekDlqBatchResult>>;

src/ServiceBusToolset.Application/DeadLetters/PeekDlq/PeekDlqBatchCommandHandler.cs

@@ -0,0 +1,146 @@
+using Ardalis.Result;
+using Azure.Messaging.ServiceBus;
+using Mediator;
+using ServiceBusToolset.Application.Common.ServiceBus.Abstractions;
+using ServiceBusToolset.Application.Common.ServiceBus.Helpers;
+using ServiceBusToolset.Application.Common.ServiceBus.Models;
+using ServiceBusToolset.Application.DeadLetters.DiagnoseDlq.Common;
+
+namespace ServiceBusToolset.Application.DeadLetters.PeekDlq;
+
+public sealed class PeekDlqBatchCommandHandler(IServiceBusClientFactory clientFactory)
+    : ICommandHandler<PeekDlqBatchCommand, Result<PeekDlqBatchResult>>
+{
+    private const int PeekSubBatchSize = 100;
+    private const int EmptyBatchThreshold = 3;
+
+    public async ValueTask<Result<PeekDlqBatchResult>> Handle(
+        PeekDlqBatchCommand command,
+        CancellationToken cancellationToken)
+    {
+        await using var client = clientFactory.CreateClient(command.FullyQualifiedNamespace);
+
+        // Get total DLQ count — use known count if provided, otherwise query admin API
+        var totalDeadLetterCount = command.KnownDeadLetterCount
+                                   ?? await GetDeadLetterCountAsync(clientFactory, command.FullyQualifiedNamespace, command.Target);
+
+        // Peek messages in sub-batches until we reach BatchSize or run out
+        await using var receiver = ReceiverFactory.CreateDlqReceiver(client, command.Target);
+
+        List<ServiceBusReceivedMessage> allMessages = [];
+        var emptyBatches = 0;
+        var isFirstPeek = true;
+        var highestSequenceNumber = command.FromSequenceNumber ?? -1;
+
+        while (allMessages.Count < command.BatchSize &&
+               emptyBatches < EmptyBatchThreshold &&
+               !cancellationToken.IsCancellationRequested)
+        {
+            var remaining = command.BatchSize - allMessages.Count;
+            var subBatchSize = Math.Min(PeekSubBatchSize, remaining);
+
+            IReadOnlyList<ServiceBusReceivedMessage> batch;
+            if (isFirstPeek && command.FromSequenceNumber.HasValue)
+            {
+                batch = await receiver.PeekMessagesAsync(subBatchSize, command.FromSequenceNumber.Value + 1, cancellationToken);
+                isFirstPeek = false;
+            }
+            else
+            {
+                batch = await receiver.PeekMessagesAsync(subBatchSize, cancellationToken:cancellationToken);
+                isFirstPeek = false;
+            }
+
+            if (batch.Count == 0)
+            {
+                emptyBatches++;
+                continue;
+            }
+
+            // Detect wrap-around: if the batch contains messages with sequence numbers
+            // we've already passed, the receiver has looped back to the beginning
+            if (batch[0].SequenceNumber <= highestSequenceNumber)
+            {
+                break;
+            }
+
+            emptyBatches = 0;
+            allMessages.AddRange(batch);
+            highestSequenceNumber = batch[^1].SequenceNumber;
+        }
+
+        if (allMessages.Count == 0)
+        {
+            return Result.Success(new PeekDlqBatchResult([],
+                                                         0,
+                                                         0,
+                                                         command.FromSequenceNumber,
+                                                         false,
+                                                         totalDeadLetterCount));
+        }
+
+        // Extract operation IDs
+        List<PeekedMessage> messages = [];
+        var skipped = 0;
+        HashSet<string> seenOperationIds = [];
+
+        foreach (var message in allMessages)
+        {
+            var operationId = MessageDiagnostics.ExtractOperationId(message);
+            if (string.IsNullOrEmpty(operationId))
+            {
+                skipped++;
+                continue;
+            }
+
+            if (seenOperationIds.Add(operationId))
+            {
+                messages.Add(new PeekedMessage(message.MessageId,
+                                               message.Subject,
+                                               operationId,
+                                               message.EnqueuedTime,
+                                               message.DeadLetterReason));
+            }
+        }
+
+        var lastSequenceNumber = allMessages[^1].SequenceNumber;
+        // We have more messages if we filled the batch (didn't run out early)
+        var hasMore = allMessages.Count >= command.BatchSize && emptyBatches < EmptyBatchThreshold;
+
+        return Result.Success(new PeekDlqBatchResult(messages,
+                                                     allMessages.Count,
+                                                     skipped,
+                                                     lastSequenceNumber,
+                                                     hasMore,
+                                                     totalDeadLetterCount));
+    }
+
+    private static async Task<long> GetDeadLetterCountAsync(
+        IServiceBusClientFactory clientFactory,
+        string fullyQualifiedNamespace,
+        EntityTarget target)
+    {
+        try
+        {
+            var adminClient = clientFactory.CreateAdministrationClient(fullyQualifiedNamespace);
+
+            if (target.IsQueueMode)
+            {
+                var props = await adminClient.GetQueueRuntimePropertiesAsync(target.Queue!);
+                return props.Value.DeadLetterMessageCount;
+            }
+
+            var subProps = await adminClient.GetSubscriptionRuntimePropertiesAsync(target.Topic!, target.Subscription!);
+            return subProps.Value.DeadLetterMessageCount;
+        }
+        catch (Azure.RequestFailedException)
+        {
+            // Admin API may not be available in all environments (e.g., emulator, unit tests)
+            return 0;
+        }
+        catch (OperationCanceledException)
+        {
+            return 0;
+        }
+    }
+}

src/ServiceBusToolset.Application/DeadLetters/PeekDlq/PeekDlqBatchResult.cs

@@ -0,0 +1,14 @@
+namespace ServiceBusToolset.Application.DeadLetters.PeekDlq;
+
+public sealed record PeekDlqBatchResult(IReadOnlyList<PeekedMessage> Messages,
+                                        int PeekedInBatch,
+                                        int SkippedNoOperationId,
+                                        long? LastSequenceNumber,
+                                        bool HasMoreMessages,
+                                        long TotalDeadLetterCount);
+
+public sealed record PeekedMessage(string? MessageId,
+                                   string? Subject,
+                                   string OperationId,
+                                   DateTimeOffset EnqueuedTime,
+                                   string? DeadLetterReason);

src/ServiceBusToolset.Application/ServiceBusToolset.Application.csproj

@@ -1,5 +1,31 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
+    <PropertyGroup>
+        <!-- NuGet Package Metadata -->
+        <PackageId>ServiceBusToolset.Application</PackageId>
+        <Authors>Gagik Kyurkchyan</Authors>
+        <Description>Application layer for Azure Service Bus diagnostics: dead letter queue analysis, Application Insights correlation, and message inspection.</Description>
+        <PackageTags>azure;service-bus;dead-letter-queue;diagnostics;application-insights</PackageTags>
+        <PackageLicenseExpression>MIT</PackageLicenseExpression>
+        <PackageProjectUrl>https://github.com/kyurkchyan/ServiceBusToolset</PackageProjectUrl>
+        <RepositoryUrl>https://github.com/kyurkchyan/ServiceBusToolset.git</RepositoryUrl>
+        <RepositoryType>git</RepositoryType>
+        <PackageReadmeFile>README.md</PackageReadmeFile>
+
+        <!-- Version - Overridden by CI/CD -->
+        <Version>1.0.0</Version>
+
+        <!-- Source Link -->
+        <PublishRepositoryUrl>true</PublishRepositoryUrl>
+        <EmbedUntrackedSources>true</EmbedUntrackedSources>
+        <IncludeSymbols>true</IncludeSymbols>
+        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <None Include="..\..\README.md" Pack="true" PackagePath="\"/>
+    </ItemGroup>
+
     <ItemGroup>
         <PackageReference Include="Mediator.Abstractions"/>
         <PackageReference Include="Mediator.SourceGenerator">
@@ -13,6 +39,7 @@
         <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions"/>
         <PackageReference Include="System.Reactive"/>
         <PackageReference Include="DynamicData"/>
+        <PackageReference Include="Microsoft.SourceLink.GitHub" PrivateAssets="All"/>
     </ItemGroup>
 
 </Project>

tests/ServiceBusToolset.Application.Tests/Common/Mocks/MockServiceBusClientFactory.cs

@@ -86,19 +86,14 @@ private void ConfigureReceiver()
         _peekCallCount = 0;
         _receiveCallCount = 0;
 
-        // Setup PeekMessagesAsync - returns all messages on first call, empty on subsequent
+        // Setup PeekMessagesAsync - returns messages in pages, respecting maxMessages per call
         Receiver.PeekMessagesAsync(Arg.Any<int>(), Arg.Any<long?>(), Arg.Any<CancellationToken>())
                 .Returns(callInfo =>
                 {
                     var maxMessages = callInfo.ArgAt<int>(0);
-                    _peekCallCount++;
-                    if (_peekCallCount == 1)
-                    {
-                        var batch = _messagesToReturn.Take(maxMessages).ToList();
-                        return Task.FromResult<IReadOnlyList<ServiceBusReceivedMessage>>(batch);
-                    }
-
-                    return Task.FromResult<IReadOnlyList<ServiceBusReceivedMessage>>([]);
+                    var batch = _messagesToReturn.Skip(_peekCallCount).Take(maxMessages).ToList();
+                    _peekCallCount += batch.Count;
+                    return Task.FromResult<IReadOnlyList<ServiceBusReceivedMessage>>(batch);
                 });
 
         // Setup ReceiveMessagesAsync - returns all messages on first call, empty on subsequent