spark-server/internal/transport/tcp/message.go

package tcp

import (
	"context"
	"encoding/json"

	msglogic "spark-server/internal/module/message/logic"
	syncentity "spark-server/internal/module/sync/entity"
	"spark-server/internal/pkg/errs"
	"spark-server/internal/transport/tcp/protocol"
	"spark-server/internal/transport/tcp/session"
)

type msgIDBody struct {
	MsgID int64 `json:"msg_id"`
}

type readBody struct {
	ConversationID int64 `json:"conversation_id"`
	MsgID          int64 `json:"msg_id"`
	LastMsgID      int64 `json:"last_msg_id"`
}

// readCursor accepts msg_id or last_msg_id. Either value above zero is the read cursor.
func (b readBody) readCursor() int64 {
	if b.MsgID > 0 {
		return b.MsgID
	}
	return b.LastMsgID
}

type syncInboxBody struct {
	MailboxSeq  int64 `json:"mailbox_seq"`
	DialogScore int64 `json:"dialog_score"`
	Limit       int   `json:"limit"`
}

type syncMessagesBody struct {
	ConversationID int64 `json:"conversation_id"`
	LastMsgID      int64 `json:"last_msg_id"`
	Limit          int   `json:"limit"`
}

func (s *Server) handleSendMessage(ctx context.Context, conn *session.Conn, frame protocol.Frame) (*protocol.Frame, error) {
	if s.messages == nil {
		return nil, errs.ErrAvailable
	}
	var in msglogic.SendInput
	if err := json.Unmarshal(frame.Body, &in); err != nil {
		return nil, errs.ErrInvalidArgument
	}
	out, err := s.messages.SendMessage(ctx, conn.UserID(), conn.DeviceID(), &in)
	if err != nil {
		return nil, err
	}
	return &protocol.Frame{
		Version: protocol.CurrentVersion, Command: protocol.CmdSendAck,
		RequestID: frame.RequestID, SequenceID: frame.SequenceID, Body: protocol.EncodeOK(conn.RequestID(), out),
	}, nil
}

func (s *Server) handleReadMessage(ctx context.Context, conn *session.Conn, frame protocol.Frame) (*protocol.Frame, error) {
	if s.messages == nil {
		return nil, errs.ErrAvailable
	}
	var in readBody
	if err := json.Unmarshal(frame.Body, &in); err != nil || in.ConversationID <= 0 || in.readCursor() <= 0 {
		return nil, errs.ErrInvalidArgument
	}
	out, err := s.messages.MarkRead(ctx, conn.UserID(), conn.DeviceID(), &msglogic.ReadInput{
		ConversationID: in.ConversationID, MsgID: in.readCursor(),
	})
	if err != nil {
		return nil, err
	}
	return &protocol.Frame{
		Version: protocol.CurrentVersion, Command: protocol.CmdReadMessage,
		RequestID: frame.RequestID, SequenceID: frame.SequenceID, Body: protocol.EncodeOK(conn.RequestID(), out),
	}, nil
}

func (s *Server) handleRecallMessage(ctx context.Context, conn *session.Conn, frame protocol.Frame) (*protocol.Frame, error) {
	if s.messages == nil {
		return nil, errs.ErrAvailable
	}
	var in msgIDBody
	if err := json.Unmarshal(frame.Body, &in); err != nil || in.MsgID <= 0 {
		return nil, errs.ErrInvalidArgument
	}
	if err := s.messages.RecallMessage(ctx, conn.UserID(), conn.DeviceID(), in.MsgID); err != nil {
		return nil, err
	}
	return &protocol.Frame{
		Version: protocol.CurrentVersion, Command: protocol.CmdRecallMessage,
		RequestID: frame.RequestID, SequenceID: frame.SequenceID, Body: protocol.EncodeOK(conn.RequestID(), in),
	}, nil
}

func (s *Server) handleDeleteMessage(ctx context.Context, conn *session.Conn, frame protocol.Frame) (*protocol.Frame, error) {
	if s.messages == nil {
		return nil, errs.ErrAvailable
	}
	var in msgIDBody
	if err := json.Unmarshal(frame.Body, &in); err != nil || in.MsgID <= 0 {
		return nil, errs.ErrInvalidArgument
	}
	if err := s.messages.DeleteMessage(ctx, conn.UserID(), conn.DeviceID(), in.MsgID); err != nil {
		return nil, err
	}
	return &protocol.Frame{
		Version: protocol.CurrentVersion, Command: protocol.CmdDeleteMessage,
		RequestID: frame.RequestID, SequenceID: frame.SequenceID, Body: protocol.EncodeOK(conn.RequestID(), in),
	}, nil
}

func (s *Server) handleSyncInbox(ctx context.Context, conn *session.Conn, frame protocol.Frame) (*protocol.Frame, error) {
	if s.sync == nil {
		return nil, errs.ErrAvailable
	}
	var in syncInboxBody
	if len(frame.Body) > 0 {
		if err := json.Unmarshal(frame.Body, &in); err != nil {
			return nil, errs.ErrInvalidArgument
		}
	}
	out, err := s.sync.SyncInbox(ctx, syncentity.InboxRequest{
		UserID: conn.UserID(), DeviceID: conn.DeviceID(),
		MailboxSeq: in.MailboxSeq, DialogScore: in.DialogScore, Limit: in.Limit,
	})
	if err != nil {
		return nil, err
	}
	return &protocol.Frame{
		Version: protocol.CurrentVersion, Command: protocol.CmdSyncInbox,
		RequestID: frame.RequestID, SequenceID: frame.SequenceID, Body: protocol.EncodeOK(conn.RequestID(), out),
	}, nil
}

func (s *Server) handleSyncMessages(ctx context.Context, conn *session.Conn, frame protocol.Frame) (*protocol.Frame, error) {
	if s.messages == nil {
		return nil, errs.ErrAvailable
	}
	var in syncMessagesBody
	if err := json.Unmarshal(frame.Body, &in); err != nil || in.ConversationID <= 0 {
		return nil, errs.ErrInvalidArgument
	}
	out, err := s.messages.GetHistory(ctx, conn.UserID(), msglogic.HistoryRequest{
		ConversationID: in.ConversationID, LastMsgID: in.LastMsgID, Limit: in.Limit,
	})
	if err != nil {
		return nil, err
	}
	return &protocol.Frame{
		Version: protocol.CurrentVersion, Command: protocol.CmdSyncMessages,
		RequestID: frame.RequestID, SequenceID: frame.SequenceID, Body: protocol.EncodeOK(conn.RequestID(), out),
	}, nil
}