Adding it back

Author: johnsimons

src/ServiceControl.Audit.Persistence.PostgreSQL/BodyStorage/PostgreSQLAttachmentsBodyStorage.cs

@@ -0,0 +1,13 @@
+namespace ServiceControl.Audit.Persistence.PostgreSQL.BodyStorage
+{
+    using System.IO;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using ServiceControl.Audit.Auditing.BodyStorage;
+
+    public class PostgreSQLAttachmentsBodyStorage : IBodyStorage
+    {
+        public Task Store(string bodyId, string contentType, int bodySize, Stream bodyStream, CancellationToken cancellationToken) => throw new System.NotImplementedException();
+        public Task<StreamResult> TryFetch(string bodyId, CancellationToken cancellationToken) => throw new System.NotImplementedException();
+    }
+}

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

@@ -0,0 +1,26 @@
+namespace ServiceControl.Audit.Persistence.PostgreSQL
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using ServiceControl.Audit.Auditing;
+    using ServiceControl.Audit.Auditing.MessagesView;
+    using ServiceControl.Audit.Infrastructure;
+    using ServiceControl.Audit.Monitoring;
+    using ServiceControl.Audit.Persistence;
+    using ServiceControl.SagaAudit;
+
+    public class PostgreSQLAuditDataStore : IAuditDataStore
+    {
+        public Task<MessageBodyView> GetMessageBody(string messageId, CancellationToken cancellationToken) => throw new NotImplementedException();
+        public Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default) => throw new NotImplementedException();
+        public Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken) => throw new NotImplementedException();
+        public Task<QueryResult<IList<KnownEndpointsView>>> QueryKnownEndpoints(CancellationToken cancellationToken) => throw new NotImplementedException();
+        public Task<QueryResult<IList<MessagesView>>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default) => throw new NotImplementedException();
+        public Task<QueryResult<IList<MessagesView>>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken) => throw new NotImplementedException();
+        public Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default) => throw new NotImplementedException();
+        public Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default) => throw new NotImplementedException();
+        public Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken) => throw new NotImplementedException();
+    }
+}

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

@@ -0,0 +1,28 @@
+namespace ServiceControl.Audit.Persistence.PostgreSQL
+{
+    using Npgsql;
+    using System.Threading.Tasks;
+    using System.Threading;
+
+    public class PostgreSQLConnectionFactory
+    {
+        readonly string connectionString;
+
+        public PostgreSQLConnectionFactory(string connectionString)
+        {
+            this.connectionString = connectionString;
+        }
+
+        public NpgsqlConnection CreateConnection()
+        {
+            return new NpgsqlConnection(connectionString);
+        }
+
+        public async Task<NpgsqlConnection> OpenConnectionAsync(CancellationToken cancellationToken = default)
+        {
+            var conn = new NpgsqlConnection(connectionString);
+            await conn.OpenAsync(cancellationToken).ConfigureAwait(false);
+            return conn;
+        }
+    }
+}

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

@@ -0,0 +1,15 @@
+namespace ServiceControl.Audit.Persistence.PostgreSQL
+{
+    using System;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using ServiceControl.Audit.Auditing;
+    using ServiceControl.Audit.Persistence;
+
+    public class PostgreSQLFailedAuditStorage : IFailedAuditStorage
+    {
+        public Task<int> GetFailedAuditsCount() => throw new NotImplementedException();
+        public Task ProcessFailedMessages(Func<FailedTransportMessage, Func<CancellationToken, Task>, CancellationToken, Task> onMessage, CancellationToken cancellationToken) => throw new NotImplementedException();
+        public Task SaveFailedAuditImport(FailedAuditImport message) => throw new NotImplementedException();
+    }
+}

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

