WIP

command/command.go

+package command
+
+import (
+	"context"
+
+	"mtoohey.com/iter/v2"
+)
+
+type Command interface {
+	Execute(ctx context.Context) error
+}
+
+type SubCommand interface {
+	Commands(ctx context.Context) iter.Iter[Command]
+}
+
+type Commands iter.Iter[Command]
+
+type EmptyCommand struct{}
+
+func (*EmptyCommand) Execute(ctx context.Context) error {
+	return nil
+}

command/history.go

+package command
+
+type History struct {
+	commands []Command
+}
+
+func NewHistory() *History {
+	return &History{}
+}
+
+func (h *History) Add(cmd Command) {
+	h.commands = append(h.commands, cmd)
+}

command/invoker.go

+package command
+
+import (
+	"context"
+
+	"mtoohey.com/iter/v2"
+)
+
+type Invoker struct {
+	history *History
+}
+
+func NewInvoker() *Invoker {
+	return &Invoker{
+		history: NewHistory(),
+	}
+}
+
+func (e *Invoker) Execute(ctx context.Context, cmds ...Command) error {
+	return e.ExecuteIter(ctx, iter.Elems(cmds))
+}
+
+func (e *Invoker) executeCommand(ctx context.Context, cmd Command) error {
+	if e.history != nil {
+		e.history.Add(cmd)
+	}
+	return cmd.Execute(ctx)
+}
+
+func (e *Invoker) ExecuteIter(ctx context.Context, cmds iter.Iter[Command]) error {
+	for {
+		cmd, cont := cmds()
+		if !cont {
+			break
+		}
+
+		if err := e.executeCommand(ctx, cmd); err != nil {
+			return err
+		}
+
+		if sub, ok := cmd.(SubCommand); ok {
+			subs := sub.Commands(ctx)
+			if subs != nil {
+				cmds = subs.Chain(cmds)
+			}
+		}
+	}
+	return nil
+}

command/invoker_test.go

+package command
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"mtoohey.com/iter/v2"
+
+	"git.perx.ru/perxis/perxis-go/pkg/errors"
+)
+
+type push struct {
+	s *string
+	r rune
+}
+
+func (p *push) Execute(ctx context.Context) error {
+	str := *(p.s)
+	str = str + string(p.r)
+	*(p.s) = str
+	return nil
+}
+
+type pop struct {
+	s *string
+}
+
+func (p *pop) Execute(ctx context.Context) error {
+	str := *(p.s)
+
+	if len(str) == 0 {
+		return errors.New("pop: empty string")
+	}
+
+	*(p.s) = str[:len(str)-1]
+	return nil
+}
+
+type repeat struct {
+	cmd Command
+	n   int
+}
+
+func (r *repeat) Execute(ctx context.Context) error {
+	return nil
+}
+
+func (r *repeat) Commands(ctx context.Context) iter.Iter[Command] {
+	n := r.n
+	return iter.GenWhile(func() (Command, error) {
+		n = n - 1
+		if n < 0 {
+			return nil, errors.New("repeat: out of bounds")
+		}
+		return r.cmd, nil
+	})
+}
+
+func TestInvoker_Execute(t *testing.T) {
+	t.Run("push", func(t *testing.T) {
+		invoker := NewInvoker()
+
+		var cmds []Command
+		var str string
+		cmds = append(cmds, &push{s: &str, r: 'a'})
+		cmds = append(cmds, &push{s: &str, r: 'b'})
+		cmds = append(cmds, &push{s: &str, r: 'c'})
+		err := invoker.Execute(context.Background(), cmds...)
+		require.NoError(t, err)
+		require.Equal(t, "abc", str)
+	})
+
+	t.Run("pop", func(t *testing.T) {
+		invoker := NewInvoker()
+
+		var cmds []Command
+		var str string
+		cmds = append(cmds, &push{s: &str, r: 'a'})
+		cmds = append(cmds, &push{s: &str, r: 'b'})
+		cmds = append(cmds, &push{s: &str, r: 'C'})
+		cmds = append(cmds, &push{s: &str, r: 'D'})
+		cmds = append(cmds, &pop{s: &str})
+		cmds = append(cmds, &pop{s: &str})
+		cmds = append(cmds, &push{s: &str, r: 'c'})
+		err := invoker.Execute(context.Background(), cmds...)
+		require.NoError(t, err)
+		require.Equal(t, "abc", str)
+	})
+	t.Run("repeat", func(t *testing.T) {
+		invoker := NewInvoker()
+
+		var cmds []Command
+		var str string
+
+		cmds = append(cmds, &repeat{cmd: &push{s: &str, r: 'a'}, n: 4})
+		cmds = append(cmds, &pop{s: &str})
+		cmds = append(cmds, &pop{s: &str})
+		cmds = append(cmds, &pop{s: &str})
+		err := invoker.Execute(context.Background(), cmds...)
+		require.NoError(t, err)
+		require.Equal(t, "a", str)
+	})
+	t.Run("error", func(t *testing.T) {
+		invoker := NewInvoker()
+
+		var cmds []Command
+		var str string
+
+		cmds = append(cmds, &repeat{cmd: &push{s: &str, r: 'a'}, n: 4})
+		cmds = append(cmds, &repeat{cmd: &pop{s: &str}, n: 5})
+		err := invoker.Execute(context.Background(), cmds...)
+		require.Error(t, err)
+	})
+}

