Connect external and internal CRS commponents
This commit is contained in:
Tomaž Jerman
+33
-76
@@ -6,29 +6,23 @@ import (
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
"github.com/cortezaproject/corteza-server/pkg/data"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type (
|
||||
// the core struct that outlines the compose record store facility
|
||||
composeRecordStore struct {
|
||||
drivers []driver
|
||||
stores map[uint64]*storeWrap
|
||||
stores map[uint64]StoreConnection
|
||||
|
||||
// Indexed by corresponding storeID
|
||||
models map[uint64]data.ModelSet
|
||||
|
||||
primary *storeWrap
|
||||
}
|
||||
primary StoreConnection
|
||||
|
||||
storeWrap struct {
|
||||
store connFunc
|
||||
driver driver
|
||||
dsn string
|
||||
capabilities capabilities.Set
|
||||
logger *zap.Logger
|
||||
inDev bool
|
||||
}
|
||||
|
||||
connFunc func(ctx context.Context) (Store, error)
|
||||
|
||||
crsDefiner interface {
|
||||
ComposeRecordStoreID() uint64
|
||||
StoreDSN() string
|
||||
@@ -41,64 +35,48 @@ const (
|
||||
)
|
||||
|
||||
// ComposeRecordStore initializes a fresh record store where the given store serves as the default
|
||||
func ComposeRecordStore(ctx context.Context, primary crsDefiner, d driver, drivers ...driver) (*composeRecordStore, error) {
|
||||
drivers = append([]driver{d}, drivers...)
|
||||
|
||||
func ComposeRecordStore(ctx context.Context, log *zap.Logger, inDev bool, primary crsDefiner, stores ...crsDefiner) (*composeRecordStore, error) {
|
||||
crs := &composeRecordStore{
|
||||
drivers: drivers,
|
||||
stores: make(map[uint64]*storeWrap),
|
||||
stores: make(map[uint64]StoreConnection),
|
||||
models: make(map[uint64]data.ModelSet),
|
||||
primary: nil,
|
||||
|
||||
logger: log,
|
||||
inDev: inDev,
|
||||
}
|
||||
|
||||
d = crs.getDriver(primary)
|
||||
if d == nil {
|
||||
return nil, fmt.Errorf("could not add default store: no supported driver found")
|
||||
var err error
|
||||
|
||||
crs.primary, err = connect(ctx, log, primary, inDev)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
crs.primary = &storeWrap{
|
||||
store: storeConnWrap(d, primary),
|
||||
driver: d,
|
||||
dsn: primary.StoreDSN(),
|
||||
capabilities: primary.Capabilities(),
|
||||
}
|
||||
|
||||
return crs, nil
|
||||
return crs, crs.AddStore(ctx, stores...)
|
||||
}
|
||||
|
||||
// AddStore registers the given store definitions as compose record stores
|
||||
func (crs *composeRecordStore) AddStore(ctx context.Context, definers ...crsDefiner) error {
|
||||
func (crs *composeRecordStore) AddStore(ctx context.Context, definers ...crsDefiner) (err error) {
|
||||
for _, definer := range definers {
|
||||
if crs.stores[definer.ComposeRecordStoreID()] != nil {
|
||||
return fmt.Errorf("can not add compose record store %d: already defined", definer.ComposeRecordStoreID())
|
||||
}
|
||||
|
||||
d := crs.getDriver(definer)
|
||||
if d == nil {
|
||||
return fmt.Errorf("could not add store %d: no supported driver found", definer.ComposeRecordStoreID())
|
||||
}
|
||||
|
||||
crs.stores[definer.ComposeRecordStoreID()] = &storeWrap{
|
||||
store: storeConnWrap(d, definer),
|
||||
driver: d,
|
||||
dsn: definer.StoreDSN(),
|
||||
capabilities: definer.Capabilities(),
|
||||
crs.stores[definer.ComposeRecordStoreID()], err = connect(ctx, crs.logger, definer, crs.inDev)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RemoveStore removes the given store definition as a compose record store
|
||||
func (crs *composeRecordStore) RemoveStore(ctx context.Context, storeIDs ...uint64) (err error) {
|
||||
for _, storeID := range storeIDs {
|
||||
func (crs *composeRecordStore) RemoveStore(ctx context.Context, storeID uint64, storeIDs ...uint64) (err error) {
|
||||
for _, storeID := range append(storeIDs, storeID) {
|
||||
s := crs.stores[storeID]
|
||||
if s == nil {
|
||||
return fmt.Errorf("can not remove compose record store %d: store does not exist", storeID)
|
||||
}
|
||||
|
||||
// Any potential driver cleanup
|
||||
err = s.driver.Close(ctx, s.dsn)
|
||||
if err != nil {
|
||||
// Potential cleanups
|
||||
if err = s.Close(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -110,6 +88,7 @@ func (crs *composeRecordStore) RemoveStore(ctx context.Context, storeIDs ...uint
|
||||
}
|
||||
|
||||
// ---
|
||||
// Utilities
|
||||
|
||||
func (crs *composeRecordStore) getModel(store uint64, ident string) *data.Model {
|
||||
for _, model := range crs.models[store] {
|
||||
@@ -122,26 +101,23 @@ func (crs *composeRecordStore) getModel(store uint64, ident string) *data.Model
|
||||
}
|
||||
|
||||
// getStore returns a store for the given identifier/capabilities combination
|
||||
func (crs *composeRecordStore) getStore(ctx context.Context, storeID uint64, cc ...capabilities.Capability) (store Store, can capabilities.Set, err error) {
|
||||
func (crs *composeRecordStore) getStore(ctx context.Context, storeID uint64, cc ...capabilities.Capability) (store StoreConnection, can capabilities.Set, err error) {
|
||||
err = func() error {
|
||||
// get the requested store
|
||||
var wrap *storeWrap
|
||||
if storeID == defaultStoreID {
|
||||
wrap = crs.primary
|
||||
store = crs.primary
|
||||
} else {
|
||||
wrap = crs.stores[storeID]
|
||||
store = crs.stores[storeID]
|
||||
}
|
||||
if wrap == nil {
|
||||
if store == nil {
|
||||
return fmt.Errorf("could not get store %d: store does not exist", storeID)
|
||||
}
|
||||
|
||||
// check if store supports requested capabilities
|
||||
if !wrap.capabilities.IsSuperset(cc...) {
|
||||
return fmt.Errorf("store does not support requested capabilities: %v", capabilities.Set(cc).Diff(wrap.capabilities))
|
||||
if !store.Can(cc...) {
|
||||
return fmt.Errorf("store does not support requested capabilities: %v", capabilities.Set(cc).Diff(store.Capabilities()))
|
||||
}
|
||||
|
||||
store, err = wrap.store(ctx)
|
||||
can = wrap.capabilities
|
||||
can = store.Capabilities()
|
||||
return nil
|
||||
}()
|
||||
|
||||
@@ -152,22 +128,3 @@ func (crs *composeRecordStore) getStore(ctx context.Context, storeID uint64, cc
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// getDriver returns a driver which can be used with the given store
|
||||
func (crs *composeRecordStore) getDriver(def crsDefiner) driver {
|
||||
for _, d := range crs.drivers {
|
||||
if !d.Can(def.StoreDSN(), def.Capabilities()...) {
|
||||
continue
|
||||
}
|
||||
|
||||
return d
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func storeConnWrap(d driver, def crsDefiner) connFunc {
|
||||
return func(ctx context.Context) (Store, error) {
|
||||
return d.Store(ctx, def.StoreDSN())
|
||||
}
|
||||
}
|
||||
|
||||
+80
-86
@@ -6,107 +6,110 @@ import (
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/cortezaproject/corteza-server/pkg/data"
|
||||
)
|
||||
|
||||
// ComposeRecordCreate creates the given records for the given module
|
||||
func (crs *composeRecordStore) ComposeRecordCreate(ctx context.Context, module *types.Module, records ...*types.Record) (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 Store
|
||||
var s StoreConnection
|
||||
if s, _, err = crs.getStore(ctx, module.Store.ComposeRecordStoreID, requiredCap...); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Prepare data
|
||||
// Get model
|
||||
model := crs.lookupModel(module)
|
||||
if model == nil {
|
||||
return crs.modelNotFoundErr(module)
|
||||
}
|
||||
|
||||
aux := make([]Getter, len(records))
|
||||
for i, r := range records {
|
||||
aux[i] = Getter(r)
|
||||
}
|
||||
return s.CreateRecords(ctx, model, aux...)
|
||||
return s.CreateRecords(ctx, model, records...)
|
||||
}
|
||||
|
||||
// @todo...
|
||||
func (crs *composeRecordStore) ComposeRecordSearch(ctx context.Context, module *types.Module, filter *types.RecordFilter) (records types.RecordSet, outFilter *types.RecordFilter, err error) {
|
||||
// Determine requiredCap we'll need
|
||||
requiredCap := capabilities.SearchCapabilities(module.Store.Capabilities...).Union(crs.recFilterCapabilities(filter))
|
||||
|
||||
// Connect to datasource
|
||||
var s Store
|
||||
var cc capabilities.Set
|
||||
_ = cc
|
||||
s, cc, err = crs.getStore(ctx, module.Store.ComposeRecordStoreID, requiredCap...)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Prepare data
|
||||
model := crs.lookupModel(module)
|
||||
if model == nil {
|
||||
return nil, nil, crs.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 := crs.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
|
||||
// // Determine requiredCap we'll need
|
||||
// requiredCap := capabilities.SearchCapabilities(module.Store.Capabilities...).Union(crs.recFilterCapabilities(filter))
|
||||
|
||||
// // Connect to datasource
|
||||
// var s Store
|
||||
// var cc capabilities.Set
|
||||
// _ = cc
|
||||
// s, cc, err = crs.getStore(ctx, module.Store.ComposeRecordStoreID, requiredCap...)
|
||||
// if err != nil {
|
||||
// return
|
||||
// }
|
||||
|
||||
// // Prepare data
|
||||
// model := crs.lookupModel(module)
|
||||
// if model == nil {
|
||||
// return nil, nil, crs.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 := crs.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 (crs *composeRecordStore) recFilterCapabilities(f *types.RecordFilter) (out capabilities.Set) {
|
||||
if f == nil {
|
||||
return
|
||||
@@ -133,12 +136,3 @@ func (crs *composeRecordStore) recFilterCapabilities(f *types.RecordFilter) (out
|
||||
func (crs composeRecordStore) modelNotFoundErr(module *types.Module) error {
|
||||
return fmt.Errorf("cannot create records for module %d: module not registered to crs", module.ID)
|
||||
}
|
||||
|
||||
func (crs composeRecordStore) extractRecords(model *data.Model, gg ...Setter) (out types.RecordSet, err error) {
|
||||
out = make(types.RecordSet, len(gg))
|
||||
for i, g := range gg {
|
||||
out[i] = g.(*types.Record)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
+111
-34
@@ -2,69 +2,103 @@ package crs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/cortezaproject/corteza-server/pkg/data"
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type (
|
||||
driver interface {
|
||||
// Capabilities returns all of the capabilities the driver is able to support
|
||||
StoreConnection interface {
|
||||
// ---
|
||||
|
||||
// Meta
|
||||
// Capabilities returns all of the capabilities the given store supports
|
||||
Capabilities() capabilities.Set
|
||||
// Can determines if the driver is able to handle the connection
|
||||
Can(dsn string, capabilities ...capabilities.Capability) bool
|
||||
// Can returns true if this store can handle the given capabilities
|
||||
Can(capabilities ...capabilities.Capability) bool
|
||||
|
||||
// Store connects and returns a store we can use
|
||||
Store(ctx context.Context, dsn string) (Store, error)
|
||||
// Close closes the connection to specified
|
||||
Close(ctx context.Context, dsn string) error
|
||||
}
|
||||
// ---
|
||||
|
||||
// Loader provides an interface for loading data from the underlying store
|
||||
Loader interface {
|
||||
More() bool
|
||||
Load(*data.Model, []Setter) (coppied int, err error)
|
||||
// Connection stuff
|
||||
// Close closes the store connection allowing the driver to perform potential
|
||||
// cleanup operations
|
||||
Close(ctx context.Context) error
|
||||
|
||||
// -1 means unknown
|
||||
Total() int
|
||||
Cursor() any
|
||||
// ... do we need anything else here?
|
||||
}
|
||||
// ---
|
||||
|
||||
// Store provides an interface which CRS uses to interact with the underlying database
|
||||
Store interface {
|
||||
// DML
|
||||
CreateRecords(ctx context.Context, sch *data.Model, cc ...Getter) error
|
||||
// @note providing a model here would allow us to control what attributes we return; would we find this useful?
|
||||
SearchRecords(ctx context.Context, sch *data.Model, filter any) (Loader, error)
|
||||
// Rest are same difference
|
||||
// CreateRecords stores the given records into the underlying database
|
||||
//
|
||||
// @todo rename to Create and replace types.Record with Getter
|
||||
CreateRecords(ctx context.Context, m *data.Model, rr ...*types.Record) error
|
||||
// SearchRecords returns an iterator we can use to load data
|
||||
//
|
||||
// @todo rename to Search and replace filter with xyz
|
||||
SearchRecords(ctx context.Context, sch *data.Model, filter any) (Iterator, error)
|
||||
// ...
|
||||
// @todo other functions; they are same difference so omitting for now
|
||||
// - LookupRecord
|
||||
// - UpdateRecords
|
||||
// - DeleteRecords
|
||||
// - TruncateRecords
|
||||
|
||||
// ---
|
||||
|
||||
// DDL
|
||||
// Models returns all of the collections the given database already defines
|
||||
// Models returns all of 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) (data.ModelSet, error)
|
||||
|
||||
// returns all attribute types that driver supports
|
||||
AttributeTypes() []data.AttributeType
|
||||
// // returns all attribute types that driver supports
|
||||
// AttributeTypes() []data.AttributeType
|
||||
|
||||
// AddModel requests the driver to support the specified collections
|
||||
// AddModel 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, *data.Model, ...*data.Model) error
|
||||
|
||||
// // RemoveModel requests the driver to remove support for the specified collections
|
||||
// RemoveModel(context.Context, *data.Model, ...*data.Model) error
|
||||
// RemoveModel removes support for the given model from the underlying database
|
||||
RemoveModel(context.Context, *data.Model, ...*data.Model) error
|
||||
|
||||
// AlterModel requests the driver to alter the general collectio parameters
|
||||
// AlterModel 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 *data.Model, new *data.Model) error
|
||||
|
||||
// AlterModelAttribute requests the driver to alter the specified attribute of the given collection
|
||||
AlterModelAttribute(ctx context.Context, sch *data.Model, old data.Attribute, new data.Attribute, trans ...func(*data.Model, data.Attribute, expr.TypedValue) (expr.TypedValue, bool, error)) error
|
||||
// AlterModelAttribute 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 *data.Model, old data.Attribute, new data.Attribute, trans ...TransformationFunction) error
|
||||
}
|
||||
|
||||
// probably somewhere else (on the db level?
|
||||
TransformationFunction func(*data.Model, data.Attribute, expr.TypedValue) (expr.TypedValue, bool, error)
|
||||
|
||||
// Iterator provides an interface for loading data from the underlying store
|
||||
Iterator interface {
|
||||
Next(ctx context.Context) bool
|
||||
Scan(r *types.Record) (err error)
|
||||
Err() error
|
||||
Close() error
|
||||
|
||||
// // -1 means unknown
|
||||
// Total() int
|
||||
// Cursor() any
|
||||
// // ... do we need anything else here?
|
||||
}
|
||||
|
||||
// Store provides an interface which CRS uses to interact with the underlying database
|
||||
|
||||
Getter interface {
|
||||
GetValue(name string, pos int) (any, error)
|
||||
}
|
||||
@@ -72,4 +106,47 @@ type (
|
||||
Setter interface {
|
||||
SetValue(name string, pos int, value any) error
|
||||
}
|
||||
|
||||
ConnectorFn func(ctx context.Context, dsn string, cc ...capabilities.Capability) (StoreConnection, error)
|
||||
)
|
||||
|
||||
var (
|
||||
registered = make(map[string]ConnectorFn)
|
||||
)
|
||||
|
||||
// Register registers a new connector for the given DSN schema
|
||||
//
|
||||
// In case of a duplicate schema the latter overwrites the prior
|
||||
func Register(fn ConnectorFn, tt ...string) {
|
||||
for _, t := range tt {
|
||||
registered[t] = fn
|
||||
}
|
||||
}
|
||||
|
||||
// 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()
|
||||
|
||||
if isDevelopment {
|
||||
if strings.Contains(dsn, "{version}") {
|
||||
log.Warn("You're using DB_DSN with {version}, It is still in EXPERIMENTAL phase")
|
||||
log.Warn("Should be used only for development mode")
|
||||
log.Warn("You may experience instability")
|
||||
}
|
||||
expr := regexp.MustCompile(`[.\-]+`)
|
||||
version := expr.ReplaceAllString(os.Getenv("BUILD_VERSION"), "_")
|
||||
dsn = strings.Replace(dsn, "{version}", version, 1)
|
||||
}
|
||||
|
||||
var storeType = strings.SplitN(dsn, "://", 2)[0]
|
||||
if storeType == "" {
|
||||
// Backward compatibility
|
||||
storeType = "mysql"
|
||||
}
|
||||
|
||||
if conn, ok := registered[storeType]; ok {
|
||||
return conn(ctx, dsn, def.Capabilities()...)
|
||||
} else {
|
||||
return nil, fmt.Errorf("unknown store type used: %q (check your storage configuration)", storeType)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,94 +0,0 @@
|
||||
package crs
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/cortezaproject/corteza-server/pkg/data"
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
"github.com/cortezaproject/corteza-server/pkg/id"
|
||||
)
|
||||
|
||||
type (
|
||||
testDriver struct{}
|
||||
|
||||
testStore map[string]*types.Record
|
||||
)
|
||||
|
||||
var (
|
||||
gStore = make(testStore)
|
||||
|
||||
// wrapper around time.Now() that will aid service testing
|
||||
now = func() *time.Time {
|
||||
c := time.Now().Round(time.Second)
|
||||
return &c
|
||||
}
|
||||
|
||||
// wrapper around nextID that will aid service testing
|
||||
nextID = func() uint64 {
|
||||
return id.Next()
|
||||
}
|
||||
)
|
||||
|
||||
func (d testDriver) Capabilities() capabilities.Set {
|
||||
return capabilities.FullCapabilities()
|
||||
}
|
||||
|
||||
func (d testDriver) Can(uri string, cc ...capabilities.Capability) bool {
|
||||
return strings.HasPrefix(uri, "noop://") && capabilities.Set(cc).IsSubset(d.Capabilities()...)
|
||||
}
|
||||
|
||||
func (d testDriver) Store(ctx context.Context, uri string) (str Store, err error) {
|
||||
return &testStore{}, nil
|
||||
}
|
||||
|
||||
func (d testDriver) Close(ctx context.Context, uri string) (err error) {
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---
|
||||
|
||||
func (ds *testStore) CreateRecords(ctx context.Context, sch *data.Model, cc ...Getter) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ds *testStore) SearchRecords(ctx context.Context, sch *data.Model, filter any) (Loader, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (ds *testStore) Models(context.Context) (data.ModelSet, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (ds *testStore) AddModel(context.Context, *data.Model, ...*data.Model) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ds *testStore) AlterModel(ctx context.Context, old *data.Model, new *data.Model) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ds *testStore) AlterModelAttribute(ctx context.Context, sch *data.Model, old data.Attribute, new data.Attribute, trans ...func(*data.Model, data.Attribute, expr.TypedValue) (expr.TypedValue, bool, error)) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func initCRS(ctx context.Context) *composeRecordStore {
|
||||
crs, _ := ComposeRecordStore(ctx, CRSConnectionWrap(0, "noop://primary", capabilities.FullCapabilities()...), testDriver{})
|
||||
crs.AddStore(ctx, CRSConnectionWrap(defaultExternalCRS, "noop://external", capabilities.FullCapabilities()...))
|
||||
return crs
|
||||
}
|
||||
|
||||
func defaultModule() *types.Module {
|
||||
return &types.Module{
|
||||
ID: nextID(),
|
||||
Store: types.CRSDef{ComposeRecordStoreID: defaultExternalCRS},
|
||||
Fields: types.ModuleFieldSet{{
|
||||
Name: "first_name",
|
||||
Kind: "String",
|
||||
}},
|
||||
}
|
||||
}
|
||||
+91
-69
@@ -26,6 +26,24 @@ const (
|
||||
urlLength = 2048
|
||||
)
|
||||
|
||||
// ----------------------
|
||||
// @todo move this out and unify with RDBMS one
|
||||
|
||||
const (
|
||||
sysID = "ID"
|
||||
sysNamespaceID = "namespaceID"
|
||||
sysModuleID = "moduleID"
|
||||
sysCreatedAt = "createdAt"
|
||||
sysCreatedBy = "createdBy"
|
||||
sysUpdatedAt = "updatedAt"
|
||||
sysUpdatedBy = "updatedBy"
|
||||
sysDeletedAt = "deletedAt"
|
||||
sysDeletedBy = "deletedBy"
|
||||
sysOwnedBy = "ownedBy"
|
||||
)
|
||||
|
||||
// ----------------------
|
||||
|
||||
// ReloadModulesFromStore resets state based on the provided cStore
|
||||
func (crs *composeRecordStore) ReloadModulesFromStore(ctx context.Context, cs cStore) (err error) {
|
||||
modules, err := crs.loadModules(ctx, cs)
|
||||
@@ -52,12 +70,17 @@ func (crs *composeRecordStore) AddModules(ctx context.Context, modules ...*types
|
||||
return
|
||||
}
|
||||
|
||||
var (
|
||||
s StoreConnection
|
||||
)
|
||||
|
||||
for storeID, models := range crs.modelByStore(models) {
|
||||
if crs.stores[storeID] == nil {
|
||||
return fmt.Errorf("can not add module to store %d: store does not exist", storeID)
|
||||
s, _, err = crs.getStore(ctx, storeID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = crs.addModel(ctx, storeID, models)
|
||||
err = crs.addModel(ctx, s, storeID, models)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -112,54 +135,45 @@ func (crs *composeRecordStore) RemoveModules(ctx context.Context, modules ...*ty
|
||||
|
||||
// AlterModule updates the old module with the new one
|
||||
func (crs *composeRecordStore) AlterModule(ctx context.Context, oldMod, newMod *types.Module) (err error) {
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
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 := crs.prepModuleDDL(ctx, oldMod)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
// store is same so we omit
|
||||
_, newModel, err := crs.prepModuleDDL(ctx, newMod)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
// store, oldModel, err := crs.prepModuleDDL(ctx, oldMod)
|
||||
// if err != nil {
|
||||
// return
|
||||
// }
|
||||
// // store is same so we omit
|
||||
// _, newModel, err := crs.prepModuleDDL(ctx, newMod)
|
||||
// if err != nil {
|
||||
// return
|
||||
// }
|
||||
|
||||
return store.AlterModel(ctx, oldModel, newModel)
|
||||
// return store.AlterModel(ctx, oldModel, newModel)
|
||||
}
|
||||
|
||||
// @todo other ddl manipupations...
|
||||
|
||||
// ---
|
||||
|
||||
func (crs *composeRecordStore) 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
|
||||
}
|
||||
// func (crs *composeRecordStore) 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]
|
||||
// models, err := crs.modulesToModel(module)
|
||||
// if err != nil {
|
||||
// return
|
||||
// }
|
||||
// model = models[0]
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// modulesByStore maps the given module set by their CRS
|
||||
func (crs *composeRecordStore) modulesByStore(modules types.ModuleSet) (out map[uint64]types.ModuleSet) {
|
||||
out = make(map[uint64]types.ModuleSet)
|
||||
for _, mod := range modules {
|
||||
out[mod.Store.ComposeRecordStoreID] = append(out[mod.Store.ComposeRecordStoreID], mod)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
// return
|
||||
// }
|
||||
|
||||
// modelByStore maps the given models by their CRS
|
||||
func (crs *composeRecordStore) modelByStore(models data.ModelSet) (out map[uint64]data.ModelSet) {
|
||||
@@ -208,6 +222,7 @@ func (crs *composeRecordStore) loadModules(ctx context.Context, cs cStore) (mm t
|
||||
return
|
||||
}
|
||||
|
||||
// moduleFieldCodec is a little utility to construct the store codec we need
|
||||
func moduleFieldCodec(f *types.ModuleField) (strat data.StoreCodec) {
|
||||
// Defaulting to alias
|
||||
strat = data.StoreCodecAlias{
|
||||
@@ -220,9 +235,8 @@ func moduleFieldCodec(f *types.ModuleField) (strat data.StoreCodec) {
|
||||
Ident: f.Encoding.EncodingStrategyAlias.Ident,
|
||||
}
|
||||
case f.Encoding.EncodingStrategyJSON != nil:
|
||||
strat = data.StoreCodecJSON{
|
||||
strat = data.StoreCodecStdRecordValueJSON{
|
||||
Ident: f.Encoding.EncodingStrategyJSON.Ident,
|
||||
Path: f.Encoding.EncodingStrategyJSON.Path,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -276,8 +290,7 @@ func (crs *composeRecordStore) moduleModelInit(mod *types.Module) (out *data.Mod
|
||||
ResourceID: mod.ID,
|
||||
ResourceType: types.ModuleResourceType,
|
||||
|
||||
// @todo this isn't exactly this
|
||||
Ident: mod.Handle,
|
||||
Ident: formatPartitionIdent(mod),
|
||||
Attributes: make(data.AttributeSet, len(mod.Fields)),
|
||||
}
|
||||
}
|
||||
@@ -354,65 +367,65 @@ func (crs *composeRecordStore) moduleModelAttributes(mod *types.Module, refIndex
|
||||
// System attrs
|
||||
out = append(out,
|
||||
&data.Attribute{
|
||||
Ident: "recordID",
|
||||
Ident: sysID,
|
||||
Store: moduleFieldCodec(&types.ModuleField{
|
||||
Name: "recordID",
|
||||
Name: sysID,
|
||||
}),
|
||||
Type: data.TypeID{},
|
||||
},
|
||||
&data.Attribute{
|
||||
Ident: "createdAt",
|
||||
Ident: sysCreatedAt,
|
||||
Store: moduleFieldCodec(&types.ModuleField{
|
||||
Name: "createdAt",
|
||||
Name: sysCreatedAt,
|
||||
}),
|
||||
Type: data.TypeTimestamp{},
|
||||
},
|
||||
&data.Attribute{
|
||||
Ident: "updatedAt",
|
||||
Ident: sysUpdatedAt,
|
||||
Store: moduleFieldCodec(&types.ModuleField{
|
||||
Name: "updatedAt",
|
||||
Name: sysUpdatedAt,
|
||||
}),
|
||||
Type: data.TypeTimestamp{},
|
||||
},
|
||||
&data.Attribute{
|
||||
Ident: "deletedAt",
|
||||
Ident: sysDeletedAt,
|
||||
Store: moduleFieldCodec(&types.ModuleField{
|
||||
Name: "deletedAt",
|
||||
Name: sysDeletedAt,
|
||||
}),
|
||||
Type: data.TypeTimestamp{},
|
||||
},
|
||||
|
||||
&data.Attribute{
|
||||
Ident: "ownedBy",
|
||||
Ident: sysOwnedBy,
|
||||
Store: moduleFieldCodec(&types.ModuleField{
|
||||
Name: "ownedBy",
|
||||
Name: sysOwnedBy,
|
||||
}),
|
||||
Type: data.TypeRef{
|
||||
RefAttribute: &data.Attribute{Ident: "id"},
|
||||
},
|
||||
},
|
||||
&data.Attribute{
|
||||
Ident: "createdBy",
|
||||
Ident: sysCreatedBy,
|
||||
Store: moduleFieldCodec(&types.ModuleField{
|
||||
Name: "createdBy",
|
||||
Name: sysCreatedBy,
|
||||
}),
|
||||
Type: data.TypeRef{
|
||||
RefAttribute: &data.Attribute{Ident: "id"},
|
||||
},
|
||||
},
|
||||
&data.Attribute{
|
||||
Ident: "updatedBy",
|
||||
Ident: sysUpdatedBy,
|
||||
Store: moduleFieldCodec(&types.ModuleField{
|
||||
Name: "updatedBy",
|
||||
Name: sysUpdatedBy,
|
||||
}),
|
||||
Type: data.TypeRef{
|
||||
RefAttribute: &data.Attribute{Ident: "id"},
|
||||
},
|
||||
},
|
||||
&data.Attribute{
|
||||
Ident: "deletedBy",
|
||||
Ident: sysDeletedBy,
|
||||
Store: moduleFieldCodec(&types.ModuleField{
|
||||
Name: "deletedBy",
|
||||
Name: sysDeletedBy,
|
||||
}),
|
||||
Type: data.TypeRef{
|
||||
RefAttribute: &data.Attribute{Ident: "id"},
|
||||
@@ -423,14 +436,14 @@ func (crs *composeRecordStore) moduleModelAttributes(mod *types.Module, refIndex
|
||||
return
|
||||
}
|
||||
|
||||
func (crs *composeRecordStore) addModel(ctx context.Context, storeID uint64, models data.ModelSet) (err error) {
|
||||
func (crs *composeRecordStore) addModel(ctx context.Context, s StoreConnection, storeID uint64, models data.ModelSet) (err error) {
|
||||
for _, model := range models {
|
||||
existing := crs.getModel(storeID, model.Ident)
|
||||
if existing != nil {
|
||||
return fmt.Errorf("cannot add model %s to store %d: already exists", model.Ident, storeID)
|
||||
}
|
||||
|
||||
err = crs.addModelToStore(ctx, storeID, model)
|
||||
err = crs.addModelToStore(ctx, s, model)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -441,12 +454,7 @@ func (crs *composeRecordStore) addModel(ctx context.Context, storeID uint64, mod
|
||||
return
|
||||
}
|
||||
|
||||
func (crs *composeRecordStore) addModelToStore(ctx context.Context, storeID uint64, model *data.Model) (err error) {
|
||||
s, _, err := crs.getStore(ctx, storeID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
func (crs *composeRecordStore) addModelToStore(ctx context.Context, s StoreConnection, model *data.Model) (err error) {
|
||||
available, err := s.Models(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -471,3 +479,17 @@ func (crs *composeRecordStore) addModelToStore(ctx context.Context, storeID uint
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---
|
||||
|
||||
func formatPartitionIdent(mod *types.Module) string {
|
||||
rpl := strings.NewReplacer(
|
||||
"{{module}}", mod.Handle,
|
||||
)
|
||||
|
||||
if mod.Store.PartitionFormat == "" {
|
||||
return mod.Handle
|
||||
}
|
||||
|
||||
return rpl.Replace(mod.Store.PartitionFormat)
|
||||
}
|
||||
|
||||
@@ -1,77 +0,0 @@
|
||||
package crs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
var (
|
||||
defaultExternalCRS uint64 = nextID()
|
||||
)
|
||||
|
||||
func TestLoadModules(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("no supporting CRS", func(t *testing.T) {
|
||||
crs := initCRS(ctx)
|
||||
|
||||
err := crs.ReloadModules(ctx, &types.Module{
|
||||
ID: nextID(),
|
||||
Store: types.CRSDef{ComposeRecordStoreID: 999},
|
||||
})
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("module added", func(t *testing.T) {
|
||||
crs := initCRS(ctx)
|
||||
|
||||
mod := defaultModule()
|
||||
|
||||
err := crs.ReloadModules(ctx, mod)
|
||||
require.NoError(t, err)
|
||||
|
||||
model := crs.lookupModel(mod)
|
||||
require.NotNil(t, model)
|
||||
|
||||
// 8 sys, 1 custom
|
||||
require.Len(t, model.Attributes, 8+1)
|
||||
|
||||
require.Equal(t, defaultExternalCRS, model.StoreID)
|
||||
})
|
||||
|
||||
t.Run("duplicate module added", func(t *testing.T) {
|
||||
crs := initCRS(ctx)
|
||||
|
||||
mod := defaultModule()
|
||||
|
||||
err := crs.ReloadModules(ctx, mod, mod)
|
||||
require.Error(t, err)
|
||||
})
|
||||
}
|
||||
|
||||
func TestRemoveModules(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("module not found", func(t *testing.T) {
|
||||
crs := initCRS(ctx)
|
||||
|
||||
err := crs.RemoveModules(ctx, defaultModule())
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("module removed", func(t *testing.T) {
|
||||
crs := initCRS(ctx)
|
||||
mod := defaultModule()
|
||||
|
||||
err := crs.ReloadModules(ctx, mod)
|
||||
require.NoError(t, err)
|
||||
|
||||
err = crs.RemoveModules(ctx, mod)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Nil(t, crs.lookupModel(mod))
|
||||
})
|
||||
}
|
||||
@@ -1,66 +0,0 @@
|
||||
// This package is just so that I have an interface conforming driver
|
||||
|
||||
package noop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs"
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/cortezaproject/corteza-server/pkg/data"
|
||||
"github.com/cortezaproject/corteza-server/pkg/expr"
|
||||
)
|
||||
|
||||
type (
|
||||
noopDriver struct{}
|
||||
|
||||
noopStore map[string]*types.Record
|
||||
)
|
||||
|
||||
var (
|
||||
gStore = make(noopStore)
|
||||
)
|
||||
|
||||
func Driver() *noopDriver {
|
||||
return &noopDriver{}
|
||||
}
|
||||
|
||||
func (d noopDriver) Capabilities() capabilities.Set {
|
||||
return capabilities.FullCapabilities()
|
||||
}
|
||||
|
||||
func (d noopDriver) Can(uri string, cc ...capabilities.Capability) bool {
|
||||
return strings.HasPrefix(uri, "noop://") && capabilities.Set(cc).IsSubset(d.Capabilities()...)
|
||||
}
|
||||
|
||||
func (d noopDriver) Store(ctx context.Context, uri string) (str crs.Store, err error) {
|
||||
return &noopStore{}, nil
|
||||
}
|
||||
|
||||
// ---
|
||||
|
||||
func (ds *noopStore) CreateRecords(ctx context.Context, sch *data.Model, cc ...crs.Getter) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ds *noopStore) SearchRecords(ctx context.Context, sch *data.Model, filter any) (crs.Loader, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (ds *noopStore) Models(context.Context) (data.ModelSet, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (ds *noopStore) AddModel(context.Context, *data.Model, ...*data.Model) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ds *noopStore) AlterModel(ctx context.Context, old *data.Model, new *data.Model) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ds *noopStore) AlterModelAttribute(ctx context.Context, sch *data.Model, old data.Attribute, new data.Attribute, trans ...func(*data.Model, data.Attribute, expr.TypedValue) (expr.TypedValue, bool, error)) error {
|
||||
return nil
|
||||
}
|
||||
@@ -1,131 +0,0 @@
|
||||
package crs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestInit(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("invalid primary store", func(t *testing.T) {
|
||||
_, err := ComposeRecordStore(
|
||||
ctx,
|
||||
CRSConnectionWrap(0, "invalid://primary", capabilities.FullCapabilities()...),
|
||||
testDriver{},
|
||||
)
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("valid primary store", func(t *testing.T) {
|
||||
crs, err := ComposeRecordStore(
|
||||
ctx,
|
||||
CRSConnectionWrap(0, "noop://primary", capabilities.FullCapabilities()...),
|
||||
testDriver{},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Len(t, crs.stores, 0)
|
||||
require.NotNil(t, crs.primary)
|
||||
|
||||
})
|
||||
}
|
||||
|
||||
func TestStoreAdd(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("invalid external store", func(t *testing.T) {
|
||||
crs, err := ComposeRecordStore(
|
||||
ctx,
|
||||
CRSConnectionWrap(0, "noop://primary", capabilities.FullCapabilities()...),
|
||||
testDriver{},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
sID := nextID()
|
||||
err = crs.AddStore(ctx, CRSConnectionWrap(sID, "invalid://external", capabilities.FullCapabilities()...))
|
||||
require.Error(t, err)
|
||||
|
||||
require.NotNil(t, crs.primary)
|
||||
require.Len(t, crs.stores, 0)
|
||||
})
|
||||
|
||||
t.Run("valid external store", func(t *testing.T) {
|
||||
crs, err := ComposeRecordStore(
|
||||
ctx,
|
||||
CRSConnectionWrap(0, "noop://primary", capabilities.FullCapabilities()...),
|
||||
testDriver{},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
sID := nextID()
|
||||
err = crs.AddStore(ctx, CRSConnectionWrap(sID, "noop://external", capabilities.FullCapabilities()...))
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NotNil(t, crs.primary)
|
||||
require.Len(t, crs.stores, 1)
|
||||
require.NotNil(t, crs.stores[sID])
|
||||
})
|
||||
|
||||
t.Run("duplicated external store", func(t *testing.T) {
|
||||
crs, err := ComposeRecordStore(
|
||||
ctx,
|
||||
CRSConnectionWrap(0, "noop://primary", capabilities.FullCapabilities()...),
|
||||
testDriver{},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
sID := nextID()
|
||||
err = crs.AddStore(ctx, CRSConnectionWrap(sID, "noop://external", capabilities.FullCapabilities()...))
|
||||
require.NoError(t, err)
|
||||
|
||||
err = crs.AddStore(ctx, CRSConnectionWrap(sID, "noop://external", capabilities.FullCapabilities()...))
|
||||
require.Error(t, err)
|
||||
|
||||
require.NotNil(t, crs.primary)
|
||||
require.Len(t, crs.stores, 1)
|
||||
require.NotNil(t, crs.stores[sID])
|
||||
})
|
||||
}
|
||||
|
||||
func TestStoreRemove(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("removing missing store", func(t *testing.T) {
|
||||
crs, err := ComposeRecordStore(
|
||||
ctx,
|
||||
CRSConnectionWrap(0, "noop://primary", capabilities.FullCapabilities()...),
|
||||
testDriver{},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
sID := nextID()
|
||||
err = crs.AddStore(ctx, CRSConnectionWrap(sID, "noop://external", capabilities.FullCapabilities()...))
|
||||
require.NoError(t, err)
|
||||
|
||||
err = crs.RemoveStore(ctx, 123)
|
||||
require.Error(t, err)
|
||||
require.Len(t, crs.stores, 1)
|
||||
})
|
||||
|
||||
t.Run("removing store", func(t *testing.T) {
|
||||
crs, err := ComposeRecordStore(
|
||||
ctx,
|
||||
CRSConnectionWrap(0, "noop://primary", capabilities.FullCapabilities()...),
|
||||
testDriver{},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
sID := nextID()
|
||||
err = crs.AddStore(ctx, CRSConnectionWrap(sID, "noop://external", capabilities.FullCapabilities()...))
|
||||
require.NoError(t, err)
|
||||
|
||||
err = crs.RemoveStore(ctx, sID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, crs.stores, 0)
|
||||
require.NotNil(t, crs.primary)
|
||||
})
|
||||
}
|
||||
@@ -8,6 +8,7 @@ type (
|
||||
Handle string
|
||||
DSN string
|
||||
Location string
|
||||
|
||||
// ...
|
||||
|
||||
// @todo IMO having it like so (instead of in a struct) allows for more
|
||||
|
||||
@@ -20,7 +20,6 @@ type (
|
||||
|
||||
Partitioned bool
|
||||
PartitionFormat string
|
||||
PartitionBase []any
|
||||
}
|
||||
|
||||
Module struct {
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
package crs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
ccrs "github.com/cortezaproject/corteza-server/compose/crs"
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/cortezaproject/corteza-server/pkg/data"
|
||||
"github.com/doug-martin/goqu/v9"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type (
|
||||
Connector struct {
|
||||
DB connection
|
||||
|
||||
// Logger for connection
|
||||
Logger *zap.Logger
|
||||
|
||||
Dialect goqu.DialectWrapper
|
||||
// ---
|
||||
|
||||
modelHandlers map[uint64]*crs
|
||||
capabilities capabilities.Set
|
||||
}
|
||||
)
|
||||
|
||||
func CRSConnector(db connection, logger *zap.Logger, di goqu.DialectWrapper, cc capabilities.Set) *Connector {
|
||||
return &Connector{
|
||||
DB: db,
|
||||
Logger: logger,
|
||||
Dialect: di,
|
||||
modelHandlers: make(map[uint64]*crs),
|
||||
capabilities: cc,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Connector) Capabilities() capabilities.Set {
|
||||
return c.capabilities
|
||||
}
|
||||
|
||||
func (c *Connector) Can(capabilities ...capabilities.Capability) bool {
|
||||
return c.Capabilities().IsSuperset(capabilities...)
|
||||
}
|
||||
|
||||
func (c *Connector) Close(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Connector) CreateRecords(ctx context.Context, m *data.Model, rr ...*types.Record) error {
|
||||
h, err := c.getModelHandlerE(m)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return h.Create(ctx, rr...)
|
||||
}
|
||||
|
||||
func (c *Connector) SearchRecords(ctx context.Context, sch *data.Model, filter any) (ccrs.Iterator, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// ---
|
||||
|
||||
func (c *Connector) RemoveModel(context.Context, *data.Model, ...*data.Model) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Connector) AlterModel(ctx context.Context, old *data.Model, new *data.Model) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Connector) AlterModelAttribute(ctx context.Context, sch *data.Model, old data.Attribute, new data.Attribute, trans ...ccrs.TransformationFunction) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Connector) Models(ctx context.Context) (data.ModelSet, error) {
|
||||
// @todo...
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (c *Connector) AddModel(ctx context.Context, m *data.Model, mm ...*data.Model) error {
|
||||
for _, model := range append(mm, m) {
|
||||
if h := c.getModelHandler(model); h != nil {
|
||||
return fmt.Errorf("cannot add model: already exists")
|
||||
}
|
||||
|
||||
_, err := c.addModelHandler(ctx, model)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---
|
||||
|
||||
func (c *Connector) getModelHandler(m *data.Model) *crs {
|
||||
return c.modelHandlers[m.ResourceID]
|
||||
}
|
||||
|
||||
func (c *Connector) getModelHandlerE(m *data.Model) (*crs, error) {
|
||||
h := c.modelHandlers[m.ResourceID]
|
||||
if h == nil {
|
||||
return nil, fmt.Errorf("cannot process model for resource %d: not supported", m.ResourceID)
|
||||
}
|
||||
return h, nil
|
||||
}
|
||||
|
||||
func (c *Connector) addModelHandler(ctx context.Context, m *data.Model) (*crs, error) {
|
||||
h := CRS(m, c.DB)
|
||||
c.modelHandlers[m.ResourceID] = h
|
||||
|
||||
return h, nil
|
||||
}
|
||||
@@ -6,10 +6,14 @@ import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs"
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/errors"
|
||||
"github.com/cortezaproject/corteza-server/pkg/logger"
|
||||
"github.com/cortezaproject/corteza-server/store"
|
||||
"github.com/cortezaproject/corteza-server/store/adapters/rdbms"
|
||||
rdbmsCRS "github.com/cortezaproject/corteza-server/store/adapters/rdbms/crs"
|
||||
"github.com/cortezaproject/corteza-server/store/adapters/rdbms/instrumentation"
|
||||
"github.com/doug-martin/goqu/v9"
|
||||
_ "github.com/doug-martin/goqu/v9/dialect/mysql"
|
||||
@@ -20,9 +24,35 @@ import (
|
||||
|
||||
func init() {
|
||||
store.Register(Connect, "mysql", "mysql+debug")
|
||||
crs.Register(ConnectCRS, "mysql", "mysql+debug")
|
||||
|
||||
sql.Register("mysql+debug", sqlmw.Driver(new(mysql.MySQLDriver), instrumentation.Debug()))
|
||||
}
|
||||
|
||||
// @todo cleanup; unify common bits and stuff
|
||||
|
||||
func ConnectCRS(ctx context.Context, dsn string, cc ...capabilities.Capability) (_ crs.StoreConnection, 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
|
||||
}
|
||||
|
||||
// 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 rdbmsCRS.CRSConnector(db, logger.Default(), goqu.Dialect("mysql"), cc), nil
|
||||
}
|
||||
|
||||
func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) {
|
||||
var (
|
||||
db *sqlx.DB
|
||||
|
||||
@@ -0,0 +1,185 @@
|
||||
package crs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs"
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/cortezaproject/corteza-server/pkg/id"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/crs/capabilities"
|
||||
_ "github.com/cortezaproject/corteza-server/store/adapters/rdbms/drivers/mysql"
|
||||
)
|
||||
|
||||
/*
|
||||
See here for my table definitions
|
||||
|
||||
CREATE TABLE `the_cake` (
|
||||
`id` bigint unsigned NOT NULL,
|
||||
`name` VARCHAR(45),
|
||||
`want` BOOLEAN,
|
||||
`ownedBy` bigint unsigned NOT NULL,
|
||||
`createdAt` datetime NOT NULL,
|
||||
`updatedAt` datetime DEFAULT NULL,
|
||||
`deletedAt` datetime DEFAULT NULL,
|
||||
`createdBy` bigint unsigned NOT NULL,
|
||||
`updatedBy` bigint unsigned NOT NULL DEFAULT '0',
|
||||
`deletedBy` bigint unsigned NOT NULL DEFAULT '0',
|
||||
PRIMARY KEY (`id`),
|
||||
KEY `compose_record_owner` (`ownedBy`)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb3;
|
||||
|
||||
CREATE TABLE `the_cookie` (
|
||||
`id` bigint unsigned NOT NULL,
|
||||
`name` VARCHAR(45),
|
||||
`is_good` BOOLEAN,
|
||||
`ownedBy` bigint unsigned NOT NULL,
|
||||
`createdAt` datetime NOT NULL,
|
||||
`updatedAt` datetime DEFAULT NULL,
|
||||
`deletedAt` datetime DEFAULT NULL,
|
||||
`createdBy` bigint unsigned NOT NULL,
|
||||
`updatedBy` bigint unsigned NOT NULL DEFAULT '0',
|
||||
`deletedBy` bigint unsigned NOT NULL DEFAULT '0',
|
||||
PRIMARY KEY (`id`),
|
||||
KEY `compose_record_owner` (`ownedBy`)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb3;
|
||||
*/
|
||||
|
||||
func TestHello(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
c, err := crs.ComposeRecordStore(
|
||||
ctx,
|
||||
nil,
|
||||
false,
|
||||
// Primary...
|
||||
crs.CRSConnectionWrap(0, "mysql://envoy:envoy@tcp(localhost:3306)/crs?collation=utf8mb4_general_ci", capabilities.FullCapabilities()...),
|
||||
|
||||
// Others...
|
||||
crs.CRSConnectionWrap(1, "mysql://envoy:envoy@tcp(localhost:3306)/crs?collation=utf8mb4_general_ci", capabilities.FullCapabilities()...),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
_ = c
|
||||
|
||||
// ---
|
||||
|
||||
cookieModule := &types.Module{
|
||||
ID: 10001,
|
||||
Handle: "cookie",
|
||||
Store: types.CRSDef{
|
||||
ComposeRecordStoreID: 0,
|
||||
Capabilities: capabilities.FullCapabilities(),
|
||||
Partitioned: true,
|
||||
PartitionFormat: "the_{{module}}",
|
||||
},
|
||||
Fields: types.ModuleFieldSet{&types.ModuleField{
|
||||
ModuleID: 10001,
|
||||
ID: 20001,
|
||||
Name: "name",
|
||||
Encoding: types.EncodingStrategy{
|
||||
EncodingStrategyAlias: &types.EncodingStrategyAlias{
|
||||
Ident: "name",
|
||||
},
|
||||
},
|
||||
}, &types.ModuleField{
|
||||
ModuleID: 10001,
|
||||
ID: 20002,
|
||||
Name: "is_good",
|
||||
Kind: "Bool",
|
||||
Encoding: types.EncodingStrategy{
|
||||
EncodingStrategyAlias: &types.EncodingStrategyAlias{
|
||||
Ident: "is_good",
|
||||
},
|
||||
},
|
||||
}},
|
||||
}
|
||||
|
||||
cakeModule := &types.Module{
|
||||
ID: 10002,
|
||||
Handle: "cake",
|
||||
Store: types.CRSDef{
|
||||
ComposeRecordStoreID: 1,
|
||||
Capabilities: capabilities.FullCapabilities(),
|
||||
Partitioned: true,
|
||||
PartitionFormat: "the_{{module}}",
|
||||
},
|
||||
Fields: types.ModuleFieldSet{&types.ModuleField{
|
||||
ModuleID: 10002,
|
||||
ID: 20003,
|
||||
Name: "name",
|
||||
Encoding: types.EncodingStrategy{
|
||||
EncodingStrategyAlias: &types.EncodingStrategyAlias{
|
||||
// this doesn't work
|
||||
Ident: "name",
|
||||
},
|
||||
},
|
||||
}, &types.ModuleField{
|
||||
ModuleID: 10002,
|
||||
ID: 20004,
|
||||
Name: "want",
|
||||
Kind: "Bool",
|
||||
Encoding: types.EncodingStrategy{
|
||||
EncodingStrategyAlias: &types.EncodingStrategyAlias{
|
||||
Ident: "want",
|
||||
},
|
||||
},
|
||||
}},
|
||||
}
|
||||
|
||||
err = c.ReloadModules(ctx, cookieModule, cakeModule)
|
||||
require.NoError(t, err)
|
||||
|
||||
// ---
|
||||
|
||||
a := time.Now()
|
||||
|
||||
cookieRecord := &types.Record{
|
||||
ID: id.Next(),
|
||||
ModuleID: 10001,
|
||||
Values: types.RecordValueSet{{
|
||||
Name: "name",
|
||||
Value: "SOME COOKIE HERE",
|
||||
}, {
|
||||
Name: "is_good",
|
||||
Value: "1",
|
||||
}},
|
||||
CreatedAt: time.Now(),
|
||||
UpdatedAt: &a,
|
||||
DeletedAt: &a,
|
||||
OwnedBy: 20003,
|
||||
CreatedBy: 20003,
|
||||
UpdatedBy: 20003,
|
||||
DeletedBy: 20003,
|
||||
}
|
||||
|
||||
cakeRecord := &types.Record{
|
||||
ID: id.Next(),
|
||||
ModuleID: 10002,
|
||||
Values: types.RecordValueSet{{
|
||||
Name: "name",
|
||||
Value: "DOME CAKE HERE",
|
||||
}, {
|
||||
Name: "want",
|
||||
Value: "1",
|
||||
}},
|
||||
CreatedAt: time.Now(),
|
||||
UpdatedAt: &a,
|
||||
DeletedAt: &a,
|
||||
OwnedBy: 20003,
|
||||
CreatedBy: 20003,
|
||||
UpdatedBy: 20003,
|
||||
DeletedBy: 20003,
|
||||
}
|
||||
|
||||
err = c.ComposeRecordCreate(ctx, cookieModule, cookieRecord)
|
||||
require.NoError(t, err)
|
||||
|
||||
err = c.ComposeRecordCreate(ctx, cakeModule, cakeRecord)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.FailNow(t, "the test dies a failNow so I can see logs!!!")
|
||||
}
|
||||
Reference in New Issue
Block a user