Outbox

phongnguyend · Jan 24, 2025 · 4d402bf · 4d402bf
1 parent 19fb792
commit 4d402bf
Show file tree

Hide file tree

Showing 30 changed files with 114 additions and 58 deletions.
diff --git a/...rvices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IOutBoxEventPublisher.cs b/...rvices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IOutBoxEventPublisher.cs
@@ -21,4 +21,6 @@ public class PublishingOutBoxEvent
     public string EventSource { get; set; }
 
     public string Payload { get; set; }
+
+    public string ActivityId { get; set; }
 }
diff --git a/src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/MetaData.cs b/src/Microservices/Common/ClassifiedAds.Domain/Infrastructure/MessageBrokers/MetaData.cs
@@ -8,7 +8,7 @@ public class MetaData
 
     public string MessageVersion { get; set; }
 
-    public string CorrelationId { get; set; }
+    public string ActivityId { get; set; }
 
     public DateTimeOffset? CreationDateTime { get; set; }
 

diff --git a/...s/Services.Product/ClassifiedAds.Services.Product.Api/Commands/AddAuditLogEntryCommand.cs b/...s/Services.Product/ClassifiedAds.Services.Product.Api/Commands/AddAuditLogEntryCommand.cs
@@ -5,6 +5,7 @@
 using ClassifiedAds.Services.Product.Entities;
 using MediatR;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 namespace ClassifiedAds.Services.Product.Commands;
@@ -50,8 +51,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = auditLog.CreatedDateTime,
             ObjectId = auditLog.Id.ToString(),
-            Message = auditLog.AsJsonString(),
-            Published = false,
+            Payload = auditLog.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/...ices/Services.Product/ClassifiedAds.Services.Product.Api/Commands/PublishEventsCommand.cs b/...ices/Services.Product/ClassifiedAds.Services.Product.Api/Commands/PublishEventsCommand.cs
@@ -49,7 +49,8 @@ public async Task Handle(PublishEventsCommand command, CancellationToken cancell
                 Id = eventLog.Id.ToString(),
                 EventType = eventLog.EventType,
                 EventSource = typeof(PublishEventsCommand).Assembly.GetName().Name,
-                Payload = eventLog.Message,
+                Payload = eventLog.Payload,
+                ActivityId = eventLog.ActivityId
             };
 
             await _messageBus.SendAsync(outbox, cancellationToken);

diff --git a/...Microservices/Services.Product/ClassifiedAds.Services.Product.Api/Entities/OutboxEvent.cs b/...Microservices/Services.Product/ClassifiedAds.Services.Product.Api/Entities/OutboxEvent.cs
@@ -19,7 +19,9 @@ public abstract class OutboxEventBase : Entity<Guid>
 
     public string ObjectId { get; set; }
 
-    public string Message { get; set; }
+    public string Payload { get; set; }
 
     public bool Published { get; set; }
+
+    public string ActivityId { get; set; }
 }
diff --git a/...es.Product/ClassifiedAds.Services.Product.Api/EventHandlers/ProductCreatedEventHandler.cs b/...es.Product/ClassifiedAds.Services.Product.Api/EventHandlers/ProductCreatedEventHandler.cs
@@ -7,6 +7,7 @@
 using ClassifiedAds.Services.Product.Entities;
 using MediatR;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -47,8 +48,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/...es.Product/ClassifiedAds.Services.Product.Api/EventHandlers/ProductDeletedEventHandler.cs b/...es.Product/ClassifiedAds.Services.Product.Api/EventHandlers/ProductDeletedEventHandler.cs
@@ -7,6 +7,7 @@
 using ClassifiedAds.Services.Product.Entities;
 using MediatR;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -47,8 +48,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/...es.Product/ClassifiedAds.Services.Product.Api/EventHandlers/ProductUpdatedEventHandler.cs b/...es.Product/ClassifiedAds.Services.Product.Api/EventHandlers/ProductUpdatedEventHandler.cs
