feat: purge the events older than the retention window

README.md

@@ -101,6 +101,13 @@ You can configure the API with environment variables:
   `WEBHOOK_DEBOUNCE_MS` is 0.
   Default: `10000`.
 
+* `EVENT_RETENTION_DAYS` (optional): how many days the audit trail is
+  kept. Every create, update and delete of a catalog entity is an
+  event recording the entity, the kind of change, who made it and
+  when, readable at `GET /v1/events`. Older events are deleted at
+  startup and once a day. Set to `0` to keep them forever.
+  Default: `365`.
+
 ## Contributing
 
 This project exists also thanks to your contributions! Here is a list of people

internal/common/env.go

@@ -30,6 +30,11 @@ type Environment struct {
 	// webhook can be deferred by repeated resets of the debounce timer.
 	// Set to 0 to disable the cap. Ignored when WebhookDebounceMS is 0.
 	WebhookDebounceMaxMS int `env:"WEBHOOK_DEBOUNCE_MAX_MS" envDefault:"10000"`
+
+	// EventRetentionDays is how many days the audit trail is kept: a
+	// background purge deletes the older events. Set to 0 to keep every
+	// event forever.
+	EventRetentionDays int `env:"EVENT_RETENTION_DAYS" envDefault:"365"`
 }
 
 func (k *Base64Key) UnmarshalText(text []byte) error {

internal/database/events.go

@@ -0,0 +1,67 @@
+package database
+
+import (
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/publiccodeyml/open-catalog-api/internal/models"
+	"gorm.io/gorm"
+)
+
+// PurgeEvents deletes the events created more than olderThan ago and
+// returns how many rows went away. The delete is Unscoped: an entry past
+// the retention window leaves the audit trail for good, a soft delete
+// would keep the row in the table.
+func PurgeEvents(gormdb *gorm.DB, olderThan time.Duration) (int64, error) {
+	cutoff := time.Now().Add(-olderThan)
+
+	result := gormdb.Unscoped().Where("created_at < ?", cutoff).Delete(&models.Event{})
+	if result.Error != nil {
+		return 0, fmt.Errorf("can't purge the events older than %s: %w", olderThan, result.Error)
+	}
+
+	return result.RowsAffected, nil
+}
+
+// StartEventPurge purges the events once and then every interval, until
+// the returned function is called. A zero olderThan keeps every event
+// and makes both the purge and the returned function a no op.
+func StartEventPurge(gormdb *gorm.DB, olderThan, interval time.Duration) func() {
+	if olderThan <= 0 || interval <= 0 {
+		return func() {}
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+
+		for {
+			purgeEvents(gormdb, olderThan)
+
+			select {
+			case <-ticker.C:
+			case <-done:
+				return
+			}
+		}
+	}()
+
+	return sync.OnceFunc(func() { close(done) })
+}
+
+func purgeEvents(gormdb *gorm.DB, olderThan time.Duration) {
+	purged, err := PurgeEvents(gormdb, olderThan)
+	if err != nil {
+		log.Println(err)
+
+		return
+	}
+
+	if purged > 0 {
+		log.Printf("purged %d events older than %s", purged, olderThan)
+	}
+}

internal/database/events_test.go

@@ -0,0 +1,121 @@
+package database
+
+import (
+	"testing"
+	"time"
+
+	"github.com/gofiber/fiber/v2/utils"
+	"github.com/publiccodeyml/open-catalog-api/internal/common"
+	"github.com/publiccodeyml/open-catalog-api/internal/models"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+)
+
+const day = 24 * time.Hour
+
+func newEventDatabase(t *testing.T) *gorm.DB {
+	t.Helper()
+
+	gormdb, err := NewDatabase("file:" + t.Name() + "?mode=memory&cache=shared")
+	require.NoError(t, err)
+
+	return gormdb
+}
+
+func createEvent(t *testing.T, gormdb *gorm.DB, age time.Duration) string {
+	t.Helper()
+
+	event := models.Event{
+		ID:         utils.UUIDv4(),
+		Type:       common.EventTypeCreate,
+		EntityType: "software",
+		EntityID:   utils.UUIDv4(),
+		CreatedAt:  time.Now().Add(-age),
+	}
+
+	require.NoError(t, gormdb.Create(&event).Error)
+
+	return event.ID
+}
+
+// countEvents counts unscoped, so a soft deleted row still counts and a
+// purge passes only when the row is gone.
+func countEvents(t *testing.T, gormdb *gorm.DB, where string, args ...any) int64 {
+	t.Helper()
+
+	var count int64
+
+	query := gormdb.Unscoped().Model(&models.Event{})
+	if where != "" {
+		query = query.Where(where, args...)
+	}
+
+	assert.NoError(t, query.Count(&count).Error)
+
+	return count
+}
+
+func eventExists(t *testing.T, gormdb *gorm.DB, id string) bool {
+	t.Helper()
+
+	return countEvents(t, gormdb, "id = ?", id) > 0
+}
+
+func TestPurgeEvents(t *testing.T) {
+	gormdb := newEventDatabase(t)
+
+	recent := createEvent(t, gormdb, day)
+	within := createEvent(t, gormdb, 20*day)
+	expired := createEvent(t, gormdb, 40*day)
+
+	purged, err := PurgeEvents(gormdb, 30*day)
+	require.NoError(t, err)
+
+	assert.Equal(t, int64(1), purged)
+	assert.Equal(t, int64(2), countEvents(t, gormdb, ""))
+	assert.True(t, eventExists(t, gormdb, recent))
+	assert.True(t, eventExists(t, gormdb, within))
+	assert.False(t, eventExists(t, gormdb, expired))
+}
+
+func TestPurgeEventsDeletesTheSoftDeletedRows(t *testing.T) {
+	gormdb := newEventDatabase(t)
+
+	expired := createEvent(t, gormdb, 40*day)
+	require.NoError(t, gormdb.Delete(&models.Event{}, "id = ?", expired).Error)
+	require.Equal(t, int64(1), countEvents(t, gormdb, ""))
+
+	purged, err := PurgeEvents(gormdb, 30*day)
+	require.NoError(t, err)
+
+	assert.Equal(t, int64(1), purged)
+	assert.Equal(t, int64(0), countEvents(t, gormdb, ""))
+}
+
+func TestStartEventPurgeDeletesOnStartup(t *testing.T) {
+	gormdb := newEventDatabase(t)
+
+	createEvent(t, gormdb, 40*day)
+	createEvent(t, gormdb, day)
+
+	stop := StartEventPurge(gormdb, 30*day, time.Hour)
+	t.Cleanup(stop)
+
+	assert.Eventually(t, func() bool {
+		return countEvents(t, gormdb, "") == 1
+	}, time.Second, 10*time.Millisecond)
+}
+
+func TestStartEventPurgeKeepsEveryEventWithoutRetention(t *testing.T) {
+	gormdb := newEventDatabase(t)
+
+	createEvent(t, gormdb, 40*day)
+
+	stop := StartEventPurge(gormdb, 0, time.Hour)
+	t.Cleanup(stop)
+
+	assert.Never(t, func() bool {
+		return countEvents(t, gormdb, "") == 0
+	}, 200*time.Millisecond, 10*time.Millisecond)
+}

main.go

@@ -29,12 +29,16 @@ import (
 	"gorm.io/gorm"
 )
 
+// The retention window is configured in days, so a shorter cadence
+// would find nothing new to delete.
+const eventPurgeInterval = 24 * time.Hour
+
 func main() {
 	rootCmd := &cobra.Command{
 		Use:          "open-catalog-api",
 		SilenceUsage: true,
 		RunE: func(_ *cobra.Command, _ []string) error {
-			app, debouncer := Setup()
+			app, debouncer, stopEventPurge := Setup()
 
 			sigCh := make(chan os.Signal, 1)
 			signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
@@ -53,6 +57,7 @@ func main() {
 			// in the debouncer's pending window is dispatched before the
 			// process exits.
 			debouncer.Drain()
+			stopEventPurge()
 
 			if err != nil {
 				return fmt.Errorf("listen: %w", err)
@@ -69,7 +74,7 @@ func main() {
 	}
 }
 
-func Setup() (*fiber.App, *webhooks.Debouncer) {
+func Setup() (*fiber.App, *webhooks.Debouncer, func()) {
 	if err := env.Parse(&common.EnvironmentConfig); err != nil {
 		panic(err)
 	}
@@ -79,6 +84,15 @@ func Setup() (*fiber.App, *webhooks.Debouncer) {
 		panic(err)
 	}
 
+	// The fixtures carry events far older than any retention window, so
+	// under the test harness the purge would delete them.
+	eventRetention := time.Duration(common.EnvironmentConfig.EventRetentionDays) * 24 * time.Hour
+	if common.EnvironmentConfig.IsTest() {
+		eventRetention = 0
+	}
+
+	stopEventPurge := database.StartEventPurge(gormDB, eventRetention, eventPurgeInterval)
+
 	// Setup a goroutine acting as a worker for events sent to the
 	// EventChan channel.
 	//
@@ -153,7 +167,7 @@ func Setup() (*fiber.App, *webhooks.Debouncer) {
 
 	setupHandlers(app, gormDB)
 
-	return app, debouncer
+	return app, debouncer, stopEventPurge
 }
 
 // requiresAuth reports whether a request must carry a valid token. Reads

main_test.go

@@ -95,7 +95,7 @@ func init() {
 	}
 
 	// Setup the app as it is done in the main function
-	app, _ = Setup()
+	app, _, _ = Setup()
 }
 
 func TestMain(m *testing.M) {
@@ -362,14 +362,18 @@ func TestRateLimiterBuckets(t *testing.T) {
 	t.Setenv("ENVIRONMENT", "production")
 	t.Setenv("MAX_REQUESTS", strconv.Itoa(maxRequests))
 
+	// Outside the test environment the purge would delete the fixture
+	// events while the other tests are loading them.
+	t.Setenv("EVENT_RETENTION_DAYS", "0")
+
 	// An empty variable leaves the already parsed value alone, so the app
 	// trusting nobody has to be built before the one with a trusted proxy.
 	t.Setenv("TRUSTED_PROXIES", "")
-	directApp, _ := Setup()
+	directApp, _, _ := Setup()
 
 	// 0.0.0.0 is the peer address of a request made through app.Test().
 	t.Setenv("TRUSTED_PROXIES", "0.0.0.0/32")
-	proxiedApp, _ := Setup()
+	proxiedApp, _, _ := Setup()
 
 	t.Run("a client varying X-Forwarded-For stays in one bucket", func(t *testing.T) {
 		codes := make([]int, 0, maxRequests+1)