95 lines
3.0 KiB
Go
95 lines
3.0 KiB
Go
package logic
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
deliverentity "spark-server/internal/module/deliver/entity"
|
|
"spark-server/internal/module/message/entity"
|
|
"spark-server/internal/pkg/errs"
|
|
"spark-server/internal/pkg/reqctx"
|
|
"spark-server/internal/pkg/validate"
|
|
)
|
|
|
|
const cmdMessage uint16 = 6
|
|
|
|
type SendInput struct {
|
|
ConversationID int64 `json:"conversation_id"`
|
|
PeerID int64 `json:"peer_id"`
|
|
ClientMsgID string `json:"client_msg_id" validate:"required,max=64" label:"field.client_msg_id" errmsg:"required:validate.message.client_msg_required"`
|
|
MsgID int64 `json:"msg_id"`
|
|
Content json.RawMessage `json:"content" validate:"required" label:"field.content" errmsg:"required:validate.message.content_required"`
|
|
}
|
|
|
|
type SendResponse struct {
|
|
MsgID int64 `json:"msg_id"`
|
|
ConversationID int64 `json:"conversation_id"`
|
|
ClientMsgID string `json:"client_msg_id"`
|
|
ServerTimeMS int64 `json:"server_time"`
|
|
}
|
|
|
|
func (s *Service) SendMessage(ctx context.Context, userID int64, deviceID uint64, in *SendInput) (*SendResponse, error) {
|
|
if !s.ready() || userID <= 0 {
|
|
return nil, errs.ErrUnauthorized
|
|
}
|
|
if in == nil {
|
|
return nil, errs.ErrInvalidArgument
|
|
}
|
|
if err := validate.StructLang(reqctx.Lang(ctx), in); err != nil {
|
|
return nil, err
|
|
}
|
|
content, err := entity.UnmarshalContent(in.Content)
|
|
if err != nil {
|
|
return nil, errs.ErrInvalidArgument
|
|
}
|
|
conv, memberIDs, err := s.prepareSend(ctx, userID, in)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.checkReplyTo(ctx, conv.ID, content.ReplyTo); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.checkMentions(ctx, userID, conv, content); err != nil {
|
|
return nil, err
|
|
}
|
|
raw, err := content.Marshal()
|
|
if err != nil {
|
|
return nil, errs.ErrInvalidArgument
|
|
}
|
|
now := time.Now()
|
|
msg := &entity.Message{
|
|
ConversationID: conv.ID, SenderID: userID, ClientMsgID: in.ClientMsgID,
|
|
Type: content.Type, Kind: kindOfConv(conv), Content: raw, Status: entity.StatusNormal, CreatedAtMS: now.UnixMilli(),
|
|
}
|
|
if err := s.stampTargets(ctx, userID, conv, memberIDs, msg); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.assignID(in.MsgID, now, msg); err != nil {
|
|
return nil, err
|
|
}
|
|
ack, dup, err := s.persistSend(ctx, msg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.enqueueMessage(ctx, ack, memberIDs, userID, deviceID, dup); err != nil {
|
|
return nil, err
|
|
}
|
|
return &SendResponse{MsgID: ack.ID, ConversationID: ack.ConversationID, ClientMsgID: ack.ClientMsgID, ServerTimeMS: ack.CreatedAtMS}, nil
|
|
}
|
|
|
|
func (s *Service) enqueueMessage(ctx context.Context, msg *entity.Message, memberIDs []int64, userID int64, deviceID uint64, retry bool) error {
|
|
if s.queue == nil {
|
|
return nil
|
|
}
|
|
job := &deliverentity.Job{
|
|
MsgID: msg.ID, Cmd: cmdMessage, ConversationID: msg.ConversationID,
|
|
FromUserID: userID, FromDeviceID: deviceID, ExcludeDevice: deviceID,
|
|
ToUserIDs: memberIDs, Payload: pushBytes(msg), Preview: previewOf(msg),
|
|
}
|
|
if retry {
|
|
job.NotMailbox = true
|
|
}
|
|
return s.queue.Push(ctx, job)
|
|
}
|