spark-server/internal/transport/tcp/message.go
2026-09-18 14:50:41 +08:00

124 lines
4.0 KiB
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 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) 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
}