diff --git a/compose/rest/record.go b/compose/rest/record.go index 75978bf97..3dbb3cd5f 100644 --- a/compose/rest/record.go +++ b/compose/rest/record.go @@ -5,7 +5,10 @@ import ( "encoding/csv" "encoding/json" "fmt" - "github.com/cortezaproject/corteza-server/compose/decoder" + "net/http" + "strconv" + "strings" + "github.com/cortezaproject/corteza-server/compose/encoder" "github.com/cortezaproject/corteza-server/compose/rest/request" "github.com/cortezaproject/corteza-server/compose/service" @@ -13,15 +16,10 @@ import ( "github.com/cortezaproject/corteza-server/pkg/api" "github.com/cortezaproject/corteza-server/pkg/corredor" "github.com/cortezaproject/corteza-server/pkg/filter" - "github.com/cortezaproject/corteza-server/pkg/mime" "github.com/cortezaproject/corteza-server/pkg/payload" "github.com/cortezaproject/corteza-server/store" systemService "github.com/cortezaproject/corteza-server/system/service" systemTypes "github.com/cortezaproject/corteza-server/system/types" - "net/http" - "path" - "strconv" - "strings" ) type ( @@ -273,14 +271,7 @@ func (ctrl *Record) Upload(ctx context.Context, r *request.RecordUpload) (interf } func (ctrl *Record) ImportInit(ctx context.Context, r *request.RecordImportInit) (interface{}, error) { - var ( - err error - recordDecoder service.Decoder - entryCount uint64 - ) - - // Access control. - if _, err = ctrl.module.With(ctx).FindByID(r.NamespaceID, r.ModuleID); err != nil { + if _, err := ctrl.module.With(ctx).FindByID(r.NamespaceID, r.ModuleID); err != nil { return nil, err } @@ -290,55 +281,7 @@ func (ctrl *Record) ImportInit(ctx context.Context, r *request.RecordImportInit) } defer f.Close() - _, ext, err := mime.Type(f) - if err != nil { - return nil, err - } - - if ext == "txt" { - if is, err := mime.JsonL(f); err != nil { - return nil, err - } else if is { - ext = "jsonl" - } else { - // As last resort, use extension of the upload filename - ext = strings.TrimLeft(path.Ext(r.Upload.Filename), ".") - } - } - - // determine decoder - switch strings.ToLower(ext) { - case "json", "jsonl", "ldjson", "ndjson": - recordDecoder = decoder.NewStructuredDecoder(json.NewDecoder(f), f) - - case "csv": - recordDecoder = decoder.NewFlatReader(csv.NewReader(f), f) - - default: - // copied here from service/errors.go for backward compatibility - // @todo move this logic to service and use action/error pattern - return nil, fmt.Errorf("compose.service.RecordImportFormatNotSupported") - } - - entryCount, err = recordDecoder.EntryCount() - if err != nil { - return nil, err - } - - header := recordDecoder.Header() - hh := make(map[string]string) - for _, h := range header { - hh[h] = "" - } - - return ctrl.importSession.SetByID( - ctx, - 0, - r.NamespaceID, - r.ModuleID, - hh, - &service.RecordImportProgress{EntryCount: entryCount}, - recordDecoder) + return ctrl.importSession.Create(ctx, f, r.Upload.Filename, r.NamespaceID, r.ModuleID) } func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) (interface{}, error) { @@ -365,9 +308,8 @@ func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) ( ses.OnError = r.OnError - // @todo routine - ctrl.record.With(ctx).Import(ses, ctrl.importSession) - + // Errors are presented in the session + ctrl.record.With(ctx).Import(ses) return ses, nil } diff --git a/compose/service/import_session.go b/compose/service/import_session.go index 905a7e1df..af843f673 100644 --- a/compose/service/import_session.go +++ b/compose/service/import_session.go @@ -3,13 +3,19 @@ package service import ( "context" "fmt" - "github.com/cortezaproject/corteza-server/pkg/auth" + "io" "sync" "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" ) type ( - recordSet []*RecordImportSession + recordSet []*recordImportSession importSession struct { l sync.Mutex @@ -17,8 +23,8 @@ type ( } ImportSessionService interface { - FindByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error) - SetByID(ctx context.Context, sessionID, namespaceID, moduleID uint64, fields map[string]string, progress *RecordImportProgress, decoder Decoder) (*RecordImportSession, error) + Create(ctx context.Context, f io.ReadSeeker, name string, namespaceID, moduleID uint64) (*recordImportSession, error) + FindByID(ctx context.Context, sessionID uint64) (*recordImportSession, error) DeleteByID(ctx context.Context, sessionID uint64) error } ) @@ -39,7 +45,74 @@ func (svc *importSession) indexOf(userID, sessionID uint64) int { return -1 } -func (svc *importSession) FindByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error) { +func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name string, namespaceID, moduleID uint64) (*recordImportSession, error) { + svc.l.Lock() + defer svc.l.Unlock() + + // Prepare the session + sh := &recordImportSession{ + Name: name, + SessionID: nextID(), + UserID: auth.GetIdentityFromContext(ctx).Identity(), + NamespaceID: namespaceID, + ModuleID: moduleID, + + OnError: IMPORT_ON_ERROR_FAIL, + Fields: make(map[string]string), + Progress: &RecordImportProgress{}, + + CreatedAt: time.Now(), + 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) { + f.Seek(0, 0) + return cd.Decode(ctx, f, do) + } + + f.Seek(0, 0) + if jd.CanDecodeFile(f) { + f.Seek(0, 0) + return jd.Decode(ctx, f, do) + } + + return nil, fmt.Errorf("compose.service.RecordImportFormatNotSupported") + }() + + if err != nil { + return nil, err + } + + // 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.Progress.EntryCount = n.P.Count() + for _, f := range n.P.Fields() { + sh.Fields[f] = "" + } + + // Create it + svc.records = append(svc.records, sh) + return sh, nil +} + +func (svc *importSession) FindByID(ctx context.Context, sessionID uint64) (*recordImportSession, error) { svc.l.Lock() defer svc.l.Unlock() @@ -51,48 +124,6 @@ func (svc *importSession) FindByID(ctx context.Context, sessionID uint64) (*Reco return nil, fmt.Errorf("compose.service.RecordImportSessionNotFound") } -func (svc *importSession) SetByID(ctx context.Context, sessionID, namespaceID, moduleID uint64, fields map[string]string, progress *RecordImportProgress, decoder Decoder) (*RecordImportSession, error) { - svc.l.Lock() - defer svc.l.Unlock() - - userID := auth.GetIdentityFromContext(ctx).Identity() - i := svc.indexOf(userID, sessionID) - var ris *RecordImportSession - - if i >= 0 { - ris = svc.records[i] - } else { - ris = &RecordImportSession{ - SessionID: nextID(), - CreatedAt: time.Now(), - } - svc.records = append(svc.records, ris) - ris.UserID = userID - } - ris.UpdatedAt = time.Now() - - if namespaceID > 0 { - ris.NamespaceID = namespaceID - } - if moduleID > 0 { - ris.ModuleID = moduleID - } - if fields != nil { - ris.Fields = fields - } - if progress != nil { - ris.Progress = *progress - } - - if ris.Progress.FinishedAt != nil { - ris.Decoder = nil - } else if decoder != nil { - ris.Decoder = decoder - } - - return ris, nil -} - // https://stackoverflow.com/a/37335777 func remove(s recordSet, i int) recordSet { s[len(s)-1], s[i] = s[i], s[len(s)-1] diff --git a/compose/service/import_session_test.go b/compose/service/import_session_test.go deleted file mode 100644 index f7672acaf..000000000 --- a/compose/service/import_session_test.go +++ /dev/null @@ -1,111 +0,0 @@ -package service - -import ( - "context" - "testing" - - "github.com/stretchr/testify/require" - "go.uber.org/zap" -) - -func TestImportSession_FindByID(t *testing.T) { - var ctx = context.Background() - DefaultLogger = zap.New(nil, nil) - svc := ImportSession() - ss, _ := svc.SetByID(ctx, 1, 0, 0, nil, nil, nil) - sid := ss.SessionID - - t.Run("Found", func(t *testing.T) { - s, err := svc.FindByID(ctx, sid) - require.True(t, - s != nil, - "Session should be found", - ) - - require.True(t, - err == nil, - "Returned with error", - ) - }) - - t.Run("Not found", func(t *testing.T) { - s, err := svc.FindByID(ctx, sid+1) - require.True(t, - s == nil, - "Session should not be found", - ) - - require.True(t, - err != nil, - "Error should not be nil", - ) - }) -} - -func TestImportSession_SetByID(t *testing.T) { - var ctx = context.Background() - DefaultLogger = zap.New(nil, nil) - svc := ImportSession() - - t.Run("New", func(t *testing.T) { - ss, err := svc.SetByID(ctx, 1, 0, 0, nil, nil, nil) - require.True(t, - len(svc.records) == 1 && ss != nil, - "Session should be created", - ) - - require.True(t, - err == nil, - "Returned with error", - ) - }) - - t.Run("Existing", func(t *testing.T) { - svc := ImportSession() - ss, err := svc.SetByID(ctx, 1, 0, 0, nil, nil, nil) - ns, err := svc.SetByID(ctx, ss.SessionID, 0, 0, nil, nil, nil) - require.True(t, - len(svc.records) == 1 && ns != nil && ss.SessionID == ns.SessionID, - "Existing session should be edited", - ) - - require.True(t, - err == nil, - "Returned with error", - ) - }) -} - -func TestImportSession_DeleteByID(t *testing.T) { - var ctx = context.Background() - DefaultLogger = zap.New(nil, nil) - svc := ImportSession() - ss, _ := svc.SetByID(ctx, 1, 0, 0, nil, nil, nil) - - t.Run("Delete existing", func(t *testing.T) { - err := svc.DeleteByID(ctx, ss.SessionID) - require.True(t, - len(svc.records) == 0, - "Session should be deleted", - ) - - require.True(t, - err == nil, - "Returned with error", - ) - }) - - t.Run("Session not found", func(t *testing.T) { - ss, _ := svc.SetByID(ctx, 1, 0, 0, nil, nil, nil) - err := svc.DeleteByID(ctx, ss.SessionID+1) - require.True(t, - len(svc.records) == 1, - "Session should not deleted", - ) - - require.True(t, - err == nil, - "Returned with error", - ) - }) -} diff --git a/compose/service/record.go b/compose/service/record.go index e4dc3fcb4..5a9d80a61 100644 --- a/compose/service/record.go +++ b/compose/service/record.go @@ -3,6 +3,10 @@ package service import ( "context" "fmt" + "regexp" + "strconv" + "time" + "github.com/cortezaproject/corteza-server/compose/decoder" "github.com/cortezaproject/corteza-server/compose/service/event" "github.com/cortezaproject/corteza-server/compose/service/values" @@ -10,13 +14,13 @@ import ( "github.com/cortezaproject/corteza-server/pkg/actionlog" "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/corredor" + "github.com/cortezaproject/corteza-server/pkg/envoy" + "github.com/cortezaproject/corteza-server/pkg/envoy/resource" + estore "github.com/cortezaproject/corteza-server/pkg/envoy/store" "github.com/cortezaproject/corteza-server/pkg/errors" "github.com/cortezaproject/corteza-server/pkg/eventbus" "github.com/cortezaproject/corteza-server/pkg/label" "github.com/cortezaproject/corteza-server/store" - "regexp" - "strconv" - "time" ) const ( @@ -81,7 +85,7 @@ type ( Report(namespaceID, moduleID uint64, metrics, dimensions, filter string) (interface{}, error) Find(filter types.RecordFilter) (set types.RecordSet, f types.RecordFilter, err error) Export(types.RecordFilter, Encoder) error - Import(*RecordImportSession, ImportSessionService) error + Import(*recordImportSession) error Create(record *types.Record) (*types.Record, error) Update(record *types.Record) (*types.Record, error) @@ -108,17 +112,21 @@ type ( Records(fields map[string]string, Create decoder.RecordCreator) error } - RecordImportSession struct { - Decoder Decoder `json:"-"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` - OnError string `json:"onError"` - SessionID uint64 `json:"sessionID,string"` - UserID uint64 `json:"userID,string"` - NamespaceID uint64 `json:"namespaceID,string"` - ModuleID uint64 `json:"moduleID,string"` - Fields map[string]string `json:"fields"` - Progress RecordImportProgress `json:"progress"` + recordImportSession struct { + Name string `json:"-"` + SessionID uint64 `json:"sessionID,string"` + UserID uint64 `json:"userID,string"` + NamespaceID uint64 `json:"namespaceID,string"` + ModuleID uint64 `json:"moduleID,string"` + + OnError string `json:"onError"` + Fields map[string]string `json:"fields"` + Progress *RecordImportProgress `json:"progress"` + + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` + + Resources []resource.Interface `json:"-"` } RecordImportProgress struct { @@ -313,50 +321,63 @@ func (svc record) Find(filter types.RecordFilter) (set types.RecordSet, f types. return set, f, svc.recordAction(svc.ctx, aProps, RecordActionSearch, err) } -func (svc record) Import(ses *RecordImportSession, ssvc ImportSessionService) (err error) { +func (svc record) Import(ses *recordImportSession) (err error) { var ( aProps = &recordActionProps{} ) - if ses.Decoder == nil { - return nil - } - err = func() (err error) { - if ses.Progress.StartedAt != nil { - return fmt.Errorf("Unable to start import: Import session already active") + return fmt.Errorf("unable to start import: import session already active") } sa := time.Now() ses.Progress.StartedAt = &sa - ssvc.SetByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil) - err = ses.Decoder.Records(ses.Fields, func(mod *types.Record) error { - mod.NamespaceID = ses.NamespaceID - mod.ModuleID = ses.ModuleID - mod.OwnedBy = ses.UserID + // Prepare additional metadata + tpl := resource.NewComposeRecordTemplate( + strconv.FormatUint(ses.ModuleID, 10), + strconv.FormatUint(ses.NamespaceID, 10), + ses.Name, + resource.MapToMappingTplSet(ses.Fields), + ) - _, err := svc.Create(mod) - if err != nil { + // Shape the data + ses.Resources = append(ses.Resources, tpl) + rt := resource.ComposeRecordShaper() + ses.Resources, err = resource.Shape(ses.Resources, rt) + + // Build + cfg := &estore.EncoderConfig{ + // For now the identifier is ignored, so this will never occur + OnExisting: estore.Skip, + Defer: func() { + ses.Progress.Completed++ + }, + } + if ses.OnError == IMPORT_ON_ERROR_SKIP { + cfg.DeferNok = func(err error) error { ses.Progress.Failed++ ses.Progress.FailReason = err.Error() - if ses.OnError == IMPORT_ON_ERROR_FAIL { - fa := time.Now() - ses.Progress.FinishedAt = &fa - ssvc.SetByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil) - return err - } - } else { - ses.Progress.Completed++ + return nil } - return nil - }) + } + se := estore.NewStoreEncoder(svc.store, cfg) + bld := envoy.NewBuilder(se) + g, err := bld.Build(svc.ctx, ses.Resources...) + if err != nil { + return err + } + + // Encode + err = envoy.Encode(svc.ctx, g, se) + ses.Progress.FinishedAt = now() + if err != nil { + ses.Progress.FailReason = err.Error() + return err + } - fa := time.Now() - ses.Progress.FinishedAt = &fa - ssvc.SetByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil) return }()