Flush pointers on chat event stream open

Author: Jeff Yanta

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

@@ -71,8 +71,17 @@ func (s *store) GetMessageById(_ context.Context, chatId chat.ChatId, messageId
 	return &cloned, nil
 }
 
-// GetAllMessagesByChat implements chat.Store.GetAllMessagesByChat
-func (s *store) GetAllMessagesByChat(_ context.Context, chatId chat.ChatId, cursor query.Cursor, direction query.Ordering, limit uint64) ([]*chat.MessageRecord, error) {
+// GetAllMembersByChatId implements chat.Store.GetAllMembersByChatId
+func (s *store) GetAllMembersByChatId(_ context.Context, chatId chat.ChatId) ([]*chat.MemberRecord, error) {
+	items := s.findMembersByChatId(chatId)
+	if len(items) == 0 {
+		return nil, chat.ErrMemberNotFound
+	}
+	return cloneMemberRecords(items), nil
+}
+
+// GetAllMessagesByChatId implements chat.Store.GetAllMessagesByChatId
+func (s *store) GetAllMessagesByChatId(_ context.Context, chatId chat.ChatId, cursor query.Cursor, direction query.Ordering, limit uint64) ([]*chat.MessageRecord, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -81,10 +90,10 @@ func (s *store) GetAllMessagesByChat(_ context.Context, chatId chat.ChatId, curs
 	if err != nil {
 		return nil, err
 	}
+
 	if len(items) == 0 {
 		return nil, chat.ErrMessageNotFound
 	}
-
 	return cloneMessageRecords(items), nil
 }
 
@@ -275,6 +284,16 @@ func (s *store) findMemberById(chatId chat.ChatId, memberId chat.MemberId) *chat
 	return nil
 }
 
+func (s *store) findMembersByChatId(chatId chat.ChatId) []*chat.MemberRecord {
+	var res []*chat.MemberRecord
+	for _, item := range s.memberRecords {
+		if bytes.Equal(chatId[:], item.ChatId[:]) {
+			res = append(res, item)
+		}
+	}
+	return res
+}
+
 func (s *store) findMessage(data *chat.MessageRecord) *chat.MessageRecord {
 	for _, item := range s.messageRecords {
 		if data.Id == item.Id {
@@ -362,6 +381,15 @@ func (s *store) reset() {
 	s.lastMessageId = 0
 }
 
+func cloneMemberRecords(items []*chat.MemberRecord) []*chat.MemberRecord {
+	res := make([]*chat.MemberRecord, len(items))
+	for i, item := range items {
+		cloned := item.Clone()
+		res[i] = &cloned
+	}
+	return res
+}
+
 func cloneMessageRecords(items []*chat.MessageRecord) []*chat.MessageRecord {
 	res := make([]*chat.MessageRecord, len(items))
 	for i, item := range items {

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

@@ -119,6 +119,20 @@ func GetPointerTypeFromProto(proto chatpb.Pointer_Kind) PointerType {
 	}
 }
 
+// ToProto returns the proto representation of the pointer type
+func (p PointerType) ToProto() chatpb.Pointer_Kind {
+	switch p {
+	case PointerTypeSent:
+		return chatpb.Pointer_SENT
+	case PointerTypeDelivered:
+		return chatpb.Pointer_DELIVERED
+	case PointerTypeRead:
+		return chatpb.Pointer_READ
+	default:
+		return chatpb.Pointer_UNKNOWN
+	}
+}
+
 // String returns the string representation of the pointer type
 func (p PointerType) String() string {
 	switch p {

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

@@ -28,10 +28,15 @@ type Store interface {
 	// GetMessageById gets a chat message by the chat and message IDs
 	GetMessageById(ctx context.Context, chatId ChatId, messageId MessageId) (*MessageRecord, error)
 
-	// GetAllMessagesByChat gets all messages for a given chat
+	// GetAllMembersByChatId gets all members for a given chat
+	//
+	// todo: Add paging when we introduce group chats
+	GetAllMembersByChatId(ctx context.Context, chatId ChatId) ([]*MemberRecord, error)
+
+	// GetAllMessagesByChatId gets all messages for a given chat
 	//
 	// Note: Cursor is a message ID
-	GetAllMessagesByChat(ctx context.Context, chatId ChatId, cursor query.Cursor, direction query.Ordering, limit uint64) ([]*MessageRecord, error)
+	GetAllMessagesByChatId(ctx context.Context, chatId ChatId, cursor query.Cursor, direction query.Ordering, limit uint64) ([]*MessageRecord, error)
 
 	// PutChat creates a new chat
 	PutChat(ctx context.Context, record *ChatRecord) error

pkg/code/data/internal.go

@@ -399,6 +399,7 @@ type DatabaseData interface {
 	GetChatByIdV2(ctx context.Context, chatId chat_v2.ChatId) (*chat_v2.ChatRecord, error)
 	GetChatMemberByIdV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId) (*chat_v2.MemberRecord, error)
 	GetChatMessageByIdV2(ctx context.Context, chatId chat_v2.ChatId, messageId chat_v2.MessageId) (*chat_v2.MessageRecord, error)
+	GetAllChatMembersV2(ctx context.Context, chatId chat_v2.ChatId) ([]*chat_v2.MemberRecord, error)
 	GetAllChatMessagesV2(ctx context.Context, chatId chat_v2.ChatId, opts ...query.Option) ([]*chat_v2.MessageRecord, error)
 	PutChatV2(ctx context.Context, record *chat_v2.ChatRecord) error
 	PutChatMemberV2(ctx context.Context, record *chat_v2.MemberRecord) error
@@ -1472,12 +1473,15 @@ func (dp *DatabaseProvider) GetChatMemberByIdV2(ctx context.Context, chatId chat
 func (dp *DatabaseProvider) GetChatMessageByIdV2(ctx context.Context, chatId chat_v2.ChatId, messageId chat_v2.MessageId) (*chat_v2.MessageRecord, error) {
 	return dp.chatv2.GetMessageById(ctx, chatId, messageId)
 }
+func (dp *DatabaseProvider) GetAllChatMembersV2(ctx context.Context, chatId chat_v2.ChatId) ([]*chat_v2.MemberRecord, error) {
+	return dp.chatv2.GetAllMembersByChatId(ctx, chatId)
+}
 func (dp *DatabaseProvider) GetAllChatMessagesV2(ctx context.Context, chatId chat_v2.ChatId, opts ...query.Option) ([]*chat_v2.MessageRecord, error) {
 	req, err := query.DefaultPaginationHandler(opts...)
 	if err != nil {
 		return nil, err
 	}
-	return dp.chatv2.GetAllMessagesByChat(ctx, chatId, req.Cursor, req.SortBy, req.Limit)
+	return dp.chatv2.GetAllMessagesByChatId(ctx, chatId, req.Cursor, req.SortBy, req.Limit)
 }
 func (dp *DatabaseProvider) PutChatV2(ctx context.Context, record *chat_v2.ChatRecord) error {
 	return dp.chatv2.PutChat(ctx, record)

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

@@ -299,8 +299,8 @@ func (s *server) StreamChatEvents(streamer chatpb.Chat_StreamChatEventsServer) e
 	sendPingCh := time.After(0)
 	streamHealthCh := monitorChatEventStreamHealth(ctx, log, streamRef, streamer)
 
-	// todo: We should also "flush" pointers for each chat member
 	go s.flushMessages(ctx, chatId, owner, stream)
+	go s.flushPointers(ctx, chatId, stream)
 
 	for {
 		select {
@@ -385,6 +385,49 @@ func (s *server) flushMessages(ctx context.Context, chatId chat.ChatId, owner *c
 	}
 }
 
+func (s *server) flushPointers(ctx context.Context, chatId chat.ChatId, stream *chatEventStream) {
+	log := s.log.WithFields(logrus.Fields{
+		"method":  "flushPointers",
+		"chat_id": chatId.String(),
+	})
+
+	memberRecords, err := s.data.GetAllChatMembersV2(ctx, chatId)
+	if err == chat.ErrMemberNotFound {
+		return
+	} else if err != nil {
+		log.WithError(err).Warn("failure getting chat members")
+		return
+	}
+
+	for _, memberRecord := range memberRecords {
+		for _, optionalPointer := range []struct {
+			kind  chat.PointerType
+			value *chat.MessageId
+		}{
+			{chat.PointerTypeDelivered, memberRecord.DeliveryPointer},
+			{chat.PointerTypeRead, memberRecord.ReadPointer},
+		} {
+			if optionalPointer.value == nil {
+				continue
+			}
+
+			event := &chatpb.ChatStreamEvent{
+				Type: &chatpb.ChatStreamEvent_Pointer{
+					Pointer: &chatpb.Pointer{
+						Kind:     optionalPointer.kind.ToProto(),
+						Value:    optionalPointer.value.ToProto(),
+						MemberId: memberRecord.MemberId.ToProto(),
+					},
+				},
+			}
+			if err := stream.notify(event, streamNotifyTimeout); err != nil {
+				log.WithError(err).Warnf("failed to notify session stream, closing streamer (stream=%p)", stream)
+				return
+			}
+		}
+	}
+}
+
 func (s *server) SendMessage(ctx context.Context, req *chatpb.SendMessageRequest) (*chatpb.SendMessageResponse, error) {
 	log := s.log.WithField("method", "SendMessage")
 	log = client.InjectLoggingMetadata(ctx, log)