status-go/api/messenger_raw_message_resend_test.go

package api

import (
	"context"
	"errors"
	"path/filepath"
	"testing"
	"time"

	"github.com/cenkalti/backoff/v3"
	"github.com/golang/protobuf/proto"
	"go.uber.org/zap"

	"github.com/status-im/status-go/eth-node/types"
	m_common "github.com/status-im/status-go/multiaccounts/common"
	"github.com/status-im/status-go/params"
	"github.com/status-im/status-go/protocol"
	"github.com/status-im/status-go/protocol/common"
	"github.com/status-im/status-go/protocol/common/shard"
	"github.com/status-im/status-go/protocol/communities"
	"github.com/status-im/status-go/protocol/protobuf"
	"github.com/status-im/status-go/protocol/requests"
	"github.com/status-im/status-go/protocol/tt"
	"github.com/status-im/status-go/services/utils"
	"github.com/status-im/status-go/signal"
	tutils "github.com/status-im/status-go/t/utils"
	"github.com/status-im/status-go/wakuv2"

	"github.com/stretchr/testify/suite"
)

type MessengerRawMessageResendTest struct {
	suite.Suite
	logger         *zap.Logger
	aliceBackend   *GethStatusBackend
	bobBackend     *GethStatusBackend
	aliceMessenger *protocol.Messenger
	bobMessenger   *protocol.Messenger
	// add exchangeBootNode to ensure alice and bob can find each other.
	// If relying on in the fleet, the test will likely be flaky
	exchangeBootNode *wakuv2.Waku
}

func TestMessengerRawMessageResendTestSuite(t *testing.T) {
	suite.Run(t, new(MessengerRawMessageResendTest))
}

func (s *MessengerRawMessageResendTest) SetupTest() {
	tutils.Init()

	var err error
	s.logger, err = zap.NewDevelopment()
	s.Require().NoError(err)

	signal.SetMobileSignalHandler(nil)

	exchangeNodeConfig := &wakuv2.Config{
		Port:                                   0,
		EnableDiscV5:                           true,
		EnablePeerExchangeServer:               true,
		ClusterID:                              16,
		DefaultShardPubsubTopic:                shard.DefaultShardPubsubTopic(),
		EnableStoreConfirmationForMessagesSent: false,
	}
	s.exchangeBootNode, err = wakuv2.New(nil, "", exchangeNodeConfig, s.logger.Named("pxServerNode"), nil, nil, nil, nil)
	s.Require().NoError(err)
	s.Require().NoError(s.exchangeBootNode.Start())

	s.createAliceBobBackendAndLogin()
	community := s.createTestCommunity(s.aliceMessenger, protobuf.CommunityPermissions_MANUAL_ACCEPT)
	s.addMutualContact()
	advertiseCommunityToUserOldWay(&s.Suite, community, s.aliceMessenger, s.bobMessenger)
	requestBob := &requests.RequestToJoinCommunity{
		CommunityID: community.ID(),
	}
	joinOnRequestCommunity(&s.Suite, community, s.aliceMessenger, s.bobMessenger, requestBob)
}

func (s *MessengerRawMessageResendTest) TearDownTest() {
	// Initialize a map to keep track of the operation status.
	operationStatus := map[string]bool{
		"Alice Logout":   false,
		"Bob Logout":     false,
		"Boot Node Stop": false,
	}

	done := make(chan string, 3) // Buffered channel to receive the names of the completed operations
	errs := make(chan error, 3)  // Channel to receive errs from operations

	// Asynchronously perform operations and report completion or errs.
	go func() {
		err := s.aliceBackend.Logout()
		if err != nil {
			errs <- err
		}
		done <- "Alice Logout"
	}()

	go func() {
		err := s.bobBackend.Logout()
		if err != nil {
			errs <- err
		}
		done <- "Bob Logout"
	}()

	go func() {
		err := s.exchangeBootNode.Stop()
		if err != nil {
			errs <- err
		}
		done <- "Boot Node Stop"
	}()

	timeout := time.After(30 * time.Second)
	operationsCompleted := 0

	for operationsCompleted < 3 {
		select {
		case opName := <-done:
			s.T().Logf("%s completed successfully.", opName)
			operationStatus[opName] = true
			operationsCompleted++
		case err := <-errs:
			s.Require().NoError(err)
		case <-timeout:
			// If a timeout occurs, check which operations have not reported completion.
			s.T().Errorf("Timeout occurred, the following operations did not complete in time:")
			for opName, completed := range operationStatus {
				if !completed {
					s.T().Errorf("%s is still pending.", opName)
				}
			}
			s.T().FailNow()
		}
	}
}

