feat: add customer session refresh functionality and improve session management
- Introduced RealtimeCustomerSessionRefreshPayload and RealtimeCustomerSessionRefreshEvent types for handling session refresh events. - Updated ws_service to verify customer session and handle session refresh notifications. - Enhanced API client to manage customer session tokens and expiration. - Implemented customer session validation and storage in session storage. - Added functions to exchange and ensure customer sessions. - Updated IM real-time connection to include customer session tokens in WebSocket requests. - Modified SDK and widget configurations to support external IDs and user tokens.
This commit is contained in:
mlogclub
@@ -28,9 +28,9 @@ func NewServer() (*iris.Application, error) {
|
||||
app := iris.New()
|
||||
corsHandler := cors.New().
|
||||
AllowOrigin("*").
|
||||
AllowHeaders("Origin", "Content-Type", "Accept", "Authorization", "X-Requested-With", "X-Guest-Id", "X-Channel-Id", "X-External-Id", "X-External-Name").
|
||||
AllowHeaders("Origin", "Content-Type", "Accept", "Authorization", "X-Requested-With", "X-Guest-Id", "X-Channel-Id", "X-External-Id", "X-External-Name", "X-Customer-Session-Token", "X-Customer-Session-Expires-At").
|
||||
MaxAge(600).
|
||||
ExposeHeaders("Content-Length", "Content-Type", "Authorization", "X-Guest-Id", "X-Channel-Id", "X-External-Id", "X-External-Name").
|
||||
ExposeHeaders("Content-Length", "Content-Type", "Authorization", "X-Guest-Id", "X-Channel-Id", "X-External-Id", "X-External-Name", "X-Customer-Session-Token", "X-Customer-Session-Expires-At").
|
||||
Handler()
|
||||
app.UseRouter(func(ctx iris.Context) {
|
||||
// WebSocket upgrade is validated by the upgrader's origin policy.
|
||||
@@ -95,6 +95,7 @@ func addRouter(app *iris.Application) {
|
||||
mvc.Configure(app.Party("/api"), func(m *mvc.Application) {
|
||||
m.Party("/auth").Handle(new(api.AuthController))
|
||||
m.Party("/channel").Handle(new(api.ChannelController))
|
||||
m.Party("/customer").Handle(new(api.CustomerController))
|
||||
m.Party("/conversation", middleware.ExternalUserMiddleware).Handle(new(api.ConversationController))
|
||||
m.Party("/message", middleware.ExternalUserMiddleware).Handle(new(api.MessageController))
|
||||
})
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"cs-agent/internal/pkg/openidentity"
|
||||
"cs-agent/internal/services"
|
||||
|
||||
"github.com/kataras/iris/v12"
|
||||
"github.com/mlogclub/simple/web"
|
||||
)
|
||||
|
||||
type CustomerController struct {
|
||||
Ctx iris.Context
|
||||
}
|
||||
|
||||
func (c *CustomerController) PostSession_exchange() *web.JsonResult {
|
||||
channel := services.ChannelService.GetEnabledChannel(c.Ctx)
|
||||
if channel == nil {
|
||||
return web.JsonErrorMsg("接入渠道不存在或已停用")
|
||||
}
|
||||
externalUser, err := openidentity.GetExternalUser(c.Ctx, services.ChannelService.GetUserTokenSecret(channel))
|
||||
if err != nil {
|
||||
return web.JsonError(err)
|
||||
}
|
||||
resp, err := services.CustomerSessionService.Exchange(channel, *externalUser)
|
||||
if err != nil {
|
||||
return web.JsonError(err)
|
||||
}
|
||||
return web.JsonData(resp)
|
||||
}
|
||||
@@ -2,7 +2,6 @@ package middleware
|
||||
|
||||
import (
|
||||
"cs-agent/internal/pkg/irisx"
|
||||
"cs-agent/internal/pkg/openidentity"
|
||||
"cs-agent/internal/services"
|
||||
|
||||
"github.com/kataras/iris/v12"
|
||||
@@ -16,13 +15,13 @@ func ExternalUserMiddleware(ctx iris.Context) {
|
||||
_ = ctx.JSON(web.JsonErrorMsg("接入渠道异常"))
|
||||
return
|
||||
}
|
||||
secret := services.ChannelService.GetUserTokenSecret(channel)
|
||||
ext, err := openidentity.GetExternalUser(ctx, secret)
|
||||
result, err := services.CustomerSessionService.VerifyRequest(ctx, channel)
|
||||
if err != nil {
|
||||
ctx.StopExecution()
|
||||
_ = ctx.JSON(web.JsonError(err))
|
||||
return
|
||||
}
|
||||
irisx.SetExternalUser(ctx, ext)
|
||||
services.CustomerSessionService.SetRefreshHeaders(ctx, result)
|
||||
irisx.SetExternalUser(ctx, result.ExternalUser)
|
||||
ctx.Next()
|
||||
}
|
||||
|
||||
@@ -9,14 +9,15 @@ import (
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Server ServerConfig `yaml:"server"`
|
||||
DB DBConfig `yaml:"db"`
|
||||
Logger LoggerConfig `yaml:"logger"`
|
||||
Auth AuthConfig `yaml:"auth"`
|
||||
Storage StorageConfig `yaml:"storage"`
|
||||
VectorDB VectorDBConfig `yaml:"vectorDB"`
|
||||
MCP MCPConfig `yaml:"mcp"`
|
||||
WxWork WxWorkConfig `yaml:"wxWork"`
|
||||
Server ServerConfig `yaml:"server"`
|
||||
DB DBConfig `yaml:"db"`
|
||||
Logger LoggerConfig `yaml:"logger"`
|
||||
Auth AuthConfig `yaml:"auth"`
|
||||
Storage StorageConfig `yaml:"storage"`
|
||||
VectorDB VectorDBConfig `yaml:"vectorDB"`
|
||||
MCP MCPConfig `yaml:"mcp"`
|
||||
WxWork WxWorkConfig `yaml:"wxWork"`
|
||||
CustomerSession CustomerSessionConfig `yaml:"customerSession"`
|
||||
}
|
||||
|
||||
type WxWorkNotifyConfig struct {
|
||||
@@ -60,6 +61,26 @@ type AuthConfig struct {
|
||||
CredentialLockMinute int `yaml:"credentialLockMinute"`
|
||||
}
|
||||
|
||||
type CustomerSessionConfig struct {
|
||||
Secret string `yaml:"secret"`
|
||||
TTLMinutes int `yaml:"ttlMinutes"`
|
||||
RefreshThresholdMinutes int `yaml:"refreshThresholdMinutes"`
|
||||
}
|
||||
|
||||
func (c CustomerSessionConfig) TTL() int {
|
||||
if c.TTLMinutes <= 0 {
|
||||
return 120
|
||||
}
|
||||
return c.TTLMinutes
|
||||
}
|
||||
|
||||
func (c CustomerSessionConfig) RefreshThreshold() int {
|
||||
if c.RefreshThresholdMinutes <= 0 {
|
||||
return 30
|
||||
}
|
||||
return c.RefreshThresholdMinutes
|
||||
}
|
||||
|
||||
type StorageConfig struct {
|
||||
Default enums.AssetProvider `yaml:"default"`
|
||||
MaxUploadSizeMB int64 `yaml:"maxUploadSizeMB"`
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
package response
|
||||
|
||||
type CustomerSessionCustomerResponse struct {
|
||||
ID int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
|
||||
type CustomerSessionExchangeResponse struct {
|
||||
CustomerSessionToken string `json:"customerSessionToken"`
|
||||
ExpiresAt string `json:"expiresAt"`
|
||||
IdentityKey string `json:"identityKey"`
|
||||
Customer CustomerSessionCustomerResponse `json:"customer"`
|
||||
}
|
||||
@@ -251,6 +251,7 @@ const (
|
||||
IMRealtimeEventConversationClosed = "conversation.closed"
|
||||
IMRealtimeEventConversationRead = "conversation.read"
|
||||
IMRealtimeEventNotificationCreated = "notification.created"
|
||||
IMRealtimeEventCustomerSessionRefresh = "customer_session.refresh"
|
||||
)
|
||||
|
||||
const (
|
||||
|
||||
@@ -0,0 +1,251 @@
|
||||
package services
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"cs-agent/internal/models"
|
||||
"cs-agent/internal/pkg/config"
|
||||
"cs-agent/internal/pkg/dto/response"
|
||||
"cs-agent/internal/pkg/enums"
|
||||
"cs-agent/internal/pkg/errorsx"
|
||||
"cs-agent/internal/pkg/openidentity"
|
||||
"cs-agent/internal/repositories"
|
||||
|
||||
"github.com/golang-jwt/jwt/v5"
|
||||
"github.com/kataras/iris/v12"
|
||||
"github.com/mlogclub/simple/sqls"
|
||||
"github.com/mlogclub/simple/web/params"
|
||||
)
|
||||
|
||||
const (
|
||||
customerSessionTokenType = "customer_session"
|
||||
customerSessionHeader = "X-Customer-Session-Token"
|
||||
customerSessionExpHeader = "X-Customer-Session-Expires-At"
|
||||
)
|
||||
|
||||
var CustomerSessionService = newCustomerSessionService()
|
||||
|
||||
func newCustomerSessionService() *customerSessionService {
|
||||
return &customerSessionService{}
|
||||
}
|
||||
|
||||
type customerSessionService struct {
|
||||
}
|
||||
|
||||
type customerSessionClaims struct {
|
||||
TokenType string `json:"typ"`
|
||||
ChannelID int64 `json:"channelId"`
|
||||
ChannelCode string `json:"channelCode"`
|
||||
CustomerID int64 `json:"customerId"`
|
||||
CustomerName string `json:"customerName"`
|
||||
IdentityKey string `json:"identityKey"`
|
||||
jwt.RegisteredClaims
|
||||
}
|
||||
|
||||
type CustomerSessionVerifyResult struct {
|
||||
ExternalUser *openidentity.ExternalUser
|
||||
Token string
|
||||
ExpiresAt time.Time
|
||||
Refreshed bool
|
||||
}
|
||||
|
||||
func (s *customerSessionService) Exchange(channel *models.Channel, externalUser openidentity.ExternalUser) (*response.CustomerSessionExchangeResponse, error) {
|
||||
if channel == nil || channel.Status != enums.StatusOk {
|
||||
return nil, errorsx.InvalidParam("接入渠道不存在或已停用")
|
||||
}
|
||||
var customerID int64
|
||||
if err := sqls.WithTransaction(func(ctx *sqls.TxContext) error {
|
||||
id, err := CustomerService.EnsureExternalCustomer(ctx.Tx, externalUser)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
customerID = id
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
customer := CustomerService.Get(customerID)
|
||||
if customer == nil || customer.Status == enums.StatusDeleted {
|
||||
return nil, errorsx.InvalidParam("客户不存在")
|
||||
}
|
||||
token, expiresAt, err := s.Sign(channel, customer, externalUser)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &response.CustomerSessionExchangeResponse{
|
||||
CustomerSessionToken: token,
|
||||
ExpiresAt: expiresAt.Format(time.DateTime),
|
||||
IdentityKey: s.identityKey(externalUser),
|
||||
Customer: response.CustomerSessionCustomerResponse{
|
||||
ID: customer.ID,
|
||||
Name: strings.TrimSpace(customer.Name),
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *customerSessionService) Sign(channel *models.Channel, customer *models.Customer, externalUser openidentity.ExternalUser) (string, time.Time, error) {
|
||||
cfg := config.Current().CustomerSession
|
||||
secret := strings.TrimSpace(cfg.Secret)
|
||||
if secret == "" {
|
||||
return "", time.Time{}, errorsx.BusinessError(1, "客服会话密钥未配置")
|
||||
}
|
||||
if channel == nil || customer == nil {
|
||||
return "", time.Time{}, errorsx.InvalidParam("客服会话参数不完整")
|
||||
}
|
||||
now := time.Now()
|
||||
expiresAt := now.Add(time.Duration(cfg.TTL()) * time.Minute)
|
||||
claims := customerSessionClaims{
|
||||
TokenType: customerSessionTokenType,
|
||||
ChannelID: channel.ID,
|
||||
ChannelCode: strings.TrimSpace(channel.ChannelID),
|
||||
CustomerID: customer.ID,
|
||||
CustomerName: strings.TrimSpace(customer.Name),
|
||||
IdentityKey: s.identityKey(externalUser),
|
||||
RegisteredClaims: jwt.RegisteredClaims{
|
||||
IssuedAt: jwt.NewNumericDate(now),
|
||||
ExpiresAt: jwt.NewNumericDate(expiresAt),
|
||||
},
|
||||
}
|
||||
token, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString([]byte(secret))
|
||||
if err != nil {
|
||||
return "", time.Time{}, err
|
||||
}
|
||||
return token, expiresAt, nil
|
||||
}
|
||||
|
||||
func (s *customerSessionService) VerifyRequest(ctx iris.Context, channel *models.Channel) (*CustomerSessionVerifyResult, error) {
|
||||
token := s.getCustomerSessionToken(ctx)
|
||||
if token == "" {
|
||||
return nil, errorsx.Unauthorized("客服会话不能为空")
|
||||
}
|
||||
claims, err := s.verifyToken(token)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if channel == nil || channel.Status != enums.StatusOk {
|
||||
return nil, errorsx.InvalidParam("接入渠道不存在或已停用")
|
||||
}
|
||||
if claims.ChannelID != channel.ID || strings.TrimSpace(claims.ChannelCode) != strings.TrimSpace(channel.ChannelID) {
|
||||
return nil, errorsx.Unauthorized("客服会话校验失败")
|
||||
}
|
||||
customer := CustomerService.Get(claims.CustomerID)
|
||||
if customer == nil || customer.Status == enums.StatusDeleted {
|
||||
return nil, errorsx.Unauthorized("客服会话校验失败")
|
||||
}
|
||||
external, err := s.externalUserFromClaims(claims, customer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result := &CustomerSessionVerifyResult{
|
||||
ExternalUser: external,
|
||||
Token: token,
|
||||
ExpiresAt: claims.ExpiresAt.Time,
|
||||
}
|
||||
if s.shouldRefresh(claims.ExpiresAt.Time) {
|
||||
newToken, expiresAt, err := s.Sign(channel, customer, *external)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result.Token = newToken
|
||||
result.ExpiresAt = expiresAt
|
||||
result.Refreshed = true
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s *customerSessionService) SetRefreshHeaders(ctx iris.Context, result *CustomerSessionVerifyResult) {
|
||||
if ctx == nil || result == nil || !result.Refreshed {
|
||||
return
|
||||
}
|
||||
ctx.Header(customerSessionHeader, result.Token)
|
||||
ctx.Header(customerSessionExpHeader, result.ExpiresAt.Format(time.DateTime))
|
||||
}
|
||||
|
||||
func (s *customerSessionService) verifyToken(rawToken string) (*customerSessionClaims, error) {
|
||||
cfg := config.Current().CustomerSession
|
||||
secret := strings.TrimSpace(cfg.Secret)
|
||||
if secret == "" {
|
||||
return nil, errorsx.BusinessError(1, "客服会话密钥未配置")
|
||||
}
|
||||
claims := &customerSessionClaims{}
|
||||
token, err := jwt.ParseWithClaims(rawToken, claims, func(token *jwt.Token) (any, error) {
|
||||
if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
|
||||
return nil, errors.New("unsupported signing method")
|
||||
}
|
||||
return []byte(secret), nil
|
||||
}, jwt.WithExpirationRequired(), jwt.WithValidMethods([]string{
|
||||
jwt.SigningMethodHS256.Alg(),
|
||||
jwt.SigningMethodHS384.Alg(),
|
||||
jwt.SigningMethodHS512.Alg(),
|
||||
}))
|
||||
if err != nil {
|
||||
if errors.Is(err, jwt.ErrTokenExpired) {
|
||||
return nil, errorsx.Unauthorized("客服会话已过期")
|
||||
}
|
||||
return nil, errorsx.Unauthorized("客服会话校验失败")
|
||||
}
|
||||
if token == nil || !token.Valid || claims.TokenType != customerSessionTokenType || claims.ExpiresAt == nil {
|
||||
return nil, errorsx.Unauthorized("客服会话校验失败")
|
||||
}
|
||||
if claims.ChannelID <= 0 || strings.TrimSpace(claims.ChannelCode) == "" || claims.CustomerID <= 0 || strings.TrimSpace(claims.IdentityKey) == "" {
|
||||
return nil, errorsx.Unauthorized("客服会话校验失败")
|
||||
}
|
||||
return claims, nil
|
||||
}
|
||||
|
||||
func (s *customerSessionService) externalUserFromClaims(claims *customerSessionClaims, customer *models.Customer) (*openidentity.ExternalUser, error) {
|
||||
identityKey := strings.TrimSpace(claims.IdentityKey)
|
||||
parts := strings.SplitN(identityKey, ":", 2)
|
||||
if len(parts) != 2 || strings.TrimSpace(parts[1]) == "" {
|
||||
return nil, errorsx.Unauthorized("客服会话校验失败")
|
||||
}
|
||||
var source enums.ExternalSource
|
||||
switch parts[0] {
|
||||
case "user":
|
||||
source = enums.ExternalSourceUser
|
||||
case "guest":
|
||||
source = enums.ExternalSourceGuest
|
||||
default:
|
||||
return nil, errorsx.Unauthorized("客服会话校验失败")
|
||||
}
|
||||
identity := repositories.CustomerIdentityRepository.GetBy(sqls.DB(), source, parts[1])
|
||||
if identity == nil || identity.CustomerID != claims.CustomerID {
|
||||
return nil, errorsx.Unauthorized("客服会话校验失败")
|
||||
}
|
||||
name := strings.TrimSpace(claims.CustomerName)
|
||||
if customer != nil && strings.TrimSpace(customer.Name) != "" {
|
||||
name = strings.TrimSpace(customer.Name)
|
||||
}
|
||||
return &openidentity.ExternalUser{
|
||||
ExternalSource: source,
|
||||
ExternalID: parts[1],
|
||||
ExternalName: name,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *customerSessionService) shouldRefresh(expiresAt time.Time) bool {
|
||||
threshold := config.Current().CustomerSession.RefreshThreshold()
|
||||
return time.Until(expiresAt) <= time.Duration(threshold)*time.Minute
|
||||
}
|
||||
|
||||
func (s *customerSessionService) identityKey(externalUser openidentity.ExternalUser) string {
|
||||
switch externalUser.ExternalSource {
|
||||
case enums.ExternalSourceUser:
|
||||
return "user:" + strings.TrimSpace(externalUser.ExternalID)
|
||||
default:
|
||||
return "guest:" + strings.TrimSpace(externalUser.ExternalID)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *customerSessionService) getCustomerSessionToken(ctx iris.Context) string {
|
||||
auth := strings.TrimSpace(ctx.GetHeader("Authorization"))
|
||||
if len(auth) > 7 && strings.EqualFold(auth[:7], "Bearer ") {
|
||||
if token := strings.TrimSpace(auth[7:]); token != "" {
|
||||
return token
|
||||
}
|
||||
}
|
||||
token, _ := params.Get(ctx, "customerSessionToken")
|
||||
return strings.TrimSpace(token)
|
||||
}
|
||||
@@ -239,6 +239,25 @@ func (e RealtimeNotificationCreatedEvent) EventPayload() RealtimeEventPayload {
|
||||
return e.Payload
|
||||
}
|
||||
|
||||
type RealtimeCustomerSessionRefreshPayload struct {
|
||||
CustomerSessionToken string `json:"customerSessionToken"`
|
||||
ExpiresAt string `json:"expiresAt"`
|
||||
}
|
||||
|
||||
func (RealtimeCustomerSessionRefreshPayload) realtimeEventPayload() {}
|
||||
|
||||
type RealtimeCustomerSessionRefreshEvent struct {
|
||||
Payload RealtimeCustomerSessionRefreshPayload
|
||||
}
|
||||
|
||||
func (e RealtimeCustomerSessionRefreshEvent) EventType() string {
|
||||
return enums.IMRealtimeEventCustomerSessionRefresh
|
||||
}
|
||||
|
||||
func (e RealtimeCustomerSessionRefreshEvent) EventPayload() RealtimeEventPayload {
|
||||
return e.Payload
|
||||
}
|
||||
|
||||
type realtimeClientMessage struct {
|
||||
Type string `json:"type"`
|
||||
Topics []string `json:"topics,omitempty"`
|
||||
|
||||
@@ -75,25 +75,27 @@ func (s *wsService) HandleOpenWS(ctx iris.Context) {
|
||||
}
|
||||
|
||||
var (
|
||||
principal = AuthService.GetAuthPrincipal(ctx)
|
||||
external *openidentity.ExternalUser
|
||||
principal = AuthService.GetAuthPrincipal(ctx)
|
||||
external *openidentity.ExternalUser
|
||||
customerSessionInfo *CustomerSessionVerifyResult
|
||||
)
|
||||
if principal == nil {
|
||||
ext, err := openidentity.GetExternalUser(ctx, ChannelService.GetUserTokenSecret(channel))
|
||||
result, err := CustomerSessionService.VerifyRequest(ctx, channel)
|
||||
if err != nil {
|
||||
_ = ctx.StopWithJSON(iris.StatusUnauthorized, web.JsonError(err))
|
||||
return
|
||||
}
|
||||
external = ext
|
||||
external = result.ExternalUser
|
||||
customerSessionInfo = result
|
||||
}
|
||||
if err := s.upgradeConnection(ctx, principal, external, realtimeRoleUser); err != nil {
|
||||
if err := s.upgradeConnection(ctx, principal, external, realtimeRoleUser, customerSessionInfo); err != nil {
|
||||
slog.Error("upgrade open im websocket failed", "error", err, "path", ctx.Path(), "channelId", channel.ChannelID, "channel_id", channel.ID)
|
||||
ctx.StopExecution()
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (s *wsService) upgradeConnection(ctx iris.Context, principal *dto.AuthPrincipal, external *openidentity.ExternalUser, role string) error {
|
||||
func (s *wsService) upgradeConnection(ctx iris.Context, principal *dto.AuthPrincipal, external *openidentity.ExternalUser, role string, customerSessionInfo ...*CustomerSessionVerifyResult) error {
|
||||
conn, err := s.upgrader.Upgrade(ctx.ResponseWriter().Naive(), ctx.Request(), nil)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -151,6 +153,14 @@ func (s *wsService) upgradeConnection(ctx iris.Context, principal *dto.AuthPrinc
|
||||
Topics: session.topicList(),
|
||||
},
|
||||
}))
|
||||
if len(customerSessionInfo) > 0 && customerSessionInfo[0] != nil && customerSessionInfo[0].Refreshed {
|
||||
session.enqueueEvent(s.newEvent("", RealtimeCustomerSessionRefreshEvent{
|
||||
Payload: RealtimeCustomerSessionRefreshPayload{
|
||||
CustomerSessionToken: customerSessionInfo[0].Token,
|
||||
ExpiresAt: customerSessionInfo[0].ExpiresAt.Format(time.DateTime),
|
||||
},
|
||||
}))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user