实例化消息时对消息进行已读处理

models/message_read_log.go

@@ -0,0 +1,14 @@
+package models
+
+import "gorm.io/gorm"
+
+type MessageReadLog struct {
+	gorm.Model
+
+	MessageId string
+	Reader    string
+}
+
+func (*MessageReadLog) TableName() string {
+	return "tb_session_message_read_log"
+}

repositories/message_read_log_repository.go

@@ -0,0 +1,27 @@
+package repositories
+
+import (
+	"context"
+
+	"gorm.io/gorm"
+	"sikey.com/websocket/models"
+)
+
+type MessageReadLogRepository interface {
+	CreateMessageReadLog(ctx context.Context, readLog *models.MessageReadLog) error
+}
+
+var _ MessageReadLogRepository = (*messageReadLogRepository)(nil)
+
+type messageReadLogRepository struct {
+	source *gorm.DB
+}
+
+// CreateMessageReadLog implements MessageReadLogRepository.
+func (repo *messageReadLogRepository) CreateMessageReadLog(ctx context.Context, readLog *models.MessageReadLog) error {
+	return repo.source.WithContext(ctx).Create(readLog).Error
+}
+
+func NewMessageReadLogRepository(source *gorm.DB) MessageReadLogRepository {
+	return &messageReadLogRepository{source: source}
+}

repositories/repositories.go

@@ -10,17 +10,30 @@ import (
 type TransactionFun func(ctx context.Context, repos *Repositories) error
 
 type Repositories struct {
+	source *gorm.DB
+	rdb    *redis.Client
+
 	SessionRepository         SessionRepository
 	OnlineRepository          OnlineRepository
 	FirebaseMessageRepository FirebaseMessageRepository
 	UnreadCounterRepository   UnreadCounterRepository
+	MessageReadLogRepository  MessageReadLogRepository
 }
 
 func NewRepositories(source *gorm.DB, rdb *redis.Client) *Repositories {
 	return &Repositories{
+		source: source,
+		rdb:    rdb,
+
 		SessionRepository:         NewSessionRepository(source),
 		OnlineRepository:          NewOnlineRepository(rdb),
 		FirebaseMessageRepository: NewFirebaseMessageRepository(source),
-		UnreadCounterRepository:   NewUnreadCounterRepository(rdb),
+		MessageReadLogRepository:  NewMessageReadLogRepository(source),
 	}
 }
+
+func (repos *Repositories) Transaction(ctx context.Context, fn TransactionFun) error {
+	return repos.source.Transaction(func(tx *gorm.DB) error {
+		return fn(ctx, repos)
+	})
+}

server/client.go

@@ -302,13 +302,25 @@ func (c *Client) persistenceMessage(ctx context.Context, message *Message) error
 		return eris.New("unknown message type")
 	}
 
-	err := c.repos.SessionRepository.CreateMessage(c.ctx, &mod)
+	err := c.repos.Transaction(ctx, func(ctx context.Context, repos *repositories.Repositories) error {
+		err := c.repos.SessionRepository.CreateMessage(c.ctx, &mod)
+		if err != nil {
+			zap.L().Error("[persistence] unable to message",
+				zap.Error(err),
+				zap.String("user_id", c.UserId),
+				zap.String("request_id", requestId))
+			return eris.Wrap(err, "unable to persistence message")
+		}
+
+		// 将自己设置为已读
+		err = c.repos.MessageReadLogRepository.CreateMessageReadLog(ctx, &models.MessageReadLog{
+			MessageId: mod.ID,
+			Reader:    mod.Sender,
+		})
+		return err
+	})
 	if err != nil {
-		zap.L().Error("[persistence] unable to message",
-			zap.Error(err),
-			zap.String("user_id", c.UserId),
-			zap.String("request_id", requestId))
-		return eris.Wrap(err, "unable to persistence message")
+		return err
 	}
 
 	zap.L().Info("[persistence] save database end",