Implement Cursor-Based Pagination in ListRepositories Endpoint (#2097)

Signed-off-by: Vyom-Yadav <[email protected]>

database/migrations/000012_repositories_pagination_idx.down.sql

@@ -0,0 +1,15 @@
+-- Copyright 2024 Stacklok, Inc
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+DROP INDEX IF EXISTS repositories_cursor_pagination_idx;

database/migrations/000012_repositories_pagination_idx.up.sql

@@ -0,0 +1,16 @@
+-- Copyright 2024 Stacklok, Inc
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+-- This adds a unique index to the repositories table to support cursor pagination.
+CREATE UNIQUE INDEX repositories_cursor_pagination_idx ON repositories(project_id, provider, repo_id);

database/query/repositories.sql

@@ -28,7 +28,9 @@ SELECT * FROM repositories WHERE provider = $1 AND repo_id = $2 AND project_id =
 -- name: ListRepositoriesByProjectID :many
 SELECT * FROM repositories
 WHERE provider = $1 AND project_id = $2
-ORDER BY repo_name;
+  AND (repo_id >= sqlc.narg('repo_id') OR sqlc.narg('repo_id') IS NULL)
+ORDER BY project_id, provider, repo_id
+LIMIT sqlc.narg('limit');
 
 -- name: ListRegisteredRepositoriesByProjectIDAndProvider :many
 SELECT * FROM repositories

internal/controlplane/cursor.go

@@ -0,0 +1,81 @@
+// Copyright 2024 Stacklok, Inc
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlplane
+
+import (
+	"encoding/base64"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// cursorDelimiter is the delimiter used to encode/decode cursors
+const cursorDelimiter = ","
+
+// RepoCursor is a cursor for listing repositories
+type RepoCursor struct {
+	ProjectId string
+	Provider  string
+	RepoId    int32
+}
+
+func (c *RepoCursor) String() string {
+	if c == nil || c.ProjectId == "" || c.Provider == "" || c.RepoId <= 0 {
+		return ""
+	}
+	key := strings.Join([]string{c.ProjectId, c.Provider, strconv.Itoa(int(c.RepoId))}, cursorDelimiter)
+	return EncodeValue(key)
+}
+
+// NewRepoCursor creates a new RepoCursor from an encoded cursor
+func NewRepoCursor(encodedCursor string) (*RepoCursor, error) {
+	if encodedCursor == "" {
+		return &RepoCursor{}, nil
+	}
+
+	cursor, err := DecodeValue(encodedCursor)
+	if err != nil {
+		return nil, err
+	}
+
+	parts := strings.Split(cursor, cursorDelimiter)
+	if len(parts) != 3 {
+		return nil, fmt.Errorf("invalid cursor: %s", encodedCursor)
+	}
+	parsedRepoId, err := strconv.ParseInt(parts[2], 10, 32)
+	if err != nil {
+		return nil, err
+	}
+
+	return &RepoCursor{
+		ProjectId: parts[0],
+		Provider:  parts[1],
+		RepoId:    int32(parsedRepoId),
+	}, nil
+}
+
+// EncodeValue encodes a string into a base64 encoded string
+func EncodeValue(value string) string {
+	return base64.StdEncoding.EncodeToString([]byte(value))
+}
+
+// DecodeValue decodes a base64 encoded string into a string
+func DecodeValue(value string) (string, error) {
+	decoded, err := base64.StdEncoding.DecodeString(value)
+	if err != nil {
+		return "", fmt.Errorf("error decoding cursor: %w", err)
+	}
+	return string(decoded), nil
+}

internal/controlplane/cursor_test.go

@@ -0,0 +1,162 @@
+// Copyright 2024 Stacklok, Inc
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlplane
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestEncodeDecodeValue(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name      string
+		input     string
+		expectErr bool
+	}{
+		{
+			name:      "non-empty string",
+			input:     "testCursor",
+			expectErr: false,
+		},
+		{
+			name:      "empty string",
+			input:     "",
+			expectErr: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		tc := tc
+
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			encoded := EncodeValue(tc.input)
+			decoded, err := DecodeValue(encoded)
+
+			if tc.expectErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tc.input, decoded)
+			}
+		})
+	}
+}
+
+func TestEncodeDecodeCursor(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name              string
+		repoCursor        RepoCursor
+		expectEmptyCursor bool
+	}{
+		{
+			name:              "all inputs valid",
+			repoCursor:        RepoCursor{ProjectId: "3b241101-e2bb-4255-8caf-4136c566a964", Provider: "testProvider", RepoId: 123},
+			expectEmptyCursor: false,
+		},
+		{
+			name:              "empty projectId",
+			repoCursor:        RepoCursor{ProjectId: "", Provider: "testProvider", RepoId: 123},
+			expectEmptyCursor: true,
+		},
+		{
+			name:              "empty provider",
+			repoCursor:        RepoCursor{ProjectId: "3b241101-e2bb-4255-8caf-4136c566a964", Provider: "", RepoId: 123},
+			expectEmptyCursor: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		tc := tc
+
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			encodedCursor := tc.repoCursor.String()
+
+			if tc.expectEmptyCursor {
+				require.Equal(t, "", encodedCursor)
+			} else {
+				decodedRepoCursor, err := NewRepoCursor(encodedCursor)
+
+				require.NoError(t, err)
+				require.Equal(t, tc.repoCursor.ProjectId, decodedRepoCursor.ProjectId)
+				require.Equal(t, tc.repoCursor.Provider, decodedRepoCursor.Provider)
+				require.Equal(t, tc.repoCursor.RepoId, decodedRepoCursor.RepoId)
+			}
+		})
+	}
+}
+
+func TestDecodeListRepositoriesByProjectIDCursor(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name                string
+		cursor              string
+		expectedRepoCursor  *RepoCursor
+		expectErrorDecoding bool
+	}{
+		{
+			name:                "compliant input",
+			cursor:              EncodeValue("3b241101-e2bb-4255-8caf-4136c566a964,testProvider,123"),
+			expectedRepoCursor:  &RepoCursor{ProjectId: "3b241101-e2bb-4255-8caf-4136c566a964", Provider: "testProvider", RepoId: 123},
+			expectErrorDecoding: false,
+		},
+		{
+			name:                "non-compliant input",
+			cursor:              EncodeValue("nonCompliantInput"),
+			expectedRepoCursor:  nil,
+			expectErrorDecoding: true,
+		},
+		{
+			name:                "non-compliant 64 bit input",
+			cursor:              EncodeValue("3b241101-e2bb-4255-8caf-4136c566a964,testProvider,12345678901234567890"),
+			expectedRepoCursor:  nil,
+			expectErrorDecoding: true,
+		},
+		{
+			name:                "empty input",
+			cursor:              "",
+			expectedRepoCursor:  &RepoCursor{},
+			expectErrorDecoding: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		tc := tc
+
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			decodedRepoCursor, err := NewRepoCursor(tc.cursor)
+
+			if tc.expectErrorDecoding {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tc.expectedRepoCursor.ProjectId, decodedRepoCursor.ProjectId)
+				require.Equal(t, tc.expectedRepoCursor.Provider, decodedRepoCursor.Provider)
+				require.Equal(t, tc.expectedRepoCursor.RepoId, decodedRepoCursor.RepoId)
+			}
+		})
+	}
+}

