Rework record importer with envoy

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
 }
 

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]

compose/service/import_session_test.go

@@ -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",
-		)
-	})
-}

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
 	}()