Hypernet/Messaging
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

🐛 Bug fixes on typing status cache

Browse Source
This commit is contained in:
LittleSheep 2024-08-23 22:26:54 +08:00
parent e5c46a89df
commit d7642c82bd
1 changed files with 38 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
pkg/internal/services/status.go
View File

@ -13,9 +13,8 @@ import (
) )
type statusQueryCacheEntry struct { type statusQueryCacheEntry struct {
Account models.Account Target []uint64
Member models.ChannelMember Data any
Channel models.Channel
} }
var statusQueryCacheLock sync.Mutex var statusQueryCacheLock sync.Mutex
@ -24,21 +23,25 @@ var statusQueryCacheLock sync.Mutex
var statusQueryCache = make(map[uint]map[uint]statusQueryCacheEntry) var statusQueryCache = make(map[uint]map[uint]statusQueryCacheEntry)
func SetTypingStatus(channelId uint, userId uint) error { func SetTypingStatus(channelId uint, userId uint) error {
var account models.Account var broadcastTarget []uint64
var member models.ChannelMember var data any
var channel models.Channel
hitCache := false
if channelLevel, ok := statusQueryCache[channelId]; ok { if channelLevel, ok := statusQueryCache[channelId]; ok {
if entry, ok := channelLevel[userId]; ok { if entry, ok := channelLevel[userId]; ok {
account = entry.Account broadcastTarget = entry.Target
member = entry.Member data = entry.Data
channel = entry.Channel hitCache = true
} }
} else { }
if !hitCache {
var account models.Account
if err := database.C.Where("external_id = ?", userId).First(&account).Error; err != nil { if err := database.C.Where("external_id = ?", userId).First(&account).Error; err != nil {
return fmt.Errorf("account not found: %v", err) return fmt.Errorf("account not found: %v", err)
} }
var member models.ChannelMember
if err := database.C. if err := database.C.
Where("account_id = ? AND channel_id = ?", account.ID, channelId). Where("account_id = ? AND channel_id = ?", account.ID, channelId).
First(&member).Error; err != nil { First(&member).Error; err != nil {
@ -47,6 +50,7 @@ func SetTypingStatus(channelId uint, userId uint) error {
member.Account = account member.Account = account
} }
var channel models.Channel
if err := database.C. if err := database.C.
Preload("Members"). Preload("Members").
Where("id = ?", channelId). Where("id = ?", channelId).
@ -54,16 +58,6 @@ func SetTypingStatus(channelId uint, userId uint) error {
return fmt.Errorf("channel not found: %v", err) return fmt.Errorf("channel not found: %v", err)
} }
// Cache queries
statusQueryCacheLock.Lock()
if _, ok := statusQueryCache[channelId]; !ok {
statusQueryCache[channelId] = make(map[uint]statusQueryCacheEntry)
}
statusQueryCache[channelId][userId] = statusQueryCacheEntry{account, member, channel}
statusQueryCacheLock.Unlock()
}
var broadcastTarget []uint64
for _, item := range channel.Members { for _, item := range channel.Members {
if item.AccountID == member.AccountID { if item.AccountID == member.AccountID {
continue continue
@ -71,20 +65,37 @@ func SetTypingStatus(channelId uint, userId uint) error {
broadcastTarget = append(broadcastTarget, uint64(item.AccountID)) broadcastTarget = append(broadcastTarget, uint64(item.AccountID))
} }
sc := proto.NewStreamControllerClient(gap.H.GetDealerGrpcConn()) data = map[string]any{
_, err := sc.PushStreamBatch(context.Background(), &proto.PushStreamBatchRequest{
UserId: broadcastTarget,
Body: hyper.NetworkPackage{
Action: "status.typing",
Payload: map[string]any{
"user_id": userId, "user_id": userId,
"member_id": member.ID, "member_id": member.ID,
"channel_id": channelId, "channel_id": channelId,
"member": member, "member": member,
"channel": channel, "channel": channel,
}, }
// Cache queries
statusQueryCacheLock.Lock()
if _, ok := statusQueryCache[channelId]; !ok {
statusQueryCache[channelId] = make(map[uint]statusQueryCacheEntry)
}
statusQueryCache[channelId][userId] = statusQueryCacheEntry{broadcastTarget, data}
statusQueryCacheLock.Unlock()
}
sc := proto.NewStreamControllerClient(gap.H.GetDealerGrpcConn())
_, err := sc.PushStreamBatch(context.Background(), &proto.PushStreamBatchRequest{
UserId: broadcastTarget,
Body: hyper.NetworkPackage{
Action: "status.typing",
Payload: data,
}.Marshal(), }.Marshal(),
}) })
if len(statusQueryCache) > 512 {
statusQueryCacheLock.Lock()
clear(statusQueryCache)
statusQueryCacheLock.Unlock()
}
return err return err
} }