Refactoring: cleanup, renaming and restructuring

internal/app/app.go

@@ -145,7 +145,7 @@ func NewApp(ctx context.Context, cfg *config.Config, logger *zap.SugaredLogger)
 	botAddress := crypto.PubkeyToAddress(cfg.BotKey.PublicKey)
 	botUserID := messaging.UserIDFromAddress(botAddress, cfg.Matrix.Host)
 
-	messageProcessor := messaging.NewProcessor(
+	messageProcessor := messaging.NewMessageProcessor(
 		matrixMessenger,
 		logger,
 		cfg.ResponseTimeout,
@@ -208,7 +208,7 @@ type App struct {
 	chequeHandler    chequehandler.ChequeHandler
 	rpcClient        *client.RPCClient
 	rpcServer        server.Server
-	messageProcessor messaging.Processor
+	messageProcessor messaging.MessageProcessor
 	messenger        messaging.Messenger
 	botUserID        id.UserID
 }

internal/compression/compress.go

@@ -10,7 +10,6 @@ import (
 )
 
 const (
-
 	// MaxChunkSize a moderate/safe max chunk size is 48KB. This is because the maximum size of a matrix event is 64KB.
 	// Megolm encryption adds an extra 33% overhead to the encrypted content due to base64 encryption. This means that
 	// the maximum size of pre-encrypted chunk should be 48KB / 1.33 ~= 36KB. We round down to 35KB to be safe.

internal/matrix/matrix_messenger.go

@@ -78,6 +78,10 @@ func (m *messenger) StartReceiver() (id.UserID, error) {
 	syncer := m.client.Syncer.(*mautrix.DefaultSyncer)
 
 	syncer.OnEventType(matrix.EventTypeC4TMessage, func(ctx context.Context, evt *event.Event) {
+		if evt.Sender == m.client.UserID { // ignore own messages
+			return
+		}
+
 		msg := evt.Content.Parsed.(*matrix.CaminoMatrixMessage)
 		traceID, err := trace.TraceIDFromHex(msg.Metadata.RequestID)
 		if err != nil {
@@ -169,7 +173,7 @@ func (m *messenger) StopReceiver() error {
 	return m.client.cryptoHelper.Close()
 }
 
-func (m *messenger) SendAsync(ctx context.Context, msg types.Message, content [][]byte, sendTo id.UserID) error {
+func (m *messenger) SendAsync(ctx context.Context, msg *types.Message, sendTo id.UserID) error {
 	m.logger.Info("Sending async message", zap.String("msg", msg.Metadata.RequestID))
 	ctx, span := m.tracer.Start(ctx, "messenger.SendAsync", trace.WithSpanKind(trace.SpanKindProducer), trace.WithAttributes(attribute.String("type", string(msg.Type))))
 	defer span.End()
@@ -181,7 +185,7 @@ func (m *messenger) SendAsync(ctx context.Context, msg types.Message, content []
 	}
 	roomSpan.End()
 
-	return m.sendMessageEvents(ctx, roomID, matrix.EventTypeC4TMessage, createMatrixMessages(&msg, content))
+	return m.sendMessageEvents(ctx, roomID, matrix.EventTypeC4TMessage, createMatrixMessages(msg))
 }
 
 func (m *messenger) sendMessageEvents(ctx context.Context, roomID id.RoomID, eventType event.Type, messages []matrix.CaminoMatrixMessage) error {
@@ -243,24 +247,24 @@ func hexWithChecksum(bytes []byte) (string, error) {
 	return fmt.Sprintf("0x%x", bytes), nil
 }
 
-func createMatrixMessages(msg *types.Message, content [][]byte) []matrix.CaminoMatrixMessage {
-	messages := make([]matrix.CaminoMatrixMessage, 0, len(content))
+func createMatrixMessages(msg *types.Message) []matrix.CaminoMatrixMessage {
+	messages := make([]matrix.CaminoMatrixMessage, 0, len(msg.CompressedContent))
 
 	// add first chunk to messages slice
 	caminoMatrixMsg := matrix.CaminoMatrixMessage{
 		MessageEventContent: event.MessageEventContent{MsgType: event.MessageType(msg.Type)},
 		Metadata:            msg.Metadata,
 	}
-	caminoMatrixMsg.Metadata.NumberOfChunks = uint64(len(content))
+	caminoMatrixMsg.Metadata.NumberOfChunks = uint64(len(msg.CompressedContent))
 	caminoMatrixMsg.Metadata.ChunkIndex = 0
-	caminoMatrixMsg.CompressedContent = content[0]
+	caminoMatrixMsg.CompressedContent = msg.CompressedContent[0]
 	messages = append(messages, caminoMatrixMsg)
 
 	// if multiple chunks were produced upon compression, add them to messages slice
-	for i, chunk := range content[1:] {
+	for i, chunk := range msg.CompressedContent[1:] {
 		messages = append(messages, matrix.CaminoMatrixMessage{
 			MessageEventContent: event.MessageEventContent{MsgType: event.MessageType(msg.Type)},
-			Metadata:            metadata.Metadata{RequestID: msg.Metadata.RequestID, NumberOfChunks: uint64(len(content)), ChunkIndex: uint64(i + 1)},
+			Metadata:            metadata.Metadata{RequestID: msg.Metadata.RequestID, NumberOfChunks: uint64(len(msg.CompressedContent)), ChunkIndex: uint64(i + 1)},
 			CompressedContent:   chunk,
 		})
 	}

internal/messaging/messenger.go

@@ -27,7 +27,7 @@ type Messenger interface {
 	StopReceiver() error
 
 	// asynchronous call (fire and forget)
-	SendAsync(ctx context.Context, m types.Message, content [][]byte, sendTo id.UserID) error
+	SendAsync(ctx context.Context, m *types.Message, sendTo id.UserID) error
 
 	// channel where incoming messages are written
 	Inbound() chan types.Message

internal/messaging/mint.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math/big"
 	"time"
@@ -17,6 +18,8 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
+var errMissingMintTxID = errors.New("missing mint transaction id")
+
 // Mints a BookingToken with the supplier private key and reserves it for the buyer address
 // For testing you can use this uri: "data:application/json;base64,eyJuYW1lIjoiQ2FtaW5vIE1lc3NlbmdlciBCb29raW5nVG9rZW4gVGVzdCJ9Cg=="
 func (h *evmResponseHandler) mint(
@@ -206,3 +209,9 @@ func verifyAndFixBuyableUntil(buyableUntil *timestamppb.Timestamp, currentTime t
 
 	return buyableUntil, nil
 }
+
+func ensureHeaderV1(responseHeader **typesv1.ResponseHeader) {
+	if *responseHeader == nil {
+		*responseHeader = &typesv1.ResponseHeader{}
+	}
+}

internal/messaging/mint_v1.go

@@ -10,114 +10,100 @@ import (
 	typesv1 "buf.build/gen/go/chain4travel/camino-messenger-protocol/protocolbuffers/go/cmp/types/v1"
 
 	"github.com/ethereum/go-ethereum/common"
-	"google.golang.org/protobuf/reflect/protoreflect"
 )
 
-func (h *evmResponseHandler) handleMintResponseV1(ctx context.Context, response protoreflect.ProtoMessage, request protoreflect.ProtoMessage) bool {
-	mintResp, ok := response.(*bookv1.MintResponse)
-	if !ok {
-		return false
-	}
-	mintReq, ok := request.(*bookv1.MintRequest)
-	if !ok {
-		return false
-	}
-	if mintResp.Header == nil {
-		mintResp.Header = &typesv1.ResponseHeader{}
-	}
-
-	if mintResp.Header.Status == typesv1.StatusType_STATUS_TYPE_FAILURE {
-		return false
+func (h *evmResponseHandler) prepareMintResponseV1(
+	ctx context.Context,
+	response *bookv1.MintResponse,
+	request *bookv1.MintRequest,
+) {
+	ensureHeaderV1(&response.Header)
+	if response.Header.Status == typesv1.StatusType_STATUS_TYPE_FAILURE {
+		return
 	}
 
 	// TODO @evlekht ensure that mintReq.BuyerAddress is c-chain address format,
 	// TODO not x/p/t chain or anything else. Currently it will not error
 	// TODO if address is invalid and will just get zero addr
-	buyerAddress := common.HexToAddress(mintReq.BuyerAddress)
+	buyerAddress := common.HexToAddress(request.BuyerAddress)
 
 	// Get a Token URI for the token.
 	jsonPlain, tokenURI, err := createTokenURIforMintResponse(
-		mintResp.MintId.Value,
-		mintReq.BookingReference,
+		response.MintId.Value,
+		request.BookingReference,
 	)
 	if err != nil {
 		errMsg := fmt.Sprintf("error creating token URI: %v", err)
 		h.logger.Debugf(errMsg) // TODO: @VjeraTurk change to Error after we stop using mocked uri data
 		h.AddErrorToResponseHeader(response, errMsg)
-		return true
+		return
 	}
 
 	h.logger.Debugf("Token URI JSON: %s\n", jsonPlain)
 
-	buyableUntil, err := verifyAndFixBuyableUntil(mintResp.BuyableUntil, time.Now())
+	buyableUntil, err := verifyAndFixBuyableUntil(response.BuyableUntil, time.Now())
 	if err != nil {
 		h.logger.Error(err)
 		h.AddErrorToResponseHeader(response, err.Error())
-		return true
+		return
 	}
-	mintResp.BuyableUntil = buyableUntil
+	response.BuyableUntil = buyableUntil
 
-	price, paymentToken, err := h.getPriceAndTokenV1(ctx, mintResp.Price)
+	price, paymentToken, err := h.getPriceAndTokenV1(ctx, response.Price)
 	if err != nil {
 		errMessage := fmt.Sprintf("error getting price and payment token: %v", err)
 		h.logger.Errorf(errMessage)
 		h.AddErrorToResponseHeader(response, errMessage)
-		return true
+		return
 	}
 
 	// MINT TOKEN
 	txID, tokenID, err := h.mint(
 		ctx,
 		buyerAddress,
 		tokenURI,
-		big.NewInt(mintResp.BuyableUntil.Seconds),
+		big.NewInt(response.BuyableUntil.Seconds),
 		price,
 		paymentToken,
 	)
 	if err != nil {
 		errMessage := fmt.Sprintf("error minting NFT: %v", err)
 		h.logger.Errorf(errMessage)
 		h.AddErrorToResponseHeader(response, errMessage)
-		return true
+		return
 	}
 
 	h.logger.Infof("NFT minted with txID: %s\n", txID)
 
-	h.onBookingTokenMint(tokenID, mintResp.MintId, mintResp.BuyableUntil.AsTime())
+	h.onBookingTokenMint(tokenID, response.MintId, response.BuyableUntil.AsTime())
 
-	mintResp.Header.Status = typesv1.StatusType_STATUS_TYPE_SUCCESS
-	mintResp.BookingToken = &typesv1.BookingToken{TokenId: int32(tokenID.Int64())} //nolint:gosec
-	mintResp.MintTransactionId = txID
-	return false
+	response.Header.Status = typesv1.StatusType_STATUS_TYPE_SUCCESS
+	response.BookingToken = &typesv1.BookingToken{TokenId: int32(tokenID.Int64())} //nolint:gosec
+	response.MintTransactionId = txID
 }
 
-func (h *evmResponseHandler) handleMintRequestV1(ctx context.Context, response protoreflect.ProtoMessage) bool {
-	mintResp, ok := response.(*bookv1.MintResponse)
-	if !ok {
-		return false
-	}
-	if mintResp.Header == nil {
-		mintResp.Header = &typesv1.ResponseHeader{}
-	}
-	if mintResp.MintTransactionId == "" {
-		h.AddErrorToResponseHeader(response, "missing mint transaction id")
-		return true
+func (h *evmResponseHandler) processMintResponseV1(ctx context.Context, response *bookv1.MintResponse) {
+	ensureHeaderV1(&response.Header)
+
+	if response.MintTransactionId == "" {
+		h.logger.Error(errMissingMintTxID)
+		h.AddErrorToResponseHeader(response, errMissingMintTxID.Error())
+		return
 	}
 
-	value64 := uint64(mintResp.BookingToken.TokenId)
+	value64 := uint64(response.BookingToken.TokenId)
 	tokenID := new(big.Int).SetUint64(value64)
 
 	receipt, err := h.bookingService.BuyBookingToken(ctx, tokenID)
 	if err != nil {
 		errMessage := fmt.Sprintf("error buying NFT: %v", err)
 		h.logger.Errorf(errMessage)
 		h.AddErrorToResponseHeader(response, errMessage)
-		return true
+		return
 	}
 
-	h.logger.Infof("Bought NFT (txID=%s) with ID: %s\n", receipt, mintResp.MintTransactionId)
-	mintResp.BuyTransactionId = receipt.TxHash.Hex()
-	return false
+	h.logger.Infof("Bought NFT (txID=%s) with ID: %s\n", receipt, response.MintTransactionId)
+	response.BuyTransactionId = receipt.TxHash.Hex()
 }
 
 func (h *evmResponseHandler) getPriceAndTokenV1(ctx context.Context, price *typesv1.Price) (*big.Int, common.Address, error) {