datasource/collection.go

+package datasource
+
+import (
+	"context"
+
+	"git.perx.ru/perxis/perxis-go/pkg/collections"
+	"github.com/barweiss/go-tuple"
+	"mtoohey.com/iter/v2"
+)
+
+// CollectionTuple - кортеж из двух коллекций
+type CollectionTuple = tuple.T2[*collections.Collection, *collections.Collection]
+
+// CollectionDataSource возвращает Datasource по коллекциям
+// type CollectionDataSource func() iter.Iter[*collections.Collection]
+
+// CollectionDataSource - это структура данных, представляющая источник данных для коллекций.
+type CollectionDataSource struct {
+	err     error
+	spaceId string
+	envId   string
+	filter  *collections.Filter
+	svc     collections.Collections
+	colls   []*collections.Collection
+}
+
+// NewCollectionsDatasource является функцией для создания нового источника данных коллекций.
+// Эта функция принимает следующие параметры:
+// svc - интерфейс Collections, который предоставляет методы для работы с коллекциями данных.
+// spaceId - строка, идентифицирующая пространство данных.
+// envId - строка, идентифицирующая среду данных.
+// filter - структура Filter, которая используется для настройки фильтрации возвращаемых коллекций.
+// Возвращает структуру CollectionDataSource, которая может быть использована
+// для итерации по коллекциям, отфильтрованным на основе предоставленных параметров.
+func NewCollectionsDatasource(svc collections.Collections, spaceId, envId string, filter *collections.Filter) *CollectionDataSource {
+	return &CollectionDataSource{
+		svc:     svc,
+		spaceId: spaceId,
+		envId:   envId,
+		filter:  filter,
+	}
+}
+
+func (ds *CollectionDataSource) Error() error {
+	return ds.err
+}
+
+func (ds *CollectionDataSource) Iter(ctx context.Context) iter.Iter[*collections.Collection] {
+	i := 0
+
+	return func() (*collections.Collection, bool) {
+		if ds.err != nil {
+			return nil, false
+		}
+
+		if ds.colls == nil {
+			ds.colls, ds.err = ds.svc.List(ctx, ds.spaceId, ds.envId, ds.filter)
+		}
+
+		if ds.err != nil {
+			return nil, false
+		}
+
+		if i >= len(ds.colls) {
+			return nil, false
+		}
+
+		c := ds.colls[i]
+		i++
+		return c, true
+	}
+}
+
+func CompareCollectionID(a, b *collections.Collection) (bool, bool) {
+	return a.ID < b.ID, a.ID == b.ID
+}
+
+func MatchCollectionByID(a, b iter.Iter[*collections.Collection]) iter.Iter[CollectionTuple] {
+	return Match(a, b, CompareCollectionID)
+}

datasource/collection_commands.go

