Have chat v2 store indicate if pointer was advanced

Author: Jeff Yanta

pkg/code/data/chat/v2/memory/store.go

@@ -201,19 +201,19 @@ func (s *store) PutMessage(_ context.Context, record *chat.MessageRecord) error
 }
 
 // AdvancePointer implements chat.Store.AdvancePointer
-func (s *store) AdvancePointer(_ context.Context, chatId chat.ChatId, memberId chat.MemberId, pointerType chat.PointerType, pointer chat.MessageId) error {
+func (s *store) AdvancePointer(_ context.Context, chatId chat.ChatId, memberId chat.MemberId, pointerType chat.PointerType, pointer chat.MessageId) (bool, error) {
 	switch pointerType {
 	case chat.PointerTypeDelivered, chat.PointerTypeRead:
 	default:
-		return chat.ErrInvalidPointerType
+		return false, chat.ErrInvalidPointerType
 	}
 
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	item := s.findMemberById(chatId, memberId)
 	if item == nil {
-		return chat.ErrMemberNotFound
+		return false, chat.ErrMemberNotFound
 	}
 
 	var currentPointer *chat.MessageId
@@ -224,20 +224,19 @@ func (s *store) AdvancePointer(_ context.Context, chatId chat.ChatId, memberId c
 		currentPointer = item.ReadPointer
 	}
 
-	if currentPointer != nil && currentPointer.After(pointer) {
-		return nil
-	}
+	if currentPointer == nil || currentPointer.Before(pointer) {
+		switch pointerType {
+		case chat.PointerTypeDelivered:
+			cloned := pointer.Clone()
+			item.DeliveryPointer = &cloned
+		case chat.PointerTypeRead:
+			cloned := pointer.Clone()
+			item.ReadPointer = &cloned
+		}
 
-	switch pointerType {
-	case chat.PointerTypeDelivered:
-		cloned := pointer.Clone()
-		item.DeliveryPointer = &cloned
-	case chat.PointerTypeRead:
-		cloned := pointer.Clone()
-		item.ReadPointer = &cloned
+		return true, nil
 	}
-
-	return nil
+	return false, nil
 }
 
 // SetMuteState implements chat.Store.SetMuteState

pkg/code/data/chat/v2/store.go

@@ -55,7 +55,7 @@ type Store interface {
 	PutMessage(ctx context.Context, record *MessageRecord) error
 
 	// AdvancePointer advances a chat pointer for a chat member
-	AdvancePointer(ctx context.Context, chatId ChatId, memberId MemberId, pointerType PointerType, pointer MessageId) error
+	AdvancePointer(ctx context.Context, chatId ChatId, memberId MemberId, pointerType PointerType, pointer MessageId) (bool, error)
 
 	// SetMuteState updates the mute state for a chat member
 	SetMuteState(ctx context.Context, chatId ChatId, memberId MemberId, isMuted bool) error

pkg/code/data/internal.go

@@ -406,7 +406,7 @@ type DatabaseData interface {
 	PutChatV2(ctx context.Context, record *chat_v2.ChatRecord) error
 	PutChatMemberV2(ctx context.Context, record *chat_v2.MemberRecord) error
 	PutChatMessageV2(ctx context.Context, record *chat_v2.MessageRecord) error
-	AdvanceChatPointerV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, pointerType chat_v2.PointerType, pointer chat_v2.MessageId) error
+	AdvanceChatPointerV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, pointerType chat_v2.PointerType, pointer chat_v2.MessageId) (bool, error)
 	SetChatMuteStateV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, isMuted bool) error
 	SetChatSubscriptionStateV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, isSubscribed bool) error
 
@@ -1504,7 +1504,7 @@ func (dp *DatabaseProvider) PutChatMemberV2(ctx context.Context, record *chat_v2
 func (dp *DatabaseProvider) PutChatMessageV2(ctx context.Context, record *chat_v2.MessageRecord) error {
 	return dp.chatv2.PutMessage(ctx, record)
 }
-func (dp *DatabaseProvider) AdvanceChatPointerV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, pointerType chat_v2.PointerType, pointer chat_v2.MessageId) error {
+func (dp *DatabaseProvider) AdvanceChatPointerV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, pointerType chat_v2.PointerType, pointer chat_v2.MessageId) (bool, error) {
 	return dp.chatv2.AdvancePointer(ctx, chatId, memberId, pointerType, pointer)
 }
 func (dp *DatabaseProvider) SetChatMuteStateV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, isMuted bool) error {

pkg/code/server/grpc/chat/v2/server.go

@@ -851,20 +851,21 @@ func (s *server) AdvancePointer(ctx context.Context, req *chatpb.AdvancePointerR
 		return nil, status.Error(codes.Internal, "")
 	}
 
-	// Note: Guarantees that pointer will never be advanced to some point in the past
-	err = s.data.AdvanceChatPointerV2(ctx, chatId, memberId, pointerType, pointerValue)
+	isAdvanced, err := s.data.AdvanceChatPointerV2(ctx, chatId, memberId, pointerType, pointerValue)
 	if err != nil {
 		log.WithError(err).Warn("failure advancing chat pointer")
 		return nil, status.Error(codes.Internal, "")
 	}
 
-	event := &chatpb.ChatStreamEvent{
-		Type: &chatpb.ChatStreamEvent_Pointer{
-			Pointer: req.Pointer,
-		},
-	}
-	if err := s.asyncNotifyAll(chatId, memberId, event); err != nil {
-		log.WithError(err).Warn("failure notifying chat event")
+	if isAdvanced {
+		event := &chatpb.ChatStreamEvent{
+			Type: &chatpb.ChatStreamEvent_Pointer{
+				Pointer: req.Pointer,
+			},
+		}
+		if err := s.asyncNotifyAll(chatId, memberId, event); err != nil {
+			log.WithError(err).Warn("failure notifying chat event")
+		}
 	}
 
 	return &chatpb.AdvancePointerResponse{