Update threads to use model/store (#97)

pkg/model/model.go

@@ -0,0 +1,175 @@
+// package model helps with data modelling on top of the store
+package model
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"github.com/micro/micro/v3/service/store"
+)
+
+var (
+	ErrNotFound      = errors.New("not found")
+	ErrAlreadyExists = errors.New("already exists")
+)
+
+type Entity interface {
+	// The primary key
+	Key(ctx context.Context) string
+	// The index for the entity
+	Index(ctx context.Context) string
+	// The raw value of the entity
+	Value() interface{}
+}
+
+type Query struct {
+	Limit  uint
+	Offset uint
+	Order  string
+}
+
+func Create(ctx context.Context, e Entity) error {
+	key := e.Key(ctx)
+	val := e.Value()
+	idx := e.Index(ctx)
+
+	// read the existing record
+	recs, err := store.Read(key, store.ReadLimit(1))
+	if err != nil && err != store.ErrNotFound {
+		return err
+	}
+
+	if len(recs) > 0 {
+		return ErrAlreadyExists
+	}
+
+	// write the record
+	if err := store.Write(store.NewRecord(key, val)); err != nil {
+		return err
+	}
+
+	// only write the index if it exists
+	if len(idx) == 0 {
+		return nil
+	}
+
+	// write the index
+	return store.Write(store.NewRecord(idx, val))
+}
+
+func ReadIndex(ctx context.Context, e Entity) error {
+	recs, err := store.Read(e.Index(ctx), store.ReadLimit(1))
+	if err == store.ErrNotFound {
+		return ErrNotFound
+	} else if err != nil {
+		return err
+	}
+	if len(recs) == 0 {
+		return ErrNotFound
+	}
+	return recs[0].Decode(e)
+}
+
+func Read(ctx context.Context, e Entity) error {
+	recs, err := store.Read(e.Key(ctx), store.ReadLimit(1))
+	if err == store.ErrNotFound {
+		return ErrNotFound
+	} else if err != nil {
+		return err
+	}
+	if len(recs) == 0 {
+		return ErrNotFound
+	}
+	return recs[0].Decode(e)
+}
+
+func Update(ctx context.Context, e Entity) error {
+	key := e.Key(ctx)
+	val := e.Value()
+	idx := e.Index(ctx)
+
+	// write the record
+	if err := store.Write(store.NewRecord(key, val)); err != nil {
+		return err
+	}
+
+	// only write the index if it exists
+	if len(idx) == 0 {
+		return nil
+	}
+
+	// write the index
+	return store.Write(store.NewRecord(idx, val))
+}
+
+func List(ctx context.Context, e Entity, rsp interface{}, q Query) error {
+	opts := []store.ReadOption{
+		store.ReadPrefix(),
+	}
+
+	if q.Limit > 0 {
+		opts = append(opts, store.ReadLimit(q.Limit))
+	}
+	if q.Offset > 0 {
+		opts = append(opts, store.ReadOffset(q.Offset))
+	}
+	if len(q.Order) > 0 {
+		if q.Order == "desc" {
+			opts = append(opts, store.ReadOrder(store.OrderDesc))
+		} else {
+			opts = append(opts, store.ReadOrder(store.OrderAsc))
+		}
+	}
+
+	recs, err := store.Read(e.Index(ctx), opts...)
+	if err != nil {
+		return err
+	}
+
+	jsBuffer := []byte("[")
+
+	for i, rec := range recs {
+		jsBuffer = append(jsBuffer, rec.Value...)
+		if i < len(recs)-1 {
+			jsBuffer = append(jsBuffer, []byte(",")...)
+		}
+	}
+
+	jsBuffer = append(jsBuffer, []byte("]")...)
+	return json.Unmarshal(jsBuffer, rsp)
+}
+
+func Delete(ctx context.Context, e Entity) error {
+	key := e.Key(ctx)
+	idx := e.Index(ctx)
+
+	if len(key) > 0 {
+		if err := store.Delete(key); err != nil && err != store.ErrNotFound {
+			return err
+		}
+	}
+
+	recs, err := store.Read(idx, store.ReadPrefix())
+	if err != nil && err != store.ErrNotFound {
+		return err
+	}
+
+	// delete every record by index
+	for _, rec := range recs {
+		var val interface{}
+		if err := rec.Decode(val); err != nil || val == nil {
+			continue
+		}
+		// convert to an entity
+		e, ok := val.(Entity)
+		if !ok {
+			continue
+		}
+		if err := store.Delete(e.Key(ctx)); err != store.ErrNotFound {
+			return err
+		}
+	}
+
+	return nil
+}

threads/handler/create_message.go

@@ -2,17 +2,16 @@ 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/services/pkg/model"
 	pb "github.com/micro/services/threads/proto"
-	"gorm.io/gorm"
 )
 
