Delete stale recommendations from cache before setting new ones (#1145)

Author: zhenghaoz

worker/pipeline.go

@@ -216,6 +216,13 @@ func (p *Pipeline) Recommend(ctx context.Context, users []data.User, progress fu
 			log.Logger().Error("failed to cache recommendation", zap.Error(err))
 			return
 		}
+		if err = p.CacheClient.DeleteScores(ctx, []string{cache.Recommend}, cache.ScoreCondition{
+			Before: &recommendTime,
+			Subset: proto.String(userId),
+		}); err != nil {
+			log.Logger().Error("failed to delete stale recommendation", zap.Error(err))
+			return
+		}
 		if err = p.CacheClient.Set(ctx,
 			cache.Time(cache.Key(cache.RecommendUpdateTime, userId), recommendTime),
 			cache.String(cache.Key(cache.RecommendDigest, userId), digest),

worker/worker_test.go

@@ -457,6 +457,10 @@ func (suite *WorkerTestSuite) TestRecommend() {
 	}, true, true, true)
 	suite.NoError(err)
 
+	// insert stale recommendation
+	err = suite.CacheClient.AddScores(ctx, cache.Recommend, "0", []cache.Score{{Id: "999", Score: 999}})
+	suite.NoError(err)
+
 	// insert non-personalized recommendation
 	err = suite.CacheClient.AddScores(ctx, cache.NonPersonalized, "popular", []cache.Score{{Id: "3", Categories: []string{""}}})
 	suite.NoError(err)