upstream-corteza/server/system/service/dal_schema_alteration.go

package service

import (
	"context"
	"fmt"

	"github.com/cortezaproject/corteza/server/pkg/actionlog"
	intAuth "github.com/cortezaproject/corteza/server/pkg/auth"
	"github.com/cortezaproject/corteza/server/pkg/dal"
	"github.com/cortezaproject/corteza/server/pkg/errors"
	"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/types"
)

type (
	dalSchemaAlteration struct {
		actionlog actionlog.Recorder
		ac        dalSchemaAlterationAccessController
		dal       dalAltManager

		store store.Storer
	}

	dalAltManager interface {
		GetConnectionByID(ID uint64) *dal.ConnectionWrap
		ApplyAlteration(ctx context.Context, alts ...*dal.Alteration) (errs []error, err error)
		ReloadModel(ctx context.Context, currentAlts []*dal.Alteration, model *dal.Model) (newAlts []*dal.Alteration, err error)
		FindModelByResourceIdent(connectionID uint64, resourceType, resourceIdent string) *dal.Model
	}

	dalSchemaAlterationAccessController interface {
		CanManageDalSchemaAlterations(ctx context.Context) bool
	}
)

func DalSchemaAlteration(dal dalAltManager) *dalSchemaAlteration {
	return &dalSchemaAlteration{
		ac:        DefaultAccessControl,
		store:     DefaultStore,
		actionlog: DefaultActionlog,
		dal:       dal,
	}
}

func (svc dalSchemaAlteration) FindByID(ctx context.Context, dalSchemaAlterationID uint64) (a *types.DalSchemaAlteration, err error) {
	var (
		uaProps = &dalSchemaAlterationActionProps{dalSchemaAlteration: &types.DalSchemaAlteration{ID: dalSchemaAlterationID}}
	)

	err = func() error {
		a, err = loadDalSchemaAlteration(ctx, svc.store, dalSchemaAlterationID)
		if err != nil {
			return err
		}

		uaProps.setDalSchemaAlteration(a)

		if !svc.ac.CanManageDalSchemaAlterations(ctx) {
			return DalSchemaAlterationErrNotAllowedToManage()
		}

		return nil
	}()

	return a, svc.recordAction(ctx, uaProps, DalSchemaAlterationActionLookup, err)
}

// Search interacts with backend storage and
//
// @todo rename to Search() for consistency
func (svc dalSchemaAlteration) Search(ctx context.Context, filter types.DalSchemaAlterationFilter) (aa types.DalSchemaAlterationSet, f types.DalSchemaAlterationFilter, err error) {
	var (
		uaProps = &dalSchemaAlterationActionProps{filter: &filter}
	)

	err = func() error {
		if !svc.ac.CanManageDalSchemaAlterations(ctx) {
			return DalSchemaAlterationErrNotAllowedToManage()
		}

		aa, f, err = store.SearchDalSchemaAlterations(ctx, svc.store, filter)
		return err
	}()

	return aa, f, svc.recordAction(ctx, uaProps, DalSchemaAlterationActionSearch, err)
}

// ModelAlterations returns all non deleted, non completed, and non dismissed alterations for the given model
func (svc dalSchemaAlteration) ModelAlterations(ctx context.Context, m *dal.Model) (out []*dal.Alteration, err error) {
	return svc.modelAlterations(ctx, svc.store, m)
}

func (svc dalSchemaAlteration) modelAlterations(ctx context.Context, s store.Storer, m *dal.Model) (out []*dal.Alteration, err error) {
	aux, _, err := store.SearchDalSchemaAlterations(ctx, s, types.DalSchemaAlterationFilter{
		Resource:     []string{m.Resource},
		ResourceType: m.ResourceType,
		Deleted:      filter.StateExcluded,
		Completed:    filter.StateExcluded,
		Dismissed:    filter.StateExcluded,
	})
	if err != nil {
		return nil, err
	}

	for _, a := range aux {
		t := &dal.Alteration{
			ID:           a.ID,
			BatchID:      a.BatchID,
			DependsOn:    a.DependsOn,
			Resource:     a.Resource,
			ResourceType: a.ResourceType,
			ConnectionID: a.ConnectionID,
		}

		switch a.Kind {
		case "attributeAdd":
			t.AttributeAdd = a.Params.AttributeAdd
		case "attributeDelete":
			t.AttributeDelete = a.Params.AttributeDelete
		case "attributeReType":
			t.AttributeReType = a.Params.AttributeReType
		case "attributeReEncode":
			t.AttributeReEncode = a.Params.AttributeReEncode
		case "modelAdd":
			t.ModelAdd = a.Params.ModelAdd
		case "modelDelete":
			t.ModelDelete = a.Params.ModelDelete
		}

		out = append(out, t)
	}

	return
}

