Fix event processing for synced events (#1053)

TeachingRecordSystem/src/TeachingRecordSystem.Core/DataStore/Postgres/Mappings/EventMapping.cs

@@ -11,6 +11,7 @@ public void Configure(EntityTypeBuilder<Event> builder)
         builder.ToTable("events");
         builder.Property(e => e.EventName).IsRequired().HasMaxLength(200);
         builder.Property(e => e.Created).IsRequired();
+        builder.Property(e => e.Inserted).IsRequired();
         builder.Property(e => e.Payload).IsRequired().HasColumnType("jsonb");
         builder.Property(e => e.Published);
         builder.HasKey(e => e.EventId);

TeachingRecordSystem/src/TeachingRecordSystem.Core/DataStore/Postgres/Migrations/20231221141110_EventInserted.cs

@@ -0,0 +1,41 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace TeachingRecordSystem.Core.DataStore.Postgres.Migrations
+{
+    /// <inheritdoc />
+    public partial class EventInserted : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.AddColumn<DateTime>(
+                name: "inserted",
+                table: "events",
+                type: "timestamp with time zone",
+                nullable: true);
+
+            migrationBuilder.Sql("update events set inserted = created where inserted is null");
+
+            migrationBuilder.AlterColumn<DateTime>(
+                name: "inserted",
+                table: "events",
+                type: "timestamp with time zone",
+                nullable: false,
+                defaultValue: new DateTime(1, 1, 1, 0, 0, 0, 0, DateTimeKind.Unspecified),
+                oldClrType: typeof(DateTime),
+                oldType: "timestamp with time zone",
+                oldNullable: true);
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropColumn(
+                name: "inserted",
+                table: "events");
+        }
+    }
+}

TeachingRecordSystem/src/TeachingRecordSystem.Core/DataStore/Postgres/Migrations/TrsDbContextModelSnapshot.cs

@@ -79,6 +79,11 @@ protected override void BuildModel(ModelBuilder modelBuilder)
                         .HasColumnType("character varying(200)")
                         .HasColumnName("event_name");
 
+                    b.Property<DateTime>("Inserted")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("timestamp with time zone")
+                        .HasColumnName("inserted");
+
                     b.Property<string>("Payload")
                         .IsRequired()
                         .HasColumnType("jsonb")

TeachingRecordSystem/src/TeachingRecordSystem.Core/DataStore/Postgres/Models/Event.cs

@@ -8,19 +8,21 @@ public class Event
     public required Guid EventId { get; init; }
     public required string EventName { get; init; }
     public required DateTime Created { get; init; }
+    public required DateTime Inserted { get; init; }
     public required string Payload { get; init; }
     public bool Published { get; set; }
 
-    public static Event FromEventBase(EventBase @event)
+    public static Event FromEventBase(EventBase @event, DateTime? inserted)
     {
         var eventName = @event.GetEventName();
         var payload = JsonSerializer.Serialize(@event, inputType: @event.GetType(), EventBase.JsonSerializerOptions);
 
         return new Event()
         {
             EventId = @event.EventId,
-            Created = @event.CreatedUtc,
             EventName = eventName,
+            Created = @event.CreatedUtc,
+            Inserted = inserted ?? @event.CreatedUtc,
             Payload = payload
         };
     }

TeachingRecordSystem/src/TeachingRecordSystem.Core/DataStore/Postgres/TrsDbContext.cs

@@ -63,9 +63,9 @@ public static void ConfigureOptions(DbContextOptionsBuilder optionsBuilder, stri
             .UseSnakeCaseNamingConvention();
     }
 
-    public void AddEvent(EventBase @event)
+    public void AddEvent(EventBase @event, DateTime? inserted = null)
     {
-        Events.Add(Event.FromEventBase(@event));
+        Events.Add(Event.FromEventBase(@event, inserted));
     }
 
     protected override void ConfigureConventions(ModelConfigurationBuilder configurationBuilder)

TeachingRecordSystem/src/TeachingRecordSystem.Core/Events/Processing/PublishEventsBackgroundService.cs

