DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Fix bulk ops logic & (record value) error handling

Browse Source
This commit is contained in:
Denis Arh
2020-05-29 15:22:58 +02:00
parent 51d8fbcbb4
commit 288e516986
7 changed files with 342 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
compose/rest/record.go
+3 -3
View File
@@ -174,7 +174,7 @@ func (ctrl *Record) Create(ctx context.Context, r *request.RecordCreate) (interf
return ctrl.handleValidationError(rve), nil
}
return ctrl.makePayloadBatch(ctx, m, err, rr...)
return ctrl.makeBulkPayload(ctx, m, err, rr...)
}
func (ctrl *Record) Update(ctx context.Context, r *request.RecordUpdate) (interface{}, error) {
@@ -223,7 +223,7 @@ func (ctrl *Record) Update(ctx context.Context, r *request.RecordUpdate) (interf
return ctrl.handleValidationError(rve), nil
}
return ctrl.makePayloadBatch(ctx, m, err, rr...)
return ctrl.makeBulkPayload(ctx, m, err, rr...)
}
func (ctrl *Record) Delete(ctx context.Context, r *request.RecordDelete) (interface{}, error) {
@@ -520,7 +520,7 @@ func (ctrl *Record) TriggerScriptOnList(ctx context.Context, r *request.RecordTr
return resputil.OK(), err
}
func (ctrl Record) makePayloadBatch(ctx context.Context, m *types.Module, err error, rr ...*types.Record) (*recordPayload, error) {
func (ctrl Record) makeBulkPayload(ctx context.Context, m *types.Module, err error, rr ...*types.Record) (*recordPayload, error) {
if err != nil || rr == nil {
return nil, err
}
compose/service/record.go
+87 -41
View File
@@ -2,6 +2,7 @@ package service
import (
"context"
"errors"
"fmt"
"regexp"
"strconv"
@@ -10,7 +11,6 @@ import (
"github.com/cortezaproject/corteza-server/compose/service/values"
"github.com/cortezaproject/corteza-server/pkg/actionlog"
"github.com/pkg/errors"
"github.com/titpetric/factory"
"github.com/cortezaproject/corteza-server/compose/decoder"
@@ -369,7 +369,7 @@ func (svc record) Import(ses *RecordImportSession, ssvc ImportSessionService) (e
err = svc.db.Transaction(func() (err error) {
if ses.Progress.StartedAt != nil {
return errors.New("Unable to start import: Import session already active")
return fmt.Errorf("Unable to start import: Import session already active")
}
sa := time.Now()
@@ -437,26 +437,47 @@ func (svc record) Export(filter types.RecordFilter, enc Encoder) (err error) {
}
// Bulk handles provided set of bulk record operations.
// It's able to create or update records in a single transaction.
//
// Feature used mainly by Record Lines, but should be used when ever we wish to perform
// multiple operations over records.
// It's able to create, update or delete records in a single transaction.
func (svc record) Bulk(oo ...*types.BulkRecordOperation) (rr types.RecordSet, err error) {
var (
ctr = map[string]uint{}
action = RecordActionCreate
aProp = &recordActionProps{}
)
return rr, svc.db.Transaction(func() (err error) {
err = svc.db.Transaction(func() error {
// pre-verify all
for _, p := range oo {
r := p.Record
switch p.Operation {
case types.OperationTypeCreate, types.OperationTypeUpdate, types.OperationTypeDelete:
// ok
default:
return RecordErrUnknownBulkOperation(&recordActionProps{bulkOperation: string(p.Operation)})
}
}
var (
// in case we get record value errors from create or update operations
// we ll merge the errors into one slice and return it all together
//
// this is done under assumption that potential before-record-update/create automation
// scripts are playing by the rules and do not do any changes before any potential
// record value errors are returned
//
// @todo all records/values could and should be pre-validated
// before we start storing any changes
rves = &types.RecordValueErrorSet{}
action func(props ...*recordActionProps) *recordAction
ctr = map[string]uint{}
r *types.Record
aProp = &recordActionProps{}
)
for _, p := range oo {
r = p.Record
res := fmt.Sprintf("compose:module:%d", r.ModuleID)
if _, ok := ctr[res]; !ok {
ctr[res] = 0
}
aProp.setChanged(r)
// Handle any pre processing, such as defining parent recordID.
if p.LinkBy != "" {
// As is, we can use the first record as the master record.
@@ -472,41 +493,62 @@ func (svc record) Bulk(oo ...*types.BulkRecordOperation) (rr types.RecordSet, er
case types.OperationTypeCreate:
action = RecordActionCreate
r, err = svc.create(r)
break
case types.OperationTypeUpdate:
action = RecordActionUpdate
r, err = svc.update(r)
break
case types.OperationTypeDelete:
action = RecordActionDelete
r, err = svc.delete(r.NamespaceID, r.ModuleID, r.ID)
break
default:
return errors.Errorf("unknown record bulk operation %s", p.Operation)
}
if err != nil {
if rve, ok := err.(*types.RecordValueErrorSet); ok {
if errors.Is(err, RecordErrValueInput()) {
wrappedRve := errors.Unwrap(err)
if rve, ok := wrappedRve.(*types.RecordValueErrorSet); ok {
// Attach additional meta to each value error for FE identification
for _, re := range rve.Set {
re.Meta["resource"] = res
re.Meta["item"] = ctr[res]
rves.Push(re)
}
return rve
// log record value error for this record
_ = svc.recordAction(svc.ctx, aProp, action, err)
// do not return errors just yet, values on other records from the payload (if any)
// might have errors too
continue
}
} else if err != nil {
return svc.recordAction(svc.ctx, aProp, action, err)
}
rr = append(rr, r)
ctr[res]++
}
if !rves.IsValid() {
// Any errors gathered?
return rves
}
return nil
})
if len(oo) == 1 {
// was not really a bulk operation and we already recorded the action
// inside transaction loop
return rr, err
} else {
// when doing bulk op (updating and/or creating more than one record at once),
// we already log action for each operation
//
// to log the fact that the bulk op was done, we do one additional recording
// without any props
return rr, svc.recordAction(svc.ctx, &recordActionProps{}, RecordActionBulk, err)
}
}
// Raw create function that is responsible for value validation, event dispatching
@@ -543,14 +585,14 @@ func (svc record) create(new *types.Record) (rec *types.Record, err error) {
if svc.optEmitEvents {
// Handle input payload
if rve = svc.procCreate(invokerID, m, new); !rve.IsValid() {
return nil, rve
return nil, RecordErrValueInput().Wrap(rve)
}
new.Values = svc.formatter.Run(m, new.Values)
if err = svc.eventbus.WaitFor(svc.ctx, event.RecordBeforeCreate(new, nil, m, ns, rve)); err != nil {
return
} else if !rve.IsValid() {
return nil, rve
return nil, RecordErrValueInput().Wrap(rve)
}
}
@@ -559,7 +601,7 @@ func (svc record) create(new *types.Record) (rec *types.Record, err error) {
// Handle payload from automation scripts
if rve = svc.procCreate(invokerID, m, new); !rve.IsValid() {
return nil, rve
return nil, RecordErrValueInput().Wrap(rve)
}
if new, err = svc.recordRepo.Create(new); err != nil {
@@ -633,7 +675,7 @@ func (svc record) update(upd *types.Record) (rec *types.Record, err error) {
if svc.optEmitEvents {
// Handle input payload
if rve = svc.procUpdate(invokerID, m, upd, old); !rve.IsValid() {
return nil, rve
return nil, RecordErrValueInput().Wrap(rve)
}
// Before we pass values to record-before-update handling events
@@ -655,13 +697,13 @@ func (svc record) update(upd *types.Record) (rec *types.Record, err error) {
if err = svc.eventbus.WaitFor(svc.ctx, event.RecordBeforeUpdate(upd, old, m, ns, rve)); err != nil {
return
} else if !rve.IsValid() {
return nil, rve
return nil, RecordErrValueInput().Wrap(rve)
}
}
// Handle payload from automation scripts
if rve = svc.procUpdate(invokerID, m, upd, old); !rve.IsValid() {
return nil, rve
return nil, RecordErrValueInput().Wrap(rve)
}
if upd, err = svc.recordRepo.Update(upd); err != nil {
@@ -963,7 +1005,7 @@ func (svc record) Organize(namespaceID, moduleID, recordID uint64, posField, pos
reorderingRecords = true
if !regexp.MustCompile(`^[0-9]+$`).MatchString(position) {
return errors.Errorf("expecting number for sorting position %q", posField)
return fmt.Errorf("expecting number for sorting position %q", posField)
}
// Check field existence and permissions
@@ -971,15 +1013,15 @@ func (svc record) Organize(namespaceID, moduleID, recordID uint64, posField, pos
sf := m.Fields.FindByName(posField)
if sf == nil {
return errors.Errorf("no such field %q", posField)
return fmt.Errorf("no such field %q", posField)
}
if !sf.IsNumeric() {
return errors.Errorf("can not reorder on non numeric field %q", posField)
return fmt.Errorf("can not reorder on non numeric field %q", posField)
}
if sf.Multi {
return errors.Errorf("can not reorder on multi-value field %q", posField)
return fmt.Errorf("can not reorder on multi-value field %q", posField)
}
if !svc.ac.CanUpdateRecordValue(svc.ctx, sf) {
@@ -999,11 +1041,11 @@ func (svc record) Organize(namespaceID, moduleID, recordID uint64, posField, pos
vf := m.Fields.FindByName(grpField)
if vf == nil {
return errors.Errorf("no such field %q", grpField)
return fmt.Errorf("no such field %q", grpField)
}
if vf.Multi {
return errors.Errorf("can not update multi-value field %q", posField)
return fmt.Errorf("can not update multi-value field %q", posField)
}
if !svc.ac.CanUpdateRecordValue(svc.ctx, vf) {
@@ -1186,7 +1228,7 @@ func (svc record) Iterator(f types.RecordFilter, fn eventbus.HandlerFn, action s
// Handle payload from automation scripts
if rve := svc.procCreate(invokerID, m, rec); !rve.IsValid() {
return rve
return RecordErrValueInput().Wrap(rve)
}
if cln, err = svc.recordRepo.Create(rec); err != nil {
@@ -1199,7 +1241,7 @@ func (svc record) Iterator(f types.RecordFilter, fn eventbus.HandlerFn, action s
// Handle input payload
if rve := svc.procUpdate(invokerID, m, rec, rec); !rve.IsValid() {
return rve
return RecordErrValueInput().Wrap(rve)
}
if rec, err = svc.recordRepo.Update(rec); err != nil {
@@ -1293,6 +1335,10 @@ func (svc record) setDefaultValues(m *types.Module, vv types.RecordValueSet) (ou
// Received values must fit the data model: on unknown fields
// or multi/single value mismatch we return an error
//
// Record value errors is intentionally NOT used here; if input fails here
// we can assume that form builder (or whatever it was that assembled the record values)
// was misconfigured and will most likely failed to properly parse the
// record value errors payload too
func (svc record) generalValueSetValidation(m *types.Module, vv types.RecordValueSet) (err error) {
var (
aProps = &recordActionProps{}
@@ -1325,7 +1371,7 @@ func (svc record) generalValueSetValidation(m *types.Module, vv types.RecordValu
// Make sure there are no multi values in a non-multi value fields
err = m.Fields.Walk(func(field *types.ModuleField) error {
if !field.Multi && len(vv.FilterByName(field.Name)) > 1 {
return RecordErrInvalidValueInput(aProps.setField(field.Name))
return RecordErrInvalidValueStructure(aProps.setField(field.Name))
}
return nil
compose/service/record_actions.gen.go
+131 -13
View File
@@ -16,13 +16,15 @@ import (
type (
recordActionProps struct {
record *types.Record
changed *types.Record
filter *types.RecordFilter
namespace *types.Namespace
module *types.Module
field string
value string
record *types.Record
changed *types.Record
filter *types.RecordFilter
namespace *types.Namespace
module *types.Module
bulkOperation string
field string
value string
valueErrors *types.RecordValueErrorSet
}
recordAction struct {
@@ -111,6 +113,17 @@ func (p *recordActionProps) setModule(module *types.Module) *recordActionProps {
return p
}
// setBulkOperation updates recordActionProps's bulkOperation
//
// Allows method chaining
//
// This function is auto-generated.
//
func (p *recordActionProps) setBulkOperation(bulkOperation string) *recordActionProps {
p.bulkOperation = bulkOperation
return p
}
// setField updates recordActionProps's field
//
// Allows method chaining
@@ -133,6 +146,17 @@ func (p *recordActionProps) setValue(value string) *recordActionProps {
return p
}
// setValueErrors updates recordActionProps's valueErrors
//
// Allows method chaining
//
// This function is auto-generated.
//
func (p *recordActionProps) setValueErrors(valueErrors *types.RecordValueErrorSet) *recordActionProps {
p.valueErrors = valueErrors
return p
}
// serialize converts recordActionProps to actionlog.Meta
//
// This function is auto-generated.
@@ -157,7 +181,6 @@ func (p recordActionProps) serialize() actionlog.Meta {
m["changed.moduleID"] = str(p.changed.ModuleID)
m["changed.namespaceID"] = str(p.changed.NamespaceID)
m["changed.ownedBy"] = str(p.changed.OwnedBy)
m["changed.values"] = str(p.changed.Values)
}
if p.filter != nil {
m["filter.query"] = str(p.filter.Query)
@@ -181,8 +204,12 @@ func (p recordActionProps) serialize() actionlog.Meta {
m["module.ID"] = str(p.module.ID)
m["module.namespaceID"] = str(p.module.NamespaceID)
}
m["bulkOperation"] = str(p.bulkOperation)
m["field"] = str(p.field)
m["value"] = str(p.value)
if p.valueErrors != nil {
m["valueErrors.set"] = str(p.valueErrors.Set)
}
return m
}
@@ -250,14 +277,12 @@ func (p recordActionProps) tr(in string, err error) string {
p.changed.ModuleID,
p.changed.NamespaceID,
p.changed.OwnedBy,
p.changed.Values,
),
)
pairs = append(pairs, "{changed.ID}", fns(p.changed.ID))
pairs = append(pairs, "{changed.moduleID}", fns(p.changed.ModuleID))
pairs = append(pairs, "{changed.namespaceID}", fns(p.changed.NamespaceID))
pairs = append(pairs, "{changed.ownedBy}", fns(p.changed.OwnedBy))
pairs = append(pairs, "{changed.values}", fns(p.changed.Values))
}
if p.filter != nil {
@@ -321,8 +346,21 @@ func (p recordActionProps) tr(in string, err error) string {
pairs = append(pairs, "{module.ID}", fns(p.module.ID))
pairs = append(pairs, "{module.namespaceID}", fns(p.module.NamespaceID))
}
pairs = append(pairs, "{bulkOperation}", fns(p.bulkOperation))
pairs = append(pairs, "{field}", fns(p.field))
pairs = append(pairs, "{value}", fns(p.value))
if p.valueErrors != nil {
// replacement for "{valueErrors}" (in order how fields are defined)
pairs = append(
pairs,
"{valueErrors}",
fns(
p.valueErrors.Set,
),
)
pairs = append(pairs, "{valueErrors.set}", fns(p.valueErrors.Set))
}
return strings.NewReplacer(pairs...).Replace(in)
}
@@ -501,6 +539,26 @@ func RecordActionReport(props ...*recordActionProps) *recordAction {
return a
}
// RecordActionBulk returns "compose:record.bulk" error
//
// This function is auto-generated.
//
func RecordActionBulk(props ...*recordActionProps) *recordAction {
a := &recordAction{
timestamp: time.Now(),
resource: "compose:record",
action: "bulk",
log: "bulk record operation",
severity: actionlog.Info,
}
if len(props) > 0 {
a.props = props[0]
}
return a
}
// RecordActionCreate returns "compose:record.create" error
//
// This function is auto-generated.
@@ -1265,16 +1323,16 @@ func RecordErrFieldNotFound(props ...*recordActionProps) *recordError {
}
// RecordErrInvalidValueInput returns "compose:record.invalidValueInput" audit event as actionlog.Error
// RecordErrInvalidValueStructure returns "compose:record.invalidValueStructure" audit event as actionlog.Error
//
//
// This function is auto-generated.
//
func RecordErrInvalidValueInput(props ...*recordActionProps) *recordError {
func RecordErrInvalidValueStructure(props ...*recordActionProps) *recordError {
var e = &recordError{
timestamp: time.Now(),
resource: "compose:record",
error: "invalidValueInput",
error: "invalidValueStructure",
action: "error",
message: "more than one value for a single-value field {field}",
log: "more than one value for a single-value field {field}",
@@ -1295,6 +1353,36 @@ func RecordErrInvalidValueInput(props ...*recordActionProps) *recordError {
}
// RecordErrUnknownBulkOperation returns "compose:record.unknownBulkOperation" audit event as actionlog.Error
//
//
// This function is auto-generated.
//
func RecordErrUnknownBulkOperation(props ...*recordActionProps) *recordError {
var e = &recordError{
timestamp: time.Now(),
resource: "compose:record",
error: "unknownBulkOperation",
action: "error",
message: "unknown bulk operation {bulkOperation}",
log: "unknown bulk operation {bulkOperation}",
severity: actionlog.Error,
props: func() *recordActionProps {
if len(props) > 0 {
return props[0]
}
return nil
}(),
}
if len(props) > 0 {
e.props = props[0]
}
return e
}
// RecordErrInvalidReferenceFormat returns "compose:record.invalidReferenceFormat" audit event as actionlog.Error
//
//
@@ -1325,6 +1413,36 @@ func RecordErrInvalidReferenceFormat(props ...*recordActionProps) *recordError {
}
// RecordErrValueInput returns "compose:record.valueInput" audit event as actionlog.Error
//
//
// This function is auto-generated.
//
func RecordErrValueInput(props ...*recordActionProps) *recordError {
var e = &recordError{
timestamp: time.Now(),
resource: "compose:record",
error: "valueInput",
action: "error",
message: "invalid record value input: {err}",
log: "invalid record value input: {err}",
severity: actionlog.Error,
props: func() *recordActionProps {
if len(props) > 0 {
return props[0]
}
return nil
}(),
}
if len(props) > 0 {
e.props = props[0]
}
return e
}
// *********************************************************************************************************************
// *********************************************************************************************************************
compose/service/record_actions.yaml
+15 -2
View File
@@ -18,7 +18,7 @@ props:
fields: [ ID, moduleID, namespaceID, ownedBy ]
- name: changed
type: "*types.Record"
fields: [ ID, moduleID, namespaceID, ownedBy, values ]
fields: [ ID, moduleID, namespaceID, ownedBy ]
- name: filter
type: "*types.RecordFilter"
fields: [ query, namespaceID, moduleID, deleted, sort, limit, offset, page, perPage ]
@@ -28,8 +28,12 @@ props:
- name: module
type: "*types.Module"
fields: [ name, handle, ID, namespaceID ]
- name: bulkOperation
- name: field
- name: value
- name: valueErrors
type: "*types.RecordValueErrorSet"
fields: [ set ]
actions:
- action: search
@@ -44,6 +48,9 @@ actions:
log: "report generated"
severity: info
- action: bulk
log: "bulk record operation"
- action: create
log: "created {record}"
@@ -145,8 +152,14 @@ errors:
- error: fieldNotFound
message: "no such field {field}"
- error: invalidValueInput
- error: invalidValueStructure
message: "more than one value for a single-value field {field}"
- error: unknownBulkOperation
message: "unknown bulk operation {bulkOperation}"
- error: invalidReferenceFormat
message: "invalid reference format"
- error: valueInput
message: "invalid record value input: {err}"
compose/types/validated.go
+3 -1
View File
@@ -1,5 +1,7 @@
package types
import "fmt"
type (
RecordValueError struct {
Kind string `json:"kind"`
@@ -21,5 +23,5 @@ func (v *RecordValueErrorSet) IsValid() bool {
}
func (v *RecordValueErrorSet) Error() string {
return "invalid values input"
return fmt.Sprintf("%d issue(s) found", len(v.Set))
}
tests/compose/record_test.go
+29
View File
@@ -172,6 +172,35 @@ func TestRecordCreate(t *testing.T) {
End()
}
func TestRecordCreateWithErrors(t *testing.T) {
h := newHelper(t)
fields := types.ModuleFieldSet{
&types.ModuleField{
Name: "name",
},
&types.ModuleField{
Name: "required",
Required: true,
},
}
module := h.repoMakeRecordModuleWithFields("record testing module", fields...)
h.allow(types.ModulePermissionResource.AppendWildcard(), "record.create")
h.apiInit().
Post(fmt.Sprintf("/namespace/%d/module/%d/record/", module.NamespaceID, module.ID)).
JSON(fmt.Sprintf(`{"values": [{"name": "name", "value": "val"}]}`)).
Expect(t).
Assert(helpers.AssertRecordValueError(
&types.RecordValueError{
Kind: "empty",
Message: "",
Meta: map[string]interface{}{"field": "required"},
},
)).
End()
}
func TestRecordUpdateForbidden(t *testing.T) {
h := newHelper(t)
tests/helpers/assert.go
+74 -6
View File
@@ -2,21 +2,36 @@ package helpers
import (
"encoding/json"
"fmt"
"net/http"
"github.com/davecgh/go-spew/spew"
"github.com/pkg/errors"
"github.com/cortezaproject/corteza-server/compose/types"
)
type (
assertFn func(*http.Response, *http.Request) error
StdErrorResponse struct{ Error struct{ Message string } }
StdErrorResponse struct {
Error struct {
Message string
}
}
RecordValueErrorSetResponse struct {
Error struct {
Message string
Details []types.RecordValueError
}
}
)
// decodes response body to given struct
func decodeBody(rsp *http.Response, s interface{}) error {
func DecodeBody(rsp *http.Response, s interface{}) error {
if err := json.NewDecoder(rsp.Body).Decode(&s); err != nil {
return errors.Wrap(err, "could not assert IsAuthorized")
return fmt.Errorf("could not decode body: %w", err)
}
return nil
@@ -47,15 +62,68 @@ func firstErr(ee ...interface{}) error {
// AssertNoErrors ensures there are no errors in the response
func AssertNoErrors(rsp *http.Response, _ *http.Request) (err error) {
tmp := StdErrorResponse{}
return firstErr(decodeBody(rsp, &tmp), tmp)
return firstErr(DecodeBody(rsp, &tmp), tmp)
}
// Asserts that all expected errors are returned
//
// Compares each error by Kind, Message and Meta.field
//
// Note: This assertion always expects errors!
func AssertRecordValueError(exp ...*types.RecordValueError) assertFn {
return func(rsp *http.Response, _ *http.Request) (err error) {
rcvd := RecordValueErrorSetResponse{}
if err = DecodeBody(rsp, &rcvd); err != nil {
return err
}
if len(rcvd.Error.Details) == 0 {
return fmt.Errorf("expecting value errors, none received")
}
expLoop:
for _, e := range exp {
for _, r := range rcvd.Error.Details {
if e.Kind != r.Kind {
continue
}
if e.Message != r.Message {
continue
}
if e.Meta["field"] != r.Meta["field"] {
continue
}
// found expected error
continue expLoop
}
// did not find expected error
return fmt.Errorf("did not find expected error %v", e)
}
return nil
}
}
// Dump can be put into Assert()
func Dump(rsp *http.Response, _ *http.Request) (err error) {
spew.Dump(rsp.Status)
spew.Dump(rsp.Header)
var payload interface{}
if err = DecodeBody(rsp, &payload); err != nil {
return err
}
spew.Dump(payload)
return nil
}
// AssertError ensures there are no errors in the response
func AssertError(expectedError string) assertFn {
return func(rsp *http.Response, _ *http.Request) (err error) {
tmp := StdErrorResponse{}
if err = decodeBody(rsp, &tmp); err != nil {
return errors.Errorf("Could not decode body: %v", err)
if err = DecodeBody(rsp, &tmp); err != nil {
return err
}
if tmp.Error.Message == "" {