// SetAlterations updates the DB state to reflect the given alterations
//
// This function should only be invoked by internal proceses so it doesn't need
// to check for permissions.
func (svc dalSchemaAlteration) SetAlterations(ctx context.Context, s store.Storer, m *dal.Model, stale []*dal.Alteration, aa ...*dal.Alteration) (err error) {
	if len(stale)+len(aa) == 0 {
		return
	}

	var (
		n = *now()
		u = intAuth.GetIdentityFromContext(ctx).Identity()
	)

	// Delete current ones
	// @todo we might be able to do some diffing to preserve the metadata/ids
	//       but for now this should be just fine.
	auxStale := make([]*types.DalSchemaAlteration, len(stale))
	for i, a := range stale {
		auxStale[i] = &types.DalSchemaAlteration{
			ID: a.ID,
		}
	}
	err = store.DeleteDalSchemaAlteration(ctx, s, auxStale...)
	if err != nil {
		return
	}

	// Set new ones
	cvt := make(types.DalSchemaAlterationSet, len(aa))
	for i, a := range aa {
		t := &types.DalSchemaAlteration{
			ID:           a.ID,
			BatchID:      a.BatchID,
			DependsOn:    a.DependsOn,
			ConnectionID: a.ConnectionID,
			Resource:     a.Resource,
			ResourceType: a.ResourceType,

			Params: &types.DalSchemaAlterationParams{},
		}

		t.ID = nextID()
		t.CreatedAt = n
		t.CreatedBy = u

		switch {
		case a.AttributeAdd != nil:
			t.Kind = "attributeAdd"
			t.Params.AttributeAdd = a.AttributeAdd

		case a.AttributeDelete != nil:
			t.Kind = "attributeDelete"
			t.Params.AttributeDelete = a.AttributeDelete

		case a.AttributeReType != nil:
			t.Kind = "attributeReType"
			t.Params.AttributeReType = a.AttributeReType

		case a.AttributeReEncode != nil:
			t.Kind = "attributeReEncode"
			t.Params.AttributeReEncode = a.AttributeReEncode

		case a.ModelAdd != nil:
			t.Kind = "modelAdd"
			t.Params.ModelAdd = a.ModelAdd

		case a.ModelDelete != nil:
			t.Kind = "modelDelete"
			t.Params.ModelDelete = a.ModelDelete

		default:
			panic(fmt.Sprintf("unknown alteration type %v", a))
		}

		cvt[i] = t
	}

	return store.UpsertDalSchemaAlteration(ctx, svc.store, cvt...)
}

func (svc dalSchemaAlteration) Apply(ctx context.Context, ids ...uint64) (err error) {
	var (
		uaProps = &dalSchemaAlterationActionProps{}
	)

	err = func() (err error) {
		if !svc.ac.CanManageDalSchemaAlterations(ctx) {
			return DalSchemaAlterationErrNotAllowedToManage()
		}

		aux, _, err := store.SearchDalSchemaAlterations(ctx, svc.store, types.DalSchemaAlterationFilter{
			AlterationID: id.Strings(ids...),
		})
		if err != nil {
			return
		}

		alts := svc.appliableAlterations(aux...)
		pkgAlts, err := svc.toPkgAlterations(ctx, alts...)
		if err != nil {
			return
		}

		ii := make([]uint64, len(alts))
		for i, a := range alts {
			ii[i] = a.ID
		}

		uaProps.setApply(ii)

		errors, err := svc.dal.ApplyAlteration(ctx, pkgAlts...)
		if err != nil {
			return
		}

		for i, e := range errors {
			if e != nil {
				aux[i].Error = e.Error()
			} else {
				aux[i].CompletedAt = now()
				aux[i].CompletedBy = intAuth.GetIdentityFromContext(ctx).Identity()
			}
		}

		err = store.UpdateDalSchemaAlteration(ctx, svc.store, aux...)
		if err != nil {
			return
		}

		return svc.reloadAlteredModels(ctx, svc.store, alts)
	}()

	return svc.recordAction(ctx, uaProps, DalSchemaAlterationActionApply, err)

}

func (svc dalSchemaAlteration) Dismiss(ctx context.Context, ids ...uint64) (err error) {
	var (
		uaProps = &dalSchemaAlterationActionProps{}
	)

	err = store.Tx(ctx, svc.store, func(ctx context.Context, s store.Storer) (err error) {
		if !svc.ac.CanManageDalSchemaAlterations(ctx) {
			return DalSchemaAlterationErrNotAllowedToManage()
		}

		alt, _, err := store.SearchDalSchemaAlterations(ctx, s, types.DalSchemaAlterationFilter{
			AlterationID: id.Strings(ids...),
		})
		if err != nil {
			return
		}

		ii := make([]uint64, len(alt))
		for i, a := range alt {
			ii[i] = a.ID
		}

		uaProps.setApply(ii)

		alt = svc.appliableAlterations(alt...)
		identity := intAuth.GetIdentityFromContext(ctx).Identity()
		for _, a := range alt {
			a.Error = ""
			a.DismissedAt = now()
			a.DismissedBy = identity
		}

		err = store.UpdateDalSchemaAlteration(ctx, s, alt...)
		if err != nil {
			return
		}

		return svc.reloadAlteredModels(ctx, s, alt)
	})

	return svc.recordAction(ctx, uaProps, DalSchemaAlterationActionDismiss, err)
}

