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

package service

import (
	"context"
	"time"

	"github.com/cortezaproject/corteza/server/pkg/actionlog"
	intAuth "github.com/cortezaproject/corteza/server/pkg/auth"
	"github.com/cortezaproject/corteza/server/pkg/filter"
	"github.com/cortezaproject/corteza/server/store"
	"github.com/cortezaproject/corteza/server/system/types"
	"go.uber.org/zap"
)

type (
	notification struct {
		ac                 notificationAccessController
		log                *zap.Logger
		actionlog          actionlog.Recorder
		store              store.Storer
		notificationSender notificationSender
	}

	notificationSender interface {
		Send(kind string, payload interface{}, userIDs ...uint64) error
	}

	notificationAccessController interface {
		CanAssignNotification(ctx context.Context) bool
	}

	NotificationService interface {
		Find(context.Context, types.NotificationFilter) (types.NotificationSet, types.NotificationFilter, error)
		FindByID(context.Context, uint64) (*types.Notification, error)

		Create(context.Context, *types.Notification) (*types.Notification, error)
		Update(context.Context, *types.Notification) (*types.Notification, error)
		Delete(context.Context, uint64) error

		MarkAsRead(context.Context, uint64) error
		MarkAsUnread(context.Context, uint64) error
		MarkAllAsRead(context.Context) error
		MarkAllAsUnread(context.Context) error
	}
)

func Notification(ctx context.Context, log *zap.Logger, ns notificationSender) NotificationService {
	return &notification{
		ac:                 DefaultAccessControl,
		log:                log,
		store:              DefaultStore,
		notificationSender: ns,
	}
}

func (svc notification) Find(ctx context.Context, filter types.NotificationFilter) (nn types.NotificationSet, f types.NotificationFilter, err error) {
	var (
		raProps = &notificationActionProps{filter: &filter}
	)

	err = func() (err error) {
		// Get current user ID
		currentUserID := intAuth.GetIdentityFromContext(ctx).Identity()

		// Override recipient filter to ensure users can only see their own notifications
		filter.Recipient = currentUserID

		nn, f, err = store.SearchNotifications(ctx, svc.store, filter)
		if err != nil {
			return err
		}

		// Ensure we never return nil, but an empty slice instead
		if nn == nil {
			nn = make(types.NotificationSet, 0)
		}

		return nil
	}()

	return nn, f, svc.recordAction(ctx, raProps, NotificationActionSearch, err)
}

func (svc notification) FindByID(ctx context.Context, ID uint64) (n *types.Notification, err error) {
	var (
		raProps = &notificationActionProps{notification: &types.Notification{ID: ID}}
	)

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

		n, err = store.LookupNotificationByID(ctx, svc.store, ID)
		if err != nil {
			return err
		}

		// Check if the notification belongs to the current user
		currentUserID := intAuth.GetIdentityFromContext(ctx).Identity()
		if n.Recipient != currentUserID {
			return NotificationErrNotFound()
		}

		raProps.setNotification(n)

		return nil
	}()

	return n, svc.recordAction(ctx, raProps, NotificationActionLookup, err)
}

func (svc notification) Create(ctx context.Context, new *types.Notification) (n *types.Notification, err error) {
	var (
		raProps = &notificationActionProps{new: new}
	)

	err = func() (err error) {
		// Check if current user has permission to assign notifications to others
		if err := svc.checkAssignee(ctx, new); err != nil {
			return err
		}

		n = new
		n.ID = nextID()
		n.CreatedAt = *now()

		// Set creator
		currentUserID := intAuth.GetIdentityFromContext(ctx).Identity()
		n.CreatedBy = currentUserID

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

		// Send the notification via websocket
		if svc.notificationSender != nil {
			// Send only to the recipient
			if err = svc.notificationSender.Send("notification", n, n.Recipient); err != nil {
				return err
			}
		}

		return nil
	}()

	return n, svc.recordAction(ctx, raProps, NotificationActionCreate, err)
}

func (svc notification) Update(ctx context.Context, upd *types.Notification) (n *types.Notification, err error) {
	var (
		raProps = &notificationActionProps{updated: upd}
	)

	err = func() (err error) {
		if upd.ID == 0 {
			return NotificationErrInvalidID()
		}

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

		// Check if the notification belongs to the current user
		currentUserID := intAuth.GetIdentityFromContext(ctx).Identity()
		if n.Recipient != currentUserID {
			return NotificationErrNotFound()
		}

		// Check if the recipient is being changed and if so, check permissions
		if upd.Recipient != 0 && upd.Recipient != n.Recipient {
			// Create a temporary notification for the permission check
			tempNotification := &types.Notification{
				Recipient: upd.Recipient,
			}

			if err := svc.checkAssignee(ctx, tempNotification); err != nil {
				return err
			}

			n.Recipient = upd.Recipient
		}

		// Assign changed values
		n.Kind = upd.Kind
		n.Config = upd.Config
		n.UpdatedAt = now()

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

		return nil
	}()

	return n, svc.recordAction(ctx, raProps, NotificationActionUpdate, err)
}

func (svc notification) Delete(ctx context.Context, ID uint64) (err error) {
	var (
		n *types.Notification

		raProps = &notificationActionProps{notification: &types.Notification{ID: ID}}
	)

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

		if n, err = svc.FindByID(ctx, ID); err != nil {
			return NotificationErrNotFound()
		}

		// Check if the notification belongs to the current user
		currentUserID := intAuth.GetIdentityFromContext(ctx).Identity()
		if n.Recipient != currentUserID {
			return NotificationErrNotFound()
		}

		raProps.setNotification(n)

		n.DeletedAt = now()

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

		// Send the deleted notification via websocket so client can update UI
		if svc.notificationSender != nil {
			if err = svc.notificationSender.Send("notification.delete", n, n.Recipient); err != nil {
				return err
			}
		}

		return nil
	}()

	return svc.recordAction(ctx, raProps, NotificationActionDelete, err)
}

