chats: add (temporary) plumbing for tip notifications.

Adds enough plumbing for the tip notifications to flow through v2
(without breaking the previous system, hopefully). When the protocol
is more flushed out, and the base messaging infra is shored up a bit,
this will likely be revisted

Author: mfycheng

pkg/code/async/geyser/messenger.go

@@ -165,7 +165,7 @@ func processPotentialBlockchainMessage(ctx context.Context, data code_data.Provi
 				return errors.Wrap(err, "error creating proto message")
 			}
 
-			canPush, err := chat_util.SendChatMessage(
+			canPush, err := chat_util.SendNotificationChatMessageV1(
 				ctx,
 				data,
 				asciiBaseDomain,

pkg/code/chat/message_cash_transactions.go

@@ -148,7 +148,7 @@ func SendCashTransactionsExchangeMessage(ctx context.Context, data code_data.Pro
 			return errors.Wrap(err, "error creating proto chat message")
 		}
 
-		_, err = SendChatMessage(
+		_, err = SendNotificationChatMessageV1(
 			ctx,
 			data,
 			CashTransactionsName,

pkg/code/chat/message_code_team.go

@@ -16,7 +16,7 @@ import (
 
 // SendCodeTeamMessage sends a message to the Code Team chat.
 func SendCodeTeamMessage(ctx context.Context, data code_data.Provider, receiver *common.Account, chatMessage *chatpb.ChatMessage) (bool, error) {
-	return SendChatMessage(
+	return SendNotificationChatMessageV1(
 		ctx,
 		data,
 		CodeTeamName,

pkg/code/chat/message_kin_purchases.go

@@ -23,7 +23,7 @@ func GetKinPurchasesChatId(owner *common.Account) chat_v1.ChatId {
 
 // SendKinPurchasesMessage sends a message to the Kin Purchases chat.
 func SendKinPurchasesMessage(ctx context.Context, data code_data.Provider, receiver *common.Account, chatMessage *chatpb.ChatMessage) (bool, error) {
-	return SendChatMessage(
+	return SendNotificationChatMessageV1(
 		ctx,
 		data,
 		KinPurchasesName,

pkg/code/chat/message_merchant.go

@@ -161,7 +161,7 @@ func SendMerchantExchangeMessage(ctx context.Context, data code_data.Provider, i
 			return nil, errors.Wrap(err, "error creating proto chat message")
 		}
 
-		canPush, err := SendChatMessage(
+		canPush, err := SendNotificationChatMessageV1(
 			ctx,
 			data,
 			chatTitle,

pkg/code/chat/message_tips.go

@@ -2,23 +2,36 @@ package chat
 
 import (
 	"context"
+	"fmt"
 
+	"github.com/mr-tron/base58"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	chatpb "github.com/code-payments/code-protobuf-api/generated/go/chat/v1"
+	chatv2pb "github.com/code-payments/code-protobuf-api/generated/go/chat/v2"
+	commonpb "github.com/code-payments/code-protobuf-api/generated/go/common/v1"
 
 	"github.com/code-payments/code-server/pkg/code/common"
 	code_data "github.com/code-payments/code-server/pkg/code/data"
 	chat_v1 "github.com/code-payments/code-server/pkg/code/data/chat/v1"
+	chat_v2 "github.com/code-payments/code-server/pkg/code/data/chat/v2"
 	"github.com/code-payments/code-server/pkg/code/data/intent"
+	chat_server "github.com/code-payments/code-server/pkg/code/server/grpc/chat/v2"
 )
 
 // SendTipsExchangeMessage sends a message to the Tips chat with exchange data
 // content related to the submitted intent. Intents that don't belong in the
 // Tips chat will be ignored.
 //
 // Note: Tests covered in SubmitIntent history tests
-func SendTipsExchangeMessage(ctx context.Context, data code_data.Provider, intentRecord *intent.Record) ([]*MessageWithOwner, error) {
+func SendTipsExchangeMessage(ctx context.Context, data code_data.Provider, notifier chat_server.Notifier, intentRecord *intent.Record) ([]*MessageWithOwner, error) {
+	intentIdRaw, err := base58.Decode(intentRecord.IntentId)
+	if err != nil {
+		return nil, fmt.Errorf("invalid intent id: %w", err)
+	}
+
 	messageId := intentRecord.IntentId
 
 	exchangeData, ok := getExchangeDataFromIntent(intentRecord)
@@ -30,7 +43,6 @@ func SendTipsExchangeMessage(ctx context.Context, data code_data.Provider, inten
 	switch intentRecord.IntentType {
 	case intent.SendPrivatePayment:
 		if !intentRecord.SendPrivatePaymentMetadata.IsTip {
-			// Not a tip
 			return nil, nil
 		}
 
@@ -61,30 +73,68 @@ func SendTipsExchangeMessage(ctx context.Context, data code_data.Provider, inten
 				},
 			},
 		}
-		protoMessage, err := newProtoChatMessage(messageId, content, intentRecord.CreatedAt)
+
+		v1Message, err := newProtoChatMessage(messageId, content, intentRecord.CreatedAt)
 		if err != nil {
 			return nil, errors.Wrap(err, "error creating proto chat message")
 		}
 
-		canPush, err := SendChatMessage(
+		v2Message := &chatv2pb.ChatMessage{
+			MessageId: chat_v2.GenerateMessageId().ToProto(),
+			Content: []*chatv2pb.Content{
+				{
+					Type: &chatv2pb.Content_ExchangeData{
+						ExchangeData: &chatv2pb.ExchangeDataContent{
+							Verb: chatv2pb.ExchangeDataContent_Verb(verb),
+							ExchangeData: &chatv2pb.ExchangeDataContent_Exact{
+								Exact: exchangeData,
+							},
+							Reference: &chatv2pb.ExchangeDataContent_Intent{
+								Intent: &commonpb.IntentId{Value: intentIdRaw},
+							},
+						},
+					},
+				},
+			},
+			Ts: timestamppb.New(intentRecord.CreatedAt),
+		}
+
+		canPush, err := SendNotificationChatMessageV1(
 			ctx,
 			data,
 			TipsName,
 			chat_v1.ChatTypeInternal,
 			true,
 			receiver,
-			protoMessage,
+			v1Message,
 			verb != chatpb.ExchangeDataContent_RECEIVED_TIP,
 		)
-		if err != nil && err != chat_v1.ErrMessageAlreadyExists {
-			return nil, errors.Wrap(err, "error persisting chat message")
+		if err != nil && !errors.Is(err, chat_v1.ErrMessageAlreadyExists) {
+			return nil, errors.Wrap(err, "error persisting v1 chat message")
+		}
+
+		_, err = SendNotificationChatMessageV2(
+			ctx,
+			data,
+			notifier,
+			TipsName,
+			true,
+			receiver,
+			v2Message,
+			intentRecord.IntentId,
+			verb != chatpb.ExchangeDataContent_RECEIVED_TIP,
+		)
+		if err != nil {
+			// TODO: Eventually we'll want to return an error, but for now we'll log
+			//       since we're not in 'prod' yet.
+			logrus.StandardLogger().WithError(err).Warn("Failed to send notification message (v2)")
 		}
 
 		if canPush {
 			messagesToPush = append(messagesToPush, &MessageWithOwner{
 				Owner:   receiver,
 				Title:   TipsName,
-				Message: protoMessage,
+				Message: v1Message,
 			})
 		}
 	}

pkg/code/chat/sender.go

@@ -2,25 +2,30 @@ package chat
 
 import (
 	"context"
+	"database/sql"
 	"errors"
+	"fmt"
 	"time"
 
 	"github.com/mr-tron/base58"
 	"google.golang.org/protobuf/proto"
 
 	chatpb "github.com/code-payments/code-protobuf-api/generated/go/chat/v1"
+	chatv2pb "github.com/code-payments/code-protobuf-api/generated/go/chat/v2"
 
 	"github.com/code-payments/code-server/pkg/code/common"
 	code_data "github.com/code-payments/code-server/pkg/code/data"
 	chat_v1 "github.com/code-payments/code-server/pkg/code/data/chat/v1"
+	chat_v2 "github.com/code-payments/code-server/pkg/code/data/chat/v2"
+	chatserver "github.com/code-payments/code-server/pkg/code/server/grpc/chat/v2"
 )
 
-// SendChatMessage sends a chat message to a receiving owner account.
+// SendNotificationChatMessageV1 sends a chat message to a receiving owner account.
 //
 // Note: This function is not responsible for push notifications. This method
 // might be called within the context of a DB transaction, which might have
-// unrelated failures. A hint as to whether a push should be sent is provided.
-func SendChatMessage(
+// unrelated failures. A hint whether a push should be sent is provided.
+func SendNotificationChatMessageV1(
 	ctx context.Context,
 	data code_data.Provider,
 	chatTitle string,
@@ -80,7 +85,7 @@ func SendChatMessage(
 		}
 
 		err = data.PutChatV1(ctx, chatRecord)
-		if err != nil && err != chat_v1.ErrChatAlreadyExists {
+		if err != nil && !errors.Is(err, chat_v1.ErrChatAlreadyExists) {
 			return false, err
 		}
 	default:
@@ -115,3 +120,131 @@ func SendChatMessage(
 
 	return canPushMessage, nil
 }
+
+func SendNotificationChatMessageV2(
+	ctx context.Context,
+	data code_data.Provider,
+	notifier chatserver.Notifier,
+	chatTitle string,
+	isVerifiedChat bool,
+	receiver *common.Account,
+	protoMessage *chatv2pb.ChatMessage,
+	intentId string,
+	isSilentMessage bool,
+) (canPushMessage bool, err error) {
+	chatId := chat_v2.GetChatId(chatTitle, receiver.PublicKey().ToBase58(), isVerifiedChat)
+
+	if protoMessage.Cursor != nil {
+		// Let the utilities and GetMessages RPC handle cursors
+		return false, errors.New("cursor must not be set")
+	}
+
+	if err := protoMessage.Validate(); err != nil {
+		return false, err
+	}
+
+	messageId, err := chat_v2.GetMessageIdFromProto(protoMessage.MessageId)
+	if err != nil {
+		return false, fmt.Errorf("invalid message id: %w", err)
+	}
+
+	// Clear out extracted metadata as a space optimization
+	cloned := proto.Clone(protoMessage).(*chatv2pb.ChatMessage)
+	cloned.MessageId = nil
+	cloned.Ts = nil
+	cloned.Cursor = nil
+
+	marshalled, err := proto.Marshal(cloned)
+	if err != nil {
+		return false, err
+	}
+
+	canPersistMessage := true
+	canPushMessage = !isSilentMessage
+
+	//
+	// Step 1: Check to see if we need to create the chat.
+	//
+	_, err = data.GetChatByIdV2(ctx, chatId)
+	if errors.Is(err, chat_v2.ErrChatNotFound) {
+		chatRecord := &chat_v2.ChatRecord{
+			ChatId:     chatId,
+			ChatType:   chat_v2.ChatTypeNotification,
+			ChatTitle:  &chatTitle,
+			IsVerified: isVerifiedChat,
+
+			CreatedAt: time.Now(),
+		}
+
+		err = data.ExecuteInTx(ctx, sql.LevelDefault, func(ctx context.Context) error {
+			err = data.PutChatV2(ctx, chatRecord)
+			if err != nil && !errors.Is(err, chat_v2.ErrChatExists) {
+				return fmt.Errorf("failed to initialize chat: %w", err)
+			}
+
+			err = data.PutChatMemberV2(ctx, &chat_v2.MemberRecord{
+				ChatId:     chatId,
+				MemberId:   chat_v2.GenerateMemberId(),
+				Platform:   chat_v2.PlatformCode,
+				PlatformId: receiver.PublicKey().ToBase58(),
+				JoinedAt:   time.Now(),
+			})
+			if err != nil {
+				return fmt.Errorf("failed to initialize chat with member: %w", err)
+			}
+
+			return nil
+		})
+		if err != nil {
+			return false, err
+		}
+	} else if err != nil {
+		return false, err
+	}
+
+	//
+	// Step 2: Ensure that there is exactly 1 member in the chat.
+	//
+	members, err := data.GetAllChatMembersV2(ctx, chatId)
+	if errors.Is(err, chat_v2.ErrMemberNotFound) { // TODO: This is a weird error...
+		return false, nil
+	} else if err != nil {
+		return false, err
+	}
+	if len(members) > 1 {
+		// TODO: This _could_ get weird if client or someone else decides to join as another member.
+		return false, errors.New("notification chat should have at most 1 member")
+	}
+
+	canPersistMessage = !members[0].IsUnsubscribed
+	canPushMessage = canPushMessage && canPersistMessage && !members[0].IsMuted
+
+	if canPersistMessage {
+		refType := chat_v2.ReferenceTypeIntent
+		messageRecord := &chat_v2.MessageRecord{
+			ChatId:    chatId,
+			MessageId: messageId,
+
+			Data:     marshalled,
+			IsSilent: isSilentMessage,
+
+			ReferenceType: &refType,
+			Reference:     &intentId,
+		}
+
+		// TODO: Once we have a better idea on the data modeling around chatv2,
+		//       we may wish to have the server manage the creation of messages
+		//       (and chats?) as well. That would also put the
+		err = data.PutChatMessageV2(ctx, messageRecord)
+		if err != nil {
+			return false, err
+		}
+
+		notifier.NotifyMessage(ctx, chatId, protoMessage)
+	}
+
+	// TODO: Once we move more things over to chatv2, we will need to increment
+	//       badge count here. We don't currently, as it would result in a double
+	//       push.
+	return canPushMessage, nil
+}

pkg/code/chat/sender_test.go

@@ -33,9 +33,8 @@ func TestSendChatMessage_HappyPath(t *testing.T) {
 		chatMessage := newRandomChatMessage(t, i+1)
 		expectedBadgeCount += 1
 
-		canPush, err := SendChatMessage(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, false)
+		canPush, err := SendNotificationChatMessageV1(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, false)
 		require.NoError(t, err)
-
 		assert.True(t, canPush)
 
 		assert.NotNil(t, chatMessage.MessageId)
@@ -56,7 +55,7 @@ func TestSendChatMessage_VerifiedChat(t *testing.T) {
 
 	for _, isVerified := range []bool{true, false} {
 		chatMessage := newRandomChatMessage(t, 1)
-		_, err := SendChatMessage(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, isVerified, receiver, chatMessage, true)
+		_, err := SendNotificationChatMessageV1(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, isVerified, receiver, chatMessage, true)
 		require.NoError(t, err)
 		env.assertChatRecordSaved(t, chatTitle, receiver, isVerified)
 	}
@@ -71,7 +70,7 @@ func TestSendChatMessage_SilentMessage(t *testing.T) {
 
 	for i, isSilent := range []bool{true, false} {
 		chatMessage := newRandomChatMessage(t, 1)
-		canPush, err := SendChatMessage(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, isSilent)
+		canPush, err := SendNotificationChatMessageV1(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, isSilent)
 		require.NoError(t, err)
 		assert.Equal(t, !isSilent, canPush)
 		env.assertChatMessageRecordSaved(t, chatId, chatMessage, isSilent)
@@ -92,7 +91,7 @@ func TestSendChatMessage_MuteState(t *testing.T) {
 		}
 
 		chatMessage := newRandomChatMessage(t, 1)
-		canPush, err := SendChatMessage(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, false)
+		canPush, err := SendNotificationChatMessageV1(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, false)
 		require.NoError(t, err)
 		assert.Equal(t, !isMuted, canPush)
 		env.assertChatMessageRecordSaved(t, chatId, chatMessage, false)
@@ -113,7 +112,7 @@ func TestSendChatMessage_SubscriptionState(t *testing.T) {
 		}
 
 		chatMessage := newRandomChatMessage(t, 1)
-		canPush, err := SendChatMessage(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, false)
+		canPush, err := SendNotificationChatMessageV1(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, false)
 		require.NoError(t, err)
 		assert.Equal(t, !isUnsubscribed, canPush)
 		if isUnsubscribed {
@@ -135,7 +134,7 @@ func TestSendChatMessage_InvalidProtoMessage(t *testing.T) {
 	chatMessage := newRandomChatMessage(t, 1)
 	chatMessage.Content = nil
 
-	canPush, err := SendChatMessage(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, false)
+	canPush, err := SendNotificationChatMessageV1(env.ctx, env.data, chatTitle, chat_v1.ChatTypeInternal, true, receiver, chatMessage, false)
 	assert.Error(t, err)
 	assert.False(t, canPush)
 	env.assertChatRecordNotSaved(t, chatId)