spark-server/internal/module/message/logic/read.go

package logic

import (
	"context"
	"errors"

	conventity "spark-server/internal/module/conversation/entity"
	deliverentity "spark-server/internal/module/deliver/entity"
	"spark-server/internal/pkg/errs"
)

const cmdRead uint16 = 8

// ReadInput is TCP command 8. msg_id is the message the member has read.
type ReadInput struct {
	ConversationID int64 `json:"conversation_id"`
	MsgID          int64 `json:"msg_id"`
}

// ReadResult is the stored cursor. msg_id is last_read_msg_id and never moves backward.
type ReadResult struct {
	ConversationID int64 `json:"conversation_id"`
	MsgID          int64 `json:"msg_id"`
}

func (s *Service) MarkRead(ctx context.Context, userID int64, deviceID uint64, in *ReadInput) (*ReadResult, error) {
	if !s.ready() || userID <= 0 {
		return nil, errs.ErrUnauthorized
	}
	if in == nil || in.ConversationID <= 0 || in.MsgID <= 0 {
		return nil, errs.ErrInvalidArgument
	}
	mem, err := s.readMember(ctx, in.ConversationID, userID)
	if err != nil {
		return nil, err
	}
	if err := s.readTarget(ctx, in); err != nil {
		return nil, err
	}
	cursor, advanced, err := s.advanceRead(ctx, mem, in)
	if err != nil {
		return nil, err
	}
	if err := s.clearDialogUnread(ctx, userID, in.ConversationID); err != nil {
		return nil, err
	}
	if advanced {
		if err := s.enqueueRead(ctx, userID, deviceID, in.ConversationID, cursor); err != nil {
			return nil, err
		}
	}
	return &ReadResult{ConversationID: in.ConversationID, MsgID: cursor}, nil
}

func (s *Service) readMember(ctx context.Context, conversationID, userID int64) (*conventity.Member, error) {
	mem, err := s.chats.GetMember(ctx, conversationID, userID)
	if err != nil {
		if errors.Is(err, errs.ErrNotFound) {
			return nil, errs.ErrForbidden
		}
		return nil, err
	}
	return mem, nil
}

func (s *Service) readTarget(ctx context.Context, in *ReadInput) error {
	msg, err := s.repo.GetByID(ctx, in.MsgID)
	if err != nil {
		return err
	}
	if msg.ConversationID != in.ConversationID {
		return errs.ErrInvalidArgument
	}
	return nil
}

func (s *Service) advanceRead(ctx context.Context, mem *conventity.Member, in *ReadInput) (int64, bool, error) {
	if mem.LastReadMsgID >= in.MsgID {
		return mem.LastReadMsgID, false, nil
	}
	if err := s.chats.UpdateLastRead(ctx, in.ConversationID, mem.UserID, in.MsgID); err != nil {
		if errors.Is(err, errs.ErrNotFound) {
			return 0, false, errs.ErrForbidden
		}
		return 0, false, err
	}
	return in.MsgID, true, nil
}

func (s *Service) clearDialogUnread(ctx context.Context, userID, conversationID int64) error {
	if s.inbox == nil {
		return nil
	}
	var score int64
	if conv, err := s.chats.Get(ctx, conversationID); err == nil && conv != nil {
		score = conv.LastMsgAt
	}
	if err := s.inbox.ClearUnread(ctx, userID, conversationID, score); err != nil {
		return errs.ErrInner.WithError("inbox", err)
	}
	return nil
}

func (s *Service) enqueueRead(ctx context.Context, userID int64, deviceID uint64, conversationID, cursor int64) error {
	if s.queue == nil {
		return nil
	}
	jobID, err := s.ids.NextID()
	if err != nil {
		return err
	}
	return s.queue.Push(ctx, &deliverentity.Job{
		MsgID: jobID, Cmd: cmdRead, ConversationID: conversationID,
		FromUserID: userID, FromDeviceID: deviceID, ExcludeDevice: deviceID,
		ToUserIDs: []int64{userID}, NotPush: true, NotMailbox: true,
		Payload: controlBytes(cursor, conversationID),
	})
}