POC graph based data migration system

Implement a POC data migration system.
See README for more details.

compose/app.go

@@ -2,6 +2,7 @@ package compose
 
 import (
 	"context"
+
 	"github.com/cortezaproject/corteza-server/pkg/automation"
 
 	"github.com/go-chi/chi"
@@ -101,6 +102,7 @@ func (app *App) RegisterCliCommands(p *cobra.Command) {
 	p.AddCommand(
 		commands.Importer(),
 		commands.Exporter(),
+		commands.Migrator(),
 		// temp command, will be removed in 2020.6
 		automation.ScriptExporter(SERVICE),
 	)

compose/commands/migrator.go

@@ -0,0 +1,91 @@
+package commands
+
+import (
+	"log"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+
+	"github.com/cortezaproject/corteza-server/compose/repository"
+	"github.com/cortezaproject/corteza-server/compose/service"
+	"github.com/cortezaproject/corteza-server/compose/types"
+	"github.com/cortezaproject/corteza-server/pkg/auth"
+	"github.com/cortezaproject/corteza-server/pkg/cli"
+	mgg "github.com/cortezaproject/corteza-server/pkg/migrate"
+	mgt "github.com/cortezaproject/corteza-server/pkg/migrate/types"
+)
+
+func Migrator() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "migrate",
+		Short: "Migrate",
+
+		Run: func(cmd *cobra.Command, args []string) {
+			var (
+				ctx     = auth.SetSuperUserContext(cli.Context())
+				nsFlag  = cmd.Flags().Lookup("namespace").Value.String()
+				dirFlag = cmd.Flags().Lookup("dir").Value.String()
+				ns      *types.Namespace
+				err     error
+
+				mg []mgt.Migrateable
+			)
+
+			svcNs := service.DefaultNamespace.With(ctx)
+
+			if nsFlag == "" {
+				cli.HandleError(errors.New("ns.undefined"))
+			}
+			if dirFlag == "" {
+				cli.HandleError(errors.New("dir.undefined"))
+			}
+
+			if namespaceID, _ := strconv.ParseUint(nsFlag, 10, 64); namespaceID > 0 {
+				ns, err = svcNs.FindByID(namespaceID)
+				if err != repository.ErrNamespaceNotFound {
+					cli.HandleError(err)
+				}
+			} else if ns, err = svcNs.FindByHandle(nsFlag); err != nil {
+				if err != repository.ErrNamespaceNotFound {
+					cli.HandleError(err)
+				}
+			}
+
+			err = filepath.Walk(dirFlag, func(path string, info os.FileInfo, err error) error {
+				if strings.HasSuffix(info.Name(), ".csv") {
+					file, err := os.Open(path)
+					if err != nil {
+						log.Fatal(err)
+					}
+
+					ext := filepath.Ext(info.Name())
+					mg = append(mg, mgt.Migrateable{
+						Name:   info.Name()[0 : len(info.Name())-len(ext)],
+						Path:   path,
+						Source: file,
+					})
+				}
+				return nil
+			})
+
+			if err != nil {
+				panic(err)
+			}
+
+			err = mgg.Migrate(mg, ns, ctx)
+			if err != nil {
+				panic(err)
+			}
+
+		},
+	}
+
+	cmd.Flags().String("namespace", "", "Import into namespace (by ID or string)")
+	cmd.Flags().String("dir", "", "Directory with migration files")
+
+	return cmd
+}

pkg/migrate/main.go

