diff --git a/server/internal/api/callback.go b/server/internal/api/callback.go index c4424a8..cbb1ba6 100644 --- a/server/internal/api/callback.go +++ b/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. diff --git a/server/internal/api/platform.go b/server/internal/api/platform.go index 0672446..98b27d9 100644 --- a/server/internal/api/platform.go +++ b/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(¶m) + 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) +} 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..103a88f 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,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() +} diff --git a/server/internal/dal/cache/redis_test.go b/server/internal/dal/cache/redis_test.go new file mode 100644 index 0000000..750383e --- /dev/null +++ b/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) + } +} diff --git a/server/internal/dal/dao/slacker_marshal_helper.go b/server/internal/dal/dao/slacker_marshal_helper.go new file mode 100644 index 0000000..0339bb7 --- /dev/null +++ b/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, + } +} 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..88f28a7 --- /dev/null +++ b/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) +} 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/msg/slack_cmd/slack.go b/server/internal/msg/slack_cmd/slack.go index b276f50..08b43a0 100644 --- a/server/internal/msg/slack_cmd/slack.go +++ b/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("", &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()) + } +} diff --git a/server/internal/params/request/msg.go b/server/internal/params/request/msg.go new file mode 100644 index 0000000..f72643d --- /dev/null +++ b/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"` +} diff --git a/server/internal/router/router.go b/server/internal/router/router.go index 5931bf7..c5c479e 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) @@ -124,6 +125,7 @@ func InitRouter() *gin.Engine { slack := platform.Group("/slack") { slack.GET("/config", api.SlackConfig) + slack.POST("/customer", api.GetSlackCustomer) } } } diff --git a/server/internal/service/slack.go b/server/internal/service/slack.go index c21de55..25688b3 100644 --- a/server/internal/service/slack.go +++ b/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 +} diff --git a/server/internal/service/user.go b/server/internal/service/user.go index a2e4f94..c6c37b1 100644 --- a/server/internal/service/user.go +++ b/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{} diff --git a/server/internal/service/user_dispatch.go b/server/internal/service/user_dispatch.go new file mode 100644 index 0000000..f418da6 --- /dev/null +++ b/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 +} 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/pkg/openim/sdk/auth/auth_test.go b/server/pkg/openim/sdk/auth/auth_test.go index 48c83fe..a65f8cd 100644 --- a/server/pkg/openim/sdk/auth/auth_test.go +++ b/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) } } diff --git a/server/pkg/openim/sdk/msg/msg.go b/server/pkg/openim/sdk/msg/msg.go index 38e833e..af22766 100644 --- a/server/pkg/openim/sdk/msg/msg.go +++ b/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"] 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/api/index/platform.ts b/web/src/api/index/platform.ts index fd2720e..aebc2b7 100644 --- a/web/src/api/index/platform.ts +++ b/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({ + url: API.SlackCustomer, + params: data, + }); +} \ No newline at end of file diff --git a/web/src/api/response/platformModel.ts b/web/src/api/response/platformModel.ts index 53e04c6..2b64c3f 100644 --- a/web/src/api/response/platformModel.ts +++ b/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, } \ No newline at end of file diff --git a/web/src/constants/index.ts b/web/src/constants/index.ts index 12e6ff1..61f826e 100644 --- a/web/src/constants/index.ts +++ b/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 diff --git a/web/src/views/session/components/SessionWindow.vue b/web/src/views/session/components/SessionWindow.vue index c4c9753..3e97481 100644 --- a/web/src/views/session/components/SessionWindow.vue +++ b/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!');