Merge pull request #440 from sgotti/cursor_fetching_GetRemoteSources

*: add cursor fetching to GetRemoteSources api

cmd/agola/cmd/remotesourcelist.go

@@ -36,19 +36,7 @@ var cmdRemoteSourceList = &cobra.Command{
 	Short: "list",
 }
 
-type remoteSourceListOptions struct {
-	limit int
-	start string
-}
-
-var remoteSourceListOpts remoteSourceListOptions
-
 func init() {
-	flags := cmdRemoteSourceList.Flags()
-
-	flags.IntVar(&remoteSourceListOpts.limit, "limit", 10, "max number of runs to show")
-	flags.StringVar(&remoteSourceListOpts.start, "start", "", "starting user name (excluded) to fetch")
-
 	cmdRemoteSource.AddCommand(cmdRemoteSourceList)
 }
 
@@ -61,12 +49,19 @@ func printRemoteSources(remoteSources []*gwapitypes.RemoteSourceResponse) {
 func remoteSourceList(cmd *cobra.Command, args []string) error {
 	gwClient := gwclient.NewClient(gatewayURL, token)
 
-	remouteSources, _, err := gwClient.GetRemoteSources(context.TODO(), remoteSourceListOpts.start, remoteSourceListOpts.limit, false)
-	if err != nil {
-		return errors.WithStack(err)
-	}
+	var cursor string
+	for {
+		remoteSources, resp, err := gwClient.GetRemoteSources(context.TODO(), &gwclient.ListOptions{Cursor: cursor})
+		if err != nil {
+			return errors.Wrapf(err, "failed to get remote sources")
+		}
+		printRemoteSources(remoteSources)
 
-	printRemoteSources(remouteSources)
+		cursor = resp.Cursor
+		if cursor == "" {
+			break
+		}
+	}
 
 	return nil
 }

internal/services/configstore/action/remotesource.go

@@ -24,6 +24,49 @@ import (
 	"agola.io/agola/services/configstore/types"
 )
 
+type GetRemoteSourcesRequest struct {
+	StartRemoteSourceName string
+
+	Limit         int
+	SortDirection types.SortDirection
+}
+
+type GetRemoteSourcesResponse struct {
+	RemoteSources []*types.RemoteSource
+
+	HasMore bool
+}
+
+func (h *ActionHandler) GetRemoteSources(ctx context.Context, req *GetRemoteSourcesRequest) (*GetRemoteSourcesResponse, error) {
+	limit := req.Limit
+	if limit > 0 {
+		limit += 1
+	}
+
+	var remoteSources []*types.RemoteSource
+	err := h.d.Do(ctx, func(tx *sql.Tx) error {
+		var err error
+		remoteSources, err = h.d.GetRemoteSources(tx, req.StartRemoteSourceName, limit, req.SortDirection)
+		return errors.WithStack(err)
+	})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	var hasMore bool
+	if req.Limit > 0 {
+		hasMore = len(remoteSources) > req.Limit
+		if hasMore {
+			remoteSources = remoteSources[0:req.Limit]
+		}
+	}
+
+	return &GetRemoteSourcesResponse{
+		RemoteSources: remoteSources,
+		HasMore:       hasMore,
+	}, nil
+}
+
 func (h *ActionHandler) ValidateRemoteSourceReq(ctx context.Context, req *CreateUpdateRemoteSourceRequest) error {
 	if req.Name == "" {
 		return util.NewAPIError(util.ErrBadRequest, errors.Errorf("remotesource name required"))

internal/services/configstore/api/remotesource.go

@@ -17,7 +17,6 @@ package api
 import (
 	"encoding/json"
 	"net/http"
-	"strconv"
 
 	"github.com/gorilla/mux"
 	"github.com/rs/zerolog"
@@ -182,63 +181,46 @@ func (h *DeleteRemoteSourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Req
 	}
 }
 
-const (
-	DefaultRemoteSourcesLimit = 10
-	MaxRemoteSourcesLimit     = 20
-)
-
 type RemoteSourcesHandler struct {
 	log zerolog.Logger
-	d   *db.DB
+	ah  *action.ActionHandler
 }
 
-func NewRemoteSourcesHandler(log zerolog.Logger, d *db.DB) *RemoteSourcesHandler {
-	return &RemoteSourcesHandler{log: log, d: d}
+func NewRemoteSourcesHandler(log zerolog.Logger, ah *action.ActionHandler) *RemoteSourcesHandler {
+	return &RemoteSourcesHandler{log: log, ah: ah}
 }
 
 func (h *RemoteSourcesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	ctx := r.Context()
-	query := r.URL.Query()
-
-	limitS := query.Get("limit")
-	limit := DefaultRemoteSourcesLimit
-	if limitS != "" {
-		var err error
-		limit, err = strconv.Atoi(limitS)
-		if err != nil {
-			util.HTTPError(w, util.NewAPIError(util.ErrBadRequest, errors.Wrapf(err, "cannot parse limit")))
-			return
-		}
-	}
-	if limit < 0 {
-		util.HTTPError(w, util.NewAPIError(util.ErrBadRequest, errors.Errorf("limit must be greater or equal than 0")))
+	res, err := h.do(w, r)
+	if util.HTTPError(w, err) {
+		h.log.Err(err).Send()
 		return
 	}
-	if limit > MaxRemoteSourcesLimit {
-		limit = MaxRemoteSourcesLimit
-	}
-	asc := false
-	if _, ok := query["asc"]; ok {
-		asc = true
+
+	if err := util.HTTPResponse(w, http.StatusOK, res); err != nil {
+		h.log.Err(err).Send()
 	}
+}
 
-	start := query.Get("start")
+func (h *RemoteSourcesHandler) do(w http.ResponseWriter, r *http.Request) ([]*types.RemoteSource, error) {
+	ctx := r.Context()
+	query := r.URL.Query()
 
-	var remoteSources []*types.RemoteSource
-	err := h.d.Do(ctx, func(tx *sql.Tx) error {
-		var err error
-		remoteSources, err = h.d.GetRemoteSources(tx, start, limit, asc)
-		return errors.WithStack(err)
-	})
+	ropts, err := parseRequestOptions(r)
 	if err != nil {
-		h.log.Err(err).Send()
-		util.HTTPError(w, err)
-		return
+		return nil, errors.WithStack(err)
 	}
 
-	if err := util.HTTPResponse(w, http.StatusOK, remoteSources); err != nil {
-		h.log.Err(err).Send()
+	startRemoteSourceName := query.Get("startremotesourcename")
+
+	ares, err := h.ah.GetRemoteSources(ctx, &action.GetRemoteSourcesRequest{StartRemoteSourceName: startRemoteSourceName, Limit: ropts.Limit, SortDirection: ropts.SortDirection})
+	if err != nil {
+		return nil, errors.WithStack(err)
 	}
+
+	addHasMoreHeader(w, ares.HasMore)
+
+	return ares.RemoteSources, nil
 }
 
 type LinkedAccountsHandler struct {

internal/services/configstore/configstore.go

@@ -195,7 +195,7 @@ func (s *Configstore) setupDefaultRouter() http.Handler {
 	removeOrgMemberHandler := api.NewRemoveOrgMemberHandler(s.log, s.ah)
 
 	remoteSourceHandler := api.NewRemoteSourceHandler(s.log, s.d)
-	remoteSourcesHandler := api.NewRemoteSourcesHandler(s.log, s.d)
+	remoteSourcesHandler := api.NewRemoteSourcesHandler(s.log, s.ah)
 	createRemoteSourceHandler := api.NewCreateRemoteSourceHandler(s.log, s.ah)
 	updateRemoteSourceHandler := api.NewUpdateRemoteSourceHandler(s.log, s.ah)
 	deleteRemoteSourceHandler := api.NewDeleteRemoteSourceHandler(s.log, s.ah)

internal/services/configstore/configstore_test.go

@@ -84,7 +84,7 @@ func getRemoteSources(ctx context.Context, cs *Configstore) ([]*types.RemoteSour
 	var users []*types.RemoteSource
 	err := cs.d.Do(ctx, func(tx *sql.Tx) error {
 		var err error
-		users, err = cs.d.GetRemoteSources(tx, "", 0, true)
+		users, err = cs.d.GetRemoteSources(tx, "", 0, types.SortDirectionAsc)
 		return errors.WithStack(err)
 	})
 
@@ -1071,6 +1071,105 @@ func TestOrgMembers(t *testing.T) {
 	})
 }
 
+func TestGetRemoteSources(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	ctx := context.Background()
+	log := testutil.NewLogger(t)
+
+	cs := setupConfigstore(ctx, t, log, dir)
+
+	t.Logf("starting cs")
+	go func() { _ = cs.Run(ctx) }()
+
+	remoteSources := []*types.RemoteSource{}
+	for i := 1; i < 10; i++ {
+		remoteSource, err := cs.ah.CreateRemoteSource(ctx, &action.CreateUpdateRemoteSourceRequest{Name: fmt.Sprintf("rs%d", i), Type: types.RemoteSourceTypeGitea, AuthType: types.RemoteSourceAuthTypePassword, APIURL: "http://example.com"})
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		remoteSources = append(remoteSources, remoteSource)
+	}
+
+	t.Run("test get all remote sources", func(t *testing.T) {
+		res, err := cs.ah.GetRemoteSources(ctx, &action.GetRemoteSourcesRequest{SortDirection: types.SortDirectionAsc})
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		expectedRemoteSources := 9
+		if len(res.RemoteSources) != expectedRemoteSources {
+			t.Fatalf("expected %d remote sources, got %d remote sources", expectedRemoteSources, len(res.RemoteSources))
+		}
+		if res.HasMore {
+			t.Fatalf("expected hasMore false, got %t", res.HasMore)
+		}
+	})
+
+	t.Run("test get remote sources with limit less than remote sources", func(t *testing.T) {
+		res, err := cs.ah.GetRemoteSources(ctx, &action.GetRemoteSourcesRequest{Limit: 5, SortDirection: types.SortDirectionAsc})
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		expectedRemoteSources := 5
+		if len(res.RemoteSources) != expectedRemoteSources {
+			t.Fatalf("expected %d remote sources, got %d remote sources", expectedRemoteSources, len(res.RemoteSources))
+		}
+		if !res.HasMore {
+			t.Fatalf("expected hasMore true, got %t", res.HasMore)
+		}
+	})
+
+	t.Run("test get remote sources with limit less than remote sources continuation", func(t *testing.T) {
+		respAllRemoteSources := []*types.RemoteSource{}
+
+		res, err := cs.ah.GetRemoteSources(ctx, &action.GetRemoteSourcesRequest{Limit: 5, SortDirection: types.SortDirectionAsc})
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+
+		expectedRemoteSources := 5
+		if len(res.RemoteSources) != expectedRemoteSources {
+			t.Fatalf("expected %d remote sources, got %d remote sources", expectedRemoteSources, len(res.RemoteSources))
+		}
+		if !res.HasMore {
+			t.Fatalf("expected hasMore true, got %t", res.HasMore)
+		}
+
+		respAllRemoteSources = append(respAllRemoteSources, res.RemoteSources...)
+		lastRemoteSource := res.RemoteSources[len(res.RemoteSources)-1]
+
+		// fetch next results
+		for {
+			res, err = cs.ah.GetRemoteSources(ctx, &action.GetRemoteSourcesRequest{StartRemoteSourceName: lastRemoteSource.Name, Limit: 5, SortDirection: types.SortDirectionAsc})
+			if err != nil {
+				t.Fatalf("unexpected err: %v", err)
+			}
+			expectedRemoteSources := 5
+			if res.HasMore && len(res.RemoteSources) != expectedRemoteSources {
+				t.Fatalf("expected %d remote sources, got %d remote sources", expectedRemoteSources, len(res.RemoteSources))
+			}
+
+			respAllRemoteSources = append(respAllRemoteSources, res.RemoteSources...)
+
+			if !res.HasMore {
+				break
+			}
+
+			lastRemoteSource = res.RemoteSources[len(res.RemoteSources)-1]
+		}
+
+		expectedRemoteSources = 9
+		if len(remoteSources) != expectedRemoteSources {
+			t.Fatalf("expected %d remote sources, got %d remote sources", expectedRemoteSources, len(remoteSources))
+		}
+
+		if diff := cmpDiffObject(remoteSources, respAllRemoteSources); diff != "" {
+			t.Fatalf("mismatch (-want +got):\n%s", diff)
+		}
+	})
+}
+
 func TestGetUsers(t *testing.T) {
 	t.Parallel()
 

internal/services/configstore/db/db.go

@@ -174,17 +174,20 @@ func (d *DB) GetRemoteSourceByName(tx *sql.Tx, name string) (*types.RemoteSource
 	return out, errors.WithStack(err)
 }
 
-func getRemoteSourcesFilteredQuery(startRemoteSourceName string, limit int, asc bool) *sq.SelectBuilder {
+func getRemoteSourcesFilteredQuery(startRemoteSourceName string, limit int, sortDirection types.SortDirection) *sq.SelectBuilder {
 	q := remoteSourceSelect()
-	if asc {
-		q = q.OrderBy("remotesource.name").Asc()
-	} else {
-		q = q.OrderBy("remotesource.name").Desc()
+	q = q.OrderBy("remotesource.name")
+	switch sortDirection {
+	case types.SortDirectionAsc:
+		q = q.Asc()
+	case types.SortDirectionDesc:
+		q = q.Desc()
 	}
 	if startRemoteSourceName != "" {
-		if asc {
+		switch sortDirection {
+		case types.SortDirectionAsc:
 			q = q.Where(q.G("remotesource.name", startRemoteSourceName))
-		} else {
+		case types.SortDirectionDesc:
 			q = q.Where(q.L("remotesource.name", startRemoteSourceName))
 		}
 	}
@@ -195,8 +198,8 @@ func getRemoteSourcesFilteredQuery(startRemoteSourceName string, limit int, asc
 	return q
 }
 
-func (d *DB) GetRemoteSources(tx *sql.Tx, startRemoteSourceName string, limit int, asc bool) ([]*types.RemoteSource, error) {
-	q := getRemoteSourcesFilteredQuery(startRemoteSourceName, limit, asc)
+func (d *DB) GetRemoteSources(tx *sql.Tx, startRemoteSourceName string, limit int, sortDirection types.SortDirection) ([]*types.RemoteSource, error) {
+	q := getRemoteSourcesFilteredQuery(startRemoteSourceName, limit, sortDirection)
 	remoteSources, _, err := d.fetchRemoteSources(tx, q)
 
 	return remoteSources, errors.WithStack(err)