Merge pull request 'Add gitlink PR.' (#1) from asklv/OpenKF:dev/dev-08-11 into main

server/internal/api/callback.go

@@ -22,7 +22,10 @@ import (
 
 	"github.com/OpenIMSDK/OpenKF/server/internal/common"
 	"github.com/OpenIMSDK/OpenKF/server/internal/common/response"
+	slackcmd "github.com/OpenIMSDK/OpenKF/server/internal/msg/slack_cmd"
 	requestparams "github.com/OpenIMSDK/OpenKF/server/internal/params/request"
+	"github.com/OpenIMSDK/OpenKF/server/internal/service"
+	"github.com/OpenIMSDK/OpenKF/server/pkg/log"
 )
 
 // OpenIMCallback
@@ -50,6 +53,21 @@ func OpenIMCallback(c *gin.Context) {
 
 // BeforeSendSingleMsg.
 func BeforeSendSingleMsg(c *gin.Context) {
+	// Msg filter
+	params := &requestparams.MsgAbstract{}
+	err := c.ShouldBindJSON(params)
+	if err != nil {
+		// Do not check
+		return
+	}
+
+	// Send message to slack user
+	ssvc := service.NewUserDispatchService(c)
+	if check := ssvc.SlackUserFilter(params.RecvID); check {
+		log.Debugf("Msg filter", "Send message to slack: %s", params.RecvID)
+		slackcmd.SendMsg(params)
+	}
+
 }
 
 // AfterSendSingleMsg.
@@ -62,10 +80,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/platform.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/OpenIMSDK/OpenKF/server/internal/common"
 	"github.com/OpenIMSDK/OpenKF/server/internal/common/response"
+	requestparams "github.com/OpenIMSDK/OpenKF/server/internal/params/request"
 	"github.com/OpenIMSDK/OpenKF/server/internal/service"
 	"github.com/OpenIMSDK/OpenKF/server/internal/utils"
 )
@@ -51,3 +52,33 @@ func SlackConfig(c *gin.Context) {
 
 	response.SuccessWithData(resp, c)
 }
+
+// GetSlackCustomer
+// @Tags platform
+// @Summary GetSlackCustomer
+// @Description get slack user info
+// @Security  ApiKeyAuth
+// Param data body param.GetUserInfoParams true "GetUserInfoParams"
+// @Success 200 {object}  response.Response{msg=string} "Success"
+// @Router /api/v1/platform/slack/customer [post].
+func GetSlackCustomer(c *gin.Context) {
+	// TODO: Check role.
+
+	param := requestparams.GetUserInfoParams{}
+	err := c.ShouldBindJSON(&param)
+	if err != nil {
+		response.FailWithCode(common.INVALID_PARAMS, c)
+
+		return
+	}
+
+	svc := service.NewSlackService(c)
+	resp, err := svc.GetSlackUser(param.UUID)
+	if err != nil {
+		response.FailWithCode(common.KF_RECORD_NOT_FOUND, c)
+
+		return
+	}
+
+	response.SuccessWithData(resp, c)
+}

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,41 @@ 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)
+
+	// Map
+	HSet(ctx context.Context, key, field string, value string) error
+	HGet(ctx context.Context, key, field string) (string, error)
+	HGetAll(ctx context.Context, key string) (map[string]string, error)
+	HDel(ctx context.Context, key string, fields ...string) (int64, error)
+	HExists(ctx context.Context, key, field string) (bool, error)
+	HIncrBy(ctx context.Context, key, field string, incr int64) (int64, error)
+	HKeys(ctx context.Context, key string) ([]string, error)
+	HLen(ctx context.Context, key string) (int64, error)
+	HMGet(ctx context.Context, key string, fields ...string) ([]interface{}, error)
+	HMSet(ctx context.Context, key string, values ...interface{}) 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 +160,188 @@ 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()
+// }
+
+// HSet hset.
+func (c *cache) HSet(ctx context.Context, key, field string, value string) error {
+	return c.client.HSet(ctx, key, field, value).Err()
+}
+
+// HGet hget.
+func (c *cache) HGet(ctx context.Context, key, field string) (string, error) {
+	return c.client.HGet(ctx, key, field).Result()
+}
+
+// HGetAll hgetall.
+func (c *cache) HGetAll(ctx context.Context, key string) (map[string]string, error) {
+	return c.client.HGetAll(ctx, key).Result()
+}
+
+// HDel hdel.
+func (c *cache) HDel(ctx context.Context, key string, fields ...string) (int64, error) {
+	return c.client.HDel(ctx, key, fields...).Result()
+}
+
+// HExists hexists.
+func (c *cache) HExists(ctx context.Context, key, field string) (bool, error) {
+	return c.client.HExists(ctx, key, field).Result()
+}
+
+// HIncrBy hincrby.
+func (c *cache) HIncrBy(ctx context.Context, key, field string, incr int64) (int64, error) {
+	return c.client.HIncrBy(ctx, key, field, incr).Result()
+}
+
+// HKeys hkeys.
+func (c *cache) HKeys(ctx context.Context, key string) ([]string, error) {
+	return c.client.HKeys(ctx, key).Result()
+}
+
+// HLen hlen.
+func (c *cache) HLen(ctx context.Context, key string) (int64, error) {
+	return c.client.HLen(ctx, key).Result()
+}
+
+// HMGet hmget.
+func (c *cache) HMGet(ctx context.Context, key string, fields ...string) ([]interface{}, error) {
+	return c.client.HMGet(ctx, key, fields...).Result()
+}
+
+// HMSet hmset.
+func (c *cache) HMSet(ctx context.Context, key string, values ...interface{}) error {
+	return c.client.HMSet(ctx, key, values).Err()
+}

