Refactor and improve DAL implementation and init

Changes:
 - Boot initialization follows standard impl
 - Improved DAL connection management (adding, reloading, removing)
 - Cleaner and more detailed logging
 - Primary store connection is now reused when added to DAL

app/boot_dal.go

@@ -0,0 +1,26 @@
+package app
+
+import (
+	"context"
+	"fmt"
+	"github.com/cortezaproject/corteza-server/pkg/dal"
+	"go.uber.org/zap"
+)
+
+func (app *CortezaApp) initDAL(ctx context.Context, log *zap.Logger) (err error) {
+	// Verify that primary store is connected
+	// or return error
+	if app.Store == nil {
+		return fmt.Errorf("primary store not connected")
+	}
+
+	// Init DAL and prepare default connection
+	svc, err := dal.New(log.Named("dal"), app.Opt.Environment.IsDevelopment())
+	if err != nil {
+		return err
+	}
+
+	dal.SetGlobal(svc)
+
+	return
+}

app/boot_levels.go

@@ -291,59 +291,6 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) {
 		return
 	}
 
-	var primaryDalConnection *types.DalConnection
-	primaryDalConnection, err = store.LookupDalConnectionByHandle(ctx, app.Store, "primary_connection")
-	if err != nil {
-		// @note we won't log the error here as this is a bigger issue then
-		// it not connecting
-		return
-	}
-
-	dalLogger := app.Log.Named("dal")
-
-	// Assure primary connection configs
-	{
-		if primaryDalConnection.Config.DefaultModelIdent != provision.DefaultComposeRecordTable {
-			dalLogger.Warn("overwriting DefaultModelIdent value", zap.String("old", primaryDalConnection.Config.DefaultModelIdent), zap.String("new", provision.DefaultComposeRecordTable))
-			primaryDalConnection.Config.DefaultModelIdent = provision.DefaultComposeRecordTable
-		}
-
-		if primaryDalConnection.Config.DefaultAttributeIdent != provision.DefaultComposeRecordValueCol {
-			dalLogger.Warn("overwriting DefaultAttributeIdent value", zap.String("old", primaryDalConnection.Config.DefaultAttributeIdent), zap.String("new", provision.DefaultComposeRecordValueCol))
-			primaryDalConnection.Config.DefaultAttributeIdent = provision.DefaultComposeRecordValueCol
-		}
-
-		if primaryDalConnection.Config.DefaultPartitionFormat != provision.DefaultPartitionFormat {
-			dalLogger.Warn("overwriting DefaultPartitionFormat value", zap.String("old", primaryDalConnection.Config.DefaultPartitionFormat), zap.String("new", provision.DefaultPartitionFormat))
-			primaryDalConnection.Config.DefaultPartitionFormat = provision.DefaultPartitionFormat
-		}
-
-		c := dal.NewDSNConnection(app.Opt.DB.DSN)
-		dalLogger.Warn("overwriting Connection value", zap.Any("old", primaryDalConnection.Config.Connection), zap.Any("new", c))
-		primaryDalConnection.Config.Connection = c
-	}
-
-	// Init DAL and prepare default connection
-	if _, err = dal.InitGlobalService(
-		ctx,
-		dalLogger,
-		app.Opt.Environment.IsDevelopment(),
-		primaryDalConnection.ID,
-
-		// DB_DSN is the default connection with full capabilities
-		primaryDalConnection.Config.Connection,
-		dal.ConnectionMeta{
-			DefaultModelIdent:      primaryDalConnection.Config.DefaultModelIdent,
-			DefaultAttributeIdent:  primaryDalConnection.Config.DefaultAttributeIdent,
-			DefaultPartitionFormat: primaryDalConnection.Config.DefaultPartitionFormat,
-			// @todo make it configurable from env
-			SensitivityLevel: primaryDalConnection.SensitivityLevel,
-			Label:            primaryDalConnection.Handle,
-		},
-		primaryDalConnection.ActiveCapabilities()...); err != nil {
-		return err
-	}
-
 	if app.Opt.Auth.DefaultClient != "" {
 		// default client will help streamline authorization with default clients
 		app.DefaultAuthClient, err = store.LookupAuthClientByHandle(ctx, app.Store, app.Opt.Auth.DefaultClient)
@@ -460,11 +407,18 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) {
 		return fmt.Errorf("could not reload resource translations: %w", err)
 	}
 
+	{
+		// Initialize Data Access Layer (DAL)
+		if err = app.initDAL(ctx, app.Log); err != nil {
+			return fmt.Errorf("can not initialize DAL: %w", err)
+		}
+	}
+
 	// Initializes system services
 	//
 	// Note: this is a legacy approach, all services from all 3 apps
 	// will most likely be merged in the future
