Implement iterator chunking

Membership iterator doesn't chunk the output as the size will
always be small.

compose/automation/records_handler.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+
 	"github.com/cortezaproject/corteza-server/compose/types"
 	. "github.com/cortezaproject/corteza-server/pkg/expr"
 	"github.com/cortezaproject/corteza-server/pkg/filter"
@@ -33,9 +34,18 @@ type (
 	}
 
 	recordSetIterator struct {
-		ptr    int
-		set    types.RecordSet
+		// Item buffer, current item pointer, and total items traversed
+		ptr    uint
+		buffer types.RecordSet
+		total  uint
+
+		// When filter limit is set, this constraints it
+		iterLimit    uint
+		useIterLimit bool
+
+		// Item loader for additional chunks
 		filter types.RecordFilter
+		loader func() error
 	}
 
 	recordLookup interface {
@@ -167,11 +177,36 @@ func (h recordsHandler) each(ctx context.Context, args *recordsEachArgs) (out wf
 	}
 
 	if args.hasLimit {
-		f.Limit = uint(args.Limit)
+		i.useIterLimit = true
+		i.iterLimit = uint(args.Limit)
+
+		if args.Limit > uint64(wfexec.MaxIteratorBufferSize) {
+			f.Limit = wfexec.MaxIteratorBufferSize
+		}
+		i.iterLimit = uint(args.Limit)
+	} else {
+		f.Limit = wfexec.MaxIteratorBufferSize
 	}
 
-	i.set, i.filter, err = h.rec.Find(ctx, f)
-	return i, err
+	i.filter = f
+	i.loader = func() (err error) {
+		// Edgecase
+		if i.filter.PageCursor != nil && i.filter.NextPage == nil {
+			return
+		}
+
+		i.total += i.ptr
+		i.ptr = 0
+
+		i.filter.PageCursor = i.filter.NextPage
+		i.filter.NextPage = nil
+		i.buffer, i.filter, err = h.rec.Find(ctx, i.filter)
+
+		return
+	}
+
+	// Initial load
+	return i, i.loader()
 }
 
 func (h recordsHandler) validate(ctx context.Context, args *recordsValidateArgs) (*recordsValidateResults, error) {
@@ -312,16 +347,22 @@ func (h recordsHandler) fetchEdge(ctx context.Context, args interface{}, first b
 }
 
 func (i *recordSetIterator) More(context.Context, *Vars) (bool, error) {
-	return i.ptr < len(i.set), nil
+	return wfexec.GenericResourceNextCheck(i.useIterLimit, i.ptr, uint(len(i.buffer)), i.total, i.iterLimit, i.filter.NextPage != nil), nil
 }
 
 func (i *recordSetIterator) Start(context.Context, *Vars) error { i.ptr = 0; return nil }
 
-func (i *recordSetIterator) Next(context.Context, *Vars) (*Vars, error) {
-	out := &Vars{}
-	out.Set("record", Must(NewComposeRecord(i.set[i.ptr])))
-	out.Set("index", i.ptr)
-	out.Set("total", i.filter.Total)
+func (i *recordSetIterator) Next(context.Context, *Vars) (out *Vars, err error) {
+	if len(i.buffer)-int(i.ptr) <= 0 {
+		if err = i.loader(); err != nil {
+			panic(err)
+		}
+	}
+
+	out = &Vars{}
+	out.Set("record", Must(NewComposeRecord(i.buffer[i.ptr])))
+	out.Set("index", Must(NewInteger(i.total+i.ptr)))
+	out.Set("total", Must(NewInteger(i.filter.Total)))
 
 	i.ptr++
 	return out, nil

pkg/wfexec/iterator.go

@@ -2,6 +2,7 @@ package wfexec
 
 import (
 	"context"
+
 	"github.com/cortezaproject/corteza-server/pkg/expr"
 )
 
@@ -45,6 +46,14 @@ type (
 	}
 )
 
+const (
+	DefaultMaxIteratorBufferSize uint = 1000
+)
+
+var (
+	MaxIteratorBufferSize uint = DefaultMaxIteratorBufferSize
+)
+
 // GenericIterator creates a wrapper around IteratorHandler and
 // returns genericIterator that implements Iterator interface
 func GenericIterator(iter, next, exit Step, h IteratorHandler) Iterator {
@@ -89,3 +98,18 @@ func (i *genericIterator) Next(ctx context.Context, scope *expr.Vars) (next Step
 
 	return
 }
+
+func GenericResourceNextCheck(useLimit bool, ptr, buffSize, total, limit uint, hasMore bool) bool {
+	// if we can go more (inverted)...
+	if useLimit && ptr+total >= limit {
+		return false
+	}
+
+	// if we have some buffer left...
+	if ptr < buffSize {
+		return true
+	}
+
+	// if we can get more...
+	return hasMore
+}

system/automation/expr_types.go

@@ -1,6 +1,7 @@
 package automation
 
 import (
+	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -31,6 +32,11 @@ func CastToUser(val interface{}) (out *types.User, err error) {
 	switch val := expr.UntypedValue(val).(type) {
 	case *types.User:
 		return val, nil
+	case map[string]interface{}:
+		out = &types.User{}
+		m, _ := json.Marshal(val)
+		_ = json.Unmarshal(m, out)
+		return out, nil
 	case nil:
 		return &types.User{}, nil
 	default:
@@ -50,6 +56,11 @@ func CastToRole(val interface{}) (out *types.Role, err error) {
 	switch val := expr.UntypedValue(val).(type) {
 	case *types.Role:
 		return val, nil
+	case map[string]interface{}:
+		out = &types.Role{}
+		m, _ := json.Marshal(val)
+		_ = json.Unmarshal(m, out)
+		return out, nil
 	case nil:
 		return &types.Role{}, nil
 	default:
@@ -69,6 +80,11 @@ func CastToTemplate(val interface{}) (out *types.Template, err error) {
 	switch val := expr.UntypedValue(val).(type) {
 	case *types.Template:
 		return val, nil
+	case map[string]interface{}:
+		out = &types.Template{}
+		m, _ := json.Marshal(val)
+		_ = json.Unmarshal(m, out)
+		return out, nil
 	case nil:
 		return &types.Template{}, nil
 	default:

system/automation/roles_handler.go

@@ -38,9 +38,18 @@ type (
 	}
 
 	roleSetIterator struct {
-		ptr    int
-		set    types.RoleSet
+		// Item buffer, current item pointer, and total items traversed
+		ptr    uint
+		buffer types.RoleSet
+		total  uint
+
+		// When filter limit is set, this constraints it
+		iterLimit    uint
+		useIterLimit bool
+
+		// Item loader for additional chunks
 		filter types.RoleFilter
+		loader func() error
 	}
 
 	roleLookup interface {
@@ -120,7 +129,7 @@ func (h rolesHandler) eachMember(ctx context.Context, args *rolesEachMemberArgs)
 	}
 
 	if len(mm) == 0 {
-		i.set = []*types.User{}
+		i.buffer = []*types.User{}
 		i.filter = types.UserFilter{}
 		return i, nil
 	}
@@ -130,7 +139,7 @@ func (h rolesHandler) eachMember(ctx context.Context, args *rolesEachMemberArgs)
 	for i, m := range mm {
 		uu[i] = m.UserID
 	}
-	i.set, i.filter, err = h.uSvc.Find(ctx, types.UserFilter{
+	i.buffer, i.filter, err = h.uSvc.Find(ctx, types.UserFilter{
 		UserID: uu,
 	})
 	return i, err
@@ -274,11 +283,36 @@ func (h rolesHandler) each(ctx context.Context, args *rolesEachArgs) (out wfexec
 	}
 
 	if args.hasLimit {
-		f.Limit = uint(args.Limit)
+		i.useIterLimit = true
+		i.iterLimit = uint(args.Limit)
+
+		if args.Limit > uint64(wfexec.MaxIteratorBufferSize) {
+			f.Limit = wfexec.MaxIteratorBufferSize
+		}
+		i.iterLimit = uint(args.Limit)
+	} else {
+		f.Limit = wfexec.MaxIteratorBufferSize
 	}
 
-	i.set, i.filter, err = h.rSvc.Find(ctx, f)
-	return i, err
+	i.filter = f
+	i.loader = func() (err error) {
+		// Edgecase
+		if i.filter.PageCursor != nil && i.filter.NextPage == nil {
+			return
+		}
+
+		i.total += i.ptr
+		i.ptr = 0
+
+		i.filter.PageCursor = i.filter.NextPage
+		i.filter.NextPage = nil
+		i.buffer, i.filter, err = h.rSvc.Find(ctx, i.filter)
+
+		return
+	}
+
+	// Initial load
+	return i, i.loader()
 }
 
 func (h rolesHandler) create(ctx context.Context, args *rolesCreateArgs) (results *rolesCreateResults, err error) {
@@ -357,16 +391,23 @@ func lookupRole(ctx context.Context, svc roleService, args roleLookup) (*types.R
 }
 
 func (i *roleSetIterator) More(context.Context, *Vars) (bool, error) {
-	return i.ptr < len(i.set), nil
+	a := wfexec.GenericResourceNextCheck(i.useIterLimit, i.ptr, uint(len(i.buffer)), i.total, i.iterLimit, i.filter.NextPage != nil)
+	return a, nil
 }
 
 func (i *roleSetIterator) Start(context.Context, *Vars) error { i.ptr = 0; return nil }
 
-func (i *roleSetIterator) Next(context.Context, *Vars) (*Vars, error) {
-	out := &Vars{}
-	out.Set("role", Must(NewRole(i.set[i.ptr])))
-	out.Set("index", i.ptr)
-	out.Set("total", i.filter.Total)
+func (i *roleSetIterator) Next(context.Context, *Vars) (out *Vars, err error) {
+	if len(i.buffer)-int(i.ptr) <= 0 {
+		if err = i.loader(); err != nil {
+			panic(err)
+		}
+	}
+
+	out = &Vars{}
+	out.Set("role", Must(NewRole(i.buffer[i.ptr])))
+	out.Set("index", Must(NewInteger(i.total+i.ptr)))
+	out.Set("total", Must(NewInteger(i.filter.Total)))
 
 	i.ptr++
 	return out, nil

system/automation/templates_handler.go

@@ -33,9 +33,18 @@ type (
 	}
 
 	templateSetIterator struct {
-		ptr    int
-		set    types.TemplateSet
+		// Item buffer, current item pointer, and total items traversed
+		ptr    uint
+		buffer types.TemplateSet
+		total  uint
+
+		// When filter limit is set, this constraints it
+		iterLimit    uint
+		useIterLimit bool
+
+		// Item loader for additional chunks
 		filter types.TemplateFilter
+		loader func() error
 	}
 
 	templateLookup interface {
@@ -125,11 +134,36 @@ func (h templatesHandler) each(ctx context.Context, args *templatesEachArgs) (ou
 	}
 
 	if args.hasLimit {
-		f.Limit = uint(args.Limit)
+		i.useIterLimit = true
+		i.iterLimit = uint(args.Limit)
+
+		if args.Limit > uint64(wfexec.MaxIteratorBufferSize) {
+			f.Limit = wfexec.MaxIteratorBufferSize
+		}
+		i.iterLimit = uint(args.Limit)
+	} else {
+		f.Limit = wfexec.MaxIteratorBufferSize
 	}
 
-	i.set, i.filter, err = h.tSvc.Search(ctx, f)
-	return i, err
+	i.filter = f
+	i.loader = func() (err error) {
+		// Edgecase
+		if i.filter.PageCursor != nil && i.filter.NextPage == nil {
+			return
+		}
+
+		i.total += i.ptr
+		i.ptr = 0
+
+		i.filter.PageCursor = i.filter.NextPage
+		i.filter.NextPage = nil
+		i.buffer, i.filter, err = h.tSvc.Search(ctx, i.filter)
+
+		return
+	}
+
+	// Initial load
+	return i, i.loader()
 }
 
 func (h templatesHandler) create(ctx context.Context, args *templatesCreateArgs) (results *templatesCreateResults, err error) {
@@ -198,16 +232,22 @@ func (h templatesHandler) render(ctx context.Context, args *templatesRenderArgs)
 }
 
 func (i *templateSetIterator) More(context.Context, *Vars) (bool, error) {
-	return i.ptr < len(i.set), nil
+	return wfexec.GenericResourceNextCheck(i.useIterLimit, i.ptr, uint(len(i.buffer)), i.total, i.iterLimit, i.filter.NextPage != nil), nil
 }
 
 func (i *templateSetIterator) Start(context.Context, *Vars) error { i.ptr = 0; return nil }
 
-func (i *templateSetIterator) Next(context.Context, *Vars) (*Vars, error) {
-	out := &Vars{}
-	out.Set("template", Must(NewTemplate(i.set[i.ptr])))
-	out.Set("index", i.ptr)
-	out.Set("total", i.filter.Total)
+func (i *templateSetIterator) Next(context.Context, *Vars) (out *Vars, err error) {
+	if len(i.buffer)-int(i.ptr) <= 0 {
+		if err = i.loader(); err != nil {
+			panic(err)
+		}
+	}
+
+	out = &Vars{}
+	out.Set("template", Must(NewTemplate(i.buffer[i.ptr])))
+	out.Set("index", Must(NewInteger(i.total+i.ptr)))
+	out.Set("total", Must(NewInteger(i.filter.Total)))
 
 	i.ptr++
 	return out, nil

system/automation/users_handler.go

@@ -34,9 +34,18 @@ type (
 	}
 
 	userSetIterator struct {
-		ptr    int
-		set    types.UserSet
+		// Item buffer, current item pointer, and total items traversed
+		ptr    uint
+		buffer types.UserSet
+		total  uint
+
+		// When filter limit is set, this constraints it
+		iterLimit    uint
+		useIterLimit bool
+
+		// Item loader for additional chunks
 		filter types.UserFilter
+		loader func() error
 	}
 
 	userLookup interface {
@@ -218,16 +227,44 @@ func (h usersHandler) each(ctx context.Context, args *usersEachArgs) (out wfexec
 		}
 	}
 
+	f.IncTotal = args.IncTotal
+	f.IncPageNavigation = args.IncPageNavigation
+
 	if args.hasLabels {
 		f.Labels = args.Labels
 	}
 
 	if args.hasLimit {
-		f.Limit = uint(args.Limit)
+		i.useIterLimit = true
+		i.iterLimit = uint(args.Limit)
+
+		if args.Limit > uint64(wfexec.MaxIteratorBufferSize) {
+			f.Limit = wfexec.MaxIteratorBufferSize
+		}
+		i.iterLimit = uint(args.Limit)
+	} else {
+		f.Limit = wfexec.MaxIteratorBufferSize
 	}
 
-	i.set, i.filter, err = h.uSvc.Find(ctx, f)
-	return i, err
+	i.filter = f
+	i.loader = func() (err error) {
+		// Edgecase
+		if i.filter.PageCursor != nil && i.filter.NextPage == nil {
+			return
+		}
+
+		i.total += i.ptr
+		i.ptr = 0
+
+		i.filter.PageCursor = i.filter.NextPage
+		i.filter.NextPage = nil
+		i.buffer, i.filter, err = h.uSvc.Find(ctx, i.filter)
+
+		return
+	}
+
+	// Initial load
+	return i, i.loader()
 }
 
 func (h usersHandler) create(ctx context.Context, args *usersCreateArgs) (results *usersCreateResults, err error) {
@@ -308,16 +345,22 @@ func lookupUser(ctx context.Context, svc userService, args userLookup) (*types.U
 }
 
 func (i *userSetIterator) More(context.Context, *Vars) (bool, error) {
-	return i.ptr < len(i.set), nil
+	return wfexec.GenericResourceNextCheck(i.useIterLimit, i.ptr, uint(len(i.buffer)), i.total, i.iterLimit, i.filter.NextPage != nil), nil
 }
 
 func (i *userSetIterator) Start(context.Context, *Vars) error { i.ptr = 0; return nil }
 
-func (i *userSetIterator) Next(context.Context, *Vars) (*Vars, error) {
-	out := &Vars{}
-	out.Set("user", Must(NewUser(i.set[i.ptr])))
-	out.Set("index", i.ptr)
-	out.Set("total", i.filter.Total)
+func (i *userSetIterator) Next(context.Context, *Vars) (out *Vars, err error) {
+	if len(i.buffer)-int(i.ptr) <= 0 {
+		if err = i.loader(); err != nil {
+			panic(err)
+		}
+	}
+
+	out = &Vars{}
+	out.Set("user", Must(NewUser(i.buffer[i.ptr])))
+	out.Set("index", Must(NewInteger(i.total+i.ptr)))
+	out.Set("total", Must(NewInteger(i.filter.Total)))
 
 	i.ptr++
 	return out, nil

tests/workflows/0005_iterator_records_test.go

@@ -0,0 +1,101 @@
+package workflows
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/cortezaproject/corteza-server/compose/automation"
+	"github.com/cortezaproject/corteza-server/pkg/expr"
+	"github.com/cortezaproject/corteza-server/pkg/wfexec"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0005_iterator_records(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateComposeRecords(ctx, nil))
+	req.NoError(defStore.TruncateComposeModules(ctx))
+	req.NoError(defStore.TruncateComposeNamespaces(ctx))
+
+	loadScenario(ctx, t)
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		rec, err := automation.NewComposeRecord(frame.Results.GetValue()["r"])
+		req.NoError(err)
+		rv := rec.GetValue().Values[0]
+		req.Equal(fmt.Sprintf("%d", ctr+1), rv.Value)
+	}
+}
+
+func Test0005_iterator_records_chunked(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = 2
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateComposeRecords(ctx, nil))
+	req.NoError(defStore.TruncateComposeModules(ctx))
+	req.NoError(defStore.TruncateComposeNamespaces(ctx))
+
+	loadScenarioWithName(ctx, t, "S0005_iterator_records")
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		rec, err := automation.NewComposeRecord(frame.Results.GetValue()["r"])
+		req.NoError(err)
+		rv := rec.GetValue().Values[0]
+		req.Equal(fmt.Sprintf("%d", ctr+1), rv.Value)
+	}
+}

tests/workflows/0006_iterator_users_test.go

@@ -0,0 +1,95 @@
+package workflows
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/cortezaproject/corteza-server/pkg/expr"
+	"github.com/cortezaproject/corteza-server/pkg/wfexec"
+	"github.com/cortezaproject/corteza-server/system/automation"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0006_iterator_users(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateUsers(ctx))
+
+	loadScenario(ctx, t)
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		usr, err := automation.NewUser(frame.Results.GetValue()["u"])
+		req.NoError(err)
+		req.Equal(fmt.Sprintf("u%d", ctr+1), usr.GetValue().Handle)
+	}
+}
+
+func Test0006_iterator_users_chunked(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = 2
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateUsers(ctx))
+
+	loadScenarioWithName(ctx, t, "S0006_iterator_users")
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		usr, err := automation.NewUser(frame.Results.GetValue()["u"])
+		req.NoError(err)
+		req.Equal(fmt.Sprintf("u%d", ctr+1), usr.GetValue().Handle)
+	}
+}

tests/workflows/0007_iterator_roles_test.go

@@ -0,0 +1,95 @@
+package workflows
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/cortezaproject/corteza-server/pkg/expr"
+	"github.com/cortezaproject/corteza-server/pkg/wfexec"
+	"github.com/cortezaproject/corteza-server/system/automation"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0007_iterator_roles(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateRoles(ctx))
+
+	loadScenario(ctx, t)
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		usr, err := automation.NewRole(frame.Results.GetValue()["r"])
+		req.NoError(err)
+		req.Equal(fmt.Sprintf("r%d", ctr+1), usr.GetValue().Handle)
+	}
+}
+
+func Test0008_iterator_roles_chunked(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = 2
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateRoles(ctx))
+
+	loadScenarioWithName(ctx, t, "S0007_iterator_roles")
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		usr, err := automation.NewRole(frame.Results.GetValue()["r"])
+		req.NoError(err)
+		req.Equal(fmt.Sprintf("r%d", ctr+1), usr.GetValue().Handle)
+	}
+}

tests/workflows/0008_iterator_role_members_test.go

@@ -0,0 +1,57 @@
+package workflows
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/cortezaproject/corteza-server/pkg/expr"
+	"github.com/cortezaproject/corteza-server/pkg/wfexec"
+	"github.com/cortezaproject/corteza-server/system/automation"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0008_iterator_role_members(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateRoleMembers(ctx))
+	req.NoError(defStore.TruncateRoles(ctx))
+	req.NoError(defStore.TruncateUsers(ctx))
+
+	loadScenario(ctx, t)
+	addRoleMember(ctx, req, "r1", "u1", "u2", "u3", "u4", "u5")
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		usr, err := automation.NewRole(frame.Results.GetValue()["u"])
+		req.NoError(err)
+		req.Equal(fmt.Sprintf("u%d", ctr+1), usr.GetValue().Handle)
+	}
+}

tests/workflows/0009_iterator_templates_test.go

@@ -0,0 +1,95 @@
+package workflows
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	autTypes "github.com/cortezaproject/corteza-server/automation/types"
+	"github.com/cortezaproject/corteza-server/pkg/expr"
+	"github.com/cortezaproject/corteza-server/pkg/wfexec"
+	"github.com/cortezaproject/corteza-server/system/automation"
+	"github.com/stretchr/testify/require"
+)
+
+func Test0009_iterator_templates(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateTemplates(ctx))
+
+	loadScenario(ctx, t)
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		usr, err := automation.NewTemplate(frame.Results.GetValue()["tpl"])
+		req.NoError(err)
+		req.Equal(fmt.Sprintf("t%d", ctr+1), usr.GetValue().Handle)
+	}
+}
+
+func Test0009_iterator_templates_chunked(t *testing.T) {
+	wfexec.MaxIteratorBufferSize = 2
+	defer func() {
+		wfexec.MaxIteratorBufferSize = wfexec.DefaultMaxIteratorBufferSize
+	}()
+
+	var (
+		ctx = superUser(context.Background())
+		req = require.New(t)
+	)
+
+	req.NoError(defStore.TruncateTemplates(ctx))
+
+	loadScenarioWithName(ctx, t, "S0009_iterator_templates")
+
+	var (
+		_, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{})
+	)
+
+	// 6x iterator, 5x continue, 1x terminator, 1x completed
+	req.Len(trace, 13)
+
+	// there are 4 iterator calls; each on the *2 index
+	ctr := int64(-1)
+	for j := 0; j <= 4; j++ {
+		ix := j * 2
+		ctr++
+
+		frame := trace[ix]
+		req.Equal(uint64(10), frame.StepID)
+
+		i, err := expr.Integer{}.Cast(frame.Results.GetValue()["i"])
+		req.NoError(err)
+		req.Equal(ctr, i.Get().(int64))
+
+		tpl, err := automation.NewTemplate(frame.Results.GetValue()["tpl"])
+		req.NoError(err)
+		req.Equal(fmt.Sprintf("t%d", ctr+1), tpl.GetValue().Handle)
+	}
+}

