Implemented initial inbox processing logic

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs

@@ -1,8 +1,10 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.EntityFrameworkCore;
 using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.Timing;
 using Volo.Abp.Uow;
 
 namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
@@ -11,11 +13,14 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
         where TDbContext : IHasEventInbox
     {
         protected IDbContextProvider<TDbContext> DbContextProvider { get; }
+        protected IClock Clock { get; }
 
         public DbContextEventInbox(
-            IDbContextProvider<TDbContext> dbContextProvider)
+            IDbContextProvider<TDbContext> dbContextProvider,
+            IClock clock)
         {
             DbContextProvider = dbContextProvider;
+            Clock = clock;
         }
 
         [UnitOfWork]
@@ -35,6 +40,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
             var outgoingEventRecords = await dbContext
                 .IncomingEvents
                 .AsNoTracking()
+                .Where(x => !x.Processed)
                 .OrderBy(x => x.CreationTime)
                 .Take(maxCount)
                 .ToListAsync();
@@ -43,5 +49,16 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
                 .Select(x => x.ToIncomingEventInfo())
                 .ToList();
         }
+
+        public async Task MarkAsProcessedAsync(Guid id)
+        {
+            //TODO: Optimize?
+            var dbContext = (IHasEventInbox) await DbContextProvider.GetDbContextAsync();
+            var incomingEvent = await dbContext.IncomingEvents.FindAsync(id);
+            if (incomingEvent != null)
+            {
+                incomingEvent.MarkAsProcessed(Clock.Now);
+            }
+        }
     }
 }

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs

@@ -48,6 +48,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
         [UnitOfWork]
         public virtual async Task DeleteAsync(Guid id)
         {
+            //TODO: Optimize?
             var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync();
             var outgoingEvent = await dbContext.OutgoingEvents.FindAsync(id);
             if (outgoingEvent != null)

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs

@@ -20,6 +20,10 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
         public byte[] EventData { get; private set; }
         
         public DateTime CreationTime { get; private set; }
+        
+        public bool Processed { get; set; }
+        
+        public DateTime? ProcessedTime { get; set; }
 
         protected IncomingEventRecord()
         {
@@ -48,5 +52,11 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
                 CreationTime
             );
         }
+
+        public void MarkAsProcessed(DateTime processedTime)
+        {
+            Processed = true;
+            ProcessedTime = processedTime;
+        }
     }
 }

framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/AbpEventBusBoxesModule.cs

@@ -12,6 +12,7 @@ namespace Volo.Abp.EventBus.Boxes
         public override void OnApplicationInitialization(ApplicationInitializationContext context)
         {
             context.AddBackgroundWorker<OutboxSenderManager>();
+            context.AddBackgroundWorker<InboxProcessManager>();
         }
     }
 }

framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/IInboxProcessor.cs

@@ -0,0 +1,13 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Volo.Abp.EventBus.Distributed;
+
+namespace Volo.Abp.EventBus.Boxes
+{
+    public interface IInboxProcessor
+    {
+        Task StartAsync(InboxConfig inboxConfig, CancellationToken cancellationToken = default);
+        
+        Task StopAsync(CancellationToken cancellationToken = default);
+    }
+}

framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/IOutboxSender.cs

@@ -7,6 +7,7 @@ namespace Volo.Abp.EventBus.Boxes
     public interface IOutboxSender
     {
         Task StartAsync(OutboxConfig outboxConfig, CancellationToken cancellationToken = default);
+        
         Task StopAsync(CancellationToken cancellationToken = default);
     }
 }

framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessManager.cs

@@ -0,0 +1,48 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+using Volo.Abp.BackgroundWorkers;
+using Volo.Abp.EventBus.Distributed;
+
+namespace Volo.Abp.EventBus.Boxes
+{
+    public class InboxProcessManager : IBackgroundWorker
+    {
+        protected AbpDistributedEventBusOptions Options { get; }
+        protected IServiceProvider ServiceProvider { get; }
+        protected List<IInboxProcessor> Processors { get; }
+
+        public InboxProcessManager(
+            IOptions<AbpDistributedEventBusOptions> options,
+            IServiceProvider serviceProvider)
+        {
+            ServiceProvider = serviceProvider;
+            Options = options.Value;
+            Processors = new List<IInboxProcessor>();
+        }
+
+        public async Task StartAsync(CancellationToken cancellationToken = default)
+        {
+            foreach (var inboxConfig in Options.Inboxes.Values)
+            {
+                if (inboxConfig.IsProcessingEnabled)
+                {
+                    var processor = ServiceProvider.GetRequiredService<IInboxProcessor>();
+                    await processor.StartAsync(inboxConfig, cancellationToken);
+                    Processors.Add(processor);
+                }
+            }
+        }
+
+        public async Task StopAsync(CancellationToken cancellationToken = default)
+        {
+            foreach (var processor in Processors)
+            {
+                await processor.StopAsync(cancellationToken);
+            }
+        }
+    }
+}

framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs

@@ -0,0 +1,111 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Medallion.Threading;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Volo.Abp.DependencyInjection;
+using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.Threading;
+using Volo.Abp.Uow;
+
+namespace Volo.Abp.EventBus.Boxes
+{
+    public class InboxProcessor : IInboxProcessor, ITransientDependency
+    {
+        protected IServiceProvider ServiceProvider { get; }
+        protected AbpTimer Timer { get; }
+        protected IDistributedEventBus DistributedEventBus { get; }
+        protected IDistributedLockProvider DistributedLockProvider { get; }
+        protected IUnitOfWorkManager UnitOfWorkManager { get; }
+        protected IEventInbox Inbox { get; private set; }
+        protected InboxConfig InboxConfig { get; private set; }
+        
+        protected string DistributedLockName => "Inbox_" + InboxConfig.Name;
+        public ILogger<InboxProcessor> Logger { get; set; }
+
+        public InboxProcessor(
+            IServiceProvider serviceProvider,
+            AbpTimer timer,
+            IDistributedEventBus distributedEventBus, 
+            IDistributedLockProvider distributedLockProvider,
+            IUnitOfWorkManager unitOfWorkManager)
+        {
+            ServiceProvider = serviceProvider;
+            Timer = timer;
+            DistributedEventBus = distributedEventBus;
+            DistributedLockProvider = distributedLockProvider;
+            UnitOfWorkManager = unitOfWorkManager;
+            Timer.Period = 2000; //TODO: Config?
+            Timer.Elapsed += TimerOnElapsed;
+            Logger = NullLogger<InboxProcessor>.Instance;
+        }
+        
+        private void TimerOnElapsed(object sender, EventArgs e)
+        {
+            AsyncHelper.RunSync(RunAsync);
+        }
+        
+        public Task StartAsync(InboxConfig inboxConfig, CancellationToken cancellationToken = default)
+        {
+            InboxConfig = inboxConfig;
+            Inbox = (IEventInbox)ServiceProvider.GetRequiredService(inboxConfig.ImplementationType);
+            Timer.Start(cancellationToken);
+            return Task.CompletedTask;    
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken = default)
+        {
+            Timer.Stop(cancellationToken);
+            return Task.CompletedTask;
+        }
+        
+        protected virtual async Task RunAsync()
+        {
+            await using (var handle = await DistributedLockProvider.TryAcquireLockAsync(DistributedLockName))
+            {
+                if (handle != null)
+                {
+                    Logger.LogDebug("Obtained the distributed lock: " + DistributedLockName);
+                    
+                    while (true)
+                    {
+                        var waitingEvents = await Inbox.GetWaitingEventsAsync(1000); //TODO: Config?
+                        if (waitingEvents.Count <= 0)
+                        {
+                            break;
+                        }
+
+                        Logger.LogInformation($"Found {waitingEvents.Count} events in the inbox.");
+                
+                        foreach (var waitingEvent in waitingEvents)
+                        {
+                            using (var uow = UnitOfWorkManager.Begin(isTransactional: true, requiresNew: true))
+                            {
+                                await DistributedEventBus
+                                    .AsRawEventPublisher()
+                                    .ProcessRawAsync(waitingEvent.EventName, waitingEvent.EventData);
+                                
+                                /*
+                                await DistributedEventBus
+                                    .AsRawEventPublisher()
+                                    .PublishRawAsync(waitingEvent.Id, waitingEvent.EventName, waitingEvent.EventData);
+                                */
+                                await Inbox.MarkAsProcessedAsync(waitingEvent.Id);
+                                await uow.CompleteAsync();
+                            }
+                            
+                            Logger.LogInformation($"Processed the incoming event with id = {waitingEvent.Id:N}");
+                        }
+                    }
+                }
+                else
+                {
+                    Logger.LogDebug("Could not obtain the distributed lock: " + DistributedLockName);
+                    await Task.Delay(7000); //TODO: Can we pass a cancellation token to cancel on shutdown? (Config?)
+                }
+            }
+        }
+    }
+}

framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSenderManager.cs

@@ -31,7 +31,7 @@ namespace Volo.Abp.EventBus.Boxes
                 if (outboxConfig.IsSendingEnabled)
                 {
                     var sender = ServiceProvider.GetRequiredService<IOutboxSender>();
-                    await sender.StartAsync(outboxConfig);
+                    await sender.StartAsync(outboxConfig, cancellationToken);
                     Senders.Add(sender);
                 }
             }
@@ -41,7 +41,7 @@ namespace Volo.Abp.EventBus.Boxes
         {
             foreach (var sender in Senders)
             {
-                await sender.StopAsync();
+                await sender.StopAsync(cancellationToken);
             }
         }
     }

framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs

@@ -211,6 +211,23 @@ namespace Volo.Abp.EventBus.Kafka
             );
         }
 
+        public override async Task ProcessRawAsync(string eventName, byte[] eventDataBytes)
+        {
+            var eventType = EventTypes.GetOrDefault(eventName);
+            if (eventType == null)
+            {
+                return;
+            }
+            
+            var eventData = Serializer.Deserialize(eventDataBytes, eventType);
+            var exceptions = new List<Exception>();
+            await TriggerHandlersAsync(eventType, eventData, exceptions);
+            if (exceptions.Any())
+            {
+                ThrowOriginalExceptions(eventType, exceptions);
+            }
+        }
+
         protected override byte[] Serialize(object eventData)
         {
             return Serializer.Serialize(eventData);

framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs

@@ -123,7 +123,7 @@ namespace Volo.Abp.EventBus.RabbitMq
                     retryAttempt = (int)ea.BasicProperties.Headers[EventErrorHandlerBase.RetryAttemptKey];
                 }
 
-                errorContext.EventData = Serializer.Deserialize(ea.Body.ToArray(), eventType);
+                errorContext.EventData = Serializer.Deserialize(eventBytes, eventType);
                 errorContext.SetProperty(EventErrorHandlerBase.HeadersKey, ea.BasicProperties);
                 errorContext.SetProperty(EventErrorHandlerBase.RetryAttemptKey, retryAttempt);
             });
@@ -217,6 +217,25 @@ namespace Volo.Abp.EventBus.RabbitMq
             return PublishAsync(eventName, eventData, null, eventId: eventId);
         }
 
+        public override async Task ProcessRawAsync(string eventName, byte[] eventDataBytes)
+        {
+            //TODO: We have a duplication in logic and also with the kafka side!
+            
+            var eventType = EventTypes.GetOrDefault(eventName);
+            if (eventType == null)
+            {
+                return;
+            }
+            
+            var eventData = Serializer.Deserialize(eventDataBytes, eventType);
+            var exceptions = new List<Exception>();
+            await TriggerHandlersAsync(eventType, eventData, exceptions);
+            if (exceptions.Any())
+            {
+                ThrowOriginalExceptions(eventType, exceptions);
+            }
+        }
+        
         protected override byte[] Serialize(object eventData)
         {
             return Serializer.Serialize(eventData);

framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs

@@ -194,6 +194,12 @@ namespace Volo.Abp.EventBus.Rebus
             throw new NotImplementedException();
         }
 
+        public override Task ProcessRawAsync(string eventName, byte[] eventDataBytes)
+        {
+            /* TODO: IMPLEMENT! */
+            throw new NotImplementedException();
+        }
+
         protected override byte[] Serialize(object eventData)
         {
             return Serializer.Serialize(eventData);

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs

@@ -81,6 +81,7 @@ namespace Volo.Abp.EventBus.Distributed
         }
 
         public abstract Task PublishRawAsync(Guid eventId, string eventName, byte[] eventData);
+        public abstract Task ProcessRawAsync(string eventName, byte[] eventDataBytes);
 
         private async Task<bool> AddToOutboxAsync(Type eventType, object eventData)
         {
@@ -128,6 +129,7 @@ namespace Volo.Abp.EventBus.Distributed
                     if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType))
                     {
                         var eventInbox = (IEventInbox) scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType);
+                        //TODO: Check if event was received before!!
                         await eventInbox.EnqueueAsync(
                             new IncomingEventInfo(
                                 GuidGenerator.Create(),

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs

@@ -1,4 +1,5 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace Volo.Abp.EventBus.Distributed
@@ -8,5 +9,7 @@ namespace Volo.Abp.EventBus.Distributed
         Task EnqueueAsync(IncomingEventInfo incomingEvent);
         
         Task<List<IncomingEventInfo>> GetWaitingEventsAsync(int maxCount);
+        
+        Task MarkAsProcessedAsync(Guid id);
     }
 }

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IRawEventPublisher.cs

@@ -3,11 +3,16 @@ using System.Threading.Tasks;
 
 namespace Volo.Abp.EventBus.Distributed
 {
-    public interface IRawEventPublisher
+    public interface IRawEventPublisher //TODO: Rename: ISupportsEventBoxes
     {
         Task PublishRawAsync(
             Guid eventId,
             string eventName,
             byte[] eventData);
+
+        Task ProcessRawAsync(
+            string eventName,
+            byte[] eventDataBytes
+        );
     }
 }

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs

@@ -162,6 +162,19 @@ namespace Volo.Abp.EventBus
                 }
             }
         }
