Enable filtering, sorting, and paging for federation nodes

lib/js/src/api-clients/federation.ts

@@ -138,6 +138,10 @@ export default class Federation {
     const {
       query,
       status,
+      limit,
+      incTotal,
+      pageCursor,
+      sort,
     } = (a as KV) || {}
     const cfg: AxiosRequestConfig = {
       ...extra,
@@ -147,6 +151,10 @@ export default class Federation {
     cfg.params = {
       query,
       status,
+      limit,
+      incTotal,
+      pageCursor,
+      sort,
     }
 
     return this.api().request(cfg).then(result => stdResolve(result))

server/federation/node.cue

@@ -7,8 +7,8 @@ import (
 node: {
 	features: {
 		labels: false
-		paging: false
-		sorting: false
+		paging: true
+		sorting: true
 	}
 
 	model: {

server/federation/rest.yaml

@@ -50,6 +50,19 @@ endpoints:
             - name: status
               type: string
               title: Filter by status
+            - type: uint
+              name: limit
+              title: Limit
+            - type: bool
+              name: incTotal
+              title: Include total counter
+            - type: string
+              name: pageCursor
+              title: Page cursor
+            - type: string
+              name: sort
+              title: Sort items
+
       - name: create
         method: POST
         title: Create a new federation node

server/federation/rest/node.go

@@ -7,6 +7,7 @@ import (
 	"github.com/cortezaproject/corteza/server/federation/service"
 	"github.com/cortezaproject/corteza/server/federation/types"
 	"github.com/cortezaproject/corteza/server/pkg/api"
+	"github.com/cortezaproject/corteza/server/pkg/filter"
 )
 
 type (
@@ -50,10 +51,25 @@ func (Node) New() *Node {
 }
 
 func (ctrl Node) Search(ctx context.Context, r *request.NodeSearch) (interface{}, error) {
-	set, f, err := ctrl.svcNode.Search(ctx, types.NodeFilter{
-		Query:  r.Query,
-		Status: r.Status,
-	})
+	var (
+		err error
+		f   = types.NodeFilter{
+			Query:  r.Query,
+			Status: r.Status,
+		}
+	)
+
+	if f.Paging, err = filter.NewPaging(r.Limit, r.PageCursor); err != nil {
+		return nil, err
+	}
+
+	f.IncTotal = r.IncTotal
+
+	if f.Sorting, err = filter.NewSorting(r.Sort); err != nil {
+		return nil, err
+	}
+
+	set, f, err := ctrl.svcNode.Search(ctx, f)
 
 	return ctrl.makeFilterPayload(ctx, set, f, err)
 }

server/federation/rest/request/node.go

@@ -43,6 +43,26 @@ type (
 		//
 		// Filter by status
 		Status string
+
+		// Limit GET parameter
+		//
+		// Limit
+		Limit uint
+
+		// IncTotal GET parameter
+		//
+		// Include total counter
+		IncTotal bool
+
+		// PageCursor GET parameter
+		//
+		// Page cursor
+		PageCursor string
+
+		// Sort GET parameter
+		//
+		// Sort items
+		Sort string
 	}
 
 	NodeCreate struct {
@@ -152,8 +172,12 @@ func NewNodeSearch() *NodeSearch {
 // Auditable returns all auditable/loggable parameters
 func (r NodeSearch) Auditable() map[string]interface{} {
 	return map[string]interface{}{
-		"query":  r.Query,
-		"status": r.Status,
+		"query":      r.Query,
+		"status":     r.Status,
+		"limit":      r.Limit,
+		"incTotal":   r.IncTotal,
+		"pageCursor": r.PageCursor,
+		"sort":       r.Sort,
 	}
 }
 
@@ -167,6 +191,26 @@ func (r NodeSearch) GetStatus() string {
 	return r.Status
 }
 
+// Auditable returns all auditable/loggable parameters
+func (r NodeSearch) GetLimit() uint {
+	return r.Limit
+}
+
+// Auditable returns all auditable/loggable parameters
+func (r NodeSearch) GetIncTotal() bool {
+	return r.IncTotal
+}
+
+// Auditable returns all auditable/loggable parameters
+func (r NodeSearch) GetPageCursor() string {
+	return r.PageCursor
+}
+
+// Auditable returns all auditable/loggable parameters
+func (r NodeSearch) GetSort() string {
+	return r.Sort
+}
+
 // Fill processes request and fills internal variables
 func (r *NodeSearch) Fill(req *http.Request) (err error) {
 
@@ -186,6 +230,30 @@ func (r *NodeSearch) Fill(req *http.Request) (err error) {
 				return err
 			}
 		}
+		if val, ok := tmp["limit"]; ok && len(val) > 0 {
+			r.Limit, err = payload.ParseUint(val[0]), nil
+			if err != nil {
+				return err
+			}
+		}
+		if val, ok := tmp["incTotal"]; ok && len(val) > 0 {
+			r.IncTotal, err = payload.ParseBool(val[0]), nil
+			if err != nil {
+				return err
+			}
+		}
+		if val, ok := tmp["pageCursor"]; ok && len(val) > 0 {
+			r.PageCursor, err = val[0], nil
+			if err != nil {
+				return err
+			}
+		}
+		if val, ok := tmp["sort"]; ok && len(val) > 0 {
+			r.Sort, err = val[0], nil
+			if err != nil {
+				return err
+			}
+		}
 	}
 
 	return err

server/federation/types/node.go

@@ -49,6 +49,8 @@ type (
 
 		Deleted filter.State `json:"deleted"`
 
-		Limit uint
+		// Standard helpers for paging and sorting
+		filter.Sorting
+		filter.Paging
 	}
 )

server/store/adapters/rdbms/rdbms.gen.go

@@ -13545,14 +13545,198 @@ func (s *Store) TruncateFederationNodes(ctx context.Context) error {
 // This function is auto-generated
 func (s *Store) SearchFederationNodes(ctx context.Context, f federationType.NodeFilter) (set federationType.NodeSet, _ federationType.NodeFilter, err error) {
 
-	set, _, err = s.QueryFederationNodes(ctx, f)
+	// Cleanup unwanted cursor values (only relevant is f.PageCursor, next&prev are reset and returned)
+	f.PrevPage, f.NextPage = nil, nil
+
+	if f.PageCursor != nil {
+		if f.IncPageNavigation || f.IncTotal {
+			return nil, f, fmt.Errorf("not allowed to fetch page navigation or total item count with page cursor")
+		}
+
+		// Page cursor exists; we need to validate it against used sort
+		// To cover the case when paging cursor is set but sorting is empty, we collect the sorting instructions
+		// from the cursor.
+		// This (extracted sorting info) is then returned as part of response
+		if f.Sort, err = f.PageCursor.Sort(f.Sort); err != nil {
+			return
+		}
+	}
+
+	// Make sure results are always sorted at least by primary keys
+	if f.Sort.Get("id") == nil {
+		f.Sort = append(f.Sort, &filter.SortExpr{
+			Column:     "id",
+			Descending: f.Sort.LastDescending(),
+		})
+	}
+
+	// Cloned sorting instructions for the actual sorting
+	// Original are passed to the etchFullPageOfFederationNodes fn used for cursor creation;
+	// direction information it MUST keep the initial
+	sort := f.Sort.Clone()
+
+	// When cursor for a previous page is used it's marked as reversed
+	// This tells us to flip the descending flag on all used sort keys
+	if f.PageCursor != nil && f.PageCursor.ROrder {
+		sort.Reverse()
+	}
+
+	set, f.PrevPage, f.NextPage, err = s.fetchFullPageOfFederationNodes(ctx, f, sort)
+
+	f.PageCursor = nil
 	if err != nil {
 		return nil, f, err
 	}
 
+	if f.IncTotal {
+		// Calc total from the number of items fetched
+		// even if we do build the page navigation
+		f.Total = uint(len(set))
+
+		if f.Limit > 0 && uint(len(set)) == f.Limit {
+			// there are fewer items fetched then requested limit
+			limit := f.Limit
+			f.Limit = 0
+			var navSet federationType.NodeSet
+			if navSet, _, _, err = s.fetchFullPageOfFederationNodes(ctx, f, sort); err != nil {
+				return
+			} else {
+				f.Total = uint(len(navSet))
+				f.Limit = limit
+			}
+		}
+	}
+
 	return set, f, nil
 }
 
+// fetchFullPageOfFederationNodes collects all requested results.
+//
+// Function applies:
+//  - cursor conditions (where ...)
+//  - limit
+//
+// Main responsibility of this function is to perform additional sequential queries in case when not enough results
+// are collected due to failed check on a specific row (by check fn).
+//
+// Function then moves cursor to the last item fetched
+//
+// This function is auto-generated
+func (s *Store) fetchFullPageOfFederationNodes(
+	ctx context.Context,
+	filter federationType.NodeFilter,
+	sort filter.SortExprSet,
+) (set []*federationType.Node, prev, next *filter.PagingCursor, err error) {
+	var (
+		aux []*federationType.Node
+
+		// When cursor for a previous page is used it's marked as reversed
+		// This tells us to flip the descending flag on all used sort keys
+		reversedOrder = filter.PageCursor != nil && filter.PageCursor.ROrder
+
+		// Copy no. of required items to limit
+		// Limit will change when doing subsequent queries to fill
+		// the set with all required items
+		limit = filter.Limit
+
+		reqItems = filter.Limit
+
+		// cursor to prev. page is only calculated when cursor is used
+		hasPrev = filter.PageCursor != nil
+
+		// next cursor is calculated when there are more pages to come
+		hasNext bool
+
+		tryFilter federationType.NodeFilter
+	)
+
+	set = make([]*federationType.Node, 0, DefaultSliceCapacity)
+
+	for try := 0; try < MaxRefetches; try++ {
+		// Copy filter & apply custom sorting that might be affected by cursor
+		tryFilter = filter
+		tryFilter.Sort = sort
+
+		if limit > 0 {
+			// fetching + 1 to peak ahead if there are more items
+			// we can fetch (next-page cursor)
+			tryFilter.Limit = limit + 1
+		}
+
+		if aux, hasNext, err = s.QueryFederationNodes(ctx, tryFilter); err != nil {
+			return nil, nil, nil, err
+		}
+
+		if len(aux) == 0 {
+			// nothing fetched
+			break
+		}
+
+		// append fetched items
+		set = append(set, aux...)
+
+		if reqItems == 0 || !hasNext {
+			// no max requested items specified, break out
+			break
+		}
+
+		collected := uint(len(set))
+
+		if reqItems > collected {
+			// not enough items fetched, try again with adjusted limit
+			limit = reqItems - collected
+
+			if limit < MinEnsureFetchLimit {
+				// In case limit is set very low and we've missed records in the first fetch,
+				// make sure next fetch limit is a bit higher
+				limit = MinEnsureFetchLimit
+			}
+
+			// Update cursor so that it points to the last item fetched
+			tryFilter.PageCursor = s.collectFederationNodeCursorValues(set[collected-1], filter.Sort...)
+
+			// Copy reverse flag from sorting
+			tryFilter.PageCursor.LThen = filter.Sort.Reversed()
+			continue
+		}
+
+		if reqItems < collected {
+			set = set[:reqItems]
+		}
+
+		break
+	}
+
+	collected := len(set)
+
+	if collected == 0 {
+		return nil, nil, nil, nil
+	}
+
+	if reversedOrder {
+		// Fetched set needs to be reversed because we've forced a descending order to get the previous page
+		for i, j := 0, collected-1; i < j; i, j = i+1, j-1 {
+			set[i], set[j] = set[j], set[i]
+		}
+
+		// when in reverse-order rules on what cursor to return change
+		hasPrev, hasNext = hasNext, hasPrev
+	}
+
+	if hasPrev {
+		prev = s.collectFederationNodeCursorValues(set[0], filter.Sort...)
+		prev.ROrder = true
+		prev.LThen = !filter.Sort.Reversed()
+	}
+
+	if hasNext {
+		next = s.collectFederationNodeCursorValues(set[collected-1], filter.Sort...)
+		next.LThen = filter.Sort.Reversed()
+	}
+
+	return set, prev, next, nil
+}
+
 // QueryFederationNodes queries the database, converts and checks each row and returns collected set
 //
 // With generics, we can remove this per-resource-generated function
@@ -13572,6 +13756,8 @@ func (s *Store) QueryFederationNodes(
 		rows        *sql.Rows
 		count       uint
 		expr, tExpr []goqu.Expression
+
+		sortExpr []exp.OrderedExpression
 	)
 
 	if s.Filters.FederationNode != nil {
@@ -13589,8 +13775,27 @@ func (s *Store) QueryFederationNodes(
 
 	expr = append(expr, tExpr...)
 
+	// paging feature is enabled
+	if f.PageCursor != nil {
+		if tExpr, err = cursorWithSorting(f.PageCursor, s.sortableFederationNodeFields()); err != nil {
+			return
+		} else {
+			expr = append(expr, tExpr...)
+		}
+	}
+
 	query := federationNodeSelectQuery(s.Dialect.GOQU()).Where(expr...)
 
+	// sorting feature is enabled
+	if sortExpr, err = order(f.Sort, s.sortableFederationNodeFields()); err != nil {
+		err = fmt.Errorf("could generate order expression for FederationNode: %w", err)
+		return
+	}
+
+	if len(sortExpr) > 0 {
+		query = query.Order(sortExpr...)
+	}
+
 	if f.Limit > 0 {
 		query = query.Limit(f.Limit)
 	}
@@ -13645,7 +13850,7 @@ func (s *Store) QueryFederationNodes(
 		set = append(set, res)
 	}
 
-	return set, false, err
+	return set, f.Limit > 0 && count >= f.Limit, err
 
 }