feat: Add user dispatch with redis zset.

Signed-off-by: IRONICBo <47499836+IRONICBo@users.noreply.github.com>

server/internal/api/callback.go

@@ -62,10 +62,12 @@ func MsgModify(c *gin.Context) {
 
 // UserOnline.
 func UserOnline(c *gin.Context) {
+	// TODO: push user id to queue
 }
 
 // UserOffline.
 func UserOffline(c *gin.Context) {
+	// TODO: remove user id to queue
 }
 
 // OfflinePush.

server/internal/api/user.go

@@ -74,7 +74,16 @@ func AccountLogin(c *gin.Context) {
 	svc := service.NewUserService(c)
 	u, err := svc.LoginWithAccount(&params)
 	if err != nil {
-		log.Debug("AdminRegister error", err)
+		log.Debug("Login error", err)
+		response.FailWithCode(common.ERROR, c)
+
+		return
+	}
+
+	// TODO: Add login status to user dispatch
+	dsvc := service.NewUserDispatchService(c.Request.Context())
+	if err = dsvc.AddUser(u.UUID); err != nil {
+		log.Error("Add to online status error", err)
 		response.FailWithCode(common.ERROR, c)
 
 		return
@@ -83,6 +92,34 @@ func AccountLogin(c *gin.Context) {
 	response.SuccessWithData(u, c)
 }
 
+// AccountExit
+// @Tags user
+// @Summary AccountExit
+// @Description logout with account
+// @Produce application/json
+// @Security  ApiKeyAuth
+// @Success 200 {object}  response.Response{msg=string} "Success"
+// @Router /api/v1/login/account [post].
+func AccountExit(c *gin.Context) {
+	uuid, err := utils.GetUserUUID(c)
+	if err != nil {
+		response.FailWithCode(common.UNAUTHORIZED, c)
+
+		return
+	}
+
+	svc := service.NewUserDispatchService(c)
+	err = svc.DeleteUser(uuid)
+	if err != nil {
+		log.Error("Delete user from online status error", err)
+		response.FailWithCode(common.ERROR, c)
+
+		return
+	}
+
+	response.Success(c)
+}
+
 // GetUserInfo
 // @Tags user
 // @Summary GetUserInfo

server/internal/dal/cache/redis.go

@@ -26,6 +26,7 @@ var _ Cache = new(cache)
 
 // Cache cache interface.
 type Cache interface {
+	// default
 	Set(ctx context.Context, key, value string, ttl time.Duration) error
 	Get(ctx context.Context, key string) (string, error)
 	TTL(ctx context.Context, key string) (time.Duration, error)
@@ -35,6 +36,29 @@ type Cache interface {
 	Exists(ctx context.Context, keys ...string) bool
 	Incr(ctx context.Context, key string) int64
 	Close() error
+
+	// List
+	LPush(ctx context.Context, key string, values ...interface{}) error
+	RPush(ctx context.Context, key string, values ...interface{}) error
+	LPop(ctx context.Context, key string) (string, error)
+	RPop(ctx context.Context, key string) (string, error)
+	LRange(ctx context.Context, key string, start, stop int64) ([]string, error)
+	LRem(ctx context.Context, key string, count int64, value interface{}) (int64, error)
+	LTrim(ctx context.Context, key string, start, stop int64) error
+	LLen(ctx context.Context, key string) (int64, error)
+
+	// ZSet
+	ZAdd(ctx context.Context, key string, members ...*redis.Z) (int64, error)
+	ZRem(ctx context.Context, key string, members ...interface{}) (int64, error)
+	ZRange(ctx context.Context, key string, start, stop int64) ([]string, error)
+
+	// tx
+	Pipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error)
+	Pipline() redis.Pipeliner
+	TxPipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error)
+	TxPipeline() redis.Pipeliner
+
+	Watch(ctx context.Context, fn func(*redis.Tx) error, keys ...string) error
 }
 
 // cache cache dao representative.
@@ -124,3 +148,138 @@ func (c *cache) Incr(ctx context.Context, key string) int64 {
 func (c *cache) Close() error {
 	return c.client.Close()
 }
+
+// LPush left push value to list.
+func (c *cache) LPush(ctx context.Context, key string, values ...interface{}) error {
+	return c.client.LPush(ctx, key, values...).Err()
+}
+
+// RPush right push value to list.
+func (c *cache) RPush(ctx context.Context, key string, values ...interface{}) error {
+	return c.client.RPush(ctx, key, values...).Err()
+}
+
+// LPop left pop value from list.
+func (c *cache) LPop(ctx context.Context, key string) (string, error) {
+	return c.client.LPop(ctx, key).Result()
+}
+
+// RPop right pop value from list.
+func (c *cache) RPop(ctx context.Context, key string) (string, error) {
+	return c.client.RPop(ctx, key).Result()
+}
+
+// LRange get list value from start to stop.
+func (c *cache) LRange(ctx context.Context, key string, start, stop int64) ([]string, error) {
+	return c.client.LRange(ctx, key, start, stop).Result()
+}
+
+// LRem remove count value from list.
+func (c *cache) LRem(ctx context.Context, key string, count int64, value interface{}) (int64, error) {
+	return c.client.LRem(ctx, key, count, value).Result()
+}
+
+// LTrim trim list from start to stop.
+func (c *cache) LTrim(ctx context.Context, key string, start, stop int64) error {
+	return c.client.LTrim(ctx, key, start, stop).Err()
+}
+
+// LLen get list length.
+func (c *cache) LLen(ctx context.Context, key string) (int64, error) {
+	return c.client.LLen(ctx, key).Result()
+}
+
+// Pipelined pipelined.
+func (c *cache) Pipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error) {
+	return c.client.Pipelined(ctx, fn)
+}
+
+// Pipline pipline.
+func (c *cache) Pipline() redis.Pipeliner {
+	return c.client.Pipeline()
+}
+
+// TxPipelined tx pipelined.
+func (c *cache) TxPipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error) {
+	return c.client.TxPipelined(ctx, fn)
+}
+
+// TxPipeline tx pipeline.
+func (c *cache) TxPipeline() redis.Pipeliner {
+	return c.client.TxPipeline()
+}
+
+// Watch watch.
+func (c *cache) Watch(ctx context.Context, fn func(*redis.Tx) error, keys ...string) error {
+	return c.client.Watch(ctx, fn, keys...)
+}
+
+// LPopAndRPush lpop and rpush.
+func (c *cache) LPopAndRPush(ctx context.Context, key string) (string, error) {
+	tx := c.client.TxPipeline()
+	defer tx.Close()
+
+	// check length
+	length, err := c.client.LLen(ctx, key).Result()
+	if err != nil || length <= 1 {
+		return "", errors.Wrapf(err, "list is not exists or length less than 1: %s err", key)
+	}
+
+	// get head value
+	headVal, _ := c.client.LIndex(ctx, key, 0).Result()
+
+	_ = tx.Process(ctx, c.client.LPop(ctx, key))
+	_ = tx.Process(ctx, c.client.RPush(ctx, key, headVal))
+
+	_, err = tx.Exec(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	return headVal, nil
+}
+
+// ZAdd zset add.
+func (c *cache) ZAdd(ctx context.Context, key string, members ...*redis.Z) (int64, error) {
+	return c.client.ZAdd(ctx, key, members...).Result()
+}
+
+// ZRem zset remove.
+func (c *cache) ZRem(ctx context.Context, key string, members ...interface{}) (int64, error) {
+	return c.client.ZRem(ctx, key, members...).Result()
+}
+
+// ZRange zset range.
+func (c *cache) ZRange(ctx context.Context, key string, start, stop int64) ([]string, error) {
+	return c.client.ZRange(ctx, key, start, stop).Result()
+}
+
+// // EnqueueStringLPush enqueue key and push left.
+// func (c *cache) EnqueueStringLPush(ctx context.Context, key, value string) error {
+// 	c.mu.Lock()
+// 	defer c.mu.Unlock()
+
+// 	err := c.client.LPush(ctx, key, value).Err()
+// 	if err != nil {
+// 		return errors.Wrapf(err, "redis enqueue key: %s , value: %s, err: %s", key, value, err.Error())
+// 	}
+
+// 	return nil
+// }
+
+// // DequeueStringRPop dequeue key and pop right.
+// func (c *cache) DequeueStringRPop(ctx context.Context, key string) (string, error) {
+// 	c.mu.Lock()
+// 	defer c.mu.Unlock()
+
+// 	value, err := c.client.RPop(ctx, key).Result()
+// 	if err != nil {
+// 		return "", errors.Wrapf(err, "redis dequeue key: %s , err: %s", key, err.Error())
+// 	}
+
+// 	return value, nil
+// }
+
+// func (c *cache) TxPipeline(ctx context.Context) redis.Pipeliner {
+// 	return c.client.TxPipeline()
+// }

server/internal/dal/dao/user_dispatch_dao.go

@@ -0,0 +1,169 @@
+// Copyright © 2023 OpenIM open source community. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dao
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+	"github.com/pkg/errors"
+
+	"github.com/OpenIMSDK/OpenKF/server/internal/conn/db"
+	"github.com/OpenIMSDK/OpenKF/server/internal/dal/cache"
+	"github.com/OpenIMSDK/OpenKF/server/internal/dal/gen"
+)
+
+// UserDispatchDao user dispatch dao.
+type UserDispatchDao struct {
+	Dao
+
+	mu sync.Mutex
+}
+
+// NewUserDispatchDao return a user dispatch dao.
+func NewUserDispatchDao() *UserDispatchDao {
+	query := gen.Use(db.GetMysqlDB())
+	cache := cache.Use(db.GetRedis())
+
+	return &UserDispatchDao{
+		Dao: Dao{
+			ctx:   context.Background(),
+			query: query,
+			cache: &cache,
+		},
+
+		mu: sync.Mutex{},
+	}
+}
+
+// EnqueueStringLPush enqueue string left push.
+func (d *UserDispatchDao) EnqueueStringLPush(key, value string) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	err := (*d.cache).LPush(d.ctx, key, value)
+	if err != nil {
+		return errors.Wrapf(err, "enqueue key: %s , value: %s, err: %s", key, value, err.Error())
+	}
+
+	return nil
+}
+
+// EnqueueStringRPush enqueue string right push.
+func (d *UserDispatchDao) EnqueueStringRPush(key, value string) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	err := (*d.cache).RPush(d.ctx, key, value)
+	if err != nil {
+		return errors.Wrapf(err, "enqueue key: %s , value: %s, err: %s", key, value, err.Error())
+	}
+
+	return nil
+}
+
+// DequeueStringRPop dequeue string right pop.
+func (d *UserDispatchDao) DequeueStringRPop(key string) (string, error) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	value, err := (*d.cache).RPop(d.ctx, key)
+	if err != nil {
+		return "", errors.Wrapf(err, "dequeue key: %s , err: %s", key, err.Error())
+	}
+
+	return value, nil
+}
+
+// DequeueStringLPop dequeue string right pop.
+func (d *UserDispatchDao) DequeueStringLPop(key string) (string, error) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	value, err := (*d.cache).LPop(d.ctx, key)
+	if err != nil {
+		return "", errors.Wrapf(err, "dequeue key: %s , err: %s", key, err.Error())
+	}
+
+	return value, nil
+}
+
+// QueueLen get queue length.
+func (d *UserDispatchDao) QueueLen(key string) (int64, error) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	length, err := (*d.cache).LLen(d.ctx, key)
+	if err != nil {
+		return 0, errors.Wrapf(err, "enqueue key: %s , err: %s", key, err.Error())
+	}
+
+	return length, nil
+}
+
+// DequeueWithEnqueue dequeue with enqueue.
+func (d *UserDispatchDao) DequeueWithEnqueue(key string) (string, error) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	value, err := (*d.cache).RPop(d.ctx, key)
+	if err != nil {
+		return "", errors.Wrapf(err, "dequeue with enqueue: %s , err: %s", key, err.Error())
+	}
+
+	err = (*d.cache).LPush(d.ctx, key, value)
+	if err != nil {
+		return "", errors.Wrapf(err, "dequeue with enqueue: %s , err: %s", key, err.Error())
+	}
+
+	return value, nil
+}
+
+// AddUser add user to zset.
+func (d *UserDispatchDao) AddUser(key, value string) (int64, error) {
+	res, err := (*d.cache).ZAdd(d.ctx, key, &redis.Z{
+		Member: value,
+		Score:  float64(time.Now().Unix()),
+	})
+
+	return res, err
+}
+
+// RemoveUser remove user from zset.
+func (d *UserDispatchDao) RemoveUser(key, value string) (int64, error) {
+	res, err := (*d.cache).ZRem(d.ctx, key, value)
+
+	return res, err
+}
+
+// GetUser get user from zset top.
+func (d *UserDispatchDao) GetUser(key string) (string, error) {
+	// Get latest and set timestamp to now
+	res, err := (*d.cache).ZRange(d.ctx, key, 0, 0)
+	if err != nil || len(res) < 1 {
+		return "", err
+	}
+
+	// Update timestamp
+	value := res[0]
+	_, err = d.AddUser(key, value)
+	if err != nil {
+		return "", err
+	}
+
+	return value, err
+}

