feat: add pagination when fetching records (#42)

* feat(records): add pagination when getting type's records

* fix(tags): inconsistent error

* feat: add cors middleware

api/handlers/record_handler.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"net/url"
+	"strconv"
 	"strings"
 
 	"github.com/gorilla/mux"
@@ -140,9 +142,13 @@ func (h *RecordHandler) GetByType(w http.ResponseWriter, r *http.Request) {
 		writeJSONError(w, status, message)
 		return
 	}
-	filterCfg := filterConfigFromValues(r.URL.Query())
+	getCfg, err := h.buildGetConfig(r.URL.Query())
+	if err != nil {
+		writeJSONError(w, http.StatusBadRequest, err.Error())
+		return
+	}
 
-	records, err := recordRepo.GetAll(r.Context(), filterCfg)
+	recordList, err := recordRepo.GetAll(r.Context(), getCfg)
 	if err != nil {
 		h.logger.WithField("type", t.Name).
 			Errorf("error fetching records: GetAll: %v", err)
@@ -153,9 +159,9 @@ func (h *RecordHandler) GetByType(w http.ResponseWriter, r *http.Request) {
 
 	fieldsToSelect := h.parseSelectQuery(r.URL.Query().Get("select"))
 	if len(fieldsToSelect) > 0 {
-		records = h.selectRecordFields(fieldsToSelect, records)
+		recordList.Data = h.selectRecordFields(fieldsToSelect, recordList.Data)
 	}
-	writeJSON(w, http.StatusOK, records)
+	writeJSON(w, http.StatusOK, recordList)
 }
 
 func (h *RecordHandler) GetOneByType(w http.ResponseWriter, r *http.Request) {
@@ -198,6 +204,25 @@ func (h *RecordHandler) GetOneByType(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, record)
 }
 
+func (h *RecordHandler) buildGetConfig(params url.Values) (cfg discovery.GetConfig, err error) {
+	if size := params.Get("size"); size != "" {
+		cfg.Size, err = strconv.Atoi(size)
+		if err != nil {
+			return
+		}
+	}
+	if from := params.Get("from"); from != "" {
+		cfg.From, err = strconv.Atoi(from)
+		if err != nil {
+			return
+		}
+	}
+
+	cfg.Filters = filterConfigFromValues(params)
+
+	return
+}
+
 func (h *RecordHandler) parseSelectQuery(raw string) (fields []string) {
 	tokens := strings.Split(raw, ",")
 	for _, token := range tokens {

api/handlers/record_handler_test.go

@@ -337,13 +337,17 @@ func TestRecordHandler(t *testing.T) {
 				Setup:        func(tc *testCase, rrf *mock.RecordRepositoryFactory) {},
 			},
 			{
-				Description:  "should return an http 200 irrespective of environment value",
+				Description:  "should get from and size from querystring and pass it to repo",
 				Type:         typeName,
-				QueryStrings: "filter.data.environment=nonexisting",
+				QueryStrings: "from=5&size=10",
 				ExpectStatus: http.StatusOK,
 				Setup: func(tc *testCase, rrf *mock.RecordRepositoryFactory) {
 					rr := new(mock.RecordRepository)
-					rr.On("GetAll", ctx, map[string][]string{"data.environment": {"nonexisting"}}).Return(records, nil)
+					rr.On("GetAll", ctx, discovery.GetConfig{
+						Filters: map[string][]string{},
+						From:    5,
+						Size:    10,
+					}).Return(discovery.RecordList{Data: records}, nil)
 					rrf.On("For", typeName).Return(rr, nil)
 				},
 			},
@@ -354,32 +358,63 @@ func TestRecordHandler(t *testing.T) {
 				ExpectStatus: http.StatusOK,
 				Setup: func(tc *testCase, rrf *mock.RecordRepositoryFactory) {
 					rr := new(mock.RecordRepository)
-					rr.On("GetAll", ctx, map[string][]string{
-						"service":      {"kafka", "rabbitmq"},
-						"data.company": {"appel"},
-					}).Return(records, nil)
+					rr.On("GetAll", ctx, discovery.GetConfig{
+						Filters: map[string][]string{
+							"service":      {"kafka", "rabbitmq"},
+							"data.company": {"appel"},
+						}}).Return(discovery.RecordList{Data: records}, nil)
 					rrf.On("For", typeName).Return(rr, nil)
 				},
 			},
 			{
-				Description:  "should return all records for an type",
+				Description:  "should return http 500 if the handler fails to construct record repository",
 				Type:         typeName,
 				QueryStrings: "filter.data.environment=test",
+				ExpectStatus: http.StatusInternalServerError,
+				Setup: func(tc *testCase, rrf *mock.RecordRepositoryFactory) {
+					rr := new(mock.RecordRepository)
+					err := fmt.Errorf("something went wrong")
+					rrf.On("For", typeName).Return(rr, err)
+				},
+			},
+			{
+				Description:  "should return an http 500 if calling recordRepository.GetAll fails",
+				Type:         typeName,
+				QueryStrings: "filter.data.environment=test",
+				ExpectStatus: http.StatusInternalServerError,
+				Setup: func(tc *testCase, rrf *mock.RecordRepositoryFactory) {
+					rr := new(mock.RecordRepository)
+					err := fmt.Errorf("temporarily unavailable")
+					rr.On("GetAll", ctx, discovery.GetConfig{
+						Filters: map[string][]string{"data.environment": {"test"}},
+					}).Return(discovery.RecordList{Data: []record.Record{}}, err)
+					rrf.On("For", typeName).Return(rr, nil)
+				},
+			},
+			{
+				Description:  "should return 200 on success and RecordList",
+				Type:         typeName,
 				ExpectStatus: http.StatusOK,
 				Setup: func(tc *testCase, rrf *mock.RecordRepositoryFactory) {
 					rr := new(mock.RecordRepository)
-					rr.On("GetAll", ctx, map[string][]string{"data.environment": {"test"}}).Return(records, nil)
+					rr.On("GetAll", ctx, discovery.GetConfig{
+						Filters: map[string][]string{},
+					}).Return(discovery.RecordList{Data: records}, nil)
 					rrf.On("For", typeName).Return(rr, nil)
 				},
 				PostCheck: func(tc *testCase, resp *http.Response) error {
-					var response []record.Record
+					var response discovery.RecordList
 					err := json.NewDecoder(resp.Body).Decode(&response)
 					if err != nil {
 						return fmt.Errorf("error parsing response payload: %v", err)
 					}
-					// TODO: more useful error messages
-					if reflect.DeepEqual(response, records) == false {
-						return fmt.Errorf("expected handler to return %v, returned %v instead", records, response)
+
+					expected := discovery.RecordList{
+						Data: records,
+					}
+
+					if reflect.DeepEqual(response, expected) == false {
+						return fmt.Errorf("expected handler to return %v, returned %v instead", expected, response)
 					}
 					return nil
 				},
@@ -391,63 +426,44 @@ func TestRecordHandler(t *testing.T) {
 				ExpectStatus: http.StatusOK,
 				Setup: func(tc *testCase, rrf *mock.RecordRepositoryFactory) {
 					rr := new(mock.RecordRepository)
-					rr.On("GetAll", ctx, map[string][]string{"data.environment": {"test"}}).Return(records, nil)
+					rr.On("GetAll", ctx, discovery.GetConfig{
+						Filters: map[string][]string{"data.environment": {"test"}},
+					}).Return(discovery.RecordList{Data: records}, nil)
 					rrf.On("For", typeName).Return(rr, nil)
 				},
 				PostCheck: func(tc *testCase, resp *http.Response) error {
-					var expectRecords = []record.Record{
-						{
-							Urn: "test-fh-1",
-							Data: map[string]interface{}{
-								"urn":   "test-fh-1",
-								"owner": "de",
-							},
-						},
-						{
-							Urn: "test-fh-2",
-							Data: map[string]interface{}{
-								"urn":   "test-fh-2",
-								"owner": "de",
-							},
-						},
-					}
-
-					var response []record.Record
+					var response discovery.RecordList
 					err := json.NewDecoder(resp.Body).Decode(&response)
 					if err != nil {
 						return fmt.Errorf("error parsing response payload: %v", err)
 					}
 
-					if reflect.DeepEqual(response, expectRecords) == false {
-						return fmt.Errorf("expected handler to return %v, returned %v instead", expectRecords, response)
+					expected := discovery.RecordList{
+						Data: []record.Record{
+							{
+								Urn: "test-fh-1",
+								Data: map[string]interface{}{
+									"urn":   "test-fh-1",
+									"owner": "de",
+								},
+							},
+							{
+								Urn: "test-fh-2",
+								Data: map[string]interface{}{
+									"urn":   "test-fh-2",
+									"owner": "de",
+								},
+							},
+						},
+					}
+
+					if reflect.DeepEqual(response, expected) == false {
+						return fmt.Errorf("expected handler to return %v, returned %v instead", expected, response)
 					}
 
 					return nil
 				},
 			},
-			{
-				Description:  "(internal) should return http 500 if the handler fails to construct record repository",
-				Type:         typeName,
-				QueryStrings: "filter.data.environment=test",
-				ExpectStatus: http.StatusInternalServerError,
-				Setup: func(tc *testCase, rrf *mock.RecordRepositoryFactory) {
-					rr := new(mock.RecordRepository)
-					err := fmt.Errorf("something went wrong")
-					rrf.On("For", typeName).Return(rr, err)
-				},
-			},
-			{
-				Description:  "(internal) should return an http 500 if calling recordRepository.GetAll fails",
-				Type:         typeName,
-				QueryStrings: "filter.data.environment=test",
-				ExpectStatus: http.StatusInternalServerError,
-				Setup: func(tc *testCase, rrf *mock.RecordRepositoryFactory) {
-					rr := new(mock.RecordRepository)
-					err := fmt.Errorf("temporarily unavailable")
-					rr.On("GetAll", ctx, map[string][]string{"data.environment": {"test"}}).Return([]record.Record{}, err)
-					rrf.On("For", typeName).Return(rr, nil)
-				},
-			},
 		}
 		for _, tc := range testCases {
 			t.Run(tc.Description, func(t *testing.T) {

cmd/serve.go

@@ -21,6 +21,7 @@ import (
 	esStore "github.com/odpf/columbus/store/elasticsearch"
 	"github.com/odpf/columbus/store/postgres"
 	"github.com/odpf/columbus/tag"
+	"github.com/rs/cors"
 	"github.com/sirupsen/logrus"
 	"gorm.io/gorm"
 )
@@ -43,10 +44,12 @@ func Serve() {
 	newRelicMonitor := initNewRelicMonitor(config)
 	statsdMonitor := initStatsdMonitor(config)
 	router := initRouter(esClient, newRelicMonitor, statsdMonitor, rootLogger)
+	c := cors.Default()
+	handler := c.Handler(router)
 
 	serverAddr := fmt.Sprintf("%s:%s", config.ServerHost, config.ServerPort)
 	log.Printf("starting http server on %s", serverAddr)
-	if err := http.ListenAndServe(serverAddr, router); err != nil {
+	if err := http.ListenAndServe(serverAddr, handler); err != nil {
 		log.Errorf("listen and serve: %v", err)
 	}
 }

discovery/model.go

@@ -1,5 +1,12 @@
 package discovery
 
+import "github.com/odpf/columbus/record"
+
+// RecordFilter is a filter intended to be used as a search
+// criteria for operations involving record search
+type RecordFilter = map[string][]string
+
+// SearchResult represents an item/result in a list of search results
 type SearchResult struct {
 	ID          string            `json:"id"`
 	Title       string            `json:"title"`
@@ -8,3 +15,54 @@ type SearchResult struct {
 	Description string            `json:"description"`
 	Labels      map[string]string `json:"labels"`
 }
+
+// SearchConfig represents a search query along
+// with any corresponding filter(s)
+type SearchConfig struct {
+	// Text to search for
+	Text string
+
+	// Filters specifies document level values to look for.
+	// Multiple values can be specified for a single key
+	Filters RecordFilter
+
+	// Number of relevant results to return
+	MaxResults int
+
+	// List of record types to search for
+	// a zero value signifies that all types should be searched
+	TypeWhiteList []string
+
+	// RankBy is a param to rank based on a specific parameter
+	RankBy string
+
+	// Queries is a param to search a resource based on record's fields
+	Queries map[string]string
+}
+
+// GetConfig represents a get query along
+// with any corresponding filter(s)
+type GetConfig struct {
+	// Filters specifies document level values to look for.
+	// Multiple values can be specified for a single key
+	Filters RecordFilter
+
+	// Number of relevant results to return
+	Size int
+
+	// Offset to fetch records from
+	From int
+}
+
+// RecordList is a struct that wraps list of records with total
+type RecordList struct {
+	// Data contains list of fetched records
+	Data []record.Record `json:"data"`
+
+	// Count is the length of Data
+	Count int `json:"count"`
+
+	// Total is the total of available data in the repository
+	// It also includes those that are not fetched
+	Total int `json:"total"`
+}

discovery/repo.go

@@ -17,9 +17,8 @@ type RecordRepository interface {
 	CreateOrReplaceMany(context.Context, []record.Record) error
 
 	// GetAll returns specific records from storage
-	// RecordFilter is an optional data structure that is
-	// used for return documents matching the search criteria.
-	GetAll(context.Context, RecordFilter) ([]record.Record, error)
+	// GetConfig is used to configure fetching such as filters and offset
+	GetAll(ctx context.Context, cfg GetConfig) (RecordList, error)
 
 	// GetAllIterator returns RecordIterator to iterate records by batches
 	GetAllIterator(context.Context) (RecordIterator, error)
@@ -33,9 +32,6 @@ type RecordRepository interface {
 	// The field that contains this ID is defined by the
 	// type to which this record belongs
 	Delete(context.Context, string) error
-
-	// TODO: we should probably switch to iterator types for returning
-	// records, or we could add options for pagination
 }
 
 // RecordRepositoryFactory represents a type capable

go.mod

@@ -23,6 +23,7 @@ require (
 	github.com/odpf/salt v0.0.0-20211028100023-de463ef825e1
 	github.com/olivere/elastic/v7 v7.0.12
 	github.com/pkg/errors v0.9.1
+	github.com/rs/cors v1.8.2
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/cobra v1.2.1
 	github.com/spf13/viper v1.8.1