@@ -7,6 +7,7 @@
 using ClassifiedAds.Services.Product.Entities;
 using MediatR;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -47,8 +48,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/...s/Services.Storage/ClassifiedAds.Services.Storage.Api/Commands/AddAuditLogEntryCommand.cs b/...s/Services.Storage/ClassifiedAds.Services.Storage.Api/Commands/AddAuditLogEntryCommand.cs
@@ -5,6 +5,7 @@
 using ClassifiedAds.Services.Storage.Constants;
 using ClassifiedAds.Services.Storage.Entities;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -50,8 +51,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = auditLog.CreatedDateTime,
             ObjectId = auditLog.Id.ToString(),
-            Message = auditLog.AsJsonString(),
-            Published = false,
+            Payload = auditLog.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/...ices/Services.Storage/ClassifiedAds.Services.Storage.Api/Commands/PublishEventsCommand.cs b/...ices/Services.Storage/ClassifiedAds.Services.Storage.Api/Commands/PublishEventsCommand.cs
@@ -49,7 +49,8 @@ public async Task HandleAsync(PublishEventsCommand command, CancellationToken ca
                 Id = eventLog.Id.ToString(),
                 EventType = eventLog.EventType,
                 EventSource = typeof(PublishEventsCommand).Assembly.GetName().Name,
-                Payload = eventLog.Message,
+                Payload = eventLog.Payload,
+                ActivityId = eventLog.ActivityId
             };
 
             await _messageBus.SendAsync(outbox, cancellationToken);

diff --git a/...Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/Entities/OutboxEvent.cs b/...Microservices/Services.Storage/ClassifiedAds.Services.Storage.Api/Entities/OutboxEvent.cs
@@ -19,7 +19,9 @@ public abstract class OutboxEventBase : Entity<Guid>
 
     public string ObjectId { get; set; }
 
-    public string Message { get; set; }
+    public string Payload { get; set; }
 
     public bool Published { get; set; }
+
+    public string ActivityId { get; set; }
 }
diff --git a/....Storage/ClassifiedAds.Services.Storage.Api/EventHandlers/FileEntryCreatedEventHandler.cs b/....Storage/ClassifiedAds.Services.Storage.Api/EventHandlers/FileEntryCreatedEventHandler.cs
@@ -7,6 +7,7 @@
 using ClassifiedAds.Services.Storage.Constants;
 using ClassifiedAds.Services.Storage.Entities;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -47,8 +48,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/....Storage/ClassifiedAds.Services.Storage.Api/EventHandlers/FileEntryDeletedEventHandler.cs b/....Storage/ClassifiedAds.Services.Storage.Api/EventHandlers/FileEntryDeletedEventHandler.cs
@@ -7,6 +7,7 @@
 using ClassifiedAds.Services.Storage.Constants;
 using ClassifiedAds.Services.Storage.Entities;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -47,8 +48,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/....Storage/ClassifiedAds.Services.Storage.Api/EventHandlers/FileEntryUpdatedEventHandler.cs b/....Storage/ClassifiedAds.Services.Storage.Api/EventHandlers/FileEntryUpdatedEventHandler.cs
@@ -7,6 +7,7 @@
 using ClassifiedAds.Services.Storage.Constants;
 using ClassifiedAds.Services.Storage.Entities;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -47,8 +48,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/...ervices.Storage/ClassifiedAds.Services.Storage.Api/MessageBusConsumers/WebhookConsumer.cs b/...ervices.Storage/ClassifiedAds.Services.Storage.Api/MessageBusConsumers/WebhookConsumer.cs
@@ -1,4 +1,5 @@
-using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
+using ClassifiedAds.CrossCuttingConcerns.Logging;
+using ClassifiedAds.Domain.Infrastructure.MessageBrokers;
 using ClassifiedAds.Services.Storage.DTOs;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