func (s *MessengerRawMessageResendTest) createAliceBobBackendAndLogin() {
	pxServerNodeENR, err := s.exchangeBootNode.GetNodeENRString()
	s.Require().NoError(err)
	// we don't support multiple logger instances, so just share the log dir
	shareLogDir := filepath.Join(s.T().TempDir(), "logs")
	s.T().Logf("shareLogDir: %s", shareLogDir)
	s.createBackendAndLogin(&s.aliceBackend, &s.aliceMessenger, "alice66", pxServerNodeENR, shareLogDir)
	s.createBackendAndLogin(&s.bobBackend, &s.bobMessenger, "bob66", pxServerNodeENR, shareLogDir)

	aliceWaku := s.aliceBackend.StatusNode().WakuV2Service()
	bobWaku := s.bobBackend.StatusNode().WakuV2Service()
	// NOTE: default MaxInterval is 10s, which is too short for the test
	// TODO(frank) figure out why it takes so long for the peers to know each other
	err = tt.RetryWithBackOff(func() error {
		if len(aliceWaku.Peerstore().Addrs(bobWaku.PeerID())) > 0 {
			return nil
		}
		s.T().Logf("alice don't know bob's addresses")
		return errors.New("alice don't know bob's addresses")
	}, func(b *backoff.ExponentialBackOff) { b.MaxInterval = 20 * time.Second })
	s.Require().NoError(err)
	err = tt.RetryWithBackOff(func() error {
		if len(bobWaku.Peerstore().Addrs(aliceWaku.PeerID())) > 0 {
			return nil
		}
		s.T().Logf("bob don't know alice's addresses")
		return errors.New("bob don't know alice's addresses")
	}, func(b *backoff.ExponentialBackOff) { b.MaxInterval = 20 * time.Second })
	s.Require().NoError(err)
}

func (s *MessengerRawMessageResendTest) createBackendAndLogin(backend **GethStatusBackend, messenger **protocol.Messenger, displayName, pxServerNodeENR, shareLogDir string) {
	*backend = NewGethStatusBackend()
	rootDir := filepath.Join(s.T().TempDir())
	s.T().Logf("%s rootDir: %s", displayName, rootDir)
	createAccountRequest := s.setCreateAccountRequest(displayName, rootDir, shareLogDir)
	_, err := (*backend).CreateAccountAndLogin(createAccountRequest,
		params.WithDiscV5BootstrapNodes([]string{pxServerNodeENR}),
		// override fleet nodes
		params.WithWakuNodes([]string{}))
	s.Require().NoError(err)
	*messenger = (*backend).Messenger()
	s.Require().NotNil(messenger)
	_, err = (*messenger).Start()
	s.Require().NoError(err)
}

func (s *MessengerRawMessageResendTest) setCreateAccountRequest(displayName, rootDataDir, logFilePath string) *requests.CreateAccount {
	nameServer := "1.1.1.1"
	verifyENSContractAddress := "0x00000000000C2E074eC69A0dFb2997BA6C7d2e1e"
	verifyTransactionChainID := int64(1)
	verifyURL := "https://eth-archival.rpc.grove.city/v1/3ef2018191814b7e1009b8d9"
	logLevel := "DEBUG"
	networkID := uint64(1)
	password := "qwerty"
	return &requests.CreateAccount{
		UpstreamConfig:           verifyURL,
		WakuV2Nameserver:         &nameServer,
		VerifyENSContractAddress: &verifyENSContractAddress,
		RootDataDir:              rootDataDir,
		Password:                 password,
		DisplayName:              displayName,
		LogEnabled:               true,
		VerifyTransactionChainID: &verifyTransactionChainID,
		VerifyTransactionURL:     &verifyURL,
		VerifyENSURL:             &verifyURL,
		LogLevel:                 &logLevel,
		LogFilePath:              logFilePath,
		NetworkID:                &networkID,
		CustomizationColor:       string(m_common.CustomizationColorPrimary),
	}
}

func (s *MessengerRawMessageResendTest) waitForMessageSent(messageID string) {
	err := tt.RetryWithBackOff(func() error {
		rawMessage, err := s.bobMessenger.RawMessageByID(messageID)
		s.Require().NoError(err)
		s.Require().NotNil(rawMessage)
		if rawMessage.SendCount > 0 {
			return nil
		}
		return errors.New("raw message should be sent finally")
	})
	s.Require().NoError(err)
}

// TestMessageSent tests if ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN is in state `sent` without resending
func (s *MessengerRawMessageResendTest) TestMessageSent() {
	ids, err := s.bobMessenger.RawMessagesIDsByType(protobuf.ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN)
	s.Require().NoError(err)
	s.Require().Len(ids, 1)

	s.waitForMessageSent(ids[0])
}