@@ -0,0 +1,26 @@
+namespace ServiceControl.Audit.Persistence.PostgreSQL
+{
+    using Microsoft.Extensions.DependencyInjection;
+    using ServiceControl.Audit.Auditing.BodyStorage;
+    using ServiceControl.Audit.Persistence;
+    using ServiceControl.Audit.Persistence.PostgreSQL.BodyStorage;
+    using ServiceControl.Audit.Persistence.PostgreSQL.UnitOfWork;
+    using ServiceControl.Audit.Persistence.UnitOfWork;
+
+    public class PostgreSQLPersistence : IPersistence
+    {
+        public void AddInstaller(IServiceCollection services)
+        {
+            AddPersistence(services);
+        }
+
+        public void AddPersistence(IServiceCollection services)
+        {
+            services.AddSingleton<IAuditDataStore, PostgreSQLAuditDataStore>();
+            services.AddSingleton<IAuditIngestionUnitOfWorkFactory, PostgreSQLAuditIngestionUnitOfWorkFactory>();
+            services.AddSingleton<IFailedAuditStorage, PostgreSQLFailedAuditStorage>();
+            services.AddSingleton<IBodyStorage, PostgreSQLAttachmentsBodyStorage>();
+            services.AddSingleton<PostgreSQLConnectionFactory>();
+        }
+    }
+}

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