@@ -27,12 +28,14 @@ public WebhookConsumer(ILogger<WebhookConsumer> logger,
 
     public async Task HandleAsync(FileUploadedEvent data, MetaData metaData, CancellationToken cancellationToken = default)
     {
+        using var activity = ActivityExtensions.StartNew("WebhookConsumer", metaData?.ActivityId);
         var url = _configuration["Webhooks:FileUploadedEvent:PayloadUrl"];
         await _httpClient.PostAsJsonAsync(url, data.FileEntry, cancellationToken: cancellationToken);
     }
 
     public async Task HandleAsync(FileDeletedEvent data, MetaData metaData, CancellationToken cancellationToken = default)
     {
+        using var activity = ActivityExtensions.StartNew("WebhookConsumer", metaData?.ActivityId);
         var url = _configuration["Webhooks:FileDeletedEvent:PayloadUrl"];
         await _httpClient.PostAsJsonAsync(url, data.FileEntry, cancellationToken: cancellationToken);
     }

diff --git a/...ClassifiedAds.Services.Storage.Api/OutBoxEventPublishers/FileEntryOutBoxEventPublisher.cs b/...ClassifiedAds.Services.Storage.Api/OutBoxEventPublishers/FileEntryOutBoxEventPublisher.cs
@@ -15,7 +15,7 @@ public class FileEntryOutBoxEventPublisher : IOutBoxEventPublisher
 
     public static string[] CanHandleEventTypes()
     {
-        return new string[] { EventTypeConstants.FileEntryCreated, EventTypeConstants.FileEntryDeleted };
+        return [EventTypeConstants.FileEntryCreated, EventTypeConstants.FileEntryDeleted];
     }
 
     public static string CanHandleEventSource()
@@ -32,11 +32,21 @@ public async Task HandleAsync(PublishingOutBoxEvent outbox, CancellationToken ca
     {
         if (outbox.EventType == EventTypeConstants.FileEntryCreated)
         {
-            await _messageBus.SendAsync(new FileUploadedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(outbox.Payload) }, cancellationToken: cancellationToken);
+            await _messageBus.SendAsync(new FileUploadedEvent
+            {
+                FileEntry = JsonSerializer.Deserialize<FileEntry>(outbox.Payload)
+            },
+            metaData: new MetaData { ActivityId = outbox.ActivityId, MessageId = outbox.Id },
+            cancellationToken: cancellationToken);
         }
         else if (outbox.EventType == EventTypeConstants.FileEntryDeleted)
         {
-            await _messageBus.SendAsync(new FileDeletedEvent { FileEntry = JsonSerializer.Deserialize<FileEntry>(outbox.Payload) }, cancellationToken: cancellationToken);
+            await _messageBus.SendAsync(new FileDeletedEvent
+            {
+                FileEntry = JsonSerializer.Deserialize<FileEntry>(outbox.Payload)
+            },
+            metaData: new MetaData { ActivityId = outbox.ActivityId, MessageId = outbox.Id },
+            cancellationToken: cancellationToken);
         }
     }
 }
diff --git a/...dularMonolith/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IOutBoxEventPublisher.cs b/...dularMonolith/ClassifiedAds.Domain/Infrastructure/MessageBrokers/IOutBoxEventPublisher.cs
@@ -21,4 +21,6 @@ public class PublishingOutBoxEvent
     public string EventSource { get; set; }
 
     public string Payload { get; set; }
+
+    public string ActivityId { get; set; }
 }
diff --git a/src/ModularMonolith/ClassifiedAds.Domain/Infrastructure/MessageBrokers/MetaData.cs b/src/ModularMonolith/ClassifiedAds.Domain/Infrastructure/MessageBrokers/MetaData.cs
@@ -8,7 +8,7 @@ public class MetaData
 
     public string MessageVersion { get; set; }
 
-    public string CorrelationId { get; set; }
+    public string ActivityId { get; set; }
 
     public DateTimeOffset? CreationDateTime { get; set; }
 

