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