@@ -0,0 +1,82 @@
+namespace ServiceControl.Audit.Persistence.PostgreSQL
+{
+    using System.Collections.Generic;
+    using Npgsql;
+    using ServiceControl.Audit.Persistence;
+
+    public class PostgreSQLPersistenceConfiguration : IPersistenceConfiguration
+    {
+        public string Name => "PostgreSQL";
+
+        public IEnumerable<string> ConfigurationKeys => new[] { "PostgreSqlConnectionString" };
+
+        public IPersistence Create(PersistenceSettings settings)
+        {
+            settings.PersisterSpecificSettings.TryGetValue("PostgreSqlConnectionString", out var connectionString);
+            using var connection = new NpgsqlConnection(connectionString);
+            connection.Open();
+
+            // Create processed_messages table
+            using (var cmd = new NpgsqlCommand(@"
+                        CREATE TABLE IF NOT EXISTS processed_messages (
+                        id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+                        unique_message_id TEXT,
+                        message_metadata JSONB,
+                        headers JSONB,
+                        processed_at TIMESTAMPTZ,
+                        body BYTEA,
+                        message_id TEXT,
+                        message_type TEXT,
+                        is_system_message BOOLEAN,
+                        status TEXT,
+                        time_sent TIMESTAMPTZ,
+                        receiving_endpoint_name TEXT,
+                        critical_time INTERVAL,
+                        processing_time INTERVAL,
+                        delivery_time INTERVAL,
+                        conversation_id TEXT,
+                        query tsvector GENERATED ALWAYS AS (
+                            setweight(to_tsvector('english', coalesce(headers::text, '')), 'A') ||
+                            setweight(to_tsvector('english', coalesce(body::text, '')), 'B')
+                        ) STORED
+                    );", connection))
+            {
+                cmd.ExecuteNonQuery();
+            }
+
+            // Create saga_snapshots table
+            using (var cmd = new NpgsqlCommand(@"
+                CREATE TABLE IF NOT EXISTS saga_snapshots (
+                    id TEXT 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
+                );", connection))
+            {
+                cmd.ExecuteNonQuery();
+            }
+
+            // Create known_endpoints table
+            using (var cmd = new NpgsqlCommand(@"
+                CREATE TABLE IF NOT EXISTS known_endpoints (
+                    id TEXT PRIMARY KEY,
+                    name TEXT,
+                    host_id UUID,
+                    host TEXT,
+                    last_seen TIMESTAMPTZ
+                );", connection))
+            {
+                cmd.ExecuteNonQuery();
+            }
+
+            return new PostgreSQLPersistence();
+        }
+    }
+}

src/ServiceControl.Audit.Persistence.PostgreSQL/ServiceControl.Audit.Persistence.PostgreSQL.csproj

@@ -0,0 +1,28 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <CopyLocalLockFileAssemblies>true</CopyLocalLockFileAssemblies>
+    <DisableTransitiveProjectReferences>true</DisableTransitiveProjectReferences>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <!-- Private=false & ExcludeAssets=runtime prevent repeatedly including binary dependencies of ServiceControl.Audit.Persistence and its depenencies in each persister directory -->
+    <ProjectReference Include="..\ServiceControl.Audit.Persistence\ServiceControl.Audit.Persistence.csproj" Private="false" ExcludeAssets="runtime" />
+    <ProjectReference Include="..\ServiceControl.Audit.Persistence.SagaAudit\ServiceControl.Audit.Persistence.SagaAudit.csproj" Private="false" ExcludeAssets="runtime" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="persistence.manifest" CopyToOutputDirectory="PreserveNewest" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <!-- Artifact does not include RavenDBServer directory. Primary instance is responsible for copying that to deploy directory. -->
+    <Artifact Include="$(OutputPath)" DestinationFolder="$(ArtifactsPath)Particular.ServiceControl.Audit\Persisters\PostgreSQL" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Npgsql" />
+  </ItemGroup>
+
+</Project>

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

@@ -0,0 +1,124 @@
+namespace ServiceControl.Audit.Persistence.PostgreSQL.UnitOfWork
+{
+    using System;
+    using System.Text.Json;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Npgsql;
+    using ServiceControl.Audit.Auditing;
+    using ServiceControl.Audit.Persistence.Monitoring;
+    using ServiceControl.Audit.Persistence.UnitOfWork;
+    using ServiceControl.SagaAudit;
+
+    public class PostgreSQLAuditIngestionUnitOfWork : IAuditIngestionUnitOfWork
+    {
+        readonly NpgsqlConnection connection;
+        readonly NpgsqlTransaction transaction;
+
+        public PostgreSQLAuditIngestionUnitOfWork(NpgsqlConnection connection, NpgsqlTransaction transaction)
+        {
+            this.connection = connection;
+            this.transaction = transaction;
+        }
+
+        public async ValueTask DisposeAsync()
+        {
+            await transaction.DisposeAsync().ConfigureAwait(false);
+            await connection.DisposeAsync().ConfigureAwait(false);
+        }
+
+        public async Task CompleteAsync(CancellationToken cancellationToken = default)
+        {
+            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
+        }
+
+        public async Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory<byte> body, CancellationToken cancellationToken = default)
+        {
+            object GetMetadata(string key) => processedMessage.MessageMetadata.TryGetValue(key, out var value) ? value ?? DBNull.Value : DBNull.Value;
+
+            // Insert ProcessedMessage into processed_messages table
+            var cmd = new NpgsqlCommand(@"
+                INSERT INTO processed_messages (
+                    unique_message_id, message_metadata, headers, processed_at, body,
+                    message_id, message_type, is_system_message, status, time_sent, receiving_endpoint_name,
+                    critical_time, processing_time, delivery_time, conversation_id
+                ) VALUES (
+                    @unique_message_id, @message_metadata, @headers, @processed_at, @body,
+                    @message_id, @message_type, @is_system_message, @status, @time_sent, @receiving_endpoint_name,
+                    @critical_time, @processing_time, @delivery_time, @conversation_id
+                )
+                ;", connection, transaction);
+
+            processedMessage.MessageMetadata["ContentLength"] = body.Length;
+            if (!body.IsEmpty)
+            {
+                cmd.Parameters.AddWithValue("body", body);
+            }
+            else
+            {
+                cmd.Parameters.AddWithValue("body", DBNull.Value);
+            }
+            cmd.Parameters.AddWithValue("unique_message_id", processedMessage.UniqueMessageId ?? (object)DBNull.Value);
+            cmd.Parameters.AddWithValue("message_metadata", JsonSerializer.Serialize(processedMessage.MessageMetadata));
+            cmd.Parameters.AddWithValue("headers", JsonSerializer.Serialize(processedMessage.Headers));
+            cmd.Parameters.AddWithValue("processed_at", processedMessage.ProcessedAt);
+            cmd.Parameters.AddWithValue("message_id", GetMetadata("MessageId"));
+            cmd.Parameters.AddWithValue("message_type", GetMetadata("MessageType"));
+            cmd.Parameters.AddWithValue("is_system_message", GetMetadata("IsSystemMessage"));
+            cmd.Parameters.AddWithValue("time_sent", GetMetadata("TimeSent"));
+            cmd.Parameters.AddWithValue("receiving_endpoint_name", GetMetadata("ReceivingEndpoint"));
+            cmd.Parameters.AddWithValue("critical_time", GetMetadata("CriticalTime"));
+            cmd.Parameters.AddWithValue("processing_time", GetMetadata("ProcessingTime"));
+            cmd.Parameters.AddWithValue("delivery_time", GetMetadata("DeliveryTime"));
+            cmd.Parameters.AddWithValue("conversation_id", GetMetadata("ConversationId"));
+
+            await cmd.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
+        }
+
+        public async Task RecordSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken = default)
+        {
+            // Insert SagaSnapshot into saga_snapshots table
+            var cmd = new NpgsqlCommand(@"
+                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;", connection, transaction);
+
+            cmd.Parameters.AddWithValue("id", sagaSnapshot.Id ?? (object)DBNull.Value);
+            cmd.Parameters.AddWithValue("saga_id", sagaSnapshot.SagaId);
+            cmd.Parameters.AddWithValue("saga_type", sagaSnapshot.SagaType ?? (object)DBNull.Value);
+            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 ?? (object)DBNull.Value);
+            cmd.Parameters.AddWithValue("initiating_message", JsonSerializer.Serialize(sagaSnapshot.InitiatingMessage));
+            cmd.Parameters.AddWithValue("outgoing_messages", JsonSerializer.Serialize(sagaSnapshot.OutgoingMessages));
+            cmd.Parameters.AddWithValue("endpoint", sagaSnapshot.Endpoint ?? (object)DBNull.Value);
+            cmd.Parameters.AddWithValue("processed_at", sagaSnapshot.ProcessedAt);
+
+            await cmd.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
+        }
+
+        public async Task RecordKnownEndpoint(KnownEndpoint knownEndpoint, CancellationToken cancellationToken = default)
+        {
+            // Insert KnownEndpoint into known_endpoints table
+            var cmd = new NpgsqlCommand(@"
+                INSERT INTO known_endpoints (
+                    id, name, host_id, host, last_seen
+                ) VALUES (
+                    @id, @name, @host_id, @host, @last_seen
+                )
+                ON CONFLICT (id) DO NOTHING;", connection, transaction);
+
+            cmd.Parameters.AddWithValue("id", knownEndpoint.Id ?? (object)DBNull.Value);
+            cmd.Parameters.AddWithValue("name", knownEndpoint.Name ?? (object)DBNull.Value);
+            cmd.Parameters.AddWithValue("host_id", knownEndpoint.HostId);
+            cmd.Parameters.AddWithValue("host", knownEndpoint.Host ?? (object)DBNull.Value);
+            cmd.Parameters.AddWithValue("last_seen", knownEndpoint.LastSeen);
+
+            await cmd.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
+        }
+    }
+}

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

@@ -0,0 +1,27 @@
+namespace ServiceControl.Audit.Persistence.PostgreSQL.UnitOfWork
+{
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Npgsql;
+    using ServiceControl.Audit.Persistence.UnitOfWork;
+    using ServiceControl.Audit.Persistence.PostgreSQL;
+
+    public class PostgreSQLAuditIngestionUnitOfWorkFactory : IAuditIngestionUnitOfWorkFactory
+    {
+        readonly PostgreSQLConnectionFactory connectionFactory;
+
+        public PostgreSQLAuditIngestionUnitOfWorkFactory(PostgreSQLConnectionFactory connectionFactory)
+        {
+            this.connectionFactory = connectionFactory;
+        }
+
+        public async ValueTask<IAuditIngestionUnitOfWork> StartNew(int batchSize, CancellationToken cancellationToken)
+        {
+            var connection = await connectionFactory.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
+            var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
+            return new PostgreSQLAuditIngestionUnitOfWork(connection, transaction);
+        }
+
+        public bool CanIngestMore() => true; // TODO: Implement logic based on storage state
+    }
+}

src/ServiceControl.Audit.Persistence.PostgreSQL/persistence.manifest

@@ -0,0 +1,7 @@
+{
+  "Name": "PostgreSQL",
+  "DisplayName": "PostgreSQL",
+  "Description": "PostgreSQL ServiceControl Audit persister",
+  "AssemblyName": "ServiceControl.Audit.Persistence.PostgreSQL",
+  "TypeName": "ServiceControl.Audit.Persistence.PostgreSQL.PostgreSQLPersistenceConfiguration, ServiceControl.Audit.Persistence.PostgreSQL"
+}