Updated retention cleanup to include saga snapshots and known endpoints

Author: johnsimons

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

@@ -162,7 +162,8 @@ CREATE TABLE IF NOT EXISTS saga_snapshots (
                 id UUID PRIMARY KEY,
                 saga_id UUID,
                 saga_type TEXT,
-                changes JSONB
+                changes JSONB,
+                updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
             );", connection))
         {
             await cmd.ExecuteNonQueryAsync(cancellationToken);
@@ -184,17 +185,38 @@ CREATE TABLE IF NOT EXISTS known_endpoints (
                 name TEXT,
                 host_id UUID,
                 host TEXT,
-                last_seen TIMESTAMPTZ
+                last_seen TIMESTAMPTZ,
+                updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
             );", connection))
         {
             await cmd.ExecuteNonQueryAsync(cancellationToken);
         }
+
+        // Create trigger to auto-update updated_at for saga_snapshots
+        await using (var cmd = new NpgsqlCommand(@"
+            CREATE OR REPLACE FUNCTION update_updated_at_column() RETURNS trigger AS $$
+            BEGIN
+            NEW.updated_at = now();
+            RETURN NEW;
+            END
+            $$ LANGUAGE plpgsql;
+
+            DROP TRIGGER IF EXISTS saga_snapshots_updated_at_trigger ON saga_snapshots;
+            CREATE TRIGGER saga_snapshots_updated_at_trigger
+            BEFORE UPDATE ON saga_snapshots
+            FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
+
+            DROP TRIGGER IF EXISTS known_endpoints_updated_at_trigger ON known_endpoints;
+            CREATE TRIGGER known_endpoints_updated_at_trigger
+            BEFORE UPDATE ON known_endpoints
+            FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();", connection))
+        {
+            await cmd.ExecuteNonQueryAsync(cancellationToken);
+        }
     }
 
     public Task StopAsync(CancellationToken cancellationToken)
     {
         return Task.CompletedTask;
     }
-}
-
-
+}

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

@@ -43,14 +43,27 @@ async Task CleanupOldMessagesAsync(CancellationToken cancellationToken)
         await using var conn = await connectionFactory.OpenConnection(cancellationToken);
 
         var cutoffDate = DateTime.UtcNow - config.AuditRetentionPeriod;
+
+        // Cleanup processed messages
+        await CleanupTable("processed_messages", "created_at", cutoffDate, conn, cancellationToken);
+
+        // Cleanup saga snapshots
+        await CleanupTable("saga_snapshots", "updated_at", cutoffDate, conn, cancellationToken);
+
+        // Cleanup known endpoints
+        await CleanupTable("known_endpoints", "updated_at", cutoffDate, conn, cancellationToken);
+    }
+
+    async Task CleanupTable(string tableName, string dateColumn, DateTime cutoffDate, NpgsqlConnection conn, CancellationToken cancellationToken)
+    {
         var totalDeleted = 0;
 
         while (!cancellationToken.IsCancellationRequested)
         {
             // Delete in batches
-            var sql = @"
-                DELETE FROM processed_messages
-                WHERE created_at < @cutoff
+            var sql = $@"
+                DELETE FROM {tableName}
+                WHERE {dateColumn} < @cutoff
                 LIMIT 1000;";
 
             await using var cmd = new NpgsqlCommand(sql, conn);
@@ -69,7 +82,7 @@ WHERE created_at < @cutoff
 
         if (totalDeleted > 0)
         {
-            logger.LogInformation("Deleted {Count} old messages older than {Cutoff}", totalDeleted, cutoffDate);
+            logger.LogInformation("Deleted {Count} old records from {Table} older than {Cutoff}", totalDeleted, tableName, cutoffDate);
         }
     }
 }