tests/workflows/main_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cortezaproject/corteza-server/store"
 	sysTypes "github.com/cortezaproject/corteza-server/system/types"
 	"github.com/cortezaproject/corteza-server/tests/helpers"
+	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -140,3 +141,21 @@ func mustExecWorkflow(ctx context.Context, t *testing.T, name string, p autTypes
 
 	return
 }
+
+func addRoleMember(ctx context.Context, req *require.Assertions, r string, uu ...string) {
+	role, err := store.LookupRoleByHandle(ctx, defStore, r)
+	req.NoError(err)
+
+	rr := make([]*sysTypes.RoleMember, len(uu))
+	for i, u := range uu {
+		usr, err := store.LookupUserByHandle(ctx, defStore, u)
+		req.NoError(err)
+
+		rr[i] = &sysTypes.RoleMember{
+			RoleID: role.ID,
+			UserID: usr.ID,
+		}
+	}
+
+	req.NoError(store.CreateRoleMember(ctx, defStore, rr...))
+}

tests/workflows/testdata/S0005_iterator_records/data_model.yaml

@@ -0,0 +1,25 @@
+namespaces:
+  ns1:
+    name: ns1 name
+
+modules:
+  mod1:
+    name: mod1 name
+    fields:
+      f1:
+        label: f1 label
+        kind: String
+        required: false
+
+records:
+  mod1:
+    - values:
+        f1: 1
+    - values:
+        f1: 2
+    - values:
+        f1: 3
+    - values:
+        f1: 4
+    - values:
+        f1: 5