-	err = sysService.Initialize(ctx, app.Log, app.Store, primaryDalConnection, app.WsServer, sysService.Config{
+	err = sysService.Initialize(ctx, app.Log, app.Store, app.WsServer, sysService.Config{
 		ActionLog: app.Opt.ActionLog,
 		Discovery: app.Opt.Discovery,
 		Storage:   app.Opt.ObjStore,
@@ -479,20 +433,6 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) {
 		return
 	}
 
-	// Initializing DAL components
-	// - Sensitivity levels
-	err = sysService.DefaultDalSensitivityLevel.ReloadSensitivityLevels(ctx, sysService.DefaultStore)
-	if err != nil {
-		// @note we're not erroring out here so Corteza can still be used
-		dalLogger.Error("failed to initialize DAL sensitivity levels", zap.Error(err))
-	}
-	// - DAL connections
-	err = sysService.DefaultDalConnection.ReloadConnections(ctx)
-	if err != nil {
-		// @note we're not erroring out here so Corteza can still be used
-		dalLogger.Error("failed to initialize DAL connections", zap.Error(err))
-	}
-
 	if app.Opt.Messagebus.Enabled {
 		// initialize all the queue handlers
 		messagebus.Service().Init(ctx, service.DefaultQueue)
@@ -527,14 +467,6 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) {
 		return fmt.Errorf("could not initialize compose services: %w", err)
 	}
 
-	// Initializing DAL components
-	// - Models
-	err = cmpService.DefaultModule.ReloadDALModels(ctx)
-	if err != nil {
-		// @note we're not erroring out here so Corteza can still be used
-		dalLogger.Error("failed to initialize DAL models", zap.Error(err))
-	}
-
 	corredor.Service().SetUserFinder(sysService.DefaultUser)
 	corredor.Service().SetRoleFinder(sysService.DefaultRole)
 

codegen/assets/templates/gocode/store/interfaces.go.tpl

@@ -32,6 +32,9 @@ type (
 		// and it's highly unlikely we'll support different/multiple logging "backend"
 		SetLogger(*zap.Logger)
 
+		// Returns underlying store as DAL connection
+		ToDalConn() dal.Connection
+
 		// Tx is a transaction handler
 		Tx(context.Context, func(context.Context, Storer) error) error
 

compose/rest/data_privacy.go

@@ -6,7 +6,6 @@ import (
 	"github.com/cortezaproject/corteza-server/compose/rest/request"
 	"github.com/cortezaproject/corteza-server/compose/service"
 	"github.com/cortezaproject/corteza-server/compose/types"
-	"github.com/cortezaproject/corteza-server/pkg/dal"
 	"github.com/cortezaproject/corteza-server/pkg/payload"
 )
 
@@ -51,14 +50,9 @@ func (DataPrivacy) New() *DataPrivacy {
 func (ctrl *DataPrivacy) SensitiveDataList(ctx context.Context, r *request.DataPrivacySensitiveDataList) (out interface{}, err error) {
 	outSet := sensitiveDataSetPayload{}
 
-	primaryConnID := dal.Service().PrimaryConnectionID(ctx)
-
 	reqConns := make(map[uint64]bool)
 	hasReqConns := len(r.ConnectionID) > 0
 	for _, connectionID := range payload.ParseUint64s(r.ConnectionID) {
-		if connectionID == 0 {
-			connectionID = primaryConnID
-		}
 		reqConns[connectionID] = true
 	}
 
@@ -78,18 +72,10 @@ func (ctrl *DataPrivacy) SensitiveDataList(ctx context.Context, r *request.DataP
 		}
 		for _, m := range modules {
 			conn := m.ModelConfig.ConnectionID
-			if conn == 0 {
-				conn = primaryConnID
-			}
 			if hasReqConns && !reqConns[conn] {
 				continue
 			}
 
-			// @todo ignoring this for now; revert after dev release
-			// if m.Privacy.SensitivityLevel == 0 {
-			// 	continue
-			// }
-
 			sData, err := ctrl.record.FindSensitive(ctx, types.RecordFilter{ModuleID: m.ID, NamespaceID: m.NamespaceID})
 			if err != nil {
 				return nil, err

compose/service/module.go

@@ -82,8 +82,8 @@ var (
 	})
 )
 
-func Module(ctx context.Context, dal dalService) (*module, error) {
-	svc := &module{
+func Module(dal dalService) *module {
+	return &module{
 		ac:        DefaultAccessControl,
 		eventbus:  eventbus.Service(),
 		actionlog: DefaultActionlog,
@@ -91,8 +91,6 @@ func Module(ctx context.Context, dal dalService) (*module, error) {
 		locale:    DefaultResourceTranslation,
 		dal:       dal,
 	}
-
-	return svc, svc.ReloadDALModels(ctx)
 }
 
 func (svc module) Find(ctx context.Context, filter types.ModuleFilter) (set types.ModuleSet, f types.ModuleFilter, err error) {

compose/service/service.go

@@ -169,9 +169,7 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, c Config)
 	}
 
 	DefaultNamespace = Namespace()
-	if DefaultModule, err = Module(ctx, dal.Service()); err != nil {
-		return
-	}
+	DefaultModule = Module(dal.Service())
 
 	DefaultImportSession = ImportSession()
 	DefaultRecord = Record(dal.Service())
@@ -217,6 +215,11 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, c Config)
 }
 
 func Activate(ctx context.Context) (err error) {
+	err = DefaultModule.ReloadDALModels(ctx)
+	if err != nil {
+		return err
+	}
+
 	return
 }
 

pkg/dal/driver.go

@@ -175,7 +175,7 @@ func RegisterDriver(d Driver) {
 // connect opens a new StoreConnection for the given CRS
 func connect(ctx context.Context, log *zap.Logger, isDevelopment bool, cp ConnectionParams, capabilities ...capabilities.Capability) (Connection, error) {
 	if cp.Type != "corteza::dal:connection:dsn" {
-		return nil, fmt.Errorf("cannot open connection: only DSN connections supported")
+		return nil, fmt.Errorf("cannot open connection: only DSN connections supported (got: %q)", cp.Type)
 	}
 	dsn := cp.Params["dsn"].(string)
 

pkg/dal/service.go

@@ -3,7 +3,6 @@ package dal
 import (
 	"context"
 	"fmt"
-
 	"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
 	"github.com/cortezaproject/corteza-server/pkg/expr"
 	"github.com/cortezaproject/corteza-server/pkg/filter"
@@ -11,30 +10,49 @@ import (
 )
 
 type (
-	connectionWrap struct {
-		connectionID     uint64
-		label            string
+	ConnectionWrap struct {
+		connectionID uint64
+
+		// @todo remove it and use value from connection meta
+		label string
+
+		// @todo remove it and use value from connection meta
 		sensitivityLevel uint64
 
 		connection Connection
-		meta       ConnectionMeta
+
+		params       ConnectionParams
+		meta         ConnectionMeta
+		capabilities capabilities.Set
 	}
 
 	ConnectionMeta struct {
 		SensitivityLevel uint64
 		Label            string
 
-		DefaultModelIdent     string
+		// When model does not specifiy the ident (table name for example), fallback to this
+		// @todo we can lose "Default" prefix
+		// @todo do we need a separate setting or can we get away with using just PartitionFormat
+		DefaultModelIdent string
+
+		// If model attribute(s) do not specify
+		// @todo needs to be more explicit that this is for  JSON encode attributes
+		// @todo we can lose "Default" prefix
 		DefaultAttributeIdent string
 
+		// If data is partitioned we fallback to this,
+		// @todo we can lose "Default" prefix
 		DefaultPartitionFormat string
 
 		PartitionValidator string
 	}
 
 	service struct {
-		connections         map[uint64]*connectionWrap
-		primaryConnectionID uint64
+		connections map[uint64]*ConnectionWrap
+
+		// Default connection ID
+		// Can not be changed in the runtime, only set to value different than zero!
+		defConnID uint64
 
 		// Indexed by corresponding storeID
 		models map[uint64]ModelSet
@@ -49,36 +67,21 @@ type (
 	}
 )
 
-const (
-	DefaultConnectionID uint64 = 0
-)
-
 var (
 	gSvc *service
 )
 
-// InitGlobalService initializes a fresh DAL where the given primary connection
-func InitGlobalService(ctx context.Context, log *zap.Logger, inDev bool, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (_ *service, err error) {
-	log.Debug("initializing DAL service with primary connection", zap.Any("connection params", cp))
-
-	// To help prevent awkward issues due to globally shared resources
-	if gSvc != nil {
-		panic("cannot initialize global DAL service: already initialized")
-	}
-
-	if gSvc == nil {
-		gSvc, err = New(ctx, log, inDev, connectionID, cp, cm, capabilities...)
-		return gSvc, err
-	}
-
-	return gSvc, nil
+func SetGlobal(svc *service) {
+	gSvc = svc
 }
 
-func New(ctx context.Context, log *zap.Logger, inDev bool, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (*service, error) {
+// New creates a DAL service with the primary connection
+//
+// It needs an established and working connection to the primary store
+func New(log *zap.Logger, inDev bool) (*service, error) {
 	svc := &service{
-		connections:         make(map[uint64]*connectionWrap),
-		models:              make(map[uint64]ModelSet),
-		primaryConnectionID: connectionID,
+		connections: make(map[uint64]*ConnectionWrap),
+		models:      make(map[uint64]ModelSet),
 
 		logger: log,
 		inDev:  inDev,
@@ -87,26 +90,12 @@ func New(ctx context.Context, log *zap.Logger, inDev bool, connectionID uint64,
 		modelIssues:      make(dalIssueIndex),
 	}
 
-	var err error
-	cw := &connectionWrap{
-		meta:             cm,
-		sensitivityLevel: cm.SensitivityLevel,
-		label:            cm.Label,
-		connectionID:     connectionID,
-	}
-	cw.connection, err = connect(ctx, log, inDev, cp, capabilities...)
-	if err != nil {
-		return nil, err
-	}
-
-	svc.connections[connectionID] = cw
-
 	return svc, nil
 }
 
 // Service returns the global initialized DAL service
 //
-// If InitGlobalService has not yet been called the function will panic
+// Function will panic if DAL service is not set (via SetGlobal)
 func Service() *service {
 	if gSvc == nil {
 		panic("DAL global service not initialized: call dal.InitGlobalService() first")
@@ -194,52 +183,132 @@ func (svc *service) DeleteSensitivityLevel(levels ...SensitivityLevel) (err erro
 // // // // // // // // // // // // // // // // // // // // // // // // //
 // Connection management
 
-// CreateConnection adds a new connection to the DAL
-func (svc *service) CreateConnection(ctx context.Context, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (err error) {
-	svc.logger.Debug("creating connection", zap.Uint64("connectionID", connectionID), zap.Any("connection params", cp))
+// MakeConnection makes and returns a new connection (wrap)
+func MakeConnection(ID uint64, conn Connection, p ConnectionParams, m ConnectionMeta, cap ...capabilities.Capability) *ConnectionWrap {
+	return &ConnectionWrap{
+		connectionID: ID,
+		connection:   conn,
 
+		params: p,
+
+		meta:             m,
+		sensitivityLevel: m.SensitivityLevel,
+		label:            m.Label,
+
+		capabilities: cap,
+	}
+}
+
+// ReplaceConnection adds new or updates an existing connection
+//
+// We rely on the user to provide stable connection IDs and
+// uses valid relations to these connections in the models.
+//
+// Is isDefault when adding a default connection. Service will then
+// compensate and use proper IDs when models refer to connection with ID=0
+func (svc *service) ReplaceConnection(ctx context.Context, cw *ConnectionWrap, isDefault bool) (err error) {
+	// @todo lock/unlock
 	var (
-		issues = newIssueHelper().addConnection(connectionID)
+		ID      = cw.connectionID
+		issues  = newIssueHelper().addConnection(ID)
+		oldConn *ConnectionWrap
+
+		log = svc.logger.Named("connection").With(
+			zap.Uint64("ID", ID),
+			zap.Any("params", cw.params),
+			zap.Any("meta", cw.meta),
+		)
 	)
+
+	if isDefault {
+		if svc.defConnID == 0 {
+			// default connection not set yet
+			log.Debug("setting as default connection")
+			svc.defConnID = ID
+		} else if svc.defConnID != ID {
+			// default connection set but ID is different.
+			// this does not make any sense
+			return fmt.Errorf("different ID for default connection detecte (old: %d, new: %d)", svc.defConnID, ID)
+		}
+	}
+
 	defer svc.updateIssues(issues)
 
-	// sensitivity levels
-	if !svc.sensitivityLevels.includes(cm.SensitivityLevel) {
-		issues.addConnectionIssue(connectionID, errConnectionCreateMissingSensitivityLevel(connectionID, cm.SensitivityLevel))
+	// Sensitivity level validations
+	if !svc.sensitivityLevels.includes(cw.meta.SensitivityLevel) {
+		issues.addConnectionIssue(ID, errConnectionCreateMissingSensitivityLevel(ID, cw.meta.SensitivityLevel))
 	}
 
-	// Prepare connection bits
-	cw := &connectionWrap{
-		connectionID:     connectionID,
-		meta:             cm,
-		sensitivityLevel: cm.SensitivityLevel,
-		label:            cm.Label,
+	if oldConn = svc.getConnectionByID(ID); oldConn != nil {
+		// Connection exists, validate models and sensitivity levels and close and remove connection at the end
+		log.Debug("found existing")
+
+		// Check already registered models and their capabilities
+		//
+		// Defer the return till the end so we can get a nicer report of what all is wrong
+		errored := false
+		for _, model := range svc.models[ID] {
+			log.Debug("validating model before connection is updated", zap.String("ident", model.Ident))
+
+			// - capabilities
+			if !model.Capabilities.IsSubset(cw.capabilities...) {
+				issues.addConnectionIssue(ID, fmt.Errorf("cannot update connection %d: new connection does not support existing models", ID))
+				errored = true
+			}
+
+			// - sensitivity levels
+			if !svc.sensitivityLevels.isSubset(model.SensitivityLevel, cw.meta.SensitivityLevel) {
+				issues.addConnectionIssue(ID, fmt.Errorf("cannot update connection %d: new connection sensitivity level does not support model %d", ID, model.ResourceID))
+				errored = true
+			}
+		}
+
+		// Don't update if meta bits are not ok
+		if errored {
+			log.Warn("update failed")
+			return
+		}
+
+		// close old connection
+		if cc, ok := oldConn.connection.(ConnectionCloser); ok {
+			if err = cc.Close(ctx); err != nil {
+				issues.addConnectionIssue(ID, err)
+				return nil
+			}
+
+			log.Debug("disconnected")
+		}
+
+		svc.removeConnection(ID)
 	}
-	if cw.connection, err = connect(ctx, svc.logger, svc.inDev, cp, capabilities...); err != nil {
-		issues.addConnectionIssue(connectionID, errConnectionCreateConnectionFailed(connectionID, err))
+
+	if cw.connection == nil {
+		cw.connection, err = connect(ctx, svc.logger, svc.inDev, cw.params, cw.capabilities...)
+		if err != nil {
+			log.Warn("could not connect", zap.Error(err))
+			issues.addConnectionIssue(ID, err)
+		} else {
+			log.Debug("connected")
+		}
+	} else {
+		log.Debug("using preexisting connection")
+
 	}
 
 	svc.addConnection(cw)
-
-	svc.logger.Debug("created connection")
+	log.Debug("added")
 	return nil
 }
 
-func (svc *service) PrimaryConnectionID(ctx context.Context) uint64 {
-	return svc.primaryConnectionID
-}
-
 // DeleteConnection removes the given connection from the DAL
-func (svc *service) DeleteConnection(ctx context.Context, connectionID uint64) (err error) {
-	svc.logger.Debug("deleting connection", zap.Uint64("connectionID", connectionID))
-
+func (svc *service) RemoveConnection(ctx context.Context, ID uint64) (err error) {
 	var (
-		issues = newIssueHelper().addConnection(connectionID)
+		issues = newIssueHelper().addConnection(ID)
 	)
 
-	c := svc.getConnectionByID(connectionID)
+	c := svc.getConnectionByID(ID)
 	if c == nil {
-		return errConnectionDeleteNotFound(connectionID)
+		return errConnectionDeleteNotFound(ID)
 	}
 
 	// Potential cleanups
@@ -253,90 +322,15 @@ func (svc *service) DeleteConnection(ctx context.Context, connectionID uint64) (
 	//
 	// @todo this is temporary until a proper update function is prepared.
 	// The primary connection must not be removable!
-	svc.removeConnection(connectionID)
+	svc.removeConnection(ID)
 
 	// Only if successful should we cleanup the issue registry
 	svc.updateIssues(issues)
 
-	svc.logger.Debug("deleted connection")
-
-	return nil
-}
-
-// UpdateConnection updates the given connection
-func (svc *service) UpdateConnection(ctx context.Context, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (err error) {
-	svc.logger.Debug("updating connection", zap.Uint64("connectionID", connectionID))
-
-	var (
-		issues  = newIssueHelper().addConnection(connectionID)
-		oldConn *connectionWrap
+	svc.logger.Named("connection").Debug("deleted",
+		zap.Uint64("ID", ID),
+		zap.Any("meta", c.meta),
 	)
-	defer svc.updateIssues(issues)
-
-	// Validation
-	{
-		// Check if connection exists
-		oldConn = svc.getConnectionByID(connectionID)
-		if oldConn == nil {
-			issues.addConnectionIssue(connectionID, errConnectionUpdateNotFound(connectionID))
-		}
-
-		// sensitivity levels
-		if !svc.sensitivityLevels.includes(cm.SensitivityLevel) {
-			issues.addConnectionIssue(connectionID, errConnectionUpdateMissingSensitivityLevel(connectionID, cm.SensitivityLevel))
-		}
-
-		// Check already registered models and their capabilities
-		//
-		// Defer the return till the end so we can get a nicer report of what all is wrong
-		errored := false
-		for _, model := range svc.models[connectionID] {
-			// - capabilities
-			if !model.Capabilities.IsSubset(capabilities...) {
-				issues.addConnectionIssue(connectionID, fmt.Errorf("cannot update connection %d: new connection does not support existing models", connectionID))
-				errored = errored || true
-			}
-			// - sensitivity levels
-			if !svc.sensitivityLevels.isSubset(model.SensitivityLevel, cm.SensitivityLevel) {
-				issues.addConnectionIssue(connectionID, fmt.Errorf("cannot update connection %d: new connection sensitivity level does not support model %d", connectionID, model.ResourceID))
-				errored = errored || true
-			}
-		}
-
-		// Don't update if meta bits are not ok
-		if errored {
-			return
-		}
-	}
-
-	// close old connection
-	{
-		if cc, ok := oldConn.connection.(ConnectionCloser); ok {
-			if err = cc.Close(ctx); err != nil {
-				issues.addConnectionIssue(connectionID, err)
-				return nil
-			}
-		}
-		svc.removeConnection(connectionID)
-	}
-
-	// open new connection
-	{
-		newConnection, err := connect(ctx, svc.logger, svc.inDev, cp, capabilities...)
-		if err != nil {
-			issues.addConnectionIssue(connectionID, err)
-		}
-
-		svc.addConnection(&connectionWrap{
-			meta:             cm,
-			sensitivityLevel: cm.SensitivityLevel,
-			label:            cm.Label,
-			connectionID:     connectionID,
-			connection:       newConnection,
-		})
-	}
-
-	svc.logger.Debug("updated connection")
 
 	return nil
 }
@@ -436,7 +430,7 @@ func (svc *service) Truncate(ctx context.Context, mf ModelFilter, capabilities c
 	return cw.connection.Truncate(ctx, model)
 }
 
-func (svc *service) storeOpPrep(ctx context.Context, mf ModelFilter, capabilities capabilities.Set) (model *Model, cw *connectionWrap, err error) {
+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 = errModelNotFound(mf.ResourceID)
@@ -485,19 +479,25 @@ func (svc *service) SearchModels(ctx context.Context) (out ModelSet, err error)
 
 // AddModel adds support for a new model
 func (svc *service) CreateModel(ctx context.Context, models ...*Model) (err error) {
-	svc.logger.Debug("creating models", zap.Int("count", len(models)))
-
 	var (
+		log       = svc.logger.Named("models")
 		issues    = newIssueHelper()
 		auxIssues = newIssueHelper()
 	)
+
+	svc.logger.Debug("creating", zap.Int("count", len(models)))
+
 	defer svc.updateIssues(issues)
 
 	// Validate models
 	for _, model := range models {
-		svc.logger.Debug("validating model", zap.Uint64("ID", model.ResourceID))
 		issues.addModel(model.ResourceID)
 
+		if model.ConnectionID == 0 {
+			// Replace model's connection ID with default one when zero
+			model.ConnectionID = svc.defConnID
+		}
+
 		// Assure the connection has no issues
 		if svc.hasConnectionIssues(model.ConnectionID) {
 			issues.addModelIssue(model.ConnectionID, model.ResourceID, errModelCreateProblematicConnection(model.ConnectionID, model.ResourceID))
@@ -536,19 +536,28 @@ func (svc *service) CreateModel(ctx context.Context, models ...*Model) (err erro
 			}
 		}
 
-		svc.logger.Debug("validated model")
+		log.Debug("validated",
+			zap.Uint64("ID", model.ResourceID),
+			zap.String("ident", model.Ident),
+			zap.String("label", model.Label),
+		)
 	}
 
 	// Add models to corresponding connections
 	for connection, models := range svc.modelByConnection(models) {
 		for _, model := range models {
-			svc.logger.Debug("adding model", zap.Uint64("model", model.ResourceID))
+			mLog := log.With(
+				zap.Uint64("connectionID", model.ConnectionID),
+				zap.Uint64("ID", model.ResourceID),
+				zap.String("ident", model.Ident),
+				zap.String("label", model.Label),
+			)
 
 			connectionIssues := svc.hasConnectionIssues(model.ConnectionID)
 			modelIssues := svc.hasModelIssues(model.ConnectionID, model.ResourceID)
 
 			if !modelIssues && !connectionIssues {
-				svc.logger.Debug("adding model to connection", zap.Uint64("connection", model.ConnectionID), zap.Uint64("model", model.ResourceID))
+				mLog.Debug("adding model to connection")
 
 				// Add model to connection
 				auxIssues, err = svc.registerModelToConnection(ctx, connection, model)
@@ -558,10 +567,10 @@ func (svc *service) CreateModel(ctx context.Context, models ...*Model) (err erro
 				}
 			} else {
 				if connectionIssues {
-					svc.logger.Warn("not adding to connection due to connection issues", zap.Uint64("connection", model.ConnectionID))
+					mLog.Warn("not adding to connection due to connection issues")
 				}
 				if modelIssues {
-					svc.logger.Warn("not adding to connection due to model issues", zap.Uint64("model", model.ResourceID))
+					mLog.Warn("not adding to connection due to model issues")
 				}
 			}
 
@@ -570,23 +579,31 @@ func (svc *service) CreateModel(ctx context.Context, models ...*Model) (err erro
 		}
 	}
 
-	svc.logger.Debug("created models")
+	log.Debug("done")
 
 	return
 }
 
 // DeleteModel removes support for the model and deletes it from the connection
 func (svc *service) DeleteModel(ctx context.Context, models ...*Model) (err error) {
-	svc.logger.Debug("deleting models", zap.Int("count", len(models)))
 
 	var (
+		log    = svc.logger.Named("models")
 		issues = newIssueHelper()
 	)
+
+	log.Debug("deleting", zap.Int("count", len(models)))
+
 	defer svc.updateIssues(issues)
 
 	// validation
 	skip := make(map[uint64]bool)
 	for _, model := range models {
+		if model.ConnectionID == 0 {
+			// Replace model's connection ID with default one when zero
+			model.ConnectionID = svc.defConnID
+		}
+
 		issues.addModel(model.ResourceID)
 
 		// Validate existence
@@ -608,10 +625,15 @@ func (svc *service) DeleteModel(ctx context.Context, models ...*Model) (err erro
 
 	// Work
 	for _, model := range models {
-		svc.logger.Debug("deleting model", zap.Uint64("model", model.ResourceID))
+		mLog := log.With(
+			zap.Uint64("connectionID", model.ConnectionID),
+			zap.Uint64("ID", model.ResourceID),
+			zap.String("ident", model.Ident),
+			zap.String("label", model.Label),
+		)
 
 		if skip[model.ResourceID] {
-			svc.logger.Debug("model does not exist; skipping")
+			mLog.Debug("model does not exist; skipping")
 			continue
 		}
 
@@ -629,20 +651,39 @@ func (svc *service) DeleteModel(ctx context.Context, models ...*Model) (err erro
 		// how should this be handled; a straight up delete doesn't sound sane to me
 		// anymore
 
-		svc.logger.Debug("deleted model")
+		mLog.Debug("deleted")
 	}
 
 	return nil
 }
 
 func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) (err error) {
-	svc.logger.Debug("updating model", zap.Uint64("model", old.ResourceID))
+	if old.ConnectionID == 0 {
+		// Replace old model's connection ID with default one when zero
+		old.ConnectionID = svc.defConnID
+	}
+
+	if new.ConnectionID == 0 {
+		// Replace new model's connection ID with default one when zero
+		new.ConnectionID = svc.defConnID
+	}
 
 	var (
-		conn *connectionWrap
+		log = svc.logger.Named("models").With(
+			zap.Uint64("connection", new.ConnectionID),
+			zap.Uint64("model", new.ResourceID),
+			zap.Uint64("ID", new.ResourceID),
+			zap.String("ident", new.Ident),
+			zap.String("label", new.Label),
+		)
+
+		conn *ConnectionWrap
 
 		issues = newIssueHelper().addModel(old.ResourceID)
 	)
+
+	log.Debug("updating", zap.Uint64("model", old.ResourceID))
+
 	defer svc.updateIssues(issues)
 
 	// Validation
@@ -693,7 +734,7 @@ func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) (er
 	modelIssues := svc.hasModelIssues(new.ConnectionID, new.ResourceID)
 
 	if !modelIssues && !connectionIssues {
-		svc.logger.Debug("updating connection's model", zap.Uint64("connection", new.ConnectionID), zap.Uint64("model", new.ResourceID))
+		log.Debug("updating connection's model")
 
 		err = conn.connection.UpdateModel(ctx, old, new)
 		if err != nil {
@@ -701,10 +742,10 @@ func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) (er
 		}
 	} else {
 		if connectionIssues {
-			svc.logger.Warn("not updating connection's model due to connection issues", zap.Uint64("connection", new.ConnectionID))
+			log.Warn("not updating connection's model due to connection issues")
 		}
 		if modelIssues {
-			svc.logger.Warn("not updating connection's model due to model issues", zap.Uint64("model", new.ResourceID))
+			log.Warn("not updating connection's model due to model issues")
 		}
 	}
 
@@ -721,7 +762,7 @@ func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) (er
 		svc.models[old.ConnectionID] = append(svc.models[old.ConnectionID], new)
 	}
 
-	svc.logger.Debug("updated model")
+	log.Debug("updated")
 
 	return
 }
@@ -730,7 +771,7 @@ func (svc *service) UpdateModelAttribute(ctx context.Context, model *Model, old,
 	svc.logger.Debug("updating model attribute", zap.Uint64("model", model.ResourceID))
 
 	var (
-		conn   *connectionWrap
+		conn   *ConnectionWrap
 		issues = newIssueHelper().addModel(model.ResourceID)
 	)
 	defer svc.updateIssues(issues)
@@ -814,6 +855,10 @@ func (svc *service) UpdateModelAttribute(ctx context.Context, model *Model, old,
 }
 
 func (svc *service) ModelIdentFormatter(connectionID uint64) (f *IdentFormatter, err error) {
+	if connectionID == 0 {
+		connectionID = svc.defConnID
+	}
+
 	c := svc.getConnectionByID(connectionID)
 
 	if c == nil {
@@ -852,15 +897,19 @@ func (svc *service) FindModelByIdent(connectionID uint64, ident string) *Model {
 // // // // // // // // // // // // // // // // // // // // // // // // //
 // Utilities
 
-func (svc *service) getConnectionByID(connectionID uint64) (cw *connectionWrap) {
-	if connectionID == DefaultConnectionID || connectionID == svc.primaryConnectionID {
-		return svc.connections[svc.primaryConnectionID]
-	}
+func (svc *service) removeConnection(connectionID uint64) {
+	delete(svc.connections, connectionID)
+}
 
+func (svc *service) addConnection(cw *ConnectionWrap) {
+	svc.connections[cw.connectionID] = cw
+}
+
+func (svc *service) getConnectionByID(connectionID uint64) (cw *ConnectionWrap) {
 	return svc.connections[connectionID]
 }
 
-func (svc *service) getConnection(connectionID uint64, cc ...capabilities.Capability) (cw *connectionWrap, can capabilities.Set, err error) {
+func (svc *service) getConnection(connectionID uint64, cc ...capabilities.Capability) (cw *ConnectionWrap, can capabilities.Set, err error) {
 	err = func() error {
 		// get the requested connection
 		cw = svc.getConnectionByID(connectionID)
@@ -885,8 +934,8 @@ func (svc *service) getConnection(connectionID uint64, cc ...capabilities.Capabi
 }
 
 // modelByConnection maps the given models by their CRS
-func (svc *service) modelByConnection(models ModelSet) (out map[*connectionWrap]ModelSet) {
-	out = make(map[*connectionWrap]ModelSet)
+func (svc *service) modelByConnection(models ModelSet) (out map[*ConnectionWrap]ModelSet) {
+	out = make(map[*ConnectionWrap]ModelSet)
 
 	for _, model := range models {
 		c := svc.getConnectionByID(model.ConnectionID)
@@ -896,7 +945,7 @@ func (svc *service) modelByConnection(models ModelSet) (out map[*connectionWrap]
 	return
 }
 
-func (svc *service) registerModelToConnection(ctx context.Context, cw *connectionWrap, model *Model) (issues *issueHelper, err error) {
+func (svc *service) registerModelToConnection(ctx context.Context, cw *ConnectionWrap, model *Model) (issues *issueHelper, err error) {
 	issues = newIssueHelper()
 
 	available, err := cw.connection.Models(ctx)
@@ -928,6 +977,10 @@ func (svc *service) registerModelToConnection(ctx context.Context, cw *connectio
 }
 
 func (svc *service) getModelByFilter(mf ModelFilter) *Model {
+	if mf.ConnectionID == 0 {
+		mf.ConnectionID = svc.defConnID
+	}
+
 	if mf.ResourceID > 0 {
 		return svc.FindModelByResourceID(mf.ConnectionID, mf.ResourceID)
 	}
@@ -995,7 +1048,7 @@ func (svc *service) newSensitivityLevelIndex(levels SensitivityLevelSet) (out se
 
 func (svc *service) validateNewSensitivityLevels(levels sensitivityLevelIndex) (err error) {
 	err = func() (err error) {
-		cIndex := make(map[uint64]*connectionWrap)
+		cIndex := make(map[uint64]*ConnectionWrap)
 
 		// - connections
 		for _, _c := range svc.connections {
@@ -1036,18 +1089,6 @@ func (svc *service) validateNewSensitivityLevels(levels sensitivityLevelIndex) (
 	return
 }
 
-func (svc *service) removeConnection(connectionID uint64) {
-	if connectionID == DefaultConnectionID || connectionID == svc.primaryConnectionID {
-		connectionID = svc.primaryConnectionID
-	}
-
-	delete(svc.connections, connectionID)
-}
-
-func (svc *service) addConnection(cw *connectionWrap) {
-	svc.connections[cw.connectionID] = cw
-}
-
 func wrapError(pfx string, err error) error {
 	return fmt.Errorf("%s: %v", pfx, err)
 }

pkg/provision/dal.go

@@ -0,0 +1,52 @@
+package provision
+
+import (
+	"context"
+	"github.com/cortezaproject/corteza-server/pkg/auth"
+	"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
+	"github.com/cortezaproject/corteza-server/pkg/id"
+	"github.com/cortezaproject/corteza-server/store"
+	"github.com/cortezaproject/corteza-server/system/types"
+)
+
+const (
+	DefaultComposeRecordTable    = "compose_record"
+	DefaultComposeRecordValueCol = "values"
+	DefaultPartitionFormat       = "compose_record"
+)
+
+// Injects primary connection
+func defaultDalConnection(ctx context.Context, s store.DalConnections) (err error) {
+	cc, err := store.LookupDalConnectionByHandle(ctx, s, types.DalPrimaryConnectionHandle)
+	if err != nil && err != store.ErrNotFound {
+		return
+	}
+
+	// Already exists
+	if cc != nil {
+		return
+	}
+
+	// Create it
+	var conn = &types.DalConnection{
+		// Using id.Next since we dropped "special" ids a while ago.
+		// If needed, use the handle
+		ID:     id.Next(),
+		Name:   "Primary Database",
+		Handle: types.DalPrimaryConnectionHandle,
+		Type:   types.DalPrimaryConnectionResourceType,
+
+		Config: types.ConnectionConfig{
+			DefaultModelIdent:      DefaultComposeRecordTable,
+			DefaultAttributeIdent:  DefaultComposeRecordValueCol,
+			DefaultPartitionFormat: DefaultPartitionFormat,
+		},
+		Capabilities: types.ConnectionCapabilities{
+			Supported: capabilities.FullCapabilities(),
+		},
+		CreatedAt: *now(),
+		CreatedBy: auth.ServiceUser().ID,
+	}
+
+	return store.CreateDalConnection(ctx, s, conn)
+}

pkg/provision/provision.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/cortezaproject/corteza-server/pkg/auth"
-	"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
 	"github.com/cortezaproject/corteza-server/pkg/errors"
 	"github.com/cortezaproject/corteza-server/pkg/id"
 	"github.com/cortezaproject/corteza-server/pkg/options"
@@ -23,12 +21,6 @@ var (
 	}
 )
 
-const (
-	DefaultComposeRecordTable    = "compose_record"
-	DefaultComposeRecordValueCol = "values"
-	DefaultPartitionFormat       = "compose_record_{{namespace}}_{{module}}"
-)
-
 func Run(ctx context.Context, log *zap.Logger, s store.Storer, provisionOpt options.ProvisionOpt, authOpt options.AuthOpt) error {
 	log = log.Named("provision")
 
@@ -54,7 +46,7 @@ func Run(ctx context.Context, log *zap.Logger, s store.Storer, provisionOpt opti
 		func() error { return oidcAutoDiscovery(ctx, log.Named("auth.oidc-auto-discovery"), s, authOpt) },
 		func() error { return defaultAuthClient(ctx, log.Named("auth.clients"), s, authOpt) },
 
-		func() error { return defaultDalConnection(ctx, log.Named("dal.connections"), s) },
+		func() error { return defaultDalConnection(ctx, s) },
 	}
 
 	for _, fn := range ffn {
@@ -118,40 +110,3 @@ func defaultAuthClient(ctx context.Context, log *zap.Logger, s store.AuthClients
 
 	return nil
 }
-
-func defaultDalConnection(ctx context.Context, log *zap.Logger, s store.DalConnections) (err error) {
-	cc, err := store.LookupDalConnectionByHandle(ctx, s, "primary_connection")
-	if err != nil && err != store.ErrNotFound {
-		return
-	}
-
-	// Already exists
-	if cc != nil {
-		return
-	}
-
-	// Create it
-	var conn = &types.DalConnection{
-		// Using id.Next since we dropped "special" ids a while ago.
-		// If needed, use the handle
-		ID:     id.Next(),
-		Name:   "Primary Connection",
-		Handle: "primary_connection",
-		Type:   types.DalPrimaryConnectionResourceType,
-
-		Config: types.ConnectionConfig{
-			DefaultModelIdent:      DefaultComposeRecordTable,
-			DefaultAttributeIdent:  DefaultComposeRecordValueCol,
-			DefaultPartitionFormat: DefaultPartitionFormat,
-
-			// The connection should be taken from the .env on boot_level
-		},
-		Capabilities: types.ConnectionCapabilities{
-			Supported: capabilities.FullCapabilities(),
-		},
-		CreatedAt: *now(),
-		CreatedBy: auth.ServiceUser().ID,
-	}
-
-	return store.CreateDalConnection(ctx, s, conn)
-}

store/adapters/rdbms/dal/connection.go

@@ -12,12 +12,15 @@ import (
 )
 
 type (
+	// connection provides (pkg/dal.Connection) interface to RDBMS implementation
+	//
+	// In other words: this allows Corteza to read Records from the supported SQL databases
 	connection struct {
 		mux          sync.RWMutex
 		models       map[string]*model
 		capabilities capabilities.Set
 
-		db      *sqlx.DB
+		db      sqlx.ExtContext
 		dialect drivers.Dialect
 	}
 )
@@ -30,7 +33,7 @@ func init() {
 	})
 }
 
-func Connection(db *sqlx.DB, dialect drivers.Dialect, cc ...capabilities.Capability) *connection {
+func Connection(db sqlx.ExtContext, dialect drivers.Dialect, cc ...capabilities.Capability) *connection {
 	return &connection{
 		db:           db,
 		dialect:      dialect,

store/adapters/rdbms/dal/iterator.go

@@ -162,8 +162,6 @@ func (i *iterator) fetch(ctx context.Context) (_ *sql.Rows, err error) {
 	}
 
 	sql, _, _ = query.Prepared(false).ToSQL()
-	println(sql)
-
 	if sql, args, err = query.ToSQL(); err != nil {
 		return nil, err
 	}

store/adapters/rdbms/drivers/mysql/connect.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
+	"github.com/cortezaproject/corteza-server/store/adapters/rdbms/dal"
 	"strings"
 
 	"github.com/cortezaproject/corteza-server/pkg/errors"
@@ -43,6 +45,8 @@ func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) {
 	s := &rdbms.Store{
 		DB: db,
 
+		DAL: dal.Connection(db, Dialect(), capabilities.FullCapabilities()...),
+
 		Dialect:           goquDialectWrapper,
 		TxRetryErrHandler: txRetryErrHandler,
 		ErrorHandler:      errorHandler,

store/adapters/rdbms/drivers/postgres/connect.go

@@ -3,6 +3,8 @@ package postgres
 import (
 	"context"
 	"database/sql"
+	"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
+	"github.com/cortezaproject/corteza-server/store/adapters/rdbms/dal"
 	"net/url"
 	"strings"
 
@@ -46,6 +48,8 @@ func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) {
 	s := &rdbms.Store{
 		DB: db,
 
+		DAL: dal.Connection(db, Dialect(), capabilities.FullCapabilities()...),
+
 		Dialect:      goquDialectWrapper,
 		ErrorHandler: errorHandler,
 

store/adapters/rdbms/drivers/sqlite/connect.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
+	"github.com/cortezaproject/corteza-server/store/adapters/rdbms/dal"
 	"regexp"
 	"strings"
 
@@ -67,6 +69,8 @@ func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) {
 	s := &rdbms.Store{
 		DB: db,
 
+		DAL: dal.Connection(db, Dialect(), capabilities.FullCapabilities()...),
+
 		Dialect:      goquDialectWrapper,
 		ErrorHandler: errorHandler,
 

store/adapters/rdbms/rdbms_store.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-
+	"github.com/cortezaproject/corteza-server/pkg/dal"
 	"github.com/cortezaproject/corteza-server/store"
 	"github.com/cortezaproject/corteza-server/store/adapters/rdbms/ddl"
 	"github.com/doug-martin/goqu/v9"
@@ -42,6 +42,9 @@ type (
 	Store struct {
 		DB sqlx.ExtContext
 
+		// DAL connection
+		DAL dal.Connection
+
 		// Logger for connection
 		Logger *zap.Logger
 
@@ -167,6 +170,11 @@ func (s Store) QueryOne(ctx context.Context, q sqlizer, dst interface{}) (err er
 	return exec.NewScanner(rows).ScanStruct(dst)
 }
 
+// ToDalConn uses store as DAL connection
+func (s Store) ToDalConn() dal.Connection {
+	return s.DAL
+}
+
 func DefaultFunctions() *Functions {
 	f := &Functions{}
 

store/dal.go

@@ -0,0 +1,11 @@
+package store
+
+import "github.com/cortezaproject/corteza-server/pkg/dal"
+
+// DAL uses given store as DAL connection
+//
+// This is mainly used to wrap primary store connection with DAL connection wrap
+// and use it to interact with records in a primary DB
+func DAL(s Storer) dal.Connection {
+	return s.ToDalConn()
+}

store/interfaces.gen.go

@@ -12,6 +12,7 @@ import (
 	composeType "github.com/cortezaproject/corteza-server/compose/types"
 	federationType "github.com/cortezaproject/corteza-server/federation/types"
 	actionlogType "github.com/cortezaproject/corteza-server/pkg/actionlog"
+	"github.com/cortezaproject/corteza-server/pkg/dal"
 	discoveryType "github.com/cortezaproject/corteza-server/pkg/discovery/types"
 	flagType "github.com/cortezaproject/corteza-server/pkg/flag/types"
 	labelsType "github.com/cortezaproject/corteza-server/pkg/label/types"
@@ -36,6 +37,9 @@ type (
 		// Tx is a transaction handler
 		Tx(context.Context, func(context.Context, Storer) error) error
 
+		// Returns underlying store as DAL connection
+		ToDalConn() dal.Connection
+
 		// Upgrade store's schema to the latest version
 		Upgrade(context.Context) error
 		Actionlogs

system/dalutils/connection.go

@@ -1,110 +0,0 @@
-package dalutils
-
-import (
-	"context"
-
-	"github.com/cortezaproject/corteza-server/pkg/dal"
-	"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
-	"github.com/cortezaproject/corteza-server/store"
-	"github.com/cortezaproject/corteza-server/system/types"
-)
-
-type (
-	connectionCreator interface {
-		CreateConnection(ctx context.Context, connectionID uint64, cp dal.ConnectionParams, cm dal.ConnectionMeta, capabilities ...capabilities.Capability) (err error)
-	}
-
-	connectionDeleter interface {
-		DeleteConnection(ctx context.Context, connectionID uint64) (err error)
-	}
-
-	connectionUpdater interface {
-		UpdateConnection(ctx context.Context, connectionID uint64, cp dal.ConnectionParams, cm dal.ConnectionMeta, capabilities ...capabilities.Capability) (err error)
-	}
-
-	connectionDeleteCreator interface {
-		connectionDeleter
-		connectionCreator
-	}
-)
-
-func DalConnectionReload(ctx context.Context, s store.Storer, dc connectionDeleteCreator) (err error) {
-	// Get all available connections
-	cc, _, err := store.SearchDalConnections(ctx, s, types.DalConnectionFilter{
-		Type: types.DalConnectionResourceType,
-	})
-	if err != nil {
-		return
-	}
-
-	for _, c := range cc {
-		var cm dal.ConnectionMeta
-		cm, err = ConnectionMeta(ctx, c)
-		if err != nil {
-			return
-		}
-		if err = dc.CreateConnection(ctx, c.ID, c.Config.Connection, cm, c.ActiveCapabilities()...); err != nil {
-			return
-		}
-	}
-
-	return
-}
-
-func DalConnectionCreate(ctx context.Context, c connectionCreator, connections ...*types.DalConnection) (err error) {
-	var cm dal.ConnectionMeta
-	for _, connection := range connections {
-		cm, err = ConnectionMeta(ctx, connection)
-		if err != nil {
-			return
-		}
-
-		if err = c.CreateConnection(ctx, connection.ID, connection.Config.Connection, cm, connection.ActiveCapabilities()...); err != nil {
-			return err
-		}
-	}
-
-	return
-}
-
-func DalConnectionUpdate(ctx context.Context, u connectionUpdater, connections ...*types.DalConnection) (err error) {
-	var cm dal.ConnectionMeta
-	for _, connection := range connections {
-		cm, err = ConnectionMeta(ctx, connection)
-		if err != nil {
-			return
-		}
-
-		if err = u.UpdateConnection(ctx, connection.ID, connection.Config.Connection, cm, connection.ActiveCapabilities()...); err != nil {
-			return err
-		}
-	}
-
-	return
-}
-
-func DalConnectionDelete(ctx context.Context, d connectionDeleter, connections ...*types.DalConnection) (err error) {
-	for _, connection := range connections {
-		if err = d.DeleteConnection(ctx, connection.ID); err != nil {
-			return err
-		}
-	}
-
-	return
-}
-
-// // // // // // // // // // // // // // // // // // // // // // // // //
-// Utils
-
-func ConnectionMeta(ctx context.Context, c *types.DalConnection) (cm dal.ConnectionMeta, err error) {
-	// @todo we could probably utilize connection params more here
-	cm = dal.ConnectionMeta{
-		DefaultModelIdent:      c.Config.DefaultModelIdent,
-		DefaultAttributeIdent:  c.Config.DefaultAttributeIdent,
-		DefaultPartitionFormat: c.Config.DefaultPartitionFormat,
-		SensitivityLevel:       c.SensitivityLevel,
-		Label:                  c.Handle,
-	}
-
-	return
-}

system/service/dal_connection.go

@@ -8,11 +8,9 @@ import (
 	"github.com/cortezaproject/corteza-server/pkg/actionlog"
 	a "github.com/cortezaproject/corteza-server/pkg/auth"
 	"github.com/cortezaproject/corteza-server/pkg/dal"
-	"github.com/cortezaproject/corteza-server/pkg/dal/capabilities"
 	"github.com/cortezaproject/corteza-server/pkg/options"
 
 	"github.com/cortezaproject/corteza-server/store"
-	"github.com/cortezaproject/corteza-server/system/dalutils"
 	"github.com/cortezaproject/corteza-server/system/types"
 )
 
@@ -21,7 +19,7 @@ type (
 		actionlog actionlog.Recorder
 		store     store.Storer
 		ac        connectionAccessController
-		dal       dalConnections
+		dal       dalConnManager
 		dbConf    options.DBOpt
 	}
 
@@ -35,15 +33,15 @@ type (
 		CanDeleteDalConnection(context.Context, *types.DalConnection) bool
 	}
 
-	dalConnections interface {
-		CreateConnection(ctx context.Context, connectionID uint64, cp dal.ConnectionParams, dft dal.ConnectionMeta, capabilities ...capabilities.Capability) (err error)
-		UpdateConnection(ctx context.Context, connectionID uint64, cp dal.ConnectionParams, dft dal.ConnectionMeta, capabilities ...capabilities.Capability) (err error)
-		DeleteConnection(ctx context.Context, connectionID uint64) (err error)
-		SearchConnectionIssues(connectionID uint64) (err []error)
+	// Connection management on DAL Service
+	dalConnManager interface {
+		ReplaceConnection(context.Context, *dal.ConnectionWrap, bool) error
+		RemoveConnection(context.Context, uint64) error
+		SearchConnectionIssues(uint64) []error
 	}
 )
 
-func Connection(ctx context.Context, dal dalConnections, dbConf options.DBOpt) *dalConnection {
+func Connection(ctx context.Context, dal dalConnManager, dbConf options.DBOpt) *dalConnection {
 	return &dalConnection{
 		ac:        DefaultAccessControl,
 		actionlog: DefaultActionlog,
@@ -108,7 +106,7 @@ func (svc *dalConnection) Create(ctx context.Context, new *types.DalConnection)
 
 		q = new
 
-		if err = dalutils.DalConnectionCreate(ctx, svc.dal, new); err != nil {
+		if err = dalConnectionReplace(ctx, svc.store.ToDalConn(), svc.dal, new); err != nil {
 			return err
 		}
 		svc.proc(q)
@@ -120,20 +118,19 @@ func (svc *dalConnection) Create(ctx context.Context, new *types.DalConnection)
 
 func (svc *dalConnection) Update(ctx context.Context, upd *types.DalConnection) (q *types.DalConnection, err error) {
 	var (
-		qProps = &dalConnectionActionProps{update: upd}
+		cProps = &dalConnectionActionProps{update: upd}
 		old    *types.DalConnection
-		e      error
 	)
 
 	err = func() (err error) {
-		if old, e = store.LookupDalConnectionByID(ctx, svc.store, upd.ID); e != nil {
-			return DalConnectionErrNotFound(qProps)
+		if old, err = store.LookupDalConnectionByID(ctx, svc.store, upd.ID); err != nil {
+			return DalConnectionErrNotFound(cProps)
 		}
 
 		svc.proc(old)
 
 		if !svc.ac.CanUpdateDalConnection(ctx, old) {
-			return DalConnectionErrNotAllowedToUpdate(qProps)
+			return DalConnectionErrNotAllowedToUpdate(cProps)
 		}
 
 		upd.UpdatedAt = now()
@@ -176,19 +173,15 @@ func (svc *dalConnection) Update(ctx context.Context, upd *types.DalConnection)
 
 		q = upd
 		defer svc.proc(q)
-		if old.HasIssues() {
-			return dalutils.DalConnectionCreate(ctx, svc.dal, upd)
-		}
-
-		return dalutils.DalConnectionUpdate(ctx, svc.dal, upd)
+		return dalConnectionReplace(ctx, svc.store.ToDalConn(), svc.dal, upd)
 	}()
-	return q, svc.recordAction(ctx, qProps, DalConnectionActionUpdate, err)
+	return q, svc.recordAction(ctx, cProps, DalConnectionActionUpdate, err)
 }
 
 func (svc *dalConnection) DeleteByID(ctx context.Context, ID uint64) (err error) {
 	var (
-		qProps = &dalConnectionActionProps{}
-		q      *types.DalConnection
+		cProps = &dalConnectionActionProps{}
+		c      *types.DalConnection
 	)
 
 	err = func() (err error) {
@@ -196,37 +189,37 @@ func (svc *dalConnection) DeleteByID(ctx context.Context, ID uint64) (err error)
 			return DalConnectionErrInvalidID()
 		}
 
-		if q, err = store.LookupDalConnectionByID(ctx, svc.store, ID); err != nil {
+		if c, err = store.LookupDalConnectionByID(ctx, svc.store, ID); err != nil {
 			return
 		}
 
-		if q.Type == types.DalPrimaryConnectionResourceType {
+		if c.Type == types.DalPrimaryConnectionResourceType {
 			return fmt.Errorf("not allowed to delete primary connections")
 		}
 
-		if !svc.ac.CanDeleteDalConnection(ctx, q) {
-			return DalConnectionErrNotAllowedToDelete(qProps)
+		if !svc.ac.CanDeleteDalConnection(ctx, c) {
+			return DalConnectionErrNotAllowedToDelete(cProps)
 		}
 
-		qProps.setConnection(q)
+		cProps.setConnection(c)
 
-		q.DeletedAt = now()
-		q.DeletedBy = a.GetIdentityFromContext(ctx).Identity()
+		c.DeletedAt = now()
+		c.DeletedBy = a.GetIdentityFromContext(ctx).Identity()
 
-		if err = store.UpdateDalConnection(ctx, svc.store, q); err != nil {
+		if err = store.UpdateDalConnection(ctx, svc.store, c); err != nil {
 			return
 		}
 
-		return dalutils.DalConnectionDelete(ctx, svc.dal, q)
+		return dalConnectionRemove(ctx, svc.dal, c)
 	}()
 
-	return svc.recordAction(ctx, qProps, DalConnectionActionDelete, err)
+	return svc.recordAction(ctx, cProps, DalConnectionActionDelete, err)
 }
 
 func (svc *dalConnection) UndeleteByID(ctx context.Context, ID uint64) (err error) {
 	var (
-		qProps = &dalConnectionActionProps{}
-		q      *types.DalConnection
+		cProps = &dalConnectionActionProps{}
+		c      *types.DalConnection
 	)
 
 	err = func() (err error) {
@@ -234,28 +227,28 @@ func (svc *dalConnection) UndeleteByID(ctx context.Context, ID uint64) (err erro
 			return DalConnectionErrInvalidID()
 		}
 
-		if q, err = store.LookupDalConnectionByID(ctx, svc.store, ID); err != nil {
+		if c, err = store.LookupDalConnectionByID(ctx, svc.store, ID); err != nil {
 			return
 		}
 
-		if !svc.ac.CanDeleteDalConnection(ctx, q) {
-			return DalConnectionErrNotAllowedToUndelete(qProps)
+		if !svc.ac.CanDeleteDalConnection(ctx, c) {
+			return DalConnectionErrNotAllowedToUndelete(cProps)
 		}
 
-		qProps.setConnection(q)
+		cProps.setConnection(c)
 
-		q.DeletedAt = nil
-		q.UpdatedBy = a.GetIdentityFromContext(ctx).Identity()
+		c.DeletedAt = nil
+		c.UpdatedBy = a.GetIdentityFromContext(ctx).Identity()
 
-		if err = store.UpdateDalConnection(ctx, svc.store, q); err != nil {
+		if err = store.UpdateDalConnection(ctx, svc.store, c); err != nil {
 			return
 		}
 
 		// We're creating it here since it was removed on delete
-		return dalutils.DalConnectionCreate(ctx, svc.dal, q)
+		return dalConnectionRemove(ctx, svc.dal, c)
 	}()
 
-	return svc.recordAction(ctx, qProps, DalConnectionActionDelete, err)
+	return svc.recordAction(ctx, cProps, DalConnectionActionDelete, err)
 }
 
 func (svc *dalConnection) Search(ctx context.Context, filter types.DalConnectionFilter) (r types.DalConnectionSet, f types.DalConnectionFilter, err error) {
@@ -288,7 +281,7 @@ func (svc *dalConnection) Search(ctx context.Context, filter types.DalConnection
 }
 
 func (svc *dalConnection) ReloadConnections(ctx context.Context) (err error) {
-	return dalutils.DalConnectionReload(ctx, svc.store, svc.dal)
+	return dalConnectionReload(ctx, svc.store, svc.dal)
 }
 
 func (svc *dalConnection) proc(connections ...*types.DalConnection) {
@@ -322,3 +315,64 @@ func (svc *dalConnection) procDal(c *types.DalConnection) {
 func (svc *dalConnection) procLocale(c *types.DalConnection) {
 	// @todo...
 }
+
+func dalConnectionReload(ctx context.Context, s store.Storer, dcm dalConnManager) (err error) {
+	// Get all available connections
+	cc, _, err := store.SearchDalConnections(ctx, s, types.DalConnectionFilter{})
+	if err != nil {
+		return
+	}
+
+	return dalConnectionReplace(ctx, s.ToDalConn(), dcm, cc...)
+}
+
+// Replaces all given connections
+func dalConnectionReplace(ctx context.Context, primary dal.Connection, dcm dalConnManager, cc ...*types.DalConnection) (err error) {
+	var (
+		cw        *dal.ConnectionWrap
+		isPrimary bool
+	)
+
+	for _, c := range cc {
+		isPrimary = c.Type == types.DalPrimaryConnectionResourceType
+
+		cw = dal.MakeConnection(
+			c.ID,
+			// When connection is primary (type) we use the primary connection
+			// passed in to the fn
+			func() dal.Connection {
+				if isPrimary {
+					return primary
+				}
+
+				return nil
+			}(),
+			c.Config.Connection,
+			dal.ConnectionMeta{
+				DefaultModelIdent:      c.Config.DefaultModelIdent,
+				DefaultAttributeIdent:  c.Config.DefaultAttributeIdent,
+				DefaultPartitionFormat: c.Config.DefaultPartitionFormat,
+				SensitivityLevel:       c.SensitivityLevel,
+				Label:                  c.Handle,
+			},
+			c.ActiveCapabilities()...,
+		)
+
+		if err = dcm.ReplaceConnection(ctx, cw, isPrimary); err != nil {
+			return err
+		}
+	}
+
+	return
+}
+
+// Removes a connection from DAL service
+func dalConnectionRemove(ctx context.Context, dcm dalConnManager, cc ...*types.DalConnection) (err error) {
+	for _, c := range cc {
+		if err = dcm.RemoveConnection(ctx, c.ID); err != nil {
+			return err
+		}
+	}
+
+	return
+}

system/service/service.go

@@ -88,7 +88,6 @@ var (
 	DefaultApigwFilter         *apigwFilter
 	DefaultApigwProfiler       *apigwProfiler
 	DefaultReport              *report
-	primaryConnectionConfig    *types.DalConnection
 	DefaultDataPrivacy         *dataPrivacy
 
 	DefaultStatistics *statistics
@@ -105,7 +104,7 @@ var (
 	}
 )
 
-func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, primaryConn *types.DalConnection, ws websocketSender, c Config) (err error) {
+func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, ws websocketSender, c Config) (err error) {
 	var (
 		hcd = healthcheck.Defaults()
 	)
@@ -151,7 +150,6 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, primaryCon
 
 	DefaultSettings = Settings(ctx, DefaultStore, DefaultLogger, DefaultAccessControl, CurrentSettings)
 
-	primaryConnectionConfig = primaryConn
 	DefaultDalConnection = Connection(ctx, dal.Service(), c.DB)
 
 	DefaultDalSensitivityLevel = SensitivityLevel(ctx, dal.Service())
@@ -277,6 +275,19 @@ func Activate(ctx context.Context) (err error) {
 		return
 	}
 
+	// Reload DAL sensitivity levels
+	err = DefaultDalSensitivityLevel.ReloadSensitivityLevels(ctx, DefaultStore)
+	if err != nil {
+		return
+	}
+
+	// Reload DAL connections
+	println("reloading connections")
+	err = DefaultDalConnection.ReloadConnections(ctx)
+	if err != nil {
+		return
+	}
+
 	return
 }
 

system/types/dal_connection.go

@@ -77,7 +77,8 @@ type (
 
 var (
 	// Used to identify the primary DAL connection instead of an extra flag
-	DalPrimaryConnectionResourceType = "corteza::system:primary_dal_connection"
+	DalPrimaryConnectionResourceType = "corteza::system:primary-dal-connection"
+	DalPrimaryConnectionHandle       = "primary-database"
 )
 
 func (c DalConnection) ActiveCapabilities() capabilities.Set {

tests/dal/dal_utils.go

@@ -14,7 +14,6 @@ import (
 	"github.com/cortezaproject/corteza-server/pkg/filter"
 	"github.com/cortezaproject/corteza-server/pkg/id"
 	"github.com/cortezaproject/corteza-server/store"
-	"github.com/cortezaproject/corteza-server/system/dalutils"
 	"github.com/cortezaproject/corteza-server/system/service"
 	"github.com/cortezaproject/corteza-server/system/types"
 	"github.com/cortezaproject/corteza-server/tests/dal/setup/mysql"
@@ -140,9 +139,12 @@ func (h helper) cleanupDal() {
 func initSvc(ctx context.Context, d driver) (dalService, error) {
 	c := makeConnectionDefinition(d.dsn)
 
-	cm, err := dalutils.ConnectionMeta(ctx, c)
-	if err != nil {
-		return nil, err
+	cm := dal.ConnectionMeta{
+		DefaultModelIdent:      c.Config.DefaultModelIdent,
+		DefaultAttributeIdent:  c.Config.DefaultAttributeIdent,
+		DefaultPartitionFormat: c.Config.DefaultPartitionFormat,
+		SensitivityLevel:       c.SensitivityLevel,
+		Label:                  c.Handle,
 	}
 
 	svc, err := dal.New(