From a8b5e3b0633d4c48fa395b0802786f0c29510a17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Mon, 26 Aug 2019 13:46:55 +0200 Subject: [PATCH] Add record importing from json & csv --- compose/decoder/decoder.go | 145 +++++++++++++++++++++ compose/decoder/record.go | 131 +++++++++++++++++++ compose/internal/service/import_session.go | 133 +++++++++++++++++++ compose/internal/service/record.go | 76 +++++++++++ compose/internal/service/service.go | 17 +-- compose/rest/record.go | 116 +++++++++++++++-- pkg/mime/mime.go | 34 +++++ 7 files changed, 636 insertions(+), 16 deletions(-) create mode 100644 compose/decoder/decoder.go create mode 100644 compose/decoder/record.go create mode 100644 compose/internal/service/import_session.go create mode 100644 pkg/mime/mime.go diff --git a/compose/decoder/decoder.go b/compose/decoder/decoder.go new file mode 100644 index 000000000..8159f8b49 --- /dev/null +++ b/compose/decoder/decoder.go @@ -0,0 +1,145 @@ +package decoder + +import ( + "io" +) + +type ( + multiple uint + + FlatReader interface { + Read() ([]string, error) + } + + StructuredDecoder interface { + Decode(interface{}) error + More() bool + } + + flatReader struct { + r FlatReader + header []string + more bool + } + + structuredDecoder struct { + header []string + d StructuredDecoder + buf []map[string]interface{} + } + + // callbacks + sdCallback func(map[string]interface{}) error + fdCallback func([]string) error +) + +// flat reader +func NewFlatReader(r FlatReader) *flatReader { + return &flatReader{ + r: r, + more: true, + } +} + +func (dec *flatReader) get(fnc fdCallback) error { + v, err := dec.r.Read() + if err == io.EOF { + dec.more = false + return nil + } else if err != nil { + return err + } + + return fnc(v) +} + +func (dec *flatReader) walk(fnc fdCallback) error { + for dec.more { + if err := dec.get(fnc); err != nil { + return err + } + } + return nil +} + +func (dec *flatReader) Header() []string { + if len(dec.header) > 0 { + return dec.header + } + + dec.get(func(rtr []string) error { + dec.header = rtr + return nil + }) + + return dec.header +} + +// structured decoder +func NewStructuredDecoder(d StructuredDecoder) *structuredDecoder { + return &structuredDecoder{ + d: d, + } +} + +func (dec *structuredDecoder) get(fnc sdCallback) error { + if !dec.d.More() { + return nil + } + + var tmp map[string]interface{} + err := dec.d.Decode(&tmp) + if err != nil { + return err + } + + return fnc(tmp) +} + +func (dec *structuredDecoder) exhaustBuffer(fnc sdCallback) error { + if dec.buf != nil { + for _, b := range dec.buf { + fnc(b) + } + dec.buf = nil + } + return nil +} + +func (dec *structuredDecoder) walk(fnc sdCallback) error { + if err := dec.exhaustBuffer(fnc); err != nil { + return err + } + + for dec.d.More() { + if err := dec.get(fnc); err != nil { + return err + } + } + + return nil +} + +func (dec *structuredDecoder) Header() []string { + if len(dec.header) > 0 { + return dec.header + } + + var tmp []string + dec.get(func(rtr map[string]interface{}) error { + // buffer first row or else it will be lost + dec.buf = append(dec.buf, rtr) + + tmp = make([]string, len(rtr)) + i := 0 + for k := range rtr { + tmp[i] = k + i++ + } + + return nil + }) + + dec.header = tmp + return tmp +} diff --git a/compose/decoder/record.go b/compose/decoder/record.go new file mode 100644 index 000000000..5aa1062b5 --- /dev/null +++ b/compose/decoder/record.go @@ -0,0 +1,131 @@ +package decoder + +import ( + "errors" + "fmt" + "strconv" + "time" + + "github.com/cortezaproject/corteza-server/compose/types" +) + +type ( + RecordCreator func(mod *types.Record) error +) + +func fmtTime(tp string) (time.Time, error) { + return time.Parse(time.RFC3339, tp) +} +func fmtTimePtr(tp string) (*time.Time, error) { + t, err := fmtTime(tp) + if err != nil { + return nil, err + } + return &t, nil +} + +func mapify(header []string, values []string) map[string]string { + rtr := make(map[string]string) + for i, v := range values { + rtr[header[i]] = v + } + + return rtr +} + +func setSystemField(r *types.Record, name, value string) (is bool, err error) { + switch name { + case "recordID", "ID": + r.ID, err = strconv.ParseUint(value, 10, 64) + case "moduleID": + r.ModuleID, err = strconv.ParseUint(value, 10, 64) + case "namespaceID": + r.NamespaceID, err = strconv.ParseUint(value, 10, 64) + case "ownedBy": + r.OwnedBy, err = strconv.ParseUint(value, 10, 64) + case "createdBy": + r.CreatedBy, err = strconv.ParseUint(value, 10, 64) + case "createdAt": + r.CreatedAt, err = fmtTime(value) + case "updatedBy": + r.UpdatedBy, err = strconv.ParseUint(value, 10, 64) + case "updatedAt": + r.UpdatedAt, err = fmtTimePtr(value) + case "deletedBy": + r.DeletedBy, err = strconv.ParseUint(value, 10, 64) + case "deletedAt": + r.DeletedAt, err = fmtTimePtr(value) + default: + return false, err + } + return true, err +} + +func (dec flatReader) Records(fields map[string]string, Create RecordCreator) error { + header := dec.Header() + + err := dec.walk(func(row []string) error { + mapped := mapify(header, row) + r := types.Record{} + rvs := types.RecordValueSet{} + + i := 0 + for imp, rec := range fields { + if rec == "" { + return errors.New("Can not import record: Record field not defined") + } + + val := mapped[imp] + if system, err := setSystemField(&r, rec, val); err != nil { + return err + } else if !system { + rv := types.RecordValue{ + Name: rec, + Value: val, + Place: uint(i), + } + i++ + + rvs = append(rvs, &rv) + } + } + + r.Values = rvs + return Create(&r) + }) + + return err +} + +func (dec structuredDecoder) Records(fields map[string]string, Create RecordCreator) error { + err := dec.walk(func(entry map[string]interface{}) error { + r := types.Record{} + rvs := types.RecordValueSet{} + + i := 0 + for imp, rec := range fields { + if rec == "" { + return errors.New("Can not import record: Record field not defined") + } + + val := fmt.Sprintf("%v", entry[imp]) + if system, err := setSystemField(&r, rec, val); err != nil { + return err + } else if !system { + rv := types.RecordValue{ + Name: rec, + Value: val, + Place: uint(i), + } + i++ + + rvs = append(rvs, &rv) + } + } + + r.Values = rvs + return Create(&r) + }) + + return err +} diff --git a/compose/internal/service/import_session.go b/compose/internal/service/import_session.go new file mode 100644 index 000000000..7717649b6 --- /dev/null +++ b/compose/internal/service/import_session.go @@ -0,0 +1,133 @@ +package service + +import ( + "context" + "sync" + "time" + + "github.com/cortezaproject/corteza-server/internal/auth" + "github.com/titpetric/factory" + + "github.com/pkg/errors" + "go.uber.org/zap" +) + +type ( + recordSet []*RecordImportSession + + importSession struct { + l sync.Mutex + logger *zap.Logger + + records recordSet + } + + ImportSessionService interface { + FindRecordByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error) + SetRecordByID(ctx context.Context, sessionID, namespaceID, moduleID uint64, fields map[string]string, progress *RecordImportProgress, decoder Decoder) (*RecordImportSession, error) + DeleteRecordByID(ctx context.Context, sessionID uint64) error + } +) + +func ImportSession() ImportSessionService { + return &importSession{ + logger: DefaultLogger.Named("importSession"), + records: recordSet{}, + } +} + +func (svc importSession) indexOf(userID, sessionID uint64) int { + for i, r := range svc.records { + if r.SessionID == sessionID && r.UserID == userID { + return i + } + } + + return -1 +} + +func (svc *importSession) FindRecordByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error) { + svc.l.Lock() + defer svc.l.Unlock() + + userID := auth.GetIdentityFromContext(ctx).Identity() + i := svc.indexOf(userID, sessionID) + if i >= 0 { + return svc.records[i], nil + } + return nil, errors.New("Can't access session: session not found") +} + +func (svc *importSession) SetRecordByID(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: factory.Sonyflake.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] + return s[:len(s)-1] +} + +func (svc *importSession) DeleteRecordByID(ctx context.Context, sessionID uint64) error { + svc.l.Lock() + defer svc.l.Unlock() + + userID := auth.GetIdentityFromContext(ctx).Identity() + i := svc.indexOf(userID, sessionID) + + if i >= 0 { + svc.records = remove(svc.records, i) + } + return nil +} + +// @todo run this in some interval +func (svc *importSession) clean(ctx context.Context) { + svc.l.Lock() + defer svc.l.Unlock() + + for i := len(svc.records) - 1; i >= 0; i-- { + r := svc.records[i] + if time.Now().After(r.UpdatedAt.Add(time.Hour * 24 * 3)) { + svc.records = remove(svc.records, i) + } + } +} diff --git a/compose/internal/service/record.go b/compose/internal/service/record.go index 846d8bf85..c67472c08 100644 --- a/compose/internal/service/record.go +++ b/compose/internal/service/record.go @@ -10,11 +10,17 @@ import ( "github.com/titpetric/factory" "go.uber.org/zap" + "github.com/cortezaproject/corteza-server/compose/decoder" "github.com/cortezaproject/corteza-server/compose/internal/repository" "github.com/cortezaproject/corteza-server/compose/types" "github.com/cortezaproject/corteza-server/internal/auth" ) +const ( + IMPORT_ON_ERROR_SKIP = "SKIP" + IMPORT_ON_ERROR_FAIL = "FAIL" +) + type ( record struct { db *factory.DB @@ -57,6 +63,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 Create(record *types.Record) (*types.Record, error) Update(record *types.Record) (*types.Record, error) @@ -67,6 +74,32 @@ type ( Encoder interface { Record(*types.Record) error } + + Decoder interface { + Header() []string + 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"` + } + + RecordImportProgress struct { + StartedAt *time.Time `json:"startedAt"` + FinishedAt *time.Time `json:"finishedAt"` + Completed uint64 `json:"completed"` + Failed uint64 `json:"failed"` + FailReason error `json:"failReason"` + } ) func Record() RecordService { @@ -184,6 +217,49 @@ func (svc record) Find(filter types.RecordFilter) (set types.RecordSet, f types. return } +func (svc record) Import(ses *RecordImportSession, ssvc ImportSessionService) error { + if ses.Decoder == nil { + return nil + } + + if ses.Progress.StartedAt != nil { + return errors.New("Unable to start import: Import session already active") + } + + sa := time.Now() + ses.Progress.StartedAt = &sa + ssvc.SetRecordByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil) + + return svc.db.Transaction(func() (err error) { + err = ses.Decoder.Records(ses.Fields, func(mod *types.Record) error { + mod.NamespaceID = ses.NamespaceID + mod.ModuleID = ses.ModuleID + mod.OwnedBy = ses.UserID + + _, err := svc.Create(mod) + if err != nil { + ses.Progress.Failed++ + ses.Progress.FailReason = err + + if ses.OnError == IMPORT_ON_ERROR_FAIL { + fa := time.Now() + ses.Progress.FinishedAt = &fa + ssvc.SetRecordByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil) + return err + } + } else { + ses.Progress.Completed++ + } + return nil + }) + + fa := time.Now() + ses.Progress.FinishedAt = &fa + ssvc.SetRecordByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil) + return + }) +} + // Export returns all records // // @todo better value handling diff --git a/compose/internal/service/service.go b/compose/internal/service/service.go index 02ef1e382..50f3932bc 100644 --- a/compose/internal/service/service.go +++ b/compose/internal/service/service.go @@ -47,14 +47,14 @@ var ( // DefaultAutomationRunner runs automation scripts by listening to triggerManager and invoking Corredor service DefaultAutomationRunner automationRunner - DefaultNamespace NamespaceService - DefaultRecord RecordService - DefaultModule ModuleService - DefaultChart ChartService - DefaultPage PageService - - DefaultAttachment AttachmentService - DefaultNotification NotificationService + DefaultNamespace NamespaceService + DefaultImportSession ImportSessionService + DefaultRecord RecordService + DefaultModule ModuleService + DefaultChart ChartService + DefaultPage PageService + DefaultAttachment AttachmentService + DefaultNotification NotificationService DefaultSystemUser *systemUser ) @@ -130,6 +130,7 @@ func Init(ctx context.Context, log *zap.Logger, c Config) (err error) { ) } + DefaultImportSession = ImportSession() DefaultRecord = Record() DefaultPage = Page() DefaultChart = Chart() diff --git a/compose/rest/record.go b/compose/rest/record.go index df57fe924..0de13d17e 100644 --- a/compose/rest/record.go +++ b/compose/rest/record.go @@ -12,11 +12,13 @@ import ( "github.com/pkg/errors" + "github.com/cortezaproject/corteza-server/compose/decoder" "github.com/cortezaproject/corteza-server/compose/encoder" "github.com/cortezaproject/corteza-server/compose/internal/repository" "github.com/cortezaproject/corteza-server/compose/internal/service" "github.com/cortezaproject/corteza-server/compose/rest/request" "github.com/cortezaproject/corteza-server/compose/types" + "github.com/cortezaproject/corteza-server/pkg/mime" "github.com/cortezaproject/corteza-server/pkg/rh" ) @@ -36,10 +38,11 @@ type ( } Record struct { - record service.RecordService - module service.ModuleService - attachment service.AttachmentService - ac recordAccessController + importSession service.ImportSessionService + record service.RecordService + module service.ModuleService + attachment service.AttachmentService + ac recordAccessController } recordAccessController interface { @@ -50,10 +53,11 @@ type ( func (Record) New() *Record { return &Record{ - record: service.DefaultRecord, - module: service.DefaultModule, - attachment: service.DefaultAttachment, - ac: service.DefaultAccessControl, + importSession: service.DefaultImportSession, + record: service.DefaultRecord, + module: service.DefaultModule, + attachment: service.DefaultAttachment, + ac: service.DefaultAccessControl, } } @@ -167,6 +171,102 @@ func (ctrl *Record) Upload(ctx context.Context, r *request.RecordUpload) (interf return makeAttachmentPayload(ctx, a, err) } +func (ctrl *Record) ImportInit(ctx context.Context, r *request.RecordImportInit) (interface{}, error) { + var ( + err error + recordDecoder service.Decoder + ) + + // Access control. + if _, err = ctrl.module.With(ctx).FindByID(r.NamespaceID, r.ModuleID); err != nil { + return nil, err + } + + f, err := r.Upload.Open() + if err != nil { + return nil, err + } + 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" + } + } + + // determine decoder + switch strings.ToLower(ext) { + case "json", "jsonl", "ldjson", "ndjson": + recordDecoder = decoder.NewStructuredDecoder(json.NewDecoder(f)) + + case "csv": + recordDecoder = decoder.NewFlatReader(csv.NewReader(f)) + + default: + return nil, errors.New(fmt.Sprintf("unsupported format (\"%s\")", ext)) + + } + + header := recordDecoder.Header() + hh := make(map[string]string) + for _, h := range header { + hh[h] = "" + } + + return ctrl.importSession.SetRecordByID(ctx, 0, r.NamespaceID, r.ModuleID, hh, nil, recordDecoder) +} + +func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) (interface{}, error) { + var ( + err error + ) + + // Access control. + if _, err = ctrl.module.With(ctx).FindByID(r.NamespaceID, r.ModuleID); err != nil { + return nil, err + } + + // Check if session ok + ses, err := ctrl.importSession.FindRecordByID(ctx, r.SessionID) + if err != nil { + return nil, err + } + + if ses.Progress.StartedAt != nil { + return nil, errors.New("Unable to start import: Import session already active") + } + + ses.Fields = make(map[string]string) + err = json.Unmarshal(r.Fields, &ses.Fields) + if err != nil { + return nil, err + } + + ses.OnError = r.OnError + + // @todo routine + ctrl.record.With(ctx).Import(ses, ctrl.importSession) + + return ses, nil +} + +func (ctrl *Record) ImportProgress(ctx context.Context, r *request.RecordImportProgress) (interface{}, error) { + // Get session + ses, err := ctrl.importSession.FindRecordByID(ctx, r.SessionID) + if err != nil { + return nil, err + } + + return ses, nil +} + func (ctrl *Record) Export(ctx context.Context, r *request.RecordExport) (interface{}, error) { type ( // ad-hoc interface for our encoder diff --git a/pkg/mime/mime.go b/pkg/mime/mime.go new file mode 100644 index 000000000..063c5adb3 --- /dev/null +++ b/pkg/mime/mime.go @@ -0,0 +1,34 @@ +package mime + +import ( + "bufio" + "io" + + "github.com/gabriel-vasile/mimetype" +) + +func Type(file io.ReadSeeker) (mt string, ext string, err error) { + if _, err = file.Seek(0, 0); err != nil { + return + } + + // Make sure we rewind when we're done + defer file.Seek(0, 0) + return mimetype.DetectReader(file) +} + +func JsonL(file io.ReadSeeker) (bool, error) { + // ExtractMimetype fails to detect json if jsonl is used + // For now check if first rune is { + r := bufio.NewReader(file) + rn, _, err := r.ReadRune() + defer file.Seek(0, 0) + if err != nil { + return false, err + } + + if string(rn) == "{" { + return true, nil + } + return false, nil +}