Merge pull request #1742 from c-teleport/feature/mongo-index-cleanup

Mongo index cleanup

Author: demorgi

src/DotNetCore.CAP.MongoDB/IStorageInitializer.MongoDB.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -67,6 +68,10 @@ await database.CreateCollectionAsync(options.PublishedCollection, cancellationTo
             await database.CreateCollectionAsync(options.LockCollection, cancellationToken: cancellationToken)
                 .ConfigureAwait(false);
 
+        await Task.WhenAll(
+            DropReceivedMessageDeprecatedIndexesAsync(), 
+            DropPublishedMessageDeprecatedIndexesAsync()).ConfigureAwait(false);
+
         await Task.WhenAll(
             CreateReceivedMessageIndexesAsync(),
             CreatePublishedMessageIndexesAsync()).ConfigureAwait(false);
@@ -98,7 +103,6 @@ async Task CreateReceivedMessageIndexesAsync()
                 new(builder.Ascending(x => x.Name)),
                 new(builder.Ascending(x => x.Added)),
                 new(builder.Ascending(x => x.ExpiresAt)),
-                new(builder.Ascending(x => x.StatusName)),
                 new(builder.Ascending(x => x.Retries)),
                 new(builder.Ascending(x => x.Version)),
                 new(builder.Ascending(x => x.StatusName).Ascending(x => x.ExpiresAt))
@@ -117,13 +121,55 @@ async Task CreatePublishedMessageIndexesAsync()
                 new(builder.Ascending(x => x.Name)),
                 new(builder.Ascending(x => x.Added)),
                 new(builder.Ascending(x => x.ExpiresAt)),
-                new(builder.Ascending(x => x.StatusName)),
                 new(builder.Ascending(x => x.Retries)),
                 new(builder.Ascending(x => x.Version)),
                 new(builder.Ascending(x => x.StatusName).Ascending(x => x.ExpiresAt))
             };
 
             await col.Indexes.CreateManyAsync(indexes, cancellationToken);
         }
+
+        async Task DropReceivedMessageDeprecatedIndexesAsync()
+        {
+            var obsoleteIndexes = new HashSet<string> { "Name", "Added", "ExpiresAt", "StatusName", "Retries", "Version" };
+           
+            var col = database.GetCollection<ReceivedMessage>(options.ReceivedCollection);
+           
+            await DropIndexesAsync(col, obsoleteIndexes);
+        }
+
+        async Task DropPublishedMessageDeprecatedIndexesAsync()
+        {
+            var obsoleteIndexes = new HashSet<string> { "Name", "Added", "ExpiresAt", "StatusName", "Retries", "Version" };
+            
+            var col = database.GetCollection<PublishedMessage>(options.PublishedCollection);
+           
+            await DropIndexesAsync(col, obsoleteIndexes);
+            
+        }
+
+        async Task DropIndexesAsync<T>(IMongoCollection<T> col, ISet<string> obsoleteIndexes)
+        {
+            using var cursor = await col.Indexes.ListAsync(cancellationToken);
+            var indexList = await cursor.ToListAsync(cancellationToken);
+
+            foreach (var index in indexList)
+            {
+                var indexName = index["name"].AsString;
+                if (!obsoleteIndexes.Contains(indexName)) continue;
+
+                try
+                {
+                    await col.Indexes.DropOneAsync(indexName, cancellationToken);
+                }
+                catch (MongoCommandException ex) when (ex.CodeName == "IndexNotFound")
+                {
+                    _logger.LogWarning(
+                        "Index '{IndexName}' on collection '{CollectionName}' was not found when attempting to drop it. " +
+                        "This may indicate concurrent initialization or an unexpected state. Verify deployment strategy if this happens repeatedly.",
+                        indexName, col.CollectionNamespace.CollectionName);
+                }
+            }
+        }
     }
 }