server/internal/dal/cache/redis_test.go

@@ -0,0 +1,60 @@
+// 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 cache
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"testing"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+)
+
+func TestSub(t *testing.T) {
+	client := redis.NewClient(&redis.Options{
+		Addr:     "localhost:6380",
+		Password: "openkf",
+		DB:       0,
+	})
+
+	ctx := context.Background()
+	pubsub := client.PSubscribe(ctx, "__keyevent@0__:expired")
+
+	ch := pubsub.Channel()
+	go func() {
+		for msg := range ch {
+			fmt.Println("Events:", msg.Payload)
+		}
+	}()
+
+	err := client.Set(ctx, "mykey", "myvalue", 5*time.Second).Err()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	time.Sleep(10 * time.Second)
+
+	err = pubsub.Close()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = client.Close()
+	if err != nil {
+		log.Fatal(err)
+	}
+}

server/internal/dal/dao/slacker_marshal_helper.go

@@ -0,0 +1,30 @@
+// 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
+
+// SlackMap is the map of slack writer and used for struct unmarshal
+type SlackMap struct {
+	StaffID        string `json:"staff_id"`
+	SlackChannelID string `json:"slack_channel_id"`
+	// SlackBotContext used to store the slack context, and use NewResponse to get writer.
+}
+
+// NewSlackMap returns a new SlackMap
+func NewSlackMap(staffID, slackChannelID string) *SlackMap {
+	return &SlackMap{
+		StaffID:        staffID,
+		SlackChannelID: slackChannelID,
+	}
+}

server/internal/dal/dao/user_dispatch_dao.go