+        
+        protected void ThrowOriginalExceptions(Type eventType, List<Exception> exceptions)
+        {
+            if (exceptions.Count == 1)
+            {
+                exceptions[0].ReThrow();
+            }
+
+            throw new AggregateException(
+                "More than one error has occurred while triggering the event: " + eventType,
+                exceptions
+            );
+        }
 
         protected virtual void SubscribeHandlers(ITypeList<IEventHandler> handlers)
         {

test/DistEvents/DistDemoApp/Migrations/20210909182251_Added_Inbox_Process_Columns.Designer.cs

@@ -0,0 +1,155 @@
+// <auto-generated />
+using System;
+using DistDemoApp;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Metadata;
+using Microsoft.EntityFrameworkCore.Migrations;
+using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
+using Volo.Abp.EntityFrameworkCore;
+
+namespace DistDemoApp.Migrations
+{
+    [DbContext(typeof(TodoDbContext))]
+    [Migration("20210909182251_Added_Inbox_Process_Columns")]
+    partial class Added_Inbox_Process_Columns
+    {
+        protected override void BuildTargetModel(ModelBuilder modelBuilder)
+        {
+#pragma warning disable 612, 618
+            modelBuilder
+                .HasAnnotation("_Abp_DatabaseProvider", EfCoreDatabaseProvider.SqlServer)
+                .HasAnnotation("Relational:MaxIdentifierLength", 128)
+                .HasAnnotation("ProductVersion", "5.0.9")
+                .HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.IdentityColumn);
+
+            modelBuilder.Entity("DistDemoApp.TodoItem", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .HasColumnType("uniqueidentifier");
+
+                    b.Property<string>("ConcurrencyStamp")
+                        .IsConcurrencyToken()
+                        .HasMaxLength(40)
+                        .HasColumnType("nvarchar(40)")
+                        .HasColumnName("ConcurrencyStamp");
+
+                    b.Property<DateTime>("CreationTime")
+                        .HasColumnType("datetime2")
+                        .HasColumnName("CreationTime");
+
+                    b.Property<Guid?>("CreatorId")
+                        .HasColumnType("uniqueidentifier")
+                        .HasColumnName("CreatorId");
+
+                    b.Property<string>("ExtraProperties")
+                        .HasColumnType("nvarchar(max)")
+                        .HasColumnName("ExtraProperties");
+
+                    b.Property<string>("Text")
+                        .IsRequired()
+                        .HasMaxLength(128)
+                        .HasColumnType("nvarchar(128)");
+
+                    b.HasKey("Id");
+
+                    b.ToTable("TodoItems");
+                });
+
+            modelBuilder.Entity("DistDemoApp.TodoSummary", b =>
+                {
+                    b.Property<int>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("int")
+                        .HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.IdentityColumn);
+
+                    b.Property<string>("ConcurrencyStamp")
+                        .IsConcurrencyToken()
+                        .HasMaxLength(40)
+                        .HasColumnType("nvarchar(40)")
+                        .HasColumnName("ConcurrencyStamp");
+
+                    b.Property<byte>("Day")
+                        .HasColumnType("tinyint");
+
+                    b.Property<string>("ExtraProperties")
+                        .HasColumnType("nvarchar(max)")
+                        .HasColumnName("ExtraProperties");
+
+                    b.Property<byte>("Month")
+                        .HasColumnType("tinyint");
+
+                    b.Property<int>("TotalCount")
+                        .HasColumnType("int");
+
+                    b.Property<int>("Year")
+                        .HasColumnType("int");
+
+                    b.HasKey("Id");
+
+                    b.ToTable("TodoSummaries");
+                });
+
+            modelBuilder.Entity("Volo.Abp.EntityFrameworkCore.DistributedEvents.IncomingEventRecord", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .HasColumnType("uniqueidentifier");
+
+                    b.Property<DateTime>("CreationTime")
+                        .HasColumnType("datetime2")
+                        .HasColumnName("CreationTime");
+
+                    b.Property<byte[]>("EventData")
+                        .IsRequired()
+                        .HasColumnType("varbinary(max)");
+
+                    b.Property<string>("EventName")
+                        .IsRequired()
+                        .HasMaxLength(256)
+                        .HasColumnType("nvarchar(256)");
+
+                    b.Property<string>("ExtraProperties")
+                        .HasColumnType("nvarchar(max)")
+                        .HasColumnName("ExtraProperties");
+
+                    b.Property<bool>("Processed")
+                        .HasColumnType("bit");
+
+                    b.Property<DateTime?>("ProcessedTime")
+                        .HasColumnType("datetime2");
+
+                    b.HasKey("Id");
+
+                    b.ToTable("AbpEventInbox");
+                });
+
+            modelBuilder.Entity("Volo.Abp.EntityFrameworkCore.DistributedEvents.OutgoingEventRecord", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .HasColumnType("uniqueidentifier");
+
+                    b.Property<DateTime>("CreationTime")
+                        .HasColumnType("datetime2")
+                        .HasColumnName("CreationTime");
+
+                    b.Property<byte[]>("EventData")
+                        .IsRequired()
+                        .HasColumnType("varbinary(max)");
+
+                    b.Property<string>("EventName")
+                        .IsRequired()
+                        .HasMaxLength(256)
+                        .HasColumnType("nvarchar(256)");
+
+                    b.Property<string>("ExtraProperties")
+                        .HasColumnType("nvarchar(max)")
+                        .HasColumnName("ExtraProperties");
+
+                    b.HasKey("Id");
+
+                    b.ToTable("AbpEventOutbox");
+                });
+#pragma warning restore 612, 618
+        }
+    }
+}

test/DistEvents/DistDemoApp/Migrations/20210909182251_Added_Inbox_Process_Columns.cs

@@ -0,0 +1,35 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+namespace DistDemoApp.Migrations
+{
+    public partial class Added_Inbox_Process_Columns : Migration
+    {
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.AddColumn<bool>(
+                name: "Processed",
+                table: "AbpEventInbox",
+                type: "bit",
+                nullable: false,
+                defaultValue: false);
+
+            migrationBuilder.AddColumn<DateTime>(
+                name: "ProcessedTime",
+                table: "AbpEventInbox",
+                type: "datetime2",
+                nullable: true);
+        }
+
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropColumn(
+                name: "Processed",
+                table: "AbpEventInbox");
+
+            migrationBuilder.DropColumn(
+                name: "ProcessedTime",
+                table: "AbpEventInbox");
+        }
+    }
+}

test/DistEvents/DistDemoApp/Migrations/TodoDbContextModelSnapshot.cs

@@ -110,6 +110,12 @@ namespace DistDemoApp.Migrations
                         .HasColumnType("nvarchar(max)")
                         .HasColumnName("ExtraProperties");
 
+                    b.Property<bool>("Processed")
+                        .HasColumnType("bit");
+
+                    b.Property<DateTime?>("ProcessedTime")
+                        .HasColumnType("datetime2");
+
                     b.HasKey("Id");
 
                     b.ToTable("AbpEventInbox");