+package datasource
+
+import (
+	"context"
+
+	"git.perx.ru/perxis/perxis-go/command"
+
+	"git.perx.ru/perxis/perxis-go/pkg/content"
+
+	"git.perx.ru/perxis/perxis-go/pkg/collections"
+	"mtoohey.com/iter/v2"
+)
+
+type Builder interface {
+	CreateCollection(c *collections.Collection) command.Command
+	DeleteCollection(c *collections.Collection) command.Command
+	CopyItems(source, target *collections.Collection) command.Command
+	SyncItems(source, target *collections.Collection) command.Command
+}
+
+func NewCreateCollectionCommand(cnt *content.Content, coll *collections.Collection) *CreateCollectionCommand {
+	return &CreateCollectionCommand{Collection: coll}
+}
+
+type CreateCollectionCommand struct {
+	Collection *collections.Collection
+}
+
+func (cmd *CreateCollectionCommand) Execute(ctx context.Context) error {
+	return nil
+}
+
+type CopyItems struct {
+	Source *collections.Collection
+	Target *collections.Collection
+}
+
+func (cmd *CopyItems) Execute(ctx context.Context) error {
+	return cmd.Executor.Handle(ctx, cmd)
+}
+
+type SyncItemsCommand struct {
+	Source *ItemDataSource
+	Target *ItemDataSource
+}
+
+type SyncCollectionsCommand struct {
+	Source    *CollectionDataSource
+	Target    *CollectionDataSource
+	Executor  Executor
+	ParentCmd Command
+	Error     error
+	Strategy  func(cmd *SyncCollectionsCommand, Source, Target *collections.Collection) iter.Iter[Command]
+}
+
+func SyncCollectionsStrategy(cmd *SyncCollectionsCommand, Source, Target *collections.Collection) iter.Iter[Command] {
+	if Source == nil {
+		return iter.Elems([]Command{&NilCommand{}})
+	}
+
+	if Target == nil {
+		coll := Source.Clone()
+		coll.SpaceID = cmd.Target.spaceId
+		coll.EnvID = cmd.Target.envId
+		return iter.Elems([]Command{
+			&CreateCollectionCommand{Collection: coll},
+			&CopyItems{Source: Source, Target: coll},
+		})
+	}
+
+	return iter.Elems([]Command{
+		&CopyItems{Source: Source, Target: Target},
+	})
+}
+
+func SyncCollectionsStrategy(cmd *SyncCollectionsCommand, Source, Target *collections.Collection) iter.Iter[Command] {
+	if Source == nil {
+		return iter.Elems([]Command{&NilCommand{}})
+	}
+
+	if Target == nil {
+		coll := Source.Clone()
+		coll.SpaceID = cmd.Target.spaceId
+		coll.EnvID = cmd.Target.envId
+		return iter.Elems([]Command{
+			&CreateCollectionCommand{Collection: coll},
+			&CopyItems{Source: Source, Target: coll},
+		})
+	}
+
+	return iter.Elems([]Command{
+		&CopyItems{Source: Source, Target: Target},
+	})
+}
+
+func (cmd *SyncCollectionsCommand) Execute(ctx context.Context) error {
+	match := MatchCollectionByID(cmd.Source.Iter(ctx), cmd.Target.Iter(ctx))
+
+	cmds := iter.FlatMap(match, func(t CollectionTuple) iter.Iter[Command] {
+		return cmd.Strategy(cmd, t.V1, t.V2)
+	})
+
+	return cmd.Executor.ExecuteIter(ctx, cmds)
+}

datasource/command.go

+package datasource

pkg/datasource/filter.go → datasource/filter.go