func (svc notification) MarkAsRead(ctx context.Context, ID uint64) (err error) {
	var (
		n *types.Notification

		raProps = &notificationActionProps{notification: &types.Notification{ID: ID}}
	)

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

		if n, err = store.LookupNotificationByID(ctx, svc.store, ID); err != nil {
			return NotificationErrNotFound()
		}

		// Check if the notification belongs to the current user
		currentUserID := intAuth.GetIdentityFromContext(ctx).Identity()
		if n.Recipient != currentUserID {
			return NotificationErrNotAllowedToRead()
		}

		raProps.setNotification(n)

		// Mark as read
		now := time.Now()
		n.ReadAt = &now
		n.UpdatedAt = &now

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

		// Send the updated notification via websocket so client can update UI
		if svc.notificationSender != nil {
			if err = svc.notificationSender.Send("notification.read", n, n.Recipient); err != nil {
				return err
			}
		}

		return nil
	}()

	return svc.recordAction(ctx, raProps, NotificationActionMarkAsRead, err)
}

func (svc notification) MarkAsUnread(ctx context.Context, ID uint64) (err error) {
	var (
		n *types.Notification

		raProps = &notificationActionProps{notification: &types.Notification{ID: ID}}
	)

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

		if n, err = store.LookupNotificationByID(ctx, svc.store, ID); err != nil {
			return NotificationErrNotFound()
		}

		// Check if the notification belongs to the current user
		currentUserID := intAuth.GetIdentityFromContext(ctx).Identity()
		if n.Recipient != currentUserID {
			return NotificationErrNotAllowedToRead()
		}

		raProps.setNotification(n)

		// Mark as unread
		n.ReadAt = nil
		now := time.Now()
		n.UpdatedAt = &now

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

		// Send the updated notification via websocket so client can update UI
		if svc.notificationSender != nil {
			if err = svc.notificationSender.Send("notification.unread", n, n.Recipient); err != nil {
				return err
			}
		}

		return nil
	}()

	return svc.recordAction(ctx, raProps, NotificationActionMarkAsUnread, err)
}

func (svc notification) MarkAllAsRead(ctx context.Context) (err error) {
	var (
		currentUserID = intAuth.GetIdentityFromContext(ctx).Identity()
		raProps       = &notificationActionProps{notification: &types.Notification{Recipient: currentUserID}}
	)

	err = func() (err error) {
		var (
			cursor *filter.PagingCursor
			nn     types.NotificationSet
			f      types.NotificationFilter
			now    = time.Now()
		)

		return store.Tx(ctx, svc.store, func(ctx context.Context, s store.Storer) error {
			// Process unread notifications in batches
			for {
				// Query unread notifications with pagination
				f = types.NotificationFilter{
					Recipient: currentUserID,
					Read:      filter.StateExcluded,
					Paging: filter.Paging{
						Limit:      100,
						PageCursor: cursor,
					},
				}

				nn, f, err = store.SearchNotifications(ctx, svc.store, f)
				if err != nil {
					return err
				}

				// Mark all notifications in this batch as read
				for _, n := range nn {
					n.ReadAt = &now
					n.UpdatedAt = &now
				}

				err = store.UpdateNotification(ctx, svc.store, nn...)
				if err != nil {
					return err
				}

				// Send the updated notifications via websocket so client can update UI
				if svc.notificationSender != nil && len(nn) > 0 {
					if err = svc.notificationSender.Send("notification.read.all", nn, currentUserID); err != nil {
						return err
					}
				}

				// Update cursor for next page or break if no more pages
				cursor = f.PageCursor
				if cursor == nil {
					break
				}
			}

			return nil
		})
	}()

	return svc.recordAction(ctx, raProps, NotificationActionMarkAllAsRead, err)
}

func (svc notification) MarkAllAsUnread(ctx context.Context) (err error) {
	var (
		currentUserID = intAuth.GetIdentityFromContext(ctx).Identity()
		raProps       = &notificationActionProps{notification: &types.Notification{Recipient: currentUserID}}
	)

	err = func() (err error) {
		var (
			cursor *filter.PagingCursor
			nn     types.NotificationSet
			f      types.NotificationFilter
			now    = time.Now()
		)

		return store.Tx(ctx, svc.store, func(ctx context.Context, s store.Storer) error {
			for {
				f = types.NotificationFilter{
					Recipient: currentUserID,
					Read:      filter.StateExclusive, // only read notifications
					Paging: filter.Paging{
						Limit:      100,
						PageCursor: cursor,
					},
				}

				nn, f, err = store.SearchNotifications(ctx, svc.store, f)
				if err != nil {
					return err
				}

				for _, n := range nn {
					n.ReadAt = nil
					n.UpdatedAt = &now
				}

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

				if svc.notificationSender != nil && len(nn) > 0 {
					if err = svc.notificationSender.Send("notification.unread.all", nn, currentUserID); err != nil {
						return err
					}
				}

				cursor = f.PageCursor
				if cursor == nil {
					break
				}
			}

			return nil
		})
	}()

	return svc.recordAction(ctx, raProps, NotificationActionMarkAllAsUnread, err)
}

func (svc notification) checkAssignee(ctx context.Context, n *types.Notification) (err error) {
	// Check if user is assigning to someone else
	if n.Recipient != svc.currentUser(ctx) {
		if !svc.ac.CanAssignNotification(ctx) {
			return NotificationErrNotAllowedToAssign()
		}
	}

	return nil
}

func (svc notification) currentUser(ctx context.Context) uint64 {
	return intAuth.GetIdentityFromContext(ctx).Identity()
}