func (svc dalSchemaAlteration) appliableAlterations(aa ...*types.DalSchemaAlteration) (out types.DalSchemaAlterationSet) {
	out = make(types.DalSchemaAlterationSet, 0, len(aa))

	altIndex := make(map[uint64]*types.DalSchemaAlteration, len(aa))
	for _, a := range aa {
		altIndex[a.ID] = a
	}

	for _, a := range aa {
		// Already completed
		if a.CompletedAt != nil {
			continue
		}

		// Dismissed for manual thing
		if a.DismissedAt != nil {
			continue
		}

		if a.DependsOn != 0 {
			// Check if the dependency is completed or inside of the index
			if _, ok := altIndex[a.DependsOn]; !ok {
				continue
			}
		}

		out = append(out, a)
	}

	return
}

func loadDalSchemaAlteration(ctx context.Context, s store.DalSchemaAlterations, ID uint64) (res *types.DalSchemaAlteration, err error) {
	if ID == 0 {
		return nil, DalSchemaAlterationErrInvalidID()
	}

	if res, err = store.LookupDalSchemaAlterationByID(ctx, s, ID); errors.IsNotFound(err) {
		return nil, DalSchemaAlterationErrNotFound()
	}

	return
}

func (svc dalSchemaAlteration) toPkgAlterations(ctx context.Context, aa ...*types.DalSchemaAlteration) (out []*dal.Alteration, err error) {
	out = make([]*dal.Alteration, len(aa))
	for i, a := range aa {
		t := &dal.Alteration{
			ID:           a.ID,
			BatchID:      a.BatchID,
			DependsOn:    a.DependsOn,
			ConnectionID: a.ConnectionID,
			Resource:     a.Resource,
			ResourceType: a.ResourceType,
		}

		switch a.Kind {
		case "attributeAdd":
			t.AttributeAdd = a.Params.AttributeAdd
		case "attributeDelete":
			t.AttributeDelete = a.Params.AttributeDelete
		case "attributeReType":
			t.AttributeReType = a.Params.AttributeReType
		case "attributeReEncode":
			t.AttributeReEncode = a.Params.AttributeReEncode
		case "modelAdd":
			t.ModelAdd = a.Params.ModelAdd
		case "modelDelete":
			t.ModelDelete = a.Params.ModelDelete
		}

		out[i] = t
	}

	return
}

func (svc dalSchemaAlteration) reloadAlteredModels(ctx context.Context, s store.Storer, alts types.DalSchemaAlterationSet) (err error) {
	// Skip any models whish were already reloaded by some alterations.
	// These might be mixed up so we'll need to do it like so.
	processed := make(map[string]bool, 3)
	mkKey := func(a *types.DalSchemaAlteration) string {
		return fmt.Sprintf("%s;%s", a.ResourceType, a.Resource)
	}

	for _, a := range alts {
		k := mkKey(a)
		if processed[k] {
			continue
		}
		processed[k] = true

		err = svc.reloadAlteredModel(ctx, s, a)
		if err != nil {
			return
		}
	}

	return
}

func (svc dalSchemaAlteration) reloadAlteredModel(ctx context.Context, s store.Storer, alt *types.DalSchemaAlteration) (err error) {
	// Fetch current alterations to see if there are any left over
	_, f, err := store.SearchDalSchemaAlterations(ctx, s, types.DalSchemaAlterationFilter{
		Resource:     []string{alt.Resource},
		ResourceType: alt.ResourceType,
		Deleted:      filter.StateExcluded,
		Completed:    filter.StateExcluded,
		Dismissed:    filter.StateExcluded,

		Paging: filter.Paging{
			IncTotal: true,
		},
	})
	if err != nil {
		return err
	}

	// There are some alterations left so we can't reload the model
	if f.Total > 0 {
		return
	}

	model := svc.dal.FindModelByResourceIdent(alt.ConnectionID, alt.ResourceType, alt.Resource)
	if model == nil {
		err = fmt.Errorf("cannot find model for resource %s", alt.Resource)
		return
	}

	currentAlts, err := svc.modelAlterations(ctx, s, model)
	if err != nil {
		return
	}

	newAlts, err := svc.dal.ReloadModel(ctx, currentAlts, model)
	if err != nil {
		return
	}

	err = svc.SetAlterations(ctx, s, model, currentAlts, newAlts...)
	if err != nil {
		return
	}

	return
}