support offset and user-id on latest items API (#1136)

Author: zhenghaoz

common/parallel/parallel_test.go

@@ -163,7 +163,7 @@ func TestParallelCancel(t *testing.T) {
 				cancel()
 			}
 			count.Add(1)
-			time.Sleep(time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			return nil
 		})
 

server/rest.go

@@ -672,20 +672,61 @@ func (s *RestServer) SearchDocuments(collection, subset string, categories []str
 }
 
 func (s *RestServer) getLatest(request *restful.Request, response *restful.Response) {
-	n, err := ParseInt(request, "n", s.Config.Server.DefaultN)
-	if err != nil {
-		BadRequest(response, errors.New("invalid n parameter"))
+	var (
+		offset int
+		n      int
+		err    error
+	)
+	ctx := request.Request.Context()
+	if offset, err = ParseInt(request, "offset", 0); err != nil {
+		BadRequest(response, errors.Errorf("invalid offset parameter: %v", err))
+		return
+	}
+	if n, err = ParseInt(request, "n", s.Config.Server.DefaultN); err != nil {
+		BadRequest(response, errors.Errorf("invalid n parameter: %v", err))
 		return
 	}
-
 	categories := ReadCategories(request, nil)
-	log.ResponseLogger(response).Debug("get category latest items in category", zap.Strings("categories", categories))
+	userId := request.QueryParameter("user-id")
+
+	readItems := mapset.NewSet[string]()
+	if userId != "" {
+		feedback, err := s.DataClient.GetUserFeedback(ctx, userId, s.Config.Now())
+		if err != nil {
+			InternalServerError(response, err)
+			return
+		}
+		for _, f := range feedback {
+			readItems.Add(f.ItemId)
+		}
+	}
+
+	limit := offset + n
+	if readItems.Cardinality() > 0 {
+		limit += readItems.Cardinality()
+	}
 
-	items, err := s.DataClient.GetLatestItems(request.Request.Context(), n, categories)
+	items, err := s.DataClient.GetLatestItems(ctx, limit, categories)
 	if err != nil {
 		InternalServerError(response, err)
 		return
 	}
+
+	if readItems.Cardinality() > 0 {
+		filtered := make([]data.Item, 0, len(items))
+		for _, item := range items {
+			if !readItems.Contains(item.ItemId) {
+				filtered = append(filtered, item)
+			}
+		}
+		items = filtered
+	}
+
+	items = items[min(offset, len(items)):]
+	if n > 0 && len(items) > n {
+		items = items[:n]
+	}
+
 	Ok(response, lo.Map(items, func(item data.Item, _ int) cache.Score {
 		return cache.Score{
 			Id:    item.ItemId,

server/rest_test.go

@@ -293,6 +293,20 @@ func (suite *ServerTestSuite) TestItems() {
 			{Id: items[1].ItemId, Score: float64(items[1].Timestamp.Unix())},
 		})).
 		End()
+	apitest.New().
+		Handler(suite.handler).
+		Get("/api/latest/").
+		Header("X-API-Key", apiKey).
+		QueryParams(map[string]string{
+			"n":      "3",
+			"offset": "1",
+		}).
+		Expect(t).
+		Status(http.StatusOK).
+		Body(suite.marshal([]cache.Score{
+			{Id: items[1].ItemId, Score: float64(items[1].Timestamp.Unix())},
+		})).
+		End()
 	apitest.New().
 		Handler(suite.handler).
 		Get("/api/latest/*").
@@ -307,6 +321,25 @@ func (suite *ServerTestSuite) TestItems() {
 			{Id: items[1].ItemId, Score: float64(items[1].Timestamp.Unix())},
 		})).
 		End()
+	err := suite.DataClient.BatchInsertFeedback(context.Background(), []data.Feedback{{
+		FeedbackKey: data.FeedbackKey{FeedbackType: "read", UserId: "0", ItemId: "6"},
+		Timestamp:   time.Now().Truncate(time.Hour),
+	}}, true, true, true)
+	suite.NoError(err)
+	apitest.New().
+		Handler(suite.handler).
+		Get("/api/latest").
+		Header("X-API-Key", apiKey).
+		QueryParams(map[string]string{
+			"n":       "3",
+			"user-id": "0",
+		}).
+		Expect(t).
+		Status(http.StatusOK).
+		Body(suite.marshal([]cache.Score{
+			{Id: items[1].ItemId, Score: float64(items[1].Timestamp.Unix())},
+		})).
+		End()
 
 	// delete item
 	apitest.New().