Chats (#96)

* stash

* replace chats sql with store

* strip unused method

chats/handler/chats.go

@@ -1,14 +1,16 @@
 package handler
 
 import (
-	"encoding/json"
+	"context"
+	"fmt"
+	"sort"
+	"strings"
 	"time"
 
 	pb "github.com/micro/services/chats/proto"
-	"github.com/micro/services/pkg/gorm"
+	"github.com/micro/services/pkg/tenant"
 
 	"github.com/micro/micro/v3/service/errors"
-	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 var (
@@ -21,13 +23,12 @@ var (
 )
 
 type Chats struct {
-	gorm.Helper
 	Time func() time.Time
 }
 
 type Chat struct {
 	ID        string
-	UserIDs   string `gorm:"uniqueIndex"` // sorted json array
+	UserIDs   []string
 	CreatedAt time.Time
 }
 
@@ -45,17 +46,69 @@ func (m *Message) Serialize() *pb.Message {
 		AuthorId: m.AuthorID,
 		ChatId:   m.ChatID,
 		Text:     m.Text,
-		SentAt:   timestamppb.New(m.SentAt),
+		SentAt:   m.SentAt.UnixNano(),
 	}
 }
 
+func (c *Chat) Index(ctx context.Context) string {
+	sort.Strings(c.UserIDs)
+	users := strings.Join(c.UserIDs, "-")
+
+	key := fmt.Sprintf("chatByUserIDs:%s", users)
+
+	t, ok := tenant.FromContext(ctx)
+	if !ok {
+		return key
+	}
+
+	return fmt.Sprintf("%s/%s", t, key)
+}
+
+func (c *Chat) Key(ctx context.Context) string {
+	key := fmt.Sprintf("chat:%s", c.ID)
+
+	t, ok := tenant.FromContext(ctx)
+	if !ok {
+		return key
+	}
+
+	return fmt.Sprintf("%s/%s", t, key)
+}
+
+func (m *Message) Key(ctx context.Context) string {
+	key := fmt.Sprintf("message:%s:%s", m.ChatID, m.ID)
+
+	t, ok := tenant.FromContext(ctx)
+	if !ok {
+		return key
+	}
+
+	return fmt.Sprintf("%s/%s", t, key)
+}
+
+func (m *Message) Index(ctx context.Context) string {
+	key := fmt.Sprintf("messagesByChatID:%s", m.ChatID)
+
+	if !m.SentAt.IsZero() {
+		key = fmt.Sprintf("%s:%d", key, m.SentAt.UnixNano())
+
+		if len(m.ID) > 0 {
+			key = fmt.Sprintf("%s:%s", key, m.ID)
+		}
+	}
+
+	t, ok := tenant.FromContext(ctx)
+	if !ok {
+		return key
+	}
+
+	return fmt.Sprintf("%s/%s", t, key)
+}
+
 func (c *Chat) Serialize() *pb.Chat {
-	var userIDs []string
-	json.Unmarshal([]byte(c.UserIDs), &userIDs)
-
 	return &pb.Chat{
 		Id:        c.ID,
-		UserIds:   userIDs,
-		CreatedAt: timestamppb.New(c.CreatedAt),
+		UserIds:   c.UserIDs,
+		CreatedAt: c.CreatedAt.UnixNano(),
 	}
 }

chats/handler/chats_test.go

@@ -2,38 +2,20 @@ package handler_test
 
 import (
 	"context"
-	"database/sql"
-	"os"
 	"testing"
 	"time"
 
 	"github.com/micro/micro/v3/service/auth"
+	"github.com/micro/micro/v3/service/store"
+	"github.com/micro/micro/v3/service/store/memory"
 	"github.com/micro/services/chats/handler"
 	pb "github.com/micro/services/chats/proto"
 	"github.com/stretchr/testify/assert"
-
-	"github.com/golang/protobuf/ptypes/timestamp"
 )
 
 func testHandler(t *testing.T) *handler.Chats {
-	// connect to the database
-	addr := os.Getenv("POSTGRES_URL")
-	if len(addr) == 0 {
-		addr = "postgresql://postgres@localhost:5432/postgres?sslmode=disable"
-	}
-	sqlDB, err := sql.Open("pgx", addr)
-	if err != nil {
-		t.Fatalf("Failed to open connection to DB %s", err)
-	}
-
-	// clean any data from a previous run
-	if _, err := sqlDB.Exec("DROP TABLE IF EXISTS micro_chats, micro_messages CASCADE"); err != nil {
-		t.Fatalf("Error cleaning database: %v", err)
-	}
-
-	h := &handler.Chats{Time: func() time.Time { return time.Unix(1611327673, 0) }}
-	h.DBConn(sqlDB).Migrations(&handler.Chat{}, &handler.Message{})
-	return h
+	store.DefaultStore = memory.NewStore()
+	return &handler.Chats{Time: func() time.Time { return time.Unix(1611327673, 0) }}
 }
 
 func assertChatsMatch(t *testing.T, exp, act *pb.Chat) {
@@ -52,18 +34,12 @@ func assertChatsMatch(t *testing.T, exp, act *pb.Chat) {
 
 	assert.Equal(t, exp.UserIds, act.UserIds)
 
-	if act.CreatedAt == nil {
+	if act.CreatedAt == 0 {
 		t.Errorf("CreatedAt not set")
 		return
 	}
 
-	assert.True(t, microSecondTime(exp.CreatedAt).Equal(microSecondTime(act.CreatedAt)))
-}
-
-// postgres has a resolution of 100microseconds so just test that it's accurate to the second
-func microSecondTime(t *timestamp.Timestamp) time.Time {
-	tt := t.AsTime()
-	return time.Unix(tt.Unix(), int64(tt.Nanosecond()-tt.Nanosecond()%1000))
+	assert.True(t, exp.CreatedAt == act.CreatedAt)
 }
 
 func assertMessagesMatch(t *testing.T, exp, act *pb.Message) {
@@ -83,11 +59,12 @@ func assertMessagesMatch(t *testing.T, exp, act *pb.Message) {
 	assert.Equal(t, exp.AuthorId, act.AuthorId)
 	assert.Equal(t, exp.ChatId, act.ChatId)
 
-	if act.SentAt == nil {
+	if act.SentAt == 0 {
 		t.Errorf("SentAt not set")
 		return
 	}
-	assert.True(t, microSecondTime(exp.SentAt).Equal(microSecondTime(act.SentAt)))
+
+	assert.True(t, exp.SentAt == act.SentAt)
 }
 
 func microAccountCtx() context.Context {

chats/handler/create_chat.go

@@ -2,8 +2,6 @@ package handler
 
 import (
 	"context"
-	"encoding/json"
-	"regexp"
 	"sort"
 	"time"
 
@@ -11,6 +9,7 @@ import (
 	"github.com/micro/micro/v3/service/auth"
 	"github.com/micro/micro/v3/service/errors"
 	"github.com/micro/micro/v3/service/logger"
+	"github.com/micro/micro/v3/service/store"
 	pb "github.com/micro/services/chats/proto"
 )
 
@@ -26,43 +25,56 @@ func (c *Chats) CreateChat(ctx context.Context, req *pb.CreateChatRequest, rsp *
 		return ErrMissingUserIDs
 	}
 
-	// sort the user ids and then marshal to json
+	// sort the user ids
 	sort.Strings(req.UserIds)
-	bytes, err := json.Marshal(req.UserIds)
-	if err != nil {
-		logger.Errorf("Error mashaling user ids: %v", err)
-		return errors.InternalServerError("ENCODING_ERROR", "Error encoding user ids")
+
+	id := uuid.New().String()
+	if len(req.Id) > 0 {
+		id = req.Id
 	}
 
 	// construct the chat
-	chat := Chat{
-		ID:        uuid.New().String(),
+	chat := &Chat{
+		ID:        id,
 		CreatedAt: time.Now(),
-		UserIDs:   string(bytes),
+		UserIDs:   req.UserIds,
 	}
 
-	db, err := c.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-	// write to the database, if we get a unique key error, the chat already exists
-	err = db.Create(&chat).Error
-	if err == nil {
+	// read the chat by the unique composition of ids
+	recs, err := store.Read(chat.Key(ctx), store.ReadLimit(1))
+	if err == nil && len(recs) == 1 {
+		// found an existing record
+		recs[0].Decode(&chat)
 		rsp.Chat = chat.Serialize()
 		return nil
 	}
 
-	if match, _ := regexp.MatchString(`idx_[\S]+_chats_user_ids`, err.Error()); !match {
+	// if not found check it exists by user index key
+	if err == store.ErrNotFound {
+		recs, err = store.Read(chat.Index(ctx), store.ReadLimit(1))
+		if err == nil && len(recs) > 0 {
+			recs[0].Decode(&chat)
+			rsp.Chat = chat.Serialize()
+			return nil
+		}
+	}
+
+	// ok otherwise we're creating an entirely new record
+	newRec := store.NewRecord(chat.Key(ctx), chat)
+	if err := store.Write(newRec); err != nil {
 		logger.Errorf("Error creating chat: %v", err)
 		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
 	}
 
-	var existing Chat
-	if err := db.Where(&Chat{UserIDs: chat.UserIDs}).First(&existing).Error; err != nil {
-		logger.Errorf("Error reading chat: %v", err)
+	// write the user composite key
+	newRec = store.NewRecord(chat.Index(ctx), chat)
+	if err := store.Write(newRec); err != nil {
+		logger.Errorf("Error creating chat: %v", err)
 		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
 	}
-	rsp.Chat = existing.Serialize()
+
+	// return the record
+	rsp.Chat = chat.Serialize()
+
 	return nil
 }

chats/handler/create_message.go

@@ -2,18 +2,17 @@ package handler
 
 import (
 	"context"
-	"strings"
 
 	"github.com/google/uuid"
 	"github.com/micro/micro/v3/service/auth"
 	"github.com/micro/micro/v3/service/errors"
 	"github.com/micro/micro/v3/service/logger"
+	"github.com/micro/micro/v3/service/store"
 	pb "github.com/micro/services/chats/proto"
-	"gorm.io/gorm"
 )
 
 // Create a message within a chat
-func (c *Chats) CreateMessage(ctx context.Context, req *pb.CreateMessageRequest, rsp *pb.CreateMessageResponse) error {
+func (c *Chats) SendMessage(ctx context.Context, req *pb.SendMessageRequest, rsp *pb.SendMessageResponse) error {
 	_, ok := auth.AccountFromContext(ctx)
 	if !ok {
 		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
@@ -29,14 +28,12 @@ func (c *Chats) CreateMessage(ctx context.Context, req *pb.CreateMessageRequest,
 		return ErrMissingText
 	}
 
-	db, err := c.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
+	chat := &Chat{
+		ID: req.ChatId,
 	}
-	// lookup the chat
-	var conv Chat
-	if err := db.Where(&Chat{ID: req.ChatId}).First(&conv).Error; err == gorm.ErrRecordNotFound {
+
+	recs, err := store.Read(chat.Key(ctx), store.ReadLimit(1))
+	if err == store.ErrNotFound {
 		return ErrNotFound
 	} else if err != nil {
 		logger.Errorf("Error reading chat: %v", err)
@@ -46,28 +43,45 @@ func (c *Chats) CreateMessage(ctx context.Context, req *pb.CreateMessageRequest,
 	// create the message
 	msg := &Message{
 		ID:       req.Id,
-		SentAt:   c.Time(),
 		Text:     req.Text,
 		AuthorID: req.AuthorId,
 		ChatID:   req.ChatId,
+		SentAt:   c.Time(),
 	}
 	if len(msg.ID) == 0 {
 		msg.ID = uuid.New().String()
 	}
-	if err := db.Create(msg).Error; err == nil {
+
+	// check if the message already exists
+	recs, err = store.Read(msg.Key(ctx), store.ReadLimit(1))
+	if err == nil && len(recs) == 1 {
+		// return the existing message
+		msg = &Message{}
+		recs[0].Decode(&msg)
 		rsp.Message = msg.Serialize()
 		return nil
-	} else if !strings.Contains(err.Error(), "messages_pkey") {
+	}
+
+	// if there's an error then return
+	if err != nil && err != store.ErrNotFound {
 		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 := db.Where(&Message{ID: msg.ID}).First(&existing).Error; err != nil {
+	// otherwise write the record
+	if err := store.Write(store.NewRecord(msg.Key(ctx), msg)); err != nil {
 		logger.Errorf("Error creating message: %v", err)
 		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
 	}
-	rsp.Message = existing.Serialize()
+
+	// write the time based index
+	if err := store.Write(store.NewRecord(msg.Index(ctx), msg)); err == nil {
+		rsp.Message = msg.Serialize()
+		return nil
+	} else if err != nil {
+		logger.Errorf("Error creating message: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+
 	return nil
 }

chats/handler/create_message_test.go

@@ -5,13 +5,12 @@ import (
 
 	"github.com/micro/services/chats/handler"
 	pb "github.com/micro/services/chats/proto"
-	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 )
 
-func TestCreateMessage(t *testing.T) {
+func TestSendMessage(t *testing.T) {
 	h := testHandler(t)
 
 	// seed some data
@@ -82,8 +81,8 @@ func TestCreateMessage(t *testing.T) {
 
 	for _, tc := range tt {
 		t.Run(tc.Name, func(t *testing.T) {
-			var rsp pb.CreateMessageResponse
-			err := h.CreateMessage(microAccountCtx(), &pb.CreateMessageRequest{
+			var rsp pb.SendMessageResponse
+			err := h.SendMessage(microAccountCtx(), &pb.SendMessageRequest{
 				AuthorId: tc.AuthorID,
 				ChatId:   tc.ChatID,
 				Text:     tc.Text,
@@ -99,7 +98,7 @@ func TestCreateMessage(t *testing.T) {
 			assertMessagesMatch(t, &pb.Message{
 				AuthorId: tc.AuthorID,
 				ChatId:   tc.ChatID,
-				SentAt:   timestamppb.New(h.Time()),
+				SentAt:   h.Time().UnixNano(),
 				Text:     tc.Text,
 				Id:       tc.ID,
 			}, rsp.Message)

chats/handler/list_messages.go

@@ -6,6 +6,7 @@ import (
 	"github.com/micro/micro/v3/service/auth"
 	"github.com/micro/micro/v3/service/errors"
 	"github.com/micro/micro/v3/service/logger"
+	"github.com/micro/micro/v3/service/store"
 	pb "github.com/micro/services/chats/proto"
 )
 
@@ -23,33 +24,46 @@ func (c *Chats) ListMessages(ctx context.Context, req *pb.ListMessagesRequest, r
 		return ErrMissingChatID
 	}
 
-	db, err := c.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-	// construct the query
-	q := db.Where(&Message{ChatID: req.ChatId}).Order("sent_at DESC")
-	if req.SentBefore != nil {
-		q = q.Where("sent_at < ?", req.SentBefore.AsTime())
-	}
-	if req.Limit != nil {
-		q.Limit(int(req.Limit.Value))
-	} else {
-		q.Limit(DefaultLimit)
+	message := &Message{
+		ChatID: req.ChatId,
 	}
 
-	// execute the query
-	var msgs []Message
-	if err := q.Find(&msgs).Error; err != nil {
+	// default order is descending
+	order := store.OrderDesc
+	if req.Order == "asc" {
+		order = store.OrderAsc
+	}
+
+	opts := []store.ReadOption{
+		store.ReadPrefix(),
+		store.ReadOrder(order),
+	}
+
+	if req.Limit > 0 {
+		opts = append(opts, store.ReadLimit(uint(req.Limit)))
+	} else {
+		opts = append(opts, store.ReadLimit(uint(DefaultLimit)))
+	}
+	if req.Offset > 0 {
+		opts = append(opts, store.ReadOffset(uint(req.Offset)))
+	}
+
+	// read all the records with the chat ID suffix
+	recs, err := store.Read(message.Index(ctx), opts...)
+	if err != nil {
 		logger.Errorf("Error reading messages: %v", err)
 		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
 	}
 
-	// serialize the response
-	rsp.Messages = make([]*pb.Message, len(msgs))
-	for i, m := range msgs {
-		rsp.Messages[i] = m.Serialize()
+	// return all the messages
+	for _, rec := range recs {
+		m := &Message{}
+		rec.Decode(&m)
+		if len(m.ID) == 0 || m.ChatID != req.ChatId {
+			continue
+		}
+		rsp.Messages = append(rsp.Messages, m.Serialize())
 	}
+
 	return nil
 }

chats/handler/list_messages_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/micro/services/chats/handler"
 	pb "github.com/micro/services/chats/proto"
 	"github.com/stretchr/testify/assert"
-	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
 func TestListMessages(t *testing.T) {
@@ -29,8 +28,8 @@ func TestListMessages(t *testing.T) {
 
 	msgs := make([]*pb.Message, 50)
 	for i := 0; i < len(msgs); i++ {
-		var rsp pb.CreateMessageResponse
-		err := h.CreateMessage(microAccountCtx(), &pb.CreateMessageRequest{
+		var rsp pb.SendMessageResponse
+		err := h.SendMessage(microAccountCtx(), &pb.SendMessageRequest{
 			ChatId:   chatRsp.Chat.Id,
 			AuthorId: uuid.New().String(),
 			Text:     strconv.Itoa(i),
@@ -68,7 +67,7 @@ func TestListMessages(t *testing.T) {
 		var rsp pb.ListMessagesResponse
 		err := h.ListMessages(microAccountCtx(), &pb.ListMessagesRequest{
 			ChatId: chatRsp.Chat.Id,
-			Limit:  &wrapperspb.Int32Value{Value: 10},
+			Limit:  10,
 		}, &rsp)
 		assert.NoError(t, err)
 
@@ -86,9 +85,9 @@ func TestListMessages(t *testing.T) {
 	t.Run("OffsetAndLimit", func(t *testing.T) {
 		var rsp pb.ListMessagesResponse
 		err := h.ListMessages(microAccountCtx(), &pb.ListMessagesRequest{
-			ChatId:     chatRsp.Chat.Id,
-			Limit:      &wrapperspb.Int32Value{Value: 5},
-			SentBefore: msgs[20].SentAt,
+			ChatId: chatRsp.Chat.Id,
+			Limit:  5,
+			Offset: 15,
 		}, &rsp)
 		assert.NoError(t, err)
 
@@ -96,7 +95,7 @@ func TestListMessages(t *testing.T) {
 			t.Fatalf("Expected %v messages but got %v", 5, len(rsp.Messages))
 			return
 		}
-		expected := msgs[15:20]
+		expected := msgs[30:35]
 		sortMessages(rsp.Messages)
 		for i, msg := range rsp.Messages {
 			assertMessagesMatch(t, expected[i], msg)
@@ -107,9 +106,9 @@ func TestListMessages(t *testing.T) {
 // sortMessages by the time they were sent
 func sortMessages(msgs []*pb.Message) {
 	sort.Slice(msgs, func(i, j int) bool {
-		if msgs[i].SentAt == nil || msgs[j].SentAt == nil {
+		if msgs[i].SentAt == 0 || msgs[j].SentAt == 0 {
 			return true
 		}
-		return msgs[i].SentAt.AsTime().Before(msgs[j].SentAt.AsTime())
+		return msgs[i].SentAt < msgs[j].SentAt
 	})
 }