Add idempotent_id to chat and threads services (#64)

* Add idempotent_id to chat and threads services

* Rename

threads/handler/create_message.go

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"context"
+	"strings"
 
 	"github.com/google/uuid"
 	"github.com/micro/micro/v3/service/errors"
@@ -23,31 +24,40 @@ func (s *Threads) CreateMessage(ctx context.Context, req *pb.CreateMessageReques
 		return ErrMissingText
 	}
 
-	return s.DB.Transaction(func(tx *gorm.DB) error {
-		// lookup the conversation
-		var conv Conversation
-		if err := tx.Where(&Conversation{ID: req.ConversationId}).First(&conv).Error; err == gorm.ErrRecordNotFound {
-			return ErrNotFound
-		} else if err != nil {
-			logger.Errorf("Error reading conversation: %v", err)
-			return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
-		}
+	// lookup the conversation
+	var conv Conversation
+	if err := s.DB.Where(&Conversation{ID: req.ConversationId}).First(&conv).Error; err == gorm.ErrRecordNotFound {
+		return ErrNotFound
+	} else if err != nil {
+		logger.Errorf("Error reading conversation: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
 
-		// create the message
-		msg := &Message{
-			ID:             uuid.New().String(),
-			SentAt:         s.Time(),
-			Text:           req.Text,
-			AuthorID:       req.AuthorId,
-			ConversationID: req.ConversationId,
-		}
-		if err := tx.Create(msg).Error; err != nil {
-			logger.Errorf("Error creating message: %v", err)
-			return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
-		}
-
-		// serialize the response
+	// create the message
+	msg := &Message{
+		ID:             req.Id,
+		SentAt:         s.Time(),
+		Text:           req.Text,
+		AuthorID:       req.AuthorId,
+		ConversationID: req.ConversationId,
+	}
+	if len(msg.ID) == 0 {
+		msg.ID = uuid.New().String()
+	}
+	if err := s.DB.Create(msg).Error; err == nil {
 		rsp.Message = msg.Serialize()
 		return nil
-	})
+	} else if !strings.Contains(err.Error(), "messages_pkey") {
+		logger.Errorf("Error creating message: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+
+	// a message already exists with this id
+	var existing Message
+	if err := s.DB.Where(&Message{ID: msg.ID}).First(&existing).Error; err != nil {
+		logger.Errorf("Error creating message: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+	rsp.Message = existing.Serialize()
+	return nil
 }

threads/handler/create_message_test.go

@@ -25,10 +25,12 @@ func TestCreateMessage(t *testing.T) {
 		return
 	}
 
+	iid := uuid.New().String()
 	tt := []struct {
 		Name           string
 		AuthorID       string
 		ConversationID string
+		ID             string
 		Text           string
 		Error          error
 	}{
@@ -58,18 +60,35 @@ func TestCreateMessage(t *testing.T) {
 			Error:          handler.ErrNotFound,
 		},
 		{
-			Name:           "Valid",
+			Name:           "NoID",
 			ConversationID: cRsp.Conversation.Id,
 			AuthorID:       uuid.New().String(),
 			Text:           "HelloWorld",
 		},
+		{
+			Name:           "WithID",
+			ConversationID: cRsp.Conversation.Id,
+			Text:           "HelloWorld",
+			AuthorID:       "johndoe",
+			ID:             iid,
+		},
+		{
+			Name:           "RepeatID",
+			ConversationID: cRsp.Conversation.Id,
+			Text:           "HelloWorld",
+			AuthorID:       "johndoe",
+			ID:             iid,
+		},
 	}
 
 	for _, tc := range tt {
 		t.Run(tc.Name, func(t *testing.T) {
 			var rsp pb.CreateMessageResponse
 			err := h.CreateMessage(context.TODO(), &pb.CreateMessageRequest{
-				Text: tc.Text, ConversationId: tc.ConversationID, AuthorId: tc.AuthorID,
+				AuthorId:       tc.AuthorID,
+				ConversationId: tc.ConversationID,
+				Text:           tc.Text,
+				Id:             tc.ID,
 			}, &rsp)
 
 			assert.Equal(t, tc.Error, err)
@@ -79,6 +98,7 @@ func TestCreateMessage(t *testing.T) {
 			}
 
 			assertMessagesMatch(t, &pb.Message{
+				Id:             tc.ID,
 				AuthorId:       tc.AuthorID,
 				ConversationId: tc.ConversationID,
 				SentAt:         timestamppb.New(h.Time()),