// TestMessageResend tests if ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN is resent
func (s *MessengerRawMessageResendTest) TestMessageResend() {
	ids, err := s.bobMessenger.RawMessagesIDsByType(protobuf.ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN)
	s.Require().NoError(err)
	s.Require().Len(ids, 1)
	// wait for Sent status for already sent message to make sure that sent message was delivered
	// before testing resend
	s.waitForMessageSent(ids[0])

	rawMessage, err := s.bobMessenger.RawMessageByID(ids[0])
	s.Require().NoError(err)
	s.Require().NotNil(rawMessage)

	s.Require().NoError(s.bobMessenger.UpdateRawMessageSent(rawMessage.ID, false))
	s.Require().NoError(s.bobMessenger.UpdateRawMessageLastSent(rawMessage.ID, 0))

	err = tt.RetryWithBackOff(func() error {
		rawMessage, err := s.bobMessenger.RawMessageByID(ids[0])
		s.Require().NoError(err)
		s.Require().NotNil(rawMessage)
		if rawMessage.SendCount < 2 {
			return errors.New("message ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN was not resent yet")
		}
		return nil
	})
	s.Require().NoError(err)

	waitOnMessengerResponse(&s.Suite, func(r *protocol.MessengerResponse) error {
		if len(r.RequestsToJoinCommunity()) > 0 {
			return nil
		}
		return errors.New("community request to join not received")
	}, s.aliceMessenger)
}

func (s *MessengerRawMessageResendTest) TestInvalidRawMessageToWatchDoesNotProduceResendLoop() {
	ids, err := s.bobMessenger.RawMessagesIDsByType(protobuf.ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN)
	s.Require().NoError(err)
	s.Require().Len(ids, 1)

	s.waitForMessageSent(ids[0])

	rawMessage, err := s.bobMessenger.RawMessageByID(ids[0])
	s.Require().NoError(err)

	requestToJoinProto := &protobuf.CommunityRequestToJoin{}
	err = proto.Unmarshal(rawMessage.Payload, requestToJoinProto)
	s.Require().NoError(err)

	requestToJoinProto.DisplayName = "invalid_ID"
	payload, err := proto.Marshal(requestToJoinProto)
	s.Require().NoError(err)
	rawMessage.Payload = payload

	_, err = s.bobMessenger.AddRawMessageToWatch(rawMessage)
	s.Require().Error(err, common.ErrModifiedRawMessage)

	// simulate storing msg with modified payload, but old message ID
	_, err = s.bobMessenger.UpsertRawMessageToWatch(rawMessage)
	s.Require().NoError(err)
	s.Require().NoError(s.bobMessenger.UpdateRawMessageSent(rawMessage.ID, false))
	s.Require().NoError(s.bobMessenger.UpdateRawMessageLastSent(rawMessage.ID, 0))

	// check counter increased for invalid message to escape the loop
	err = tt.RetryWithBackOff(func() error {
		rawMessage, err := s.bobMessenger.RawMessageByID(ids[0])
		s.Require().NoError(err)
		s.Require().NotNil(rawMessage)
		if rawMessage.SendCount < 2 {
			return errors.New("message ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN was not resent yet")
		}
		return nil
	})
	s.Require().NoError(err)
}

// To be removed in https://github.com/status-im/status-go/issues/4437
func advertiseCommunityToUserOldWay(s *suite.Suite, community *communities.Community, alice *protocol.Messenger, bob *protocol.Messenger) {
	chat := protocol.CreateOneToOneChat(bob.IdentityPublicKeyString(), bob.IdentityPublicKey(), bob.GetTransport())

	inputMessage := common.NewMessage()
	inputMessage.ChatId = chat.ID
	inputMessage.Text = "some text"
	inputMessage.CommunityID = community.IDString()

	err := alice.SaveChat(chat)
	s.Require().NoError(err)
	_, err = alice.SendChatMessage(context.Background(), inputMessage)
	s.Require().NoError(err)

	// Ensure community is received
	response, err := protocol.WaitOnMessengerResponse(
		bob,
		func(r *protocol.MessengerResponse) bool {
			return len(r.Communities()) > 0
		},
		"bob did not receive community request to join",
	)
	s.Require().NoError(err)
	communityInResponse := response.Communities()[0]
	s.Require().Equal(community.ID(), communityInResponse.ID())
}

