Implemented support for dumping the summary of the dead letter messages

Author: kyurkchyan

ServiceBusToolset/Commands/PurgeDlqCommand.cs

@@ -11,6 +11,8 @@ public class PurgeDlqCommand(IServiceBusClientFactory clientFactory, IConsoleOut
     private static readonly TimeSpan MaxWaitTime = TimeSpan.FromSeconds(5);
     private const int EmptyBatchThreshold = 3;
 
+    private record DlqCategory(string Label, string DeadLetterReason, int Count);
+
     public async Task<int> ExecuteAsync(PurgeDlqOptions options, CancellationToken cancellationToken = default)
     {
         var validationError = options.Validate();
@@ -34,6 +36,14 @@ public async Task<int> ExecuteAsync(PurgeDlqOptions options, CancellationToken c
                                                 cancellationToken);
             }
 
+            if (options.Interactive)
+            {
+                return await ExecuteInteractivePurgeAsync(client,
+                                                          options,
+                                                          entityDescription,
+                                                          cancellationToken);
+            }
+
             return await ExecutePurgeAsync(client,
                                            options,
                                            entityDescription,
@@ -70,7 +80,6 @@ private async Task<int> ExecuteDryRunAsync(
         {
             return await ExecuteFilteredDryRunAsync(client,
                                                     options,
-                                                    entityDescription,
                                                     cancellationToken);
         }
 
@@ -103,7 +112,6 @@ private async Task<int> ExecuteFastDryRunAsync(
     private async Task<int> ExecuteFilteredDryRunAsync(
         ServiceBusClient client,
         PurgeDlqOptions options,
-        string entityDescription,
         CancellationToken cancellationToken)
     {
         Output.Verbose("Using slow count due to --before filter", options.Verbose);
@@ -257,4 +265,248 @@ private async Task<int> ExecuteFilteredPurgeAsync(
         Output.Success($"Purged {totalDeleted} messages from DLQ for {entityDescription} (skipped {totalSkipped} newer messages)");
         return 0;
     }
+
+    private async Task<int> ExecuteInteractivePurgeAsync(
+        ServiceBusClient client,
+        PurgeDlqOptions options,
+        string entityDescription,
+        CancellationToken cancellationToken)
+    {
+        Output.Info($"Analyzing DLQ for {entityDescription}...");
+
+        // Step 1: Peek all messages and build category dictionary
+        var categories = await BuildCategoryListAsync(client, options, cancellationToken);
+
+        if (categories.Count == 0)
+        {
+            Output.Info("No messages found in DLQ.");
+            return 0;
+        }
+
+        // Step 2: Display category table
+        DisplayCategoryTable(categories);
+
+        // Step 3: Get user selection
+        Output.Info("");
+        Console.Write("Select categories to purge (comma-separated numbers, 'all', or 'q' to quit): ");
+        var input = Output.ReadLine();
+
+        var selectedIndices = ParseSelection(input, categories.Count);
+        if (selectedIndices == null)
+        {
+            Output.Info("Operation cancelled.");
+            return 0;
+        }
+
+        if (selectedIndices.Count == 0)
+        {
+            Output.Warning("No valid categories selected.");
+            return 0;
+        }
+
+        // Step 4: Build set of selected category keys
+        var selectedCategories = new HashSet<(string Label, string Reason)>();
+        var totalToPurge = 0;
+        foreach(var cat in selectedIndices.Select(idx => categories[idx]))
+        {
+            selectedCategories.Add((cat.Label, cat.DeadLetterReason));
+            totalToPurge += cat.Count;
+        }
+
+        Output.Info($"Purging {totalToPurge} messages from {selectedIndices.Count} categories...");
+
+        // Step 5: Receive messages and complete only those matching selected categories
+        var totalDeleted = await PurgeByCategoriesAsync(client, options, selectedCategories, cancellationToken);
+
+        Console.WriteLine();
+        Output.Success($"Purged {totalDeleted} messages from DLQ for {entityDescription}.");
+        return 0;
+    }
+
+    private async Task<List<DlqCategory>> BuildCategoryListAsync(
+        ServiceBusClient client,
+        PurgeDlqOptions options,
+        CancellationToken cancellationToken)
+    {
+        await using var receiver = CreateDlqReceiver(client,
+                                                     options.Queue,
+                                                     options.Topic,
+                                                     options.Subscription,
+                                                     ServiceBusReceiveMode.PeekLock);
+
+        var categoryCounts = new Dictionary<(string Label, string Reason), int>();
+        var totalPeeked = 0;
+        long? fromSequenceNumber = null;
+
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            IReadOnlyList<ServiceBusReceivedMessage> messages;
+
+            if (fromSequenceNumber.HasValue)
+            {
+                messages = await receiver.PeekMessagesAsync(MaxBatchSize, fromSequenceNumber.Value, cancellationToken);
+            }
+            else
+            {
+                messages = await receiver.PeekMessagesAsync(MaxBatchSize, cancellationToken: cancellationToken);
+            }
+
+            if (messages.Count == 0)
+            {
+                break;
+            }
+
+            foreach (var msg in messages)
+            {
+                var label = msg.Subject ?? "(none)";
+                var reason = msg.DeadLetterReason ?? "(none)";
+                var key = (label, reason);
+
+                var count = categoryCounts.GetValueOrDefault(key, 0);
+                categoryCounts[key] = count + 1;
+            }
+
+            totalPeeked += messages.Count;
+            fromSequenceNumber = messages[^1].SequenceNumber + 1;
+
+            Output.Progress($"Peeked {totalPeeked} messages...");
+        }
+
+        Console.WriteLine();
+
+        return categoryCounts
+            .OrderByDescending(kvp => kvp.Value)
+            .Select(kvp => new DlqCategory(kvp.Key.Label, kvp.Key.Reason, kvp.Value))
+            .ToList();
+    }
+
+    private void DisplayCategoryTable(IReadOnlyCollection<DlqCategory> categories)
+    {
+        Output.Info("");
+        Output.Info("Dead Letter Summary:");
+
+        var headers = new[] { "#", "Label", "DeadLetterReason", "Count" };
+        var rows = categories.Select((cat, index) => new[]
+        {
+            (index + 1).ToString(),
+            cat.Label.ReplaceLineEndings(" "),
+            cat.DeadLetterReason.ReplaceLineEndings(" "),
+            cat.Count.ToString()
+        });
+
+        Output.Table(headers, rows);
+
+        var total = categories.Sum(c => c.Count);
+        Output.Info($"Total: {total} messages");
+    }
+
+    private static List<int>? ParseSelection(string? input, int maxIndex)
+    {
+        if (string.IsNullOrWhiteSpace(input))
+        {
+            return null;
+        }
+
+        var trimmed = input.Trim().ToLowerInvariant();
+
+        if (trimmed == "q" || trimmed == "quit")
+        {
+            return null;
+        }
+
+        if (trimmed == "all" || trimmed == "a")
+        {
+            return Enumerable.Range(0, maxIndex).ToList();
+        }
+
+        var indices = new List<int>();
+        var parts = input.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
+
+        foreach (var part in parts)
+        {
+            // Handle ranges like "1-5"
+            if (part.Contains('-'))
+            {
+                var rangeParts = part.Split('-', 2);
+                if (rangeParts.Length == 2 &&
+                    int.TryParse(rangeParts[0], out var start) &&
+                    int.TryParse(rangeParts[1], out var end))
+                {
+                    for (var i = start; i <= end; i++)
+                    {
+                        var idx = i - 1; // Convert to 0-based
+                        if (idx >= 0 && idx < maxIndex && !indices.Contains(idx))
+                        {
+                            indices.Add(idx);
+                        }
+                    }
+                }
+            }
+            else if (int.TryParse(part, out var num))
+            {
+                var idx = num - 1; // Convert to 0-based
+                if (idx >= 0 && idx < maxIndex && !indices.Contains(idx))
+                {
+                    indices.Add(idx);
+                }
+            }
+        }
+
+        return indices;
+    }
+
+    private async Task<int> PurgeByCategoriesAsync(
+        ServiceBusClient client,
+        PurgeDlqOptions options,
+        HashSet<(string Label, string Reason)> selectedCategories,
+        CancellationToken cancellationToken)
+    {
+        await using var receiver = CreateDlqReceiver(client,
+                                                     options.Queue,
+                                                     options.Topic,
+                                                     options.Subscription,
+                                                     ServiceBusReceiveMode.PeekLock);
+
+        var totalDeleted = 0;
+        var totalSkipped = 0;
+        var emptyBatches = 0;
+
+        while (!cancellationToken.IsCancellationRequested && emptyBatches < EmptyBatchThreshold)
+        {
+            var messages = await receiver.ReceiveMessagesAsync(MaxBatchSize,
+                                                               MaxWaitTime,
+                                                               cancellationToken);
+
+            if (messages.Count == 0)
+            {
+                emptyBatches++;
+                Output.Verbose($"Empty batch {emptyBatches}/{EmptyBatchThreshold}", options.Verbose);
+                continue;
+            }
+
+            emptyBatches = 0;
+
+            foreach (var message in messages)
+            {
+                var label = message.Subject ?? "(none)";
+                var reason = message.DeadLetterReason ?? "(none)";
+                var key = (label, reason);
+
+                if (selectedCategories.Contains(key))
+                {
+                    await receiver.CompleteMessageAsync(message, cancellationToken);
+                    totalDeleted++;
+                }
+                else
+                {
+                    await receiver.AbandonMessageAsync(message, cancellationToken: cancellationToken);
+                    totalSkipped++;
+                }
+            }
+
+            Output.Progress($"Purged {totalDeleted} messages (skipped {totalSkipped})...");
+        }
+
+        return totalDeleted;
+    }
 }

ServiceBusToolset/Options/PurgeDlqOptions.cs

@@ -41,6 +41,12 @@ public class PurgeDlqOptions
             HelpText = "Enable verbose output")]
     public bool Verbose { get; set; }
 
+    [Option('i',
+            "interactive",
+            Default = false,
+            HelpText = "Interactive mode: show categories and select which to purge")]
+    public bool Interactive { get; set; }
+
     public bool IsQueueMode => !string.IsNullOrEmpty(Queue);
     public bool IsSubscriptionMode => !string.IsNullOrEmpty(Topic) && !string.IsNullOrEmpty(Subscription);
 

ServiceBusToolset/ServiceBusToolset.csproj

@@ -8,9 +8,10 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="Azure.Messaging.ServiceBus" Version="7.18.2"/>
-        <PackageReference Include="Azure.Identity" Version="1.13.2"/>
-        <PackageReference Include="CommandLineParser" Version="2.9.1"/>
+        <PackageReference Include="Azure.Messaging.ServiceBus" Version="7.18.2" />
+        <PackageReference Include="Azure.Identity" Version="1.13.2" />
+        <PackageReference Include="CommandLineParser" Version="2.9.1" />
+        <PackageReference Include="Spectre.Console" Version="0.54.0" />
     </ItemGroup>
 
 </Project>

ServiceBusToolset/Services/ConsoleOutput.cs

@@ -1,3 +1,5 @@
+using Spectre.Console;
+
 namespace ServiceBusToolset.Services;
 
 public class ConsoleOutput : IConsoleOutput
@@ -44,4 +46,25 @@ public void Progress(string message)
         Console.Write($"\r{message}");
         Console.ResetColor();
     }
+
+    public void Table(IEnumerable<string> headers, IEnumerable<string[]> rows)
+    {
+        var table = new Table();
+        table.Border(TableBorder.Rounded);
+        table.Expand();
+
+        foreach (var header in headers)
+        {
+            table.AddColumn(new TableColumn(header));
+        }
+
+        foreach (var row in rows)
+        {
+            table.AddRow(row.Select(Markup.Escape).ToArray());
+        }
+
+        AnsiConsole.Write(table);
+    }
+
+    public string? ReadLine() => Console.ReadLine();
 }

ServiceBusToolset/Services/IConsoleOutput.cs

@@ -8,4 +8,6 @@ public interface IConsoleOutput
     void Error(string message);
     void Verbose(string message, bool isVerbose);
     void Progress(string message);
+    void Table(IEnumerable<string> headers, IEnumerable<string[]> rows);
+    string? ReadLine();
 }