@@ -17,5 +17,3 @@ func FilterEqual[T Value[T]](t tuple.T2[T, T]) bool {
 func FilterDiff[T Value[T]](t tuple.T2[T, T]) bool {
 	return !FilterEqual(t)
 }
-
-func Chanded
\ No newline at end of file

datasource/match.go

+package datasource
+
+import (
+	"github.com/barweiss/go-tuple"
+	"mtoohey.com/iter/v2"
+)
+
+type Compare uint
+
+const (
+	EQUAL Compare = iota
+	LESS
+	GREATER
+)
+
+type CompareFunc[V, W any] func(V, W) (less bool, equal bool)
+
+// MatchID возвращает Datasource по двум наборам коллекций, объединяющие коллекции с одинаковыми ID
+// Оба итератора должны возвращать записи в порядке возрастания ID.
+// Match читает значения вперед на 1 запись, чтобы определить, какие записи объединить.
+func Match[V, W any](i1 iter.Iter[V], i2 iter.Iter[W], f CompareFunc[V, W]) iter.Iter[tuple.T2[V, W]] {
+	var (
+		c1    V    // Current value from i1
+		c2    W    // Current value from i2
+		cont1 bool // Continue reading from i1
+		cont2 bool // Continue reading from i2
+		skip1 bool // Skip reading from i1
+		skip2 bool // Skip reading from i2
+	)
+
+	return func() (tuple.T2[V, W], bool) {
+		var t tuple.T2[V, W]
+
+		if !skip1 {
+			c1, cont1 = i1()
+		}
+
+		if !skip2 {
+			c2, cont2 = i2()
+		}
+
+		skip1 = false
+		skip2 = false
+
+		if !cont1 && !cont2 {
+			skip1 = true
+			skip2 = true
+			return t, false
+		}
+
+		if !cont1 {
+			t = tuple.New2(Zero[V](), c2)
+			skip1 = true
+			return t, true
+		}
+
+		if !cont2 {
+			t = tuple.New2(c1, Zero[W]())
+			skip2 = true
+			return t, true
+		}
+
+		less, equal := f(c1, c2)
+
+		switch {
+		case equal:
+			t = tuple.New2(c1, c2)
+		case less:
+			t = tuple.New2(c1, Zero[W]())
+			skip2 = true
+		default:
+			t = tuple.New2(Zero[V](), c2)
+			skip1 = true
+		}
+
+		return t, true
+	}
+}

datasource/value.go

+package datasource
+
+func Zero[V any]() V {
+	var e V
+	return e
+}
+
+type Value[V comparable] interface {
+	IsEqual(other V) bool
+	comparable
+}
+
+type ComparableValue[V comparable] interface {
+	Value[V]
+	IsLess(other V) bool
+}
+
+type IDValue[V comparable] interface {
+	Value[V]
+	GetID() string
+}
+
+func IsZero[V comparable](t V) bool {
+	return t == Zero[V]()
+}

pkg/datasource/copy.go

@@ -34,6 +34,16 @@ type Copy struct {
 	env2    string
 }
 
+type SyncConfig struct {
+	SrcCollectionFilter  func(*collections.Collection) bool                          // Фильтр коллекций в источнике
+	DstCollectionFilter  func(*collections.Collection) bool                          // Фильтр коллекций в приемнике
+	SyncCollectionFilter func(tuple *CollectionTuple) bool                           // Фильтр пар коллекций для синхронизации
+	CollectionMatcher    func(*collections.Collection, *collections.Collection) bool // Функция сравнения коллекций ??
+	SrcItemFilter        func(*items.Item) bool                                      // Фильтр элементов в источнике
+	DstItemFilter        func(*items.Item) bool                                      // Фильтр элементов в приемнике
+	SyncItemFilter       func(tuple *ItemTuple) bool                                 // Функция сравнения элементов ??
+}
+
 func NewCopy(colSvc collections.Collections, itemSvc items.Items, space1, space2, env1, env2 string) *Copy {
 	return &Copy{
 		colSvc:  colSvc,

sync/command.go

+package sync
+
+import (
+	"context"
+
+	"mtoohey.com/iter/v2"
+)
+
+type Command interface {
+	Execute(ctx context.Context) error
+	Commands(ctx context.Context) iter.Iter[Command]
+}
+
+type Commands iter.Iter[Command]
+
+type Executor interface {
+	Execute(ctx context.Context, cmds ...Command) error
+	ExecuteIter(ctx context.Context, cmds iter.Iter[Command]) error
+}
+
+type executor struct {
+}
+
+func (e *executor) Execute(ctx context.Context, cmds ...Command) error {
+	return nil
+}
+
+func (e *executor) ExecuteIter(ctx context.Context, cmds iter.Iter[Command]) error {
+	cmds.ForEach(func(cmd Command) {
+		_ = cmd.Execute(ctx)
+		sub := cmd.Commands(ctx)
+		cmds = cmds.Chain(sub)
+	})
+}

sync/doc.go

+package sync

sync/sync.go

+package sync
+
+import (
+	"git.perx.ru/perxis/perxis-go/pkg/collections"
+	"mtoohey.com/iter/v2"
+)
+
+type Sync struct {
+	i iter.Iter[Command]
+}
+
+// NewSync - создает новый Sync
+func NewSync(i iter.Iter[Command]) *Sync {
+	return &Sync{
+		i: i,
+	}
+}
+
+func (s *Sync) AddCommands(commands iter.Iter[Command]) {
+	s.i = s.i.Chain(commands)
+}
+
+// Sync - выполняет синхронизацию
+func (s *Sync) Sync() error {
+
+}
+
+// AddSourceCollectionFilter - добавляет фильтр по коллекции
+func (s *Sync) AddSourceCollectionFilter(filter *collections.Filter) {
+
+}