More refactors of chat stuff

Author: Jeff Yanta

go.mod

@@ -6,7 +6,7 @@ require (
 	firebase.google.com/go/v4 v4.8.0
 	github.com/aws/aws-sdk-go-v2 v0.17.0
 	github.com/bits-and-blooms/bloom/v3 v3.1.0
-	github.com/code-payments/code-protobuf-api v1.16.7-0.20240618171829-bb587a7f24ce
+	github.com/code-payments/code-protobuf-api v1.16.7-0.20240619134635-bde7104d4467
 	github.com/emirpasic/gods v1.12.0
 	github.com/envoyproxy/protoc-gen-validate v1.0.4
 	github.com/golang-jwt/jwt/v5 v5.0.0

go.sum

@@ -121,8 +121,8 @@ github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWH
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I=
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
-github.com/code-payments/code-protobuf-api v1.16.7-0.20240618171829-bb587a7f24ce h1:wdPvuDHfEmymPn4zSaVD2ymK4pRKHlegAhTnl52d/IY=
-github.com/code-payments/code-protobuf-api v1.16.7-0.20240618171829-bb587a7f24ce/go.mod h1:pHQm75vydD6Cm2qHAzlimW6drysm489Z4tVxC2zHSsU=
+github.com/code-payments/code-protobuf-api v1.16.7-0.20240619134635-bde7104d4467 h1:BQgr91ASdCdxs8loz54TzmZQ0e6gDk1T7pkQxChJpdg=
+github.com/code-payments/code-protobuf-api v1.16.7-0.20240619134635-bde7104d4467/go.mod h1:pHQm75vydD6Cm2qHAzlimW6drysm489Z4tVxC2zHSsU=
 github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6 h1:NmTXa/uVnDyp0TY5MKi197+3HWcnYWfnHGyaFthlnGw=
 github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=

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

@@ -767,33 +767,15 @@ func (s *server) SendMessage(ctx context.Context, req *chatpb.SendMessageRequest
 	chatLock.Lock()
 	defer chatLock.Unlock()
 
-	messageId := chat.GenerateMessageId()
-	ts, _ := messageId.GetTimestamp()
-
-	chatMessage := &chatpb.ChatMessage{
-		MessageId: messageId.ToProto(),
-		SenderId:  req.MemberId,
-		Content:   req.Content,
-		Ts:        timestamppb.New(ts),
-		Cursor:    &chatpb.Cursor{Value: messageId[:]},
-	}
+	chatMessage := newProtoChatMessage(memberId, req.Content...)
 
 	err = s.persistChatMessage(ctx, chatId, chatMessage)
 	if err != nil {
 		log.WithError(err).Warn("failure persisting chat message")
 		return nil, status.Error(codes.Internal, "")
 	}
 
-	event := &chatpb.ChatStreamEvent{
-		Type: &chatpb.ChatStreamEvent_Message{
-			Message: chatMessage,
-		},
-	}
-	if err := s.asyncNotifyAll(chatId, memberId, event); err != nil {
-		log.WithError(err).Warn("failure notifying chat event")
-	}
-
-	// todo: send the push
+	s.onPersistChatMessage(log, chatId, chatMessage)
 
 	return &chatpb.SendMessageResponse{
 		Result:  chatpb.SendMessageResponse_OK,
@@ -946,7 +928,7 @@ func (s *server) AdvancePointer(ctx context.Context, req *chatpb.AdvancePointerR
 				Pointer: req.Pointer,
 			},
 		}
-		if err := s.asyncNotifyAll(chatId, memberId, event); err != nil {
+		if err := s.asyncNotifyAll(chatId, event); err != nil {
 			log.WithError(err).Warn("failure notifying chat event")
 		}
 	}
@@ -1061,20 +1043,17 @@ func (s *server) RevealIdentity(ctx context.Context, req *chatpb.RevealIdentityR
 	chatLock.Lock()
 	defer chatLock.Unlock()
 
-	messageId := chat.GenerateMessageId()
-	ts, _ := messageId.GetTimestamp()
-
-	chatMessage := &chatpb.ChatMessage{
-		MessageId: messageId.ToProto(),
-		SenderId:  req.MemberId,
-		Content: []*chatpb.Content{
-			{
-				Type: &chatpb.Content_IdentityRevealed{},
+	chatMessage := newProtoChatMessage(
+		memberId,
+		&chatpb.Content{
+			Type: &chatpb.Content_IdentityRevealed{
+				IdentityRevealed: &chatpb.IdentityRevealedContent{
+					MemberId: req.MemberId,
+					Identity: req.Identity,
+				},
 			},
 		},
-		Ts:     timestamppb.New(ts),
-		Cursor: &chatpb.Cursor{Value: messageId[:]},
-	}
+	)
 
 	err = s.data.ExecuteInTx(ctx, sql.LevelDefault, func(ctx context.Context) error {
 		err = s.data.UpgradeChatMemberIdentityV2(ctx, chatId, memberId, platform, req.Identity.Username)
@@ -1094,16 +1073,7 @@ func (s *server) RevealIdentity(ctx context.Context, req *chatpb.RevealIdentityR
 	})
 
 	if err == nil {
-		event := &chatpb.ChatStreamEvent{
-			Type: &chatpb.ChatStreamEvent_Message{
-				Message: chatMessage,
-			},
-		}
-		if err := s.asyncNotifyAll(chatId, memberId, event); err != nil {
-			log.WithError(err).Warn("failure notifying chat event")
-		}
-
-		// todo: send the push
+		s.onPersistChatMessage(log, chatId, chatMessage)
 	}
 
 	switch err {
@@ -1251,66 +1221,11 @@ func (s *server) SetSubscriptionState(ctx context.Context, req *chatpb.SetSubscr
 	}, nil
 }
 
-func (s *server) getProtoChatMessages(ctx context.Context, chatId chat.ChatId, owner *common.Account, queryOptions ...query.Option) ([]*chatpb.ChatMessage, error) {
-	messageRecords, err := s.data.GetAllChatMessagesV2(
-		ctx,
-		chatId,
-		queryOptions...,
-	)
-	if err == chat.ErrMessageNotFound {
-		return nil, err
-	}
-
-	var userLocale *language.Tag // Loaded lazily when required
-	var res []*chatpb.ChatMessage
-	for _, messageRecord := range messageRecords {
-		var protoChatMessage chatpb.ChatMessage
-		err = proto.Unmarshal(messageRecord.Data, &protoChatMessage)
-		if err != nil {
-			return nil, errors.Wrap(err, "error unmarshalling proto chat message")
-		}
-
-		ts, err := messageRecord.GetTimestamp()
-		if err != nil {
-			return nil, errors.Wrap(err, "error getting message timestamp")
-		}
-
-		for _, content := range protoChatMessage.Content {
-			switch typed := content.Type.(type) {
-			case *chatpb.Content_Localized:
-				if userLocale == nil {
-					loadedUserLocale, err := s.data.GetUserLocale(ctx, owner.PublicKey().ToBase58())
-					if err != nil {
-						return nil, errors.Wrap(err, "error getting user locale")
-					}
-					userLocale = &loadedUserLocale
-				}
-
-				typed.Localized.KeyOrText = localization.LocalizeWithFallback(
-					*userLocale,
-					localization.GetLocalizationKeyForUserAgent(ctx, typed.Localized.KeyOrText),
-					typed.Localized.KeyOrText,
-				)
-			}
-		}
-
-		protoChatMessage.MessageId = messageRecord.MessageId.ToProto()
-		if messageRecord.Sender != nil {
-			protoChatMessage.SenderId = messageRecord.Sender.ToProto()
-		}
-		protoChatMessage.Ts = timestamppb.New(ts)
-		protoChatMessage.Cursor = &chatpb.Cursor{Value: messageRecord.MessageId[:]}
-
-		res = append(res, &protoChatMessage)
-	}
-
-	return res, nil
-}
-
 func (s *server) toProtoChat(ctx context.Context, chatRecord *chat.ChatRecord, memberRecords []*chat.MemberRecord, myIdentitiesByPlatform map[chat.Platform]string) (*chatpb.ChatMetadata, error) {
 	protoChat := &chatpb.ChatMetadata{
 		ChatId: chatRecord.ChatId.ToProto(),
 		Kind:   chatRecord.ChatType.ToProto(),
+		Cursor: &chatpb.Cursor{Value: query.ToCursor(uint64(chatRecord.Id))},
 	}
 
 	switch chatRecord.ChatType {
@@ -1390,6 +1305,75 @@ func (s *server) toProtoChat(ctx context.Context, chatRecord *chat.ChatRecord, m
 	return protoChat, nil
 }
 
+func (s *server) getProtoChatMessages(ctx context.Context, chatId chat.ChatId, owner *common.Account, queryOptions ...query.Option) ([]*chatpb.ChatMessage, error) {
+	messageRecords, err := s.data.GetAllChatMessagesV2(
+		ctx,
+		chatId,
+		queryOptions...,
+	)
+	if err == chat.ErrMessageNotFound {
+		return nil, err
+	}
+
+	var userLocale *language.Tag // Loaded lazily when required
+	var res []*chatpb.ChatMessage
+	for _, messageRecord := range messageRecords {
+		var protoChatMessage chatpb.ChatMessage
+		err = proto.Unmarshal(messageRecord.Data, &protoChatMessage)
+		if err != nil {
+			return nil, errors.Wrap(err, "error unmarshalling proto chat message")
+		}
+
+		ts, err := messageRecord.GetTimestamp()
+		if err != nil {
+			return nil, errors.Wrap(err, "error getting message timestamp")
+		}
+
+		for _, content := range protoChatMessage.Content {
+			switch typed := content.Type.(type) {
+			case *chatpb.Content_Localized:
+				if userLocale == nil {
+					loadedUserLocale, err := s.data.GetUserLocale(ctx, owner.PublicKey().ToBase58())
+					if err != nil {
+						return nil, errors.Wrap(err, "error getting user locale")
+					}
+					userLocale = &loadedUserLocale
+				}
+
+				typed.Localized.KeyOrText = localization.LocalizeWithFallback(
+					*userLocale,
+					localization.GetLocalizationKeyForUserAgent(ctx, typed.Localized.KeyOrText),
+					typed.Localized.KeyOrText,
+				)
+			}
+		}
+
+		protoChatMessage.MessageId = messageRecord.MessageId.ToProto()
+		if messageRecord.Sender != nil {
+			protoChatMessage.SenderId = messageRecord.Sender.ToProto()
+		}
+		protoChatMessage.Ts = timestamppb.New(ts)
+		protoChatMessage.Cursor = &chatpb.Cursor{Value: messageRecord.MessageId[:]}
+
+		res = append(res, &protoChatMessage)
+	}
+
+	return res, nil
+}
+
+func (s *server) onPersistChatMessage(log *logrus.Entry, chatId chat.ChatId, chatMessage *chatpb.ChatMessage) {
+	event := &chatpb.ChatStreamEvent{
+		Type: &chatpb.ChatStreamEvent_Message{
+			Message: chatMessage,
+		},
+	}
+	if err := s.asyncNotifyAll(chatId, event); err != nil {
+		log.WithError(err).Warn("failure notifying chat event")
+	}
+
+	// todo: send the push
+}
+
 func (s *server) getAllIdentities(ctx context.Context, owner *common.Account) (map[chat.Platform]string, error) {
 	identities := map[chat.Platform]string{
 		chat.PlatformCode: owner.PublicKey().ToBase58(),
@@ -1466,3 +1450,16 @@ func (s *server) getOwnedTwitterUsername(ctx context.Context, owner *common.Acco
 		return "", false, errors.Wrap(err, "error getting twitter user")
 	}
 }
+
+func newProtoChatMessage(sender chat.MemberId, content ...*chatpb.Content) *chatpb.ChatMessage {
+	messageId := chat.GenerateMessageId()
+	ts, _ := messageId.GetTimestamp()
+
+	return &chatpb.ChatMessage{
+		MessageId: messageId.ToProto(),
+		SenderId:  sender.ToProto(),
+		Content:   content,
+		Ts:        timestamppb.New(ts),
+		Cursor:    &chatpb.Cursor{Value: messageId[:]},
+	}
+}

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

@@ -93,15 +93,14 @@ func boundedStreamChatEventsRecv(
 }
 
 type chatEventNotification struct {
-	chatId   chat.ChatId
-	memberId chat.MemberId
-	event    *chatpb.ChatStreamEvent
-	ts       time.Time
+	chatId chat.ChatId
+	event  *chatpb.ChatStreamEvent
+	ts     time.Time
 }
 
-func (s *server) asyncNotifyAll(chatId chat.ChatId, memberId chat.MemberId, event *chatpb.ChatStreamEvent) error {
+func (s *server) asyncNotifyAll(chatId chat.ChatId, event *chatpb.ChatStreamEvent) error {
 	m := proto.Clone(event).(*chatpb.ChatStreamEvent)
-	ok := s.chatEventChans.Send(chatId[:], &chatEventNotification{chatId, memberId, m, time.Now()})
+	ok := s.chatEventChans.Send(chatId[:], &chatEventNotification{chatId, m, time.Now()})
 	if !ok {
 		return errors.New("chat event channel is full")
 	}