increase error handling and mutex encapsulation (#12)

* increase error handling and mutex encapsulation

* undo method rename for now

* set cant return error

---------

Co-authored-by: Chris Clayton <chris.clayton@contino.io>

README.md

@@ -16,6 +16,8 @@ This project is in active use by [telegram-bot.app](https://telegram-bot.app), a
   - [Speed](#speed)
     - [Caching](#caching)
     - [Read-only endpoint](#read-only-endpoint)
+  - [Maintenance](#maintenance)
+    - [Hasura event cleaner](#hasura-event-cleaner)
   - [Security](#security)
     - [Role-based rate limiting](#role-based-rate-limiting)
     - [Read-only mode](#read-only-mode)
@@ -101,6 +103,7 @@ In this case, both proxy and websockets will be available under the `/v1/graphql
 | security   | Blocking mutations in read-only mode                                  |
 | security   | Allow access only to listed URLs                                      |
 | security   | Ban / unban specific user from accessing the application              |
+| maintenance | Hasura event cleaner                                                 |
 
 
 ### Configuration
@@ -138,6 +141,9 @@ You can still use the non-prefixed environment variables in the spirit of the ba
 | `PROXIED_CLIENT_TIMEOUT` | The timeout for the proxied client in seconds     | `120`                      |
 | `PURGE_METRICS_ON_CRAWL` | Purge metrics on each /metrics crawl    | `false`                      |
 | `PURGE_METRICS_ON_TIMER` | Purge metrics every x seconds. `0` - disabled | `0`                      |
+| `HASURA_EVENT_CLEANER`   | Enable the hasura event cleaner          | `false`                    |
+| `HASURA_EVENT_CLEANER_OLDER_THAN` | The interval for the hasura event cleaner (in days) | `1`                  |
+| `HASURA_EVENT_METADATA_DB` | URL to the hasura metadata database    | `postgresql://localhost:5432/hasura` |
 
 ### Speed
 
@@ -171,6 +177,20 @@ You can now specify the read-only GraphQL endpoint by setting the `HOST_GRAPHQL_
 
 You can check out the [example of combined deployment with RW and read-only hasura](static/kubernetes-single-deployment-with-ro.yaml).
 
+### Maintenance
+
+#### Hasura event cleaner
+
+When enabled via `HASURA_EVENT_CLEANER=true` - proxy needs to have a direct access to the database to execute simple delete queries on schedule. You can specify number of days the logs should be kept for using `HASURA_EVENT_CLEANER_OLDER_THAN`, for example `HASURA_EVENT_CLEANER_OLDER_THAN=14` will keep 14 days of event execution logs. Ticker managing the cleaner routine will be executed every hour.
+
+Following tables are being cleaned:
+- `hdb_catalog.event_invocation_logs`
+- `hdb_catalog.event_log`
+- `hdb_catalog.hdb_action_log`
+- `hdb_catalog.hdb_cron_event_invocation_logs`
+- `hdb_catalog.hdb_scheduled_event_invocation_logs`
+
+
 ### Security
 
 #### Role-based rate limiting

cache/memory/cache.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"io"
+	"log"
 	"sync"
 	"time"
 )
@@ -14,11 +15,11 @@ type CacheEntry struct {
 }
 
 type Cache struct {
-	entries        sync.Map
-	globalTTL      time.Duration
 	compressPool   sync.Pool
 	decompressPool sync.Pool
-	sync.RWMutex   // Reintroduced to provide lock methods
+	entries        sync.Map
+	globalTTL      time.Duration
+	mu             sync.RWMutex // Added sync.RWMutex field for locking
 }
 
 func New(globalTTL time.Duration) *Cache {
@@ -26,13 +27,11 @@ func New(globalTTL time.Duration) *Cache {
 		globalTTL: globalTTL,
 		compressPool: sync.Pool{
 			New: func() interface{} {
-				w := gzip.NewWriter(nil)
-				return w
+				return gzip.NewWriter(nil)
 			},
 		},
 		decompressPool: sync.Pool{
 			New: func() interface{} {
-				// Ensure that new is returning a new reader initialized with an empty byte buffer
 				r, _ := gzip.NewReader(bytes.NewReader([]byte{}))
 				return r
 			},
@@ -53,13 +52,14 @@ func (c *Cache) cleanupRoutine(globalTTL time.Duration) {
 }
 
 func (c *Cache) Set(key string, value []byte, ttl time.Duration) {
-	c.Lock() // use the lock
-	defer c.Unlock()
+	c.lock()
+	defer c.unlock()
 
 	expiresAt := time.Now().Add(ttl)
 
 	compressedValue, err := c.compress(value)
 	if err != nil {
+		log.Printf("Error compressing value for key %s: %v", key, err)
 		return
 	}
 
@@ -71,8 +71,8 @@ func (c *Cache) Set(key string, value []byte, ttl time.Duration) {
 }
 
 func (c *Cache) Get(key string) ([]byte, bool) {
-	c.RLock() // use the read lock
-	defer c.RUnlock()
+	c.rlock()
+	defer c.runlock()
 
 	entry, ok := c.entries.Load(key)
 	if !ok || entry.(CacheEntry).ExpiresAt.Before(time.Now()) {
@@ -81,30 +81,33 @@ func (c *Cache) Get(key string) ([]byte, bool) {
 	compressedValue := entry.(CacheEntry).Value
 	value, err := c.decompress(compressedValue)
 	if err != nil {
+		log.Printf("Error decompressing value for key %s: %v", key, err)
 		return nil, false
 	}
 	return value, true
 }
 
 func (c *Cache) Delete(key string) {
-	c.Lock()
-	defer c.Unlock()
-
-	_, ok := c.entries.Load(key)
-	if !ok {
-		return
-	}
+	c.lock()
+	defer c.unlock()
 
 	c.entries.Delete(key)
 }
 
 func (c *Cache) Clear() {
-	c.entries = sync.Map{}
+	c.lock()
+	defer c.unlock()
+
+	c.entries.Range(func(key, value interface{}) bool {
+		c.entries.Delete(key)
+		return true
+	})
 }
 
 func (c *Cache) CountQueries() int {
-	c.RLock()
-	defer c.RUnlock()
+	c.rlock()
+	defer c.runlock()
+
 	var count int
 	c.entries.Range(func(_, _ interface{}) bool {
 		count++
@@ -131,27 +134,24 @@ func (c *Cache) compress(data []byte) ([]byte, error) {
 func (c *Cache) decompress(data []byte) ([]byte, error) {
 	r, ok := c.decompressPool.Get().(*gzip.Reader)
 	if !ok || r == nil {
-		// If r is nil or type assertion fails, create a new gzip.Reader
 		var err error
 		r, err = gzip.NewReader(bytes.NewReader(data))
 		if err != nil {
-			return nil, err // Handle the error if gzip.NewReader fails
+			return nil, err
 		}
 	} else {
-		// Reset the existing reader with new data
 		if err := r.Reset(bytes.NewReader(data)); err != nil {
-			return nil, err // Handle the error if Reset fails
+			return nil, err
 		}
 	}
-	defer r.Close()
+	defer func() {
+		r.Close()
+		c.decompressPool.Put(r)
+	}()
 
-	// Ensure the reader is returned to the pool
-	defer c.decompressPool.Put(r)
-
-	// Read all the data from the reader
 	decompressedData, err := io.ReadAll(r)
 	if err != nil {
-		return nil, err // Handle the error if reading fails
+		return nil, err
 	}
 	return decompressedData, nil
 }
@@ -166,3 +166,21 @@ func (c *Cache) CleanExpiredEntries() {
 		return true
 	})
 }
+
+// Private methods to handle locking
+
+func (c *Cache) lock() {
+	c.mu.Lock()
+}
+
+func (c *Cache) unlock() {
+	c.mu.Unlock()
+}
+
+func (c *Cache) rlock() {
+	c.mu.RLock()
+}
+
+func (c *Cache) runlock() {
+	c.mu.RUnlock()
+}

events.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+func enableHasuraEventCleaner() {
+	if cfg.HasuraEventCleaner.Enable {
+		if cfg.HasuraEventCleaner.EventMetadataDb == "" {
+			cfg.Logger.Warning("Event metadata db URL not specified, event cleaner not active", nil)
+			return
+		}
+
+		ticker := time.NewTicker(1 * time.Hour)
+		defer ticker.Stop()
+		cfg.Logger.Info("Event cleaner enabled", map[string]interface{}{"interval_in_days": cfg.HasuraEventCleaner.ClearOlderThan})
+
+		time.Sleep(60 * time.Second) // wait for everything to start and settle down
+		cfg.Logger.Info("Initial cleanup of old events", nil)
+		cleanEvents()
+
+		for {
+			select {
+			case <-ticker.C:
+				cfg.Logger.Info("Cleaning up old events", nil)
+				cleanEvents()
+			}
+		}
+	}
+}
+
+func cleanEvents() {
+	conn, err := pgx.Connect(context.Background(), cfg.HasuraEventCleaner.EventMetadataDb)
+	if err != nil {
+		cfg.Logger.Error("Failed to connect to event metadata db", map[string]interface{}{"error": err})
+		return
+	}
+	defer conn.Close(context.Background())
+
+	delQueries := []string{
+		fmt.Sprintf("DELETE FROM hdb_catalog.event_invocation_logs WHERE created_at < now() - interval '%d days';", cfg.HasuraEventCleaner.ClearOlderThan),
+		fmt.Sprintf("DELETE FROM hdb_catalog.event_log WHERE created_at < now() - interval '%d days';", cfg.HasuraEventCleaner.ClearOlderThan),
+		fmt.Sprintf("DELETE FROM hdb_catalog.hdb_action_log WHERE created_at < NOW() - INTERVAL '%d days';", cfg.HasuraEventCleaner.ClearOlderThan),
+		fmt.Sprintf("DELETE FROM hdb_catalog.hdb_cron_event_invocation_logs WHERE created_at < NOW() - INTERVAL '%d days';", cfg.HasuraEventCleaner.ClearOlderThan),
+		fmt.Sprintf("DELETE FROM hdb_catalog.hdb_scheduled_event_invocation_logs WHERE created_at < NOW() - INTERVAL '%d days';", cfg.HasuraEventCleaner.ClearOlderThan),
+	}
+
+	for _, query := range delQueries {
+		_, err := conn.Exec(context.Background(), query)
+		if err != nil {
+			cfg.Logger.Debug("Failed to execute query", map[string]interface{}{"query": query, "error": err})
+		}
+	}
+}

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/gookit/goutil v0.6.15
 	github.com/graphql-go/graphql v0.8.1
+	github.com/jackc/pgx/v5 v5.6.0
 	github.com/lukaszraczylo/ask v0.0.0-20230927103145-2ff1123b4415
 	github.com/lukaszraczylo/go-ratecounter v0.1.8
 	github.com/lukaszraczylo/go-simple-graphql v1.2.14
@@ -22,11 +23,13 @@ require (
 
 require (
 	github.com/andybalholm/brotli v1.1.0 // indirect
-	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/gookit/color v1.5.4 // indirect
-	github.com/klauspost/compress v1.17.8 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
@@ -39,12 +42,12 @@ require (
 	github.com/valyala/histogram v1.2.0 // indirect
 	github.com/valyala/tcplisten v1.0.0 // indirect
 	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
+	golang.org/x/crypto v0.24.0 // indirect
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
 	golang.org/x/net v0.26.0 // indirect
 	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/term v0.21.0 // indirect
 	golang.org/x/text v0.16.0 // indirect
-	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -8,10 +8,11 @@ github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
-github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
-github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
@@ -31,13 +32,18 @@ github.com/gookit/goutil v0.6.15 h1:mMQ0ElojNZoyPD0eVROk5QXJPh2uKR4g06slgPDF5Jo=
 github.com/gookit/goutil v0.6.15/go.mod h1:qdKdYEHQdEtyH+4fNdQNZfJHhI0jUZzHxQVAV3DaMDY=
 github.com/graphql-go/graphql v0.8.1 h1:p7/Ou/WpmulocJeEx7wjQy611rtXGQaAcXGqanuMMgc=
 github.com/graphql-go/graphql v0.8.1/go.mod h1:nKiHzRM0qopJEwCITUuIsxk9PlVlwIiiI8pnJEhordQ=
-github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
-github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
-github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
+github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
+github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
+github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
-github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lukaszraczylo/ask v0.0.0-20230927103145-2ff1123b4415 h1:lvI8Wlbg4PxkRcg2f10wgoaRpfN19v+YdRek3+dLtlM=
@@ -69,6 +75,9 @@ github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUz
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.33.0 h1:1cU2KZkvPxNyfgEmhHAz/1A9Bz+llsdYzklWFzgp0r8=
 github.com/rs/zerolog v1.33.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
@@ -83,6 +92,8 @@ github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVS
 github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
 github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no=
 github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM=
+golang.org/x/crypto v0.24.0 h1:mnl8DM0o513X8fdIkmyFE/5hTYxbwYOjDS/+rK6qpRI=
+golang.org/x/crypto v0.24.0/go.mod h1:Z1PMYSOR5nyMcyAVAIQSKCDwalqy85Aqn1x3Ws4L5DM=
 golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI=
 golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
 golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ=
@@ -101,5 +112,6 @@ golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

logging/logging.go

@@ -76,6 +76,9 @@ func (lw *LogConfig) log(w io.Writer, level zerolog.Level, message string, field
 }
 
 func (lw *LogConfig) logWithLevel(level zerolog.Level, message string, fields map[string]interface{}) {
+	if lw.logger.GetLevel() > level {
+		return
+	}
 	if lw.logger.GetLevel() <= level {
 		w := os.Stdout
 		if level >= zerolog.ErrorLevel {

main.go

@@ -83,12 +83,16 @@ func parseConfig() {
 	c.Api.BannedUsersFile = getDetailsFromEnv("BANNED_USERS_FILE", "/go/src/app/banned_users.json")
 	c.Server.PurgeOnCrawl = getDetailsFromEnv("PURGE_METRICS_ON_CRAWL", false)
 	c.Server.PurgeEvery = getDetailsFromEnv("PURGE_METRICS_ON_TIMER", 0)
+	c.HasuraEventCleaner.Enable = getDetailsFromEnv("HASURA_EVENT_CLEANER", false)
+	c.HasuraEventCleaner.ClearOlderThan = getDetailsFromEnv("HASURA_EVENT_CLEANER_OLDER_THAN", 1)
+	c.HasuraEventCleaner.EventMetadataDb = getDetailsFromEnv("HASURA_EVENT_METADATA_DB", "")
 	cfg = &c
 
 	enableCache() // takes close to no resources, but can be used with dynamic query cache
 	loadRatelimitConfig()
 	once.Do(func() {
 		go enableApi()
+		go enableHasuraEventCleaner()
 	})
 	prepareQueriesAndExemptions()
 }

struct_config.go

@@ -9,15 +9,6 @@ import (
 
 // config is a struct that holds the configuration of the application.
 type config struct {
-	Cache struct {
-		Client             CacheClient
-		CacheTTL           int
-		CacheEnable        bool
-		CacheRedisEnable   bool
-		CacheRedisURL      string
-		CacheRedisPassword string
-		CacheRedisDB       int
-	}
 	Logger     *libpack_logging.LogConfig
 	Monitoring *libpack_monitoring.MetricsSetup
 	Api        struct{ BannedUsersFile string }
@@ -35,6 +26,20 @@ type config struct {
 		IntrospectionAllowed []string
 		BlockIntrospection   bool
 	}
+	HasuraEventCleaner struct {
+		EventMetadataDb string
+		ClearOlderThan  int
+		Enable          bool
+	}
+	Cache struct {
+		Client             CacheClient
+		CacheRedisURL      string
+		CacheRedisPassword string
+		CacheTTL           int
+		CacheRedisDB       int
+		CacheEnable        bool
+		CacheRedisEnable   bool
+	}
 	Server struct {
 		HostGraphQL         string
 		HostGraphQLReadOnly string