server/internal/middleware/hooks/gen_jwt_hook.go

@@ -44,6 +44,7 @@ func (h *JWT) Patterns() []string {
 		"/api/v1/community/*",
 		"/api/v1/admin/*",
 		"/api/v1/platform/*",
+		"/api/v1/login/exit", // temp, will remove to OpenIM Callback
 	}
 }
 

server/internal/router/router.go

@@ -61,6 +61,7 @@ func InitRouter() *gin.Engine {
 		login := apiv1.Group("/login")
 		{
 			login.POST("/account", api.AccountLogin)
+			login.POST("/exit", api.AccountExit)
 			// user.POST("/email", api.GithubRegister)
 			// user.POST("/github", api.GithubRegister)
 			// user.POST("/wechat", api.GithubRegister)

server/internal/service/user_dispatch.go

@@ -0,0 +1,61 @@
+// Copyright © 2023 OpenIM open source community. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/OpenKF/server/internal/dal/dao"
+)
+
+// USER_DISPATCH_QUEUE_KEY user dispatch queue key.
+const USER_DISPATCH_QUEUE_KEY = "openkf:user_dispatch_queue"
+
+// UserDispatchService user service.
+type UserDispatchService struct {
+	Service
+
+	UserDispatchDao *dao.UserDispatchDao
+}
+
+// NewUserDispatchService return new service with context.
+func NewUserDispatchService(c context.Context) *UserDispatchService {
+	return &UserDispatchService{
+		Service: Service{
+			ctx: c,
+		},
+
+		UserDispatchDao: dao.NewUserDispatchDao(),
+	}
+}
+
+// AddUser add user to enqueue.
+func (s *UserDispatchService) AddUser(uuid string) error {
+	_, err := s.UserDispatchDao.AddUser(USER_DISPATCH_QUEUE_KEY, uuid)
+
+	return err
+}
+
+// GetUser get user and update timestamp.
+func (s *UserDispatchService) GetUser() (string, error) {
+	return s.UserDispatchDao.GetUser(USER_DISPATCH_QUEUE_KEY)
+}
+
+// DeleteUser delete user from queue.
+func (s *UserDispatchService) DeleteUser(uuid string) error {
+	_, err := s.UserDispatchDao.RemoveUser(USER_DISPATCH_QUEUE_KEY, uuid)
+
+	return err
+}

server/internal/service/user_dispatch_test.go

@@ -0,0 +1,44 @@
+// Copyright © 2023 OpenIM open source community. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"context"
+	"testing"
+
+	"github.com/OpenIMSDK/OpenKF/server/internal/config"
+	"github.com/OpenIMSDK/OpenKF/server/internal/conn/db"
+	"github.com/OpenIMSDK/OpenKF/server/pkg/log"
+)
+
+// TestUserQueue test user queue.
+func TestUserQueue(t *testing.T) {
+	// Init
+	config.ConfigInit("../../config.yaml")
+	log.InitLogger()
+	db.InitMysqlDB()
+	db.InitRedisDB()
+
+	s := NewUserDispatchService(context.Background())
+	// err := s.AddUser("test3")
+	// if err != nil {
+	// 	t.Error(err)
+	// }
+	value, err := s.GetUser()
+	if err != nil {
+		t.Error(err)
+	}
+	t.Error(value)
+}

