112 lines
3.0 KiB
Go
112 lines
3.0 KiB
Go
// Package logic is SYNC_INBOX. HTTP and TCP share it.
|
|
package logic
|
|
|
|
import (
|
|
"context"
|
|
|
|
inboxentity "spark-server/internal/module/inbox/entity"
|
|
inboxlogic "spark-server/internal/module/inbox/logic"
|
|
"spark-server/internal/module/sync/entity"
|
|
"spark-server/internal/pkg/errs"
|
|
)
|
|
|
|
const defaultInboxLimit = 200
|
|
|
|
type Mailbox interface {
|
|
Put(ctx context.Context, userID int64, eventType entity.EventType, conversationID int64, payload []byte) error
|
|
ListSince(ctx context.Context, userID int64, deviceID uint64, seq int64, limit int) ([]*entity.MailboxItem, error)
|
|
SaveCursor(ctx context.Context, userID int64, deviceID uint64, mailboxSeq, dialogScore int64) error
|
|
PurgeUser(ctx context.Context, userID int64) error
|
|
}
|
|
|
|
type InboxResponse struct {
|
|
Dialogs []inboxentity.Dialog `json:"dialogs"`
|
|
Mailbox []*entity.MailboxItem `json:"mailbox"`
|
|
DialogScore int64 `json:"dialog_score"`
|
|
MailboxSeq int64 `json:"mailbox_seq"`
|
|
HasMore bool `json:"has_more"`
|
|
}
|
|
|
|
// Service is SYNC_INBOX. HTTP and TCP share it.
|
|
type Service struct {
|
|
inbox inboxlogic.Store
|
|
box Mailbox
|
|
}
|
|
|
|
func NewService(inbox inboxlogic.Store, box Mailbox) *Service {
|
|
return &Service{inbox: inbox, box: box}
|
|
}
|
|
|
|
func (s *Service) ready() bool {
|
|
return s != nil && s.inbox != nil
|
|
}
|
|
|
|
func (s *Service) SyncInbox(ctx context.Context, req entity.InboxRequest) (*InboxResponse, error) {
|
|
if !s.ready() || req.UserID <= 0 {
|
|
return nil, errs.ErrUnauthorized
|
|
}
|
|
if req.MailboxSeq < 0 || req.DialogScore < 0 {
|
|
return nil, errs.ErrInvalidArgument
|
|
}
|
|
limit := req.Limit
|
|
if limit <= 0 || limit > defaultInboxLimit {
|
|
limit = defaultInboxLimit
|
|
}
|
|
hits, err := s.inbox.ListUpdatedSince(ctx, req.UserID, req.DialogScore, limit+1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
hasMore := len(hits) > limit
|
|
if hasMore {
|
|
hits = hits[:limit]
|
|
}
|
|
dialogs, err := s.inbox.LoadDialogs(ctx, req.UserID, hits)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if dialogs == nil {
|
|
dialogs = []inboxentity.Dialog{}
|
|
}
|
|
nextScore := maxScore(req.DialogScore, dialogs)
|
|
mailbox, nextSeq, err := s.loadMailbox(ctx, req, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if s.box != nil {
|
|
_ = s.box.SaveCursor(ctx, req.UserID, req.DeviceID, nextSeq, nextScore)
|
|
}
|
|
return &InboxResponse{
|
|
Dialogs: dialogs, Mailbox: mailbox,
|
|
DialogScore: nextScore, MailboxSeq: nextSeq, HasMore: hasMore,
|
|
}, nil
|
|
}
|
|
|
|
func (s *Service) loadMailbox(ctx context.Context, req entity.InboxRequest, limit int) ([]*entity.MailboxItem, int64, error) {
|
|
nextSeq := req.MailboxSeq
|
|
if s.box == nil {
|
|
return []*entity.MailboxItem{}, nextSeq, nil
|
|
}
|
|
mailbox, err := s.box.ListSince(ctx, req.UserID, req.DeviceID, req.MailboxSeq, limit)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
if mailbox == nil {
|
|
mailbox = []*entity.MailboxItem{}
|
|
}
|
|
for _, item := range mailbox {
|
|
if item != nil && item.Seq > nextSeq {
|
|
nextSeq = item.Seq
|
|
}
|
|
}
|
|
return mailbox, nextSeq, nil
|
|
}
|
|
|
|
func maxScore(cur int64, dialogs []inboxentity.Dialog) int64 {
|
|
for i := range dialogs {
|
|
if dialogs[i].Score > cur {
|
|
cur = dialogs[i].Score
|
|
}
|
|
}
|
|
return cur
|
|
}
|