internal/controlplane/handlers_repositories.go

@@ -34,6 +34,9 @@ import (
 	pb "github.com/stacklok/minder/pkg/api/protobuf/go/minder/v1"
 )
 
+// maxFetchLimit is the maximum number of repositories that can be fetched from the database in one call
+const maxFetchLimit = 100
+
 // RegisterRepository adds repositories to the database and registers a webhook
 // Once a user had enrolled in a project (they have a valid token), they can register
 // repositories to be monitored by the minder by provisioning a webhook on the
@@ -142,7 +145,6 @@ func (s *Server) RegisterRepository(ctx context.Context,
 // ListRepositories returns a list of repositories for a given project
 // This function will typically be called by the client to get a list of
 // repositories that are registered present in the minder database
-// The API is called with a project id, limit and offset
 func (s *Server) ListRepositories(ctx context.Context,
 	in *pb.ListRepositoriesRequest) (*pb.ListRepositoriesResponse, error) {
 	projectID, err := getProjectFromRequestOrDefault(ctx, in)
@@ -160,9 +162,30 @@ func (s *Server) ListRepositories(ctx context.Context,
 		return nil, providerError(err)
 	}
 
+	reqRepoCursor, err := NewRepoCursor(in.GetCursor())
+	if err != nil {
+		return nil, util.UserVisibleError(codes.InvalidArgument, err.Error())
+	}
+
+	repoId := sql.NullInt32{Valid: false, Int32: 0}
+	if reqRepoCursor.ProjectId == projectID.String() && reqRepoCursor.Provider == provider.Name {
+		repoId = sql.NullInt32{Valid: true, Int32: reqRepoCursor.RepoId}
+	}
+
+	limit := sql.NullInt32{Valid: false, Int32: 0}
+	reqLimit := in.GetLimit()
+	if reqLimit > 0 {
+		if reqLimit > maxFetchLimit {
+			return nil, util.UserVisibleError(codes.InvalidArgument, "limit too high, max is %d", maxFetchLimit)
+		}
+		limit = sql.NullInt32{Valid: true, Int32: reqLimit + 1}
+	}
+
 	repos, err := s.store.ListRepositoriesByProjectID(ctx, db.ListRepositoriesByProjectIDParams{
 		Provider:  provider.Name,
 		ProjectID: projectID,
+		RepoID:    repoId,
+		Limit:     limit,
 	})
 
 	if err != nil {
@@ -184,7 +207,21 @@ func (s *Server) ListRepositories(ctx context.Context,
 		results = append(results, r)
 	}
 
+	var respRepoCursor *RepoCursor
+	if limit.Valid && len(repos) == int(limit.Int32) {
+		lastRepo := repos[len(repos)-1]
+		respRepoCursor = &RepoCursor{
+			ProjectId: projectID.String(),
+			Provider:  provider.Name,
+			RepoId:    lastRepo.RepoID,
+		}
+
+		// remove the (limit + 1)th element from the results
+		results = results[:len(results)-1]
+	}
+
 	resp.Results = results
+	resp.Cursor = respRepoCursor.String()
 
 	return &resp, nil
 }