DeforaNetworks/corteza
3
0
Fork 0
Code Activity

DAL service refactor & RDBMS interface tweak

Browse Source 
* Remove Compose service related bits out of the DAL service
* Minor DB connectivity tweaks and include capabilities
* Add values column to the compose_record table
* Compose types tweaks
This commit is contained in:
Tomaž Jerman
2022-05-20 15:37:33 +02:00
parent cb37b7beb7
commit 81dda274eb
14 changed files with 386 additions and 731 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/dal/attribute_codec.go
+3 -3
View File
@@ -51,8 +51,8 @@ type (
}
)
func (CodecPlain) Type() AttributeCodecType { return "corteza::dal:attribute-codec:plain" }
func (CodecRecordValueSetJSON) Type() AttributeCodecType {
func (*CodecPlain) Type() AttributeCodecType { return "corteza::dal:attribute-codec:plain" }
func (*CodecRecordValueSetJSON) Type() AttributeCodecType {
return "corteza::dal:attribute-codec:record-value-set-json"
}
func (CodecAlias) Type() AttributeCodecType { return "corteza::dal:attribute-codec:alias" }
func (*CodecAlias) Type() AttributeCodecType { return "corteza::dal:attribute-codec:alias" }
pkg/dal/capabilities/capabilities.go
+20 -8
View File
@@ -6,14 +6,15 @@ type (
)
const (
Create Capability = "create"
Update Capability = "update"
Delete Capability = "delete"
Search Capability = "search"
Paging Capability = "paging"
Stats Capability = "stats"
Sorting Capability = "sorting"
RBAC Capability = "RBAC"
Create Capability = "corteza::dal:capability:create"
Update Capability = "corteza::dal:capability:update"
Delete Capability = "corteza::dal:capability:delete"
Search Capability = "corteza::dal:capability:search"
Lookup Capability = "corteza::dal:capability:lookup"
Paging Capability = "corteza::dal:capability:paging"
Stats Capability = "corteza::dal:capability:stats"
Sorting Capability = "corteza::dal:capability:sorting"
RBAC Capability = "corteza::dal:capability:RBAC"
)
var (
@@ -21,6 +22,7 @@ var (
Create,
Update,
Search,
Lookup,
Paging,
Stats,
Sorting,
@@ -53,6 +55,11 @@ var (
Stats,
RBAC,
}
lookupCapabilities = Set{
Lookup,
RBAC,
}
)
// FullCapabilities returns all base system defined capabilities
@@ -86,6 +93,11 @@ func SearchCapabilities(requested ...Capability) (required Set) {
return common(searchCapabilities, requested)
}
// LookupCapabilities returns only requested capabilities used for Search operations
func LookupCapabilities(requested ...Capability) (required Set) {
return common(lookupCapabilities, requested)
}
func common(aa, bb Set) Set {
return aa.Intersect(bb)
}
pkg/dal/driver.go
+35 -42
View File
@@ -16,72 +16,66 @@ import (
type (
PKValues map[string]any
StoreConnection interface {
// ---
Connection interface {
// Meta
// Models returns all the models the underlying connection already supports
//
// This is useful when adding support for new models since we can find out what
// can work out of the box.
Models(context.Context) (ModelSet, error)
// Capabilities returns all of the capabilities the given store supports
Capabilities() capabilities.Set
// Can returns true if this store can handle the given capabilities
Can(capabilities ...capabilities.Capability) bool
// ---
// DML stuff
// Connection stuff
// Create stores the given data into the underlying database
Create(ctx context.Context, m *Model, rr ...ValueGetter) error
// Close closes the store connection allowing the driver to perform potential
// cleanup operations
Close(ctx context.Context) error
// Update(ctx context.Context, m *data.Model, rr ...ValueGetter) error
// Delete(ctx context.Context, m *data.Model, rr ...ValueGetter) error
// Truncate(ctx context.Context, m *data.Model) error
// ---
// Lookup returns one bit of data
Lookup(context.Context, *Model, ValueGetter, ValueSetter) error
// DML
// Search returns an iterator which can be used to access all if the bits
Search(context.Context, *Model, filter.Filter) (Iterator, error)
// CreateRecords stores the given records into the underlying database
CreateRecords(ctx context.Context, m *Model, rr ...ValueGetter) error
//UpdateRecords(ctx context.Context, m *data.Model, rr ...ValueGetter) error
//DeleteRecordsByPK(ctx context.Context, m *data.Model, rr ...ValueGetter) error
//TruncateRecords(ctx context.Context, m *data.Model) error
LookupRecord(context.Context, *Model, ValueGetter, ValueSetter) error
SearchRecords(context.Context, *Model, filter.Filter) (Iterator, error)
// ---
// DDL
// Models returns all the models the underlying database already supports
//
// This is useful when adding support for new modules since we can find out what
// can work out of the box.
Models(context.Context) (ModelSet, error)
// DDL stuff
// // returns all attribute types that driver supports
// AttributeTypes() []data.AttributeType
// AddModel adds support for the given models to the underlying database
// CreateModel adds support for the given models to the underlying database
//
// The operation returns an error if any of the models already exists.
AddModel(context.Context, *Model, ...*Model) error
CreateModel(context.Context, *Model, ...*Model) error
// RemoveModel removes support for the given model from the underlying database
RemoveModel(context.Context, *Model, ...*Model) error
// DeleteModel removes support for the given model from the underlying database
DeleteModel(context.Context, *Model, ...*Model) error
// AlterModel requests for metadata changes to the existing model
// UpdateModel requests for metadata changes to the existing model
//
// Only metadata (such as idents) are affected; attributes can not be changed here
AlterModel(ctx context.Context, old *Model, new *Model) error
UpdateModel(ctx context.Context, old *Model, new *Model) error
// AlterModelAttribute requests for the model attribute change
// 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.
AlterModelAttribute(ctx context.Context, sch *Model, old Attribute, new Attribute, trans ...TransformationFunction) error
UpdateModelAttribute(ctx context.Context, sch *Model, old Attribute, new Attribute, trans ...TransformationFunction) error
}
ConnectionCloser interface {
// Close closes the store connection allowing the driver to perform potential
// cleanup operations
Close(ctx context.Context) error
}
TransformationFunction func(*Model, Attribute, expr.TypedValue) (expr.TypedValue, bool, error)
@@ -113,7 +107,7 @@ type (
SetValue(string, uint, any) error
}
ConnectorFn func(ctx context.Context, dsn string, cc ...capabilities.Capability) (StoreConnection, error)
ConnectorFn func(ctx context.Context, dsn string, cc ...capabilities.Capability) (Connection, error)
)
var (
@@ -147,8 +141,7 @@ func Register(fn ConnectorFn, tt ...string) {
}
// connect opens a new StoreConnection for the given CRS
func connect(ctx context.Context, log *zap.Logger, def crsDefiner, isDevelopment bool) (StoreConnection, error) {
dsn := def.StoreDSN()
func connect(ctx context.Context, log *zap.Logger, isDevelopment bool, dsn string, capabilities ...capabilities.Capability) (Connection, error) {
if isDevelopment {
if strings.Contains(dsn, "{version}") {
@@ -168,7 +161,7 @@ func connect(ctx context.Context, log *zap.Logger, def crsDefiner, isDevelopment
}
if conn, ok := registered[storeType]; ok {
return conn(ctx, dsn, def.Capabilities()...)
return conn(ctx, dsn, capabilities...)
} else {
return nil, fmt.Errorf("unknown store type used: %q (check your storage configuration)", storeType)
}
pkg/dal/model.go
+59 -11
View File
@@ -9,11 +9,22 @@ import (
)
type (
// ModelFilter is used to retrieve a model from the DAL based on given params
ModelFilter struct {
ConnectionID uint64
ResourceID uint64
ResourceType string
Resource string
}
// Model describes the underlying data and its shape
Model struct {
StoreID uint64
Ident string
ConnectionID uint64
Ident string
Resource string
ResourceID uint64
ResourceType string
@@ -52,10 +63,47 @@ type (
AttributeSet []*Attribute
)
// FindByIdent returns the model that matches the ident
func (mm ModelSet) FindByIdent(ident string) *Model {
func PrimaryAttribute(ident string, codec Codec) *Attribute {
out := FullAttribute(ident, TypeID{}, codec)
out.Type = &TypeID{}
out.PrimaryKey = true
return out
}
func FullAttribute(ident string, at Type, codec Codec) *Attribute {
return &Attribute{
Ident: ident,
Sortable: true,
Filterable: true,
Store: codec,
Type: at,
}
}
func (a *Attribute) WithSoftDelete() *Attribute {
a.SoftDeleteFlag = true
return a
}
func (a *Attribute) WithMultiValue() *Attribute {
a.MultiValue = true
return a
}
// FindByResource returns the model that matches the resource
func (mm ModelSet) FindByResource(resType string, resource string) *Model {
for _, m := range mm {
if m.Ident == ident {
if m.ResourceType == resType && m.Resource == resource {
return m
}
}
return nil
}
func (mm ModelSet) FindByID(id uint64) *Model {
for _, m := range mm {
if m.ResourceID == id {
return m
}
}
@@ -66,14 +114,14 @@ func (mm ModelSet) FindByIdent(ident string) *Model {
// FilterByReferenced returns all of the models that reference b
func (aa ModelSet) FilterByReferenced(b *Model) (out ModelSet) {
for _, aModel := range aa {
if aModel.Ident == b.Ident {
if aModel.Resource == b.Resource {
continue
}
for _, aAttribute := range aModel.Attributes {
switch casted := aAttribute.Type.(type) {
case *TypeRef:
if casted.RefModel.Ident == b.Ident {
if casted.RefModel.Resource == b.Resource {
out = append(out, aModel)
}
}
@@ -83,14 +131,14 @@ func (aa ModelSet) FilterByReferenced(b *Model) (out ModelSet) {
return
}
// HasAttribute returns true when the model includes the specified ident
// HasAttribute returns true when the model includes the specified attribute
func (m Model) HasAttribute(ident string) bool {
return m.Attributes.FindByIdent(ident) != nil
}
func (aa AttributeSet) FindByIdent(ident string) *Attribute {
for _, a := range aa {
if strings.ToLower(a.Ident) == strings.ToLower(ident) {
if strings.EqualFold(a.Ident, ident) {
return a
}
}
@@ -100,8 +148,8 @@ func (aa AttributeSet) FindByIdent(ident string) *Attribute {
// Validate performs a base model validation before it is passed down
func (m Model) Validate() error {
if m.Ident == "" {
return fmt.Errorf("ident not defined")
if m.Resource == "" {
return fmt.Errorf("resource not defined")
}
seen := make(map[string]bool)
pkg/dal/service.go
+207 -592
View File
@@ -3,323 +3,201 @@ package dal
import (
"context"
"fmt"
"strings"
"github.com/cortezaproject/corteza-server/compose/types"
"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
"github.com/cortezaproject/corteza-server/pkg/filter"
"go.uber.org/zap"
)
type (
// the core struct that outlines the DAL service facility
connectionWrap struct {
connection Connection
Defaults ConnectionDefaults
}
ConnectionDefaults struct {
ModelIdent string
AttributeIdent string
PartitionFormat string
}
service struct {
stores map[uint64]StoreConnection
connections map[uint64]*connectionWrap
primary *connectionWrap
// Indexed by corresponding storeID
models map[uint64]ModelSet
primary StoreConnection
logger *zap.Logger
inDev bool
}
crsDefiner interface {
ComposeRecordStoreID() uint64
StoreDSN() string
Capabilities() capabilities.Set
}
// cStore is a simplified interface so we can use the store.Storer to assert a valid schema
cStore interface {
SearchComposeModules(ctx context.Context, f types.ModuleFilter) (types.ModuleSet, types.ModuleFilter, error)
SearchComposeModuleFields(ctx context.Context, f types.ModuleFieldFilter) (types.ModuleFieldSet, types.ModuleFieldFilter, error)
SearchComposeNamespaces(ctx context.Context, f types.NamespaceFilter) (types.NamespaceSet, types.NamespaceFilter, error)
}
)
const (
// https://www.rfc-editor.org/errata/eid1690
emailLength = 254
// Generally the upper most limit
urlLength = 2048
defaultStoreID uint64 = 0
sysID = "ID"
sysNamespaceID = "namespaceID"
sysModuleID = "moduleID"
sysCreatedAt = "createdAt"
sysCreatedBy = "createdBy"
sysUpdatedAt = "updatedAt"
sysUpdatedBy = "updatedBy"
sysDeletedAt = "deletedAt"
sysDeletedBy = "deletedBy"
sysOwnedBy = "ownedBy"
DefaultConnectionID uint64 = 0
)
// Service initializes a fresh record store where the given store serves as the default
func Service(ctx context.Context, log *zap.Logger, inDev bool, primary crsDefiner, stores ...crsDefiner) (*service, error) {
crs := &service{
stores: make(map[uint64]StoreConnection),
models: make(map[uint64]ModelSet),
primary: nil,
var (
gSvc *service
)
logger: log,
inDev: inDev,
}
// InitGlobalService initializes a fresh DAL where the given primary connection
func InitGlobalService(ctx context.Context, log *zap.Logger, inDev bool, dsn string, dft ConnectionDefaults, capabilities ...capabilities.Capability) (*service, error) {
if gSvc == nil {
gSvc = &service{
connections: make(map[uint64]*connectionWrap),
models: make(map[uint64]ModelSet),
primary: nil,
var err error
logger: log,
inDev: inDev,
}
crs.primary, err = connect(ctx, log, primary, inDev)
if err != nil {
return nil, err
}
return crs, crs.AddStore(ctx, stores...)
}
// ComposeRecordCreate creates the given records for the given module
func (svc *service) ComposeRecordCreate(ctx context.Context, module *types.Module, records ...ValueGetter) (err error) {
if !module.Store.Partitioned {
return fmt.Errorf("only partitioned modules work right now")
}
// Determine required capabilities
requiredCap := capabilities.CreateCapabilities(module.Store.Capabilities...)
// Determine store
var s StoreConnection
if s, _, err = svc.getStore(ctx, module.Store.ComposeRecordStoreID, requiredCap...); err != nil {
return err
}
// Get model
model := svc.lookupModel(module)
if model == nil {
return svc.modelNotFoundErr(module)
}
return s.CreateRecords(ctx, model, records...)
}
// @todo...
func (svc *service) ComposeRecordSearch(ctx context.Context, module *types.Module, filter *types.RecordFilter) (records types.RecordSet, outFilter *types.RecordFilter, err error) {
return
// // Determine requiredCap we'll need
// requiredCap := capabilities.SearchCapabilities(module.Store.Capabilities...).Union(svc.recFilterCapabilities(filter))
// // Connect to datasource
// var s Store
// var cc capabilities.Set
// _ = cc
// s, cc, err = svc.getStore(ctx, module.Store.ComposeRecordStoreID, requiredCap...)
// if err != nil {
// return
// }
// // Prepare data
// model := svc.lookupModel(module)
// if model == nil {
// return nil, nil, svc.modelNotFoundErr(module)
// }
// loader, err := s.SearchRecords(ctx, model, nil)
// if err != nil {
// return
// }
// limit := int(filter.Limit)
// if limit == 0 {
// limit = 10
// }
// auxCC := make([]Setter, limit)
// for i := range auxCC {
// auxCC[i] = &types.Record{}
// }
// var ok bool
// _ = ok
// for loader.More() && len(records) < int(limit) {
// _, err = loader.Load(model, auxCC)
// if err != nil {
// return
// }
// auxRecords, err := svc.extractRecords(model, auxCC...)
// if err != nil {
// return nil, nil, err
// }
// if !capabilities.AccessControlCapabilities().IsSubset(cc...) && filter.Check != nil {
// for _, r := range auxRecords {
// if r == nil {
// continue
// }
// if ok, err = filter.Check(r); err != nil {
// return nil, nil, err
// } else if !ok {
// continue
// }
// records = append(records, r)
// }
// } else {
// for _, r := range auxRecords {
// if r == nil {
// break
// }
// records = append(records, r)
// }
// }
// }
// return
}
// ---
// recFilterCapabilities utility helps construct required filter capabilities based on the provided record filter
func (svc *service) recFilterCapabilities(f *types.RecordFilter) (out capabilities.Set) {
if f == nil {
return
}
if f.PageCursor != nil {
out = append(out, capabilities.Paging)
}
if f.IncPageNavigation {
out = append(out, capabilities.Paging)
}
if f.IncTotal {
out = append(out, capabilities.Stats)
}
if f.Sort != nil {
out = append(out, capabilities.Sorting)
}
return
}
func (svc service) modelNotFoundErr(module *types.Module) error {
return fmt.Errorf("cannot create records for module %d: module not registered to svc", module.ID)
}
// AddStore registers the given store definitions as compose record stores
func (svc *service) AddStore(ctx context.Context, definers ...crsDefiner) (err error) {
for _, definer := range definers {
svc.stores[definer.ComposeRecordStoreID()], err = connect(ctx, svc.logger, definer, svc.inDev)
var err error
cw := &connectionWrap{
Defaults: dft,
}
cw.connection, err = connect(ctx, log, inDev, dsn, capabilities...)
if err != nil {
return
return nil, err
}
gSvc.primary = cw
}
return nil
return gSvc, nil
}
// RemoveStore removes the given store definition as a compose record store
func (svc *service) RemoveStore(ctx context.Context, storeID uint64, storeIDs ...uint64) (err error) {
for _, storeID := range append(storeIDs, storeID) {
s := svc.stores[storeID]
if s == nil {
return fmt.Errorf("can not remove compose record store %d: store does not exist", storeID)
}
// Potential cleanups
if err = s.Close(ctx); err != nil {
return
}
// Remove from registry
delete(svc.stores, storeID)
// Service returns the global initialized DAL service
//
// If InitGlobalService has not yet been called the function will panic
func Service() *service {
if gSvc == nil {
panic("DAL global service not initialized: call dal.InitGlobalService() first")
}
return nil
return gSvc
}
// ---
// Utilities
// // // // // // // // // // // // // // // // // // // // // // // // //
// Connection management
func (svc *service) getModel(store uint64, ident string) *Model {
for _, model := range svc.models[store] {
if model.Ident == ident {
return model
}
// AddConnection adds a new connection to the DAL
func (svc *service) AddConnection(ctx context.Context, connectionID uint64, dsn string, dft ConnectionDefaults, capabilities ...capabilities.Capability) (err error) {
cw := &connectionWrap{
Defaults: dft,
}
cw.connection, err = connect(ctx, svc.logger, svc.inDev, dsn, capabilities...)
if err != nil {
return
}
svc.connections[connectionID] = cw
return
}
// RemoveConnection removes the given connection from the DAL
func (svc *service) RemoveConnection(ctx context.Context, connectionID uint64) (err error) {
c := svc.connections[connectionID]
if c == nil {
return fmt.Errorf("can not remove connection %d: connection does not exist", connectionID)
}
// Potential cleanups
if cc, ok := c.connection.(ConnectionCloser); ok {
if err = cc.Close(ctx); err != nil {
return err
}
}
// Remove from registry
delete(svc.connections, connectionID)
return nil
}
// ConnectionDefaultreturns the defaults we can use with this connection
func (svc *service) ConnectionDefaults(ctx context.Context, connectionID uint64) (dft ConnectionDefaults, err error) {
wrap, _, err := svc.getConnection(ctx, connectionID)
if err != nil {
return
}
return wrap.Defaults, nil
}
// // // // // // // // // // // // // // // // // // // // // // // // //
// // // // // // // // // // // // // // // // // // // // // // // // //
// DML
func (svc *service) Create(ctx context.Context, mf ModelFilter, capabilities capabilities.Set, rr ...ValueGetter) (err error) {
model, cw, err := svc.storeOpPrep(ctx, mf, capabilities)
if err != nil {
return
}
return cw.connection.Create(ctx, model, rr...)
}
func (svc *service) Search(ctx context.Context, mf ModelFilter, capabilities capabilities.Set, f filter.Filter) (iter Iterator, err error) {
model, cw, err := svc.storeOpPrep(ctx, mf, capabilities)
if err != nil {
return
}
return cw.connection.Search(ctx, model, f)
}
// getStore returns a store for the given identifier/capabilities combination
func (svc *service) getStore(ctx context.Context, storeID uint64, cc ...capabilities.Capability) (store StoreConnection, can capabilities.Set, err error) {
err = func() error {
// get the requested store
if storeID == defaultStoreID {
store = svc.primary
} else {
store = svc.stores[storeID]
}
if store == nil {
return fmt.Errorf("could not get store %d: store does not exist", storeID)
}
func (svc *service) Lookup(ctx context.Context, mf ModelFilter, capabilities capabilities.Set, lookup ValueGetter, dst ValueSetter) (err error) {
model, cw, err := svc.storeOpPrep(ctx, mf, capabilities)
if err != nil {
return
}
return cw.connection.Lookup(ctx, model, lookup, dst)
}
// check if store supports requested capabilities
if !store.Can(cc...) {
return fmt.Errorf("store does not support requested capabilities: %v", capabilities.Set(cc).Diff(store.Capabilities()))
}
can = store.Capabilities()
return nil
}()
func (svc *service) storeOpPrep(ctx context.Context, mf ModelFilter, capabilities capabilities.Set) (model *Model, cw *connectionWrap, err error) {
model = svc.getModelByFilter(mf)
if model == nil {
err = fmt.Errorf("cannot perform operation: model not registered")
return
}
cw, _, err = svc.getConnection(ctx, model.ConnectionID, capabilities...)
if err != nil {
err = fmt.Errorf("could not connect to store %d: %v", storeID, err)
return
}
return
}
// ReloadModulesFromStore resets state based on the provided cStore
func (svc *service) ReloadModulesFromStore(ctx context.Context, cs cStore) (err error) {
modules, err := svc.loadModules(ctx, cs)
if err != nil {
return
}
// // // // // // // // // // // // // // // // // // // // // // // // //
return svc.ReloadModules(ctx, modules...)
}
// // // // // // // // // // // // // // // // // // // // // // // // //
// DDL
// ReloadModulesFromStore resets state based on the provided set of modules
func (svc *service) ReloadModules(ctx context.Context, modules ...*types.Module) (err error) {
// ReloadModel unregister old models and register the new ones
func (svc *service) ReloadModel(ctx context.Context, models ...*Model) (err error) {
// Clear up the old ones
// @todo profile if manually removing nested pointers makes it faster
svc.models = make(map[uint64]ModelSet)
return svc.AddModules(ctx, modules...)
return svc.AddModel(ctx, models...)
}
// AddModules adds new modules without affecting existing ones
func (svc *service) AddModules(ctx context.Context, modules ...*types.Module) (err error) {
models, err := svc.modulesToModel(modules...)
if err != nil {
return
}
// AddModel adds support for a new model
func (svc *service) AddModel(ctx context.Context, models ...*Model) (err error) {
var (
s StoreConnection
cw *connectionWrap
)
for storeID, models := range svc.modelByStore(models) {
s, _, err = svc.getStore(ctx, storeID)
for connectionID, models := range svc.modelByConnection(models) {
cw, _, err = svc.getConnection(ctx, connectionID)
if err != nil {
return err
}
err = svc.addModel(ctx, s, storeID, models)
err = svc.registerModel(ctx, cw.connection, connectionID, models)
if err != nil {
return
}
@@ -328,19 +206,14 @@ func (svc *service) AddModules(ctx context.Context, modules ...*types.Module) (e
return
}
// RemoveModules removes the specified modules
func (svc *service) RemoveModules(ctx context.Context, modules ...*types.Module) (err error) {
models, err := svc.modulesToModel(modules...)
if err != nil {
return
}
// RemoveModel removes support for the given model
func (svc *service) RemoveModel(ctx context.Context, models ...*Model) (err error) {
// validation
for _, model := range models {
// Validate existence
old := svc.getModel(model.StoreID, model.Ident)
old := svc.GetModelByResource(model.ConnectionID, model.ResourceType, model.Resource)
if old == nil {
return fmt.Errorf("cannot remove module %s: not registered", model.Ident)
return fmt.Errorf("cannot remove module %s: not registered", model.Resource)
}
// Validate no leftover references
@@ -348,22 +221,21 @@ func (svc *service) RemoveModules(ctx context.Context, modules ...*types.Module)
for _, registered := range svc.models {
refs := registered.FilterByReferenced(model)
if len(refs) > 0 {
return fmt.Errorf("cannot remove module %s: referenced by other modules", model.Ident)
return fmt.Errorf("cannot remove module %s: referenced by other modules", model.Resource)
}
}
}
// Work
for _, model := range models {
oldModels := svc.models[model.StoreID]
svc.models[model.StoreID] = make(ModelSet, 0, len(oldModels))
oldModels := svc.models[model.ConnectionID]
svc.models[model.ConnectionID] = make(ModelSet, 0, len(oldModels))
for _, o := range oldModels {
if o.Ident == model.Ident {
if o.Resource == model.Resource {
continue
}
svc.models[model.StoreID] = append(svc.models[model.StoreID], o)
svc.models[model.ConnectionID] = append(svc.models[model.ConnectionID], o)
}
// @todo should the underlying store be notified about this?
@@ -372,327 +244,79 @@ func (svc *service) RemoveModules(ctx context.Context, modules ...*types.Module)
return nil
}
// AlterModule updates the old module with the new one
func (svc *service) AlterModule(ctx context.Context, oldMod, newMod *types.Module) (err error) {
return
// // validation
// {
// if oldMod.Store.ComposeRecordStoreID != newMod.Store.ComposeRecordStoreID {
// return fmt.Errorf("cannot alter module stored in different record stores: old: %d, new: %d", oldMod.Store.ComposeRecordStoreID, newMod.Store.ComposeRecordStoreID)
// }
// }
// store, oldModel, err := svc.prepModuleDDL(ctx, oldMod)
// if err != nil {
// return
// }
// // store is same so we omit
// _, newModel, err := svc.prepModuleDDL(ctx, newMod)
// if err != nil {
// return
// }
// return store.AlterModel(ctx, oldModel, newModel)
// DeleteModel removes support for the model and deletes it from the connection
//
// @todo do we really want this?
func (svc *service) DeleteModel(ctx context.Context, models ...*Model) (err error) {
panic("implement DeleteModel")
}
// @todo other ddl manipupations...
// ---
// func (crs *service) prepModuleDDL(ctx context.Context, module *types.Module) (s Store, model *data.Model, err error) {
// s, _, err = crs.getStore(ctx, module.Store.ComposeRecordStoreID)
// if err != nil {
// return
// }
// models, err := crs.modulesToModel(module)
// if err != nil {
// return
// }
// model = models[0]
// return
// }
// modelByStore maps the given models by their CRS
func (svc *service) modelByStore(models ModelSet) (out map[uint64]ModelSet) {
out = make(map[uint64]ModelSet)
for _, model := range models {
out[model.StoreID] = append(out[model.StoreID], model)
}
return
func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) error {
panic("implement UpdateModel")
}
// loadModules is a utility to load available modules with all their metadata included
func (svc *service) loadModules(ctx context.Context, cs cStore) (mm types.ModuleSet, err error) {
var (
namespaces types.NamespaceSet
modules types.ModuleSet
fields types.ModuleFieldSet
)
func (svc *service) UpdateModelAttribute(ctx context.Context, sch *Model, old Attribute, new Attribute, trans ...TransformationFunction) error {
panic("implement UpdateModelAttribute")
}
// // // // // // // // // // // // // // // // // // // // // // // // //
// Utilities
func (svc *service) GetModelByID(connectionID uint64, id uint64) *Model {
return svc.models[connectionID].FindByID(id)
}
func (svc *service) GetModelByResource(connectionID uint64, resType string, resource string) *Model {
return svc.models[connectionID].FindByResource(resType, resource)
}
func (svc *service) getConnection(ctx context.Context, connectionID uint64, cc ...capabilities.Capability) (cw *connectionWrap, can capabilities.Set, err error) {
err = func() error {
// get the requested connection
if connectionID == DefaultConnectionID {
cw = svc.primary
} else {
cw = svc.connections[connectionID]
}
if cw == nil {
return fmt.Errorf("could not get connection %d: store does not exist", connectionID)
}
// check if connection supports requested capabilities
if !cw.connection.Can(cc...) {
return fmt.Errorf("connection does not support requested capabilities: %v", capabilities.Set(cc).Diff(cw.connection.Capabilities()))
}
can = cw.connection.Capabilities()
return nil
}()
namespaces, _, err = cs.SearchComposeNamespaces(ctx, types.NamespaceFilter{})
if err != nil {
err = fmt.Errorf("could not connect to %d: %v", connectionID, err)
return
}
for _, ns := range namespaces {
modules, _, err = cs.SearchComposeModules(ctx, types.ModuleFilter{
NamespaceID: ns.ID,
})
if err != nil {
return
}
for _, mod := range modules {
fields, _, err = cs.SearchComposeModuleFields(ctx, types.ModuleFieldFilter{
ModuleID: []uint64{mod.ID},
})
if err != nil {
return
}
mod.Fields = append(mod.Fields, fields...)
}
}
return
}
// moduleFieldCodec is a little utility to construct the store codec we need
// @todo compose/types.Module to dal.Model conversion MUST happen inside
// compose/service package
func moduleFieldCodec(f *types.ModuleField) (strat Codec) {
// Defaulting to alias
strat = CodecAlias{
Ident: f.Name,
}
// modelByConnection maps the given models by their CRS
func (svc *service) modelByConnection(models ModelSet) (out map[uint64]ModelSet) {
out = make(map[uint64]ModelSet)
switch {
case f.Encoding.EncodingStrategyAlias != nil:
strat = CodecAlias{
Ident: f.Encoding.EncodingStrategyAlias.Ident,
}
case f.Encoding.EncodingStrategyJSON != nil:
strat = CodecRecordValueSetJSON{
Ident: f.Encoding.EncodingStrategyJSON.Ident,
}
}
return
}
// ----------
// @todo compose/types.Module to dal.Model conversion MUST happen inside
// compose/service package
func (svc *service) lookupModel(module *types.Module) (out *Model) {
for _, model := range svc.models[module.Store.ComposeRecordStoreID] {
if model.ResourceID == module.ID {
return model
}
}
return nil
}
// @todo compose/types.Module to dal.Model conversion MUST happen inside
// compose/service package
func (svc *service) modulesToModel(modules ...*types.Module) (out ModelSet, err error) {
refIndex := make(map[uint64]*Model)
out = make(ModelSet, 0, len(modules))
// Initial pass to get everything we can
for _, module := range modules {
model := svc.moduleModelInit(module)
refIndex[module.ID] = model
out = append(out, model)
}
// Add stuff we already have
for _, models := range svc.models {
for _, model := range models {
refIndex[model.ResourceID] = model
}
}
// Build up fields
for i, mod := range modules {
out[i].Attributes, err = svc.moduleModelAttributes(mod, refIndex)
if err != nil {
return
}
}
return
}
// @todo compose/types.Module to dal.Model conversion MUST happen inside
// compose/service package
func (svc *service) moduleModelInit(mod *types.Module) (out *Model) {
return &Model{
StoreID: mod.Store.ComposeRecordStoreID,
ResourceID: mod.ID,
ResourceType: types.ModuleResourceType,
Ident: formatPartitionIdent(mod),
Attributes: make(AttributeSet, len(mod.Fields)),
}
}
// @todo compose/types.Module to dal.Model conversion MUST happen inside
// compose/service package
func (svc *service) moduleModelAttributes(mod *types.Module, refIndex map[uint64]*Model) (out AttributeSet, err error) {
for _, f := range mod.Fields {
attr := &Attribute{
Ident: f.Name,
MultiValue: f.Multi,
Store: moduleFieldCodec(f),
}
out = append(out, attr)
switch strings.ToLower(f.Kind) {
case "bool":
attr.Type = TypeBoolean{}
case "datetime":
switch {
case f.IsDateOnly():
attr.Type = TypeDate{}
case f.IsTimeOnly():
attr.Type = TypeTime{}
default:
attr.Type = TypeTimestamp{}
}
case "email":
attr.Type = TypeText{Length: emailLength}
case "file":
attr.Type = TypeRef{
RefModel: &Model{Ident: "attachments"},
RefAttribute: &Attribute{Ident: "id"},
}
case "number":
attr.Type = TypeNumber{
Precision: f.Options.Precision(),
// Scale: ,
}
case "record":
var refModel *Model
mRefID := f.Options.UInt64("moduleID")
if mRefID > 0 {
refModel = refIndex[mRefID]
}
attr.Type = TypeRef{
RefModel: refModel,
RefAttribute: &Attribute{
Ident: "id",
},
}
case "select":
attr.Type = TypeEnum{
Values: f.SelectOptions(),
}
case "string":
attr.Type = TypeText{
Length: 0,
}
case "url":
attr.Type = TypeText{
Length: urlLength,
}
case "user":
attr.Type = TypeRef{
// @todo...
RefAttribute: &Attribute{
Ident: "id",
},
}
}
}
// System attrs
out = append(out,
&Attribute{
Ident: sysID,
Store: moduleFieldCodec(&types.ModuleField{
Name: sysID,
}),
Type: TypeID{},
},
&Attribute{
Ident: sysCreatedAt,
Store: moduleFieldCodec(&types.ModuleField{
Name: sysCreatedAt,
}),
Type: TypeTimestamp{},
},
&Attribute{
Ident: sysUpdatedAt,
Store: moduleFieldCodec(&types.ModuleField{
Name: sysUpdatedAt,
}),
Type: TypeTimestamp{},
},
&Attribute{
Ident: sysDeletedAt,
Store: moduleFieldCodec(&types.ModuleField{
Name: sysDeletedAt,
}),
Type: TypeTimestamp{},
},
&Attribute{
Ident: sysOwnedBy,
Store: moduleFieldCodec(&types.ModuleField{
Name: sysOwnedBy,
}),
Type: TypeRef{
RefAttribute: &Attribute{Ident: "id"},
},
},
&Attribute{
Ident: sysCreatedBy,
Store: moduleFieldCodec(&types.ModuleField{
Name: sysCreatedBy,
}),
Type: TypeRef{
RefAttribute: &Attribute{Ident: "id"},
},
},
&Attribute{
Ident: sysUpdatedBy,
Store: moduleFieldCodec(&types.ModuleField{
Name: sysUpdatedBy,
}),
Type: TypeRef{
RefAttribute: &Attribute{Ident: "id"},
},
},
&Attribute{
Ident: sysDeletedBy,
Store: moduleFieldCodec(&types.ModuleField{
Name: sysDeletedBy,
}),
Type: TypeRef{
RefAttribute: &Attribute{Ident: "id"},
},
},
)
return
}
func (svc *service) addModel(ctx context.Context, s StoreConnection, storeID uint64, models ModelSet) (err error) {
for _, model := range models {
existing := svc.getModel(storeID, model.Ident)
out[model.ConnectionID] = append(out[model.ConnectionID], model)
}
return
}
func (svc *service) registerModel(ctx context.Context, s Connection, storeID uint64, models ModelSet) (err error) {
for _, model := range models {
existing := svc.GetModelByResource(storeID, model.ResourceType, model.Resource)
if existing != nil {
return fmt.Errorf("cannot add model %s to store %d: already exists", model.Ident, storeID)
return fmt.Errorf("cannot add model %s to store %d: already exists", model.Resource, storeID)
}
err = svc.addModelToStore(ctx, s, model)
err = svc.registerModelToConnection(ctx, s, model)
if err != nil {
return
}
@@ -703,25 +327,25 @@ func (svc *service) addModel(ctx context.Context, s StoreConnection, storeID uin
return
}
func (svc *service) addModelToStore(ctx context.Context, s StoreConnection, model *Model) (err error) {
func (svc *service) registerModelToConnection(ctx context.Context, s Connection, model *Model) (err error) {
available, err := s.Models(ctx)
if err != nil {
return err
}
// Check if already in there
if existing := available.FindByIdent(model.Ident); existing != nil {
if existing := available.FindByResource(model.ResourceType, model.Resource); existing != nil {
// Assert validity
diff := existing.Diff(model)
if len(diff) > 0 {
return fmt.Errorf("model %s exists: model not compatible: %v", existing.Ident, diff)
return fmt.Errorf("model %s exists: model not compatible: %v", existing.Resource, diff)
}
return nil
}
// Try to add to store
err = s.AddModel(ctx, model)
err = s.CreateModel(ctx, model)
if err != nil {
return
}
@@ -729,18 +353,9 @@ func (svc *service) addModelToStore(ctx context.Context, s StoreConnection, mode
return nil
}
// ---
// @todo compose/types.Module to dal.Model conversion MUST happen inside
// compose/service package
func formatPartitionIdent(mod *types.Module) string {
rpl := strings.NewReplacer(
"{{module}}", mod.Handle,
)
if mod.Store.PartitionFormat == "" {
return mod.Handle
func (svc *service) getModelByFilter(mf ModelFilter) *Model {
if mf.ResourceID > 0 {
return svc.GetModelByID(mf.ConnectionID, mf.ResourceID)
}
return rpl.Replace(mod.Store.PartitionFormat)
return svc.GetModelByResource(mf.ConnectionID, mf.ResourceType, mf.Resource)
}
pkg/dal/test/driver.go
+8 -8
View File
@@ -14,12 +14,12 @@ import (
"github.com/stretchr/testify/require"
)
func All(t *testing.T, d dal.StoreConnection) {
func All(t *testing.T, d dal.Connection) {
t.Run("RecordCodec", func(t *testing.T) { RecordCodec(t, d) })
t.Run("RecordSearch", func(t *testing.T) { RecordSearch(t, d) })
}
func RecordCodec(t *testing.T, d dal.StoreConnection) {
func RecordCodec(t *testing.T, d dal.Connection) {
var (
req = require.New(t)
@@ -113,10 +113,10 @@ func RecordCodec(t *testing.T, d dal.StoreConnection) {
rIn.Values = rIn.Values.Set(&types.RecordValue{Name: "pUUID", Value: "ba485865-54f9-44de-bde8-6965556c022a"})
rIn.Values = rIn.Values.GetClean()
req.NoError(d.CreateRecords(ctx, m, &rIn))
req.NoError(d.Create(ctx, m, &rIn))
rOut = new(types.Record)
req.NoError(d.LookupRecord(ctx, m, dal.PKValues{"id": rIn.ID}, rOut))
req.NoError(d.Lookup(ctx, m, dal.PKValues{"id": rIn.ID}, rOut))
{
// normalize timezone on timestamps
@@ -134,7 +134,7 @@ func RecordCodec(t *testing.T, d dal.StoreConnection) {
}
}
func RecordSearch(t *testing.T, d dal.StoreConnection) {
func RecordSearch(t *testing.T, d dal.Connection) {
const (
totalRecords = 10
)
@@ -173,7 +173,7 @@ func RecordSearch(t *testing.T, d dal.StoreConnection) {
r.Values = r.Values.Set(&types.RecordValue{Name: "p_number", Value: strconv.Itoa(i)})
r.Values = r.Values.Set(&types.RecordValue{Name: "p_is_odd", Value: strconv.FormatBool(i%2 == 1)})
req.NoError(d.CreateRecords(ctx, m, r))
req.NoError(d.Create(ctx, m, r))
}
cases := []struct {
@@ -219,7 +219,7 @@ func RecordSearch(t *testing.T, d dal.StoreConnection) {
req = require.New(t)
)
i, err := d.SearchRecords(ctx, m, c.f.ToFilter())
i, err := d.Search(ctx, m, c.f.ToFilter())
req.NoError(err)
rr, err := drain(ctx, i)
@@ -239,7 +239,7 @@ func RecordSearch(t *testing.T, d dal.StoreConnection) {
f.PageCursor = cur
f.Limit = lim
req.NoError(f.Sort.Set(orderBy))
i, err := d.SearchRecords(ctx, m, f.ToFilter())
i, err := d.Search(ctx, m, f.ToFilter())
req.NoError(err)
req.NoError(i.Err())
store/adapters/rdbms/dal/connection.go
+27 -30
View File
@@ -13,84 +13,81 @@ import (
type (
connection struct {
mux sync.RWMutex
models map[string]*model
mux sync.RWMutex
models map[string]*model
capabilities capabilities.Set
db *sqlx.DB
dialect drivers.Dialect
}
)
func Connection(db *sqlx.DB, dialect drivers.Dialect) *connection {
func Connection(db *sqlx.DB, dialect drivers.Dialect, cc ...capabilities.Capability) *connection {
return &connection{
db: db,
dialect: dialect,
models: make(map[string]*model),
db: db,
dialect: dialect,
models: make(map[string]*model),
capabilities: cc,
}
}
func (c *connection) model(m *dal.Model) *model {
c.mux.RLock()
if c.models[m.Ident] == nil {
if c.models[m.Resource] == nil {
c.mux.RUnlock()
c.mux.Lock()
c.models[m.Ident] = Model(m, c.db, c.dialect)
c.models[m.Resource] = Model(m, c.db, c.dialect)
defer c.mux.Unlock()
return c.models[m.Ident]
return c.models[m.Resource]
}
defer c.mux.RUnlock()
return c.models[m.Ident]
return c.models[m.Resource]
}
func (c *connection) Capabilities() capabilities.Set {
//TODO implement me
panic("implement me")
return c.capabilities
}
func (c *connection) Can(capabilities ...capabilities.Capability) bool {
//TODO implement me
panic("implement me")
return c.capabilities.IsSuperset(capabilities...)
}
func (c *connection) Close(ctx context.Context) error {
//return c.db.Close() // <<= should we really?
return nil
}
func (c *connection) CreateRecords(ctx context.Context, m *dal.Model, rr ...dal.ValueGetter) error {
func (c *connection) Create(ctx context.Context, m *dal.Model, rr ...dal.ValueGetter) error {
return c.model(m).Create(ctx, rr...)
}
func (c *connection) LookupRecord(ctx context.Context, m *dal.Model, pkv dal.ValueGetter, r dal.ValueSetter) error {
func (c *connection) Lookup(ctx context.Context, m *dal.Model, pkv dal.ValueGetter, r dal.ValueSetter) error {
return c.model(m).Lookup(ctx, pkv, r)
}
func (c *connection) SearchRecords(ctx context.Context, m *dal.Model, f filter.Filter) (dal.Iterator, error) {
func (c *connection) Search(ctx context.Context, m *dal.Model, f filter.Filter) (dal.Iterator, error) {
return c.model(m).Search(f)
}
func (c *connection) Models(ctx context.Context) (dal.ModelSet, error) {
//TODO implement me
return nil, nil
panic("implement me")
}
func (c *connection) AddModel(ctx context.Context, model *dal.Model, model2 ...*dal.Model) error {
func (c *connection) CreateModel(ctx context.Context, model *dal.Model, model2 ...*dal.Model) error {
//TODO implement me
return nil
panic("implement me")
}
func (c *connection) DeleteModel(ctx context.Context, model *dal.Model, model2 ...*dal.Model) error {
//TODO implement me
panic("implement me")
}
func (c *connection) RemoveModel(ctx context.Context, model *dal.Model, model2 ...*dal.Model) error {
func (c *connection) UpdateModel(ctx context.Context, old *dal.Model, new *dal.Model) error {
//TODO implement me
panic("implement me")
}
func (c *connection) AlterModel(ctx context.Context, old *dal.Model, new *dal.Model) error {
//TODO implement me
panic("implement me")
}
func (c *connection) AlterModelAttribute(ctx context.Context, sch *dal.Model, old dal.Attribute, new dal.Attribute, trans ...dal.TransformationFunction) error {
func (c *connection) UpdateModelAttribute(ctx context.Context, sch *dal.Model, old dal.Attribute, new dal.Attribute, trans ...dal.TransformationFunction) error {
//TODO implement me
panic("implement me")
}
store/adapters/rdbms/dal/fetcher.go → store/adapters/rdbms/dal/iterator.go
View File
store/adapters/rdbms/dal/model.go
+1
View File
@@ -227,6 +227,7 @@ func (d *model) searchSql(f filter.Filter) *goqu.SelectDataset {
return base.SetError(fmt.Errorf("unknown attribute %q used for state constrant", ident))
}
// @note why?
if !attr.PrimaryKey {
continue
}
store/adapters/rdbms/drivers/mysql/connect.go
+19 -14
View File
@@ -31,20 +31,8 @@ func init() {
}
func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) {
var (
db *sqlx.DB
cfg *rdbms.ConnConfig
)
if cfg, err = NewConfig(dsn); err != nil {
return
}
if db, err = rdbms.Connect(ctx, logger.Default(), cfg); err != nil {
return
}
if err = connSetup(ctx, db); err != nil {
db, cfg, err := connectBase(ctx, dsn)
if err != nil {
return
}
@@ -63,6 +51,23 @@ func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) {
return s, nil
}
func connectBase(ctx context.Context, dsn string) (db *sqlx.DB, cfg *rdbms.ConnConfig, err error) {
if cfg, err = NewConfig(dsn); err != nil {
return
}
if db, err = rdbms.Connect(ctx, logger.Default(), cfg); err != nil {
return
}
// See https://dev.mysql.com/doc/refman/8.0/en/sql-mode.html#sqlmode_ansi for details
if _, err = db.ExecContext(ctx, `SET SESSION sql_mode = 'ANSI'`); err != nil {
return
}
return
}
// NewConfig validates given DSN and ensures
// params are present and correct
//
store/adapters/rdbms/drivers/mysql/dal.go
+4 -20
View File
@@ -5,33 +5,17 @@ import (
"github.com/cortezaproject/corteza-server/pkg/dal"
"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
"github.com/cortezaproject/corteza-server/pkg/logger"
"github.com/cortezaproject/corteza-server/store/adapters/rdbms"
rdbmsdal "github.com/cortezaproject/corteza-server/store/adapters/rdbms/dal"
"github.com/jmoiron/sqlx"
)
func init() {
dal.Register(dalConnector, baseSchema, debugSchema)
}
func dalConnector(ctx context.Context, dsn string, cc ...capabilities.Capability) (_ dal.StoreConnection, err error) {
var (
db *sqlx.DB
cfg *rdbms.ConnConfig
)
if cfg, err = NewConfig(dsn); err != nil {
func dalConnector(ctx context.Context, dsn string, cc ...capabilities.Capability) (_ dal.Connection, err error) {
db, _, err := connectBase(ctx, dsn)
if err != nil {
return
}
if db, err = rdbms.Connect(ctx, logger.Default(), cfg); err != nil {
return
}
if err = connSetup(ctx, db); err != nil {
return
}
return rdbmsdal.Connection(db, Dialect()), nil
return rdbmsdal.Connection(db, Dialect(), cc...), nil
}
store/adapters/rdbms/drivers/postgres/dal.go
+1 -1
View File
@@ -15,7 +15,7 @@ func init() {
dal.Register(dalConnector, baseSchema, debugSchema)
}
func dalConnector(ctx context.Context, dsn string, cc ...capabilities.Capability) (_ dal.StoreConnection, err error) {
func dalConnector(ctx context.Context, dsn string, cc ...capabilities.Capability) (_ dal.Connection, err error) {
var (
db *sqlx.DB
cfg *rdbms.ConnConfig
store/adapters/rdbms/drivers/sqlite/dal.go
+1 -1
View File
@@ -15,7 +15,7 @@ func init() {
dal.Register(dalConnector, baseSchema, altSchema, debugSchema)
}
func dalConnector(ctx context.Context, dsn string, cc ...capabilities.Capability) (_ dal.StoreConnection, err error) {
func dalConnector(ctx context.Context, dsn string, cc ...capabilities.Capability) (_ dal.Connection, err error) {
var (
db *sqlx.DB
cfg *rdbms.ConnConfig
store/adapters/rdbms/drivers/types.go
+1 -1
View File
@@ -71,7 +71,7 @@ func TypeWrap(dt dal.Type) Type {
return &TypeUUID{c}
}
panic("type implementation missing")
panic(fmt.Sprintf("type implementation missing: %s", dt.Type()))
}
func (*TypeID) MakeScanBuffer() any { return new(ID) }