diff --git a/compose/app.go b/compose/app.go index 85a8e0b80..8c954d532 100644 --- a/compose/app.go +++ b/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), ) diff --git a/compose/commands/migrator.go b/compose/commands/migrator.go new file mode 100644 index 000000000..9bd5def67 --- /dev/null +++ b/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 +} diff --git a/pkg/migrate/main.go b/pkg/migrate/main.go new file mode 100644 index 000000000..23eede57b --- /dev/null +++ b/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 +} diff --git a/pkg/migrate/types/migrateable.go b/pkg/migrate/types/migrateable.go new file mode 100644 index 000000000..3445a7bc9 --- /dev/null +++ b/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 + } +) diff --git a/pkg/migrate/types/node.go b/pkg/migrate/types/node.go new file mode 100644 index 000000000..284e304ec --- /dev/null +++ b/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 +}