mirror of
				https://github.com/openimsdk/open-im-server.git
				synced 2025-11-01 00:42:13 +08:00 
			
		
		
		
	* fix: GroupApplicationAcceptedNotification * fix: GroupApplicationAcceptedNotification * fix: NotificationUserInfoUpdate * cicd: robot automated Change * fix: component * fix: getConversationInfo * feat: cron task * feat: cron task * feat: cron task * feat: cron task * feat: cron task * fix: minio config url recognition error * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * friend incr sync * friend incr sync * friend incr sync * friend incr sync * friend incr sync * mage * optimization version log * optimization version log * sync * sync * sync * group sync * sync option * sync option * refactor: replace `friend` package with `realtion`. * refactor: update lastest commit to relation. * sync option * sync option * sync option * sync * sync * go.mod * seq * update: go mod * refactor: change incremental to full * feat: get full friend user ids * feat: api and config * seq * group version * merge * seq * seq * seq * fix: sort by id avoid unstable sort friends. * group * group * group * fix: sort by id avoid unstable sort friends. * fix: sort by id avoid unstable sort friends. * fix: sort by id avoid unstable sort friends. * user version * seq * seq * seq user * user online * implement minio expire delete. * user online * config * fix * fix * implement minio expire delete logic. * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * feat: implement scheduled delete outdated object in minio. * update gomake version * update gomake version * implement FindExpires pagination. * remove unnesseary incr. * fix uncorrect args call. * online push * online push * online push * resolving conflicts * resolving conflicts * test * api prommetrics * api prommetrics * api prommetrics * api prommetrics * api prommetrics * rpc prommetrics * rpc prommetrics * online status * online status * online status * online status * sub * conversation version incremental * merge seq * merge online * merge online * merge online * merge seq * GetOwnerConversation * fix: change incremental syncer router name. * rockscache batch get * rockscache seq batch get * fix: GetMsgDocModelByIndex bug * update go.mod * update go.mod * merge * feat: prometheus * feat: prometheus * group member sort * sub * sub * fix: seq conversion bug * fix: redis pipe exec * sort version * sort version * sort version * remove old version online subscription * remove old version online subscription * version log index * version log index * batch push * batch push * seq void filling * fix: batchGetMaxSeq * fix: batchGetMaxSeq * cache db error log * 111 * fix bug * fix: ImportFriends * add online cache * add some logs --------- Co-authored-by: withchao <withchao@users.noreply.github.com> Co-authored-by: Monet Lee <monet_lee@163.com> Co-authored-by: OpenIM-Gordon <46924906+FGadvancer@users.noreply.github.com> Co-authored-by: icey-yu <1186114839@qq.com>
		
			
				
	
	
		
			94 lines
		
	
	
		
			2.7 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
			
		
		
	
	
			94 lines
		
	
	
		
			2.7 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
| package redis
 | |
| 
 | |
| import (
 | |
| 	"context"
 | |
| 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 | |
| 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 | |
| 	"github.com/openimsdk/tools/errs"
 | |
| 	"github.com/openimsdk/tools/log"
 | |
| 	"github.com/redis/go-redis/v9"
 | |
| 	"strconv"
 | |
| 	"time"
 | |
| )
 | |
| 
 | |
| func NewUserOnline(rdb redis.UniversalClient) cache.OnlineCache {
 | |
| 	return &userOnline{
 | |
| 		rdb:         rdb,
 | |
| 		expire:      cachekey.OnlineExpire,
 | |
| 		channelName: cachekey.OnlineChannel,
 | |
| 	}
 | |
| }
 | |
| 
 | |
| type userOnline struct {
 | |
| 	rdb         redis.UniversalClient
 | |
| 	expire      time.Duration
 | |
| 	channelName string
 | |
| }
 | |
| 
 | |
| func (s *userOnline) getUserOnlineKey(userID string) string {
 | |
| 	return cachekey.GetOnlineKey(userID)
 | |
| }
 | |
| 
 | |
| func (s *userOnline) GetOnline(ctx context.Context, userID string) ([]int32, error) {
 | |
| 	members, err := s.rdb.ZRangeByScore(ctx, s.getUserOnlineKey(userID), &redis.ZRangeBy{
 | |
| 		Min: strconv.FormatInt(time.Now().Unix(), 10),
 | |
| 		Max: "+inf",
 | |
| 	}).Result()
 | |
| 	if err != nil {
 | |
| 		return nil, errs.Wrap(err)
 | |
| 	}
 | |
| 	platformIDs := make([]int32, 0, len(members))
 | |
| 	for _, member := range members {
 | |
| 		val, err := strconv.Atoi(member)
 | |
| 		if err != nil {
 | |
| 			return nil, errs.Wrap(err)
 | |
| 		}
 | |
| 		platformIDs = append(platformIDs, int32(val))
 | |
| 	}
 | |
| 	return platformIDs, nil
 | |
| }
 | |
| 
 | |
| func (s *userOnline) SetUserOnline(ctx context.Context, userID string, online, offline []int32) error {
 | |
| 	script := `
 | |
| 	local key = KEYS[1]
 | |
| 	local score = ARGV[3]
 | |
| 	local num1 = redis.call("ZCARD", key)
 | |
| 	redis.call("ZREMRANGEBYSCORE", key, "-inf", ARGV[2])
 | |
| 	for i = 5, tonumber(ARGV[4])+4 do
 | |
| 		redis.call("ZREM", key, ARGV[i])
 | |
| 	end
 | |
| 	local num2 = redis.call("ZCARD", key)
 | |
| 	for i = 5+tonumber(ARGV[4]), #ARGV do
 | |
| 		redis.call("ZADD", key, score, ARGV[i])
 | |
| 	end
 | |
| 	redis.call("EXPIRE", key, ARGV[1])
 | |
| 	local num3 = redis.call("ZCARD", key)
 | |
| 	local change = (num1 ~= num2) or (num2 ~= num3)
 | |
| 	if change then
 | |
| 		local members = redis.call("ZRANGE", key, 0, -1)
 | |
| 		table.insert(members, KEYS[2])
 | |
| 		redis.call("PUBLISH", KEYS[3], table.concat(members, ":"))
 | |
| 		return 1
 | |
| 	else
 | |
| 		return 0
 | |
| 	end
 | |
| `
 | |
| 	now := time.Now()
 | |
| 	argv := make([]any, 0, 2+len(online)+len(offline))
 | |
| 	argv = append(argv, int32(s.expire/time.Second), now.Unix(), now.Add(s.expire).Unix(), int32(len(offline)))
 | |
| 	for _, platformID := range offline {
 | |
| 		argv = append(argv, platformID)
 | |
| 	}
 | |
| 	for _, platformID := range online {
 | |
| 		argv = append(argv, platformID)
 | |
| 	}
 | |
| 	keys := []string{s.getUserOnlineKey(userID), userID, s.channelName}
 | |
| 	status, err := s.rdb.Eval(ctx, script, keys, argv).Result()
 | |
| 	if err != nil {
 | |
| 		log.ZError(ctx, "redis SetUserOnline", err, "userID", userID, "online", online, "offline", offline)
 | |
| 		return err
 | |
| 	}
 | |
| 	log.ZDebug(ctx, "redis SetUserOnline", "userID", userID, "online", online, "offline", offline, "status", status)
 | |
| 	return nil
 | |
| }
 |