100 lines
2.5 KiB
Go
100 lines
2.5 KiB
Go
package services
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"git.solsynth.dev/hydrogen/dealer/pkg/hyper"
|
|
"git.solsynth.dev/hydrogen/dealer/pkg/proto"
|
|
"git.solsynth.dev/hydrogen/messaging/pkg/internal/database"
|
|
"git.solsynth.dev/hydrogen/messaging/pkg/internal/gap"
|
|
"git.solsynth.dev/hydrogen/messaging/pkg/internal/models"
|
|
)
|
|
|
|
type statusQueryCacheEntry struct {
|
|
Target []uint64
|
|
Data any
|
|
}
|
|
|
|
var statusQueryCacheLock sync.Mutex
|
|
|
|
// Map for caching typing status queries [channel id][user id]
|
|
var statusQueryCache = make(map[uint]map[uint]statusQueryCacheEntry)
|
|
|
|
func SetTypingStatus(channelId uint, userId uint) error {
|
|
var broadcastTarget []uint64
|
|
var data any
|
|
|
|
hitCache := false
|
|
if channelLevel, ok := statusQueryCache[channelId]; ok {
|
|
if entry, ok := channelLevel[userId]; ok {
|
|
broadcastTarget = entry.Target
|
|
data = entry.Data
|
|
hitCache = true
|
|
}
|
|
}
|
|
|
|
if !hitCache {
|
|
var account models.Account
|
|
if err := database.C.Where("external_id = ?", userId).First(&account).Error; err != nil {
|
|
return fmt.Errorf("account not found: %v", err)
|
|
}
|
|
|
|
var member models.ChannelMember
|
|
if err := database.C.
|
|
Where("account_id = ? AND channel_id = ?", account.ID, channelId).
|
|
First(&member).Error; err != nil {
|
|
return fmt.Errorf("channel member not found: %v", err)
|
|
} else {
|
|
member.Account = account
|
|
}
|
|
|
|
var channel models.Channel
|
|
if err := database.C.
|
|
Preload("Members").
|
|
Preload("Members.Account").
|
|
Where("id = ?", channelId).
|
|
First(&channel).Error; err != nil {
|
|
return fmt.Errorf("channel not found: %v", err)
|
|
}
|
|
|
|
for _, item := range channel.Members {
|
|
broadcastTarget = append(broadcastTarget, uint64(item.Account.ID))
|
|
}
|
|
|
|
data = map[string]any{
|
|
"user_id": userId,
|
|
"member_id": member.ID,
|
|
"channel_id": channelId,
|
|
"member": member,
|
|
"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(),
|
|
})
|
|
|
|
if len(statusQueryCache) > 512 {
|
|
statusQueryCacheLock.Lock()
|
|
clear(statusQueryCache)
|
|
statusQueryCacheLock.Unlock()
|
|
}
|
|
|
|
return err
|
|
}
|