chat/v2: remove tip hooks, sketch out latest rpcs

Author: mfycheng

go.mod

@@ -42,6 +42,7 @@ require (
 	golang.org/x/crypto v0.21.0
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842
 	golang.org/x/net v0.22.0
+	golang.org/x/sync v0.7.0
 	golang.org/x/text v0.14.0
 	golang.org/x/time v0.5.0
 	google.golang.org/api v0.170.0
@@ -120,7 +121,6 @@ require (
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	golang.org/x/oauth2 v0.18.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/appengine/v2 v2.0.1 // indirect

pkg/code/chat/message_tips.go

@@ -2,22 +2,12 @@ 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"
+	"github.com/pkg/errors"
 )
 
 // SendTipsExchangeMessage sends a message to the Tips chat with exchange data
@@ -26,11 +16,6 @@ import (
 //
 // Note: Tests covered in SubmitIntent history tests
 func SendTipsExchangeMessage(ctx context.Context, data code_data.Provider, notifier 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)
@@ -78,26 +63,6 @@ func SendTipsExchangeMessage(ctx context.Context, data code_data.Provider, notif
 			return nil, errors.Wrap(err, "error creating proto chat message")
 		}
 
-		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,
@@ -112,23 +77,6 @@ func SendTipsExchangeMessage(ctx context.Context, data code_data.Provider, notif
 			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,

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

@@ -100,12 +100,13 @@ func (s *store) GetAllMembersByPlatformIds(_ context.Context, idByPlatform map[c
 }
 
 // GetUnreadCount implements chat.store.GetUnreadCount
-func (s *store) GetUnreadCount(_ context.Context, chatId chat.ChatId, readPointer chat.MessageId) (uint32, error) {
+func (s *store) GetUnreadCount(_ context.Context, chatId chat.ChatId, memberId chat.MemberId, readPointer chat.MessageId) (uint32, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	items := s.findMessagesByChatId(chatId)
 	items = s.filterMessagesAfter(items, readPointer)
+	items = s.filterMessagesNotSentBy(items, memberId)
 	items = s.filterNotifiedMessages(items)
 	return uint32(len(items)), nil
 }
@@ -446,6 +447,16 @@ func (s *store) filterMessagesAfter(items []*chat.MessageRecord, pointer chat.Me
 	return res
 }
 
+func (s *store) filterMessagesNotSentBy(items []*chat.MessageRecord, sender chat.MemberId) []*chat.MessageRecord {
+	var res []*chat.MessageRecord
+	for _, item := range items {
+		if item.Sender == nil || !bytes.Equal(item.Sender[:], sender[:]) {
+			res = append(res, item)
+		}
+	}
+	return res
+}
+
 func (s *store) filterNotifiedMessages(items []*chat.MessageRecord) []*chat.MessageRecord {
 	var res []*chat.MessageRecord
 	for _, item := range items {

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

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 
+	"github.com/code-payments/code-protobuf-api/generated/go/common/v1"
+
 	"github.com/code-payments/code-server/pkg/database/query"
 )
 
@@ -43,7 +45,7 @@ type Store interface {
 	GetAllMessagesByChatId(ctx context.Context, chatId ChatId, cursor query.Cursor, direction query.Ordering, limit uint64) ([]*MessageRecord, error)
 
 	// GetUnreadCount gets the unread message count for a chat ID at a read pointer
-	GetUnreadCount(ctx context.Context, chatId ChatId, readPointer MessageId) (uint32, error)
+	GetUnreadCount(ctx context.Context, chatId ChatId, memberId MemberId, readPointer MessageId) (uint32, error)
 
 	// PutChat creates a new chat
 	PutChat(ctx context.Context, record *ChatRecord) error
@@ -66,3 +68,23 @@ type Store interface {
 	// SetSubscriptionState updates the subscription state for a chat member
 	SetSubscriptionState(ctx context.Context, chatId ChatId, memberId MemberId, isSubscribed bool) error
 }
+
+type PaymentStore interface {
+	// MarkFriendshipPaid marks a friendship as paid.
+	//
+	// The intentId is the intent that paid for the friendship.
+	MarkFriendshipPaid(ctx context.Context, payer, other *common.SolanaAccountId, intentId *common.IntentId) error
+
+	// IsFriendshipPaid returns whether a payment has been made for a friendship.
+	//
+	// IsFriendshipPaid  is reflexive, with only a single payment being required.
+	IsFriendshipPaid(ctx context.Context, user, other *common.SolanaAccountId) (bool, error)
+
+	// MarkChatPaid marks a chat as paid.
+	MarkChatPaid(ctx context.Context, payer *common.SolanaAccountId, chat ChatId) error
+
+	// IsChatPaid returns whether a member paid to be part of a chat.
+	//
+	// This is only valid for non-two way chats.
+	IsChatPaid(ctx context.Context, chatId ChatId, member *common.SolanaAccountId) (bool, error)
+}

pkg/code/data/internal.go

@@ -71,7 +71,7 @@ import (
 	intent_memory_client "github.com/code-payments/code-server/pkg/code/data/intent/memory"
 	login_memory_client "github.com/code-payments/code-server/pkg/code/data/login/memory"
 	merkletree_memory_client "github.com/code-payments/code-server/pkg/code/data/merkletree/memory"
-	messaging "github.com/code-payments/code-server/pkg/code/data/messaging"
+	"github.com/code-payments/code-server/pkg/code/data/messaging"
 	messaging_memory_client "github.com/code-payments/code-server/pkg/code/data/messaging/memory"
 	nonce_memory_client "github.com/code-payments/code-server/pkg/code/data/nonce/memory"
 	onramp_memory_client "github.com/code-payments/code-server/pkg/code/data/onramp/memory"
@@ -402,7 +402,7 @@ type DatabaseData interface {
 	GetAllChatMembersV2(ctx context.Context, chatId chat_v2.ChatId) ([]*chat_v2.MemberRecord, error)
 	GetPlatformUserChatMembershipV2(ctx context.Context, idByPlatform map[chat_v2.Platform]string, opts ...query.Option) ([]*chat_v2.MemberRecord, error)
 	GetAllChatMessagesV2(ctx context.Context, chatId chat_v2.ChatId, opts ...query.Option) ([]*chat_v2.MessageRecord, error)
-	GetChatUnreadCountV2(ctx context.Context, chatId chat_v2.ChatId, readPointer chat_v2.MessageId) (uint32, error)
+	GetChatUnreadCountV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, readPointer chat_v2.MessageId) (uint32, error)
 	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
@@ -1493,8 +1493,8 @@ func (dp *DatabaseProvider) GetAllChatMessagesV2(ctx context.Context, chatId cha
 	}
 	return dp.chatv2.GetAllMessagesByChatId(ctx, chatId, req.Cursor, req.SortBy, req.Limit)
 }
-func (dp *DatabaseProvider) GetChatUnreadCountV2(ctx context.Context, chatId chat_v2.ChatId, readPointer chat_v2.MessageId) (uint32, error) {
-	return dp.chatv2.GetUnreadCount(ctx, chatId, readPointer)
+func (dp *DatabaseProvider) GetChatUnreadCountV2(ctx context.Context, chatId chat_v2.ChatId, memberId chat_v2.MemberId, readPointer chat_v2.MessageId) (uint32, error) {
+	return dp.chatv2.GetUnreadCount(ctx, chatId, memberId, readPointer)
 }
 func (dp *DatabaseProvider) PutChatV2(ctx context.Context, record *chat_v2.ChatRecord) error {
 	return dp.chatv2.PutChat(ctx, record)

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

@@ -54,18 +54,18 @@ type server struct {
 	chatLocks      *sync_util.StripedLock
 	chatEventChans *sync_util.StripedChannel
 
-	streamsMu sync.RWMutex
-	streams   map[string]*chatEventStream
-
 	chatpb.UnimplementedChatServer
 }
 
-func NewChatServer(data code_data.Provider, auth *auth_util.RPCSignatureVerifier) chatpb.ChatServer {
-	return &server{
-		log:     logrus.StandardLogger().WithField("type", "chat/server"),
-		data:    data,
-		auth:    auth,
-		streams: make(map[string]*chatEventStream),
+func NewChatServer(data code_data.Provider, auth *auth_util.RPCSignatureVerifier, pusher push_lib.Provider) chatpb.ChatServer {
+	s := &server{
+		log:            logrus.StandardLogger().WithField("type", "chat/v1/server"),
+		data:           data,
+		auth:           auth,
+		pusher:         pusher,
+		streams:        make(map[string]*chatEventStream),
+		chatLocks:      sync_util.NewStripedLock(64),             // todo: configurable parameters
+		chatEventChans: sync_util.NewStripedChannel(64, 100_000), // todo: configurable parameters
 	}
 
 	for i, channel := range s.chatEventChans.GetChannels() {
@@ -403,21 +403,9 @@ func (s *server) AdvancePointer(ctx context.Context, req *chatpb.AdvancePointerR
 			Pointers: []*chatpb.Pointer{req.Pointer},
 		}
 
-		s.streamsMu.RLock()
-		for key, stream := range s.streams {
-			if !strings.HasPrefix(key, chatId.String()) {
-				continue
-			}
-
-			if strings.HasSuffix(key, owner.PublicKey().ToBase58()) {
-				continue
-			}
-
-			if err := stream.notify(event, streamNotifyTimeout); err != nil {
-				log.WithError(err).Warnf("failed to notify session stream, closing streamer (stream=%p)", stream)
-			}
+		if err := s.asyncNotifyAll(chatId, owner, event); err != nil {
+			log.WithError(err).Warn("failure notifying chat event")
 		}
-		s.streamsMu.RUnlock()
 
 		return &chatpb.AdvancePointerResponse{
 			Result: chatpb.AdvancePointerResponse_OK,
@@ -428,7 +416,7 @@ func (s *server) AdvancePointer(ctx context.Context, req *chatpb.AdvancePointerR
 		return nil, status.Error(codes.InvalidArgument, "Pointer.Kind must be READ")
 	}
 
-	chatRecord, err := s.data.GetChatById(ctx, chatId)
+	chatRecord, err := s.data.GetChatByIdV1(ctx, chatId)
 	if err == chat.ErrChatNotFound {
 		return &chatpb.AdvancePointerResponse{
 			Result: chatpb.AdvancePointerResponse_CHAT_NOT_FOUND,
@@ -656,6 +644,22 @@ func (s *server) StreamChatEvents(streamer chatpb.Chat_StreamChatEventsServer) e
 
 	s.streamsMu.Unlock()
 
+	defer func() {
+		s.streamsMu.Lock()
+
+		log.Tracef("closing streamer (stream=%s)", streamRef)
+
+		// We check to see if the current active stream is the one that we created.
+		// If it is, we can just remove it since it's closed. Otherwise, we leave it
+		// be, as another OpenMessageStream() call is handling it.
+		liveStream, exists := s.streams[streamKey]
+		if exists && liveStream == stream {
+			delete(s.streams, streamKey)
+		}
+
+		s.streamsMu.Unlock()
+	}()
+
 	sendPingCh := time.After(0)
 	streamHealthCh := monitorChatEventStreamHealth(ctx, log, streamRef, streamer)
 
@@ -734,6 +738,10 @@ func (s *server) SendMessage(ctx context.Context, req *chatpb.SendMessageRequest
 		return nil, status.Error(codes.InvalidArgument, "content[0] must be Text or ThankYou")
 	}
 
+	chatLock := s.chatLocks.Get(chatId[:])
+	chatLock.Lock()
+	defer chatLock.Unlock()
+
 	// todo: Revisit message IDs
 	messageId, err := common.NewRandomAccount()
 	if err != nil {
@@ -749,28 +757,54 @@ func (s *server) SendMessage(ctx context.Context, req *chatpb.SendMessageRequest
 		Cursor:    nil, // todo: Don't have cursor until we save it to the DB
 	}
 
+	// todo: Save the message to the DB
+
 	event := &chatpb.ChatStreamEvent{
 		Messages: []*chatpb.ChatMessage{chatMessage},
 	}
 
-	s.streamsMu.RLock()
-	for key, stream := range s.streams {
-		if !strings.HasPrefix(key, chatId.String()) {
-			continue
-		}
-
-		if strings.HasSuffix(key, owner.PublicKey().ToBase58()) {
-			continue
-		}
-
-		if err := stream.notify(event, streamNotifyTimeout); err != nil {
-			log.WithError(err).Warnf("failed to notify session stream, closing streamer (stream=%p)", stream)
-		}
+	if err := s.asyncNotifyAll(chatId, owner, event); err != nil {
+		log.WithError(err).Warn("failure notifying chat event")
 	}
-	s.streamsMu.RUnlock()
+
+	s.asyncPushChatMessage(owner, chatId, chatMessage)
 
 	return &chatpb.SendMessageResponse{
 		Result:  chatpb.SendMessageResponse_OK,
 		Message: chatMessage,
 	}, nil
 }
+
+// todo: doesn't respect mute/unsubscribe rules
+// todo: only sends pushes to active stream listeners instead of all message recipients
+func (s *server) asyncPushChatMessage(sender *common.Account, chatId chat.ChatId, chatMessage *chatpb.ChatMessage) {
+	ctx := context.TODO()
+
+	go func() {
+		s.streamsMu.RLock()
+		for key := range s.streams {
+			if !strings.HasPrefix(key, chatId.String()) {
+				continue
+			}
+
+			receiver, err := common.NewAccountFromPublicKeyString(strings.Split(key, ":")[1])
+			if err != nil {
+				continue
+			}
+
+			if bytes.Equal(sender.PublicKey().ToBytes(), receiver.PublicKey().ToBytes()) {
+				continue
+			}
+
+			go push_util.SendChatMessagePushNotification(
+				ctx,
+				s.data,
+				s.pusher,
+				"TontonTwitch",
+				receiver,
+				chatMessage,
+			)
+		}
+		s.streamsMu.RUnlock()
+	}()
+}