chore!: delete old connection events from audit log (#18735)

### Breaking change (changelog note):
>With new connection events appearing in the Connection Log, connection events older than 90 days will now be deleted from the Audit Log. If you require this legacy data, we recommend querying it from the REST API or making a backup of the database/these events before upgrading your Coder deployment. Please see the PR for details on what exactly will be deleted. 
Of note is that there are currently no plans to delete connection events from the Connection Log.


### Context

This is the fifth PR for moving connection events out of the audit log.

In previous PRs:
- **New** connection logs have been routed to the `connection_logs` table. They will *not* appear in the audit log.
- These new connection logs are served from the new `/api/v2/connectionlog` endpoint.

In this PR:
- We'll now clean existing connection events out of the audit log, if they are older than 90 days, We do this in batches of 1000, every 10 minutes.

The criteria for deletion is simple:
```
WHERE
(
     action = 'connect'
     OR action = 'disconnect'
     OR action = 'open'
     OR action = 'close'
)
AND "time" < @before_time::timestamp with time zone
```
where `@before_time` is currently configured to 90 days in the past.


Future PRs:
- Write documentation for the endpoint / feature

coderd/database/dbauthz/dbauthz.go

@@ -1552,6 +1552,16 @@ func (q *querier) DeleteOAuth2ProviderAppTokensByAppAndUserID(ctx context.Contex
 	return q.db.DeleteOAuth2ProviderAppTokensByAppAndUserID(ctx, arg)
 }
 
+func (q *querier) DeleteOldAuditLogConnectionEvents(ctx context.Context, threshold database.DeleteOldAuditLogConnectionEventsParams) error {
+	// `ResourceSystem` is deprecated, but it doesn't make sense to add
+	// `policy.ActionDelete` to `ResourceAuditLog`, since this is the one and
+	// only time we'll be deleting from the audit log.
+	if err := q.authorizeContext(ctx, policy.ActionDelete, rbac.ResourceSystem); err != nil {
+		return err
+	}
+	return q.db.DeleteOldAuditLogConnectionEvents(ctx, threshold)
+}
+
 func (q *querier) DeleteOldNotificationMessages(ctx context.Context) error {
 	if err := q.authorizeContext(ctx, policy.ActionDelete, rbac.ResourceNotificationMessage); err != nil {
 		return err

coderd/database/dbauthz/dbauthz_test.go

@@ -337,6 +337,10 @@ func (s *MethodTestSuite) TestAuditLogs() {
 		_ = dbgen.AuditLog(s.T(), db, database.AuditLog{})
 		check.Args(database.CountAuditLogsParams{}, emptyPreparedAuthorized{}).Asserts(rbac.ResourceAuditLog, policy.ActionRead)
 	}))
+	s.Run("DeleteOldAuditLogConnectionEvents", s.Subtest(func(db database.Store, check *expects) {
+		_ = dbgen.AuditLog(s.T(), db, database.AuditLog{})
+		check.Args(database.DeleteOldAuditLogConnectionEventsParams{}).Asserts(rbac.ResourceSystem, policy.ActionDelete)
+	}))
 }
 
 func (s *MethodTestSuite) TestConnectionLogs() {

coderd/database/dbmetrics/querymetrics.go

@@ -362,6 +362,13 @@ func (m queryMetricsStore) DeleteOAuth2ProviderAppTokensByAppAndUserID(ctx conte
 	return r0
 }
 
+func (m queryMetricsStore) DeleteOldAuditLogConnectionEvents(ctx context.Context, threshold database.DeleteOldAuditLogConnectionEventsParams) error {
+	start := time.Now()
+	r0 := m.s.DeleteOldAuditLogConnectionEvents(ctx, threshold)
+	m.queryLatencies.WithLabelValues("DeleteOldAuditLogConnectionEvents").Observe(time.Since(start).Seconds())
+	return r0
+}
+
 func (m queryMetricsStore) DeleteOldNotificationMessages(ctx context.Context) error {
 	start := time.Now()
 	r0 := m.s.DeleteOldNotificationMessages(ctx)

coderd/database/dbmock/dbmock.go

@@ -635,6 +635,20 @@ func (mr *MockStoreMockRecorder) DeleteOAuth2ProviderAppTokensByAppAndUserID(ctx
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteOAuth2ProviderAppTokensByAppAndUserID", reflect.TypeOf((*MockStore)(nil).DeleteOAuth2ProviderAppTokensByAppAndUserID), ctx, arg)
 }
 
+// DeleteOldAuditLogConnectionEvents mocks base method.
+func (m *MockStore) DeleteOldAuditLogConnectionEvents(ctx context.Context, arg database.DeleteOldAuditLogConnectionEventsParams) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DeleteOldAuditLogConnectionEvents", ctx, arg)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// DeleteOldAuditLogConnectionEvents indicates an expected call of DeleteOldAuditLogConnectionEvents.
+func (mr *MockStoreMockRecorder) DeleteOldAuditLogConnectionEvents(ctx, arg any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteOldAuditLogConnectionEvents", reflect.TypeOf((*MockStore)(nil).DeleteOldAuditLogConnectionEvents), ctx, arg)
+}
+
 // DeleteOldNotificationMessages mocks base method.
 func (m *MockStore) DeleteOldNotificationMessages(ctx context.Context) error {
 	m.ctrl.T.Helper()

coderd/database/dbpurge/dbpurge.go

@@ -18,6 +18,11 @@ import (
 const (
 	delay          = 10 * time.Minute
 	maxAgentLogAge = 7 * 24 * time.Hour
+	// Connection events are now inserted into the `connection_logs` table.
+	// We'll slowly remove old connection events from the `audit_logs` table,
+	// but we won't touch the `connection_logs` table.
+	maxAuditLogConnectionEventAge    = 90 * 24 * time.Hour // 90 days
+	auditLogConnectionEventBatchSize = 1000
 )
 
 // New creates a new periodically purging database instance.
@@ -63,6 +68,14 @@ func New(ctx context.Context, logger slog.Logger, db database.Store, clk quartz.
 				return xerrors.Errorf("failed to delete old notification messages: %w", err)
 			}
 
+			deleteOldAuditLogConnectionEventsBefore := start.Add(-maxAuditLogConnectionEventAge)
+			if err := tx.DeleteOldAuditLogConnectionEvents(ctx, database.DeleteOldAuditLogConnectionEventsParams{
+				BeforeTime: deleteOldAuditLogConnectionEventsBefore,
+				LimitCount: auditLogConnectionEventBatchSize,
+			}); err != nil {
+				return xerrors.Errorf("failed to delete old audit log connection events: %w", err)
+			}
+
 			logger.Debug(ctx, "purged old database entries", slog.F("duration", clk.Since(start)))
 
 			return nil

coderd/database/dbpurge/dbpurge_test.go

@@ -490,3 +490,148 @@ func containsProvisionerDaemon(daemons []database.ProvisionerDaemon, name string
 		return d.Name == name
 	})
 }
+
+//nolint:paralleltest // It uses LockIDDBPurge.
+func TestDeleteOldAuditLogConnectionEvents(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
+	defer cancel()
+
+	clk := quartz.NewMock(t)
+	now := dbtime.Now()
+	afterThreshold := now.Add(-91 * 24 * time.Hour)       // 91 days ago (older than 90 day threshold)
+	beforeThreshold := now.Add(-30 * 24 * time.Hour)      // 30 days ago (newer than 90 day threshold)
+	closeBeforeThreshold := now.Add(-89 * 24 * time.Hour) // 89 days ago
+	clk.Set(now).MustWait(ctx)
+
+	db, _ := dbtestutil.NewDB(t, dbtestutil.WithDumpOnFailure())
+	logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+	user := dbgen.User(t, db, database.User{})
+	org := dbgen.Organization(t, db, database.Organization{})
+
+	oldConnectLog := dbgen.AuditLog(t, db, database.AuditLog{
+		UserID:         user.ID,
+		OrganizationID: org.ID,
+		Time:           afterThreshold,
+		Action:         database.AuditActionConnect,
+		ResourceType:   database.ResourceTypeWorkspace,
+	})
+
+	oldDisconnectLog := dbgen.AuditLog(t, db, database.AuditLog{
+		UserID:         user.ID,
+		OrganizationID: org.ID,
+		Time:           afterThreshold,
+		Action:         database.AuditActionDisconnect,
+		ResourceType:   database.ResourceTypeWorkspace,
+	})
+
+	oldOpenLog := dbgen.AuditLog(t, db, database.AuditLog{
+		UserID:         user.ID,
+		OrganizationID: org.ID,
+		Time:           afterThreshold,
+		Action:         database.AuditActionOpen,
+		ResourceType:   database.ResourceTypeWorkspace,
+	})
+
+	oldCloseLog := dbgen.AuditLog(t, db, database.AuditLog{
+		UserID:         user.ID,
+		OrganizationID: org.ID,
+		Time:           afterThreshold,
+		Action:         database.AuditActionClose,
+		ResourceType:   database.ResourceTypeWorkspace,
+	})
+
+	recentConnectLog := dbgen.AuditLog(t, db, database.AuditLog{
+		UserID:         user.ID,
+		OrganizationID: org.ID,
+		Time:           beforeThreshold,
+		Action:         database.AuditActionConnect,
+		ResourceType:   database.ResourceTypeWorkspace,
+	})
+
+	oldNonConnectionLog := dbgen.AuditLog(t, db, database.AuditLog{
+		UserID:         user.ID,
+		OrganizationID: org.ID,
+		Time:           afterThreshold,
+		Action:         database.AuditActionCreate,
+		ResourceType:   database.ResourceTypeWorkspace,
+	})
+
+	nearThresholdConnectLog := dbgen.AuditLog(t, db, database.AuditLog{
+		UserID:         user.ID,
+		OrganizationID: org.ID,
+		Time:           closeBeforeThreshold,
+		Action:         database.AuditActionConnect,
+		ResourceType:   database.ResourceTypeWorkspace,
+	})
+
+	// Run the purge
+	done := awaitDoTick(ctx, t, clk)
+	closer := dbpurge.New(ctx, logger, db, clk)
+	defer closer.Close()
+	// Wait for tick
+	testutil.TryReceive(ctx, t, done)
+
+	// Verify results by querying all audit logs
+	logs, err := db.GetAuditLogsOffset(ctx, database.GetAuditLogsOffsetParams{})
+	require.NoError(t, err)
+
+	// Extract log IDs for comparison
+	logIDs := make([]uuid.UUID, len(logs))
+	for i, log := range logs {
+		logIDs[i] = log.AuditLog.ID
+	}
+
+	require.NotContains(t, logIDs, oldConnectLog.ID, "old connect log should be deleted")
+	require.NotContains(t, logIDs, oldDisconnectLog.ID, "old disconnect log should be deleted")
+	require.NotContains(t, logIDs, oldOpenLog.ID, "old open log should be deleted")
+	require.NotContains(t, logIDs, oldCloseLog.ID, "old close log should be deleted")
+	require.Contains(t, logIDs, recentConnectLog.ID, "recent connect log should be kept")
+	require.Contains(t, logIDs, nearThresholdConnectLog.ID, "near threshold connect log should be kept")
+	require.Contains(t, logIDs, oldNonConnectionLog.ID, "old non-connection log should be kept")
+}
+
+func TestDeleteOldAuditLogConnectionEventsLimit(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
+	defer cancel()
+
+	db, _ := dbtestutil.NewDB(t, dbtestutil.WithDumpOnFailure())
+	user := dbgen.User(t, db, database.User{})
+	org := dbgen.Organization(t, db, database.Organization{})
+
+	now := dbtime.Now()
+	threshold := now.Add(-90 * 24 * time.Hour)
+
+	for i := 0; i < 5; i++ {
+		dbgen.AuditLog(t, db, database.AuditLog{
+			UserID:         user.ID,
+			OrganizationID: org.ID,
+			Time:           threshold.Add(-time.Duration(i+1) * time.Hour),
+			Action:         database.AuditActionConnect,
+			ResourceType:   database.ResourceTypeWorkspace,
+		})
+	}
+
+	err := db.DeleteOldAuditLogConnectionEvents(ctx, database.DeleteOldAuditLogConnectionEventsParams{
+		BeforeTime: threshold,
+		LimitCount: 1,
+	})
+	require.NoError(t, err)
+
+	logs, err := db.GetAuditLogsOffset(ctx, database.GetAuditLogsOffsetParams{})
+	require.NoError(t, err)
+
+	require.Len(t, logs, 4)
+
+	err = db.DeleteOldAuditLogConnectionEvents(ctx, database.DeleteOldAuditLogConnectionEventsParams{
+		BeforeTime: threshold,
+		LimitCount: 100,
+	})
+	require.NoError(t, err)
+
+	logs, err = db.GetAuditLogsOffset(ctx, database.GetAuditLogsOffsetParams{})
+	require.NoError(t, err)
+
+	require.Len(t, logs, 0)
+}

coderd/database/querier.go

@@ -96,6 +96,7 @@ type sqlcQuerier interface {
 	DeleteOAuth2ProviderAppCodesByAppAndUserID(ctx context.Context, arg DeleteOAuth2ProviderAppCodesByAppAndUserIDParams) error
 	DeleteOAuth2ProviderAppSecretByID(ctx context.Context, id uuid.UUID) error
 	DeleteOAuth2ProviderAppTokensByAppAndUserID(ctx context.Context, arg DeleteOAuth2ProviderAppTokensByAppAndUserIDParams) error
+	DeleteOldAuditLogConnectionEvents(ctx context.Context, arg DeleteOldAuditLogConnectionEventsParams) error
 	// Delete all notification messages which have not been updated for over a week.
 	DeleteOldNotificationMessages(ctx context.Context) error
 	// Delete provisioner daemons that have been created at least a week ago

coderd/database/queries.sql.go

@@ -566,6 +566,33 @@ func (q *sqlQuerier) CountAuditLogs(ctx context.Context, arg CountAuditLogsParam
 	return count, err
 }
 
+const deleteOldAuditLogConnectionEvents = `-- name: DeleteOldAuditLogConnectionEvents :exec
+DELETE FROM audit_logs
+WHERE id IN (
+    SELECT id FROM audit_logs
+    WHERE
+        (
+            action = 'connect'
+            OR action = 'disconnect'
+            OR action = 'open'
+            OR action = 'close'
+        )
+        AND "time" < $1::timestamp with time zone
+    ORDER BY "time" ASC
+    LIMIT $2
+)
+`
+
+type DeleteOldAuditLogConnectionEventsParams struct {
+	BeforeTime time.Time `db:"before_time" json:"before_time"`
+	LimitCount int32     `db:"limit_count" json:"limit_count"`
+}
+
+func (q *sqlQuerier) DeleteOldAuditLogConnectionEvents(ctx context.Context, arg DeleteOldAuditLogConnectionEventsParams) error {
+	_, err := q.db.ExecContext(ctx, deleteOldAuditLogConnectionEvents, arg.BeforeTime, arg.LimitCount)
+	return err
+}
+
 const getAuditLogsOffset = `-- name: GetAuditLogsOffset :many
 SELECT audit_logs.id, audit_logs.time, audit_logs.user_id, audit_logs.organization_id, audit_logs.ip, audit_logs.user_agent, audit_logs.resource_type, audit_logs.resource_id, audit_logs.resource_target, audit_logs.action, audit_logs.diff, audit_logs.status_code, audit_logs.additional_fields, audit_logs.request_id, audit_logs.resource_icon,
 	-- sqlc.embed(users) would be nice but it does not seem to play well with

coderd/database/queries/auditlogs.sql

@@ -237,3 +237,19 @@ WHERE
 	-- Authorize Filter clause will be injected below in CountAuthorizedAuditLogs
 	-- @authorize_filter
 ;
+
+-- name: DeleteOldAuditLogConnectionEvents :exec
+DELETE FROM audit_logs
+WHERE id IN (
+    SELECT id FROM audit_logs
+    WHERE
+        (
+            action = 'connect'
+            OR action = 'disconnect'
+            OR action = 'open'
+            OR action = 'close'
+        )
+        AND "time" < @before_time::timestamp with time zone
+    ORDER BY "time" ASC
+    LIMIT @limit_count
+);