From 399f9a7cc490dc8849ae285568a1e86d9c78703d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Mon, 9 Oct 2023 14:45:01 +0200 Subject: [PATCH] Plug new envoy to record import endpoints --- server/compose/envoy/record_datasource.go | 38 +-- server/compose/envoy/store_decode.go | 4 +- server/compose/envoy/store_encode.go | 12 + server/compose/envoy/yaml_decode.go | 89 +------- server/compose/rest/record.go | 267 +++++++++++++++------- server/compose/service/import_session.go | 62 ++--- server/compose/service/record.go | 4 +- server/pkg/envoyx/datasource.go | 92 +++++++- server/pkg/envoyx/envoy.go | 1 + 9 files changed, 341 insertions(+), 228 deletions(-) diff --git a/server/compose/envoy/record_datasource.go b/server/compose/envoy/record_datasource.go index 8b0d74f42..9193cc7f1 100644 --- a/server/compose/envoy/record_datasource.go +++ b/server/compose/envoy/record_datasource.go @@ -2,6 +2,7 @@ package envoy import ( "context" + "strconv" "strings" "github.com/cortezaproject/corteza/server/pkg/dal" @@ -13,8 +14,10 @@ type ( // RecordDatasource provides a mechanism for you to access large // record datasets optimally RecordDatasource struct { - mapping datasourceMapping - provider envoyx.Provider + Mapping envoyx.DatasourceMapping + Provider envoyx.Provider + + currentIndex int // Reusable buffer for reading records rowCache map[string]string @@ -36,11 +39,11 @@ type ( ) func (rd *RecordDatasource) SetProvider(s envoyx.Provider) bool { - if rd.mapping.SourceIdent != s.Ident() { + if rd.Mapping.SourceIdent != s.Ident() { return false } - rd.provider = s + rd.Provider = s return true } @@ -49,27 +52,34 @@ func (rd *RecordDatasource) Next(ctx context.Context, out map[string]string) (id rd.rowCache = make(map[string]string) } - more, err = rd.provider.Next(ctx, rd.rowCache) + more, err = rd.Provider.Next(ctx, rd.rowCache) if err != nil || !more { return } rd.applyMapping(rd.rowCache, out) - for _, k := range rd.mapping.KeyField { - ident = append(ident, rd.rowCache[k]) + if len(rd.Mapping.KeyField) == 0 { + ident = append(ident, strconv.FormatInt(int64(rd.currentIndex), 10)) + } else { + for _, k := range rd.Mapping.KeyField { + ident = append(ident, rd.rowCache[k]) + } } + rd.currentIndex++ + return } func (rd *RecordDatasource) Reset(ctx context.Context) (err error) { - return rd.provider.Reset(ctx) + rd.currentIndex = 0 + return rd.Provider.Reset(ctx) } func (rd *RecordDatasource) applyMapping(in, out map[string]string) { - if len(rd.mapping.Mapping.m) == 0 { - if !rd.mapping.Defaultable { + if len(rd.Mapping.Mapping.Map) == 0 { + if !rd.Mapping.Defaultable { return } @@ -79,7 +89,7 @@ func (rd *RecordDatasource) applyMapping(in, out map[string]string) { return } - if rd.mapping.Defaultable { + if rd.Mapping.Defaultable { rd.applyMappingWithDefaults(in, out) } else { rd.applyMappingWoDefaults(in, out) @@ -87,8 +97,8 @@ func (rd *RecordDatasource) applyMapping(in, out map[string]string) { } func (rd *RecordDatasource) applyMappingWithDefaults(in, out map[string]string) { - maps := make(map[string]mapEntry) - for k, v := range rd.mapping.Mapping.m { + maps := make(map[string]envoyx.MapEntry) + for k, v := range rd.Mapping.Mapping.Map { maps[k] = v } @@ -105,7 +115,7 @@ func (rd *RecordDatasource) applyMappingWithDefaults(in, out map[string]string) } func (rd *RecordDatasource) applyMappingWoDefaults(in, out map[string]string) { - for _, m := range rd.mapping.Mapping.m { + for _, m := range rd.Mapping.Mapping.Map { if m.Skip { continue } diff --git a/server/compose/envoy/store_decode.go b/server/compose/envoy/store_decode.go index a7abc7940..7251ab2e2 100644 --- a/server/compose/envoy/store_decode.go +++ b/server/compose/envoy/store_decode.go @@ -240,10 +240,10 @@ func (d StoreDecoder) decodeRecordDatasource(ctx context.Context, s store.Storer } ou := &RecordDatasource{ - provider: &iteratorProvider{iter: iter}, + Provider: &iteratorProvider{iter: iter}, refToID: make(map[string]uint64), // @todo consider providing defaults from the outside - mapping: datasourceMapping{ + Mapping: envoyx.DatasourceMapping{ KeyField: []string{"id"}, Defaultable: true, }, diff --git a/server/compose/envoy/store_encode.go b/server/compose/envoy/store_encode.go index f3a8df54f..f1378fa06 100644 --- a/server/compose/envoy/store_encode.go +++ b/server/compose/envoy/store_encode.go @@ -13,6 +13,18 @@ import ( ) func (e StoreEncoder) encode(ctx context.Context, p envoyx.EncodeParams, s store.Storer, rt string, nn envoyx.NodeSet, tree envoyx.Traverser) (err error) { + dl, err := e.grabDal(p) + if err != nil { + return + } + + switch rt { + case ComposeRecordDatasourceAuxType: + err = e.encodeRecordDatasources(ctx, p, s, dl, nn, tree) + if err != nil { + return + } + } return } diff --git a/server/compose/envoy/yaml_decode.go b/server/compose/envoy/yaml_decode.go index c473847e6..1135ad516 100644 --- a/server/compose/envoy/yaml_decode.go +++ b/server/compose/envoy/yaml_decode.go @@ -12,32 +12,6 @@ import ( "gopkg.in/yaml.v3" ) -type ( - mapEntry struct { - Column string - Field string - Skip bool - } - - fieldMapping struct { - m map[string]mapEntry - } - - datasourceMapping struct { - SourceIdent string `yaml:"source"` - KeyField []string `yaml:"key"` - References map[string]string - Scope map[string]string - - // Defaultable indicates wether the mapping should keep the values where - // the ident is not explicitly mapped. - // - // When true, the value is assigned to the given identifier. - Defaultable bool `yaml:"defaultable"` - Mapping fieldMapping - } -) - const ( ComposeRecordDatasourceAuxType = "corteza::compose:record-datasource" ) @@ -432,7 +406,7 @@ func (d *auxYamlDoc) unmarshalPagesExtendedNode(dctx documentContext, n *yaml.No } func (d *auxYamlDoc) unmarshalSourceExtendedNode(dctx documentContext, n *yaml.Node, meta ...*yaml.Node) (out envoyx.NodeSet, err error) { - var r datasourceMapping + var r envoyx.DatasourceMapping // @todo we're omitting errors because there will be a bunch due to invalid // resource field types. This might be a bit unstable as other errors may @@ -474,7 +448,7 @@ func (d *auxYamlDoc) unmarshalSourceExtendedNode(dctx documentContext, n *yaml.N // @todo for now we only support record datasources; extend when needed auxN := &envoyx.Node{ Datasource: &RecordDatasource{ - mapping: r, + Mapping: r, }, ResourceType: ComposeRecordDatasourceAuxType, @@ -486,65 +460,6 @@ func (d *auxYamlDoc) unmarshalSourceExtendedNode(dctx documentContext, n *yaml.N return } -// UnmarshalYAML is used to get the yaml parsed into a series of nodes so -// we can easily pass it down -func (d *fieldMapping) UnmarshalYAML(n *yaml.Node) (err error) { - d.m = make(map[string]mapEntry) - if y7s.IsSeq(n) { - err = y7s.EachSeq(n, func(n *yaml.Node) error { - a, err := d.unmarshalMappingNode(n) - d.m[a.Column] = a - return err - }) - } else { - err = y7s.EachMap(n, func(k, n *yaml.Node) error { - a, err := d.unmarshalMappingNode(n) - if a.Column == "" { - err = y7s.DecodeScalar(k, "fieldMapping column", &a.Column) - if err != nil { - return err - } - } - - d.m[a.Column] = a - return err - }) - } - - return -} - -func (d *fieldMapping) unmarshalMappingNode(n *yaml.Node) (out mapEntry, err error) { - if y7s.IsKind(n, yaml.ScalarNode) { - err = y7s.DecodeScalar(n, "Column", &out.Column) - if err != nil { - return - } - err = y7s.DecodeScalar(n, "Field", &out.Field) - return - } - - // @todo we're omitting errors because there will be a bunch due to invalid - // resource field types. This might be a bit unstable as other errors may - // also get ignored. - // - // A potential fix would be to firstly unmarshal into an any, check errors - // and then unmarshal into the resource while omitting errors. - n.Decode(&out) - - err = y7s.EachMap(n, func(k, v *yaml.Node) error { - switch strings.ToLower(k.Value) { - case "skip": - if v.Value == "/" { - out.Skip = true - } - } - return nil - }) - - return -} - func (d *auxYamlDoc) procMappingRefs(in map[string]string) (out map[string]envoyx.Ref, scope envoyx.Scope) { out = make(map[string]envoyx.Ref) diff --git a/server/compose/rest/record.go b/server/compose/rest/record.go index 21854bfc9..786b11467 100644 --- a/server/compose/rest/record.go +++ b/server/compose/rest/record.go @@ -11,6 +11,7 @@ import ( "strings" "time" + composeEnvoy "github.com/cortezaproject/corteza/server/compose/envoy" "github.com/cortezaproject/corteza/server/compose/rest/request" "github.com/cortezaproject/corteza/server/compose/service" "github.com/cortezaproject/corteza/server/compose/types" @@ -19,9 +20,9 @@ import ( "github.com/cortezaproject/corteza/server/pkg/dal" "github.com/cortezaproject/corteza/server/pkg/envoy" "github.com/cortezaproject/corteza/server/pkg/envoy/csv" - ejson "github.com/cortezaproject/corteza/server/pkg/envoy/json" - "github.com/cortezaproject/corteza/server/pkg/envoy/resource" + envoyJson "github.com/cortezaproject/corteza/server/pkg/envoy/json" estore "github.com/cortezaproject/corteza/server/pkg/envoy/store" + "github.com/cortezaproject/corteza/server/pkg/envoyx" "github.com/cortezaproject/corteza/server/pkg/filter" "github.com/cortezaproject/corteza/server/pkg/revisions" "github.com/cortezaproject/corteza/server/store" @@ -393,113 +394,207 @@ func (ctrl *Record) ImportInit(ctx context.Context, r *request.RecordImportInit) return ctrl.importSession.Create(ctx, f, r.Upload.Filename, ct, r.NamespaceID, r.ModuleID) } -func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) (interface{}, error) { +// @todo :') +func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) (_ interface{}, err error) { var ( - err error + ns *types.Namespace + mod *types.Module ) - - // Access control. - if _, err = ctrl.module.FindByID(ctx, r.NamespaceID, r.ModuleID); err != nil { + if mod, err = ctrl.module.FindByID(ctx, r.NamespaceID, r.ModuleID); err != nil { + return nil, err + } + if ns, err = ctrl.namespace.FindByID(ctx, r.NamespaceID); err != nil { return nil, err } - // Check if session ok - ses, err := ctrl.importSession.FindByID(ctx, r.SessionID) - if err != nil { - return nil, err - } - - ses.Fields = make(map[string]string) - err = json.Unmarshal(r.Fields, &ses.Fields) - if err != nil { - return nil, err - } - - ses.OnError = r.OnError - - // Errors are presented in the session + var importSession *service.RecordImportSession err = func() (err error) { - if ses.Progress.StartedAt != nil { - return fmt.Errorf("unable to start import: import session already active") + // Check if session ok + { + importSession, err = ctrl.importSession.FindByID(ctx, r.SessionID) + if err != nil { + return + } + + if importSession.Progress.StartedAt != nil { + return fmt.Errorf("unable to start import: import session already active") + } } sa := time.Now() - ses.Progress.StartedAt = &sa + importSession.Progress.StartedAt = &sa - // Prepare additional metadata - tpl := resource.NewComposeRecordTemplate( - strconv.FormatUint(ses.ModuleID, 10), - strconv.FormatUint(ses.NamespaceID, 10), - ses.Name, - false, - resource.MapToMappingTplSet(ses.Fields), - ses.Key, + // Some prereq + { + importSession.Fields = make(map[string]string) + err = json.Unmarshal(r.Fields, &importSession.Fields) + if err != nil { + return err + } + + importSession.OnError = r.OnError + } + + // Prep envoy bits + var ( + envoySvc *envoyx.Service + encodeParams envoyx.EncodeParams + + nodeScope envoyx.Scope + nodes envoyx.NodeSet + node *envoyx.Node + storeEncoder = composeEnvoy.StoreEncoder{} ) + { + envoySvc = envoyx.New() + // @todo add when/if needed + envoySvc.AddEncoder(envoyx.EncodeTypeStore, + storeEncoder, + ) - // Shape the data - ses.Resources = append(ses.Resources, tpl) - rt := resource.ComposeRecordShaper() - ses.Resources, err = resource.Shape(ses.Resources, rt) + encodeParams = envoyx.EncodeParams{ + Type: envoyx.EncodeTypeStore, + Params: map[string]any{ + "storer": service.DefaultStore, + "dal": dal.Service(), + }, + DeferOk: func() { + importSession.Progress.Completed++ + }, + DeferNok: func(err error) error { + importSession.Progress.Failed++ - // Build - cfg := &estore.EncoderConfig{ - // For now the identifier is ignored, so this will never occur - OnExisting: resource.Skip, - DeferOk: func() { - ses.Progress.Completed++ - }, - } - cfg.DeferNok = func(err error) error { - ses.Progress.Failed++ - - if ses.Progress.FailLog == nil { - ses.Progress.FailLog = &service.FailLog{ - Errors: make(service.ErrorIndex), - } - } - - if rve, is := err.(*types.RecordValueErrorSet); is { - for _, ve := range rve.Set { - for k, v := range ve.Meta { - ses.Progress.FailLog.Errors.Add(fmt.Sprintf("%s %s %v", ve.Kind, k, v)) + if importSession.Progress.FailLog == nil { + importSession.Progress.FailLog = &service.FailLog{ + Errors: make(service.ErrorIndex), + } } + + if rve, is := err.(*types.RecordValueErrorSet); is { + for _, ve := range rve.Set { + for k, v := range ve.Meta { + importSession.Progress.FailLog.Errors.Add(fmt.Sprintf("%s %s %v", ve.Kind, k, v)) + } + } + } else { + importSession.Progress.FailLog.Errors.Add(err.Error()) + } + + if len(importSession.Progress.FailLog.Records) < service.IMPORT_ERROR_MAX_INDEX_COUNT { + // +1 because we indexed them with 1 before + importSession.Progress.FailLog.Records = append(importSession.Progress.FailLog.Records, int(importSession.Progress.Completed)+1) + } else { + importSession.Progress.FailLog.RecordsTruncated = true + } + + if importSession.OnError == service.IMPORT_ON_ERROR_SKIP { + return nil + } + return err + }, + } + + nodeScope = envoyx.Scope{ + ResourceType: types.NamespaceResourceType, + Identifiers: envoyx.MakeIdentifiers(importSession.NamespaceID), + } + + fieldMapping := map[string]envoyx.MapEntry{} + for c, f := range importSession.Fields { + fieldMapping[c] = envoyx.MapEntry{ + Column: c, + Field: f, } - } else { - ses.Progress.FailLog.Errors.Add(err.Error()) } - if len(ses.Progress.FailLog.Records) < service.IMPORT_ERROR_MAX_INDEX_COUNT { - // +1 because we indexed them with 1 before - ses.Progress.FailLog.Records = append(ses.Progress.FailLog.Records, int(ses.Progress.Completed)+1) - } else { - ses.Progress.FailLog.RecordsTruncated = true + nsNode := &envoyx.Node{ + Resource: ns, + ResourceType: types.NamespaceResourceType, + Identifiers: envoyx.MakeIdentifiers(ns.Slug, ns.ID), + Scope: nodeScope, + Placeholder: true, } - if ses.OnError == service.IMPORT_ON_ERROR_SKIP { - return nil + modNode := &envoyx.Node{ + Resource: mod, + ResourceType: types.ModuleResourceType, + Identifiers: envoyx.MakeIdentifiers(mod.Handle, mod.ID), + Scope: nodeScope, + References: map[string]envoyx.Ref{ + "NamespaceID": { + ResourceType: types.NamespaceResourceType, + Identifiers: nsNode.Identifiers, + Scope: nsNode.Scope, + }, + }, + Placeholder: true, } - return err - } - se := estore.NewStoreEncoder(service.DefaultStore, dal.Service(), cfg) - bld := envoy.NewBuilder(se) - g, err := bld.Build(ctx, ses.Resources...) - if err != nil { - return err + + node = &envoyx.Node{ + Datasource: &composeEnvoy.RecordDatasource{ + Mapping: envoyx.DatasourceMapping{ + SourceIdent: importSession.Name, + References: map[string]string{}, + Scope: map[string]string{}, + Defaultable: false, + Mapping: envoyx.FieldMapping{ + Map: fieldMapping, + }, + }, + }, + ResourceType: composeEnvoy.ComposeRecordDatasourceAuxType, + Identifiers: envoyx.MakeIdentifiers(importSession.Name), + + References: map[string]envoyx.Ref{ + "ModuleID": { + ResourceType: types.ModuleResourceType, + Identifiers: envoyx.MakeIdentifiers(importSession.ModuleID), + Scope: nodeScope, + }, + "NamespaceID": { + ResourceType: types.NamespaceResourceType, + Identifiers: envoyx.MakeIdentifiers(importSession.NamespaceID), + Scope: nodeScope, + }, + }, + Scope: nodeScope, + } + + nodes = envoyx.NodeSet{nsNode, modNode, node} } - // Encode - err = envoy.Encode(ctx, g, se) - now := time.Now() - ses.Progress.FinishedAt = &now - if err != nil { - ses.Progress.FailReason = err.Error() - return err - } + // encoding stuff + var ( + depGraph *envoyx.DepGraph + ) + { + depGraph, err = envoySvc.Bake(ctx, encodeParams, importSession.Providers, nodes...) + if err != nil { + return + } - return + { + // @todo this is temporary because the service's logic is a bit flawed for this case + err = storeEncoder.Prepare(ctx, encodeParams, composeEnvoy.ComposeRecordDatasourceAuxType, envoyx.NodeSet{node}) + if err != nil { + return + } + + err = storeEncoder.Encode(ctx, encodeParams, composeEnvoy.ComposeRecordDatasourceAuxType, envoyx.NodeSet{node}, depGraph) + // @note err is handled lower down; bare with + } + + // err = envoySvc.Encode(ctx, encodeParams, depGraph) + now := time.Now() + importSession.Progress.FinishedAt = &now + if err != nil { + importSession.Progress.FailReason = err.Error() + return + } + return + } }() - - return ses, ctrl.record.RecordImport(ctx, err) + return importSession, ctrl.record.RecordImport(ctx, err) } func (ctrl *Record) ImportProgress(ctx context.Context, r *request.RecordImportProgress) (interface{}, error) { @@ -559,7 +654,7 @@ func (ctrl *Record) Export(ctx context.Context, r *request.RecordExport) (interf switch strings.ToLower(r.Ext) { case "json", "jsonl", "ldjson", "ndjson": contentType = "application/jsonl" - encoder = ejson.NewBulkRecordEncoder(&ejson.EncoderConfig{ + encoder = envoyJson.NewBulkRecordEncoder(&envoyJson.EncoderConfig{ Fields: fx, Timezone: r.Timezone, }) diff --git a/server/compose/service/import_session.go b/server/compose/service/import_session.go index f25fd1173..ac2a6e3bb 100644 --- a/server/compose/service/import_session.go +++ b/server/compose/service/import_session.go @@ -9,23 +9,29 @@ import ( "time" "github.com/cortezaproject/corteza/server/pkg/auth" - "github.com/cortezaproject/corteza/server/pkg/envoy" - "github.com/cortezaproject/corteza/server/pkg/envoy/csv" - "github.com/cortezaproject/corteza/server/pkg/envoy/json" - "github.com/cortezaproject/corteza/server/pkg/envoy/resource" + "github.com/cortezaproject/corteza/server/pkg/envoyx" + + envoyCsv "github.com/cortezaproject/corteza/server/pkg/envoyx/csv" + envoyJson "github.com/cortezaproject/corteza/server/pkg/envoyx/json" ) type ( - recordSet []*recordImportSession + recordSet []*RecordImportSession importSession struct { l sync.Mutex records recordSet } + countableProvider interface { + envoyx.Provider + Count() uint64 + Fields() []string + } + ImportSessionService interface { - Create(ctx context.Context, f io.ReadSeeker, name, contentType string, namespaceID, moduleID uint64) (*recordImportSession, error) - FindByID(ctx context.Context, sessionID uint64) (*recordImportSession, error) + Create(ctx context.Context, f io.ReadSeeker, name, contentType string, namespaceID, moduleID uint64) (*RecordImportSession, error) + FindByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error) DeleteByID(ctx context.Context, sessionID uint64) error } ) @@ -46,12 +52,12 @@ func (svc *importSession) indexOf(userID, sessionID uint64) int { return -1 } -func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, contentType string, namespaceID, moduleID uint64) (*recordImportSession, error) { +func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, contentType string, namespaceID, moduleID uint64) (_ *RecordImportSession, err error) { svc.l.Lock() defer svc.l.Unlock() // Prepare the session - sh := &recordImportSession{ + sh := &RecordImportSession{ Name: name, SessionID: nextID(), UserID: auth.GetIdentityFromContext(ctx).Identity(), @@ -66,49 +72,31 @@ func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, con UpdatedAt: time.Now(), } - // Decoders; We only need to do csv & yaml here - cd := csv.Decoder() - jd := json.Decoder() - - // This will really be at most 1 - var err error - do := &envoy.DecoderOpts{ - Name: name, - Path: "", - } - - sh.Resources, err = func() ([]resource.Interface, error) { - if cd.CanDecodeFile(f) || cd.CanDecodeMime(contentType) { + rd, err := func() (pp countableProvider, err error) { + if envoyCsv.CanDecodeFile(f) || envoyCsv.CanDecodeMime(contentType) { f.Seek(0, 0) - return cd.Decode(ctx, f, do) + return envoyCsv.Decoder(f, name) } - f.Seek(0, 0) - if jd.CanDecodeFile(f) || jd.CanDecodeMime(contentType) { + if envoyJson.CanDecodeFile(f) || envoyJson.CanDecodeMime(contentType) { f.Seek(0, 0) - return jd.Decode(ctx, f, do) + return envoyJson.Decoder(f, name) } return nil, fmt.Errorf("compose.service.RecordImportFormatNotSupported") }() - if err != nil { - return nil, err + return } - // Get some metadata - n, ok := (sh.Resources[0]).(*resource.ResourceDataset) - if !ok { - // @todo move this logic to service and use action/error pattern - return nil, fmt.Errorf("compose.service.RecordImportFormatNotSupported") - } + sh.Providers = append(sh.Providers, rd) prepKey := func(k string) string { return strings.TrimSpace(strings.ToLower(k)) } - sh.Progress.EntryCount = n.P.Count() - for _, f := range n.P.Fields() { + sh.Progress.EntryCount = rd.Count() + for _, f := range rd.Fields() { sh.Fields[f] = "" // @todo improve this bit @@ -123,7 +111,7 @@ func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, con return sh, nil } -func (svc *importSession) FindByID(ctx context.Context, sessionID uint64) (*recordImportSession, error) { +func (svc *importSession) FindByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error) { svc.l.Lock() defer svc.l.Unlock() diff --git a/server/compose/service/record.go b/server/compose/service/record.go index e460c30ef..43623d086 100644 --- a/server/compose/service/record.go +++ b/server/compose/service/record.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "github.com/cortezaproject/corteza/server/pkg/envoyx" "github.com/cortezaproject/corteza/server/pkg/filter" "github.com/cortezaproject/corteza/server/pkg/revisions" "github.com/spf13/cast" @@ -153,7 +154,7 @@ type ( Country() string } - recordImportSession struct { + RecordImportSession struct { Name string `json:"-"` SessionID uint64 `json:"sessionID,string"` UserID uint64 `json:"userID,string"` @@ -169,6 +170,7 @@ type ( UpdatedAt time.Time `json:"updatedAt"` Resources []resource.Interface `json:"-"` + Providers []envoyx.Provider `json:"-"` } RecordImportProgress struct { diff --git a/server/pkg/envoyx/datasource.go b/server/pkg/envoyx/datasource.go index 3aae7864d..4aeb7f3a4 100644 --- a/server/pkg/envoyx/datasource.go +++ b/server/pkg/envoyx/datasource.go @@ -1,6 +1,12 @@ package envoyx -import "context" +import ( + "context" + "strings" + + "github.com/cortezaproject/corteza/server/pkg/y7s" + "gopkg.in/yaml.v3" +) type ( Provider interface { @@ -15,6 +21,31 @@ type ( Reset(ctx context.Context) error SetProvider(Provider) bool } + + MapEntry struct { + Column string + Field string + Skip bool + } + + FieldMapping struct { + // @note This had to be like so to simplify decoding + Map map[string]MapEntry + } + + DatasourceMapping struct { + SourceIdent string `yaml:"source"` + KeyField []string `yaml:"key"` + References map[string]string + Scope map[string]string + + // Defaultable indicates wether the mapping should keep the values where + // the ident is not explicitly mapped. + // + // When true, the value is assigned to the given identifier. + Defaultable bool `yaml:"defaultable"` + Mapping FieldMapping + } ) func SetDecoderSources(nn NodeSet, dd ...Provider) { @@ -30,3 +61,62 @@ func SetDecoderSources(nn NodeSet, dd ...Provider) { } } } + +// UnmarshalYAML is used to get the yaml parsed into a series of nodes so +// we can easily pass it down +func (d *FieldMapping) UnmarshalYAML(n *yaml.Node) (err error) { + d.Map = make(map[string]MapEntry) + if y7s.IsSeq(n) { + err = y7s.EachSeq(n, func(n *yaml.Node) error { + a, err := d.unmarshalMappingNode(n) + d.Map[a.Column] = a + return err + }) + } else { + err = y7s.EachMap(n, func(k, n *yaml.Node) error { + a, err := d.unmarshalMappingNode(n) + if a.Column == "" { + err = y7s.DecodeScalar(k, "fieldMapping column", &a.Column) + if err != nil { + return err + } + } + + d.Map[a.Column] = a + return err + }) + } + + return +} + +func (d *FieldMapping) unmarshalMappingNode(n *yaml.Node) (out MapEntry, err error) { + if y7s.IsKind(n, yaml.ScalarNode) { + err = y7s.DecodeScalar(n, "Column", &out.Column) + if err != nil { + return + } + err = y7s.DecodeScalar(n, "Field", &out.Field) + return + } + + // @todo we're omitting errors because there will be a bunch due to invalid + // resource field types. This might be a bit unstable as other errors may + // also get ignored. + // + // A potential fix would be to firstly unmarshal into an any, check errors + // and then unmarshal into the resource while omitting errors. + n.Decode(&out) + + err = y7s.EachMap(n, func(k, v *yaml.Node) error { + switch strings.ToLower(k.Value) { + case "skip": + if v.Value == "/" { + out.Skip = true + } + } + return nil + }) + + return +} diff --git a/server/pkg/envoyx/envoy.go b/server/pkg/envoyx/envoy.go index d0dd6e3b5..5a74b8a93 100644 --- a/server/pkg/envoyx/envoy.go +++ b/server/pkg/envoyx/envoy.go @@ -87,6 +87,7 @@ type ( Encoder EncoderConfig // @note these are only used by records since v1 did just that + // @todo make these more expanded DeferOk func() DeferNok func(error) error Defer func()