@@ -0,0 +1,212 @@
+// 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"
+	"encoding/json"
+	"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
+}
+
+// SetSlackMap set staffId and slack channel id.
+func (d *UserDispatchDao) SetSlackMap(key, customID, staffID, slackChannelID string) error {
+	sMap := NewSlackMap(staffID, slackChannelID)
+
+	// JSON encode
+	value, err := json.Marshal(sMap)
+	if err != nil {
+		return errors.Wrapf(err, "Marshal user map: %s , err: %s", key, err.Error())
+	}
+
+	// SetToMap
+	err = (*d.cache).HSet(d.ctx, key, customID, string(value))
+	if err != nil {
+		return errors.Wrapf(err, "SetToMap user map: %s , err: %s", key, err.Error())
+	}
+
+	return nil
+}
+
+// GetSlackMap get staffId and slack channel id.
+func (d *UserDispatchDao) GetSlackMap(key, customID string) *SlackMap {
+	// Get from map
+	value, err := (*d.cache).HGet(d.ctx, key, customID)
+	if err != nil {
+		return &SlackMap{}
+	}
+
+	// JSON decode
+	sMap := &SlackMap{}
+	err = json.Unmarshal([]byte(value), sMap)
+	if err != nil {
+		return &SlackMap{}
+	}
+
+	return sMap
+}
+
+// GetSlackIDs get slack user ids.
+func (d *UserDispatchDao) GetSlackIDs(key string) ([]string, error) {
+	return (*d.cache).HKeys(d.ctx, key)
+}

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/msg/slack_cmd/slack.go

@@ -16,6 +16,8 @@ package slackcmd
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"strings"
 
 	"github.com/shomali11/slacker"
@@ -25,18 +27,30 @@ import (
 	"github.com/OpenIMSDK/OpenKF/server/pkg/log"
 )
 
+// SLACK_PERFIX do not use separator.
+const SLACK_PERFIX = "SLACK"
+
+// BOT_CONTEXT_MAP bot context map in memory. TODO: use better way to store this map.
+var BOT_CONTEXT_MAP map[string]slacker.BotContext
+
 // InitSlackListen init slack socket listen.
 func InitSlackListen() {
 	bot := client.InitSlack()
 
-	svc := service.NewSlackService(context.Background())
+	// init bot context map
+	BOT_CONTEXT_MAP = make(map[string]slacker.BotContext)
+	ctx := context.Background()
+	svc := service.NewSlackService(ctx)
 
 	// receive all @ message
 	bot.Command("<question>", &slacker.CommandDefinition{
 		Handler: func(bc slacker.BotContext, r slacker.Request, w slacker.ResponseWriter) {
+			BOT_CONTEXT_MAP[bc.Event().ChannelID] = bc
+
 			// find or create a customer
 			messageEvent := bc.Event()
-			senderId, _, err := svc.CreateCustomer(messageEvent.UserID, messageEvent.UserProfile)
+			slackUserId := fmt.Sprintf("%s%s", SLACK_PERFIX, messageEvent.UserID)
+			senderId, _, err := svc.CreateCustomer(slackUserId, messageEvent.UserProfile)
 			if err != nil {
 				log.Errorf("Slack", "CreateCustomer failed: %s", err.Error())
 
@@ -48,10 +62,12 @@ func InitSlackListen() {
 			words := strings.Split(question, " ")
 			question = strings.Join(words[1:], " ")
 
-			// TODO: get staff id and get into im process.
-			_ = senderId
-			_ = question
-			_ = w.Reply("senderId", slacker.WithThreadReply(true))
+			// send message to staff and cache pair
+			err = svc.SendMsg(senderId, question, bc)
+			if err != nil {
+				log.Errorf("Slack", "SendMsg failed: %s", err.Error())
+				_ = w.Reply("[OpenKF] Error in system.", slacker.WithThreadReply(true))
+			}
 		},
 	})
 
@@ -62,3 +78,37 @@ func InitSlackListen() {
 		log.Panicf("Slack Client", "Connection failed: %s", err.Error())
 	}
 }
+
+// SendMsg to slack user.
+func SendMsg(params *requestparams.MsgAbstract) {
+	recvID := params.RecvID
+	content := params.Content
+	ctx := context.Background()
+
+	// Parse content {"content": "xxx"}
+	type Content struct {
+		Content string `json:"content"`
+	}
+	c := &Content{}
+	err := json.Unmarshal([]byte(content), &c)
+	if err != nil {
+		log.Errorf("Slack", "SendMsg json.Unmarshal failed: %s", err.Error())
+
+		return
+	}
+
+	udSvc := service.NewUserDispatchService(ctx)
+	sMap := udSvc.GetSlackMap(recvID)
+	slackCtx, ok := BOT_CONTEXT_MAP[sMap.SlackChannelID]
+	if !ok {
+		log.Errorf("Slack", "Slack context not found")
+
+		return
+	}
+
+	// Reply in thread and return response
+	err = slacker.NewResponse(slackCtx).Reply(c.Content, slacker.WithThreadReply(true))
+	if err != nil {
+		log.Errorf("Slack", "SendToSlack failed: %s", err.Error())
+	}
+}

server/internal/params/request/msg.go

@@ -0,0 +1,22 @@
+// 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 requestparams
+
+// MsgAbstract msg abstract info.
+type MsgAbstract struct {
+	SendID  string `json:"sendID"           binding:"required"`
+	RecvID  string `json:"recvID"           binding:"required"`
+	Content string `json:"content"          binding:"required"`
+}

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)
@@ -124,6 +125,7 @@ func InitRouter() *gin.Engine {
 			slack := platform.Group("/slack")
 			{
 				slack.GET("/config", api.SlackConfig)
+				slack.POST("/customer", api.GetSlackCustomer)
 			}
 		}
 	}