diff --git a/src/ModularMonolith/ClassifiedAds.Modules.Product/Commands/PublishEventsCommand.cs b/src/ModularMonolith/ClassifiedAds.Modules.Product/Commands/PublishEventsCommand.cs
@@ -49,7 +49,8 @@ public async Task HandleAsync(PublishEventsCommand command, CancellationToken ca
                 Id = eventLog.Id.ToString(),
                 EventType = eventLog.EventType,
                 EventSource = typeof(PublishEventsCommand).Assembly.GetName().Name,
-                Payload = eventLog.Message,
+                Payload = eventLog.Payload,
+                ActivityId = eventLog.ActivityId
             };
 
             await _messageBus.SendAsync(outbox, cancellationToken);

diff --git a/src/ModularMonolith/ClassifiedAds.Modules.Product/Entities/OutboxEvent.cs b/src/ModularMonolith/ClassifiedAds.Modules.Product/Entities/OutboxEvent.cs
@@ -19,7 +19,9 @@ public abstract class OutboxEventBase : Entity<Guid>
 
     public string ObjectId { get; set; }
 
-    public string Message { get; set; }
+    public string Payload { get; set; }
 
     public bool Published { get; set; }
+
+    public string ActivityId { get; set; }
 }
diff --git a/...ModularMonolith/ClassifiedAds.Modules.Product/EventHandlers/ProductCreatedEventHandler.cs b/...ModularMonolith/ClassifiedAds.Modules.Product/EventHandlers/ProductCreatedEventHandler.cs
@@ -5,6 +5,7 @@
 using ClassifiedAds.Modules.Product.Constants;
 using ClassifiedAds.Modules.Product.Entities;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -45,8 +46,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = auditLog.CreatedDateTime,
             ObjectId = auditLog.Id.ToString(),
-            Message = auditLog.AsJsonString(),
-            Published = false,
+            Payload = auditLog.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
@@ -55,8 +56,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/...ModularMonolith/ClassifiedAds.Modules.Product/EventHandlers/ProductDeletedEventHandler.cs b/...ModularMonolith/ClassifiedAds.Modules.Product/EventHandlers/ProductDeletedEventHandler.cs
@@ -5,6 +5,7 @@
 using ClassifiedAds.Modules.Product.Constants;
 using ClassifiedAds.Modules.Product.Entities;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -45,8 +46,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = auditLog.CreatedDateTime,
             ObjectId = auditLog.Id.ToString(),
-            Message = auditLog.AsJsonString(),
-            Published = false,
+            Payload = auditLog.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
@@ -55,8 +56,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);

diff --git a/...ModularMonolith/ClassifiedAds.Modules.Product/EventHandlers/ProductUpdatedEventHandler.cs b/...ModularMonolith/ClassifiedAds.Modules.Product/EventHandlers/ProductUpdatedEventHandler.cs
@@ -5,6 +5,7 @@
 using ClassifiedAds.Modules.Product.Constants;
 using ClassifiedAds.Modules.Product.Entities;
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -45,8 +46,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = auditLog.CreatedDateTime,
             ObjectId = auditLog.Id.ToString(),
-            Message = auditLog.AsJsonString(),
-            Published = false,
+            Payload = auditLog.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
@@ -55,8 +56,8 @@ await _outboxEventRepository.AddOrUpdateAsync(new OutboxEvent
             TriggeredById = _currentUser.UserId,
             CreatedDateTime = domainEvent.EventDateTime,
             ObjectId = domainEvent.Entity.Id.ToString(),
-            Message = domainEvent.Entity.AsJsonString(),
-            Published = false,
+            Payload = domainEvent.Entity.AsJsonString(),
+            ActivityId = Activity.Current.Id,
         }, cancellationToken);
 
         await _outboxEventRepository.UnitOfWork.SaveChangesAsync(cancellationToken);