@@ -0,0 +1,319 @@
+package migrate
+
+import (
+	"context"
+	"encoding/csv"
+	"errors"
+	"fmt"
+	"io"
+	"strconv"
+	"time"
+
+	"github.com/cortezaproject/corteza-server/compose/repository"
+	"github.com/cortezaproject/corteza-server/compose/service"
+	cct "github.com/cortezaproject/corteza-server/compose/types"
+	"github.com/cortezaproject/corteza-server/pkg/migrate/types"
+	sysRepo "github.com/cortezaproject/corteza-server/system/repository"
+	sysTypes "github.com/cortezaproject/corteza-server/system/types"
+	"github.com/davecgh/go-spew/spew"
+)
+
+var (
+	userModHandle = "User"
+)
+
+type (
+	Migrator struct {
+		// a set of nodes included in the migration
+		nodes []*types.Node
+
+		// list of leaf nodes, that we might be able to migrate
+		Leafs []*types.Node
+	}
+)
+
+func Migrate(mg []types.Migrateable, ns *cct.Namespace, ctx context.Context) error {
+	mig := &Migrator{}
+	svcMod := service.DefaultModule.With(ctx)
+
+	// 1. migrate all the users, so we can reference then accross the entire system
+	var mgUsr types.Migrateable
+	for _, m := range mg {
+		if m.Name == userModHandle {
+			mgUsr = m
+			break
+		}
+	}
+
+	uMap, err := migrateUsers(mgUsr, ns, ctx)
+	mgUsr.Source.Seek(0, 0)
+	if err != nil {
+		return err
+	}
+
+	// 2. prepare and link migration nodes
+	for _, m := range mg {
+		fmt.Printf("mg.processing > %s\n", m.Name)
+
+		// 2.1 load module
+		mod, err := svcMod.FindByHandle(ns.ID, m.Name)
+		if err != nil {
+			return err
+		}
+
+		// 2.2 get header fields
+		r := csv.NewReader(m.Source)
+		header, err := r.Read()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			panic(err)
+		}
+
+		// 2.3 create migration node
+		n := &types.Node{
+			Name:      m.Name,
+			Module:    mod,
+			Namespace: ns,
+			Reader:    r,
+			Header:    header,
+		}
+		n = mig.AddNode(n)
+
+		// 2.4 prepare additional migration nodes, to provide dep. constraints
+		for _, f := range mod.Fields {
+			if f.Kind == "Record" {
+				refMod := f.Options["moduleID"]
+				if refMod == nil {
+					return errors.New("moduleField.record.missingRef")
+				}
+
+				modID, ok := refMod.(string)
+				if !ok {
+					return errors.New("moduleField.record.invalidRefFormat")
+				}
+				fmt.Printf("mg.node.link > %s [%s]\n", f.Name, modID)
+
+				vv, err := strconv.ParseUint(modID, 10, 64)
+				if err != nil {
+					return err
+				}
+
+				mm, err := svcMod.FindByID(ns.ID, vv)
+				if err != nil {
+					return err
+				}
+
+				nn := &types.Node{
+					Name:      mm.Handle,
+					Module:    mm,
+					Namespace: ns,
+				}
+
+				nn = mig.AddNode(nn)
+				n.LinkAdd(nn)
+			}
+		}
+
+		fmt.Printf("mg.processed > %s\n\n\n", m.Name)
+	}
+
+	fmt.Printf("graph.remove.cycles\n")
+	mig.MakeAcyclic()
+
+	for _, n := range mig.nodes {
+		// keep track of leaf nodes for later importing
+		if !n.HasChildren() {
+			mig.Leafs = append(mig.Leafs, n)
+		}
+	}
+
+	fmt.Printf("migration.prepared\n")
+	fmt.Printf("no. of nodes %d\n", len(mig.nodes))
+	fmt.Printf("no. of entry points %d\n", len(mig.Leafs))
+
+	fmt.Printf("\n\nmigrator.migrating\n")
+	err = mig.Migrate(ctx, uMap)
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("\n\nmigrator.migrating.finished\n")
+
+	return nil
+}
+
+// if function resolves an existing node, it will merge with the provided node
+// and return the new reference
+func (m *Migrator) AddNode(n *types.Node) *types.Node {
+	var fn *types.Node
+	for _, nn := range m.nodes {
+		if nn.Compare(n) {
+			fn = nn
+			break
+		}
+	}
+
+	if fn == nil {
+		spew.Dump(len(n.Children))
+		m.nodes = append(m.nodes, n)
+
+		return n
+	}
+
+	fn.Merge(n)
+	return fn
+}
+
+// it converts the graph from a cyclic (unsafe) graph to an acyclic (safe) graph
+// that can be processed with a single algorithm
+func (m *Migrator) MakeAcyclic() {
+	// splices the node from the cycle and thus preventing the cycle
+	splice := func(n *types.Node, from *types.Node) {
+		spl := n.Splice(from)
+		m.AddNode(spl)
+	}
+
+	for _, n := range m.nodes {
+		if !n.Visited {
+			n.Traverse(splice)
+		}
+	}
+}
+
+// processess migration nodes and migrates the data from the provided source files
+func (m *Migrator) Migrate(ctx context.Context, users map[string]uint64) error {
+	db := repository.DB(ctx)
+	repoRecord := repository.Record(ctx, db)
+
+	for len(m.Leafs) > 0 {
+		for i := len(m.Leafs) - 1; i >= 0; i-- {
+			n := m.Leafs[i]
+			// only nodes with satisfied deps can be migrated
+			if n.Satisfied() {
+
+				// migrate & update leaf nodes
+				add, err := n.Migrate(repoRecord, users)
+				if err != nil {
+					return err
+				}
+
+				// this will maintain order
+				copy(m.Leafs[i:], m.Leafs[i+1:])
+				m.Leafs = m.Leafs[:len(m.Leafs)-1]
+
+				// update leaf nodes (entry points)
+				// take care to not duplicate the given nodes.
+				// That would be a bit not optimal :)
+				for _, a := range add {
+					for _, n := range m.Leafs {
+						if a.Compare(n) {
+							goto skip
+						}
+					}
+
+					m.Leafs = append(m.Leafs, a)
+				skip:
+				}
+
+			}
+		}
+	}
+
+	return nil
+}
+
+// migrates provided users
+// this should be a pre-requisite to any further migration, as user information is required
+func migrateUsers(mg types.Migrateable, ns *cct.Namespace, ctx context.Context) (map[string]uint64, error) {
+	db := repository.DB(ctx)
+	repoUser := sysRepo.User(ctx, db)
+	// this provides a map between SF ID -> CortezaID
+	mapping := make(map[string]uint64)
+
+	// get fields
+	r := csv.NewReader(mg.Source)
+	header, err := r.Read()
+	if err != nil {
+		return nil, err
+	}
+
+	// create users
+	for {
+	looper:
+		record, err := r.Read()
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		u := &sysTypes.User{}
+		for i, h := range header {
+			val := record[i]
+
+			// when creating users we only care about a handfull of values.
+			// the rest are included in the module
+			switch h {
+			case "Username":
+				u.Username = record[i]
+				break
+
+			case "Email":
+				u.Email = record[i]
+				break
+
+			case "FirstName":
+				u.Name = record[i]
+				break
+
+			case "LastName":
+				u.Name = u.Name + " " + record[i]
+				break
+
+			case "CreatedDate":
+				if val != "" {
+					u.CreatedAt, err = time.Parse(types.SfDateTime, val)
+					if err != nil {
+						return nil, err
+					}
+				}
+				break
+
+			case "LastModifiedDate":
+				if val != "" {
+					tt, err := time.Parse(types.SfDateTime, val)
+					u.UpdatedAt = &tt
+					if err != nil {
+						return nil, err
+					}
+				}
+				break
+
+				// ignore deleted values, as SF provides minimal info about those
+			case "IsDeleted":
+				if val != "" {
+					goto looper
+				}
+			}
+		}
+
+		// this allows us to reuse existing users
+		uu, err := repoUser.FindByEmail(u.Email)
+		if err == nil {
+			u = uu
+		} else {
+			u, err = repoUser.Create(u)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		mapping[record[0]] = u.ID
+	}
+
+	return mapping, nil
+}

pkg/migrate/types/migrateable.go

@@ -0,0 +1,18 @@
+package types
+
+import "os"
+
+var (
+	SfDateTime = "2006-01-02 15:04:05"
+)
+
+type (
+	Migrateable struct {
+		Name string
+		Path string
+
+		Source *os.File
+		// @todo?
+		Mapping *os.File
+	}
+)

pkg/migrate/types/node.go

@@ -0,0 +1,422 @@
+package types
+
+import (
+	"encoding/csv"
+	"errors"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/cortezaproject/corteza-server/compose/repository"
+	"github.com/cortezaproject/corteza-server/compose/types"
+)
+
+type (
+	// graph node
+	Node struct {
+		// unique node name
+		Name string
+
+		// keep note of parent refs, so we don't need to inverse it ;)
+		Parents []*Node
+		// keep note of Children, as they are our dependencies
+		Children []*Node
+		// mapping from migrated IDs to Corteza IDs
+		mapping map[string]Map
+		// determines if node is in current path; used for loop detection
+		inPath bool
+		// determines if node was spliced; used to break the loop
+		spliced  bool
+		original *Node
+
+		// records are applicable in the case of spliced nodes
+		records []*types.Record
+
+		// some refs
+		Module    *types.Module
+		Namespace *types.Namespace
+		Reader    *csv.Reader
+
+		// meta
+		Header  []string
+		Visited bool
+	}
+
+	// map between migrated ID and Corteza ID
+	Map map[string]string
+)
+
+// helper, to determine if the two nodes are equal
+func (n *Node) Compare(to *Node) bool {
+	return n.Name == to.Name && n.spliced == to.spliced
+}
+
+// helper to stringify the node
+func (n *Node) Stringify() string {
+	return fmt.Sprintf("NODE > n: %s; spliced: %t; inPath: %t;", n.Name, n.spliced, n.inPath)
+}
+
+// adds a new map to the given node
+func (n *Node) addMap(key string, m Map) {
+	if n.mapping == nil {
+		n.mapping = map[string]Map{}
+	}
+
+	n.mapping[key] = m
+}
+
+// does the actual data migration for the given node
+func (n *Node) Migrate(repoRecord repository.RecordRepository, users map[string]uint64) ([]*Node, error) {
+	fmt.Printf("node.migrate > %s\n", n.Stringify())
+	var err error
+
+	mapping := make(Map)
+	if n.Reader != nil {
+		// if records exist (from spliced node); correct refs
+		if !n.spliced && n.records != nil && len(n.records) > 0 {
+			fmt.Printf("node.refs.update\n")
+			// we can just reuse the mapping object, since it will remain the same
+			mapping = n.mapping[fmt.Sprint(n.Module.ID)]
+
+			err := updateRefs(n, repoRecord)
+			if err != nil {
+				return nil, err
+			}
+			fmt.Printf("node.refs.update.done\n")
+		} else {
+			fmt.Printf("node.migrate.source\n")
+			mapping, err = importNodeSource(n, users, repoRecord)
+			if err != nil {
+				return nil, err
+			}
+			fmt.Printf("node.migrate.source.done\n")
+		}
+	}
+
+	var rtr []*Node
+
+	// update node refs
+	for _, p := range n.Parents {
+		rtr = append(rtr, p)
+
+		// pass mapping object to the node's parend so it can migrate it's data
+		p.addMap(fmt.Sprint(n.Module.ID), mapping)
+		p.LinkRemove(n)
+	}
+
+	return rtr, nil
+}
+
+// determines if node is Satisfied and can be imported
+// it is Satisfied, when all of it's dependencies have been imported ie. no
+// more child refs
+func (n *Node) Satisfied() bool {
+	return !n.HasChildren()
+}
+
+func (n *Node) HasChildren() bool {
+	return n.Children != nil && len(n.Children) > 0
+}
+
+// partially Merge the two nodes
+func (n *Node) Merge(nn *Node) {
+	if nn.Module != nil {
+		n.Module = nn.Module
+	}
+	if nn.Reader != nil {
+		n.Reader = nn.Reader
+	}
+	if nn.Header != nil {
+		n.Header = nn.Header
+	}
+}
+
+// link the two nodes
+func (n *Node) LinkAdd(to *Node) {
+	n.addChild(to)
+	to.addParent(n)
+}
+
+// remove the link between the two nodes
+func (n *Node) LinkRemove(from *Node) {
+	n.Children = n.removeIfPresent(from, n.Children)
+	from.Parents = from.removeIfPresent(n, from.Parents)
+}
+
+// adds a parent node to the given node
+func (n *Node) addParent(add *Node) {
+	n.Parents = n.addIfMissing(add, n.Parents)
+}
+
+// adds a child node to the given node
+func (n *Node) addChild(add *Node) {
+	n.Children = n.addIfMissing(add, n.Children)
+}
+
+// adds a node, if it doesn't yet exist
+func (n *Node) addIfMissing(add *Node, list []*Node) []*Node {
+	var fn *Node
+
+	for _, nn := range list {
+		if add.Compare(nn) {
+			fn = nn
+		}
+	}
+
+	if fn != nil {
+		fn.Merge(add)
+		return list
+	}
+	return append(list, add)
+}
+
+// removes the node, if it exists
+func (n *Node) removeIfPresent(rem *Node, list []*Node) []*Node {
+	for i, nn := range list {
+		if rem.Compare(nn) {
+			// https://stackoverflow.com/a/37335777
+			list[len(list)-1], list[i] = list[i], list[len(list)-1]
+			return list[:len(list)-1]
+		}
+	}
+
+	return list
+}
+
+// traverses the graph and notifies us of any cycles
+func (n *Node) Traverse(cycle func(n *Node, to *Node)) {
+	n.inPath = true
+	n.Visited = true
+
+	var cc []*Node
+	for _, nn := range n.Children {
+		cc = append(cc, nn)
+	}
+
+	for _, nn := range cc {
+		if n.Name == "client" {
+		}
+
+		if nn.inPath {
+			cycle(n, nn)
+		} else {
+			nn.Traverse(cycle)
+		}
+	}
+
+	n.inPath = false
+}
+
+func (n *Node) DFS() {
+	n.inPath = true
+
+	for _, nn := range n.Children {
+		if !nn.inPath {
+			nn.DFS()
+		}
+	}
+
+	n.inPath = false
+}
+
+// clones the given node
+func (n *Node) clone() *Node {
+	return &Node{
+		Name:      n.Name,
+		Parents:   n.Parents,
+		Children:  n.Children,
+		mapping:   n.mapping,
+		inPath:    n.inPath,
+		spliced:   n.spliced,
+		original:  n.original,
+		records:   n.records,
+		Visited:   n.Visited,
+		Module:    n.Module,
+		Namespace: n.Namespace,
+		Reader:    n.Reader,
+		Header:    n.Header,
+	}
+}
+
+// splices the node from the original graph and removes the cycle
+func (n *Node) Splice(from *Node) *Node {
+	splicedN := from.clone()
+	splicedN.spliced = true
+	splicedN.Parents = nil
+	splicedN.Children = nil
+	splicedN.inPath = false
+
+	splicedN.original = from
+
+	n.LinkRemove(from)
+	n.LinkAdd(splicedN)
+
+	return splicedN
+}
+
+func sysField(f string) bool {
+	switch f {
+	case "CreatedDate",
+		"CreatedById",
+		"LastModifiedById",
+		"LastModifiedDate",
+		"IsDeleted":
+		return true
+	}
+	return false
+}
+
+func updateRefs(n *Node, repo repository.RecordRepository) error {
+	// correct references
+	for _, r := range n.records {
+		for _, v := range r.Values {
+			var f *types.ModuleField
+			// find the applicable module field
+			for _, ff := range n.Module.Fields {
+				if ff.Name == v.Name {
+					f = ff
+					break
+				}
+			}
+
+			val := v.Value
+			// determine value based on the provided map
+			if f != nil && f.Options["moduleID"] != nil {
+				ref, ok := f.Options["moduleID"].(string)
+				if !ok {
+					return errors.New("moduleField.record.invalidRefFormat")
+				}
+
+				val = n.mapping[ref][val]
+				v.Value = val
+			}
+		}
+
+		// update values
+		err := repo.UpdateValues(r.ID, r.Values)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func importNodeSource(n *Node, users map[string]uint64, repo repository.RecordRepository) (Map, error) {
+	mapping := make(Map)
+
+	for {
+	looper:
+		record, err := n.Reader.Read()
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		rr := &types.Record{
+			ModuleID:    n.Module.ID,
+			NamespaceID: n.Namespace.ID,
+			CreatedAt:   time.Now(),
+		}
+
+		vals := types.RecordValueSet{}
+		for i, h := range n.Header {
+			val := record[i]
+
+			if sysField(h) {
+				switch h {
+				case "CreatedDate":
+					if val != "" {
+						rr.CreatedAt, err = time.Parse(SfDateTime, val)
+						if err != nil {
+							return nil, err
+						}
+					}
+					break
+
+				case "CreatedById":
+					rr.CreatedBy = users[val]
+					break
+
+				case "LastModifiedById":
+					rr.UpdatedBy = users[val]
+					break
+
+				case "LastModifiedDate":
+					if val != "" {
+						tt, err := time.Parse(SfDateTime, val)
+						rr.UpdatedAt = &tt
+						if err != nil {
+							return nil, err
+						}
+					}
+					break
+
+					// ignore deleted values, as SF provides minimal info about those
+				case "IsDeleted":
+					if val != "" {
+						goto looper
+					}
+				}
+			} else {
+				var f *types.ModuleField
+				for _, ff := range n.Module.Fields {
+					if ff.Name == h {
+						f = ff
+						break
+					}
+				}
+
+				// spliced nodes should NOT manage their references
+				if !n.spliced && f != nil && f.Options["moduleID"] != nil {
+					ref, ok := f.Options["moduleID"].(string)
+					if !ok {
+						return nil, errors.New("moduleField.record.invalidRefFormat")
+					}
+
+					if n.mapping[ref] != nil {
+						if n.mapping[ref] != nil {
+							val = n.mapping[ref][val]
+						}
+					}
+				}
+
+				if f != nil && f.Kind == "User" {
+					val = fmt.Sprint(users[val])
+				}
+
+				vals = append(vals, &types.RecordValue{
+					Name:  h,
+					Value: val,
+				})
+			}
+		}
+
+		// create record
+		r, err := repo.Create(rr)
+		if err != nil {
+			return nil, err
+		}
+
+		// update record values with recordID
+		for _, v := range vals {
+			v.RecordID = r.ID
+		}
+		err = repo.UpdateValues(r.ID, vals)
+		if err != nil {
+			return nil, err
+		}
+
+		// spliced nodes should preserve their records for later ref processing
+		if n.spliced {
+			rr.Values = vals
+			n.original.records = append(n.original.records, rr)
+		}
+
+		mapping[record[0]] = fmt.Sprint(rr.ID)
+	}
+
+	return mapping, nil
+}