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), }) }