fix: align private and channel update semantics
(cherry picked from commit c65f76f56278f74082c4fa792ed49104d5d33c38)
This commit is contained in:
A
parent
dce7b92772
commit
d84fa6e126
36 changed files with 1765 additions and 382 deletions
|
|
@ -4937,6 +4937,7 @@ func (s *ChannelStore) ListChannelDifference(_ context.Context, req domain.Chann
|
|||
}
|
||||
messages = append(messages, cloneChannelMessage(msg))
|
||||
}
|
||||
s.populateChannelMessageUnreadFlagsLocked(req.UserID, messages)
|
||||
return domain.ChannelDifference{
|
||||
Channel: channel,
|
||||
Self: member,
|
||||
|
|
@ -4991,6 +4992,15 @@ func (s *ChannelStore) ListChannelDifference(_ context.Context, req domain.Chann
|
|||
diff.OtherUpdates = append(diff.OtherUpdates, cloneChannelEvent(event))
|
||||
}
|
||||
}
|
||||
s.populateChannelMessageUnreadFlagsLocked(req.UserID, diff.NewMessages)
|
||||
for i := range diff.OtherUpdates {
|
||||
if diff.OtherUpdates[i].Message.ID == 0 {
|
||||
continue
|
||||
}
|
||||
messages := []domain.ChannelMessage{diff.OtherUpdates[i].Message}
|
||||
s.populateChannelMessageUnreadFlagsLocked(req.UserID, messages)
|
||||
diff.OtherUpdates[i].Message = messages[0]
|
||||
}
|
||||
return diff, nil
|
||||
}
|
||||
|
||||
|
|
@ -5025,6 +5035,46 @@ func (s *ChannelStore) ListActiveChannelIDsForUser(_ context.Context, userID, af
|
|||
return out, nil
|
||||
}
|
||||
|
||||
func (s *ChannelStore) ListDirtyActiveChannelsForUser(_ context.Context, userID int64, sinceDate int, afterChannelID int64, limit int) ([]domain.DirtyChannel, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
if userID == 0 || sinceDate <= 0 || afterChannelID < 0 {
|
||||
return nil, domain.ErrChannelInvalid
|
||||
}
|
||||
if limit <= 0 || limit > domain.MaxChannelDifferenceLimit {
|
||||
limit = domain.MaxChannelDifferenceLimit
|
||||
}
|
||||
out := make([]domain.DirtyChannel, 0, limit)
|
||||
for channelID, members := range s.members {
|
||||
if channelID <= afterChannelID {
|
||||
continue
|
||||
}
|
||||
channel, ok := s.channels[channelID]
|
||||
if !ok || channel.Deleted {
|
||||
continue
|
||||
}
|
||||
member, ok := members[userID]
|
||||
if !ok || member.Status != domain.ChannelMemberActive {
|
||||
continue
|
||||
}
|
||||
dirty := false
|
||||
for _, event := range s.events[channelID] {
|
||||
if event.Date > sinceDate {
|
||||
dirty = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if dirty {
|
||||
out = append(out, domain.DirtyChannel{ChannelID: channelID, Pts: channel.Pts})
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].ChannelID < out[j].ChannelID })
|
||||
if len(out) > limit {
|
||||
out = out[:limit]
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *ChannelStore) ListActiveChannelMemberIDs(_ context.Context, viewerUserID, channelID int64, limit int) ([]int64, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
|
@ -6495,6 +6545,7 @@ func (s *ChannelStore) populateChannelMessageReactionsLocked(viewerUserID int64,
|
|||
if len(messages) == 0 || channel.ID == 0 {
|
||||
return
|
||||
}
|
||||
s.populateChannelMessageUnreadFlagsLocked(viewerUserID, messages)
|
||||
for i := range messages {
|
||||
if messages[i].ChannelID != channel.ID || messages[i].ID <= 0 {
|
||||
continue
|
||||
|
|
@ -6511,6 +6562,7 @@ func (s *ChannelStore) populateChannelMessagesReactionsLocked(viewerUserID int64
|
|||
if len(messages) == 0 {
|
||||
return
|
||||
}
|
||||
s.populateChannelMessageUnreadFlagsLocked(viewerUserID, messages)
|
||||
channelsByID := make(map[int64]domain.Channel, len(channels))
|
||||
for _, ch := range channels {
|
||||
if ch.ID != 0 {
|
||||
|
|
@ -6533,6 +6585,22 @@ func (s *ChannelStore) populateChannelMessagesReactionsLocked(viewerUserID int64
|
|||
}
|
||||
}
|
||||
|
||||
func (s *ChannelStore) populateChannelMessageUnreadFlagsLocked(viewerUserID int64, messages []domain.ChannelMessage) {
|
||||
if viewerUserID == 0 || len(messages) == 0 {
|
||||
return
|
||||
}
|
||||
for i := range messages {
|
||||
if messages[i].ChannelID == 0 || messages[i].ID <= 0 {
|
||||
continue
|
||||
}
|
||||
if _, ok := s.mentions[viewerUserID][messages[i].ChannelID][messages[i].ID]; !ok {
|
||||
continue
|
||||
}
|
||||
messages[i].Mentioned = true
|
||||
messages[i].MediaUnread = !messages[i].Media.IsZero()
|
||||
}
|
||||
}
|
||||
|
||||
type memoryReactionCursor struct {
|
||||
date int
|
||||
userID int64
|
||||
|
|
|
|||
|
|
@ -242,13 +242,17 @@ func (s *TempAuthKeyBindingStore) GetByTemp(_ context.Context, tempAuthKeyID [8]
|
|||
|
||||
// ContactStore 是 store.ContactStore 的内存实现。
|
||||
type ContactStore struct {
|
||||
mu sync.RWMutex
|
||||
m map[int64]domain.ContactList
|
||||
mu sync.RWMutex
|
||||
m map[int64]domain.ContactList
|
||||
blocks map[int64]map[int64]domain.BlockedContact
|
||||
}
|
||||
|
||||
// NewContactStore 创建内存 ContactStore。
|
||||
func NewContactStore() *ContactStore {
|
||||
return &ContactStore{m: make(map[int64]domain.ContactList)}
|
||||
return &ContactStore{
|
||||
m: make(map[int64]domain.ContactList),
|
||||
blocks: make(map[int64]map[int64]domain.BlockedContact),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ContactStore) ListByUser(_ context.Context, userID int64) (domain.ContactList, error) {
|
||||
|
|
@ -400,6 +404,68 @@ func (s *ContactStore) Delete(_ context.Context, userID int64, contactUserIDs []
|
|||
return deleted, nil
|
||||
}
|
||||
|
||||
func (s *ContactStore) Block(_ context.Context, userID, blockedUserID int64, date int) (bool, error) {
|
||||
if userID == 0 || blockedUserID == 0 || userID == blockedUserID {
|
||||
return false, nil
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.blocks[userID] == nil {
|
||||
s.blocks[userID] = make(map[int64]domain.BlockedContact)
|
||||
}
|
||||
_, existed := s.blocks[userID][blockedUserID]
|
||||
s.blocks[userID][blockedUserID] = domain.BlockedContact{
|
||||
User: domain.User{ID: blockedUserID},
|
||||
Date: date,
|
||||
}
|
||||
return !existed, nil
|
||||
}
|
||||
|
||||
func (s *ContactStore) Unblock(_ context.Context, userID, blockedUserID int64) (bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.blocks[userID] == nil {
|
||||
return false, nil
|
||||
}
|
||||
_, existed := s.blocks[userID][blockedUserID]
|
||||
delete(s.blocks[userID], blockedUserID)
|
||||
return existed, nil
|
||||
}
|
||||
|
||||
func (s *ContactStore) IsBlocked(_ context.Context, userID, blockedUserID int64) (bool, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
_, blocked := s.blocks[userID][blockedUserID]
|
||||
return blocked, nil
|
||||
}
|
||||
|
||||
func (s *ContactStore) ListBlocked(_ context.Context, userID int64, offset, limit int) (domain.BlockedContactList, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
items := make([]domain.BlockedContact, 0, len(s.blocks[userID]))
|
||||
for _, item := range s.blocks[userID] {
|
||||
items = append(items, item)
|
||||
}
|
||||
sort.Slice(items, func(i, j int) bool {
|
||||
if items[i].Date == items[j].Date {
|
||||
return items[i].User.ID > items[j].User.ID
|
||||
}
|
||||
return items[i].Date > items[j].Date
|
||||
})
|
||||
total := len(items)
|
||||
if offset < 0 {
|
||||
offset = 0
|
||||
}
|
||||
if offset >= len(items) {
|
||||
return domain.BlockedContactList{Count: total}, nil
|
||||
}
|
||||
if limit <= 0 || limit > len(items)-offset {
|
||||
limit = len(items) - offset
|
||||
}
|
||||
out := append([]domain.BlockedContact(nil), items[offset:offset+limit]...)
|
||||
return domain.BlockedContactList{Blocked: out, Count: total}, nil
|
||||
}
|
||||
|
||||
// SaveList 保存一份用户通讯录，供测试和本地替身使用。
|
||||
func (s *ContactStore) SaveList(_ context.Context, userID int64, list domain.ContactList) error {
|
||||
list.Contacts = cloneContacts(list.Contacts)
|
||||
|
|
@ -956,21 +1022,30 @@ func (s *MessageStore) SendPrivateText(_ context.Context, req domain.SendPrivate
|
|||
Forward: cloneMessageForward(req.Forward),
|
||||
Pts: s.nextPtsLocked(req.SenderUserID),
|
||||
}
|
||||
recipient := sender
|
||||
if req.SenderUserID != req.RecipientUserID {
|
||||
recipient := domain.Message{}
|
||||
if req.SenderUserID == req.RecipientUserID {
|
||||
recipient = sender
|
||||
}
|
||||
if req.SenderUserID != req.RecipientUserID && !req.RecipientBlocked {
|
||||
recipient = sender
|
||||
recipient.ID = s.nextBoxIDLocked(req.RecipientUserID)
|
||||
recipient.OwnerUserID = req.RecipientUserID
|
||||
recipient.Peer = domain.Peer{Type: domain.PeerTypeUser, ID: req.SenderUserID}
|
||||
recipient.Out = false
|
||||
recipient.ReplyTo = cloneMessageReply(recipientReply)
|
||||
recipient.Pts = s.nextPtsLocked(req.RecipientUserID)
|
||||
recipient.MediaUnread = !req.Media.IsZero()
|
||||
}
|
||||
s.m[req.SenderUserID] = append(s.m[req.SenderUserID], sender)
|
||||
if req.SenderUserID != req.RecipientUserID {
|
||||
if req.SenderUserID != req.RecipientUserID && !req.RecipientBlocked {
|
||||
s.m[req.RecipientUserID] = append(s.m[req.RecipientUserID], recipient)
|
||||
}
|
||||
if s.dialogs != nil {
|
||||
s.upsertMemoryDialogsLocked(sender, recipient)
|
||||
if recipient.ID != 0 {
|
||||
s.upsertMemoryDialogsLocked(sender, recipient)
|
||||
} else {
|
||||
s.upsertMemoryDialogsLocked(sender, sender)
|
||||
}
|
||||
}
|
||||
return domain.SendPrivateTextResult{
|
||||
SenderMessage: cloneMessage(sender),
|
||||
|
|
@ -1078,18 +1153,19 @@ func (s *MessageStore) ForwardPrivateMessages(ctx context.Context, req domain.Fo
|
|||
}
|
||||
}
|
||||
sent, err := s.SendPrivateText(ctx, domain.SendPrivateTextRequest{
|
||||
SenderUserID: req.OwnerUserID,
|
||||
RecipientUserID: req.ToUserID,
|
||||
RandomID: req.RandomIDs[i],
|
||||
Message: source.Body,
|
||||
Entities: append([]domain.MessageEntity(nil), source.Entities...),
|
||||
Silent: req.Silent,
|
||||
NoForwards: req.NoForwards,
|
||||
ReplyTo: req.ReplyTo,
|
||||
Forward: forward,
|
||||
Date: req.Date,
|
||||
OriginAuthKeyID: req.OriginAuthKeyID,
|
||||
OriginSessionID: req.OriginSessionID,
|
||||
SenderUserID: req.OwnerUserID,
|
||||
RecipientUserID: req.ToUserID,
|
||||
RandomID: req.RandomIDs[i],
|
||||
Message: source.Body,
|
||||
Entities: append([]domain.MessageEntity(nil), source.Entities...),
|
||||
Silent: req.Silent,
|
||||
NoForwards: req.NoForwards,
|
||||
ReplyTo: req.ReplyTo,
|
||||
Forward: forward,
|
||||
Date: req.Date,
|
||||
OriginAuthKeyID: req.OriginAuthKeyID,
|
||||
OriginSessionID: req.OriginSessionID,
|
||||
RecipientBlocked: req.RecipientBlocked,
|
||||
})
|
||||
if err != nil {
|
||||
return res, err
|
||||
|
|
@ -1277,14 +1353,52 @@ func (s *MessageStore) ReadMessageContents(_ context.Context, req domain.ReadMes
|
|||
if len(wanted) == 0 {
|
||||
return res, nil
|
||||
}
|
||||
s.mu.RLock()
|
||||
for _, msg := range s.m[req.OwnerUserID] {
|
||||
if _, ok := wanted[msg.ID]; ok {
|
||||
res.MessageIDs = append(res.MessageIDs, msg.ID)
|
||||
}
|
||||
if req.Date == 0 {
|
||||
req.Date = int(time.Now().Unix())
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
affectedPeers := make(map[domain.Peer]struct{})
|
||||
for i := range s.m[req.OwnerUserID] {
|
||||
msg := &s.m[req.OwnerUserID][i]
|
||||
if _, ok := wanted[msg.ID]; !ok {
|
||||
continue
|
||||
}
|
||||
if !msg.MediaUnread && !msg.ReactionUnread {
|
||||
continue
|
||||
}
|
||||
if msg.ReactionUnread && msg.Peer.ID != 0 {
|
||||
affectedPeers[msg.Peer] = struct{}{}
|
||||
}
|
||||
msg.MediaUnread = false
|
||||
msg.ReactionUnread = false
|
||||
res.MessageIDs = append(res.MessageIDs, msg.ID)
|
||||
}
|
||||
s.mu.RUnlock()
|
||||
sort.Ints(res.MessageIDs)
|
||||
if len(res.MessageIDs) == 0 {
|
||||
return res, nil
|
||||
}
|
||||
if s.dialogs != nil && len(affectedPeers) > 0 {
|
||||
s.dialogs.mu.Lock()
|
||||
list := s.dialogs.m[req.OwnerUserID]
|
||||
for i := range list.Dialogs {
|
||||
if _, ok := affectedPeers[list.Dialogs[i].Peer]; !ok {
|
||||
continue
|
||||
}
|
||||
list.Dialogs[i].UnreadReactions = s.countPrivateUnreadReactionsLocked(req.OwnerUserID, list.Dialogs[i].Peer)
|
||||
}
|
||||
s.dialogs.m[req.OwnerUserID] = list
|
||||
s.dialogs.mu.Unlock()
|
||||
}
|
||||
pts := s.nextPtsNLocked(req.OwnerUserID, len(res.MessageIDs))
|
||||
res.Event = domain.UpdateEvent{
|
||||
UserID: req.OwnerUserID,
|
||||
Type: domain.UpdateEventReadMessageContents,
|
||||
Pts: pts,
|
||||
PtsCount: len(res.MessageIDs),
|
||||
Date: req.Date,
|
||||
MessageIDs: append([]int(nil), res.MessageIDs...),
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
|
|
@ -1355,6 +1469,27 @@ func (s *MessageStore) SetMessageReactions(_ context.Context, req domain.SetPriv
|
|||
} else {
|
||||
s.privateReactions[target.UID][req.UserID] = rows
|
||||
}
|
||||
if target.From.ID != 0 && target.From.ID != req.UserID {
|
||||
for i := range s.m[target.From.ID] {
|
||||
if s.m[target.From.ID][i].UID != target.UID {
|
||||
continue
|
||||
}
|
||||
s.m[target.From.ID][i].ReactionUnread = len(rows) > 0
|
||||
if s.dialogs != nil {
|
||||
s.dialogs.mu.Lock()
|
||||
list := s.dialogs.m[target.From.ID]
|
||||
peer := domain.Peer{Type: domain.PeerTypeUser, ID: req.UserID}
|
||||
for j := range list.Dialogs {
|
||||
if list.Dialogs[j].Peer == peer {
|
||||
list.Dialogs[j].UnreadReactions = s.countPrivateUnreadReactionsLocked(target.From.ID, peer)
|
||||
}
|
||||
}
|
||||
s.dialogs.m[target.From.ID] = list
|
||||
s.dialogs.mu.Unlock()
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
return s.privateReactionResultLocked(target.UID), nil
|
||||
}
|
||||
|
||||
|
|
@ -1585,6 +1720,16 @@ func (s *MessageStore) nextPtsNLocked(userID int64, count int) int {
|
|||
return next
|
||||
}
|
||||
|
||||
func (s *MessageStore) countPrivateUnreadReactionsLocked(ownerUserID int64, peer domain.Peer) int {
|
||||
count := 0
|
||||
for _, msg := range s.m[ownerUserID] {
|
||||
if msg.Peer == peer && msg.ReactionUnread {
|
||||
count++
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
func (s *MessageStore) upsertMemoryDialogsLocked(sender, recipient domain.Message) {
|
||||
s.dialogs.mu.Lock()
|
||||
defer s.dialogs.mu.Unlock()
|
||||
|
|
|
|||
|
|
@ -120,6 +120,31 @@ func TestMessageStorePrivateMessageReactionsAreSharedAcrossOwnerBoxes(t *testing
|
|||
if got := aliceReactions.Messages[0].Reactions.Recent; len(got) != 1 || got[0].UserID != bobID || !got[0].Big || got[0].My {
|
||||
t.Fatalf("alice recent reactions = %+v, want bob non-my big reaction", got)
|
||||
}
|
||||
aliceBox, err := messages.GetByIDs(ctx, aliceID, []int{sent.SenderMessage.ID})
|
||||
if err != nil {
|
||||
t.Fatalf("alice GetByIDs after reaction: %v", err)
|
||||
}
|
||||
if len(aliceBox.Messages) != 1 || !aliceBox.Messages[0].ReactionUnread {
|
||||
t.Fatalf("alice box after reaction = %+v, want reaction_unread", aliceBox.Messages)
|
||||
}
|
||||
read, err := messages.ReadMessageContents(ctx, domain.ReadMessageContentsRequest{
|
||||
OwnerUserID: aliceID,
|
||||
IDs: []int{sent.SenderMessage.ID},
|
||||
Date: 1700000210,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ReadMessageContents reaction: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(read.MessageIDs, []int{sent.SenderMessage.ID}) || read.Event.Type != domain.UpdateEventReadMessageContents || read.Event.Pts == 0 {
|
||||
t.Fatalf("read reaction contents = %+v, want one read_message_contents event", read)
|
||||
}
|
||||
aliceBox, err = messages.GetByIDs(ctx, aliceID, []int{sent.SenderMessage.ID})
|
||||
if err != nil {
|
||||
t.Fatalf("alice GetByIDs after read reaction: %v", err)
|
||||
}
|
||||
if len(aliceBox.Messages) != 1 || aliceBox.Messages[0].ReactionUnread {
|
||||
t.Fatalf("alice box after read reaction = %+v, want reaction_unread cleared", aliceBox.Messages)
|
||||
}
|
||||
|
||||
bobReactions, err := messages.GetMessageReactions(ctx, domain.PrivateMessageReactionsRequest{
|
||||
OwnerUserID: bobID,
|
||||
|
|
@ -344,7 +369,7 @@ func TestMessageStoreReadHistoryEmitsInboxAndOutboxReceipts(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestMessageStoreReadMessageContentsReturnsExistingOwnerIDs(t *testing.T) {
|
||||
func TestMessageStoreReadMessageContentsClearsUnreadContentOnce(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
messages := NewMessageStore()
|
||||
sent, err := messages.SendPrivateText(ctx, domain.SendPrivateTextRequest{
|
||||
|
|
@ -352,20 +377,39 @@ func TestMessageStoreReadMessageContentsReturnsExistingOwnerIDs(t *testing.T) {
|
|||
RecipientUserID: 1002,
|
||||
RandomID: 88,
|
||||
Message: "voice placeholder",
|
||||
Media: &domain.MessageMedia{Kind: domain.MessageMediaKindDocument, Voice: true},
|
||||
Date: 1700000300,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("SendPrivateText: %v", err)
|
||||
}
|
||||
if !sent.RecipientMessage.MediaUnread {
|
||||
t.Fatalf("recipient MediaUnread = false, want true for incoming media")
|
||||
}
|
||||
got, err := messages.ReadMessageContents(ctx, domain.ReadMessageContentsRequest{
|
||||
OwnerUserID: 1002,
|
||||
IDs: []int{sent.RecipientMessage.ID, domain.MaxMessageBoxID},
|
||||
Date: 1700000400,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ReadMessageContents: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(got.MessageIDs, []int{sent.RecipientMessage.ID}) {
|
||||
t.Fatalf("MessageIDs = %v, want existing recipient id", got.MessageIDs)
|
||||
t.Fatalf("MessageIDs = %v, want unread recipient id", got.MessageIDs)
|
||||
}
|
||||
if got.Event.Type != domain.UpdateEventReadMessageContents || got.Event.Pts == 0 || got.Event.PtsCount != 1 {
|
||||
t.Fatalf("Event = %+v, want read_message_contents pts update", got.Event)
|
||||
}
|
||||
repeated, err := messages.ReadMessageContents(ctx, domain.ReadMessageContentsRequest{
|
||||
OwnerUserID: 1002,
|
||||
IDs: []int{sent.RecipientMessage.ID},
|
||||
Date: 1700000500,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ReadMessageContents repeat: %v", err)
|
||||
}
|
||||
if len(repeated.MessageIDs) != 0 || repeated.Event.Pts != 0 {
|
||||
t.Fatalf("repeat = %+v, want no affected messages and no pts", repeated)
|
||||
}
|
||||
if _, err := messages.ReadMessageContents(ctx, domain.ReadMessageContentsRequest{
|
||||
OwnerUserID: 1002,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue