Implemented saga logic

Author: johnsimons

src/ServiceControl.Audit.Persistence.PostgreSQL/PostgreSQLAuditDataStore.cs

@@ -163,7 +163,39 @@ public Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpointAn
         return ExecuteMessagesQuery(builder, cancellationToken);
     }
 
-    public Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken) => throw new NotImplementedException();
+    public async Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken)
+    {
+        await using var conn = await connectionFactory.OpenConnection(cancellationToken);
+        await using var cmd = new NpgsqlCommand(@"
+            SELECT 
+                id,
+                saga_id,
+                saga_type,
+                changes
+            FROM saga_snapshots
+            WHERE saga_id = @saga_id
+            LIMIT 1", conn);
+
+        cmd.Parameters.AddWithValue("saga_id", input);
+
+        await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
+
+        if (await reader.ReadAsync(cancellationToken))
+        {
+            var changes = GetValue<List<SagaStateChange>>(reader, "changes") ?? [];
+            var sagaHistory = new SagaHistory
+            {
+                Id = GetValue<Guid>(reader, "id"),
+                SagaId = GetValue<Guid>(reader, "saga_id"),
+                SagaType = GetValue<string>(reader, "saga_type"),
+                Changes = changes
+            };
+
+            return new QueryResult<SagaHistory>(sagaHistory, new QueryStatsInfo(string.Empty, changes.Count));
+        }
+
+        return QueryResult<SagaHistory>.Empty();
+    }
 
     async Task<QueryResult<IList<MessagesView>>> ExecuteMessagesQuery(
         PostgresqlMessagesQueryBuilder builder,

src/ServiceControl.Audit.Persistence.PostgreSQL/PostgreSQLPersistenceInstaller.cs

@@ -130,17 +130,19 @@ CREATE INDEX IF NOT EXISTS idx_processed_messages_by_message_id ON processed_mes
             await cmd.ExecuteNonQueryAsync(cancellationToken);
         }
 
+
         await using (var cmd = new NpgsqlCommand(@"
-            CREATE INDEX IF NOT EXISTS idx_processed_messages_by_conversation_id ON processed_messages (
-                conversation_id
+            CREATE INDEX IF NOT EXISTS idx_processed_messages_by_created_at ON processed_messages (
+                created_at
             );", connection))
         {
             await cmd.ExecuteNonQueryAsync(cancellationToken);
         }
 
         await using (var cmd = new NpgsqlCommand(@"
-            CREATE INDEX IF NOT EXISTS idx_processed_messages_by_created_at ON processed_messages (
-                created_at
+            CREATE INDEX IF NOT EXISTS idx_processed_messages_by_conversation ON processed_messages (
+                conversation_id,
+                time_sent
             );", connection))
         {
             await cmd.ExecuteNonQueryAsync(cancellationToken);
@@ -157,17 +159,19 @@ CREATE INDEX IF NOT EXISTS idx_processed_messages_by_query ON processed_messages
         // Create saga_snapshots table
         await using (var cmd = new NpgsqlCommand(@"
             CREATE TABLE IF NOT EXISTS saga_snapshots (
-                id TEXT PRIMARY KEY,
+                id UUID PRIMARY KEY,
                 saga_id UUID,
                 saga_type TEXT,
-                start_time TIMESTAMPTZ,
-                finish_time TIMESTAMPTZ,
-                status TEXT,
-                state_after_change TEXT,
-                initiating_message JSONB,
-                outgoing_messages JSONB,
-                endpoint TEXT,
-                processed_at TIMESTAMPTZ
+                changes JSONB
+            );", connection))
+        {
+            await cmd.ExecuteNonQueryAsync(cancellationToken);
+        }
+
+        // Create index on saga_snapshots for faster saga_id lookups
+        await using (var cmd = new NpgsqlCommand(@"
+            CREATE INDEX IF NOT EXISTS idx_saga_snapshots_saga_id ON saga_snapshots (
+                saga_id
             );", connection))
         {
             await cmd.ExecuteNonQueryAsync(cancellationToken);

src/ServiceControl.Audit.Persistence.PostgreSQL/UnitOfWork/PostgreSQLAuditIngestionUnitOfWork.cs

@@ -78,27 +78,29 @@ INSERT INTO processed_messages (
 
     public Task RecordSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken)
     {
-        // Insert SagaSnapshot into saga_snapshots table
+        var newChange = new
+        {
+            sagaSnapshot.StartTime,
+            sagaSnapshot.FinishTime,
+            sagaSnapshot.Status,
+            sagaSnapshot.StateAfterChange,
+            sagaSnapshot.InitiatingMessage,
+            sagaSnapshot.OutgoingMessages,
+            sagaSnapshot.Endpoint
+        };
+
+        // Insert or update saga_snapshots table - add new change to the changes array
         var cmd = batch.CreateBatchCommand();
         cmd.CommandText = @"
-                INSERT INTO saga_snapshots (
-                    id, saga_id, saga_type, start_time, finish_time, status, state_after_change, initiating_message, outgoing_messages, endpoint, processed_at
-                ) VALUES (
-                    @id, @saga_id, @saga_type, @start_time, @finish_time, @status, @state_after_change, @initiating_message, @outgoing_messages, @endpoint, @processed_at
-                )
-                ON CONFLICT (id) DO NOTHING;";
+                INSERT INTO saga_snapshots (id, saga_id, saga_type, changes)
+                VALUES (@saga_id, @saga_id, @saga_type, @new_change)
+                ON CONFLICT (id) DO UPDATE SET
+                    changes = COALESCE(saga_snapshots.changes, '[]'::jsonb) || @new_change::jsonb;";
 
-        cmd.Parameters.AddWithValue("id", sagaSnapshot.Id);
         cmd.Parameters.AddWithValue("saga_id", sagaSnapshot.SagaId);
         cmd.Parameters.AddWithValue("saga_type", sagaSnapshot.SagaType);
-        cmd.Parameters.AddWithValue("start_time", sagaSnapshot.StartTime);
-        cmd.Parameters.AddWithValue("finish_time", sagaSnapshot.FinishTime);
-        cmd.Parameters.AddWithValue("status", sagaSnapshot.Status.ToString());
-        cmd.Parameters.AddWithValue("state_after_change", sagaSnapshot.StateAfterChange);
-        cmd.Parameters.AddWithValue("initiating_message", NpgsqlTypes.NpgsqlDbType.Jsonb, sagaSnapshot.InitiatingMessage);
-        cmd.Parameters.AddWithValue("outgoing_messages", NpgsqlTypes.NpgsqlDbType.Jsonb, sagaSnapshot.OutgoingMessages);
-        cmd.Parameters.AddWithValue("endpoint", sagaSnapshot.Endpoint);
-        cmd.Parameters.AddWithValue("processed_at", sagaSnapshot.ProcessedAt);
+        cmd.Parameters.AddWithValue("new_change", NpgsqlTypes.NpgsqlDbType.Jsonb, new[] { newChange });
+
         batch.BatchCommands.Add(cmd);
 
         return Task.CompletedTask;