server/main.go

@@ -43,7 +43,6 @@ func init() {
 	client.InitMinio()
 	// client.InitMail()
 	hooks.InitHooks()
-	slackcmd.InitSlackListen()
 }
 
 //go:generate go env -w GO111MODULE=on
@@ -60,6 +59,9 @@ func init() {
 func main() {
 	serverAddress := fmt.Sprintf("%s:%d", config.Config.Server.Ip, config.Config.Server.Port)
 
+	// Add slack server
+	go slackcmd.InitSlackListen()
+
 	r := router.InitRouter()
 	s := server.InitServer(serverAddress, r)
 	log.Error("server start error: %v", s.ListenAndServe().Error())

web/src/api/index/login.ts

@@ -25,6 +25,7 @@ const API = {
     RegisterAdmin: '/register/admin',
     RegisterStaff: '/register/staff',
     AccountLogin: '/login/account',
+    AccountExit: '/login/exit',
 };
 
 // Send email verification code
@@ -60,3 +61,10 @@ export function accountLogin(data: AccountLoginParam) {
         data,
     });
 }
+
+// User logout
+export function accountLogout() {
+    return request.post<any>({
+        url: API.AccountExit,
+    });
+}

web/src/views/layouts/components/LayoutSideNav.vue

@@ -3,6 +3,7 @@ import { ref } from 'vue';
 import { computed, onMounted } from 'vue';
 import { useRouter } from 'vue-router';
 import { useMenuStore } from '@/store';
+import { accountLogout } from '@/api/index/login';
 
 const router = useRouter();
 const menuStore = useMenuStore();
@@ -24,8 +25,9 @@ const changeHandler = (active:string) => {
 };
 
 
-const goHome = () => {
+const goHome = async () => {
   router.push('/home/dashboard');
+  await accountLogout()
 };
 
 </script>