corteza/system/service/queue.go

package service

import (
	"context"

	"github.com/cortezaproject/corteza-server/pkg/actionlog"
	"github.com/cortezaproject/corteza-server/pkg/messagebus"
	"github.com/cortezaproject/corteza-server/store"
)

type (
	queue struct {
		actionlog actionlog.Recorder
		store     store.Storer
		ac        queueAccessController
	}

	queueAccessController interface {
		CanCreateQueue(ctx context.Context) bool
		CanSearchQueues(ctx context.Context) bool
		CanReadQueue(ctx context.Context, c *messagebus.QueueSettings) bool
		CanUpdateQueue(ctx context.Context, c *messagebus.QueueSettings) bool
		CanDeleteQueue(ctx context.Context, c *messagebus.QueueSettings) bool
	}
)

func Queue() *queue {
	return &queue{
		ac:        DefaultAccessControl,
		actionlog: DefaultActionlog,
		store:     DefaultStore,
	}
}

func (svc *queue) FindByID(ctx context.Context, ID uint64) (q *messagebus.QueueSettings, err error) {
	var (
		qProps = &queueActionProps{}
	)

	err = func() error {
		if ID == 0 {
			return QueueErrInvalidID()
		}

		if q, err = store.LookupMessagebusQueueSettingByID(ctx, svc.store, ID); err != nil {
			return TemplateErrInvalidID().Wrap(err)
		}

		qProps.setQueue(q)

		if !svc.ac.CanReadQueue(ctx, q) {
			return QueueErrNotAllowedToRead(qProps)
		}

		return nil
	}()

	return q, svc.recordAction(ctx, qProps, QueueActionLookup, err)
}

func (svc *queue) Create(ctx context.Context, new *messagebus.QueueSettings) (q *messagebus.QueueSettings, err error) {
	var (
		qProps = &queueActionProps{new: new}
	)

	err = func() (err error) {
		if !svc.ac.CanCreateQueue(ctx) {
			return QueueErrNotAllowedToCreate(qProps)
		}

		if !svc.isValidHandler(messagebus.ConsumerType(new.Consumer)) {
			return QueueErrInvalidConsumer(qProps)
		}

		// Set new values after beforeCreate events are emitted
		new.ID = nextID()
		new.CreatedAt = *now()

		if err = store.CreateMessagebusQueueSetting(ctx, svc.store, new); err != nil {
			return
		}

		q = new

		// send the signal to reload all queues
		messagebus.Service().ReloadQueues()

		return nil
	}()

	return q, svc.recordAction(ctx, qProps, QueueActionCreate, err)
}

func (svc *queue) Update(ctx context.Context, upd *messagebus.QueueSettings) (q *messagebus.QueueSettings, err error) {
	var (
		qProps = &queueActionProps{update: upd}
		qq     *messagebus.QueueSettings
		e      error
	)

	err = func() (err error) {
		if !svc.ac.CanUpdateQueue(ctx, upd) {
			return QueueErrNotAllowedToUpdate(qProps)
		}

		if qq, e = store.LookupMessagebusQueueSettingByID(ctx, svc.store, upd.ID); e != nil {
			return QueueErrNotFound(qProps)
		}

		if qq, e := store.LookupMessagebusQueueSettingByQueue(ctx, svc.store, upd.Queue); e == nil && qq != nil && qq.ID != upd.ID {
			return QueueErrAlreadyExists(qProps)
		}

		if !svc.isValidHandler(messagebus.ConsumerType(upd.Consumer)) {
			return QueueErrInvalidConsumer(qProps)
		}

		// Set new values after beforeCreate events are emitted
		upd.UpdatedAt = now()
		upd.CreatedAt = qq.CreatedAt

		if err = store.UpdateMessagebusQueueSetting(ctx, svc.store, upd); err != nil {
			return
		}

		q = upd

		// send the signal to reload all queues
		messagebus.Service().ReloadQueues()

		return nil
	}()

	return q, svc.recordAction(ctx, qProps, QueueActionUpdate, err)
}

func (svc *queue) DeleteByID(ctx context.Context, ID uint64) (err error) {
	var (
		qProps = &queueActionProps{}
		q      *messagebus.QueueSettings
	)

	err = func() (err error) {
		if ID == 0 {
			return QueueErrInvalidID()
		}

		if q, err = store.LookupMessagebusQueueSettingByID(ctx, svc.store, ID); err != nil {
			return
		}

		qProps.setQueue(q)

		if !svc.ac.CanDeleteQueue(ctx, q) {
			return QueueErrNotAllowedToDelete(qProps)
		}

		q.DeletedAt = now()
		if err = store.UpdateMessagebusQueueSetting(ctx, svc.store, q); err != nil {
			return
		}

		// send the signal to reload all queues
		messagebus.Service().ReloadQueues()

		return nil
	}()

	return svc.recordAction(ctx, qProps, QueueActionDelete, err)
}

func (svc *queue) UndeleteByID(ctx context.Context, ID uint64) (err error) {
	var (
		qProps = &queueActionProps{}
		q      *messagebus.QueueSettings
	)

	err = func() (err error) {
		if ID == 0 {
			return QueueErrInvalidID()
		}

		if q, err = store.LookupMessagebusQueueSettingByID(ctx, svc.store, ID); err != nil {
			return
		}

		qProps.setQueue(q)

		if !svc.ac.CanDeleteQueue(ctx, q) {
			return QueueErrNotAllowedToDelete(qProps)
		}

		q.DeletedAt = nil
		if err = store.UpdateMessagebusQueueSetting(ctx, svc.store, q); err != nil {
			return
		}

		// send the signal to reload all queues
		messagebus.Service().ReloadQueues()

		return nil
	}()

	return svc.recordAction(ctx, qProps, QueueActionDelete, err)
}

func (svc *queue) Search(ctx context.Context, filter messagebus.QueueSettingsFilter) (q messagebus.QueueSettingsSet, f messagebus.QueueSettingsFilter, err error) {
	var (
		aProps = &queueActionProps{search: &filter}
	)

	// For each fetched item, store backend will check if it is valid or not
	filter.Check = func(res *messagebus.QueueSettings) (bool, error) {
		if !svc.ac.CanReadQueue(ctx, res) {
			return false, nil
		}

		return true, nil
	}

	err = func() error {
		if !svc.ac.CanSearchQueues(ctx) {
			return QueueErrNotAllowedToSearch()
		}

		if q, f, err = store.SearchMessagebusQueueSettings(ctx, svc.store, filter); err != nil {
			return err
		}

		return nil
	}()

	return q, f, svc.recordAction(ctx, aProps, QueueActionSearch, err)
}

func (svc *queue) isValidHandler(h messagebus.ConsumerType) bool {
	for _, hh := range messagebus.ConsumerTypes() {
		if h == hh {
			return true
		}
	}
	return false
}