@@ -60,9 +60,11 @@ public async Task PublishEvents(CancellationToken cancellationToken)
                 .FromSql($"""
                     select * from events
                     where published is false
-                    and created > {lastProcessedEventTimestamp}
+                    and inserted > {lastProcessedEventTimestamp}
                     and event_id != {lastProcessedEventId}
-                    for update skip locked limit {BatchSize}
+                    order by inserted, created
+                    for update skip locked
+                    limit {BatchSize}
                     """)
                 .ToListAsync(cancellationToken: cancellationToken);
 
@@ -88,7 +90,7 @@ and created > {lastProcessedEventTimestamp}
                 }
                 finally
                 {
-                    lastProcessedEventTimestamp = e.Created;
+                    lastProcessedEventTimestamp = e.Inserted;
                     lastProcessedEventId = e.EventId;
                 }
             }

TeachingRecordSystem/src/TeachingRecordSystem.Core/Services/TrsDataSync/TrsDataSyncHelper.cs

@@ -582,6 +582,7 @@ private async Task<int> SyncEvents(IReadOnlyCollection<EventBase> events, Npgsql
             "event_id",
             "event_name",
             "created",
+            "inserted",
             "payload"
         };
 
@@ -592,6 +593,7 @@ private async Task<int> SyncEvents(IReadOnlyCollection<EventBase> events, Npgsql
                 event_id UUID NOT NULL,
                 event_name VARCHAR(200) NOT NULL,
                 created TIMESTAMP WITH TIME ZONE NOT NULL,
+                inserted TIMESTAMP WITH TIME ZONE NOT NULL,
                 payload JSONB NOT NULL
             )
             """;
@@ -622,6 +624,7 @@ ON CONFLICT (event_id) DO NOTHING
             writer.WriteValueOrNull(e.EventId, NpgsqlDbType.Uuid);
             writer.WriteValueOrNull(e.GetEventName(), NpgsqlDbType.Varchar);
             writer.WriteValueOrNull(e.CreatedUtc, NpgsqlDbType.TimestampTz);
+            writer.WriteValueOrNull(clock.UtcNow, NpgsqlDbType.TimestampTz);
             writer.WriteValueOrNull(payload, NpgsqlDbType.Jsonb);
         }
 

TeachingRecordSystem/tests/TeachingRecordSystem.Core.Tests/Events/Processing/PublishEventsBackgroundServiceTests.cs

@@ -8,14 +8,9 @@
 namespace TeachingRecordSystem.Core.Tests.Events.Processing;
 
 [Collection(nameof(DisableParallelization))]
-public class PublishEventsBackgroundServiceTests : IAsyncLifetime
+public class PublishEventsBackgroundServiceTests(DbFixture dbFixture) : IAsyncLifetime
 {
-    private readonly DbFixture _dbFixture;
-
-    public PublishEventsBackgroundServiceTests(DbFixture dbFixture)
-    {
-        _dbFixture = dbFixture;
-    }
+    private readonly DbFixture _dbFixture = dbFixture;
 
     public Task InitializeAsync() =>
         _dbFixture.WithDbContext(dbContext => dbContext.Database.ExecuteSqlAsync($"delete from events"));

TeachingRecordSystem/tests/TeachingRecordSystem.TestCommon/DbFixture.cs

@@ -4,17 +4,11 @@
 
 namespace TeachingRecordSystem.TestCommon;
 
-public class DbFixture
+public class DbFixture(DbHelper dbHelper, IServiceProvider serviceProvider)
 {
-    public DbFixture(DbHelper dbHelper, IServiceProvider serviceProvider)
-    {
-        DbHelper = dbHelper;
-        Services = serviceProvider;
-    }
-
-    public DbHelper DbHelper { get; }
+    public DbHelper DbHelper { get; } = dbHelper;
 
-    public IServiceProvider Services { get; }
+    public IServiceProvider Services { get; } = serviceProvider;
 
     public TrsDbContext GetDbContext() => Services.GetRequiredService<TrsDbContext>();