server/internal/service/slack.go

@@ -16,8 +16,13 @@ package service
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"net"
 
+	"github.com/gin-gonic/gin"
+	"github.com/pkg/errors"
+	"github.com/shomali11/slacker"
 	"github.com/slack-go/slack"
 
 	"github.com/OpenIMSDK/OpenKF/server/internal/config"
@@ -25,7 +30,10 @@ import (
 	"github.com/OpenIMSDK/OpenKF/server/internal/models/base"
 	customerroles "github.com/OpenIMSDK/OpenKF/server/internal/models/customer_roles"
 	responseparams "github.com/OpenIMSDK/OpenKF/server/internal/params/response"
+	"github.com/OpenIMSDK/OpenKF/server/pkg/log"
 	"github.com/OpenIMSDK/OpenKF/server/pkg/openim/param/request"
+	"github.com/OpenIMSDK/OpenKF/server/pkg/openim/sdk/constant"
+	"github.com/OpenIMSDK/OpenKF/server/pkg/openim/sdk/msg"
 	"github.com/OpenIMSDK/OpenKF/server/pkg/utils"
 )
 
@@ -77,7 +85,7 @@ func (svc *SlackService) CreateCustomer(userId string, profile *slack.UserProfil
 
 	customerSlack := customerroles.CustomerSlack{
 		UserBase: base.UserBase{
-			// UUID:        fmt.Sprintf("%s%s", SLACK_PERFIX, userId),
+			// UUID: fmt.Sprintf("%s%s", dao.SLACK_PERFIX, userId),
 			UUID:        userId,
 			Email:       profile.Email,
 			Nickname:    profile.FirstName,
@@ -134,3 +142,74 @@ func (svc *SlackService) CreateCustomer(userId string, profile *slack.UserProfil
 
 	return userId, s.Id, nil
 }
+
+// GetSlackUser get slack user.
+func (svc *SlackService) GetSlackUser(userId string) (*customerroles.CustomerSlack, error) {
+	return svc.CustomerSlackDao.FindFirstByUUID(userId)
+}
+
+// SendMsg send message to openkf.
+func (svc *SlackService) SendMsg(uid, question string, botContext slacker.BotContext) error {
+	udSvc := NewUserDispatchService(svc.ctx)
+
+	// Get default staff id
+	sMap := udSvc.GetSlackMap(uid)
+	staffId := sMap.StaffID
+
+	if sMap == nil || sMap.StaffID == "" || sMap.SlackChannelID == "" {
+		// Get staff id
+		tempStaffId, err := udSvc.GetUser()
+		if err != nil {
+			return err
+		}
+
+		// Store staff, writer to cache redis map
+		err = udSvc.SetSlackMap(uid, tempStaffId, botContext)
+		if err != nil {
+			log.Error("wer", err.Error())
+
+			return errors.Wrapf(err, "set slack map failed")
+		}
+		staffId = tempStaffId
+	}
+
+	// Get OpenIM admin token
+	uSvc := NewUserService(&gin.Context{}) // TODO: Change context to same
+	token, err := uSvc.GetAdminToken()
+	if err != nil {
+		return errors.Wrapf(err, "get admin token failed")
+	}
+
+	// Get custom info
+	customer, err := svc.CustomerSlackDao.FindFirstByUUID(uid)
+	if err != nil {
+		return errors.Wrapf(err, "find customer failed")
+	}
+
+	msgInfo := &request.MsgInfo{
+		SendID:           uid,
+		RecvID:           staffId,
+		GroupID:          "",
+		SenderNickname:   customer.Nickname,
+		SenderFaceURL:    customer.Avatar,
+		SenderPlatformID: constant.PLATFORMID_WEB,
+		Content: &request.TextContent{
+			Text: fmt.Sprintf("{\"content\":\"%s\"}", question),
+		},
+		ContentType:     constant.CONTENT_TYPE_TEXT,
+		SessionType:     constant.SESSION_TYPE_SINGLE_CHAT,
+		IsOnlineOnly:    false,
+		NotOfflinePush:  false,
+		OfflinePushInfo: &request.OfflinePushInfo{},
+	}
+	res, _ := json.Marshal(msgInfo)
+	log.Error("msgInfo", string(res))
+	host := fmt.Sprintf("http://%s", net.JoinHostPort(config.Config.OpenIM.Ip, fmt.Sprintf("%d", config.Config.OpenIM.ApiPort)))
+	resp, err := msg.AdminSendMsg(msgInfo, "sendMsg:"+uid, host, token.Token)
+	if err != nil {
+		return errors.Wrapf(err, "send msg failed")
+	}
+	log.Debugf("AdminSendMsg", "Resp: %+v", resp)
+
+	return nil
+}

server/internal/service/user.go

@@ -277,6 +277,19 @@ func getUserIMToken(param *request.UserTokenParams) (*response.TokenData, error)
 	return &resp.Data, nil
 }
 
+// GetAdminToken get admin token.
+func (svc *UserService) GetAdminToken() (*response.TokenData, error) {
+	params := &request.UserTokenParams{
+		Secret:     config.Config.OpenIM.Secret,
+		PlatformID: uint(config.Config.OpenIM.PlatformID),
+		UserID:     config.Config.OpenIM.AdminID,
+	}
+
+	// TODO: Get cache from redis
+
+	return getUserIMToken(params)
+}
+
 // GetUserInfoByUUID get user info by uuid.
 func (svc *UserService) GetUserInfoByUUID(uid string) (*responseparams.UserInfoResponse, error) {
 	resp := &responseparams.UserInfoResponse{}

server/internal/service/user_dispatch.go

@@ -0,0 +1,112 @@
+// 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"
+	"errors"
+
+	"github.com/OpenIMSDK/OpenKF/server/internal/dal/dao"
+	"github.com/shomali11/slacker"
+)
+
+// USER_DISPATCH_QUEUE_KEY user dispatch queue key.
+const USER_DISPATCH_QUEUE_KEY = "openkf:user_dispatch_queue"
+const USER_SLACK_MAP_KEY = "openkf:user_slack_map"
+
+// UserDispatchService user service.
+type UserDispatchService struct {
+	Service
+
+	UserDispatchDao *dao.UserDispatchDao
+	SysUserDao      *dao.SysUserDao
+}
+
+// NewUserDispatchService return new service with context.
+func NewUserDispatchService(c context.Context) *UserDispatchService {
+	return &UserDispatchService{
+		Service: Service{
+			ctx: c,
+		},
+
+		UserDispatchDao: dao.NewUserDispatchDao(),
+		SysUserDao:      dao.NewSysUserDao(),
+	}
+}
+
+// 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) {
+	res, err := s.UserDispatchDao.GetUser(USER_DISPATCH_QUEUE_KEY)
+	if err != nil {
+		return "", errors.New("can not find a user")
+	}
+
+	// return a default value if res is empty
+	if res == "" {
+		u, err := s.SysUserDao.First()
+		if err != nil {
+			return "", err
+		}
+
+		return u.UUID, nil
+	}
+
+	return res, nil
+}
+
+// DeleteUser delete user from queue.
+func (s *UserDispatchService) DeleteUser(uuid string) error {
+	_, err := s.UserDispatchDao.RemoveUser(USER_DISPATCH_QUEUE_KEY, uuid)
+
+	return err
+}
+
+// SetSlackMap set slack map.
+func (s *UserDispatchService) SetSlackMap(customID, staffID string, botContext slacker.BotContext) error {
+	return s.UserDispatchDao.SetSlackMap(USER_SLACK_MAP_KEY, customID, staffID, botContext.Event().ChannelID)
+}
+
+// GetSlackMap get slack map.
+func (s *UserDispatchService) GetSlackMap(customID string) *dao.SlackMap {
+	return s.UserDispatchDao.GetSlackMap(USER_SLACK_MAP_KEY, customID)
+}
+
+// GetStaffID get all staff id.
+func (s *UserDispatchService) GetSlackIDs() ([]string, error) {
+	return s.UserDispatchDao.GetSlackIDs(USER_SLACK_MAP_KEY)
+}
+
+// SlackUserFilter filter user by slack id.
+func (s *UserDispatchService) SlackUserFilter(uid string) bool {
+	ids, err := s.GetSlackIDs()
+	if err != nil || len(ids) == 0 {
+		return false
+	}
+
+	for _, id := range ids {
+		if id == uid {
+			return true
+		}
+	}
+
+	return false
+}

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/pkg/openim/sdk/auth/auth_test.go

@@ -30,15 +30,15 @@ func TestGetUserToken(t *testing.T) {
 		userID     string
 	}{
 		{
-			secret:     "openkf",
-			platformID: 1,
+			secret:     "openIM123",
+			platformID: 5,
 			userID:     "openIMAdmin",
 		},
 	}
 
 	// range test case
 	for _, data := range testData {
-		_, err := auth.GetUserToken(&request.UserTokenParams{
+		token, err := auth.GetUserToken(&request.UserTokenParams{
 			Secret:     data.secret,
 			PlatformID: data.platformID,
 			UserID:     data.userID,
@@ -48,5 +48,6 @@ func TestGetUserToken(t *testing.T) {
 		if err != nil {
 			t.Error(err)
 		}
+		t.Error(token)
 	}
 }

server/pkg/openim/sdk/msg/msg.go

@@ -45,7 +45,7 @@ func AdminSendMsg(param *request.MsgInfo, operationID, host, adminToken string)
 	r.ErrDlt = resp["errDlt"].(string)
 
 	if resp["data"] == nil {
-		return r, errors.New("data is nil")
+		return r, errors.New("data is nil: " + r.ErrMsg)
 	}
 
 	r.Data = resp["data"]

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/api/index/platform.ts

@@ -13,10 +13,13 @@
 // limitations under the License.
 
 import { request } from '@/utils/request';
-import { GetSlackConfigResponse } from '@/api/response/platformModel';
+import { GetSlackConfigResponse, GetSlackCustomerResponse } from '@/api/response/platformModel';
+import { GetUserInfoParam } from '@/api/request/userModel';
+
 
 const API = {
     SlackConfig: '/platform/slack/config',
+    SlackCustomer: '/platform/slack/customer',
 };
 
 // get slack info
@@ -25,3 +28,11 @@ export function getSlackConfig() {
         url: API.SlackConfig,
     });
 }
+
+// get slack customer info
+export function getSlackCustomerInfo(data: GetUserInfoParam) {
+    return request.post<GetSlackCustomerResponse>({
+        url: API.SlackCustomer,
+        params: data,
+    });
+}

web/src/api/response/platformModel.ts

@@ -6,4 +6,34 @@ export interface GetSlackConfigResponse {
     client_secret: string,
     signing_secret: string,
     verification_token: string,
+}
+
+export interface GetSlackCustomerResponse {
+    id: number,
+    created_at: string,
+    updated_at: string,
+    uuid: string,
+    email: string,
+    nickname: string,
+    avatar: string,
+    description: string,
+    is_enable: boolean,
+    first_name: string,
+    last_name: string,
+    real_name: string,
+    real_name_normalized: string,
+    display_name: string,
+    display_name_normalized: string,
+    skype: string,
+    phone: string,
+    image_24: string,
+    image_32: string,
+    image_48: string,
+    image_72: string,
+    image_192: string,
+    image_512: string,
+    image_original: string,
+    title: string,
+    status_expiration: number,
+    team: string,
 }

web/src/constants/index.ts

@@ -29,8 +29,8 @@ export enum ContentTypeEnum {
 
 // Platform type
 export enum PlatformType {
-    Slack = 'Slack',
-    Web = 'Web',
+    Slack = 'SLACK',
+    Web = 'WEB',
 }
 
 // Define support platforms

web/src/views/session/components/SessionWindow.vue

@@ -3,6 +3,7 @@ import { ErrorCodes, reactive, ref, watch, nextTick } from 'vue';
 import { MessagePlugin } from 'tdesign-vue-next';
 import { Icon } from 'tdesign-icons-vue-next';
 import { getMyInfo, updateUserInfo } from '@/api/index/user';
+import { getSlackCustomerInfo } from '@/api/index/platform';
 import { getMyCommunityInfo, updateCommunityInfo } from '@/api/index/community';
 import { updateStaffEnableStatus, deleteStaff } from '@/api/index/admin';
 import { UpdateUserInfoParam } from '@/api/request/userModel';
@@ -20,6 +21,7 @@ import { CbEvents } from "@/utils/open-im-sdk-wasm/constant";
 import { WSEvent } from "@/utils/open-im-sdk-wasm/types/entity";
 import useUserStore from '@/store/user';
 import useMessageStore from '@/store/openim_message';
+import { PlatformType } from '@/constants';
 
 const InitUserInfo:GetUserInfoResponse = {
   uuid: '',
@@ -63,12 +65,27 @@ watch(() => messageStore, (newVal, oldVal) => {
 
 const fetchRecvInfo = async () => {
   try {
+    const uuid = getRecvIdFromSessionId(props.session?.conversationID || '')
     const params: GetUserInfoParam = {
-      uuid: getRecvIdFromSessionId(props.session?.conversationID || '')
+      uuid: uuid
+    }
+
+    if (uuid.includes(PlatformType.Slack)) {
+      // fetch info from slack
+      const info = await getSlackCustomerInfo(params);
+      recvInfo.value.uuid = info.uuid;
+      recvInfo.value.email = info.email;
+      recvInfo.value.nickname = info.nickname;
+      recvInfo.value.avatar = info.avatar;
+      recvInfo.value.description = info.description;
+      recvInfo.value.is_enable = info.is_enable;
+      recvInfo.value.created_at = info.created_at;
+    } else {
+      // fetch info from openkf
+      const info = await getUserInfo(params);
+      recvInfo.value = info;
     }
 
-    const info = await getUserInfo(params);
-    recvInfo.value = info;
   } catch (e) {
     console.log(e);
     MessagePlugin.error('Fetch recv info error!');