Update pkg/dal to handle schema alterations
This commit is contained in:
Tomaž Jerman
parent
0695eb7118
commit
4224ba9828
193
server/pkg/dal/alteration.go
Normal file
193
server/pkg/dal/alteration.go
Normal file
@ -0,0 +1,193 @@
|
||||
package dal
|
||||
|
||||
import "encoding/json"
|
||||
|
||||
type (
|
||||
Alteration struct {
|
||||
ID uint64
|
||||
BatchID uint64
|
||||
DependsOn uint64
|
||||
Resource string
|
||||
ResourceType string
|
||||
ConnectionID uint64
|
||||
|
||||
AttributeAdd *AttributeAdd
|
||||
AttributeDelete *AttributeDelete
|
||||
AttributeReType *AttributeReType
|
||||
AttributeReEncode *AttributeReEncode
|
||||
ModelAdd *ModelAdd
|
||||
ModelDelete *ModelDelete
|
||||
}
|
||||
|
||||
AlterationSet []*Alteration
|
||||
|
||||
AttributeAdd struct {
|
||||
Attr *Attribute `json:"attr"`
|
||||
}
|
||||
|
||||
AttributeDelete struct {
|
||||
Attr *Attribute `json:"attr"`
|
||||
}
|
||||
|
||||
AttributeReType struct {
|
||||
Attr *Attribute `json:"attr"`
|
||||
To Type `json:"to"`
|
||||
}
|
||||
|
||||
AttributeReEncode struct {
|
||||
Attr *Attribute `json:"attr"`
|
||||
To *Attribute `json:"to"`
|
||||
}
|
||||
|
||||
ModelAdd struct {
|
||||
Model *Model `json:"model"`
|
||||
}
|
||||
|
||||
ModelDelete struct {
|
||||
Model *Model `json:"model"`
|
||||
}
|
||||
|
||||
// auxAttributeType is a helper struct used for marshaling/unmarshaling
|
||||
//
|
||||
// This is required since the Type inside the Attribute is an interface and we
|
||||
// need to help the encoding/json a bit.
|
||||
auxAttributeReType struct {
|
||||
Attr *Attribute
|
||||
To *auxAttributeType
|
||||
}
|
||||
)
|
||||
|
||||
// Merge merges the two alteration slices
|
||||
func (a AlterationSet) Merge(b AlterationSet) (c AlterationSet) {
|
||||
// @todo don't blindly append the two slices since there can be duplicates
|
||||
// or overlapping alterations which would cause needles processing
|
||||
//
|
||||
// A quick list of overlapping alterations:
|
||||
// * attribute A added and then renamed from A to A'
|
||||
// * attribute A renamed to A' and then renamed to A''
|
||||
// * attribute A deleted and then created
|
||||
//
|
||||
// For now we'll simply append them and worry about improvements on a later stage
|
||||
|
||||
return append(a, b...)
|
||||
}
|
||||
|
||||
func (a AttributeReType) MarshalJSON() ([]byte, error) {
|
||||
aux := auxAttributeReType{
|
||||
Attr: a.Attr,
|
||||
To: &auxAttributeType{},
|
||||
}
|
||||
|
||||
switch t := a.To.(type) {
|
||||
case *TypeID:
|
||||
aux.To.Type = "ID"
|
||||
aux.To.ID = t
|
||||
|
||||
case *TypeRef:
|
||||
aux.To.Type = "Ref"
|
||||
aux.To.Ref = t
|
||||
|
||||
case *TypeTimestamp:
|
||||
aux.To.Type = "Timestamp"
|
||||
aux.To.Timestamp = t
|
||||
|
||||
case *TypeTime:
|
||||
aux.To.Type = "Time"
|
||||
aux.To.Time = t
|
||||
|
||||
case *TypeDate:
|
||||
aux.To.Type = "Date"
|
||||
aux.To.Date = t
|
||||
|
||||
case *TypeNumber:
|
||||
aux.To.Type = "Number"
|
||||
aux.To.Number = t
|
||||
|
||||
case *TypeText:
|
||||
aux.To.Type = "Text"
|
||||
aux.To.Text = t
|
||||
|
||||
case *TypeBoolean:
|
||||
aux.To.Type = "Boolean"
|
||||
aux.To.Boolean = t
|
||||
|
||||
case *TypeEnum:
|
||||
aux.To.Type = "Enum"
|
||||
aux.To.Enum = t
|
||||
|
||||
case *TypeGeometry:
|
||||
aux.To.Type = "Geometry"
|
||||
aux.To.Geometry = t
|
||||
|
||||
case *TypeJSON:
|
||||
aux.To.Type = "JSON"
|
||||
aux.To.JSON = t
|
||||
|
||||
case *TypeBlob:
|
||||
aux.To.Type = "Blob"
|
||||
aux.To.Blob = t
|
||||
|
||||
case *TypeUUID:
|
||||
aux.To.Type = "UUID"
|
||||
aux.To.UUID = t
|
||||
}
|
||||
|
||||
return json.Marshal(aux)
|
||||
}
|
||||
|
||||
func (a *AttributeReType) UnmarshalJSON(data []byte) (err error) {
|
||||
aux := &auxAttributeReType{}
|
||||
err = json.Unmarshal(data, &aux)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if a == nil {
|
||||
*a = AttributeReType{}
|
||||
}
|
||||
|
||||
a.Attr = aux.Attr
|
||||
|
||||
switch aux.To.Type {
|
||||
case "ID":
|
||||
a.To = aux.To.ID
|
||||
|
||||
case "Ref":
|
||||
a.To = aux.To.Ref
|
||||
|
||||
case "Timestamp":
|
||||
a.To = aux.To.Timestamp
|
||||
|
||||
case "Time":
|
||||
a.To = aux.To.Time
|
||||
|
||||
case "Date":
|
||||
a.To = aux.To.Date
|
||||
|
||||
case "Number":
|
||||
a.To = aux.To.Number
|
||||
|
||||
case "Text":
|
||||
a.To = aux.To.Text
|
||||
|
||||
case "Boolean":
|
||||
a.To = aux.To.Boolean
|
||||
|
||||
case "Enum":
|
||||
a.To = aux.To.Enum
|
||||
|
||||
case "Geometry":
|
||||
a.To = aux.To.Geometry
|
||||
|
||||
case "JSON":
|
||||
a.To = aux.To.JSON
|
||||
|
||||
case "Blob":
|
||||
a.To = aux.To.Blob
|
||||
|
||||
case "UUID":
|
||||
a.To = aux.To.UUID
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
@ -52,11 +52,17 @@ type (
|
||||
}
|
||||
)
|
||||
|
||||
func (*CodecPlain) Type() AttributeCodecType { return "corteza::dal:attribute-codec:plain" }
|
||||
const (
|
||||
AttributeCodecPlain AttributeCodecType = "corteza::dal:attribute-codec:plain"
|
||||
AttributeCodecRecordValueSetJSON AttributeCodecType = "corteza::dal:attribute-codec:record-value-set-json"
|
||||
AttributeCodecAlias AttributeCodecType = "corteza::dal:attribute-codec:alias"
|
||||
)
|
||||
|
||||
func (*CodecPlain) Type() AttributeCodecType { return AttributeCodecPlain }
|
||||
func (*CodecRecordValueSetJSON) Type() AttributeCodecType {
|
||||
return "corteza::dal:attribute-codec:record-value-set-json"
|
||||
return AttributeCodecRecordValueSetJSON
|
||||
}
|
||||
func (*CodecAlias) Type() AttributeCodecType { return "corteza::dal:attribute-codec:alias" }
|
||||
func (*CodecAlias) Type() AttributeCodecType { return AttributeCodecAlias }
|
||||
|
||||
func (*CodecPlain) SingleValueOnly() bool { return true }
|
||||
func (*CodecRecordValueSetJSON) SingleValueOnly() bool { return false }
|
||||
|
||||
@ -28,6 +28,13 @@ const (
|
||||
|
||||
// Diff calculates the diff between models a and b where a is used as base
|
||||
func (a *Model) Diff(b *Model) (out ModelDiffSet) {
|
||||
if a == nil {
|
||||
a = &Model{}
|
||||
}
|
||||
if b == nil {
|
||||
b = &Model{}
|
||||
}
|
||||
|
||||
bIndex := make(map[string]struct {
|
||||
found bool
|
||||
attr *Attribute
|
||||
@ -124,3 +131,69 @@ func (a *Model) Diff(b *Model) (out ModelDiffSet) {
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (dd ModelDiffSet) Alterations() (out []*Alteration) {
|
||||
add := func(a *Alteration) {
|
||||
out = append(out, a)
|
||||
}
|
||||
|
||||
for _, d := range dd {
|
||||
switch d.Type {
|
||||
case AttributeMissing:
|
||||
if d.Inserted == nil {
|
||||
// @todo if this was the last attribute we can consider dropping this column
|
||||
if d.Original.Store.Type() == AttributeCodecRecordValueSetJSON {
|
||||
break
|
||||
}
|
||||
|
||||
add(&Alteration{
|
||||
AttributeDelete: &AttributeDelete{
|
||||
Attr: d.Original,
|
||||
},
|
||||
})
|
||||
} else {
|
||||
if d.Inserted.Store.Type() == AttributeCodecRecordValueSetJSON {
|
||||
add(&Alteration{
|
||||
AttributeAdd: &AttributeAdd{
|
||||
Attr: &Attribute{
|
||||
Ident: d.Inserted.StoreIdent(),
|
||||
Type: &TypeJSON{Nullable: false},
|
||||
Store: &CodecPlain{},
|
||||
},
|
||||
},
|
||||
})
|
||||
} else {
|
||||
add(&Alteration{
|
||||
AttributeAdd: &AttributeAdd{
|
||||
Attr: d.Inserted,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
case AttributeTypeMissmatch:
|
||||
// @todo we might have to do some validation earlier on
|
||||
if d.Original.Store.Type() == AttributeCodecRecordValueSetJSON {
|
||||
break
|
||||
}
|
||||
|
||||
add(&Alteration{
|
||||
AttributeReType: &AttributeReType{
|
||||
Attr: d.Original,
|
||||
To: d.Inserted.Type,
|
||||
},
|
||||
})
|
||||
|
||||
case AttributeCodecMismatch:
|
||||
add(&Alteration{
|
||||
AttributeReEncode: &AttributeReEncode{
|
||||
Attr: d.Original,
|
||||
To: d.Inserted,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@ -7,7 +7,6 @@ import (
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
"github.com/cortezaproject/corteza/server/pkg/expr"
|
||||
"github.com/cortezaproject/corteza/server/pkg/filter"
|
||||
"github.com/cortezaproject/corteza/server/pkg/ql"
|
||||
"go.uber.org/zap"
|
||||
@ -80,12 +79,12 @@ type (
|
||||
// Only metadata (such as idents) are affected; attributes can not be changed here
|
||||
UpdateModel(ctx context.Context, old *Model, new *Model) error
|
||||
|
||||
// UpdateModelAttribute requests for the model attribute change
|
||||
//
|
||||
// Specific operations require data transformations (type change).
|
||||
// Some basic ops. should be implemented on DB driver level, but greater controll can be
|
||||
// achieved via the trans functions.
|
||||
UpdateModelAttribute(ctx context.Context, sch *Model, diff *ModelDiff, hasRecords bool, trans ...TransformationFunction) error
|
||||
// AssertSchemaAlterations returns a new set of Alterations based on what the underlying
|
||||
// schema already provides -- it discards alterations for column additions that already exist, etc.
|
||||
AssertSchemaAlterations(ctx context.Context, sch *Model, aa ...*Alteration) ([]*Alteration, error)
|
||||
|
||||
// ApplyAlteration applies the given alterations to the underlying schema
|
||||
ApplyAlteration(ctx context.Context, sch *Model, aa ...*Alteration) []error
|
||||
}
|
||||
|
||||
ConnectionCloser interface {
|
||||
@ -94,8 +93,6 @@ type (
|
||||
Close(ctx context.Context) error
|
||||
}
|
||||
|
||||
TransformationFunction func(*Model, Attribute, expr.TypedValue) (expr.TypedValue, bool, error)
|
||||
|
||||
// Store provides an interface which CRS uses to interact with the underlying database
|
||||
|
||||
ValueGetter interface {
|
||||
|
||||
@ -2,6 +2,7 @@ package dal
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/cortezaproject/corteza/server/pkg/errors"
|
||||
)
|
||||
|
||||
@ -67,6 +68,9 @@ func errModelCreateGreaterAttributeSensitivityLevel(connectionID, modelID, attrS
|
||||
func errModelCreateConnectionModelUnsupported(connectionID, modelID uint64) error {
|
||||
return fmt.Errorf("cannot create model %d on connection %d: model already exists for connection but is not compatible with provided definition", modelID, connectionID)
|
||||
}
|
||||
func errModelCreateInvalidIdent(connectionID, modelID uint64, ident string) error {
|
||||
return fmt.Errorf("cannot create model %d on connection %d: malformed model ident %s", modelID, connectionID, ident)
|
||||
}
|
||||
|
||||
// - update
|
||||
func errModelUpdateProblematicConnection(connectionID, modelID uint64) error {
|
||||
@ -94,6 +98,11 @@ func errModelUpdateGreaterSensitivityLevel(connectionID, modelID, modelSensitivi
|
||||
return fmt.Errorf("cannot update model %d on connection %d: sensitivity level %d exceeds connection supported sensitivity level %d", modelID, connectionID, modelSensitivityLevelID, connSensitivityLevelID)
|
||||
}
|
||||
|
||||
// - alterations
|
||||
func errModelRequiresAlteration(connectionID, modelID, batchID uint64) error {
|
||||
return fmt.Errorf("model %d on connection %d requires schema alterations: alteration batchID %d", modelID, connectionID, batchID)
|
||||
}
|
||||
|
||||
// Attribute errors
|
||||
// - Update
|
||||
func errAttributeUpdateProblematicConnection(connectionID, modelID uint64) error {
|
||||
|
||||
@ -3,11 +3,14 @@ package dal
|
||||
import "go.uber.org/zap"
|
||||
|
||||
type (
|
||||
issue struct {
|
||||
kind issueKind
|
||||
err error
|
||||
Issue struct {
|
||||
err error
|
||||
Issue string `json:"issue,omitempty"`
|
||||
|
||||
Kind issueKind `json:"kind,omitempty"`
|
||||
Meta map[string]any `json:"meta,omitempty"`
|
||||
}
|
||||
issueSet []issue
|
||||
issueSet []Issue
|
||||
|
||||
issueHelper struct {
|
||||
// these two will be used to help clear out unneeded errors
|
||||
@ -34,27 +37,12 @@ func newIssueHelper() *issueHelper {
|
||||
}
|
||||
}
|
||||
|
||||
func makeIssue(kind issueKind, err error) issue {
|
||||
return issue{
|
||||
kind: kind,
|
||||
err: err,
|
||||
}
|
||||
func (svc *service) SearchConnectionIssues(connectionID uint64) (out []Issue) {
|
||||
return svc.connectionIssues[connectionID]
|
||||
}
|
||||
|
||||
func (svc *service) SearchConnectionIssues(connectionID uint64) (out []error) {
|
||||
for _, issue := range svc.connectionIssues[connectionID] {
|
||||
out = append(out, issue.err)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (svc *service) SearchModelIssues(resourceID uint64) (out []error) {
|
||||
for _, issue := range svc.modelIssues[resourceID] {
|
||||
out = append(out, issue.err)
|
||||
}
|
||||
|
||||
return
|
||||
func (svc *service) SearchModelIssues(resourceID uint64) (out []Issue) {
|
||||
return svc.modelIssues[resourceID]
|
||||
}
|
||||
|
||||
func (svc *service) hasConnectionIssues(connectionID uint64) bool {
|
||||
@ -95,12 +83,24 @@ func (rd *issueHelper) addModel(modelID uint64) *issueHelper {
|
||||
return rd
|
||||
}
|
||||
|
||||
func (rd *issueHelper) addConnectionIssue(connectionID uint64, err error) {
|
||||
rd.connectionIssues[connectionID] = append(rd.connectionIssues[connectionID], makeIssue(connectionIssue, err))
|
||||
func (rd *issueHelper) addConnectionIssue(connectionID uint64, i Issue) {
|
||||
i.Kind = connectionIssue
|
||||
i.Issue = i.err.Error()
|
||||
rd.connectionIssues[connectionID] = append(rd.connectionIssues[connectionID], i)
|
||||
}
|
||||
|
||||
func (rd *issueHelper) addModelIssue(resourceID uint64, err error) {
|
||||
rd.modelIssues[resourceID] = append(rd.modelIssues[resourceID], makeIssue(modelIssue, err))
|
||||
func (rd *issueHelper) addModelIssue(resourceID uint64, i Issue) {
|
||||
i.Kind = modelIssue
|
||||
i.Issue = i.err.Error()
|
||||
rd.modelIssues[resourceID] = append(rd.modelIssues[resourceID], i)
|
||||
}
|
||||
|
||||
func (rd *issueHelper) hasConnectionIssues() bool {
|
||||
return len(rd.connectionIssues) > 0
|
||||
}
|
||||
|
||||
func (rd *issueHelper) hasModelIssues() bool {
|
||||
return len(rd.modelIssues) > 0
|
||||
}
|
||||
|
||||
func (a *issueHelper) mergeWith(b *issueHelper) {
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
@ -79,6 +80,56 @@ type (
|
||||
Type Type
|
||||
}
|
||||
|
||||
// auxAttribute is a helper struct used for marshaling/unmarshaling
|
||||
//
|
||||
// This is required since some fields are interfaces
|
||||
auxAttribute struct {
|
||||
Ident string `json:"ident"`
|
||||
Label string `json:"label"`
|
||||
SensitivityLevelID uint64 `json:"sensitivityLevelID"`
|
||||
MultiValue bool `json:"multiValue"`
|
||||
PrimaryKey bool `json:"primaryKey"`
|
||||
SoftDeleteFlag bool `json:"softDeleteFlag"`
|
||||
System bool `json:"system"`
|
||||
Sortable bool `json:"sortable"`
|
||||
Filterable bool `json:"filterable"`
|
||||
|
||||
Store *auxAttributeStore `json:"store"`
|
||||
Type *auxAttributeType `json:"type"`
|
||||
}
|
||||
|
||||
// auxAttributeStore is a helper struct used for marshaling/unmarshaling
|
||||
//
|
||||
// This is required since some fields are interfaces
|
||||
auxAttributeStore struct {
|
||||
Type string `json:"type"`
|
||||
|
||||
Plain *CodecPlain `json:"plain,omitempty"`
|
||||
RecordValueSetJSON *CodecRecordValueSetJSON `json:"recordValueSetJSON,omitempty"`
|
||||
Alias *CodecAlias `json:"alias,omitempty"`
|
||||
}
|
||||
|
||||
// auxAttributeType is a helper struct used for marshaling/unmarshaling
|
||||
//
|
||||
// This is required since some fields are interfaces
|
||||
auxAttributeType struct {
|
||||
Type string `json:"type"`
|
||||
|
||||
ID *TypeID `json:"id,omitempty"`
|
||||
Ref *TypeRef `json:"ref,omitempty"`
|
||||
Timestamp *TypeTimestamp `json:"timestamp,omitempty"`
|
||||
Time *TypeTime `json:"time,omitempty"`
|
||||
Date *TypeDate `json:"date,omitempty"`
|
||||
Number *TypeNumber `json:"number,omitempty"`
|
||||
Text *TypeText `json:"text,omitempty"`
|
||||
Boolean *TypeBoolean `json:"boolean,omitempty"`
|
||||
Enum *TypeEnum `json:"enum,omitempty"`
|
||||
Geometry *TypeGeometry `json:"geometry,omitempty"`
|
||||
JSON *TypeJSON `json:"jSON,omitempty"`
|
||||
Blob *TypeBlob `json:"blob,omitempty"`
|
||||
UUID *TypeUUID `json:"uuid,omitempty"`
|
||||
}
|
||||
|
||||
AttributeSet []*Attribute
|
||||
|
||||
Index struct {
|
||||
@ -295,3 +346,175 @@ func (m Model) Validate() error {
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *Attribute) MarshalJSON() ([]byte, error) {
|
||||
aux := &auxAttribute{
|
||||
Ident: a.Ident,
|
||||
Label: a.Label,
|
||||
SensitivityLevelID: a.SensitivityLevelID,
|
||||
MultiValue: a.MultiValue,
|
||||
PrimaryKey: a.PrimaryKey,
|
||||
SoftDeleteFlag: a.SoftDeleteFlag,
|
||||
System: a.System,
|
||||
Sortable: a.Sortable,
|
||||
Filterable: a.Filterable,
|
||||
|
||||
Store: &auxAttributeStore{},
|
||||
Type: &auxAttributeType{},
|
||||
}
|
||||
|
||||
switch s := a.Store.(type) {
|
||||
case *CodecPlain:
|
||||
aux.Store.Type = "plain"
|
||||
aux.Store.Plain = s
|
||||
|
||||
case *CodecRecordValueSetJSON:
|
||||
aux.Store.Type = "recordValueSetJSON"
|
||||
aux.Store.RecordValueSetJSON = s
|
||||
|
||||
case *CodecAlias:
|
||||
aux.Store.Type = "alias"
|
||||
aux.Store.Alias = s
|
||||
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown store codec type: %T", s)
|
||||
}
|
||||
|
||||
switch t := a.Type.(type) {
|
||||
case *TypeID:
|
||||
aux.Type.Type = "ID"
|
||||
aux.Type.ID = t
|
||||
|
||||
case *TypeRef:
|
||||
aux.Type.Type = "Ref"
|
||||
aux.Type.Ref = t
|
||||
|
||||
case *TypeTimestamp:
|
||||
aux.Type.Type = "Timestamp"
|
||||
aux.Type.Timestamp = t
|
||||
|
||||
case *TypeTime:
|
||||
aux.Type.Type = "Time"
|
||||
aux.Type.Time = t
|
||||
|
||||
case *TypeDate:
|
||||
aux.Type.Type = "Date"
|
||||
aux.Type.Date = t
|
||||
|
||||
case *TypeNumber:
|
||||
aux.Type.Type = "Number"
|
||||
aux.Type.Number = t
|
||||
|
||||
case *TypeText:
|
||||
aux.Type.Type = "Text"
|
||||
aux.Type.Text = t
|
||||
|
||||
case *TypeBoolean:
|
||||
aux.Type.Type = "Boolean"
|
||||
aux.Type.Boolean = t
|
||||
|
||||
case *TypeEnum:
|
||||
aux.Type.Type = "Enum"
|
||||
aux.Type.Enum = t
|
||||
|
||||
case *TypeGeometry:
|
||||
aux.Type.Type = "Geometry"
|
||||
aux.Type.Geometry = t
|
||||
|
||||
case *TypeJSON:
|
||||
aux.Type.Type = "JSON"
|
||||
aux.Type.JSON = t
|
||||
|
||||
case *TypeBlob:
|
||||
aux.Type.Type = "Blob"
|
||||
aux.Type.Blob = t
|
||||
|
||||
case *TypeUUID:
|
||||
aux.Type.Type = "UUID"
|
||||
aux.Type.UUID = t
|
||||
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown attribute type type: %T", t)
|
||||
}
|
||||
|
||||
return json.Marshal(aux)
|
||||
}
|
||||
|
||||
func (a *Attribute) UnmarshalJSON(data []byte) (err error) {
|
||||
aux := &auxAttribute{}
|
||||
err = json.Unmarshal(data, &aux)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if a == nil {
|
||||
*a = Attribute{}
|
||||
}
|
||||
|
||||
a.Ident = aux.Ident
|
||||
a.Label = aux.Label
|
||||
a.SensitivityLevelID = aux.SensitivityLevelID
|
||||
a.MultiValue = aux.MultiValue
|
||||
a.PrimaryKey = aux.PrimaryKey
|
||||
a.SoftDeleteFlag = aux.SoftDeleteFlag
|
||||
a.System = aux.System
|
||||
a.Sortable = aux.Sortable
|
||||
a.Filterable = aux.Filterable
|
||||
|
||||
switch aux.Store.Type {
|
||||
case "plain":
|
||||
a.Store = aux.Store.Plain
|
||||
|
||||
case "recordValueSetJSON":
|
||||
a.Store = aux.Store.RecordValueSetJSON
|
||||
|
||||
case "alias":
|
||||
a.Store = aux.Store.Alias
|
||||
}
|
||||
|
||||
switch aux.Type.Type {
|
||||
case "ID":
|
||||
a.Type = aux.Type.ID
|
||||
|
||||
case "Ref":
|
||||
a.Type = aux.Type.Ref
|
||||
|
||||
case "Timestamp":
|
||||
a.Type = aux.Type.Timestamp
|
||||
|
||||
case "Time":
|
||||
a.Type = aux.Type.Time
|
||||
|
||||
case "Date":
|
||||
a.Type = aux.Type.Date
|
||||
|
||||
case "Number":
|
||||
a.Type = aux.Type.Number
|
||||
|
||||
case "Text":
|
||||
a.Type = aux.Type.Text
|
||||
|
||||
case "Boolean":
|
||||
a.Type = aux.Type.Boolean
|
||||
|
||||
case "Enum":
|
||||
a.Type = aux.Type.Enum
|
||||
|
||||
case "Geometry":
|
||||
a.Type = aux.Type.Geometry
|
||||
|
||||
case "JSON":
|
||||
a.Type = aux.Type.JSON
|
||||
|
||||
case "Blob":
|
||||
a.Type = aux.Type.Blob
|
||||
|
||||
case "UUID":
|
||||
a.Type = aux.Type.UUID
|
||||
|
||||
default:
|
||||
return fmt.Errorf("unknown attribute type type: %s", aux.Type.Type)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@ -3,10 +3,10 @@ package dal
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
|
||||
"github.com/cortezaproject/corteza/server/pkg/filter"
|
||||
"github.com/cortezaproject/corteza/server/pkg/id"
|
||||
"github.com/cortezaproject/corteza/server/pkg/logger"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
@ -29,6 +29,13 @@ type (
|
||||
|
||||
connectionIssues dalIssueIndex
|
||||
modelIssues dalIssueIndex
|
||||
|
||||
Alterations alterations
|
||||
}
|
||||
|
||||
alterations interface {
|
||||
ModelAlterations(context.Context, *Model) (out []*Alteration, err error)
|
||||
SetAlterations(ctx context.Context, m *Model, stale []*Alteration, set ...*Alteration) (err error)
|
||||
}
|
||||
|
||||
FullService interface {
|
||||
@ -45,7 +52,6 @@ type (
|
||||
SearchModels(ctx context.Context) (out ModelSet, err error)
|
||||
ReplaceModel(ctx context.Context, model *Model) (err error)
|
||||
RemoveModel(ctx context.Context, connectionID, ID uint64) (err error)
|
||||
ReplaceModelAttribute(ctx context.Context, model *Model, dif *ModelDiff, hasRecords bool, trans ...TransformationFunction) (err error)
|
||||
FindModelByResourceID(connectionID uint64, resourceID uint64) *Model
|
||||
FindModelByResourceIdent(connectionID uint64, resourceType, resourceIdent string) *Model
|
||||
FindModelByIdent(connectionID uint64, ident string) *Model
|
||||
@ -60,8 +66,8 @@ type (
|
||||
Run(ctx context.Context, pp Pipeline) (iter Iterator, err error)
|
||||
Dryrun(ctx context.Context, pp Pipeline) (err error)
|
||||
|
||||
SearchConnectionIssues(connectionID uint64) (out []error)
|
||||
SearchModelIssues(resourceID uint64) (out []error)
|
||||
SearchConnectionIssues(connectionID uint64) (out []Issue)
|
||||
SearchModelIssues(resourceID uint64) (out []Issue)
|
||||
}
|
||||
)
|
||||
|
||||
@ -319,7 +325,9 @@ func (svc *service) ReplaceConnection(ctx context.Context, conn *ConnectionWrap,
|
||||
|
||||
// Sensitivity level validations
|
||||
if !svc.sensitivityLevels.includes(conn.Config.SensitivityLevelID) {
|
||||
issues.addConnectionIssue(ID, errConnectionCreateMissingSensitivityLevel(ID, conn.Config.SensitivityLevelID))
|
||||
issues.addConnectionIssue(ID, Issue{
|
||||
err: errConnectionCreateMissingSensitivityLevel(ID, conn.Config.SensitivityLevelID),
|
||||
})
|
||||
}
|
||||
|
||||
if oldConn = svc.GetConnectionByID(ID); oldConn != nil {
|
||||
@ -335,7 +343,9 @@ func (svc *service) ReplaceConnection(ctx context.Context, conn *ConnectionWrap,
|
||||
|
||||
// - sensitivity levels
|
||||
if !svc.sensitivityLevels.isSubset(model.SensitivityLevelID, conn.Config.SensitivityLevelID) {
|
||||
issues.addConnectionIssue(ID, fmt.Errorf("cannot update connection %d: new connection sensitivity level does not support model %d", ID, model.ResourceID))
|
||||
issues.addConnectionIssue(ID, Issue{
|
||||
err: fmt.Errorf("cannot update connection %d: new connection sensitivity level does not support model %d", ID, model.ResourceID),
|
||||
})
|
||||
errored = true
|
||||
}
|
||||
}
|
||||
@ -349,7 +359,9 @@ func (svc *service) ReplaceConnection(ctx context.Context, conn *ConnectionWrap,
|
||||
// close old connection
|
||||
if cc, ok := oldConn.connection.(ConnectionCloser); ok {
|
||||
if err = cc.Close(ctx); err != nil {
|
||||
issues.addConnectionIssue(ID, err)
|
||||
issues.addConnectionIssue(ID, Issue{
|
||||
err: err,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -363,7 +375,9 @@ func (svc *service) ReplaceConnection(ctx context.Context, conn *ConnectionWrap,
|
||||
conn.connection, err = connect(ctx, svc.logger, svc.inDev, conn.params)
|
||||
if err != nil {
|
||||
log.Warn("could not connect", zap.Error(err))
|
||||
issues.addConnectionIssue(ID, err)
|
||||
issues.addConnectionIssue(ID, Issue{
|
||||
err: err,
|
||||
})
|
||||
} else {
|
||||
log.Debug("connected")
|
||||
}
|
||||
@ -667,18 +681,17 @@ func (svc *service) SearchModels(ctx context.Context) (out ModelSet, err error)
|
||||
|
||||
// ReplaceModel adds new or updates an existing model
|
||||
//
|
||||
// ReplaceModel only affects the metadata (the connection, identifier, ...)
|
||||
// and leaves attributes untouched.
|
||||
// Use the ReplaceModelAttribute to upsert attributes.
|
||||
//
|
||||
// We rely on the user to provide stable and valid model definitions.
|
||||
func (svc *service) ReplaceModel(ctx context.Context, model *Model) (err error) {
|
||||
var (
|
||||
ID = model.ResourceID
|
||||
oldModel *Model
|
||||
issues = newIssueHelper().addModel(ID)
|
||||
auxIssues = newIssueHelper()
|
||||
upd bool
|
||||
ID = model.ResourceID
|
||||
connection = svc.GetConnectionByID(model.ConnectionID)
|
||||
oldModel *Model
|
||||
issues = newIssueHelper().addModel(ID)
|
||||
upd bool
|
||||
|
||||
modelIssues bool
|
||||
connectionIssues bool
|
||||
|
||||
log = svc.logger.Named("models").With(
|
||||
logger.Uint64("ID", ID),
|
||||
@ -693,8 +706,8 @@ func (svc *service) ReplaceModel(ctx context.Context, model *Model) (err error)
|
||||
model.ConnectionID = svc.defConnID
|
||||
}
|
||||
|
||||
// Check if update
|
||||
if oldModel = svc.FindModelByResourceID(model.ConnectionID, model.ResourceID); oldModel != nil {
|
||||
// Check if we're creating or updating the model
|
||||
if oldModel = svc.FindModel(model.ToFilter()); oldModel != nil {
|
||||
log.Debug("found existing")
|
||||
|
||||
if oldModel.ConnectionID != model.ConnectionID {
|
||||
@ -704,54 +717,134 @@ func (svc *service) ReplaceModel(ctx context.Context, model *Model) (err error)
|
||||
upd = true
|
||||
}
|
||||
|
||||
// Validation
|
||||
svc.validateModel(issues, model, oldModel)
|
||||
for _, attr := range model.Attributes {
|
||||
svc.validateAttribute(issues, model, attr)
|
||||
}
|
||||
// @todo consider adding some logging to validators
|
||||
svc.validateModel(issues, connection, model, oldModel)
|
||||
svc.validateAttributes(issues, model, model.Attributes...)
|
||||
|
||||
// Add to connection
|
||||
connectionIssues := svc.hasConnectionIssues(model.ConnectionID)
|
||||
connectionIssues = issues.hasConnectionIssues()
|
||||
if connectionIssues {
|
||||
log.Warn(
|
||||
"not adding to connection due to connection issues",
|
||||
"not adding to store due to connection issues",
|
||||
zap.Any("issues", svc.SearchConnectionIssues(model.ConnectionID)),
|
||||
)
|
||||
}
|
||||
|
||||
modelIssues := svc.hasModelIssues(model.ResourceID)
|
||||
modelIssues = issues.hasModelIssues()
|
||||
if modelIssues {
|
||||
log.Warn(
|
||||
"not adding to connection due to model issues",
|
||||
"not adding to store due to model issues",
|
||||
zap.Any("issues", svc.SearchModelIssues(model.ResourceID)),
|
||||
)
|
||||
}
|
||||
|
||||
connection := svc.GetConnectionByID(model.ConnectionID)
|
||||
if !modelIssues && !connectionIssues {
|
||||
if !checkIdent(model.Ident, connection.Config.ModelIdentCheck...) {
|
||||
log.Warn("can not add model to connection, invalid ident")
|
||||
return nil
|
||||
}
|
||||
|
||||
auxIssues, err = svc.registerModelToConnection(ctx, connection, model)
|
||||
issues.mergeWith(auxIssues)
|
||||
if err != nil {
|
||||
log.Error("failed with errors", zap.Error(err))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Add to registry
|
||||
// @note models should be added to the registry regardless of issues
|
||||
svc.addModelToRegistry(model, upd)
|
||||
log.Debug(
|
||||
"added",
|
||||
"added to registry",
|
||||
logger.Uint64("connectionID", model.ConnectionID),
|
||||
)
|
||||
|
||||
// Can't continue with connection alterations if there are issues already
|
||||
if connectionIssues || modelIssues {
|
||||
return
|
||||
}
|
||||
|
||||
// Get alterations
|
||||
// - base from the model diff
|
||||
df := oldModel.Diff(model)
|
||||
batchID := id.Next()
|
||||
aa := df.Alterations()
|
||||
for _, a := range aa {
|
||||
a.BatchID = batchID
|
||||
a.Resource = model.Resource
|
||||
a.ResourceType = model.ResourceType
|
||||
a.ConnectionID = model.ConnectionID
|
||||
}
|
||||
// - merge with existing
|
||||
baseAa, err := svc.Alterations.ModelAlterations(ctx, model)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
// - cleanup to remove duplicates, squash overlapping changes, ...
|
||||
aa = svc.mergeAlterations(baseAa, aa)
|
||||
if len(aa) > 0 {
|
||||
batchID = aa[0].BatchID
|
||||
}
|
||||
// - updated with an assertion over the connection
|
||||
aa, err = connection.connection.AssertSchemaAlterations(ctx, model, aa...)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, a := range aa {
|
||||
a.BatchID = batchID
|
||||
}
|
||||
|
||||
err = svc.Alterations.SetAlterations(ctx, model, baseAa, aa...)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if len(aa) > 0 {
|
||||
issues.addModelIssue(model.ResourceID, Issue{
|
||||
err: errModelRequiresAlteration(connection.ID, model.ResourceID, batchID),
|
||||
Meta: map[string]any{
|
||||
"batchID": strconv.FormatUint(batchID, 10),
|
||||
},
|
||||
})
|
||||
log.Info("not adding to store: alterations required", zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
if upd {
|
||||
err = connection.connection.UpdateModel(ctx, oldModel, model)
|
||||
} else {
|
||||
err = connection.connection.CreateModel(ctx, model)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
log.Error("failed with errors", zap.Error(err))
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// ApplyAlteration updates the underlying schema with the requested changes
|
||||
func (svc *service) ApplyAlteration(ctx context.Context, alts ...*Alteration) (errs []error, err error) {
|
||||
if len(alts) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
connectionID := alts[0].ConnectionID
|
||||
resource := alts[0].Resource
|
||||
resourceType := alts[0].ResourceType
|
||||
for _, alt := range alts {
|
||||
if alt.ConnectionID != connectionID {
|
||||
return nil, fmt.Errorf("alterations must be for the same connection")
|
||||
}
|
||||
|
||||
if alt.ResourceType != resourceType {
|
||||
return nil, fmt.Errorf("alterations must be for the same resource type")
|
||||
}
|
||||
|
||||
if alt.Resource != resource {
|
||||
return nil, fmt.Errorf("alterations must be for the same resource")
|
||||
}
|
||||
}
|
||||
|
||||
connection := svc.GetConnectionByID(connectionID)
|
||||
if connection == nil {
|
||||
return nil, fmt.Errorf("connection not found")
|
||||
}
|
||||
|
||||
model := svc.getModelByRef(ModelRef{Resource: resource, ResourceType: resourceType})
|
||||
if model == nil {
|
||||
return nil, fmt.Errorf("model not found xd")
|
||||
}
|
||||
|
||||
return connection.connection.ApplyAlteration(ctx, model, alts...), nil
|
||||
}
|
||||
|
||||
// RemoveModel removes the given model from DAL
|
||||
//
|
||||
// @todo potentially add more interaction with the connection as in letting it know a model was removed.
|
||||
@ -798,44 +891,64 @@ func (svc *service) RemoveModel(ctx context.Context, connectionID, ID uint64) (e
|
||||
return nil
|
||||
}
|
||||
|
||||
func (svc *service) validateModel(issues *issueHelper, model, oldModel *Model) {
|
||||
func (svc *service) validateModel(issues *issueHelper, c *ConnectionWrap, model, oldModel *Model) {
|
||||
// Connection ok?
|
||||
if svc.hasConnectionIssues(model.ConnectionID) {
|
||||
issues.addModelIssue(model.ResourceID, errModelCreateProblematicConnection(model.ConnectionID, model.ResourceID))
|
||||
issues.addModelIssue(model.ResourceID, Issue{
|
||||
err: errModelCreateProblematicConnection(model.ConnectionID, model.ResourceID),
|
||||
})
|
||||
}
|
||||
// Connection exists?
|
||||
conn := svc.GetConnectionByID(model.ConnectionID)
|
||||
if conn == nil {
|
||||
issues.addModelIssue(model.ResourceID, errModelCreateMissingConnection(model.ConnectionID, model.ResourceID))
|
||||
if c == nil {
|
||||
issues.addModelIssue(model.ResourceID, Issue{
|
||||
err: errModelCreateMissingConnection(model.ConnectionID, model.ResourceID),
|
||||
})
|
||||
}
|
||||
|
||||
// If ident changed, check for duplicate
|
||||
if oldModel != nil && oldModel.Ident != model.Ident {
|
||||
if tmp := svc.FindModelByIdent(model.ConnectionID, model.Ident); tmp == nil {
|
||||
issues.addModelIssue(oldModel.ResourceID, errModelUpdateDuplicate(model.ConnectionID, model.ResourceID))
|
||||
issues.addModelIssue(oldModel.ResourceID, Issue{
|
||||
err: errModelUpdateDuplicate(model.ConnectionID, model.ResourceID),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Sensitivity level ok and valid?
|
||||
if !svc.sensitivityLevels.includes(model.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, errModelCreateMissingSensitivityLevel(model.ConnectionID, model.ResourceID, model.SensitivityLevelID))
|
||||
issues.addModelIssue(model.ResourceID, Issue{
|
||||
err: errModelCreateMissingSensitivityLevel(model.ConnectionID, model.ResourceID, model.SensitivityLevelID),
|
||||
})
|
||||
} else {
|
||||
// Only check if it is present
|
||||
if !svc.sensitivityLevels.isSubset(model.SensitivityLevelID, conn.Config.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, errModelCreateGreaterSensitivityLevel(model.ConnectionID, model.ResourceID, model.SensitivityLevelID, conn.Config.SensitivityLevelID))
|
||||
if !svc.sensitivityLevels.isSubset(model.SensitivityLevelID, c.Config.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, Issue{
|
||||
err: errModelCreateGreaterSensitivityLevel(model.ConnectionID, model.ResourceID, model.SensitivityLevelID, c.Config.SensitivityLevelID),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
if c != nil && !checkIdent(model.Ident, c.Config.ModelIdentCheck...) {
|
||||
issues.addModelIssue(model.ResourceID, Issue{
|
||||
err: errModelCreateInvalidIdent(model.ConnectionID, model.ResourceID, model.Ident),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (svc *service) validateAttribute(issues *issueHelper, model *Model, attr *Attribute) {
|
||||
if !svc.sensitivityLevels.includes(attr.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, errModelCreateMissingAttributeSensitivityLevel(model.ConnectionID, model.ResourceID, attr.SensitivityLevelID))
|
||||
} else {
|
||||
if !svc.sensitivityLevels.isSubset(attr.SensitivityLevelID, model.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, errModelCreateGreaterAttributeSensitivityLevel(model.ConnectionID, model.ResourceID, attr.SensitivityLevelID, model.SensitivityLevelID))
|
||||
func (svc *service) validateAttributes(issues *issueHelper, model *Model, attr ...*Attribute) {
|
||||
for _, a := range attr {
|
||||
if !svc.sensitivityLevels.includes(a.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, Issue{
|
||||
err: errModelCreateMissingAttributeSensitivityLevel(model.ConnectionID, model.ResourceID, a.SensitivityLevelID),
|
||||
})
|
||||
} else {
|
||||
if !svc.sensitivityLevels.isSubset(a.SensitivityLevelID, model.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, Issue{
|
||||
err: errModelCreateGreaterAttributeSensitivityLevel(model.ConnectionID, model.ResourceID, a.SensitivityLevelID, model.SensitivityLevelID),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (svc *service) addModelToRegistry(model *Model, upd bool) {
|
||||
@ -869,100 +982,6 @@ func (svc *service) removeModelFromRegistry(model *Model) {
|
||||
}
|
||||
}
|
||||
|
||||
// ReplaceModelAttribute adds new or updates an existing attribute for the given model
|
||||
//
|
||||
// We rely on the user to provide stable and valid attribute definitions.
|
||||
func (svc *service) ReplaceModelAttribute(ctx context.Context, model *Model, diff *ModelDiff, hasRecords bool, trans ...TransformationFunction) (err error) {
|
||||
svc.logger.Debug("updating model attribute", logger.Uint64("model", model.ResourceID))
|
||||
|
||||
var (
|
||||
conn *ConnectionWrap
|
||||
issues = newIssueHelper().addModel(model.ResourceID)
|
||||
)
|
||||
defer svc.updateIssues(issues)
|
||||
|
||||
if model.ConnectionID == 0 {
|
||||
model.ConnectionID = svc.defConnID
|
||||
}
|
||||
|
||||
// Validation
|
||||
{
|
||||
// Connection issues
|
||||
if svc.hasConnectionIssues(model.ConnectionID) {
|
||||
issues.addModelIssue(model.ResourceID, errAttributeUpdateProblematicConnection(model.ConnectionID, model.ResourceID))
|
||||
}
|
||||
|
||||
// Check if it exists
|
||||
auxModel := svc.FindModelByResourceID(model.ConnectionID, model.ResourceID)
|
||||
if auxModel == nil {
|
||||
issues.addModelIssue(model.ResourceID, errAttributeUpdateMissingModel(model.ConnectionID, model.ResourceID))
|
||||
}
|
||||
|
||||
// In case we're deleting it we can ignore this check
|
||||
if diff.Inserted != nil {
|
||||
if !svc.sensitivityLevels.includes(diff.Inserted.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, errAttributeUpdateMissingSensitivityLevel(model.ConnectionID, model.ResourceID, diff.Inserted.SensitivityLevelID))
|
||||
} else {
|
||||
if !svc.sensitivityLevels.isSubset(diff.Inserted.SensitivityLevelID, model.SensitivityLevelID) {
|
||||
issues.addModelIssue(model.ResourceID, errAttributeUpdateGreaterSensitivityLevel(model.ConnectionID, model.ResourceID, diff.Inserted.SensitivityLevelID, model.SensitivityLevelID))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
conn = svc.GetConnectionByID(model.ConnectionID)
|
||||
}
|
||||
|
||||
// Update attribute
|
||||
// Update connection
|
||||
connectionIssues := svc.hasConnectionIssues(model.ConnectionID)
|
||||
modelIssues := svc.hasModelIssues(model.ResourceID)
|
||||
|
||||
if !modelIssues && !connectionIssues {
|
||||
svc.logger.Debug("updating model attribute", logger.Uint64("connection", model.ConnectionID), logger.Uint64("model", model.ResourceID))
|
||||
|
||||
err = conn.connection.UpdateModelAttribute(ctx, model, diff, hasRecords, trans...)
|
||||
if err != nil {
|
||||
issues.addModelIssue(model.ResourceID, err)
|
||||
}
|
||||
} else {
|
||||
if connectionIssues {
|
||||
svc.logger.Warn("not updating model attribute due to connection issues", logger.Uint64("connection", model.ConnectionID))
|
||||
}
|
||||
if modelIssues {
|
||||
svc.logger.Warn("not updating model attribute due to model issues", logger.Uint64("model", model.ResourceID))
|
||||
}
|
||||
}
|
||||
|
||||
// Update registry
|
||||
if diff.Original == nil {
|
||||
// adding
|
||||
model.Attributes = append(model.Attributes, diff.Original)
|
||||
} else if diff.Original == nil {
|
||||
// removing
|
||||
model = svc.FindModelByResourceID(model.ConnectionID, model.ResourceID)
|
||||
nSet := make(AttributeSet, 0, len(model.Attributes))
|
||||
for _, attribute := range model.Attributes {
|
||||
if attribute.Ident != diff.Original.Ident {
|
||||
nSet = append(nSet, attribute)
|
||||
}
|
||||
}
|
||||
model.Attributes = nSet
|
||||
} else {
|
||||
// updating
|
||||
model = svc.FindModelByResourceID(model.ConnectionID, model.ResourceID)
|
||||
for i, attribute := range model.Attributes {
|
||||
if attribute.Ident == diff.Original.Ident {
|
||||
model.Attributes[i] = diff.Inserted
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
svc.logger.Debug("updated model attribute")
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// FindModelByRefs returns the model with all of the given refs matching
|
||||
//
|
||||
// @note refs are primarily used for DAL pipelines where steps can reference models
|
||||
@ -1059,48 +1078,6 @@ func (svc *service) getConnection(connectionID uint64, oo ...Operation) (cw *Con
|
||||
return
|
||||
}
|
||||
|
||||
func (svc *service) registerModelToConnection(ctx context.Context, cw *ConnectionWrap, model *Model) (issues *issueHelper, err error) {
|
||||
issues = newIssueHelper()
|
||||
|
||||
available, err := cw.connection.Models(ctx)
|
||||
if err != nil {
|
||||
issues.addConnectionIssue(model.ConnectionID, err)
|
||||
return issues, nil
|
||||
}
|
||||
|
||||
// Check if already in there
|
||||
if existing := available.FindByResourceIdent(model.ResourceType, model.Resource); existing != nil {
|
||||
// Assert validity
|
||||
diff := existing.Diff(model)
|
||||
if len(diff) > 0 {
|
||||
issues.addModelIssue(model.ResourceID, errModelCreateConnectionModelUnsupported(model.ConnectionID, model.ResourceID))
|
||||
return issues, nil
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// make sure connection supports model's ident
|
||||
var (
|
||||
rre []*regexp.Regexp
|
||||
)
|
||||
|
||||
for _, re := range rre {
|
||||
if re.MatchString(model.Ident) {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Try to add to store
|
||||
err = cw.connection.CreateModel(ctx, model)
|
||||
if err != nil {
|
||||
issues.addModelIssue(model.ResourceID, err)
|
||||
return issues, nil
|
||||
}
|
||||
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (svc *service) getModelByRef(mr ModelRef) *Model {
|
||||
if mr.ConnectionID == 0 {
|
||||
mr.ConnectionID = svc.defConnID
|
||||
@ -1234,3 +1211,19 @@ func (svc *service) analyzePipeline(ctx context.Context, pp Pipeline) (err error
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (svc *service) mergeAlterations(base, added AlterationSet) (out AlterationSet) {
|
||||
var (
|
||||
batchID uint64
|
||||
)
|
||||
|
||||
if len(base) > 0 {
|
||||
batchID = base[0].BatchID
|
||||
}
|
||||
|
||||
for _, a := range base {
|
||||
a.BatchID = batchID
|
||||
}
|
||||
|
||||
return base.Merge(added)
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user