diff --git a/server/internal/api/callback.go b/server/internal/api/callback.go index c4424a8..a71ead3 100644 --- a/server/internal/api/callback.go +++ b/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. diff --git a/server/internal/api/user.go b/server/internal/api/user.go index 5b0c2e7..fcecae1 100644 --- a/server/internal/api/user.go +++ b/server/internal/api/user.go @@ -74,7 +74,16 @@ func AccountLogin(c *gin.Context) { svc := service.NewUserService(c) u, err := svc.LoginWithAccount(¶ms) 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 diff --git a/server/internal/dal/cache/redis.go b/server/internal/dal/cache/redis.go index 95312b3..0a28593 100644 --- a/server/internal/dal/cache/redis.go +++ b/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() +// } diff --git a/server/internal/dal/dao/user_dispatch_dao.go b/server/internal/dal/dao/user_dispatch_dao.go new file mode 100644 index 0000000..9cba8de --- /dev/null +++ b/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 +} diff --git a/server/internal/middleware/hooks/gen_jwt_hook.go b/server/internal/middleware/hooks/gen_jwt_hook.go index 42c91cd..c089fcf 100644 --- a/server/internal/middleware/hooks/gen_jwt_hook.go +++ b/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 } } diff --git a/server/internal/router/router.go b/server/internal/router/router.go index 5931bf7..89b704c 100644 --- a/server/internal/router/router.go +++ b/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) diff --git a/server/internal/service/user_dispatch.go b/server/internal/service/user_dispatch.go new file mode 100644 index 0000000..f35fce6 --- /dev/null +++ b/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 +} diff --git a/server/internal/service/user_dispatch_test.go b/server/internal/service/user_dispatch_test.go new file mode 100644 index 0000000..e9d25d8 --- /dev/null +++ b/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) +} diff --git a/server/main.go b/server/main.go index a01b745..41ac86d 100644 --- a/server/main.go +++ b/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()) diff --git a/web/src/api/index/login.ts b/web/src/api/index/login.ts index faca2f9..d57dcfa 100644 --- a/web/src/api/index/login.ts +++ b/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({ + url: API.AccountExit, + }); +} \ No newline at end of file diff --git a/web/src/views/layouts/components/LayoutSideNav.vue b/web/src/views/layouts/components/LayoutSideNav.vue index c254a90..84c23a9 100644 --- a/web/src/views/layouts/components/LayoutSideNav.vue +++ b/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() };