tests/workflows/testdata/S0005_iterator_records/workflow.yaml

@@ -0,0 +1,29 @@
+workflows:
+  testing:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: iterator
+        ref: composeRecordsEach
+        arguments:
+          - { "target": "module", "value": "mod1", "type": "Handle" }
+          - { "target": "namespace", "value": "ns1", "type": "Handle" }
+        results:
+          - { "target": "r", "expr": "record" }
+          - { "target": "i", "expr": "index" }
+          - { "target": "t", "expr": "total" }
+
+      - stepID: 11
+        kind: continue
+
+      - stepID: 12
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11 }
+      - { parentID: 10, childID: 12 }

tests/workflows/testdata/S0006_iterator_users/data_model.yaml

@@ -0,0 +1,6 @@
+users:
+  u1: u1@example.tld
+  u2: u2@example.tld
+  u3: u3@example.tld
+  u4: u4@example.tld
+  u5: u5@example.tld

tests/workflows/testdata/S0006_iterator_users/workflow.yaml

@@ -0,0 +1,28 @@
+workflows:
+  testing:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: iterator
+        ref: usersEach
+        arguments:
+          - { "target": "incTotal", "value": "false", "type": "Boolean" }
+        results:
+          - { "target": "u", "expr": "user" }
+          - { "target": "i", "expr": "index" }
+          - { "target": "t", "expr": "total" }
+
+      - stepID: 11
+        kind: continue
+
+      - stepID: 12
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11 }
+      - { parentID: 10, childID: 12 }