func (s *MessengerRawMessageResendTest) addMutualContact() {
	bobPubkey := s.bobMessenger.IdentityPublicKeyCompressed()
	bobZQ3ID, err := utils.SerializePublicKey(bobPubkey)
	s.Require().NoError(err)
	mr, err := s.aliceMessenger.AddContact(context.Background(), &requests.AddContact{
		ID:          bobZQ3ID,
		DisplayName: "bob666",
	})
	s.Require().NoError(err)
	s.Require().Len(mr.Messages(), 2)

	var contactRequest *common.Message
	waitOnMessengerResponse(&s.Suite, func(r *protocol.MessengerResponse) error {
		for _, m := range r.Messages() {
			if m.GetContentType() == protobuf.ChatMessage_CONTACT_REQUEST {
				contactRequest = m
				return nil
			}
		}
		return errors.New("contact request not received")
	}, s.bobMessenger)

	mr, err = s.bobMessenger.AcceptContactRequest(context.Background(), &requests.AcceptContactRequest{
		ID: types.FromHex(contactRequest.ID),
	})
	s.Require().NoError(err)
	s.Require().Len(mr.Contacts, 1)

	waitOnMessengerResponse(&s.Suite, func(r *protocol.MessengerResponse) error {
		if len(r.Contacts) > 0 {
			return nil
		}
		return errors.New("contact accepted not received")
	}, s.aliceMessenger)
}

type MessageResponseValidator func(*protocol.MessengerResponse) error

func waitOnMessengerResponse(s *suite.Suite, fnWait MessageResponseValidator, user *protocol.Messenger) {
	_, err := protocol.WaitOnMessengerResponse(
		user,
		func(r *protocol.MessengerResponse) bool {
			err := fnWait(r)
			if err != nil {
				s.T().Logf("response error: %s", err.Error())
			}
			return err == nil
		},
		"MessengerResponse data not received",
	)
	s.Require().NoError(err)
}

func requestToJoinCommunity(s *suite.Suite, controlNode *protocol.Messenger, user *protocol.Messenger, request *requests.RequestToJoinCommunity) types.HexBytes {
	response, err := user.RequestToJoinCommunity(request)
	s.Require().NoError(err)
	s.Require().NotNil(response)
	s.Require().Len(response.RequestsToJoinCommunity(), 1)

	requestToJoin := response.RequestsToJoinCommunity()[0]
	s.Require().Equal(requestToJoin.PublicKey, user.IdentityPublicKeyString())

	_, err = protocol.WaitOnMessengerResponse(
		controlNode,
		func(r *protocol.MessengerResponse) bool {
			if len(r.RequestsToJoinCommunity()) == 0 {
				return false
			}

			for _, resultRequest := range r.RequestsToJoinCommunity() {
				if resultRequest.PublicKey == user.IdentityPublicKeyString() {
					return true
				}
			}
			return false
		},
		"control node did not receive community request to join",
	)
	s.Require().NoError(err)

	return requestToJoin.ID
}

func joinOnRequestCommunity(s *suite.Suite, community *communities.Community, controlNode *protocol.Messenger, user *protocol.Messenger, request *requests.RequestToJoinCommunity) {
	// Request to join the community
	requestToJoinID := requestToJoinCommunity(s, controlNode, user, request)

	// accept join request
	acceptRequestToJoin := &requests.AcceptRequestToJoinCommunity{ID: requestToJoinID}
	response, err := controlNode.AcceptRequestToJoinCommunity(acceptRequestToJoin)
	s.Require().NoError(err)
	s.Require().NotNil(response)

	updatedCommunity := response.Communities()[0]
	s.Require().NotNil(updatedCommunity)
	s.Require().True(updatedCommunity.HasMember(user.IdentityPublicKey()))

	// receive request to join response
	_, err = protocol.WaitOnMessengerResponse(
		user,
		func(r *protocol.MessengerResponse) bool {
			return len(r.Communities()) > 0 && r.Communities()[0].HasMember(user.IdentityPublicKey())
		},
		"user did not receive request to join response",
	)
	s.Require().NoError(err)

	userCommunity, err := user.GetCommunityByID(community.ID())
	s.Require().NoError(err)
	s.Require().True(userCommunity.HasMember(user.IdentityPublicKey()))

	_, err = protocol.WaitOnMessengerResponse(
		controlNode,
		func(r *protocol.MessengerResponse) bool {
			return len(r.Communities()) > 0 && r.Communities()[0].HasMember(user.IdentityPublicKey())
		},
		"control node did not receive request to join response",
	)
	s.Require().NoError(err)
}

func (s *MessengerRawMessageResendTest) createTestCommunity(controlNode *protocol.Messenger, membershipType protobuf.CommunityPermissions_Access) *communities.Community {
	description := &requests.CreateCommunity{
		Membership:                  membershipType,
		Name:                        "status",
		Color:                       "#ffffff",
		Description:                 "status community description",
		PinMessageAllMembersEnabled: false,
	}
	response, err := controlNode.CreateCommunity(description, true)
	s.Require().NoError(err)
	s.Require().NotNil(response)
	s.Require().Len(response.Communities(), 1)
	s.Require().Len(response.Chats(), 1)
	return response.Communities()[0]
}