-// Create a message within a conversation
+// Create a message within a thread
 func (s *Threads) CreateMessage(ctx context.Context, req *pb.CreateMessageRequest, rsp *pb.CreateMessageResponse) error {
 	_, ok := auth.AccountFromContext(ctx)
 	if !ok {
@@ -22,24 +21,20 @@ func (s *Threads) CreateMessage(ctx context.Context, req *pb.CreateMessageReques
 	if len(req.AuthorId) == 0 {
 		return ErrMissingAuthorID
 	}
-	if len(req.ConversationId) == 0 {
-		return ErrMissingConversationID
+	if len(req.ThreadId) == 0 {
+		return ErrMissingThreadID
 	}
 	if len(req.Text) == 0 {
 		return ErrMissingText
 	}
 
-	db, err := s.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-	// lookup the conversation
-	var conv Conversation
-	if err := db.Where(&Conversation{ID: req.ConversationId}).First(&conv).Error; err == gorm.ErrRecordNotFound {
+	// lookup the thread
+	conv := Thread{ID: req.ThreadId}
+
+	if err := model.Read(ctx, &conv); err == model.ErrNotFound {
 		return ErrNotFound
 	} else if err != nil {
-		logger.Errorf("Error reading conversation: %v", err)
+		logger.Errorf("Error reading thread: %v", err)
 		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
 	}
 
@@ -49,25 +44,31 @@ func (s *Threads) CreateMessage(ctx context.Context, req *pb.CreateMessageReques
 		SentAt:   s.Time(),
 		Text:     req.Text,
 		AuthorID: req.AuthorId,
-		ConversationID: req.ConversationId,
+		ThreadID: req.ThreadId,
 	}
 	if len(msg.ID) == 0 {
 		msg.ID = uuid.New().String()
 	}
-	if err := db.Create(msg).Error; err == nil {
+
+	if err := model.Create(ctx, msg); err == nil {
 		rsp.Message = msg.Serialize()
 		return nil
-	} else if !strings.Contains(err.Error(), "messages_pkey") {
+	} else if err != model.ErrAlreadyExists {
 		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 {
+	existing := &Message{ID: msg.ID, ThreadID: req.ThreadId}
+
+	if err := model.Read(ctx, existing); err == model.ErrNotFound {
+		return ErrNotFound
+	} else if err != nil {
 		logger.Errorf("Error creating message: %v", err)
 		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
 	}
+
+	// return the message
 	rsp.Message = existing.Serialize()
 	return nil
 }

threads/handler/create_message_test.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/micro/services/threads/handler"
 	pb "github.com/micro/services/threads/proto"
-	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
@@ -15,12 +14,12 @@ func TestCreateMessage(t *testing.T) {
 	h := testHandler(t)
 
 	// seed some data
-	var cRsp pb.CreateConversationResponse
-	err := h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
+	var cRsp pb.CreateThreadResponse
+	err := h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
 		Topic: "HelloWorld", GroupId: uuid.New().String(),
 	}, &cRsp)
 	if err != nil {
-		t.Fatalf("Error creating conversation: %v", err)
+		t.Fatalf("Error creating thread: %v", err)
 		return
 	}
 
@@ -28,52 +27,52 @@ func TestCreateMessage(t *testing.T) {
 	tt := []struct {
 		Name     string
 		AuthorID string
-		ConversationID string
+		ThreadID string
 		ID       string
 		Text     string
 		Error    error
 	}{
 		{
-			Name:     "MissingConversationID",
+			Name:     "MissingThreadID",
 			Text:     "HelloWorld",
 			AuthorID: uuid.New().String(),
-			Error:    handler.ErrMissingConversationID,
+			Error:    handler.ErrMissingThreadID,
 		},
 		{
 			Name:     "MissingAuthorID",
-			ConversationID: uuid.New().String(),
+			ThreadID: uuid.New().String(),
 			Text:     "HelloWorld",
 			Error:    handler.ErrMissingAuthorID,
 		},
 		{
 			Name:     "MissingText",
-			ConversationID: uuid.New().String(),
+			ThreadID: uuid.New().String(),
 			AuthorID: uuid.New().String(),
 			Error:    handler.ErrMissingText,
 		},
 		{
-			Name:           "ConversationNotFound",
-			ConversationID: uuid.New().String(),
+			Name:     "ThreadNotFound",
+			ThreadID: uuid.New().String(),
 			AuthorID: uuid.New().String(),
 			Text:     "HelloWorld",
 			Error:    handler.ErrNotFound,
 		},
 		{
 			Name:     "NoID",
-			ConversationID: cRsp.Conversation.Id,
+			ThreadID: cRsp.Thread.Id,
 			AuthorID: uuid.New().String(),
 			Text:     "HelloWorld",
 		},
 		{
 			Name:     "WithID",
-			ConversationID: cRsp.Conversation.Id,
+			ThreadID: cRsp.Thread.Id,
 			Text:     "HelloWorld",
 			AuthorID: "johndoe",
 			ID:       iid,
 		},
 		{
 			Name:     "RepeatID",
-			ConversationID: cRsp.Conversation.Id,
+			ThreadID: cRsp.Thread.Id,
 			Text:     "HelloWorld",
 			AuthorID: "johndoe",
 			ID:       iid,
@@ -85,7 +84,7 @@ func TestCreateMessage(t *testing.T) {
 			var rsp pb.CreateMessageResponse
 			err := h.CreateMessage(microAccountCtx(), &pb.CreateMessageRequest{
 				AuthorId: tc.AuthorID,
-				ConversationId: tc.ConversationID,
+				ThreadId: tc.ThreadID,
 				Text:     tc.Text,
 				Id:       tc.ID,
 			}, &rsp)
@@ -99,8 +98,8 @@ func TestCreateMessage(t *testing.T) {
 			assertMessagesMatch(t, &pb.Message{
 				Id:       tc.ID,
 				AuthorId: tc.AuthorID,
-				ConversationId: tc.ConversationID,
-				SentAt:         timestamppb.New(h.Time()),
+				ThreadId: tc.ThreadID,
+				SentAt:   handler.FormatTime(h.Time()),
 				Text:     tc.Text,
 			}, rsp.Message)
 		})

threads/handler/create_thread.go

@@ -7,11 +7,12 @@ 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/services/pkg/model"
 	pb "github.com/micro/services/threads/proto"
 )
 
-// Create a conversation
-func (s *Threads) CreateConversation(ctx context.Context, req *pb.CreateConversationRequest, rsp *pb.CreateConversationResponse) error {
+// Create a thread
+func (s *Threads) CreateThread(ctx context.Context, req *pb.CreateThreadRequest, rsp *pb.CreateThreadResponse) error {
 	_, ok := auth.AccountFromContext(ctx)
 	if !ok {
 		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
@@ -24,24 +25,21 @@ func (s *Threads) CreateConversation(ctx context.Context, req *pb.CreateConversa
 		return ErrMissingTopic
 	}
 
-	// write the conversation to the database
-	conv := &Conversation{
+	// write the thread to the database
+	thread := &Thread{
 		ID:        uuid.New().String(),
 		Topic:     req.Topic,
 		GroupID:   req.GroupId,
 		CreatedAt: s.Time(),
 	}
-	db, err := s.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-	if err := db.Create(conv).Error; err != nil {
-		logger.Errorf("Error creating conversation: %v", err)
-		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+
+	// write the thread to the database
+	if err := model.Create(ctx, thread); err != nil {
+		logger.Errorf("Error creating thread: %v", err)
+		return err
 	}
 
 	// serialize the response
-	rsp.Conversation = conv.Serialize()
+	rsp.Thread = thread.Serialize()
 	return nil
 }

threads/handler/create_thread_test.go

@@ -5,13 +5,12 @@ import (
 
 	"github.com/micro/services/threads/handler"
 	pb "github.com/micro/services/threads/proto"
-	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 )
 
-func TestCreateConversation(t *testing.T) {
+func TestCreateThread(t *testing.T) {
 	tt := []struct {
 		Name    string
 		GroupID string
@@ -38,22 +37,22 @@ func TestCreateConversation(t *testing.T) {
 	h := testHandler(t)
 	for _, tc := range tt {
 		t.Run(tc.Name, func(t *testing.T) {
-			var rsp pb.CreateConversationResponse
-			err := h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
+			var rsp pb.CreateThreadResponse
+			err := h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
 				Topic: tc.Topic, GroupId: tc.GroupID,
 			}, &rsp)
 
 			assert.Equal(t, tc.Error, err)
 			if tc.Error != nil {
-				assert.Nil(t, rsp.Conversation)
+				assert.Nil(t, rsp.Thread)
 				return
 			}
 
-			assertConversationsMatch(t, &pb.Conversation{
-				CreatedAt: timestamppb.New(h.Time()),
+			assertThreadsMatch(t, &pb.Thread{
+				CreatedAt: handler.FormatTime(h.Time()),
 				GroupId:   tc.GroupID,
 				Topic:     tc.Topic,
-			}, rsp.Conversation)
+			}, rsp.Thread)
 		})
 	}
 }

threads/handler/delete_conversation.go

@@ -1,44 +0,0 @@
-package handler
-
-import (
-	"context"
-
-	"github.com/micro/micro/v3/service/auth"
-	"github.com/micro/micro/v3/service/errors"
-	"github.com/micro/micro/v3/service/logger"
-	pb "github.com/micro/services/threads/proto"
-	"gorm.io/gorm"
-)
-
-// Delete a conversation and all the messages within
-func (s *Threads) DeleteConversation(ctx context.Context, req *pb.DeleteConversationRequest, rsp *pb.DeleteConversationResponse) error {
-	_, ok := auth.AccountFromContext(ctx)
-	if !ok {
-		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
-	}
-	// validate the request
-	if len(req.Id) == 0 {
-		return ErrMissingID
-	}
-	db, err := s.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-
-	return db.Transaction(func(tx *gorm.DB) error {
-		// delete all the messages
-		if err := tx.Where(&Message{ConversationID: req.Id}).Delete(&Message{}).Error; err != nil {
-			logger.Errorf("Error deleting messages: %v", err)
-			return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
-		}
-
-		// delete the conversation
-		if err := tx.Where(&Conversation{ID: req.Id}).Delete(&Conversation{}).Error; err != nil {
-			logger.Errorf("Error deleting conversation: %v", err)
-			return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
-		}
-
-		return nil
-	})
-}

threads/handler/delete_conversation_test.go

@@ -1,49 +0,0 @@
-package handler_test
-
-import (
-	"testing"
-
-	"github.com/micro/services/threads/handler"
-	pb "github.com/micro/services/threads/proto"
-
-	"github.com/google/uuid"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestDeleteConversation(t *testing.T) {
-	h := testHandler(t)
-
-	// seed some data
-	var cRsp pb.CreateConversationResponse
-	err := h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
-		Topic: "HelloWorld", GroupId: uuid.New().String(),
-	}, &cRsp)
-	if err != nil {
-		t.Fatalf("Error creating conversation: %v", err)
-		return
-	}
-
-	t.Run("MissingID", func(t *testing.T) {
-		err := h.DeleteConversation(microAccountCtx(), &pb.DeleteConversationRequest{}, &pb.DeleteConversationResponse{})
-		assert.Equal(t, handler.ErrMissingID, err)
-	})
-
-	t.Run("Valid", func(t *testing.T) {
-		err := h.DeleteConversation(microAccountCtx(), &pb.DeleteConversationRequest{
-			Id: cRsp.Conversation.Id,
-		}, &pb.DeleteConversationResponse{})
-		assert.NoError(t, err)
-
-		err = h.ReadConversation(microAccountCtx(), &pb.ReadConversationRequest{
-			Id: cRsp.Conversation.Id,
-		}, &pb.ReadConversationResponse{})
-		assert.Equal(t, handler.ErrNotFound, err)
-	})
-
-	t.Run("Retry", func(t *testing.T) {
-		err := h.DeleteConversation(microAccountCtx(), &pb.DeleteConversationRequest{
-			Id: cRsp.Conversation.Id,
-		}, &pb.DeleteConversationResponse{})
-		assert.NoError(t, err)
-	})
-}

threads/handler/delete_thread.go

@@ -0,0 +1,41 @@
+package handler
+
+import (
+	"context"
+
+	"github.com/micro/micro/v3/service/auth"
+	"github.com/micro/micro/v3/service/errors"
+	"github.com/micro/micro/v3/service/logger"
+	"github.com/micro/services/pkg/model"
+	pb "github.com/micro/services/threads/proto"
+)
+
+// Delete a thread and all the messages within
+func (s *Threads) DeleteThread(ctx context.Context, req *pb.DeleteThreadRequest, rsp *pb.DeleteThreadResponse) error {
+	_, ok := auth.AccountFromContext(ctx)
+	if !ok {
+		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
+	}
+	// validate the request
+	if len(req.Id) == 0 {
+		return ErrMissingID
+	}
+
+	thread := Thread{ID: req.Id}
+
+	// delete the thread
+	if err := model.Delete(ctx, &thread); err != nil {
+		logger.Errorf("Error deleting thread: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+
+	message := Message{ThreadID: req.Id}
+
+	// delete the messages
+	if err := model.Delete(ctx, &message); err != nil {
+		logger.Errorf("Error deleting messages: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+
+	return nil
+}

threads/handler/delete_thread_test.go

@@ -0,0 +1,49 @@
+package handler_test
+
+import (
+	"testing"
+
+	"github.com/micro/services/threads/handler"
+	pb "github.com/micro/services/threads/proto"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDeleteThread(t *testing.T) {
+	h := testHandler(t)
+
+	// seed some data
+	var cRsp pb.CreateThreadResponse
+	err := h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
+		Topic: "HelloWorld", GroupId: uuid.New().String(),
+	}, &cRsp)
+	if err != nil {
+		t.Fatalf("Error creating thread: %v", err)
+		return
+	}
+
+	t.Run("MissingID", func(t *testing.T) {
+		err := h.DeleteThread(microAccountCtx(), &pb.DeleteThreadRequest{}, &pb.DeleteThreadResponse{})
+		assert.Equal(t, handler.ErrMissingID, err)
+	})
+
+	t.Run("Valid", func(t *testing.T) {
+		err := h.DeleteThread(microAccountCtx(), &pb.DeleteThreadRequest{
+			Id: cRsp.Thread.Id,
+		}, &pb.DeleteThreadResponse{})
+		assert.NoError(t, err)
+
+		err = h.ReadThread(microAccountCtx(), &pb.ReadThreadRequest{
+			Id: cRsp.Thread.Id,
+		}, &pb.ReadThreadResponse{})
+		assert.Equal(t, handler.ErrNotFound, err)
+	})
+
+	t.Run("Retry", func(t *testing.T) {
+		err := h.DeleteThread(microAccountCtx(), &pb.DeleteThreadRequest{
+			Id: cRsp.Thread.Id,
+		}, &pb.DeleteThreadResponse{})
+		assert.NoError(t, err)
+	})
+}

threads/handler/handler.go

@@ -0,0 +1,144 @@
+package handler
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/micro/micro/v3/service/errors"
+	"github.com/micro/services/pkg/tenant"
+	pb "github.com/micro/services/threads/proto"
+)
+
+var (
+	ErrMissingID        = errors.BadRequest("MISSING_ID", "Missing ID")
+	ErrMissingGroupID   = errors.BadRequest("MISSING_GROUP_ID", "Missing GroupID")
+	ErrMissingTopic     = errors.BadRequest("MISSING_TOPIC", "Missing Topic")
+	ErrMissingAuthorID  = errors.BadRequest("MISSING_AUTHOR_ID", "Missing Author ID")
+	ErrMissingText      = errors.BadRequest("MISSING_TEXT", "Missing text")
+	ErrMissingThreadID  = errors.BadRequest("MISSING_CONVERSATION_ID", "Missing Thread ID")
+	ErrMissingThreadIDs = errors.BadRequest("MISSING_CONVERSATION_IDS", "One or more Thread IDs are required")
+	ErrNotFound         = errors.NotFound("NOT_FOUND", "Thread not found")
+)
+
+type Threads struct {
+	Time func() time.Time
+}
+
+type Message struct {
+	ID       string
+	AuthorID string
+	ThreadID string
+	Text     string
+	SentAt   time.Time
+}
+
+func (m *Message) Serialize() *pb.Message {
+	return &pb.Message{
+		Id:       m.ID,
+		AuthorId: m.AuthorID,
+		ThreadId: m.ThreadID,
+		Text:     m.Text,
+		SentAt:   m.SentAt.Format(time.RFC3339Nano),
+	}
+}
+
+type Thread struct {
+	ID        string
+	GroupID   string
+	Topic     string
+	CreatedAt time.Time
+}
+
+func (c *Thread) Serialize() *pb.Thread {
+	return &pb.Thread{
+		Id:        c.ID,
+		GroupId:   c.GroupID,
+		Topic:     c.Topic,
+		CreatedAt: c.CreatedAt.Format(time.RFC3339Nano),
+	}
+}
+
+func ParseTime(v string) time.Time {
+	t, err := time.Parse(time.RFC3339Nano, v)
+	if err == nil {
+		return t
+	}
+	t, err = time.Parse(time.RFC3339, v)
+	if err == nil {
+		return t
+	}
+	return time.Time{}
+}
+
+func FormatTime(t time.Time) string {
+	return t.Format(time.RFC3339Nano)
+}
+
+func (t *Thread) Key(ctx context.Context) string {
+	if len(t.ID) == 0 {
+		return ""
+	}
+
+	key := fmt.Sprintf("thread:%s", t.ID)
+
+	tnt, ok := tenant.FromContext(ctx)
+	if !ok {
+		return key
+	}
+
+	return fmt.Sprintf("%s/%s", tnt, key)
+}
+
+func (t *Thread) Index(ctx context.Context) string {
+	key := fmt.Sprintf("threadsByGroupID:%s:%s", t.GroupID, t.ID)
+
+	tnt, ok := tenant.FromContext(ctx)
+	if !ok {
+		return key
+	}
+
+	return fmt.Sprintf("%s/%s", tnt, key)
+}
+
+func (t *Thread) Value() interface{} {
+	return t
+}
+
+func (m *Message) Key(ctx context.Context) string {
+	if len(m.ID) == 0 {
+		return ""
+	}
+
+	key := fmt.Sprintf("message:%s:%s", m.ID, m.ThreadID)
+
+	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("messagesByThreadID:%s", m.ThreadID)
+
+	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 (m *Message) Value() interface{} {
+	return m
+}

threads/handler/handler_test.go

@@ -2,43 +2,25 @@ package handler_test
 
 import (
 	"context"
-	"database/sql"
-	"os"
 	"testing"
 	"time"
 
-	"github.com/golang/protobuf/ptypes/timestamp"
 	"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/threads/handler"
 	pb "github.com/micro/services/threads/proto"
 	"github.com/stretchr/testify/assert"
 )
 
 func testHandler(t *testing.T) *handler.Threads {
-	// 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_conversations, micro_messages CASCADE"); err != nil {
-		t.Fatalf("Error cleaning database: %v", err)
-	}
-
-	h := &handler.Threads{Time: func() time.Time { return time.Unix(1611327673, 0) }}
-	h.DBConn(sqlDB).Migrations(&handler.Conversation{}, &handler.Message{})
-
-	return h
+	store.DefaultStore = memory.NewStore()
+	return &handler.Threads{Time: func() time.Time { return time.Unix(1611327673, 0) }}
 }
 
-func assertConversationsMatch(t *testing.T, exp, act *pb.Conversation) {
+func assertThreadsMatch(t *testing.T, exp, act *pb.Thread) {
 	if act == nil {
-		t.Errorf("Conversation not returned")
+		t.Errorf("Thread not returned")
 		return
 	}
 
@@ -53,7 +35,7 @@ func assertConversationsMatch(t *testing.T, exp, act *pb.Conversation) {
 	assert.Equal(t, exp.Topic, act.Topic)
 	assert.Equal(t, exp.GroupId, act.GroupId)
 
-	if act.CreatedAt == nil {
+	if act.CreatedAt == "" {
 		t.Errorf("CreatedAt not set")
 		return
 	}
@@ -77,9 +59,9 @@ func assertMessagesMatch(t *testing.T, exp, act *pb.Message) {
 
 	assert.Equal(t, exp.Text, act.Text)
 	assert.Equal(t, exp.AuthorId, act.AuthorId)
-	assert.Equal(t, exp.ConversationId, act.ConversationId)
+	assert.Equal(t, exp.ThreadId, act.ThreadId)
 
-	if act.SentAt == nil {
+	if act.SentAt == "" {
 		t.Errorf("SentAt not set")
 		return
 	}
@@ -88,8 +70,8 @@ func assertMessagesMatch(t *testing.T, exp, act *pb.Message) {
 }
 
 // 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()
+func microSecondTime(t string) time.Time {
+	tt := handler.ParseTime(t)
 	return time.Unix(tt.Unix(), int64(tt.Nanosecond()-tt.Nanosecond()%1000))
 }
 

threads/handler/list_conversations.go

@@ -1,41 +0,0 @@
-package handler
-
-import (
-	"context"
-
-	"github.com/micro/micro/v3/service/auth"
-	"github.com/micro/micro/v3/service/errors"
-	"github.com/micro/micro/v3/service/logger"
-	pb "github.com/micro/services/threads/proto"
-)
-
-// List all the conversations for a group
-func (s *Threads) ListConversations(ctx context.Context, req *pb.ListConversationsRequest, rsp *pb.ListConversationsResponse) error {
-	_, ok := auth.AccountFromContext(ctx)
-	if !ok {
-		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
-	}
-	// validate the request
-	if len(req.GroupId) == 0 {
-		return ErrMissingGroupID
-	}
-
-	db, err := s.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-	// query the database
-	var convs []Conversation
-	if err := db.Where(&Conversation{GroupID: req.GroupId}).Find(&convs).Error; err != nil {
-		logger.Errorf("Error reading conversation: %v", err)
-		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
-	}
-
-	// serialize the response
-	rsp.Conversations = make([]*pb.Conversation, len(convs))
-	for i, c := range convs {
-		rsp.Conversations[i] = c.Serialize()
-	}
-	return nil
-}

threads/handler/list_conversations_test.go

@@ -1,54 +0,0 @@
-package handler_test
-
-import (
-	"testing"
-
-	"github.com/google/uuid"
-	"github.com/micro/services/threads/handler"
-	pb "github.com/micro/services/threads/proto"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestListConversations(t *testing.T) {
-	h := testHandler(t)
-
-	// seed some data
-	var cRsp1 pb.CreateConversationResponse
-	err := h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
-		Topic: "HelloWorld", GroupId: uuid.New().String(),
-	}, &cRsp1)
-	if err != nil {
-		t.Fatalf("Error creating conversation: %v", err)
-		return
-	}
-	var cRsp2 pb.CreateConversationResponse
-	err = h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
-		Topic: "FooBar", GroupId: uuid.New().String(),
-	}, &cRsp2)
-	if err != nil {
-		t.Fatalf("Error creating conversation: %v", err)
-		return
-	}
-
-	t.Run("MissingGroupID", func(t *testing.T) {
-		var rsp pb.ListConversationsResponse
-		err := h.ListConversations(microAccountCtx(), &pb.ListConversationsRequest{}, &rsp)
-		assert.Equal(t, handler.ErrMissingGroupID, err)
-		assert.Nil(t, rsp.Conversations)
-	})
-
-	t.Run("Valid", func(t *testing.T) {
-		var rsp pb.ListConversationsResponse
-		err := h.ListConversations(microAccountCtx(), &pb.ListConversationsRequest{
-			GroupId: cRsp1.Conversation.GroupId,
-		}, &rsp)
-
-		assert.NoError(t, err)
-		if len(rsp.Conversations) != 1 {
-			t.Fatalf("Expected 1 conversation to be returned, got %v", len(rsp.Conversations))
-			return
-		}
-
-		assertConversationsMatch(t, cRsp1.Conversation, rsp.Conversations[0])
-	})
-}

threads/handler/list_messages.go

@@ -6,12 +6,13 @@ 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/threads/proto"
 )
 
 const DefaultLimit = 25
 
-// List the messages within a conversation in reverse chronological order, using sent_before to
+// List the messages within a thread in reverse chronological order, using sent_before to
 // offset as older messages need to be loaded
 func (s *Threads) ListMessages(ctx context.Context, req *pb.ListMessagesRequest, rsp *pb.ListMessagesResponse) error {
 	_, ok := auth.AccountFromContext(ctx)
@@ -19,37 +20,50 @@ func (s *Threads) ListMessages(ctx context.Context, req *pb.ListMessagesRequest,
 		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
 	}
 	// validate the request
-	if len(req.ConversationId) == 0 {
-		return ErrMissingConversationID
+	if len(req.ThreadId) == 0 {
+		return ErrMissingThreadID
 	}
 
-	db, err := s.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
+	// default order is descending
+	order := store.OrderDesc
+	if req.Order == "asc" {
+		order = store.OrderAsc
 	}
-	// construct the query
-	q := db.Where(&Message{ConversationID: req.ConversationId}).Order("sent_at DESC")
-	if req.SentBefore != nil {
-		q = q.Where("sent_at < ?", req.SentBefore.AsTime())
+
+	opts := []store.ReadOption{
+		store.ReadPrefix(),
+		store.ReadOrder(order),
 	}
-	if req.Limit != nil {
-		q.Limit(int(req.Limit.Value))
+
+	if req.Limit > 0 {
+		opts = append(opts, store.ReadLimit(uint(req.Limit)))
 	} else {
-		q.Limit(DefaultLimit)
+		opts = append(opts, store.ReadLimit(uint(DefaultLimit)))
+	}
+	if req.Offset > 0 {
+		opts = append(opts, store.ReadOffset(uint(req.Offset)))
 	}
 
-	// execute the query
-	var msgs []Message
-	if err := q.Find(&msgs).Error; err != nil {
+	message := &Message{
+		ThreadID: req.ThreadId,
+	}
+
+	// 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.ThreadID != req.ThreadId {
+			continue
 		}
+		rsp.Messages = append(rsp.Messages, m.Serialize())
+	}
+
 	return nil
 }

threads/handler/list_messages_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/micro/services/threads/handler"
 	pb "github.com/micro/services/threads/proto"
 	"github.com/stretchr/testify/assert"
-	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
 func TestListMessages(t *testing.T) {
@@ -18,8 +17,8 @@ func TestListMessages(t *testing.T) {
 	h.Time = time.Now
 
 	// seed some data
-	var convRsp pb.CreateConversationResponse
-	err := h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
+	var convRsp pb.CreateThreadResponse
+	err := h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
 		Topic: "TestListMessages", GroupId: uuid.New().String(),
 	}, &convRsp)
 	assert.NoError(t, err)
@@ -31,7 +30,7 @@ func TestListMessages(t *testing.T) {
 	for i := 0; i < len(msgs); i++ {
 		var rsp pb.CreateMessageResponse
 		err := h.CreateMessage(microAccountCtx(), &pb.CreateMessageRequest{
-			ConversationId: convRsp.Conversation.Id,
+			ThreadId: convRsp.Thread.Id,
 			AuthorId: uuid.New().String(),
 			Text:     strconv.Itoa(i),
 		}, &rsp)
@@ -39,17 +38,17 @@ func TestListMessages(t *testing.T) {
 		msgs[i] = rsp.Message
 	}
 
-	t.Run("MissingConversationID", func(t *testing.T) {
+	t.Run("MissingThreadID", func(t *testing.T) {
 		var rsp pb.ListMessagesResponse
 		err := h.ListMessages(microAccountCtx(), &pb.ListMessagesRequest{}, &rsp)
-		assert.Equal(t, handler.ErrMissingConversationID, err)
+		assert.Equal(t, handler.ErrMissingThreadID, err)
 		assert.Nil(t, rsp.Messages)
 	})
 
 	t.Run("NoOffset", func(t *testing.T) {
 		var rsp pb.ListMessagesResponse
 		err := h.ListMessages(microAccountCtx(), &pb.ListMessagesRequest{
-			ConversationId: convRsp.Conversation.Id,
+			ThreadId: convRsp.Thread.Id,
 		}, &rsp)
 		assert.NoError(t, err)
 
@@ -67,8 +66,8 @@ func TestListMessages(t *testing.T) {
 	t.Run("LimitSet", func(t *testing.T) {
 		var rsp pb.ListMessagesResponse
 		err := h.ListMessages(microAccountCtx(), &pb.ListMessagesRequest{
-			ConversationId: convRsp.Conversation.Id,
-			Limit:          &wrapperspb.Int32Value{Value: 10},
+			ThreadId: convRsp.Thread.Id,
+			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{
-			ConversationId: convRsp.Conversation.Id,
-			Limit:          &wrapperspb.Int32Value{Value: 5},
-			SentBefore:     msgs[20].SentAt,
+			ThreadId: convRsp.Thread.Id,
+			Limit:    5,
+			Offset:   30,
 		}, &rsp)
 		assert.NoError(t, err)
 
@@ -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 == "" || msgs[j].SentAt == "" {
 			return true
 		}
-		return msgs[i].SentAt.AsTime().Before(msgs[j].SentAt.AsTime())
+		return handler.ParseTime(msgs[i].SentAt).Before(handler.ParseTime(msgs[j].SentAt))
 	})
 }

threads/handler/list_threads.go

@@ -0,0 +1,39 @@
+package handler
+
+import (
+	"context"
+
+	"github.com/micro/micro/v3/service/auth"
+	"github.com/micro/micro/v3/service/errors"
+	"github.com/micro/micro/v3/service/logger"
+	"github.com/micro/services/pkg/model"
+	pb "github.com/micro/services/threads/proto"
+)
+
+// List all the threads for a group
+func (s *Threads) ListThreads(ctx context.Context, req *pb.ListThreadsRequest, rsp *pb.ListThreadsResponse) error {
+	_, ok := auth.AccountFromContext(ctx)
+	if !ok {
+		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
+	}
+	// validate the request
+	if len(req.GroupId) == 0 {
+		return ErrMissingGroupID
+	}
+
+	var threads []*Thread
+	thread := &Thread{GroupID: req.GroupId}
+
+	// get all the threads
+	if err := model.List(ctx, thread, &threads, model.Query{}); err != nil {
+		logger.Errorf("Error reading thread: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+
+	// return the response
+	for _, thread := range threads {
+		rsp.Threads = append(rsp.Threads, thread.Serialize())
+	}
+
+	return nil
+}

threads/handler/list_threads_test.go

@@ -0,0 +1,54 @@
+package handler_test
+
+import (
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/micro/services/threads/handler"
+	pb "github.com/micro/services/threads/proto"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestListThreads(t *testing.T) {
+	h := testHandler(t)
+
+	// seed some data
+	var cRsp1 pb.CreateThreadResponse
+	err := h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
+		Topic: "HelloWorld", GroupId: uuid.New().String(),
+	}, &cRsp1)
+	if err != nil {
+		t.Fatalf("Error creating thread: %v", err)
+		return
+	}
+	var cRsp2 pb.CreateThreadResponse
+	err = h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
+		Topic: "FooBar", GroupId: uuid.New().String(),
+	}, &cRsp2)
+	if err != nil {
+		t.Fatalf("Error creating thread: %v", err)
+		return
+	}
+
+	t.Run("MissingGroupID", func(t *testing.T) {
+		var rsp pb.ListThreadsResponse
+		err := h.ListThreads(microAccountCtx(), &pb.ListThreadsRequest{}, &rsp)
+		assert.Equal(t, handler.ErrMissingGroupID, err)
+		assert.Nil(t, rsp.Threads)
+	})
+
+	t.Run("Valid", func(t *testing.T) {
+		var rsp pb.ListThreadsResponse
+		err := h.ListThreads(microAccountCtx(), &pb.ListThreadsRequest{
+			GroupId: cRsp1.Thread.GroupId,
+		}, &rsp)
+
+		assert.NoError(t, err)
+		if len(rsp.Threads) != 1 {
+			t.Fatalf("Expected 1 thread to be returned, got %v", len(rsp.Threads))
+			return
+		}
+
+		assertThreadsMatch(t, cRsp1.Thread, rsp.Threads[0])
+	})
+}

threads/handler/read_conversation.go

@@ -1,48 +0,0 @@
-package handler
-
-import (
-	"context"
-
-	"github.com/micro/micro/v3/service/auth"
-	"gorm.io/gorm"
-
-	"github.com/micro/micro/v3/service/errors"
-	"github.com/micro/micro/v3/service/logger"
-	pb "github.com/micro/services/threads/proto"
-)
-
-// Read a conversation using its ID, can filter using group ID if provided
-func (s *Threads) ReadConversation(ctx context.Context, req *pb.ReadConversationRequest, rsp *pb.ReadConversationResponse) error {
-	_, ok := auth.AccountFromContext(ctx)
-	if !ok {
-		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
-	}
-	// validate the request
-	if len(req.Id) == 0 {
-		return ErrMissingID
-	}
-
-	// construct the query
-	q := Conversation{ID: req.Id}
-	if req.GroupId != nil {
-		q.GroupID = req.GroupId.Value
-	}
-
-	db, err := s.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-	// execute the query
-	var conv Conversation
-	if err := db.Where(&q).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")
-	}
-
-	// serialize the response
-	rsp.Conversation = conv.Serialize()
-	return nil
-}

threads/handler/read_thread.go

@@ -0,0 +1,46 @@
+package handler
+
+import (
+	"context"
+
+	"github.com/micro/micro/v3/service/auth"
+	"github.com/micro/micro/v3/service/errors"
+	"github.com/micro/micro/v3/service/logger"
+	"github.com/micro/services/pkg/model"
+	pb "github.com/micro/services/threads/proto"
+)
+
+// Read a thread using its ID, can filter using group ID if provided
+func (s *Threads) ReadThread(ctx context.Context, req *pb.ReadThreadRequest, rsp *pb.ReadThreadResponse) error {
+	_, ok := auth.AccountFromContext(ctx)
+	if !ok {
+		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
+	}
+	// validate the request
+	if len(req.Id) == 0 {
+		return ErrMissingID
+	}
+
+	// construct the query
+	thread := &Thread{ID: req.Id}
+
+	var err error
+
+	if len(req.GroupId) > 0 {
+		thread.GroupID = req.GroupId
+		err = model.ReadIndex(ctx, thread)
+	} else {
+		err = model.Read(ctx, thread)
+	}
+
+	if err == model.ErrNotFound {
+		return ErrNotFound
+	} else if err != nil {
+		logger.Errorf("Error reading thread: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+
+	// serialize the response
+	rsp.Thread = thread.Serialize()
+	return nil
+}

threads/handler/read_thread_test.go

@@ -7,28 +7,27 @@ import (
 	"github.com/micro/services/threads/handler"
 	pb "github.com/micro/services/threads/proto"
 	"github.com/stretchr/testify/assert"
-	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
-func TestReadConversation(t *testing.T) {
+func TestReadThread(t *testing.T) {
 	h := testHandler(t)
 
 	// seed some data
-	var cRsp pb.CreateConversationResponse
-	err := h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
+	var cRsp pb.CreateThreadResponse
+	err := h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
 		Topic: "HelloWorld", GroupId: uuid.New().String(),
 	}, &cRsp)
 	if err != nil {
-		t.Fatalf("Error creating conversation: %v", err)
+		t.Fatalf("Error creating thread: %v", err)
 		return
 	}
 
 	tt := []struct {
 		Name    string
 		ID      string
-		GroupID *wrapperspb.StringValue
+		GroupID string
 		Error   error
-		Result  *pb.Conversation
+		Result  *pb.Thread
 	}{
 		{
 			Name:  "MissingID",
@@ -41,28 +40,28 @@ func TestReadConversation(t *testing.T) {
 		},
 		{
 			Name:   "FoundUsingIDOnly",
-			ID:     cRsp.Conversation.Id,
-			Result: cRsp.Conversation,
+			ID:     cRsp.Thread.Id,
+			Result: cRsp.Thread,
 		},
 		{
 			Name:    "IncorrectGroupID",
-			ID:      cRsp.Conversation.Id,
+			ID:      cRsp.Thread.Id,
 			Error:   handler.ErrNotFound,
-			GroupID: &wrapperspb.StringValue{Value: uuid.New().String()},
+			GroupID: uuid.New().String(),
 		},
 	}
 	for _, tc := range tt {
 		t.Run(tc.Name, func(t *testing.T) {
-			var rsp pb.ReadConversationResponse
-			err := h.ReadConversation(microAccountCtx(), &pb.ReadConversationRequest{
+			var rsp pb.ReadThreadResponse
+			err := h.ReadThread(microAccountCtx(), &pb.ReadThreadRequest{
 				Id: tc.ID, GroupId: tc.GroupID,
 			}, &rsp)
 			assert.Equal(t, tc.Error, err)
 
 			if tc.Result == nil {
-				assert.Nil(t, rsp.Conversation)
+				assert.Nil(t, rsp.Thread)
 			} else {
-				assertConversationsMatch(t, tc.Result, rsp.Conversation)
+				assertThreadsMatch(t, tc.Result, rsp.Thread)
 			}
 		})
 	}

threads/handler/recent_messages.go

@@ -6,54 +6,42 @@ 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/services/pkg/model"
 	pb "github.com/micro/services/threads/proto"
-	"gorm.io/gorm"
 )
 
-// RecentMessages returns the most recent messages in a group of conversations. By default the
-// most messages retrieved per conversation is 25, however this can be overriden using the
-// limit_per_conversation option
+// RecentMessages returns the most recent messages in a group of threads. By default the
+// most messages retrieved per thread is 25, however this can be overriden using the
+// limit_per_thread option
 func (s *Threads) RecentMessages(ctx context.Context, req *pb.RecentMessagesRequest, rsp *pb.RecentMessagesResponse) error {
 	_, ok := auth.AccountFromContext(ctx)
 	if !ok {
 		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
 	}
 	// validate the request
-	if len(req.ConversationIds) == 0 {
-		return ErrMissingConversationIDs
+	if len(req.ThreadIds) == 0 {
+		return ErrMissingThreadIDs
 	}
 
-	limit := DefaultLimit
-	if req.LimitPerConversation != nil {
-		limit = int(req.LimitPerConversation.Value)
+	limit := uint(DefaultLimit)
+	if req.LimitPerThread > 0 {
+		limit = uint(req.LimitPerThread)
 	}
 
-	db, err := s.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-	// query the database
-	var msgs []Message
-	err = db.Transaction(func(tx *gorm.DB) error {
-		for _, id := range req.ConversationIds {
-			var cms []Message
-			if err := tx.Where(&Message{ConversationID: id}).Order("sent_at DESC").Limit(limit).Find(&cms).Error; err != nil {
-				return err
-			}
-			msgs = append(msgs, cms...)
-		}
-		return nil
-	})
-	if err != nil {
+	for _, thread := range req.ThreadIds {
+		q := model.Query{Limit: limit, Order: "desc"}
+		m := &Message{ThreadID: thread}
+		var messages []*Message
+
+		if err := model.List(ctx, m, &messages, q); 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()
+		for _, msg := range messages {
+			rsp.Messages = append(rsp.Messages, msg.Serialize())
 		}
+	}
+
 	return nil
 }

threads/handler/recent_messages_test.go

@@ -9,7 +9,6 @@ import (
 	"github.com/micro/services/threads/handler"
 	pb "github.com/micro/services/threads/proto"
 	"github.com/stretchr/testify/assert"
-	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
 func TestRecentMessages(t *testing.T) {
@@ -20,8 +19,8 @@ func TestRecentMessages(t *testing.T) {
 	ids := make([]string, 3)
 	convos := make(map[string][]*pb.Message, 3)
 	for i := 0; i < 3; i++ {
-		var convRsp pb.CreateConversationResponse
-		err := h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
+		var convRsp pb.CreateThreadResponse
+		err := h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
 			Topic: "TestRecentMessages", GroupId: uuid.New().String(),
 		}, &convRsp)
 		assert.NoError(t, err)
@@ -29,33 +28,33 @@ func TestRecentMessages(t *testing.T) {
 			return
 		}
 
-		convos[convRsp.Conversation.Id] = make([]*pb.Message, 50)
-		ids[i] = convRsp.Conversation.Id
+		convos[convRsp.Thread.Id] = make([]*pb.Message, 50)
+		ids[i] = convRsp.Thread.Id
 
 		for j := 0; j < 50; j++ {
 			var rsp pb.CreateMessageResponse
 			err := h.CreateMessage(microAccountCtx(), &pb.CreateMessageRequest{
-				ConversationId: convRsp.Conversation.Id,
+				ThreadId: convRsp.Thread.Id,
 				AuthorId: uuid.New().String(),
-				Text:           fmt.Sprintf("Conversation %v, Message %v", i, j),
+				Text:     fmt.Sprintf("Thread %v, Message %v", i, j),
 			}, &rsp)
 			assert.NoError(t, err)
-			convos[convRsp.Conversation.Id][j] = rsp.Message
+			convos[convRsp.Thread.Id][j] = rsp.Message
 		}
 	}
 
-	t.Run("MissingConversationIDs", func(t *testing.T) {
+	t.Run("MissingThreadIDs", func(t *testing.T) {
 		var rsp pb.RecentMessagesResponse
 		err := h.RecentMessages(microAccountCtx(), &pb.RecentMessagesRequest{}, &rsp)
-		assert.Equal(t, handler.ErrMissingConversationIDs, err)
+		assert.Equal(t, handler.ErrMissingThreadIDs, err)
 		assert.Nil(t, rsp.Messages)
 	})
 
 	t.Run("LimitSet", func(t *testing.T) {
 		var rsp pb.RecentMessagesResponse
 		err := h.RecentMessages(microAccountCtx(), &pb.RecentMessagesRequest{
-			ConversationIds:      ids,
-			LimitPerConversation: &wrapperspb.Int32Value{Value: 10},
+			ThreadIds:      ids,
+			LimitPerThread: 10,
 		}, &rsp)
 		assert.NoError(t, err)
 
@@ -79,7 +78,7 @@ func TestRecentMessages(t *testing.T) {
 
 		var rsp pb.RecentMessagesResponse
 		err := h.RecentMessages(microAccountCtx(), &pb.RecentMessagesRequest{
-			ConversationIds: reducedIDs,
+			ThreadIds: reducedIDs,
 		}, &rsp)
 		assert.NoError(t, err)
 

threads/handler/streams.go

@@ -1,60 +0,0 @@
-package handler
-
-import (
-	"time"
-
-	"github.com/micro/micro/v3/service/errors"
-	gorm2 "github.com/micro/services/pkg/gorm"
-	pb "github.com/micro/services/threads/proto"
-	"google.golang.org/protobuf/types/known/timestamppb"
-)
-
-var (
-	ErrMissingID              = errors.BadRequest("MISSING_ID", "Missing ID")
-	ErrMissingGroupID         = errors.BadRequest("MISSING_GROUP_ID", "Missing GroupID")
-	ErrMissingTopic           = errors.BadRequest("MISSING_TOPIC", "Missing Topic")
-	ErrMissingAuthorID        = errors.BadRequest("MISSING_AUTHOR_ID", "Missing Author ID")
-	ErrMissingText            = errors.BadRequest("MISSING_TEXT", "Missing text")
-	ErrMissingConversationID  = errors.BadRequest("MISSING_CONVERSATION_ID", "Missing Conversation ID")
-	ErrMissingConversationIDs = errors.BadRequest("MISSING_CONVERSATION_IDS", "One or more Conversation IDs are required")
-	ErrNotFound               = errors.NotFound("NOT_FOUND", "Conversation not found")
-)
-
-type Threads struct {
-	gorm2.Helper
-	Time func() time.Time
-}
-
-type Message struct {
-	ID             string
-	AuthorID       string
-	ConversationID string
-	Text           string
-	SentAt         time.Time
-}
-
-func (m *Message) Serialize() *pb.Message {
-	return &pb.Message{
-		Id:             m.ID,
-		AuthorId:       m.AuthorID,
-		ConversationId: m.ConversationID,
-		Text:           m.Text,
-		SentAt:         timestamppb.New(m.SentAt),
-	}
-}
-
-type Conversation struct {
-	ID        string
-	GroupID   string
-	Topic     string
-	CreatedAt time.Time
-}
-
-func (c *Conversation) Serialize() *pb.Conversation {
-	return &pb.Conversation{
-		Id:        c.ID,
-		GroupId:   c.GroupID,
-		Topic:     c.Topic,
-		CreatedAt: timestamppb.New(c.CreatedAt),
-	}
-}

threads/handler/update_conversation.go

@@ -1,51 +0,0 @@
-package handler
-
-import (
-	"context"
-
-	"github.com/micro/micro/v3/service/auth"
-	"github.com/micro/micro/v3/service/errors"
-	"github.com/micro/micro/v3/service/logger"
-	pb "github.com/micro/services/threads/proto"
-	"gorm.io/gorm"
-)
-
-// Update a conversations topic
-func (s *Threads) UpdateConversation(ctx context.Context, req *pb.UpdateConversationRequest, rsp *pb.UpdateConversationResponse) error {
-	_, ok := auth.AccountFromContext(ctx)
-	if !ok {
-		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
-	}
-	// validate the request
-	if len(req.Id) == 0 {
-		return ErrMissingID
-	}
-	if len(req.Topic) == 0 {
-		return ErrMissingTopic
-	}
-
-	db, err := s.GetDBConn(ctx)
-	if err != nil {
-		logger.Errorf("Error connecting to DB: %v", err)
-		return errors.InternalServerError("DB_ERROR", "Error connecting to DB")
-	}
-	// lookup the conversation
-	var conv Conversation
-	if err := db.Where(&Conversation{ID: req.Id}).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")
-	}
-
-	// update the conversation
-	conv.Topic = req.Topic
-	if err := db.Save(&conv).Error; err != nil {
-		logger.Errorf("Error updating conversation: %v", err)
-		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
-	}
-
-	// serialize the result
-	rsp.Conversation = conv.Serialize()
-	return nil
-}

threads/handler/update_conversation_test.go

@@ -1,65 +0,0 @@
-package handler_test
-
-import (
-	"testing"
-
-	"github.com/google/uuid"
-	"github.com/micro/services/threads/handler"
-	pb "github.com/micro/services/threads/proto"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestUpdateConversation(t *testing.T) {
-	h := testHandler(t)
-
-	// seed some data
-	var cRsp pb.CreateConversationResponse
-	err := h.CreateConversation(microAccountCtx(), &pb.CreateConversationRequest{
-		Topic: "HelloWorld", GroupId: uuid.New().String(),
-	}, &cRsp)
-	if err != nil {
-		t.Fatalf("Error creating conversation: %v", err)
-		return
-	}
-
-	t.Run("MissingID", func(t *testing.T) {
-		err := h.UpdateConversation(microAccountCtx(), &pb.UpdateConversationRequest{
-			Topic: "NewTopic",
-		}, &pb.UpdateConversationResponse{})
-		assert.Equal(t, handler.ErrMissingID, err)
-	})
-
-	t.Run("MissingTopic", func(t *testing.T) {
-		err := h.UpdateConversation(microAccountCtx(), &pb.UpdateConversationRequest{
-			Id: uuid.New().String(),
-		}, &pb.UpdateConversationResponse{})
-		assert.Equal(t, handler.ErrMissingTopic, err)
-	})
-
-	t.Run("InvalidID", func(t *testing.T) {
-		err := h.UpdateConversation(microAccountCtx(), &pb.UpdateConversationRequest{
-			Id:    uuid.New().String(),
-			Topic: "NewTopic",
-		}, &pb.UpdateConversationResponse{})
-		assert.Equal(t, handler.ErrNotFound, err)
-	})
-
-	t.Run("Valid", func(t *testing.T) {
-		err := h.UpdateConversation(microAccountCtx(), &pb.UpdateConversationRequest{
-			Id:    cRsp.Conversation.Id,
-			Topic: "NewTopic",
-		}, &pb.UpdateConversationResponse{})
-		assert.NoError(t, err)
-
-		var rsp pb.ReadConversationResponse
-		err = h.ReadConversation(microAccountCtx(), &pb.ReadConversationRequest{
-			Id: cRsp.Conversation.Id,
-		}, &rsp)
-		assert.NoError(t, err)
-		if rsp.Conversation == nil {
-			t.Fatal("No conversation returned")
-			return
-		}
-		assert.Equal(t, "NewTopic", rsp.Conversation.Topic)
-	})
-}

threads/handler/update_thread.go

@@ -0,0 +1,47 @@
+package handler
+
+import (
+	"context"
+
+	"github.com/micro/micro/v3/service/auth"
+	"github.com/micro/micro/v3/service/errors"
+	"github.com/micro/micro/v3/service/logger"
+	"github.com/micro/services/pkg/model"
+	pb "github.com/micro/services/threads/proto"
+)
+
+// Update a threads topic
+func (s *Threads) UpdateThread(ctx context.Context, req *pb.UpdateThreadRequest, rsp *pb.UpdateThreadResponse) error {
+	_, ok := auth.AccountFromContext(ctx)
+	if !ok {
+		errors.Unauthorized("UNAUTHORIZED", "Unauthorized")
+	}
+	// validate the request
+	if len(req.Id) == 0 {
+		return ErrMissingID
+	}
+	if len(req.Topic) == 0 {
+		return ErrMissingTopic
+	}
+
+	t := &Thread{ID: req.Id}
+
+	if err := model.Read(ctx, t); err == model.ErrNotFound {
+		return ErrNotFound
+	} else if err != nil {
+		logger.Errorf("Error reading thread: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+
+	// update the thread
+	t.Topic = req.Topic
+	if err := model.Update(ctx, t); err != nil {
+		logger.Errorf("Error updating thread: %v", err)
+		return errors.InternalServerError("DATABASE_ERROR", "Error connecting to database")
+	}
+
+	// serialize the result
+	rsp.Thread = t.Serialize()
+
+	return nil
+}

threads/handler/update_thread_test.go

@@ -0,0 +1,65 @@
+package handler_test
+
+import (
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/micro/services/threads/handler"
+	pb "github.com/micro/services/threads/proto"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUpdateThread(t *testing.T) {
+	h := testHandler(t)
+
+	// seed some data
+	var cRsp pb.CreateThreadResponse
+	err := h.CreateThread(microAccountCtx(), &pb.CreateThreadRequest{
+		Topic: "HelloWorld", GroupId: uuid.New().String(),
+	}, &cRsp)
+	if err != nil {
+		t.Fatalf("Error creating thread: %v", err)
+		return
+	}
+
+	t.Run("MissingID", func(t *testing.T) {
+		err := h.UpdateThread(microAccountCtx(), &pb.UpdateThreadRequest{
+			Topic: "NewTopic",
+		}, &pb.UpdateThreadResponse{})
+		assert.Equal(t, handler.ErrMissingID, err)
+	})
+
+	t.Run("MissingTopic", func(t *testing.T) {
+		err := h.UpdateThread(microAccountCtx(), &pb.UpdateThreadRequest{
+			Id: uuid.New().String(),
+		}, &pb.UpdateThreadResponse{})
+		assert.Equal(t, handler.ErrMissingTopic, err)
+	})
+
+	t.Run("InvalidID", func(t *testing.T) {
+		err := h.UpdateThread(microAccountCtx(), &pb.UpdateThreadRequest{
+			Id:    uuid.New().String(),
+			Topic: "NewTopic",
+		}, &pb.UpdateThreadResponse{})
+		assert.Equal(t, handler.ErrNotFound, err)
+	})
+
+	t.Run("Valid", func(t *testing.T) {
+		err := h.UpdateThread(microAccountCtx(), &pb.UpdateThreadRequest{
+			Id:    cRsp.Thread.Id,
+			Topic: "NewTopic",
+		}, &pb.UpdateThreadResponse{})
+		assert.NoError(t, err)
+
+		var rsp pb.ReadThreadResponse
+		err = h.ReadThread(microAccountCtx(), &pb.ReadThreadRequest{
+			Id: cRsp.Thread.Id,
+		}, &rsp)
+		assert.NoError(t, err)
+		if rsp.Thread == nil {
+			t.Fatal("No thread returned")
+			return
+		}
+		assert.Equal(t, "NewTopic", rsp.Thread.Topic)
+	})
+}

threads/main.go

@@ -1,21 +1,15 @@
 package main
 
 import (
-	"database/sql"
 	"time"
 
 	"github.com/micro/services/threads/handler"
 	pb "github.com/micro/services/threads/proto"
 
 	"github.com/micro/micro/v3/service"
-	"github.com/micro/micro/v3/service/config"
 	"github.com/micro/micro/v3/service/logger"
-
-	_ "github.com/jackc/pgx/v4/stdlib"
 )
 
-var dbAddress = "postgresql://postgres:postgres@localhost:5432/threads?sslmode=disable"
-
 func main() {
 	// Create service
 	srv := service.New(
@@ -23,19 +17,7 @@ func main() {
 		service.Version("latest"),
 	)
 
-	// Connect to the database
-	cfg, err := config.Get("threads.database")
-	if err != nil {
-		logger.Fatalf("Error loading config: %v", err)
-	}
-	addr := cfg.String(dbAddress)
-	sqlDB, err := sql.Open("pgx", addr)
-	if err != nil {
-		logger.Fatalf("Failed to open connection to DB %s", err)
-	}
-
 	h := &handler.Threads{Time: time.Now}
-	h.DBConn(sqlDB).Migrations(&handler.Conversation{}, &handler.Message{})
 	// Register handler
 	pb.RegisterThreadsHandler(srv.Server(), h)
 

threads/proto/threads.pb.micro.go

@@ -6,8 +6,6 @@ package threads
 import (
 	fmt "fmt"
 	proto "github.com/golang/protobuf/proto"
-	_ "google.golang.org/protobuf/types/known/timestamppb"
-	_ "google.golang.org/protobuf/types/known/wrapperspb"
 	math "math"
 )
 
@@ -44,24 +42,24 @@ func NewThreadsEndpoints() []*api.Endpoint {
 // Client API for Threads service
 
 type ThreadsService interface {
-	// Create a conversation
-	CreateConversation(ctx context.Context, in *CreateConversationRequest, opts ...client.CallOption) (*CreateConversationResponse, error)
-	// Read a conversation using its ID, can filter using group ID if provided
-	ReadConversation(ctx context.Context, in *ReadConversationRequest, opts ...client.CallOption) (*ReadConversationResponse, error)
-	// Update a conversations topic
-	UpdateConversation(ctx context.Context, in *UpdateConversationRequest, opts ...client.CallOption) (*UpdateConversationResponse, error)
-	// Delete a conversation and all the messages within
-	DeleteConversation(ctx context.Context, in *DeleteConversationRequest, opts ...client.CallOption) (*DeleteConversationResponse, error)
-	// List all the conversations for a group
-	ListConversations(ctx context.Context, in *ListConversationsRequest, opts ...client.CallOption) (*ListConversationsResponse, error)
-	// Create a message within a conversation
+	// Create a thread
+	CreateThread(ctx context.Context, in *CreateThreadRequest, opts ...client.CallOption) (*CreateThreadResponse, error)
+	// Read a thread using its ID, can filter using group ID if provided
+	ReadThread(ctx context.Context, in *ReadThreadRequest, opts ...client.CallOption) (*ReadThreadResponse, error)
+	// Update a threads topic
+	UpdateThread(ctx context.Context, in *UpdateThreadRequest, opts ...client.CallOption) (*UpdateThreadResponse, error)
+	// Delete a thread and all the messages within
+	DeleteThread(ctx context.Context, in *DeleteThreadRequest, opts ...client.CallOption) (*DeleteThreadResponse, error)
+	// List all the threads for a group
+	ListThreads(ctx context.Context, in *ListThreadsRequest, opts ...client.CallOption) (*ListThreadsResponse, error)
+	// Create a message within a thread
 	CreateMessage(ctx context.Context, in *CreateMessageRequest, opts ...client.CallOption) (*CreateMessageResponse, error)
-	// List the messages within a conversation in reverse chronological order, using sent_before to
+	// List the messages within a thread in reverse chronological order, using sent_before to
 	// offset as older messages need to be loaded
 	ListMessages(ctx context.Context, in *ListMessagesRequest, opts ...client.CallOption) (*ListMessagesResponse, error)
-	// RecentMessages returns the most recent messages in a group of conversations. By default the
-	// most messages retrieved per conversation is 25, however this can be overriden using the
-	// limit_per_conversation option
+	// RecentMessages returns the most recent messages in a group of threads. By default the
+	// most messages retrieved per thread is 25, however this can be overriden using the
+	// limit_per_thread option
 	RecentMessages(ctx context.Context, in *RecentMessagesRequest, opts ...client.CallOption) (*RecentMessagesResponse, error)
 }
 
@@ -77,9 +75,9 @@ func NewThreadsService(name string, c client.Client) ThreadsService {
 	}
 }
 
-func (c *threadsService) CreateConversation(ctx context.Context, in *CreateConversationRequest, opts ...client.CallOption) (*CreateConversationResponse, error) {
-	req := c.c.NewRequest(c.name, "Threads.CreateConversation", in)
-	out := new(CreateConversationResponse)
+func (c *threadsService) CreateThread(ctx context.Context, in *CreateThreadRequest, opts ...client.CallOption) (*CreateThreadResponse, error) {
+	req := c.c.NewRequest(c.name, "Threads.CreateThread", in)
+	out := new(CreateThreadResponse)
 	err := c.c.Call(ctx, req, out, opts...)
 	if err != nil {
 		return nil, err
@@ -87,9 +85,9 @@ func (c *threadsService) CreateConversation(ctx context.Context, in *CreateConve
 	return out, nil
 }
 
-func (c *threadsService) ReadConversation(ctx context.Context, in *ReadConversationRequest, opts ...client.CallOption) (*ReadConversationResponse, error) {
-	req := c.c.NewRequest(c.name, "Threads.ReadConversation", in)
-	out := new(ReadConversationResponse)
+func (c *threadsService) ReadThread(ctx context.Context, in *ReadThreadRequest, opts ...client.CallOption) (*ReadThreadResponse, error) {
+	req := c.c.NewRequest(c.name, "Threads.ReadThread", in)
+	out := new(ReadThreadResponse)
 	err := c.c.Call(ctx, req, out, opts...)
 	if err != nil {
 		return nil, err
@@ -97,9 +95,9 @@ func (c *threadsService) ReadConversation(ctx context.Context, in *ReadConversat
 	return out, nil
 }
 
-func (c *threadsService) UpdateConversation(ctx context.Context, in *UpdateConversationRequest, opts ...client.CallOption) (*UpdateConversationResponse, error) {
-	req := c.c.NewRequest(c.name, "Threads.UpdateConversation", in)
-	out := new(UpdateConversationResponse)
+func (c *threadsService) UpdateThread(ctx context.Context, in *UpdateThreadRequest, opts ...client.CallOption) (*UpdateThreadResponse, error) {
+	req := c.c.NewRequest(c.name, "Threads.UpdateThread", in)
+	out := new(UpdateThreadResponse)
 	err := c.c.Call(ctx, req, out, opts...)
 	if err != nil {
 		return nil, err
@@ -107,9 +105,9 @@ func (c *threadsService) UpdateConversation(ctx context.Context, in *UpdateConve
 	return out, nil
 }
 
-func (c *threadsService) DeleteConversation(ctx context.Context, in *DeleteConversationRequest, opts ...client.CallOption) (*DeleteConversationResponse, error) {
-	req := c.c.NewRequest(c.name, "Threads.DeleteConversation", in)
-	out := new(DeleteConversationResponse)
+func (c *threadsService) DeleteThread(ctx context.Context, in *DeleteThreadRequest, opts ...client.CallOption) (*DeleteThreadResponse, error) {
+	req := c.c.NewRequest(c.name, "Threads.DeleteThread", in)
+	out := new(DeleteThreadResponse)
 	err := c.c.Call(ctx, req, out, opts...)
 	if err != nil {
 		return nil, err
@@ -117,9 +115,9 @@ func (c *threadsService) DeleteConversation(ctx context.Context, in *DeleteConve
 	return out, nil
 }
 
-func (c *threadsService) ListConversations(ctx context.Context, in *ListConversationsRequest, opts ...client.CallOption) (*ListConversationsResponse, error) {
-	req := c.c.NewRequest(c.name, "Threads.ListConversations", in)
-	out := new(ListConversationsResponse)
+func (c *threadsService) ListThreads(ctx context.Context, in *ListThreadsRequest, opts ...client.CallOption) (*ListThreadsResponse, error) {
+	req := c.c.NewRequest(c.name, "Threads.ListThreads", in)
+	out := new(ListThreadsResponse)
 	err := c.c.Call(ctx, req, out, opts...)
 	if err != nil {
 		return nil, err
@@ -160,34 +158,34 @@ func (c *threadsService) RecentMessages(ctx context.Context, in *RecentMessagesR
 // Server API for Threads service
 
 type ThreadsHandler interface {
-	// Create a conversation
-	CreateConversation(context.Context, *CreateConversationRequest, *CreateConversationResponse) error
-	// Read a conversation using its ID, can filter using group ID if provided
-	ReadConversation(context.Context, *ReadConversationRequest, *ReadConversationResponse) error
-	// Update a conversations topic
-	UpdateConversation(context.Context, *UpdateConversationRequest, *UpdateConversationResponse) error
-	// Delete a conversation and all the messages within
-	DeleteConversation(context.Context, *DeleteConversationRequest, *DeleteConversationResponse) error
-	// List all the conversations for a group
-	ListConversations(context.Context, *ListConversationsRequest, *ListConversationsResponse) error
-	// Create a message within a conversation
+	// Create a thread
+	CreateThread(context.Context, *CreateThreadRequest, *CreateThreadResponse) error
+	// Read a thread using its ID, can filter using group ID if provided
+	ReadThread(context.Context, *ReadThreadRequest, *ReadThreadResponse) error
+	// Update a threads topic
+	UpdateThread(context.Context, *UpdateThreadRequest, *UpdateThreadResponse) error
+	// Delete a thread and all the messages within
+	DeleteThread(context.Context, *DeleteThreadRequest, *DeleteThreadResponse) error
+	// List all the threads for a group
+	ListThreads(context.Context, *ListThreadsRequest, *ListThreadsResponse) error
+	// Create a message within a thread
 	CreateMessage(context.Context, *CreateMessageRequest, *CreateMessageResponse) error
-	// List the messages within a conversation in reverse chronological order, using sent_before to
+	// List the messages within a thread in reverse chronological order, using sent_before to
 	// offset as older messages need to be loaded
 	ListMessages(context.Context, *ListMessagesRequest, *ListMessagesResponse) error
-	// RecentMessages returns the most recent messages in a group of conversations. By default the
-	// most messages retrieved per conversation is 25, however this can be overriden using the
-	// limit_per_conversation option
+	// RecentMessages returns the most recent messages in a group of threads. By default the
+	// most messages retrieved per thread is 25, however this can be overriden using the
+	// limit_per_thread option
 	RecentMessages(context.Context, *RecentMessagesRequest, *RecentMessagesResponse) error
 }
 
 func RegisterThreadsHandler(s server.Server, hdlr ThreadsHandler, opts ...server.HandlerOption) error {
 	type threads interface {
-		CreateConversation(ctx context.Context, in *CreateConversationRequest, out *CreateConversationResponse) error
-		ReadConversation(ctx context.Context, in *ReadConversationRequest, out *ReadConversationResponse) error
-		UpdateConversation(ctx context.Context, in *UpdateConversationRequest, out *UpdateConversationResponse) error
-		DeleteConversation(ctx context.Context, in *DeleteConversationRequest, out *DeleteConversationResponse) error
-		ListConversations(ctx context.Context, in *ListConversationsRequest, out *ListConversationsResponse) error
+		CreateThread(ctx context.Context, in *CreateThreadRequest, out *CreateThreadResponse) error
+		ReadThread(ctx context.Context, in *ReadThreadRequest, out *ReadThreadResponse) error
+		UpdateThread(ctx context.Context, in *UpdateThreadRequest, out *UpdateThreadResponse) error
+		DeleteThread(ctx context.Context, in *DeleteThreadRequest, out *DeleteThreadResponse) error
+		ListThreads(ctx context.Context, in *ListThreadsRequest, out *ListThreadsResponse) error
 		CreateMessage(ctx context.Context, in *CreateMessageRequest, out *CreateMessageResponse) error
 		ListMessages(ctx context.Context, in *ListMessagesRequest, out *ListMessagesResponse) error
 		RecentMessages(ctx context.Context, in *RecentMessagesRequest, out *RecentMessagesResponse) error
@@ -203,24 +201,24 @@ type threadsHandler struct {
 	ThreadsHandler
 }
 
-func (h *threadsHandler) CreateConversation(ctx context.Context, in *CreateConversationRequest, out *CreateConversationResponse) error {
-	return h.ThreadsHandler.CreateConversation(ctx, in, out)
+func (h *threadsHandler) CreateThread(ctx context.Context, in *CreateThreadRequest, out *CreateThreadResponse) error {
+	return h.ThreadsHandler.CreateThread(ctx, in, out)
 }
 
-func (h *threadsHandler) ReadConversation(ctx context.Context, in *ReadConversationRequest, out *ReadConversationResponse) error {
-	return h.ThreadsHandler.ReadConversation(ctx, in, out)
+func (h *threadsHandler) ReadThread(ctx context.Context, in *ReadThreadRequest, out *ReadThreadResponse) error {
+	return h.ThreadsHandler.ReadThread(ctx, in, out)
 }
 
-func (h *threadsHandler) UpdateConversation(ctx context.Context, in *UpdateConversationRequest, out *UpdateConversationResponse) error {
-	return h.ThreadsHandler.UpdateConversation(ctx, in, out)
+func (h *threadsHandler) UpdateThread(ctx context.Context, in *UpdateThreadRequest, out *UpdateThreadResponse) error {
+	return h.ThreadsHandler.UpdateThread(ctx, in, out)
 }
 
-func (h *threadsHandler) DeleteConversation(ctx context.Context, in *DeleteConversationRequest, out *DeleteConversationResponse) error {
-	return h.ThreadsHandler.DeleteConversation(ctx, in, out)
+func (h *threadsHandler) DeleteThread(ctx context.Context, in *DeleteThreadRequest, out *DeleteThreadResponse) error {
+	return h.ThreadsHandler.DeleteThread(ctx, in, out)
 }
 
-func (h *threadsHandler) ListConversations(ctx context.Context, in *ListConversationsRequest, out *ListConversationsResponse) error {
-	return h.ThreadsHandler.ListConversations(ctx, in, out)
+func (h *threadsHandler) ListThreads(ctx context.Context, in *ListThreadsRequest, out *ListThreadsResponse) error {
+	return h.ThreadsHandler.ListThreads(ctx, in, out)
 }
 
 func (h *threadsHandler) CreateMessage(ctx context.Context, in *CreateMessageRequest, out *CreateMessageResponse) error {

threads/proto/threads.proto

@@ -1,91 +1,90 @@
 syntax = "proto3";
 
 package threads;
+
 option go_package = "./proto;threads";
-import "google/protobuf/timestamp.proto";
-import "google/protobuf/wrappers.proto";
 
 service Threads {
-	// Create a conversation
-	rpc CreateConversation(CreateConversationRequest) returns (CreateConversationResponse);
-	// Read a conversation using its ID, can filter using group ID if provided
-	rpc ReadConversation(ReadConversationRequest) returns (ReadConversationResponse);
-	// Update a conversations topic
-	rpc UpdateConversation(UpdateConversationRequest) returns (UpdateConversationResponse);
-	// Delete a conversation and all the messages within
-	rpc DeleteConversation(DeleteConversationRequest) returns (DeleteConversationResponse);
-	// List all the conversations for a group
-	rpc ListConversations(ListConversationsRequest) returns (ListConversationsResponse);
-	// Create a message within a conversation
+	// Create a thread
+	rpc CreateThread(CreateThreadRequest) returns (CreateThreadResponse);
+	// Read a thread using its ID, can filter using group ID if provided
+	rpc ReadThread(ReadThreadRequest) returns (ReadThreadResponse);
+	// Update a threads topic
+	rpc UpdateThread(UpdateThreadRequest) returns (UpdateThreadResponse);
+	// Delete a thread and all the messages within
+	rpc DeleteThread(DeleteThreadRequest) returns (DeleteThreadResponse);
+	// List all the threads for a group
+	rpc ListThreads(ListThreadsRequest) returns (ListThreadsResponse);
+	// Create a message within a thread
 	rpc CreateMessage(CreateMessageRequest) returns (CreateMessageResponse);
-	// List the messages within a conversation in reverse chronological order, using sent_before to
+	// List the messages within a thread in reverse chronological order, using sent_before to
 	// offset as older messages need to be loaded
 	rpc ListMessages(ListMessagesRequest) returns (ListMessagesResponse);
-	// RecentMessages returns the most recent messages in a group of conversations. By default the 
-	// most messages retrieved per conversation is 25, however this can be overriden using the 
-	// limit_per_conversation option
+	// RecentMessages returns the most recent messages in a group of threads. By default the 
+	// most messages retrieved per thread is 25, however this can be overriden using the 
+	// limit_per_thread option
 	rpc RecentMessages(RecentMessagesRequest) returns (RecentMessagesResponse);
 }
 
-message Conversation {
+message Thread {
 	string id = 1;
 	string group_id = 2;
 	string topic = 3;
-	google.protobuf.Timestamp created_at = 4;
+	string created_at = 4;
 }
 
 message Message {
 	string id = 1;
 	string author_id = 2;
-	string conversation_id = 3;
+	string thread_id = 3;
 	string text = 4;
-	google.protobuf.Timestamp sent_at = 5;
+	string sent_at = 5;
 }
 
-message CreateConversationRequest {
+message CreateThreadRequest {
 	string group_id = 1;
 	string topic = 2;
 }
 
-message CreateConversationResponse {
-	Conversation conversation = 1;
+message CreateThreadResponse {
+	Thread thread = 1;
 }
 
-message ReadConversationRequest {
+message ReadThreadRequest {
 	string id = 1;
-	google.protobuf.StringValue group_id = 2;
+	string group_id = 2;
 }
 
-message ReadConversationResponse {
-	Conversation conversation = 1;
+message ReadThreadResponse {
+	Thread thread = 1;
 }
 
-message ListConversationsRequest {
+message ListThreadsRequest {
 	string group_id = 1;
 }
 
-message ListConversationsResponse {
-	repeated Conversation conversations = 1;
+message ListThreadsResponse {
+	repeated Thread threads = 1;
 }
 
-message UpdateConversationRequest {
+message UpdateThreadRequest {
 	string id = 1;
 	string topic = 2;
 }
 
-message UpdateConversationResponse {
-	Conversation conversation = 1;
+message UpdateThreadResponse {
+	Thread thread = 1;
 }
 
-message DeleteConversationRequest {
+message DeleteThreadRequest {
 	string id = 1;
 }
 
-message DeleteConversationResponse {}
+message DeleteThreadResponse {}
 
 message CreateMessageRequest {
 	string id = 1;
-	string conversation_id = 2;
+	string thread_id = 2;
 	string author_id = 3;
 	string text = 4;
 }
@@ -95,9 +94,10 @@ message CreateMessageResponse {
 }
 
 message ListMessagesRequest {
-	string conversation_id = 1;
-	google.protobuf.Timestamp sent_before = 2;
-	google.protobuf.Int32Value limit = 3;
+	string thread_id = 1;
+	int64 limit = 2;
+	int64 offset = 3;
+	string order = 4;
 }
 
 message ListMessagesResponse {
@@ -105,8 +105,8 @@ message ListMessagesResponse {
 }
 
 message RecentMessagesRequest {
-	repeated string conversation_ids = 1;
-	google.protobuf.Int32Value limit_per_conversation = 2;
+	repeated string thread_ids = 1;
+	int64 limit_per_thread = 2;
 }
 
 message RecentMessagesResponse {