feat: implement expiration for latest feed data

node/pkg/db/redis.go

@@ -68,14 +68,39 @@ func MSetObject(ctx context.Context, values map[string]any) error {
 		data, err := json.Marshal(value)
 		if err != nil {
 			log.Error().Err(err).Msg("Error marshalling object")
-			return err
+			continue
 		}
 		pairs = append(pairs, key, string(data))
 	}
 
 	return rdbConn.MSet(ctx, pairs...).Err()
 }
 
+func MSetObjectWithExp(ctx context.Context, values map[string]any, exp time.Duration) error {
+	rdbConn, err := GetRedisConn(ctx)
+	if err != nil {
+		return err
+	}
+
+	var pairs []any
+	for key, value := range values {
+		data, err := json.Marshal(value)
+		if err != nil {
+			log.Error().Err(err).Msg("Error marshalling object")
+			continue
+		}
+		pairs = append(pairs, key, string(data))
+	}
+
+	pipe := rdbConn.TxPipeline()
+	pipe.MSet(ctx, pairs...)
+	for key := range values {
+		pipe.Expire(ctx, key, exp)
+	}
+	_, err = pipe.Exec(ctx)
+	return err
+}
+
 func Set(ctx context.Context, key string, value string, exp time.Duration) error {
 	rdbConn, err := GetRedisConn(ctx)
 	if err != nil {

node/pkg/db/redis_test.go

@@ -480,3 +480,47 @@ func TestPopAllObject(t *testing.T) {
 		t.Errorf("Expected empty list, got %+v", result)
 	}
 }
+func TestMSetObjectWithExp(t *testing.T) {
+	ctx := context.Background()
+
+	values := map[string]any{
+		"key1": "value1",
+		"key2": "value2",
+	}
+
+	exp := 1 * time.Second
+
+	err := MSetObjectWithExp(ctx, values, exp)
+	if err != nil {
+		t.Errorf("Error setting objects with expiration: %v", err)
+	}
+
+	// Check if the values were set correctly
+	for key, value := range values {
+		gotValue, getValueErr := Get(ctx, key)
+		if getValueErr != nil {
+			t.Errorf("Error getting key: %v", getValueErr)
+		}
+		expectedValue, marshalErr := json.Marshal(value)
+		if marshalErr != nil {
+			t.Errorf("Error marshalling value: %v", marshalErr)
+			continue
+		}
+		if gotValue != string(expectedValue) {
+			t.Errorf("Value did not match expected. Got %v, expected %v", gotValue, string(expectedValue))
+		}
+	}
+
+	time.Sleep(1001 * time.Millisecond)
+
+	// Check if the values were expired
+	for key := range values {
+		gotValue, getValueErr := Get(ctx, key)
+		if getValueErr == nil || !strings.Contains(getValueErr.Error(), "redis: nil") {
+			t.Errorf("Expected to have err")
+		}
+		if gotValue != "" {
+			t.Errorf("Expected empty value, got %v", gotValue)
+		}
+	}
+}

node/pkg/fetcher/fetcher.go

@@ -55,7 +55,7 @@ func (f *Fetcher) fetcherJob(ctx context.Context, chainHelpers map[string]ChainH
 		return err
 	}
 
-	err = setLatestFeedData(ctx, result)
+	err = setLatestFeedData(ctx, result, time.Duration(f.FetchInterval)*time.Millisecond)
 	if err != nil {
 		log.Error().Str("Player", "Fetcher").Err(err).Msg("error in setLatestFeedData")
 		return err

node/pkg/fetcher/utils.go

@@ -50,12 +50,12 @@ func getTokenPrice(sqrtPriceX96 *big.Int, definition *Definition) (float64, erro
 	return math.Round(result), nil
 }
 
-func setLatestFeedData(ctx context.Context, feedData []FeedData) error {
+func setLatestFeedData(ctx context.Context, feedData []FeedData, expiration time.Duration) error {
 	latestData := make(map[string]any)
 	for _, data := range feedData {
 		latestData["latestFeedData:"+strconv.Itoa(int(data.FeedID))] = data
 	}
-	return db.MSetObject(ctx, latestData)
+	return db.MSetObjectWithExp(ctx, latestData, expiration)
 }
 
 func getLatestFeedData(ctx context.Context, feedIds []int32) ([]FeedData, error) {

node/pkg/fetcher/utils_test.go

@@ -146,7 +146,7 @@ func TestSetLatestFeedData(t *testing.T) {
 		},
 	}
 
-	err := setLatestFeedData(ctx, feedData)
+	err := setLatestFeedData(ctx, feedData, 1*time.Second)
 	if err != nil {
 		t.Fatalf("error setting latest feed data: %v", err)
 	}
@@ -179,7 +179,7 @@ func TestGetLatestFeedData(t *testing.T) {
 	}
 
 	keys := []string{"latestFeedData:1", "latestFeedData:2"}
-	err := setLatestFeedData(ctx, feedData)
+	err := setLatestFeedData(ctx, feedData, 1*time.Second)
 	if err != nil {
 		t.Fatalf("error setting latest feed data: %v", err)
 	}