tests/workflows/testdata/S0007_iterator_roles/data_model.yaml

@@ -0,0 +1,6 @@
+roles:
+  r1: r1 name
+  r2: r2 name
+  r3: r3 name
+  r4: r4 name
+  r5: r5 name

tests/workflows/testdata/S0007_iterator_roles/workflow.yaml

@@ -0,0 +1,28 @@
+workflows:
+  testing:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: iterator
+        ref: rolesEach
+        arguments:
+          - { "target": "incTotal", "value": "false", "type": "Boolean" }
+        results:
+          - { "target": "r", "expr": "role" }
+          - { "target": "i", "expr": "index" }
+          - { "target": "t", "expr": "total" }
+
+      - stepID: 11
+        kind: continue
+
+      - stepID: 12
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11 }
+      - { parentID: 10, childID: 12 }

tests/workflows/testdata/S0008_iterator_role_members/data_model.yaml

@@ -0,0 +1,9 @@
+roles:
+  r1: r1 name
+
+users:
+  u1: u1@mail.tld
+  u2: u2@mail.tld
+  u3: u3@mail.tld
+  u4: u4@mail.tld
+  u5: u5@mail.tld

tests/workflows/testdata/S0008_iterator_role_members/workflow.yaml

@@ -0,0 +1,28 @@
+workflows:
+  testing:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: iterator
+        ref: rolesEachMember
+        arguments:
+          - { "target": "lookup", "value": "r1", "type": "Handle" }
+        results:
+          - { "target": "u", "expr": "user" }
+          - { "target": "i", "expr": "index" }
+          - { "target": "t", "expr": "total" }
+
+      - stepID: 11
+        kind: continue
+
+      - stepID: 12
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11 }
+      - { parentID: 10, childID: 12 }

tests/workflows/testdata/S0009_iterator_templates/data_model.yaml

@@ -0,0 +1,11 @@
+templates:
+  t1:
+    type: text/html
+  t2:
+    type: text/html
+  t3:
+    type: text/html
+  t4:
+    type: text/html
+  t5:
+    type: text/html

tests/workflows/testdata/S0009_iterator_templates/workflow.yaml

@@ -0,0 +1,28 @@
+workflows:
+  testing:
+    enabled: true
+    trace: true
+    triggers:
+      - enabled: true
+        stepID: 10
+
+    steps:
+      - stepID: 10
+        kind: iterator
+        ref: templatesEach
+        arguments:
+          - { "target": "incTotal", "value": "false", "type": "Boolean" }
+        results:
+          - { "target": "tpl", "expr": "template" }
+          - { "target": "i", "expr": "index" }
+          - { "target": "t", "expr": "total" }
+
+      - stepID: 11
+        kind: continue
+
+      - stepID: 12
+        kind: termination
+
+    paths:
+      - { parentID: 10, childID: 11 }
+      - { parentID: 10, childID: 12 }