From f1ae3d33a9d37683aaa873a7232ae1a2eb67a2e1 Mon Sep 17 00:00:00 2001 From: Fabio Bonelli Date: Thu, 3 Sep 2026 07:44:55 +0200 Subject: [PATCH] feat: purge the events older than the retention window The audit trail grew for the life of the deployment, nothing ever deleted an event. A daily purge now drops the ones older than EVENT_RETENTION_DAYS, a year by default. --- README.md | 7 ++ internal/common/env.go | 5 ++ internal/database/events.go | 67 +++++++++++++++++ internal/database/events_test.go | 121 +++++++++++++++++++++++++++++++ main.go | 20 ++++- main_test.go | 10 ++- 6 files changed, 224 insertions(+), 6 deletions(-) create mode 100644 internal/database/events.go create mode 100644 internal/database/events_test.go diff --git a/README.md b/README.md index fa47f99..d43b3ba 100644 --- a/README.md +++ b/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 diff --git a/internal/common/env.go b/internal/common/env.go index bf68451..377b7f7 100644 --- a/internal/common/env.go +++ b/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 { diff --git a/internal/database/events.go b/internal/database/events.go new file mode 100644 index 0000000..3efa068 --- /dev/null +++ b/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) + } +} diff --git a/internal/database/events_test.go b/internal/database/events_test.go new file mode 100644 index 0000000..444fd48 --- /dev/null +++ b/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) +} diff --git a/main.go b/main.go index 95fdbb0..27290e6 100644 --- a/main.go +++ b/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 diff --git a/main_test.go b/main_test.go index bef205e..923c229 100644 --- a/main_test.go +++ b/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)