Added pagination for operations (#639)

* Pagination for operations

* Added integration test.

* Skip the test when not in integration.

* feedback

doc/swagger.yaml

@@ -1802,6 +1802,10 @@ definitions:
     type: object
   pkg_server_router.ListOperationsResponse:
     properties:
+      nextPageToken:
+        description: Pagination token to retrieve the next page of results. If the
+          value is "", it means no further results for the request.
+        type: string
       operations:
         items:
           $ref: '#/definitions/pkg_server_router.Operation'
@@ -3593,6 +3597,16 @@ paths:
         in: query
         name: filter
         type: string
+      - description: Hint to the server of the maximum elements to return. More may
+          be returned. When not set, the server will return all elements.
+        in: query
+        name: pageSize
+        type: number
+      - description: Used to indicate to the server to return a specific page of the
+          list results. Must match a previous requests' `nextPageToken`.
+        in: query
+        name: pageToken
+        type: string
       produces:
       - application/json
       responses:

integration/presentation_exchange_integration_test.go

@@ -268,3 +268,24 @@ func TestSubmissionFlowExternalCredential(t *testing.T) {
 	s, _ := getJSONElement(reviewOutput, "$")
 	assert.Equal(t, s, opResponse)
 }
+
+func TestListOperationsWithPagination(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	// This test simply ensures that we can retrieve all operations from a parent using pagination.
+	schemasPage, err := get(endpoint + version + "operations?parent=presentations/submissions&pageSize=1")
+	assert.NoError(t, err)
+
+	nextPageToken, err := getJSONElement(schemasPage, "$.nextPageToken")
+	assert.NoError(t, err)
+
+	for nextPageToken != "" {
+		schemasPage, err := get(endpoint + version + "operations?parent=presentations/submissions&pageSize=1&pageToken=" + nextPageToken)
+		assert.NoError(t, err)
+
+		nextPageToken, err = getJSONElement(schemasPage, "$.nextPageToken")
+		assert.NoError(t, err)
+	}
+}

pkg/server/router/operation.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/pkg/errors"
+	"github.com/tbd54566975/ssi-service/pkg/server/pagination"
 	"go.einride.tech/aip/filtering"
 
 	"github.com/tbd54566975/ssi-service/pkg/server/framework"
@@ -104,7 +105,7 @@ const (
 
 const FilterCharacterLimit = 1024
 
-func (r listOperationsRequest) toServiceRequest() (operation.ListOperationsRequest, error) {
+func (r listOperationsRequest) toServiceRequest(pageRequest pagination.PageRequest) (operation.ListOperationsRequest, error) {
 	var opReq operation.ListOperationsRequest
 	opReq.Parent = r.Parent
 
@@ -130,11 +131,15 @@ func (r listOperationsRequest) toServiceRequest() (operation.ListOperationsReque
 		return opReq, errors.Wrap(err, "parsing filter")
 	}
 	opReq.Filter = filter
+	opReq.PageRequest = &pageRequest
 	return opReq, nil
 }
 
 type ListOperationsResponse struct {
 	Operations []Operation `json:"operations"`
+
+	// Pagination token to retrieve the next page of results. If the value is "", it means no further results for the request.
+	NextPageToken string `json:"nextPageToken"`
 }
 
 // ListOperations godoc
@@ -144,15 +149,17 @@ type ListOperationsResponse struct {
 //	@Tags			Operations
 //	@Accept			json
 //	@Produce		json
-//	@Param			parent	query		string					false	"The name of the parent's resource. For example: `?parent=/presentation/submissions`"
-//	@Param			filter	query		string					false	"A standard filter expression conforming to https://google.aip.dev/160. For example: `?filter=done="true"`"
-//	@Success		200		{object}	ListOperationsResponse	"OK"
-//	@Failure		400		{string}	string					"Bad request"
-//	@Failure		500		{string}	string					"Internal server error"
+//	@Param			parent		query		string					false	"The name of the parent's resource. For example: `?parent=/presentation/submissions`"
+//	@Param			filter		query		string					false	"A standard filter expression conforming to https://google.aip.dev/160. For example: `?filter=done="true"`"
+//	@Param			pageSize	query		number					false	"Hint to the server of the maximum elements to return. More may be returned. When not set, the server will return all elements."
+//	@Param			pageToken	query		string					false	"Used to indicate to the server to return a specific page of the list results. Must match a previous requests' `nextPageToken`."
+//	@Success		200			{object}	ListOperationsResponse	"OK"
+//	@Failure		400			{string}	string					"Bad request"
+//	@Failure		500			{string}	string					"Internal server error"
 //	@Router			/v1/operations [get]
 func (o OperationRouter) ListOperations(c *gin.Context) {
-	parentParam := framework.GetParam(c, ParentParam)
-	filterParam := framework.GetParam(c, FilterParam)
+	parentParam := framework.GetQueryValue(c, ParentParam)
+	filterParam := framework.GetQueryValue(c, FilterParam)
 	var request listOperationsRequest
 	if parentParam != nil {
 		unescaped, err := url.QueryUnescape(*parentParam)
@@ -179,23 +186,30 @@ func (o OperationRouter) ListOperations(c *gin.Context) {
 		return
 	}
 
-	req, err := request.toServiceRequest()
+	var pageRequest pagination.PageRequest
+	if pagination.ParsePaginationQueryValues(c, &pageRequest) {
+		return
+	}
+
+	req, err := request.toServiceRequest(pageRequest)
 	if err != nil {
 		framework.LoggingRespondErrWithMsg(c, err, invalidGetOperationsErr, http.StatusBadRequest)
 		return
 	}
-
-	ops, err := o.service.ListOperations(c, req)
+	listOpsResp, err := o.service.ListOperations(c, req)
 	if err != nil {
 		errMsg := "getting operations from service"
 		framework.LoggingRespondErrWithMsg(c, err, errMsg, http.StatusInternalServerError)
 		return
 	}
 
-	resp := ListOperationsResponse{Operations: make([]Operation, 0, len(ops.Operations))}
-	for _, op := range ops.Operations {
+	resp := ListOperationsResponse{Operations: make([]Operation, 0, len(listOpsResp.Operations))}
+	for _, op := range listOpsResp.Operations {
 		resp.Operations = append(resp.Operations, routerModel(op))
 	}
+	if pagination.MaybeSetNextPageToken(c, listOpsResp.NextPageToken, &resp.NextPageToken) {
+		return
+	}
 	framework.Respond(c, resp, http.StatusOK)
 }
 

pkg/service/operation/model.go

@@ -2,6 +2,7 @@ package operation
 
 import (
 	"github.com/TBD54566975/ssi-sdk/util"
+	"github.com/tbd54566975/ssi-service/pkg/server/pagination"
 	"go.einride.tech/aip/filtering"
 )
 
@@ -17,16 +18,18 @@ type Operation struct {
 }
 
 type ListOperationsRequest struct {
-	Parent string `validate:"required"`
-	Filter filtering.Filter
+	Parent      string `validate:"required"`
+	Filter      filtering.Filter
+	PageRequest *pagination.PageRequest
 }
 
 func (r ListOperationsRequest) Validate() error {
 	return util.NewValidator().Struct(r)
 }
 
 type ListOperationsResponse struct {
-	Operations []Operation
+	Operations    []Operation
+	NextPageToken string
 }
 
 type GetOperationRequest struct {

pkg/service/operation/service.go

@@ -48,13 +48,16 @@ func (s Service) ListOperations(ctx context.Context, request ListOperationsReque
 		return nil, errors.Wrap(err, "invalid request")
 	}
 
-	ops, err := s.storage.ListOperations(ctx, request.Parent, request.Filter)
+	ops, err := s.storage.ListOperations(ctx, request.Parent, request.Filter, request.PageRequest.ToServicePage())
 	if err != nil {
 		return nil, errors.Wrap(err, "fetching ops from storage")
 	}
 
-	resp := ListOperationsResponse{Operations: make([]Operation, len(ops))}
-	for i, op := range ops {
+	resp := ListOperationsResponse{
+		Operations:    make([]Operation, len(ops.StoredOperations)),
+		NextPageToken: ops.NextPageToken,
+	}
+	for i, op := range ops.StoredOperations {
 		op := op
 		newOp, err := ServiceModel(op)
 		if err != nil {

pkg/service/operation/storage.go

@@ -8,6 +8,7 @@ import (
 	"github.com/goccy/go-json"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"github.com/tbd54566975/ssi-service/pkg/service/common"
 	"go.einride.tech/aip/filtering"
 
 	"github.com/tbd54566975/ssi-service/pkg/service/operation/credential"
@@ -101,8 +102,10 @@ func (s Storage) GetOperation(ctx context.Context, id string) (opstorage.StoredO
 	return stored, nil
 }
 
-func (s Storage) ListOperations(ctx context.Context, parent string, filter filtering.Filter) ([]opstorage.StoredOperation, error) {
-	operations, err := s.db.ReadAll(ctx, namespace.FromParent(parent))
+func (s Storage) ListOperations(ctx context.Context, parent string, filter filtering.Filter, page *common.Page) (*opstorage.StoredOperations, error) {
+	token, size := page.ToStorageArgs()
+
+	operations, nextPageToken, err := s.db.ReadPage(ctx, namespace.FromParent(parent), token, size)
 	if err != nil {
 		return nil, sdkutil.LoggingErrorMsgf(err, "could not get all operations")
 	}
@@ -123,7 +126,10 @@ func (s Storage) ListOperations(ctx context.Context, parent string, filter filte
 			stored = append(stored, nextOp)
 		}
 	}
-	return stored, nil
+	return &opstorage.StoredOperations{
+		StoredOperations: stored,
+		NextPageToken:    nextPageToken,
+	}, nil
 }
 
 func (s Storage) DeleteOperation(ctx context.Context, id string) error {

pkg/service/operation/storage/storage.go

@@ -6,6 +6,11 @@ import (
 	"go.einride.tech/aip/filtering"
 )
 
+type StoredOperations struct {
+	StoredOperations []StoredOperation
+	NextPageToken    string
+}
+
 type StoredOperation struct {
 	ID string `json:"id"`