models

package
v0.12.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Nov 1, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

README

models(数据层)

纯 struct 零依赖——全部域共享的数据词汇表

这个域干什么

models 放所有域共享的数据结构:客户端、消息、群组、连接记录、枚举、错误哨兵 它是依赖链的最底层——零外部依赖、零逻辑,只有字段与简单方法

核心概念

概念 说明
Client 服务端视角的连接客户端:身份、状态、元数据(并发安全的 JSON 快照 marshal)
HubMessage 消息主体:链式 Setter、Clone、trace 上下文注入(InjectContext)
ConnectionRecord / ConnectionQuality 连接落库记录 / 质量打分(LiveScore 加权)
枚举族 MessageType、Priority、VIPLevel、UserStatus、NodeStatus 等(带 FromInt 解析)
错误哨兵 ErrUserOffline ErrQueueFull 等 + IsRetryableError 分类判定
DeliveryGuarantee 消息级投递保证解析:type 表 / 显式覆盖 / 分类分兜底
DeliverResult / BroadcastResult 投递与广播的结果结构

怎么用

package main

import (
    "errors"

    "github.com/kamalyes/go-wsc/models"
)

msg := models.NewHubMessage().
    SetMessageType(models.MessageTypeText).
    SetContent("hello")

if err := models.ErrUserOffline; errors.Is(err, models.ErrUserOffline) {
    // 走离线路径
}

quality := models.ConnectionQuality{}
quality.UpdatePingStats(rtt)
score := quality.LiveScore()

端口与依赖

  • 零依赖(import 链终点);除标准库外不 import 任何包
  • pb 相关测试随 go-wsc-grpc-adapter 走,本包不含生成代码

性能与陷阱

  • HubMessage 的元数据 Map 并发读写有锁保护——高频字段尽量用具名列,别塞 Map
  • JSON marshal 走安全快照(已验证并发正确),新增字段必须进快照测试
  • 陷阱:往 models 加方法时先问「有没有逻辑」——有逻辑的去对应域

后端绑定

无纯数据结构,所有后端适配器共享本包词汇

Documentation

Overview

* @Author: kamalyes 501893067@qq.com * @Date: 2025-02-24 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-02-24 00:00:00 * @FilePath: \go-wsc\models\agent_workload.go * @Description: 客服负载模型 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-01-21 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-12-28 00:00:00 * @FilePath: \go-wsc\models\classification.go * @Description: 消息分类相关定义 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-01-02 12:20:22 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-10-26 13:58:33 * @FilePath: \go-wsc\models\client.go * @Description: * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-08-20 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-08-20 00:00:00 * @FilePath: \go-wsc\models\deliver_result.go * @Description: 统一投递结果类型 — 覆盖 P2P/群组/命名空间/全局广播所有场景 * * 由 Hub.Deliver(ctx, msg, excludeSender) 统一入口返回,替代历史 GroupSendResult / * SendResult / BroadcastResult 三套割裂的返回类型。调用方按 result.Mode + 计数字段判断结果。 * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-09-06 20:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-10-05 10:06:18 * @FilePath: \go-wsc\models\delivery_guarantee.go * @Description: 消息送达分级模型 —— 分级送达保证的地基 * * 三级送达保证(DeliveryGuarantee): * - GuaranteeGuaranteed 必达:控制/资金/安全类消息,控制通道直发 + ACK + 离线兜底,任何情况可达 * - GuaranteeStandard 普通:聊天/通知类消息,实时投递,队列满/过载时转离线补发(拒绝≠丢弃) * - GuaranteeEphemeral 高频流式:输入状态/已读/实时状态类消息,latest-wins 合并,只保证最新送达 * * 决策树(ResolveGuarantee):显式设置(WithGuarantee) > 消息类型默认表 > Classification 评分(≥90 必达) * > PriorityCritical 提级 > 默认普通 * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-02-10 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-02-10 00:00:00 * @FilePath: \go-wsc\models\device_type_mapper.go * @Description: 设备类型映射器 - 将 go-toolbox 的 DeviceType 映射到 go-wsc 的 ClientType * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-01-21 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-12-28 00:00:00 * @FilePath: \go-wsc\models\enums.go * @Description: 枚举类型定义 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-09-06 09:50:55 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-12-28 20:10:18 * @FilePath: \go-wsc\models\errors.go * @Description: WebSocket 通信错误定义 - 基于errorx.BaseError模式 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2026-01-13 09:57:05 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-01-13 10:17:55 * @FilePath: \go-wsc\models\event.go * @Description: * * Copyright (c) 2026 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-09-06 09:50:55 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-10-27 17:26:03 * @FilePath: \go-wsc\models\message.go * @Description: 消息处理逻辑 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-09-06 09:50:55 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2026-10-05 10:06:18 * @FilePath: \go-wsc\models\message_record.go * @Description: 消息发送记录管理 - 使用 GORM 数据库持久化 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-01-21 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-12-28 00:00:00 * @FilePath: \go-wsc\models\message_type.go * @Description: 消息类型和优先级定义 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-09-06 09:50:55 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-12-28 00:00:00 * @FilePath: \go-wsc\models\offline_message.go * @Description: 消息处理逻辑 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-01-21 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-12-28 17:15:25 * @FilePath: \go-wsc\models\templates.go * @Description: 模板相关定义 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

* @Author: kamalyes 501893067@qq.com * @Date: 2025-12-28 00:00:00 * @LastEditors: kamalyes 501893067@qq.com * @LastEditTime: 2025-12-28 00:00:00 * @FilePath: \go-wsc\models\validator.go * @Description: 枚举验证器集中管理 * * Copyright (c) 2025 by kamalyes, All Rights Reserved.

Index

Constants

View Source
const (
	DeviceTypeBot     = "bot"     // 机器人
	DeviceTypeTablet  = "tablet"  // 平板
	DeviceTypeMobile  = "mobile"  // 移动设备
	DeviceTypeDesktop = "desktop" // 桌面设备
	DeviceTypeUnknown = "unknown" // 未知设备
)

DeviceType 常量(来自 go-toolbox/pkg/useragent)

View Source
const (
	// 基础错误恢复类型 (80000-80099) - error_recovery.go中使用
	ErrorTypeConnection    errorx.ErrorType = 80000 // 连接错误
	ErrorTypeMessage       errorx.ErrorType = 80001 // 消息错误
	ErrorTypeSystem        errorx.ErrorType = 80002 // 系统错误
	ErrorTypeNetwork       errorx.ErrorType = 80003 // 网络错误
	ErrorTypeConcurrency   errorx.ErrorType = 80004 // 并发错误
	ErrorTypeMemory        errorx.ErrorType = 80005 // 内存错误
	ErrorTypeConfiguration errorx.ErrorType = 80006 // 配置错误

	// 连接相关错误 (80100-80199) - 可重试
	ErrTypeConnectionClosed   errorx.ErrorType = 80101 // 连接已关闭
	ErrTypeConnectionReset    errorx.ErrorType = 80102 // 连接重置
	ErrTypeConnectionTimeout  errorx.ErrorType = 80103 // 连接超时
	ErrTypeNetworkUnreachable errorx.ErrorType = 80104 // 网络不可达
	ErrTypeServiceUnavailable errorx.ErrorType = 80105 // 服务不可用

	// 队列和缓冲区错误 (80200-80299) - 可重试
	ErrTypeQueueFull           errorx.ErrorType = 80201 // 队列已满
	ErrTypeMessageBufferFull   errorx.ErrorType = 80202 // 消息缓冲区已满
	ErrTypePendingQueueFull    errorx.ErrorType = 80203 // 待处理队列已满
	ErrTypeQueueAndPendingFull errorx.ErrorType = 80204 // 队列和待处理队列均已满
	ErrTypeSendChannelFull     errorx.ErrorType = 80205 // 发送通道已满

	// 用户和认证错误 (80300-80399) - 不可重试
	ErrTypeUserOffline          errorx.ErrorType = 80301 // 用户离线
	ErrTypeUserNotFound         errorx.ErrorType = 80302 // 用户未找到
	ErrTypePermissionDenied     errorx.ErrorType = 80303 // 权限被拒绝
	ErrTypeAuthenticationFailed errorx.ErrorType = 80304 // 认证失败
	ErrTypeUnauthorized         errorx.ErrorType = 80305 // 未经授权的访问

	// 消息错误 (80400-80499) - 不可重试
	ErrTypeInvalidMessageFormat   errorx.ErrorType = 80401 // 无效的消息格式
	ErrTypeMessageTooLarge        errorx.ErrorType = 80402 // 消息过大
	ErrTypeMessageTargetMissing   errorx.ErrorType = 80403 // 消息目标未指定
	ErrTypeMessageFiltered        errorx.ErrorType = 80404 // 消息被规则过滤
	ErrTypeMessageDeliveryTimeout errorx.ErrorType = 80405 // 消息投递超时

	// 客户端错误 (80500-80599) - 不可重试
	ErrTypeClientNotFound     errorx.ErrorType = 80501 // 客户端未找到
	ErrTypeClientDisconnected errorx.ErrorType = 80502 // 客户端已断开连接
	ErrTypeNoAvailableAgents  errorx.ErrorType = 80503 // 没有可用的代理

	// 集线器操作错误 (80600-80699) - 混合可重试性
	ErrTypeHubStartupTimeout  errorx.ErrorType = 80601 // 集线器启动超时 - 可重试
	ErrTypeHubShutdownTimeout errorx.ErrorType = 80602 // 集线器关闭超时 - 可重试
	ErrTypeHubNotRunning      errorx.ErrorType = 80603 // 集线器未运行 - 不可重试
	ErrTypeCircuitBreakerOpen errorx.ErrorType = 80604 // 电路断路器已打开 - 可重试

	// 记录管理错误 (80700-80799) - 不可重试
	ErrTypeRecordManagerDisabled        errorx.ErrorType = 80701 // 记录管理器已禁用
	ErrTypeMessageRecordNotFound        errorx.ErrorType = 80702 // 消息记录未找到
	ErrTypeMessageAlreadySent           errorx.ErrorType = 80703 // 消息已成功发送
	ErrTypeMaxRetriesExceeded           errorx.ErrorType = 80704 // 超过最大重试次数
	ErrTypeRecordManagerNotInitialized  errorx.ErrorType = 80705 // 记录管理器未初始化
	ErrTypeMaxRetriesExceededForMessage errorx.ErrorType = 80706 // 消息重试次数超过最大限制
	ErrTypeRecordRepositoryNotSet       errorx.ErrorType = 80707 // 消息记录仓库未设置
	ErrTypeOnlineStatusRepositoryNotSet errorx.ErrorType = 80708 // 在线状态仓库未设置
	ErrTypeStatsRepositoryNotSet        errorx.ErrorType = 80709 // 统计仓库未设置

	// 速率限制错误 (80800-80899) - 不可重试
	ErrTypeRateLimitExceeded     errorx.ErrorType = 80801 // 超过速率限制
	ErrTypeFrequencyLimitReached errorx.ErrorType = 80802 // 达到频率限制

	// 操作错误 (80900-80999) - 可重试
	ErrTypeOperationTimeout errorx.ErrorType = 80901 // 操作超时
	ErrTypeTemporaryFailure errorx.ErrorType = 80902 // 临时故障
	ErrTypeResourceBusy     errorx.ErrorType = 80903 // 资源繁忙
	ErrTypeUnknownError     errorx.ErrorType = 80999 // 未知错误

	// ACK相关错误 (81000-81099) - 混合可重试性
	ErrTypeAckTimeout        errorx.ErrorType = 81001 // ACK超时 - 可重试
	ErrTypeAckTimeoutRetries errorx.ErrorType = 81002 // ACK经重试后超时 - 不可重试
	ErrTypeContextCancelled  errorx.ErrorType = 81003 // 上下文取消 - 不可重试

	// 配置相关错误 (81100-81199) - 不可重试
	ErrTypeConfigValidatorNotInitialized errorx.ErrorType = 81101 // 配置验证器未初始化
	ErrTypeConfigValidationFailed        errorx.ErrorType = 81102 // 配置验证失败
	ErrTypeConfigAutoFixFailed           errorx.ErrorType = 81103 // 配置自动修复失败

	// 安全相关错误 (81200-81299) - 不可重试
	ErrTypeIPInBlacklist      errorx.ErrorType = 81201 // IP在黑名单中
	ErrTypeBruteForceDetected errorx.ErrorType = 81202 // 检测到暴力攻击
	ErrTypeThreatDetected     errorx.ErrorType = 81203 // 检测到威胁内容
	ErrTypeAccessDeniedByRule errorx.ErrorType = 81204 // 被访问规则拒绝

	// PubSub相关错误 (81300-81399) - 不可重试
	ErrTypePubSubNotSet           errorx.ErrorType = 81301 // PubSub未设置
	ErrTypePubSubPublishFailed    errorx.ErrorType = 81302 // 事件发布失败
	ErrTypeEventSerializeFailed   errorx.ErrorType = 81303 // 事件序列化失败
	ErrTypeEventDeserializeFailed errorx.ErrorType = 81304 // 事件反序列化失败

	// 群组相关错误 (81400-81499) - 不可重试
	ErrTypeGroupNotFound         errorx.ErrorType = 81401 // 群组未找到
	ErrTypeGroupMemberExisted    errorx.ErrorType = 81402 // 用户已是群组成员
	ErrTypeGroupFull             errorx.ErrorType = 81403 // 群组已满
	ErrTypeGroupRepoNotSet       errorx.ErrorType = 81404 // 群组仓库未设置
	ErrTypeGroupExisted          errorx.ErrorType = 81405 // 群组已存在(同命名空间下 groupID 唯一)
	ErrTypeGroupReserved         errorx.ErrorType = 81406 // 群组名为系统保留名(__ 前缀)
	ErrTypeRouteNamespaceMissing errorx.ErrorType = 81407 // 路由信封缺少命名空间(群组投递必要参数)
)

WebSocket 通信错误码常量定义 使用 8xxxx 区间,避免与其他包冲突(WSC = WebSocket Client)

View Source
const (
	// ErrMsgClientIDFormat 客户端ID错误消息格式
	ErrMsgClientIDFormat = "client_id: %s"

	// ErrMsgDecompressFailed 解压消息失败错误格式
	ErrMsgDecompressFailed = "decompress message failed: %w"
)
View Source
const (
	// EventUserOnline 用户上线事件(包括客服、客户、访客等所有用户)
	EventUserOnline = "user.online"
	// EventUserOffline 用户下线事件
	EventUserOffline = "user.offline"

	// EventTicketQueuePushed 工单入队事件(由外部系统发布)
	EventTicketQueuePushed = "ticket.queue.pushed"

	// EventTicketAssigned 工单分配事件(工单已分配给客服)
	EventTicketAssigned = "ticket.assigned"
	// EventTicketAssignmentFailed 工单分配失败事件
	EventTicketAssignmentFailed = "ticket.assignment.failed"
)

事件类型常量

View Source
const (
	DataKeyContentExtra   = "content_extra"  // 扩展内容
	DataKeyMetadata       = "metadata"       // 元数据
	DataKeyMediaInfo      = "media_info"     // 媒体信息
	DataKeyClassification = "classification" // 消息综合分类(ResolveGuarantee 消费其评分)
)

Data 字段的常量 key

View Source
const (
	QueryMessageIDWhere = "message_id = ?"
	// QueryMessageIDReceiverWhere 复合键定位(P2P 同一 message_id 多 receiver 记录互不干扰)
	QueryMessageIDReceiverWhere = "message_id = ? AND receiver = ?"
	OrderByCreateTimeDesc       = "create_time DESC"
	OrderByCreateTimeAsc        = "create_time ASC"
	OrderByExpiresAtAsc         = "expires_at ASC"
)

数据库查询常量

View Source
const ClassificationGuaranteedScore = 90

ClassificationGuaranteedScore 分类评分必达阈值 GetFinalPriority() 评分 ≥ 90(IsCriticalMessage 判定)的消息提升为必达级

Variables

View Source
var (
	ErrConnectionClosed       errorx.BaseError
	ErrMessageBufferFull      errorx.BaseError
	ErrHubStartupTimeout      errorx.BaseError
	ErrHubShutdownTimeout     errorx.BaseError
	ErrQueueAndPendingFull    errorx.BaseError
	ErrMessageTargetMissing   errorx.BaseError
	ErrUserOffline            errorx.BaseError
	ErrMessageDeliveryTimeout errorx.BaseError
	ErrCircuitBreakerOpen     errorx.BaseError
)

连接相关错误变量(在 init() 末尾统一赋值,详见 init() 注释)

View Source
var (
	ErrAckTimeout        errorx.BaseError
	ErrAckTimeoutRetries errorx.BaseError
	ErrContextCancelled  errorx.BaseError
)

ACK相关错误变量

View Source
var (
	ErrRecordManagerNotInitialized errorx.BaseError
	ErrMaxRetriesExceeded          errorx.BaseError
)

记录管理相关错误变量

View Source
var (
	ErrMessageFiltered              errorx.BaseError
	ErrNoAvailableAgents            errorx.BaseError
	ErrQueueFull                    errorx.BaseError
	ErrRecordRepositoryNotSet       errorx.BaseError
	ErrOnlineStatusRepositoryNotSet errorx.BaseError
	ErrStatsRepositoryNotSet        errorx.BaseError
	ErrSendChannelFull              errorx.BaseError
	ErrUserNotFound                 errorx.BaseError
	ErrClientNotFound               errorx.BaseError
	ErrClientDisconnected           errorx.BaseError
)

业务逻辑错误变量

View Source
var (
	ErrPubSubNotSet           errorx.BaseError
	ErrPubSubPublishFailed    errorx.BaseError
	ErrEventSerializeFailed   errorx.BaseError
	ErrEventDeserializeFailed errorx.BaseError
)

PubSub相关错误变量

View Source
var (
	ErrGroupNotFound         errorx.BaseError
	ErrGroupMemberExisted    errorx.BaseError
	ErrGroupFull             errorx.BaseError
	ErrGroupRepoNotSet       errorx.BaseError
	ErrGroupExisted          errorx.BaseError
	ErrGroupReserved         errorx.BaseError
	ErrRouteNamespaceMissing errorx.BaseError
)

群组相关错误变量

View Source
var (
	// UserRoleValidator 用户角色验证器
	UserRoleValidator = validator.NewEnumValidator(
		UserRoleCustomer,
		UserRoleAgent,
		UserRoleAdmin,
	)

	// UserTypeValidator 用户类型验证器
	UserTypeValidator = validator.NewEnumValidator(
		UserTypeVisitor,
		UserTypeCustomer,
		UserTypeAgent,
		UserTypeAdmin,
		UserTypeBot,
		UserTypeVIP,
		UserTypeSystem,
		UserTypeObserver,
	)

	// UserStatusValidator 用户状态验证器
	UserStatusValidator = validator.NewEnumValidator(
		UserStatusOnline,
		UserStatusOffline,
		UserStatusBusy,
		UserStatusAway,
		UserStatusInvisible,
	)

	// DisconnectReasonValidator 断开原因验证器
	DisconnectReasonValidator = validator.NewEnumValidator(
		DisconnectReasonReadError,
		DisconnectReasonWriteError,
		DisconnectReasonContextDone,
		DisconnectReasonCloseMessage,
		DisconnectReasonHeartbeatFail,
		DisconnectReasonKickOut,
		DisconnectReasonForceOffline,
		DisconnectReasonTimeout,
		DisconnectReasonClientRequest,
		DisconnectReasonServerShutdown,
		DisconnectReasonUnknown,
	)

	// ErrorSeverityValidator 错误严重程度验证器
	ErrorSeverityValidator = validator.NewEnumValidator(
		ErrorSeverityInfo,
		ErrorSeverityWarning,
		ErrorSeverityError,
		ErrorSeverityCritical,
		ErrorSeverityFatal,
	)

	// QueueTypeValidator 队列类型验证器
	QueueTypeValidator = validator.NewEnumValidator(
		QueueTypeBroadcast,
		QueueTypePending,
		QueueTypeAllQueues,
		QueueTypeMessageQueue,
		QueueTypeClientBuffer,
	)

	// MessageTypeValidator 消息类型验证器
	MessageTypeValidator = validator.NewEnumValidator(
		MessageTypeText, MessageTypeImage, MessageTypeFile, MessageTypeAudio, MessageTypeVideo,
		MessageTypeSystem, MessageTypeNotice, MessageTypeEvent, MessageTypeAck, MessageTypeLocation,
		MessageTypeCard, MessageTypeEmoji, MessageTypeSticker, MessageTypeLink, MessageTypeQuote,
		MessageTypeForward, MessageTypeCommand, MessageTypeMarkdown, MessageTypeRichText, MessageTypeCode,
		MessageTypeJSON, MessageTypeXML, MessageTypeBinary, MessageTypeVoice, MessageTypeGIF,
		MessageTypeDocument, MessageTypeSpreadsheet, MessageTypePresentation, MessageTypeContact,
		MessageTypeCalendar, MessageTypeTask, MessageTypePoll, MessageTypeForm, MessageTypePayment,
		MessageTypeOrder, MessageTypeProduct, MessageTypeInvite, MessageTypeAnnouncement, MessageTypeAlert,
		MessageTypeError, MessageTypeInfo, MessageTypeSuccess, MessageTypeWarning, MessageTypeHeartbeat,
		MessageTypePing, MessageTypePong, MessageTypeTyping, MessageTypeRead, MessageTypeDelivered,
		MessageTypeRecall, MessageTypeEdit, MessageTypeReaction, MessageTypeThread, MessageTypeReply,
		MessageTypeMention, MessageTypeCustom, MessageTypeUnknown, MessageTypeTicketAssigned,
		MessageTypeTicketClosed, MessageTypeTicketTransfer, MessageTypeTicketActive, MessageTypeTest,
		MessageTypeWelcome, MessageTypeTerminate, MessageTypeTransferred, MessageTypeSessionCreated,
		MessageTypeSessionClosed, MessageTypeSessionQueued, MessageTypeSessionTimeout, MessageTypeSessionPaused,
		MessageTypeSessionResumed, MessageTypeSessionTransferred, MessageTypeSessionMemberJoined,
		MessageTypeSessionMemberLeft, MessageTypeSessionStatusChanged, MessageTypeCheckUserStatus,
		MessageTypeUserStatusResponse, MessageTypeGetOnlineUsers, MessageTypeOnlineUsersList,
		MessageTypeGetUserInfo, MessageTypeUserInfoResponse, MessageTypeSystemQuery, MessageTypeSystemResponse,
		MessageTypeUserJoined, MessageTypeUserLeft, MessageTypeUserStatusChanged, MessageTypeServerStatus,
		MessageTypeServerStats, MessageTypeClientConfig, MessageTypeConfigUpdate, MessageTypeHealthCheck,
		MessageTypeHealthResponse, MessageTypeConnected, MessageTypeDisconnected, MessageTypeReconnected,
		MessageTypeConnectionError, MessageTypeConnectionTimeout, MessageTypeKickOut, MessageTypeForceOffline,
		MessageTypeGroupChanged,
	)

	// PushTypeValidator 推送类型验证器
	PushTypeValidator = validator.NewEnumValidator(
		PushTypeDirect,
		PushTypeQueue,
		PushTypeOffline,
		PushTypeNone,
	)

	// BroadcastTypeValidator 广播类型验证器
	BroadcastTypeValidator = validator.NewEnumValidator(
		BroadcastTypeNone,
		BroadcastTypeSession,
		BroadcastTypeGlobal,
	)

	// VIPLevelValidator VIP等级验证器
	VIPLevelValidator = validator.NewEnumValidator(
		VIPLevelV0, VIPLevelV1, VIPLevelV2, VIPLevelV3,
		VIPLevelV4, VIPLevelV5, VIPLevelV6, VIPLevelV7, VIPLevelV8,
	)

	// UrgencyLevelValidator 紧急等级验证器
	UrgencyLevelValidator = validator.NewEnumValidator(
		UrgencyLevelLow,
		UrgencyLevelNormal,
		UrgencyLevelHigh,
	)

	// BusinessCategoryValidator 业务分类验证器
	BusinessCategoryValidator = validator.NewEnumValidator(
		BusinessCategoryGeneral,
		BusinessCategoryCustomer,
		BusinessCategorySales,
		BusinessCategoryTechnical,
		BusinessCategoryFinance,
		BusinessCategorySecurity,
		BusinessCategoryOperations,
		BusinessCategorySupport,
		BusinessCategoryIT,
		BusinessCategoryQuality,
		BusinessCategoryOther,
	)

	// MessageStatusValidator 消息状态验证器
	MessageStatusValidator = validator.NewEnumValidator(
		MessageStatusPending,
		MessageStatusSent,
		MessageStatusDelivered,
		MessageStatusRead,
		MessageStatusFailed,
	)

	// NodeStatusValidator 节点状态验证器
	NodeStatusValidator = validator.NewEnumValidator(
		NodeStatusActive,
		NodeStatusInactive,
		NodeStatusOffline,
	)

	// ConnectionStatusValidator 连接状态验证器
	ConnectionStatusValidator = validator.NewEnumValidator(
		ConnectionStatusConnecting,
		ConnectionStatusConnected,
		ConnectionStatusDisconnected,
		ConnectionStatusReconnecting,
		ConnectionStatusError,
	)

	// OperationTypeValidator 操作类型验证器
	OperationTypeValidator = validator.NewEnumValidator(
		OperationTypeJoin,
		OperationTypeLeave,
		OperationTypeMessage,
		OperationTypeBroadcast,
		OperationTypeNotify,
		OperationTypeHeartbeat,
		OperationTypeAuth,
		OperationTypeSync,
		OperationTypeSendMessage,
		OperationTypeKickUser,
		OperationTypeNodeRegister,
		OperationTypeObserverNotify,
		OperationTypeGroupBroadcast,
		OperationTypeGroupsBroadcast,
		OperationTypeUserNotFound,
		OperationTypeClientReclaim,
		OperationTypeGroupInvalidate,
	)

	// ClientTypeValidator 客户端类型验证器
	ClientTypeValidator = validator.NewEnumValidator(
		ClientTypeWeb,
		ClientTypeMobile,
		ClientTypeDesktop,
		ClientTypeAPI,
	)

	// PriorityValidator 优先级验证器
	PriorityValidator = validator.NewEnumValidator(
		PriorityLow,
		PriorityNormal,
		PriorityHigh,
		PriorityUrgent,
		PriorityCritical,
	)

	// DepartmentValidator 部门验证器
	DepartmentValidator = validator.NewEnumValidator(
		DepartmentSales,
		DepartmentSupport,
		DepartmentBilling,
		DepartmentGeneral,
		DepartmentTechnical,
	)

	// SkillValidator 技能验证器
	SkillValidator = validator.NewEnumValidator(
		SkillTechnical,
		SkillSales,
		SkillBilling,
		SkillGeneral,
		SkillLanguageEN,
		SkillLanguageZH,
		SkillVIP,
	)

	// MessageSendStatusValidator 消息发送状态验证器
	MessageSendStatusValidator = validator.NewEnumValidator(
		MessageSendStatusPending,
		MessageSendStatusSending,
		MessageSendStatusSuccess,
		MessageSendStatusFailed,
		MessageSendStatusRetrying,
		MessageSendStatusAckTimeout,
		MessageSendStatusUserOffline,
		MessageSendStatusExpired,
	)

	// MessageSourceValidator 消息来源验证器
	MessageSourceValidator = validator.NewEnumValidator(
		MessageSourceOnline,
		MessageSourceOffline,
	)

	// PendingOfflineStatuses 待推送的离线消息状态列表
	// 用于查询未完成推送的离线消息
	PendingOfflineStatuses = []MessageSendStatus{
		MessageSendStatusUserOffline,
		MessageSendStatusPending,
		MessageSendStatusFailed,
	}
)

全局枚举验证器实例

AllWorkloadDimensions 所有维度

View Source
var (
	ErrConfigValidatorNotInitialized errorx.BaseError
)

配置相关错误变量

View Source
var MessageTypeEmojiMap = map[MessageType]string{

	MessageTypeOpenWindow:  "🟢",
	MessageTypeCloseWindow: "🔴",

	MessageTypeTyping:          "⌨️",
	MessageTypeRead:            "👁️",
	MessageTypeDelivered:       "✅",
	MessageTypeAck:             "✔️",
	MessageTypeReaction:        "❤️",
	MessageTypeEdit:            "✏️",
	MessageTypeRecall:          "↩️",
	MessageTypeBStatusReminder: "⏰",
}

MessageTypeEmojiMap 消息类型对应的日志 emoji 映射表

View Source
var WsCloseCodeMap = map[int]struct {
	IsNormal bool   // 是否正常关闭
	Desc     string // 描述
}{

	websocket.CloseNormalClosure: {IsNormal: true, Desc: "正常关闭"},
	websocket.CloseGoingAway:     {IsNormal: true, Desc: "客户端离开(关闭标签页/浏览器)"},

	websocket.CloseProtocolError:           {IsNormal: false, Desc: "协议错误"},
	websocket.CloseUnsupportedData:         {IsNormal: false, Desc: "不支持的数据类型"},
	websocket.CloseNoStatusReceived:        {IsNormal: false, Desc: "未收到状态码"},
	websocket.CloseInvalidFramePayloadData: {IsNormal: false, Desc: "无效的帧数据"},

	websocket.ClosePolicyViolation:    {IsNormal: false, Desc: "策略违规"},
	websocket.CloseMessageTooBig:      {IsNormal: false, Desc: "消息过大"},
	websocket.CloseMandatoryExtension: {IsNormal: false, Desc: "强制扩展未协商"},

	websocket.CloseInternalServerErr: {IsNormal: false, Desc: "服务器内部错误"},
	websocket.CloseServiceRestart:    {IsNormal: false, Desc: "服务重启"},
	websocket.CloseTryAgainLater:     {IsNormal: false, Desc: "稍后重试"},

	websocket.CloseAbnormalClosure: {IsNormal: false, Desc: "异常关闭(网络中断/连接丢失)"},
	websocket.CloseTLSHandshake:    {IsNormal: false, Desc: "TLS握手失败"},
}

WsCloseCodeMap WebSocket 关闭码映射表 (RFC 6455, section 11.7)

Functions

func IsAckTimeoutError

func IsAckTimeoutError(err error) bool

IsAckTimeoutError 判断是否为ACK超时错误

func IsQueueFullError

func IsQueueFullError(err error) bool

IsQueueFullError 判断是否为队列满错误

func IsRetryableError

func IsRetryableError(err error) bool

IsRetryableError 判断错误是否可以重试

func IsRetryableErrorType

func IsRetryableErrorType(errType errorx.ErrorType) bool

IsRetryableErrorType 判断错误类型是否可以重试

func IsSendTimeoutError

func IsSendTimeoutError(err error) bool

IsSendTimeoutError 判断是否为发送超时错误

func IsSystemGroup

func IsSystemGroup(groupID string) bool

IsSystemGroup 判断 groupID 是否为系统保留组(`__` 前缀)

func IsUserOfflineError

func IsUserOfflineError(err error) bool

IsUserOfflineError 判断是否为用户离线错误

func MapClientTypeToDeviceType

func MapClientTypeToDeviceType(clientType ClientType) string

MapClientTypeToDeviceType 将 go-wsc 的 ClientType 反向映射到 go-toolbox 的 DeviceType

映射规则:

  • api -> bot
  • mobile -> mobile
  • desktop -> desktop
  • web -> desktop

Types

type AgentWorkloadModel

type AgentWorkloadModel struct {
	ID                uint              `gorm:"primaryKey;autoIncrement;comment:自增主键" json:"id"`
	Namespace         string            `` /* 166-byte string literal not displayed */
	AgentID           string            `gorm:"column:agent_id;size:255;not null;index:idx_ns_agent_dimension_time;comment:客服ID" json:"agent_id"`
	WorkloadDimension WorkloadDimension `` /* 131-byte string literal not displayed */
	TimeKey           string            `` /* 154-byte string literal not displayed */
	Workload          int64             `gorm:"column:workload;not null;default:0;index;comment:当前负载" json:"workload"`
	CreatedAt         time.Time         `gorm:"column:created_at;not null;comment:创建时间" json:"created_at"`
	UpdatedAt         time.Time         `gorm:"column:updated_at;not null;comment:更新时间" json:"updated_at"`
}

AgentWorkloadModel 客服负载模型(DB 持久化,支持多维度+多命名空间隔离) 用于存储客服的工作负载信息,支持负载均衡和故障恢复 通过 dimension 和 time_key 字段支持多时间维度统计 namespace 字段实现多租户隔离,不同命名空间下同名 agent_id 互不干扰

func (AgentWorkloadModel) TableComment

func (AgentWorkloadModel) TableComment() string

TableComment 表注释

func (AgentWorkloadModel) TableName

func (AgentWorkloadModel) TableName() string

TableName 表名

type BroadcastResult

type BroadcastResult struct {
	Total      int              // 总用户数量
	Success    int              // 成功发送数量
	Offline    int              // 离线用户数量
	Failed     int              // 发送失败数量
	Errors     map[string]error // 错误详情 map[userID]error
	OfflineIDs []string         // 离线用户ID列表
	FailedIDs  []string         // 发送失败的用户ID列表
}

BroadcastResult 广播发送结果

type BroadcastType

type BroadcastType string

BroadcastType 广播类型

const (
	BroadcastTypeNone    BroadcastType = "none"    // 不广播(单播)
	BroadcastTypeSession BroadcastType = "session" // 会话成员广播
	BroadcastTypeGlobal  BroadcastType = "global"  // 全站广播
)

func (BroadcastType) IsValid

func (b BroadcastType) IsValid() bool

IsValid 检查广播类型是否有效

func (BroadcastType) String

func (b BroadcastType) String() string

String 实现Stringer接口

type BusinessCategory

type BusinessCategory string

BusinessCategory 业务分类

const (
	// 基本分类
	BusinessCategoryGeneral    BusinessCategory = "general"    // 通用
	BusinessCategoryCustomer   BusinessCategory = "customer"   // 客户服务
	BusinessCategorySales      BusinessCategory = "sales"      // 销售
	BusinessCategoryTechnical  BusinessCategory = "technical"  // 技术
	BusinessCategoryFinance    BusinessCategory = "finance"    // 财务
	BusinessCategorySecurity   BusinessCategory = "security"   // 安全
	BusinessCategoryOperations BusinessCategory = "operations" // 运营
	BusinessCategorySupport    BusinessCategory = "support"    // 支持
	BusinessCategoryIT         BusinessCategory = "it"         // 信息技术
	BusinessCategoryQuality    BusinessCategory = "quality"    // 质量
	BusinessCategoryOther      BusinessCategory = "other"      // 其他
)

func GetAllBusinessCategories

func GetAllBusinessCategories() []BusinessCategory

GetAllBusinessCategories 获取所有业务分类

func (BusinessCategory) GetCategoryType

func (b BusinessCategory) GetCategoryType() string

GetCategoryType 获取业务分类类型

func (BusinessCategory) IsValid

func (b BusinessCategory) IsValid() bool

IsValid 检查业务分类是否有效

func (BusinessCategory) String

func (b BusinessCategory) String() string

String 实现Stringer接口

type Client

type Client struct {
	ID             string          `json:"id"`                  // 客户端ID
	UserID         string          `json:"user_id"`             // 用户ID
	AppID          string          `json:"app_id,omitempty"`    // 应用ID(最上层隔离维度,入口层归一化为 DefaultAppID,用于应用间消息隔离)
	Namespace      string          `json:"namespace"`           // 命名空间ID(默认 "default",类似 k8s namespace,用于观察者过滤与命名空间隔离)
	GroupID        string          `json:"group_id,omitempty"`  // 默认群组ID(普通用户=连接后自动加入;观察者=订阅的观察群组,空=观察整个命名空间)
	GroupIDs       []string        `json:"group_ids,omitempty"` // 连接时自动加入的全部群组列表(管理端一人多组场景);GetGroupIDs 回退链:GroupIDs→[GroupID]→[DefaultGroupID]
	UserType       UserType        `json:"user_type"`           // 用户类型
	VIPLevel       VIPLevel        `json:"vip_level"`           // VIP等级
	Role           UserRole        `json:"role"`                // 用户角色
	ClientIP       string          `json:"client_ip"`           // 客户端IP
	Conn           *websocket.Conn `json:"-"`                   // WebSocket连接(不序列化,仅WS使用)
	ConnectedAt    time.Time       `json:"connected_at"`        // 连接时间
	Status         UserStatus      `json:"status"`              // 用户状态
	Department     Department      `json:"department"`          // 部门
	Skills         []Skill         `json:"skills"`              // 技能列表
	MaxTickets     int             `json:"max_tickets"`         // 最大工单数
	NodeID         string          `json:"node_id"`             // 所在节点ID
	NodeIP         string          `json:"node_ip"`             // 所在节点IP
	NodePort       int             `json:"node_port"`           // 所在节点端口
	ClientType     ClientType      `json:"client_type"`         // 客户端类型(web/mobile/desktop)
	ConnectionType ConnectionType  `json:"connection_type"`     // 连接类型(websocket/sse)
	// Metadata 惰性分配:NewClient 不预建 map(省 ~300B/conn),首次写入经
	// SetMetadataValue/WithMetadataMap 惰性创建;序列化边界输出空对象保持 JSON 形态不变
	Metadata map[string]interface{} `json:"metadata"` // 元数据
	SendChan chan []byte            `json:"-"`        // 发送通道(不序列化,仅WS使用)
	CtrlCh   chan []byte            `json:"-"`        // 控制通道(必达级:KickOut/ForceOffline/Ack 等;写泵 select 前优先排空,容量见 constants.CtrlChanCapacity)
	PongCh   chan []byte            `json:"-"`        // pong 控制帧队列(不序列化,cap 1;读协程收到协议级 PING 后非阻塞投递,写泵统一写出——gorilla 单写者模式)
	DoneCh   chan struct{}          `json:"-"`        // 生命周期关闭信号(注销时 close,写泵 select 退出;数据通道永不 close,消除 send/close 竞态)
	Context  context.Context        `json:"-"`        // 上下文(不序列化)

	CloseMu sync.Mutex `json:"-"` // 保护channel关闭的互斥锁(不序列化)

	// SSE 专用字段(仅当 ConnectionType 为 SSE 时使用)
	SSEWriter    http.ResponseWriter `json:"-"` // SSE Writer(不序列化)
	SSEFlusher   http.Flusher        `json:"-"` // SSE Flusher(不序列化)
	SSEMessageCh chan *HubMessage    `json:"-"` // SSE 消息通道(不序列化)
	SSECloseCh   chan struct{}       `json:"-"` // SSE 关闭通道(不序列化)
	// contains filtered or unexported fields
}

Client 客户端连接(统一管理 WebSocket 和 SSE 连接)

func NewClient

func NewClient(id, userID string, userType UserType) *Client

NewClient 创建新的客户端实例

func (*Client) BacklogRatio

func (c *Client) BacklogRatio() float64

BacklogRatio 读取 SendChan 利用率(0.0-1.0,无锁;扫描器/治理逻辑用)

func (*Client) CachedPongBytes

func (c *Client) CachedPongBytes(build func() []byte) []byte

CachedPongBytes 返回应用层心跳 pong 的预序列化缓存,缓存缺失时经 build 构建一次并缓存 pong 内容 per-connection 恒定(类型 + 接收者 + 连接级唯一 ID,幂等保活响应无逐次唯一性需求), 构建后心跳热路径零分配零编码;持 mu 与 MarshalJSON 序列化路径互斥,build 返回空切片时不缓存

func (*Client) CompletePendingWrites

func (c *Client) CompletePendingWrites(n int64)

CompletePendingWrites 结算已写出、写出失败或断连放弃的数据帧,避免重复扣减

func (*Client) GetAppID

func (c *Client) GetAppID() string

GetAppID 获取应用ID(空值归一化为 DefaultAppID,与 GetNamespace 语义一致) 入口层已统一归一化,下游过滤层无需处理空值

func (*Client) GetClientIP

func (c *Client) GetClientIP() string

GetClientIP 获取客户端IP地址

func (*Client) GetGroupID

func (c *Client) GetGroupID() string

GetGroupID 获取群组ID,空值返回默认群组ID

func (*Client) GetGroupIDRaw

func (c *Client) GetGroupIDRaw() string

GetGroupIDRaw 并发安全获取原始群组ID(不加默认值,用于索引判断)

func (*Client) GetGroupIDs

func (c *Client) GetGroupIDs() []string

GetGroupIDs 获取群组ID列表,空值回退到 [GroupID]→[DefaultGroupID] 保证 joinMemberGroupOnConnect 空值时仍加入默认组(保留单值时代行为)

func (*Client) GetGroupIDsRaw

func (c *Client) GetGroupIDsRaw() []string

GetGroupIDsRaw 并发安全获取原始群组ID列表(不补默认值,用于索引判断)

func (*Client) GetLastHeartbeat

func (c *Client) GetLastHeartbeat() time.Time

GetLastHeartbeat 原子读最后心跳时间(并发安全)

func (*Client) GetLastPong

func (c *Client) GetLastPong() time.Time

GetLastPong 原子读最后 Pong 响应时间(并发安全)

func (*Client) GetLastSeen

func (c *Client) GetLastSeen() time.Time

GetLastSeen 原子读最后活跃时间(并发安全)

func (*Client) GetMetadataSnapshot

func (c *Client) GetMetadataSnapshot() map[string]interface{}

GetMetadataSnapshot 线程安全地获取元数据的只读副本(用于序列化/日志) 惰性分配下无元数据时返回空对象(非 nil),保持序列化输出形态与预分配时代一致(metadata:{})

func (*Client) GetMetadataValue

func (c *Client) GetMetadataValue(key string) (interface{}, bool)

GetMetadataValue 线程安全地读取元数据值

func (*Client) GetNamespace

func (c *Client) GetNamespace() string

GetNamespace 获取命名空间ID,空值返回默认命名空间

func (*Client) GetStatus

func (c *Client) GetStatus() UserStatus

GetStatus 原子读用户状态(并发安全,广播遍历中使用)

func (*Client) GetUserAgent

func (c *Client) GetUserAgent() string

GetUserAgent 获取用户代理

func (*Client) GetVIPLevel

func (c *Client) GetVIPLevel() VIPLevel

GetVIPLevel 原子读VIP等级(并发安全,广播遍历中使用)

func (*Client) IsClosed

func (c *Client) IsClosed() bool

IsClosed 检查客户端channel是否已关闭

func (*Client) MarkClosed

func (c *Client) MarkClosed()

MarkClosed 标记客户端channel为已关闭

func (*Client) MarshalJSON

func (c *Client) MarshalJSON() ([]byte, error)

MarshalJSON 自定义 JSON 序列化,确保对仍在 Hub 中、被业务并发修改的活 *Client 进行 json.Marshal 时不会触发 "fatal error: concurrent map iteration and map write", 也不会与 With*/Set* 并发写产生数据竞争

背景:Client.Metadata 是普通 map,encoding/json 通过反射遍历它时不持 mu, 一旦与 SetMetadataValue / Hub.UpdateClientMetadata 并发即导致进程级 fatal(不可 recover) 此外 Namespace/Role/ClientIP/Department/Skills/MaxTickets/NodeID/NodeIP/NodePort/ ClientType/ConnectionType 等字段注册后仍会被 With* 并发写入,反射裸读同样构成数据竞争 该方法一次堵住所有 json.Marshal(client) 调用点(如在线状态入库 BatchSetClientsOnline)

实现要点:

  1. 整段 json.Marshal 期间持 mu.RLock,与所有 With*/SetMetadataValue/SetGroupID 的 Lock 互斥, 覆盖别名提升字段(*clientAlias)的反射读路径;
  2. type clientAlias Client 定义本地别名类型,剥离 *Client 的 MarshalJSON 方法, 避免 json.Marshal 递归调用自身;别名以指针嵌入,不拷贝整个 Client(内含 sync.RWMutex/sync.Mutex/atomic,值拷贝会触发 copylocks),其余字段沿用默认序列化;
  3. sync.RWMutex 不可重入:已持 RLock 时不能再调 GetMetadataSnapshot/GetGroupIDRaw(会再次 RLock), 故 Metadata 快照与 GroupID 读均内联在此处直接读取;
  4. 对“注册后仍会被并发写入”的字段,覆盖别名提升的同名字段,改用并发安全取值: - Metadata 内联快照(与 hub/connection_record.go 入库语义一致) - GroupID 直接读字段(已持 RLock,与 SetGroupID/WithGroupID 互斥) - LastHeartbeat/LastSeen/LastPong 走原子读(唯一存储为原子字段,无 time.Time 字段) - Status/VIPLevel 走原子读(消除与 SetStatus/SetVIPLevel 并发写的数据竞争,atomic 不依赖 mu 可安全调用)

JSON 字段集合与原 json.Marshal(client) 完全一致,仅取值路径变为并发安全

func (*Client) SetBacklogRatio

func (c *Client) SetBacklogRatio(used, capacity int)

SetBacklogRatio 写泵单写者更新 SendChan 利用率(万分比整数原子 store,无锁) 写泵每批 drain 后调用;慢消费者扫描器经 BacklogRatio 无锁读取

func (*Client) SetGroupID

func (c *Client) SetGroupID(groupID string)

SetGroupID 并发安全更新群组ID(用于 MoveGroupMember 等运行时变更场景)

func (*Client) SetLastHeartbeat

func (c *Client) SetLastHeartbeat(t time.Time)

SetLastHeartbeat 原子更新最后心跳时间(仅写原子镜像,消除 time.Time 并发写数据竞争)

func (*Client) SetLastPong

func (c *Client) SetLastPong(t time.Time)

SetLastPong 原子更新最后 Pong 响应时间(仅写原子镜像,消除 time.Time 并发写数据竞争)

func (*Client) SetLastSeen

func (c *Client) SetLastSeen(t time.Time)

SetLastSeen 原子更新最后活跃时间(仅写原子镜像,消除 time.Time 并发写数据竞争)

func (*Client) SetMetadataValue

func (c *Client) SetMetadataValue(key string, value interface{})

SetMetadataValue 线程安全地写入元数据值

func (*Client) SetPendingCounter

func (c *Client) SetPendingCounter(counter *atomic.Int64)

SetPendingCounter 在连接注册并对外可见前,绑定所属 Hub 的共享积压计数

func (*Client) SetStatus

func (c *Client) SetStatus(status UserStatus)

SetStatus 原子更新用户状态(同时更新 Status 字段和原子镜像,并发安全)

func (*Client) SetVIPLevel

func (c *Client) SetVIPLevel(level VIPLevel)

SetVIPLevel 原子更新VIP等级(同时更新 VIPLevel 字段和原子镜像,并发安全)

func (*Client) StatsFlushDue

func (c *Client) StatsFlushDue(interval time.Duration) bool

StatsFlushDue 心跳统计落库节流判定:距上次落库超过 interval 才放行并抢占本次slot(CAS 抢占:并发路径提交同一连接时仅一次放行;零值状态=新连接首条心跳立即落库) 节流动机:心跳落库写的是 dashboard 活跃统计的分钟桶口径源数据,逐心跳全量落库 在 CRDB 上构成 range 写热点(小批被大批排队,实测 2 行 UPDATE 长尾 900ms+),按节流粒度落库写量随节流系数等比下降、桶归属统计无损

func (*Client) TrySend

func (c *Client) TrySend(data []byte) (sent bool)

TrySend 尝试向客户端发送数据(WebSocket),如果已关闭或失败则返回false

性能:无锁热路径——发送是全库最热操作(广播扇出时每客户端一次), closeClientChannel 持 CloseMu 先 MarkClosed 再 close(channel),此处不抢锁: 闭锁竞态窗口内的 send panic 由 recover 兜住并标记关闭,语义与持锁版一致 (消息要么在 close 前入队被写泵排空,要么被丢弃返回 false,永不外泄 panic)

func (*Client) TrySendControl

func (c *Client) TrySendControl(data []byte) (sent bool)

TrySendControl 尝试向控制通道发送必达级数据(KickOut/ForceOffline/Ack 等)

双 lane 单写者模型(Slack async_priority 模式):控制通道与数据通道分离, 数据 SendChan 满时控制消息仍可入 CtrlCh;写泵在 select 前先非阻塞排空 CtrlCh, 保证控制帧严格优先于数据帧。无锁热路径 + recover 闭锁竞态防护,风格同 TrySend

返回 false:通道满或已关闭,调用方(SendControl)走降级语义(断链/ACK 重试)

func (*Client) TrySendSSE

func (c *Client) TrySendSSE(msg *HubMessage) (sent bool)

TrySendSSE 尝试向SSE客户端发送消息,如果已关闭或失败则返回false 无锁热路径,闭锁竞态处理同 TrySend

func (*Client) WithAppID

func (c *Client) WithAppID(appID string) *Client

WithAppID 设置应用ID(最上层隔离维度) 线程安全(持 mu),与 MarshalJSON 反射读互斥 通常在连接建立时设置一次,运行期不变 空值归一化为 DefaultAppID(与 WithNamespace 归一化策略一致)

func (*Client) WithClientIP

func (c *Client) WithClientIP(ip string) *Client

WithClientIP 设置客户端IP 线程安全(持 mu),与 MarshalJSON 反射读互斥

func (*Client) WithClientType

func (c *Client) WithClientType(clientType ClientType) *Client

WithClientType 设置客户端类型 线程安全(持 mu),与 MarshalJSON 反射读互斥

func (*Client) WithContext

func (c *Client) WithContext(ctx context.Context) *Client

WithContext 设置上下文

func (*Client) WithDepartment

func (c *Client) WithDepartment(dept Department) *Client

WithDepartment 设置部门 线程安全(持 mu),与 MarshalJSON 反射读互斥

func (*Client) WithGroupID

func (c *Client) WithGroupID(groupID string) *Client

WithGroupID 设置群组ID 普通用户:连接后自动加入该群组;观察者:只接收指定命名空间+群组的消息通知 线程安全(复用 SetGroupID 持 mu),注册后并发调用也安全

func (*Client) WithGroupIDs

func (c *Client) WithGroupIDs(groupIDs []string) *Client

WithGroupIDs 设置群组ID列表(管理端一人多组场景) 持 mu;写 GroupIDs,len>0 时同步 GroupID=groupIDs[0](首项=主群组,保证单值字段与观察者 scope/MarshalJSON 一致) len==0 时清空 GroupIDs,不触碰 GroupID(留给 GetGroupIDs 回退读取)

func (*Client) WithMaxTickets

func (c *Client) WithMaxTickets(max int) *Client

WithMaxTickets 设置最大工单数 线程安全(持 mu),与 MarshalJSON 反射读互斥

func (*Client) WithMetadata

func (c *Client) WithMetadata(key string, value interface{}) *Client

WithMetadata 设置元数据 线程安全(复用 SetMetadataValue 持 mu),注册后并发调用也安全; 通常用于构造期链式构造,运行期写入推荐直接使用 SetMetadataValue / Hub.UpdateClientMetadata

func (*Client) WithMetadataMap

func (c *Client) WithMetadataMap(metadata map[string]interface{}) *Client

WithMetadataMap 批量设置元数据 线程安全(持锁一次性写入,避免注册后与 MarshalJSON 等序列化路径在 map 上数据竞争)

func (*Client) WithNamespace

func (c *Client) WithNamespace(namespace string) *Client

WithNamespace 设置命名空间ID 线程安全(持 mu),与 MarshalJSON 反射读互斥

func (*Client) WithNodeInfo

func (c *Client) WithNodeInfo(nodeID, nodeIP string, nodePort int) *Client

WithNodeInfo 设置节点信息 线程安全(持 mu),与 MarshalJSON 反射读互斥

func (*Client) WithRole

func (c *Client) WithRole(role UserRole) *Client

WithRole 设置用户角色 线程安全(持 mu),与 MarshalJSON 反射读互斥

func (*Client) WithSSEChannels

func (c *Client) WithSSEChannels(messageCh chan *HubMessage, closeCh chan struct{}) *Client

WithSSEChannels 设置SSE通道

func (*Client) WithSSEWriter

func (c *Client) WithSSEWriter(w http.ResponseWriter, flusher http.Flusher) *Client

WithSSEWriter 设置SSE Writer 线程安全(持 mu 保护 ConnectionType 序列化字段)

func (*Client) WithSendChan

func (c *Client) WithSendChan(ch chan []byte) *Client

WithSendChan 设置发送通道

func (*Client) WithSkills

func (c *Client) WithSkills(skills []Skill) *Client

WithSkills 设置技能列表 线程安全(持 mu),与 MarshalJSON 反射读互斥

func (*Client) WithStatus

func (c *Client) WithStatus(status UserStatus) *Client

WithStatus 设置用户状态 线程安全(复用 SetStatus 同步更新 Status 字段与 statusVal 原子镜像, 避免构造期设置后 GetStatus/MarshalJSON 读到 NewClient 的旧原子值)

func (*Client) WithVIPLevel

func (c *Client) WithVIPLevel(level VIPLevel) *Client

WithVIPLevel 设置VIP等级 线程安全(复用 SetVIPLevel 同步更新 VIPLevel 字段与 vipLevelVal 原子镜像)

func (*Client) WithWebSocketConn

func (c *Client) WithWebSocketConn(conn *websocket.Conn) *Client

WithWebSocketConn 设置WebSocket连接 线程安全(持 mu 保护 ConnectionType 序列化字段)

type ClientType

type ClientType string

ClientType 客户端类型

const (
	ClientTypeWeb     ClientType = "web"     // Web客户端
	ClientTypeMobile  ClientType = "mobile"  // 移动端
	ClientTypeDesktop ClientType = "desktop" // 桌面客户端
	ClientTypeAPI     ClientType = "api"     // API客户端
)

func MapDeviceTypeToClientType

func MapDeviceTypeToClientType(deviceType string) ClientType

MapDeviceTypeToClientType 将 go-toolbox 的 DeviceType 映射到 go-wsc 的 ClientType

映射规则:

  • bot -> api (机器人视为 API 客户端)
  • mobile -> mobile (移动设备)
  • tablet -> mobile (平板视为移动设备)
  • desktop -> web (桌面设备视为 Web 客户端)
  • unknown -> web (未知设备默认为 Web 客户端)

func (ClientType) IsValid

func (c ClientType) IsValid() bool

IsValid 检查客户端类型是否有效

func (ClientType) String

func (c ClientType) String() string

String 实现Stringer接口

type ClusterStats

type ClusterStats struct {
	TotalNodes        int          `json:"total_nodes"`
	ActiveNodes       int          `json:"active_nodes"`
	TotalConnections  int64        `json:"total_connections"`
	ActiveConnections int64        `json:"active_connections"`
	MessagesSent      int64        `json:"messages_sent"`
	MessagesReceived  int64        `json:"messages_received"`
	BroadcastsSent    int64        `json:"broadcasts_sent"`
	NodesStats        []*NodeStats `json:"nodes_stats"`
	UpdateTime        time.Time    `json:"update_time"`
}

ClusterStats 集群统计信息

type ConnectionQuality

type ConnectionQuality struct {
	// ========== 关联标识 ==========
	ID           uint64 `gorm:"primaryKey;autoIncrement;comment:自增主键" json:"id"`
	ConnectionID string `` /* 147-byte string literal not displayed */
	UserID       string `gorm:"column:user_id;size:64;index;comment:用户ID(冗余索引,便于按用户查质量)" json:"user_id"`

	// ========== 多租户/命名空间隔离 ==========
	// 与 connect 表 + Bitmap 分桶维度一致,支持按 app+namespace 查连接质量
	// 索引名与 connect 表区分(SQLite/PostgreSQL 索引名为库级作用域,重名会建表失败)
	AppID     string `` /* 139-byte string literal not displayed */
	Namespace string `` /* 138-byte string literal not displayed */

	// ========== 质量指标 ==========
	ReconnectCount   int     `gorm:"column:reconnect_count;default:0;comment:重连次数" json:"reconnect_count"`
	MessagesSent     int64   `gorm:"column:messages_sent;default:0;comment:发送消息总数" json:"messages_sent"`
	MessagesReceived int64   `gorm:"column:messages_received;default:0;comment:接收消息总数" json:"messages_received"`
	BytesSent        int64   `gorm:"column:bytes_sent;default:0;comment:发送字节数" json:"bytes_sent"`
	BytesReceived    int64   `gorm:"column:bytes_received;default:0;comment:接收字节数" json:"bytes_received"`
	AveragePingMs    float64 `gorm:"column:average_ping_ms;comment:平均Ping延迟(毫秒)" json:"average_ping_ms,omitempty"`
	MaxPingMs        float64 `gorm:"column:max_ping_ms;comment:最大Ping延迟(毫秒)" json:"max_ping_ms,omitempty"`
	MinPingMs        float64 `gorm:"column:min_ping_ms;comment:最小Ping延迟(毫秒)" json:"min_ping_ms,omitempty"`
	PacketLossRate   float64 `gorm:"column:packet_loss_rate;comment:丢包率(%)" json:"packet_loss_rate,omitempty"`

	// ========== 错误信息 ==========
	ErrorCount  int        `gorm:"column:error_count;default:0;comment:错误次数" json:"error_count"`
	LastError   string     `gorm:"column:last_error;type:text;comment:最后错误信息" json:"last_error,omitempty"`
	LastErrorAt *time.Time `gorm:"column:last_error_at;comment:最后错误时间" json:"last_error_at,omitempty"`

	// ========== 评分与活跃时间 ==========
	QualityScore float64    `gorm:"column:quality_score;index;comment:连接质量评分(0-100)" json:"quality_score"`
	LastActiveAt *time.Time `gorm:"column:last_active_at;index;comment:最后活跃时间(供清理)" json:"last_active_at,omitempty"`

	// ========== 系统字段 ==========
	CreatedAt time.Time `gorm:"autoCreateTime;comment:记录创建时间" json:"created_at"`
	UpdatedAt time.Time `gorm:"autoUpdateTime;comment:记录更新时间" json:"updated_at"`
}

ConnectionQuality 连接质量记录 - 与 ConnectionRecord 1:1 关联 承载运行时质量指标(心跳延迟/消息统计/错误/重连)与评分,随 batcher 高频批量更新

func (*ConnectionQuality) AddError

func (q *ConnectionQuality) AddError(err error)

AddError 添加错误记录

func (*ConnectionQuality) FinalScore

func (q *ConnectionQuality) FinalScore(durationSec int64) float64

FinalScore 断开终评(0-100,可归零) 在 LiveScore 基础上追加时长维度(最多扣20),供断开时终评 时长梯度:<30s扣20(短连接即断视为低质量)/<5min扣10/<30min扣5/≥30min不扣

func (*ConnectionQuality) IncrementReconnect

func (q *ConnectionQuality) IncrementReconnect()

IncrementReconnect 增加重连次数

func (*ConnectionQuality) LiveScore

func (q *ConnectionQuality) LiveScore() float64

LiveScore 在线实时评分(0-100,保底20) 4 维扣分:丢包(最多25)+延迟(最多25)+错误率(最多15)+重连(最多15),合计最多扣80,保底20 供 batcher flush 周期性计算(连接在线时无 duration 终值)

func (ConnectionQuality) TableComment

func (ConnectionQuality) TableComment() string

TableComment 表注释

func (ConnectionQuality) TableName

func (ConnectionQuality) TableName() string

TableName 指定表名

func (*ConnectionQuality) UpdateBytesStats

func (q *ConnectionQuality) UpdateBytesStats(sent, received int64)

UpdateBytesStats 更新字节统计

func (*ConnectionQuality) UpdateMessageStats

func (q *ConnectionQuality) UpdateMessageStats(sent, received int64)

UpdateMessageStats 更新消息统计

func (*ConnectionQuality) UpdatePingStats

func (q *ConnectionQuality) UpdatePingStats(pingMs float64)

UpdatePingStats 更新Ping延迟统计(简单平均,batcher 路径用 SQL 移动平均)

type ConnectionQueryOptions

type ConnectionQueryOptions struct {
	UserID     string // 用户ID过滤
	NodeID     string // 节点ID过滤
	IsActive   *bool  // 是否活跃(nil表示不过滤)
	IsAbnormal *bool  // 是否异常(nil表示不过滤)
	ClientIP   string // 客户端IP过滤
	Limit      int    // 限制数量
	Offset     int    // 偏移量
	OrderBy    string // 排序字段(默认 connected_at DESC)
}

ConnectionQueryOptions 连接查询选项

type ConnectionRecord

type ConnectionRecord struct {
	// ========== 基础标识信息 ==========
	ID           uint64 `gorm:"primaryKey;autoIncrement;comment:自增主键" json:"id"`
	ConnectionID string `` /* 130-byte string literal not displayed */
	UserID       string `gorm:"column:user_id;size:64;not null;index;comment:用户ID(同一用户可有多条记录)" json:"user_id"`

	// ========== 多租户/命名空间隔离 ==========
	// 与 Bitmap 在线状态层分桶维度一致,支持按 app+namespace 过滤连接
	AppID     string `gorm:"column:app_id;size:64;index:idx_app_namespace;comment:应用ID(多租户隔离,与 Bitmap 分桶一致)" json:"app_id"`
	Namespace string `gorm:"column:namespace;size:64;index:idx_app_namespace;comment:命名空间(与 Bitmap 分桶一致)" json:"namespace"`

	// ========== 服务器节点信息 ==========
	NodeID   string `gorm:"column:node_id;size:100;index;comment:服务器节点ID" json:"node_id"`
	NodeIP   string `gorm:"column:node_ip;size:45;comment:服务器IP" json:"node_ip"`
	NodePort int    `gorm:"column:node_port;comment:服务器端口" json:"node_port"`

	// ========== 客户端信息 ==========
	ClientIP   string     `gorm:"column:client_ip;size:45;index;comment:客户端IP地址(用于索引查询)" json:"client_ip"`
	ClientType ClientType `gorm:"column:client_type;size:20;comment:客户端类型(web/mobile/desktop/sdk)" json:"client_type"`

	// ========== 连接协议信息 ==========
	Protocol ConnectionType `gorm:"column:protocol;size:20;default:websocket;comment:协议类型(websocket/sse/http)" json:"protocol"`

	// ========== 连接时间信息 ==========
	ConnectedAt    time.Time  `gorm:"column:connected_at;index;not null;comment:连接建立时间" json:"connected_at"`
	DisconnectedAt *time.Time `gorm:"column:disconnected_at;index;comment:断开连接时间" json:"disconnected_at,omitempty"`
	Duration       int64      `gorm:"column:duration;comment:连接持续时长(秒)" json:"duration,omitempty"`
	LastPingAt     *time.Time `gorm:"column:last_ping_at;comment:最后心跳时间" json:"last_ping_at,omitempty"`
	LastPongAt     *time.Time `gorm:"column:last_pong_at;comment:最后Pong响应时间" json:"last_pong_at,omitempty"`

	// ========== 断开连接信息 ==========
	DisconnectReason string `` /* 142-byte string literal not displayed */
	DisconnectCode   int    `gorm:"column:disconnect_code;comment:断开代码" json:"disconnect_code,omitempty"`

	// ========== 状态标识 ==========
	IsActive        bool `gorm:"column:is_active;index;comment:是否活跃连接" json:"is_active"`
	IsForcedOffline bool `gorm:"column:is_forced_offline;comment:是否被强制下线" json:"is_forced_offline"`
	IsAbnormal      bool `gorm:"column:is_abnormal;index;comment:是否异常断开" json:"is_abnormal"`

	// ========== 元数据 ==========
	Metadata sqlbuilder.MapAny `gorm:"column:metadata;type:json;comment:请求元数据JSON(包含所有HTTP头信息)" json:"metadata,omitempty"`

	// ========== 系统字段 ==========
	CreatedAt time.Time `gorm:"autoCreateTime;comment:记录创建时间" json:"created_at"`
	UpdatedAt time.Time `gorm:"autoUpdateTime;comment:记录更新时间" json:"updated_at"`
}

ConnectionRecord WebSocket连接记录模型 - 详细记录每次连接的完整信息 设计说明:支持多设备登录,每个连接维护一条独立记录

func (*ConnectionRecord) CalculateDuration

func (c *ConnectionRecord) CalculateDuration()

CalculateDuration 计算连接持续时长

func (*ConnectionRecord) IsOnline

func (c *ConnectionRecord) IsOnline() bool

IsOnline 判断是否在线

func (ConnectionRecord) TableComment

func (ConnectionRecord) TableComment() string

TableComment 表注释

func (ConnectionRecord) TableName

func (ConnectionRecord) TableName() string

TableName 指定表名

type ConnectionStats

type ConnectionStats struct {
	TotalConnections      int64   `json:"total_connections"`       // 总连接数
	ActiveConnections     int64   `json:"active_connections"`      // 活跃连接数
	AverageDuration       float64 `json:"average_duration"`        // 平均连接时长(秒)
	TotalMessagesSent     int64   `json:"total_messages_sent"`     // 总发送消息数(拆表后零填充,跨表补充由调用方)
	TotalMessagesReceived int64   `json:"total_messages_received"` // 总接收消息数(拆表后零填充)
	TotalBytesSent        int64   `json:"total_bytes_sent"`        // 总发送字节数(拆表后零填充)
	TotalBytesReceived    int64   `json:"total_bytes_received"`    // 总接收字节数(拆表后零填充)
	AveragePingMs         float64 `json:"average_ping_ms"`         // 平均Ping延迟(拆表后零填充)
	AbnormalRate          float64 `json:"abnormal_rate"`           // 异常断开率
	AverageReconnectCount float64 `json:"average_reconnect_count"` // 平均重连次数(拆表后零填充)
}

ConnectionStats 连接统计信息 质量维度字段(TotalMessages*/TotalBytes*/AveragePingMs/AverageReconnectCount)保留兼容调用方 拆表后由本方法零填充,跨表补充由调用方按需 JOIN qualityRepo

type ConnectionStatus

type ConnectionStatus string

ConnectionStatus 连接状态

const (
	ConnectionStatusConnecting   ConnectionStatus = "connecting"   // 连接中
	ConnectionStatusConnected    ConnectionStatus = "connected"    // 已连接
	ConnectionStatusDisconnected ConnectionStatus = "disconnected" // 已断开
	ConnectionStatusReconnecting ConnectionStatus = "reconnecting" // 重连中
	ConnectionStatusError        ConnectionStatus = "error"        // 连接错误
)

func (ConnectionStatus) IsValid

func (s ConnectionStatus) IsValid() bool

IsValid 检查连接状态是否有效

func (ConnectionStatus) String

func (s ConnectionStatus) String() string

String 实现Stringer接口

type ConnectionType

type ConnectionType string

ConnectionType 连接类型(WebSocket/SSE)

const (
	ConnectionTypeWebSocket ConnectionType = "websocket" // WebSocket连接
	ConnectionTypeSSE       ConnectionType = "sse"       // SSE连接
)

func (ConnectionType) IsValid

func (c ConnectionType) IsValid() bool

IsValid 检查连接类型是否有效

func (ConnectionType) String

func (c ConnectionType) String() string

String 实现Stringer接口

type DeliverResult

type DeliverResult struct {
	Mode      DeliveryMode // 投递模式
	AppID     string       // 应用ID(已归一化)
	Namespace string       // 命名空间ID(全局广播时为空)
	GroupIDs  []string     // 群组ID列表(非群组场景为空)

	// P2P / 群组可靠投递场景(per-member 统计)
	TotalMembers   int // 目标成员数(P2P=1,群组=去重后成员数,广播=0)
	OnlineMembers  int // 在线成员数(至少在一个节点有连接)
	OfflineMembers int // 离线成员数
	Sent           int // 在线且成功投递数(含跨节点路由)
	StoredOffline  int // 离线成员存储离线消息数
	Failed         int // 投递失败数(在线但队列满等)

	// 广播场景(fire-and-forget 本地成功投递数)
	// 群组广播/命名空间广播/全局广播的本地送达数;全局广播为异步,此值为 0
	LocalDelivered int

	// 群组广播场景:可聚合档消息已入本地聚合窗口(drain 窗口批量扇出,LocalDelivered 不含窗口内消息;跨节点意图照常已发出)
	Aggregated bool

	Errors []error // 投递过程中的错误
}

DeliverResult 统一投递结果(覆盖 P2P/群组/广播所有场景)

设计原则:

  • 始终非 nil:错误收集到 Errors,调用方按 Mode + 计数字段判断结果
  • 字段语义按 Mode 分组: P2P / GroupReliable:TotalMembers/OnlineMembers/OfflineMembers/Sent/StoredOffline/Failed 有效 GroupBroadcast / Namespace / Global:LocalDelivered 有效(本地成功投递数)
  • 跨节点投递为异步,不计入返回值(仅统计本地同步投递结果)

func (*DeliverResult) AddError

func (r *DeliverResult) AddError(err error)

AddError 安全追加错误(避免 nil 追加)

func (*DeliverResult) HasError

func (r *DeliverResult) HasError() bool

HasError 是否有错误

type DeliveryGuarantee

type DeliveryGuarantee int8

DeliveryGuarantee 送达分级

int8 底层:仅 4 个枚举值,1 字节存储塞入结构体末尾标志区(与 bool 同区), 零 padding 净增长(结构体大小守护测试 432 字节门限)

零值为 GuaranteeUnset:未显设置时走 ResolveGuarantee 决策树(类型默认表/分类评分), 保证零值构造的消息(&HubMessage{})也能得到正确的分级,无需强制初始化

const (
	// GuaranteeUnset 未设置(零值,走决策树推导)
	GuaranteeUnset DeliveryGuarantee = iota
	// GuaranteeStandard 普通:实时投递 + 离线兜底(默认档)
	GuaranteeStandard
	// GuaranteeEphemeral 高频流式:latest-wins 合并,只保证最新送达
	GuaranteeEphemeral
	// GuaranteeGuaranteed 必达:控制通道 + ACK + 离线多级兜底,零丢失
	GuaranteeGuaranteed
	// GuaranteeAggregatable 可聚合:全保留窗口攒批(N 条合 1 条数组帧),容忍窗口级延迟与 Ephemeral 的区别:latest-wins 丢旧保新(只需最新值),可聚合全保留合并(条条要送达)
	GuaranteeAggregatable
)

func GuaranteeForType

func GuaranteeForType(mt MessageType) (DeliveryGuarantee, bool)

GuaranteeForType 查询消息类型的默认送达分级 第二返回值表示该类型是否在默认表中(未登记的类型默认走普通档)

func (DeliveryGuarantee) IsExpress

func (g DeliveryGuarantee) IsExpress() bool

IsExpress 是否快速通道档(高频流式/可聚合)

express 语义:fire-and-forget——不落 message_record、不进离线队列、不走重试链、 队列满语义丢弃(latest-wins 语义下同 key 下一条消息自然覆盖,丢弃是正确行为); 可靠档(Standard/Guaranteed)保持现状全机制

func (DeliveryGuarantee) String

func (g DeliveryGuarantee) String() string

String 送达分级的日志/序列化友好表示

type DeliveryMode

type DeliveryMode int

DeliveryMode 投递模式(Deliver 内部根据 ctx+msg 路由决策)

const (
	DeliveryModeP2P            DeliveryMode = iota // 点对点(msg.Receiver 非空)
	DeliveryModeGroupReliable                      // 群组可靠投递(RequireAck=true,per-member SendToUserWithRetry + 离线存储)
	DeliveryModeGroupBroadcast                     // 群组广播(RequireAck=false,fire-and-forget)
	DeliveryModeNamespace                          // 命名空间广播(namespace 非空,无 groupIDs)
	DeliveryModeGlobal                             // 全局广播(namespace 为空,无 groupIDs)
)

func (DeliveryMode) String

func (m DeliveryMode) String() string

String 返回投递模式的可读名称(日志/调试用)

type Department

type Department string

Department 部门类型

const (
	DepartmentSales     Department = "sales"     // 销售部门
	DepartmentSupport   Department = "support"   // 技术支持
	DepartmentBilling   Department = "billing"   // 计费部门
	DepartmentGeneral   Department = "general"   // 综合部门
	DepartmentTechnical Department = "technical" // 技术部门
)

func (Department) IsValid

func (d Department) IsValid() bool

IsValid 检查部门是否有效

func (Department) String

func (d Department) String() string

String 实现Stringer接口

type DisconnectReason

type DisconnectReason string

DisconnectReason 断开连接原因

const (
	DisconnectReasonReadError      DisconnectReason = "read_error"      // 读取错误
	DisconnectReasonWriteError     DisconnectReason = "write_error"     // 写入错误
	DisconnectReasonContextDone    DisconnectReason = "context_done"    // 上下文结束
	DisconnectReasonCloseMessage   DisconnectReason = "close_message"   // 关闭消息
	DisconnectReasonHeartbeatFail  DisconnectReason = "heartbeat_fail"  // 心跳失败
	DisconnectReasonKickOut        DisconnectReason = "kick_out"        // 被踢出
	DisconnectReasonForceOffline   DisconnectReason = "force_offline"   // 强制下线
	DisconnectReasonTimeout        DisconnectReason = "timeout"         // 超时
	DisconnectReasonClientRequest  DisconnectReason = "client_request"  // 客户端主动断开
	DisconnectReasonServerShutdown DisconnectReason = "server_shutdown" // 服务器关闭
	DisconnectReasonUnknown        DisconnectReason = "unknown"         // 未知原因
)

func (DisconnectReason) IsValid

func (r DisconnectReason) IsValid() bool

IsValid 检查断开原因是否有效

func (DisconnectReason) String

func (r DisconnectReason) String() string

String 实现Stringer接口

type DisconnectionEntry

type DisconnectionEntry struct {
	ConnectionID   string
	ConnectedAt    time.Time
	DisconnectedAt time.Time
	Reason         DisconnectReason
	Code           int
}

DisconnectionEntry 断连终态条目(由连接记录仓库批量消费) 快照语义:断连瞬间冻结 DisconnectedAt/ConnectedAt,flush 延迟不虚增 duration; 同一连接重复提交时去重保留最后一条(断连是单次事件,终态以最后一次为准)

type DistributedMessage

type DistributedMessage struct {
	Type          OperationType `json:"type"`                     // 操作类型
	NodeID        string        `json:"node_id"`                  // 源节点ID
	TargetID      string        `json:"target_id"`                // 目标ID(用户ID、节点ID等)
	TraceID       string        `json:"trace_id,omitempty"`       // 全链路追踪ID(从 ctx 自动注入,跨节点序列化携带)
	Message       *HubMessage   `json:"message"`                  // 消息数据(用于 send_message, broadcast, observer_notify)
	Reason        string        `json:"reason"`                   // 原因
	Timestamp     time.Time     `json:"timestamp"`                // 时间戳
	AppID         string        `json:"app_id,omitempty"`         // 应用ID(最上层隔离维度,路由信封携带,空=全局共享)
	Namespace     string        `json:"namespace,omitempty"`      // 命名空间ID(路由信封携带,空=全命名空间广播,非空=指定命名空间)
	GroupIDs      []string      `json:"group_ids,omitempty"`      // 群组ID列表(支持多群组,观察者可订阅多个组;空表示无群组操作)
	ExcludeSender bool          `json:"exclude_sender,omitempty"` // 是否排除发送者(跨节点群组广播 PubSub 兜底携带,与 gRPC BroadcastGroupRequest 对齐)
	SenderID      string        `json:"sender_id,omitempty"`      // 发送者ID(排除发送者时用,跨节点 PubSub 兜底场景)
}

DistributedMessage 分布式消息结构

func (*DistributedMessage) ContextFrom

func (dm *DistributedMessage) ContextFrom(parent context.Context) context.Context

ContextFrom 基于分布式消息的 trace_id 创建一个携带 trace 信息的 context 用于消息流转路径中恢复 ctx(如 PubSub 消费端、回调等场景)

func (*DistributedMessage) InjectContext

func (dm *DistributedMessage) InjectContext(ctx context.Context) *DistributedMessage

InjectContext 从 ctx 注入上下文信息到分布式消息(trace_id 等) 优先从 OTel span 提取 trace_id,fallback 到 ctx.Value(logger.ContextKeyTraceID) 已有 trace_id 时不覆盖(跨节点消息保留源 trace)

type ErrorSeverity

type ErrorSeverity string

ErrorSeverity 错误严重程度

const (
	ErrorSeverityInfo     ErrorSeverity = "info"     // 信息
	ErrorSeverityWarning  ErrorSeverity = "warning"  // 警告
	ErrorSeverityError    ErrorSeverity = "error"    // 错误
	ErrorSeverityCritical ErrorSeverity = "critical" // 严重错误
	ErrorSeverityFatal    ErrorSeverity = "fatal"    // 致命错误
)

func (ErrorSeverity) IsValid

func (s ErrorSeverity) IsValid() bool

IsValid 检查严重程度是否有效

func (ErrorSeverity) String

func (s ErrorSeverity) String() string

String 实现Stringer接口

type ErrorUpdateEntry

type ErrorUpdateEntry struct {
	ConnectionID string
	ErrorCount   int64
	LastError    string
	LastErrorAt  time.Time
}

ErrorUpdateEntry 错误更新条目(由连接质量仓库批量消费) 同一连接的多次错误在攒批 flush 时合并:ErrorCount 为合并次数,LastError/LastErrorAt 为最新一次

type EventStatus

type EventStatus string

EventStatus 事件状态类型

const (
	// EventTypeOnline 上线状态
	EventTypeOnline EventStatus = "online"
	// EventTypeOffline 下线状态
	EventTypeOffline EventStatus = "offline"
)

type FailureReason

type FailureReason string

FailureReason 失败原因

const (
	FailureReasonQueueFull    FailureReason = "queue_full"    // 队列已满
	FailureReasonUserOffline  FailureReason = "user_offline"  // 用户离线
	FailureReasonConnError    FailureReason = "conn_error"    // 连接错误
	FailureReasonAckTimeout   FailureReason = "ack_timeout"   // ACK超时
	FailureReasonSendTimeout  FailureReason = "send_timeout"  // 发送超时
	FailureReasonNetworkError FailureReason = "network_error" // 网络错误
	FailureReasonUnknown      FailureReason = "unknown"       // 未知错误
	FailureReasonMaxRetry     FailureReason = "max_retry"     // 超过最大重试次数
	FailureReasonExpired      FailureReason = "expired"       // 消息过期
)

type Group

type Group struct {
	AppID      string                 `json:"app_id"`             // 应用ID(默认 "__default_app__")
	Namespace  string                 `json:"namespace"`          // 命名空间ID(默认 "default")
	GroupID    string                 `json:"group_id"`           // 群组ID(命名空间内唯一)
	Name       string                 `json:"name"`               // 群组名称
	OwnerID    string                 `json:"owner_id"`           // 群主用户ID
	MaxMembers int                    `json:"max_members"`        // 最大成员数(0 表示不限)
	CreatedAt  time.Time              `json:"created_at"`         // 创建时间
	Metadata   map[string]interface{} `json:"metadata,omitempty"` // 扩展元数据
}

Group 群组模型 群组成员关系为"用户级"持久关系,存于 Redis,与连接生命周期无关:

  • 用户加入群组后即使下线仍是成员,上线后通过离线消息补发群组消息
  • 跨节点共享,任意节点均可查询/管理群组成员
  • 群组隶属于命名空间(Namespace),默认为 "default",类似 k8s namespace

func (*Group) GetAppID

func (g *Group) GetAppID() string

GetAppID 获取应用ID,空值返回默认应用ID

func (*Group) GetNamespace

func (g *Group) GetNamespace() string

GetNamespace 获取命名空间ID,空值返回默认命名空间

type GroupObserverStats

type GroupObserverStats struct {
	GroupID       string           `json:"group_id"`       // 群组ID
	TotalUsers    int              `json:"total_users"`    // 该群组下观察者用户数
	TotalDevices  int              `json:"total_devices"`  // 该群组下观察者设备数
	ObserverStats []*ObserverStats `json:"observer_stats"` // 群组级观察者详细统计
}

GroupObserverStats 群组级观察者统计

type GroupSendResult

type GroupSendResult struct {
	AppID          string  // 应用ID
	GroupID        string  // 群组ID
	TotalMembers   int     // 群组总成员数
	OnlineMembers  int     // 在线成员数(至少在一个节点有连接)
	OfflineMembers int     // 离线成员数
	Sent           int     // 在线且成功投递数(含跨节点路由)
	StoredOffline  int     // 离线成员存储离线消息数
	Failed         int     // 投递失败数(在线但队列满等)
	Errors         []error // 投递过程中的错误
}

GroupSendResult 群组消息投递结果

type HeartbeatUpdateEntry

type HeartbeatUpdateEntry struct {
	ConnectionID string
	PingTime     *time.Time
	PingMs       float64 // >0 时更新 average/max/min_ping_ms
}

HeartbeatUpdateEntry 心跳更新条目(由连接记录仓库批量消费)

type HubHealthInfo

type HubHealthInfo struct {
	Status           string `json:"status"`
	IsRunning        bool   `json:"is_running"`
	WebSocketCount   int    `json:"websocket_count"`
	SSECount         int    `json:"sse_count"`
	TotalConnections int    `json:"total_connections"`
	NodeID           string `json:"node_id"`
}

HubHealthInfo Hub健康状态信息

type HubMessage

type HubMessage struct {
	// ========== 标识与路由(热点字段,置于头部以利用 cache line) ==========
	ID           string      `json:"id"`                      // 消息ID(用于ACK)
	TraceID      string      `json:"trace_id,omitempty"`      // 全链路追踪ID(从 ctx 自动注入,跨节点序列化携带)
	MessageType  MessageType `json:"message_type"`            // 消息类型
	Sender       string      `json:"sender"`                  // 发送者 (从上下文获取)
	SenderName   string      `json:"sender_name"`             // 发送者昵称
	SenderType   UserType    `json:"sender_type"`             // 发送者类型
	SenderClient string      `json:"sender_client,omitempty"` // 发送者客户端ID(多端同步标识)
	Receiver     string      `json:"receiver"`                // 接收者用户ID
	ReceiverName string      `json:"receiver_name"`           // 接收者昵称
	ReceiverType UserType    `json:"receiver_type"`           // 接收者用户类型

	// ========== 接收与节点路由 ==========
	ReceiverClient string `json:"receiver_client,omitempty"` // 接收者客户端ID
	ReceiverNode   string `json:"receiver_node,omitempty"`   // 接收者所在节点ID
	SessionID      string `json:"session_id"`                // 会话ID

	// ========== 内容 ==========
	Content string `json:"content"` // 消息内容

	// ========== 扩展数据(8 字节对齐字段) ==========
	Data     map[string]interface{} `json:"data,omitempty"` // 扩展数据(包含 content_extra、metadata、media_info)
	CreateAt time.Time              `json:"create_at"`      // 创建时间

	// ========== 消息ID与序列 ==========
	MessageID    string `json:"message_id"`                // 业务消息ID
	ReplyToMsgID string `json:"reply_to_msg_id,omitempty"` // 回复的消息ID
	SeqNo        int64  `json:"seq_no"`                    // 消息序列号

	// ========== 类型与策略(string,16 字节对齐) ==========
	Priority      Priority      `json:"priority"`                 // 优先级
	Source        MessageSource `json:"source,omitempty"`         // 消息来源(online/offline)
	PushType      PushType      `json:"push_type,omitempty"`      // 推送类型
	BroadcastType BroadcastType `json:"broadcast_type,omitempty"` // 广播类型(会话成员/全站)

	// ========== 路由信封(投递精确隔离,跨节点随消息体携带) ==========
	// ctx 路由元数据在异步队列消费时会丢失,故信封必须随消息体流转。
	// 入口层注入(InjectRoute),投递/跨节点/离线回放统一从 msg 取。
	// 隔离层次:AppID(应用) > Namespace(租户) > GroupID(平台) > UserID
	// 空 AppID = 全局共享(老部署兼容,不参与 appId 严格匹配过滤)
	AppID     string   `json:"app_id,omitempty"`    // 应用ID(最上层隔离维度,空=全局共享)
	Namespace string   `json:"namespace,omitempty"` // 命名空间ID(空=全局,已归一化时非空)
	GroupIDs  []string `json:"group_ids,omitempty"` // 群组ID列表(P2P 为 nil;单群组 len==1;多群组 len>1)

	// ========== 标志区(1 字节对齐字段集中末尾,零 padding) ==========
	DeliveryGuarantee   DeliveryGuarantee `json:"delivery_guarantee,omitempty"`    // 送达分级(必达/普通/高频流式;零值走决策树推导,详见 delivery_guarantee.go)
	RequireAck          bool              `json:"require_ack,omitempty"`           // 是否需要ACK确认
	SkipDatabaseStorage bool              `json:"skip_database_storage,omitempty"` // 是否跳过主数据库存储
	SkipSendToClient    bool              `json:"skip_send_to_client,omitempty"`   // 是否跳过发送到客户端
	// contains filtered or unexported fields
}

HubMessage Hub消息结构

字段对齐优化说明:

  • 所有 16 字节 string 字段集中在结构体前部(8 字节对齐,无 padding)
  • map/time.Time/int64 等其他 8 字节对齐字段紧随其后
  • 3 个 bool 字段(1 字节对齐)集中放在末尾,避免散落导致的 7 字节 padding
  • 优化后结构体大小:360 → 352 字节(每实例节省 8 字节,海量消息场景下可观)
  • 热点字段(ID/MessageType/Sender/Receiver)保持在结构体头部,位于首个 cache line 内

并发安全说明:

  • mu 为 *sync.RWMutex 指针(非值字段),避免 Clone/浅拷贝时的 go vet copylocks
  • 所有 Set*/With* 写方法持 Lock,所有 Get* 读方法持 RLock
  • 内部 setMapValue/getMapValue 不持锁,由调用方(With*/Get*)持锁保护,避免 RLock 不可重入导致的死锁
  • Clone 持 RLock 深拷贝 Data,副本使用独立 mu(避免与原对象共享锁互相阻塞)
  • mu 为 nil(如直接 &HubMessage{} 构造)时退化为无锁,兼容测试与零值构造场景

func NewHubMessage

func NewHubMessage() *HubMessage

NewHubMessage 创建通用消息结构

func (*HubMessage) Clone

func (m *HubMessage) Clone() *HubMessage

Clone 创建消息的深拷贝,避免并发修改问题 实现 syncx.Cloner 接口,DeepCopy 调用时走零反射快速路径 持 RLock 读取全部字段(与 Set*/With* 的 Lock 互斥),Data map 深拷贝(含嵌套子 map) 副本使用独立 mu(避免与原对象共享锁互相阻塞)

func (*HubMessage) CloneDeep

func (m *HubMessage) CloneDeep() any

CloneDeep 实现 syncx.Cloner 接口 DeepCopy(&dst, src) 检测到 src 实现 Cloner 时,直接调用此方法跳过反射

func (*HubMessage) CloneForFanout

func (m *HubMessage) CloneForFanout() *HubMessage

CloneForFanout 扇出克隆:标量信封字段独立可写,Data 与 GroupIDs 只读共享

适用场景为群组可靠投递/批量发送/离线回放/P2P 发送等逐成员扇出,仅改写 Receiver/ReceiverNode/ID/CreateAt/Source 等信封标量字段 Data 与 GroupIDs 全链路只读,所有读方(本地投递 Marshal/落库 SetMessage/跨节点 Publish/离线转存)均同步序列化 依据:深拷贝的 map/slice 重建是 10w 级扇出的 GC 热点,共享底层结构消除该开销

契约:扇出进行期间调用方不得并发修改原消息的 Data/GroupIDs 现状扇出均为同步 Execute,扇出完成后才可能触碰原消息,天然满足

func (*HubMessage) ContextFrom

func (m *HubMessage) ContextFrom(parent context.Context) context.Context

ContextFrom 基于消息的 trace_id 创建一个携带 trace 信息的 context 用于消息流转路径中恢复 ctx(如 PubSub 消费端、回调等场景) ctx 已有 trace 时不覆盖(尊重上游)

func (*HubMessage) ContextWithRoute

func (m *HubMessage) ContextWithRoute(parent context.Context, appID, ns string, groupIDs []string) context.Context

ContextWithRoute 将给定 appID+namespace+groupIDs + trace_id 同时写入 msg 信封和 ctx

与 InjectRoute 的区别:路由参数显式传入(不从 ctx 取),适用于 Deliver 等带 命名空间参数的公开 API;InjectRoute 适用于从 ctx 取路由的内部发送链路

入口层首选;没有上下文时单独调用 SetAppID/SetNamespace/SetGroupIDs 空值 appID 归一化为 DefaultAppID;namespace 保持原值(广播场景可传空)

func (*HubMessage) FirstGroupID

func (m *HubMessage) FirstGroupID() string

FirstGroupID 获取第一个群组ID(单群组场景用;多群组/无群组返回空)

func (*HubMessage) GetAllContentExtra

func (m *HubMessage) GetAllContentExtra() map[string]any

GetAllContentExtra 获取整个 content_extra map

func (*HubMessage) GetAllMetadata

func (m *HubMessage) GetAllMetadata() map[string]any

GetAllMetadata 获取整个 metadata map

func (*HubMessage) GetAppID

func (m *HubMessage) GetAppID() string

GetAppID 获取应用ID(空值返回空串,不补默认值,保持「空=全局共享」语义)

func (*HubMessage) GetClassification

func (m *HubMessage) GetClassification() *MessageClassification

GetClassification 获取消息综合分类(nil 安全)

func (*HubMessage) GetContentExtra

func (m *HubMessage) GetContentExtra(key string) (any, bool)

GetContentExtra 获取 content_extra 字段值

func (*HubMessage) GetContentExtraJSON

func (m *HubMessage) GetContentExtraJSON() string

GetContentExtraJSON 获取 content_extra 的 JSON 字符串表示

func (*HubMessage) GetGroupIDs

func (m *HubMessage) GetGroupIDs() []string

GetGroupIDs 获取群组ID列表(返回副本,避免外部修改污染内部)

func (*HubMessage) GetMediaInfo

func (m *HubMessage) GetMediaInfo() (any, bool)

GetMediaInfo 获取媒体信息(返回原始值)

func (*HubMessage) GetMediaInfoJSON

func (m *HubMessage) GetMediaInfoJSON() string

GetMediaInfoJSON 获取媒体信息的 JSON 字符串表示

func (*HubMessage) GetMessageID

func (m *HubMessage) GetMessageID() string

GetMessageID 获取消息ID,空值返回默认消息ID

func (*HubMessage) GetMetadata

func (m *HubMessage) GetMetadata(key string) (string, bool)

GetMetadata 获取 metadata 字段值 安全:非 string 类型返回 ("", false),不再 panic

func (*HubMessage) GetMetadataJSON

func (m *HubMessage) GetMetadataJSON() string

GetMetadataJSON 获取 metadata 的 JSON 字符串表示

func (*HubMessage) GetNamespace

func (m *HubMessage) GetNamespace() string

GetNamespace 获取命名空间

func (*HubMessage) GetOption

func (m *HubMessage) GetOption(key string) (interface{}, bool)

GetOption 获取扩展数据选项

func (*HubMessage) GetTraceID

func (m *HubMessage) GetTraceID() string

GetTraceID 获取 trace_id,优先自身字段,fallback 到 metadata

func (*HubMessage) InjectContext

func (m *HubMessage) InjectContext(ctx context.Context) *HubMessage

InjectContext 从 ctx 注入上下文信息到消息(trace_id 等) 优先从 OTel span 提取 trace_id,fallback 到 ctx.Value(logger.ContextKeyTraceID) 已有 trace_id 时不覆盖(跨节点消息保留源 trace)

func (*HubMessage) InjectRoute

func (m *HubMessage) InjectRoute(ctx context.Context) context.Context

InjectRoute 从 ctx 提取路由元数据 + trace_id 注入信封(入口层统一调用,一次性写入 msg+ctx)

一个方法搞定四件事(广播与 P2P 共用同一套路由提取逻辑,无特殊分支):

  1. 注入 trace_id:已有不覆盖(跨节点消息保留源 trace),优先 OTel span,fallback ctx.Value
  2. 归一化 appID:空值补 DefaultAppID(appID 无全局语义,必填)
  3. 注入信封:已有值不覆盖(跨节点消息保留源路由),空值从 ctx 补;namespace 保持 ctx 原值
  4. 回写 ctx:保证下游 routing.FromContext 与信封一致

namespace 不归一化的设计意图(一套逻辑打通广播与 P2P):

  • 全局广播:ctx 无 namespace → msg.Namespace="" → ClientMatchesEnvelope 跳过 ns 过滤匹配所有
  • 命名空间广播/P2P:ctx 有 namespace → msg.Namespace=ns → 严格匹配
  • P2P 严格场景需 namespace 归一化时,调用方在入口 EnsureRouteDefaults 或 NormalizeRoute

func (*HubMessage) MarshalJSON

func (m *HubMessage) MarshalJSON() ([]byte, error)

MarshalJSON 自定义序列化:持 RLock 贯穿 json.Marshal,与所有 Set*/With* 的 Lock 互斥 消除反射遍历 Data map 与并发写入导致的 "concurrent map iteration and map write" fatal

func (*HubMessage) ObserverCopy

func (m *HubMessage) ObserverCopy() (*HubMessage, []byte, error)

ObserverCopy 构建观察者投递专用信封(观察者投递统一入口共用): Clone 副本 + 旁路标记(observer_mode / 原始收发双方)+ 预序列化一次

投递语义(旁路投递,非主投递链路):

  • Clone 保证不污染原消息(观察者收到的副本与原消息互不影响)
  • 旁路标记保证接收方能区分观察者副本与直投消息
  • 预序列化一次:所有 WS 观察者复用同一份字节,SSE 观察者复用同一份对象,消除逐观察者 Clone+Marshal

func (*HubMessage) ResolveGuarantee

func (m *HubMessage) ResolveGuarantee() DeliveryGuarantee

ResolveGuarantee 解析消息的最终送达分级

决策树(优先级从高到低):

  1. 显式设置(WithGuarantee / 字段赋值,非 Unset)
  2. 消息类型默认表(guaranteeTypeDefaults)
  3. Classification 评分 ≥ ClassificationGuaranteedScore → 必达 (消费 SendToUserWithClassification 写入 Data["classification"] 的打分,让分类系统从 dead data 变活)
  4. PriorityCritical → 必达(与现有优先级体系打通)
  5. 兜底普通档

并发安全:持读锁一次性完成全部字段读取(不调用其他持锁方法——锁不可重入)

func (*HubMessage) RouteContext

func (m *HubMessage) RouteContext(parent context.Context) context.Context

RouteContext 从信封恢复路由 ctx(异步队列/跨节点/离线回放场景,ctx 已丢失时重建) 如果信封为空则返回原 ctx;非空则归一化 appID 后重建 ctx

func (*HubMessage) SetAppID

func (m *HubMessage) SetAppID(appID string) *HubMessage

SetAppID 设置应用ID(链式调用,最上层隔离维度)

func (*HubMessage) SetBroadcastType

func (m *HubMessage) SetBroadcastType(broadcastType BroadcastType) *HubMessage

SetBroadcastType 设置广播类型

func (*HubMessage) SetContent

func (m *HubMessage) SetContent(content string) *HubMessage

SetContent 设置消息内容

func (*HubMessage) SetGroupIDs

func (m *HubMessage) SetGroupIDs(groupIDs []string) *HubMessage

SetGroupIDs 设置群组ID列表(链式调用,P2P 传 nil)

func (*HubMessage) SetID

func (m *HubMessage) SetID(id string) *HubMessage

SetID 设置消息ID

func (*HubMessage) SetMessageID

func (m *HubMessage) SetMessageID(messageID string) *HubMessage

SetMessageID 设置业务消息ID

func (*HubMessage) SetMessageType

func (m *HubMessage) SetMessageType(messageType MessageType) *HubMessage

SetMessageType 设置消息类型

func (*HubMessage) SetNamespace

func (m *HubMessage) SetNamespace(ns string) *HubMessage

SetNamespace 设置命名空间(链式调用)

func (*HubMessage) SetPriority

func (m *HubMessage) SetPriority(priority Priority) *HubMessage

SetPriority 设置优先级

func (*HubMessage) SetPushType

func (m *HubMessage) SetPushType(pushType PushType) *HubMessage

SetPushType 设置推送类型

func (*HubMessage) SetReceiver

func (m *HubMessage) SetReceiver(receiver string) *HubMessage

SetReceiver 设置接收者

func (*HubMessage) SetReceiverClient

func (m *HubMessage) SetReceiverClient(clientID string) *HubMessage

SetReceiverClient 设置接收者客户端ID

func (*HubMessage) SetReceiverName

func (m *HubMessage) SetReceiverName(name string) *HubMessage

SetReceiverName 设置接收者昵称

func (*HubMessage) SetReceiverNode

func (m *HubMessage) SetReceiverNode(nodeID string) *HubMessage

SetReceiverNode 设置接收者所在节点ID

func (*HubMessage) SetReceiverType

func (m *HubMessage) SetReceiverType(receiverType UserType) *HubMessage

SetReceiverType 设置接收者用户类型

func (*HubMessage) SetReplyToMsgID

func (m *HubMessage) SetReplyToMsgID(replyToMsgID string) *HubMessage

SetReplyToMsgID 设置回复的消息ID

func (*HubMessage) SetRequireAck

func (m *HubMessage) SetRequireAck(requireAck bool) *HubMessage

SetRequireAck 设置是否需要ACK确认

func (*HubMessage) SetSender

func (m *HubMessage) SetSender(sender string) *HubMessage

SetSender 设置发送者

func (*HubMessage) SetSenderName

func (m *HubMessage) SetSenderName(name string) *HubMessage

SetSenderName 设置发送者昵称

func (*HubMessage) SetSenderType

func (m *HubMessage) SetSenderType(senderType UserType) *HubMessage

SetSenderType 设置发送者类型

func (*HubMessage) SetSeqNo

func (m *HubMessage) SetSeqNo(seqNo int64) *HubMessage

SetSeqNo 设置消息序列号

func (*HubMessage) SetSessionID

func (m *HubMessage) SetSessionID(sessionID string) *HubMessage

SetSessionID 设置会话ID

func (*HubMessage) SetSkipDatabaseStorage

func (m *HubMessage) SetSkipDatabaseStorage(skip bool) *HubMessage

SetSkipDatabaseStorage 设置是否跳过主数据库存储

func (*HubMessage) SetSkipSendToClient

func (m *HubMessage) SetSkipSendToClient(skip bool) *HubMessage

SetSkipSendToClient 设置是否跳过发送到客户端

func (*HubMessage) TraceContext

func (m *HubMessage) TraceContext(parent context.Context) context.Context

TraceContext 以消息信封的 trace_id 为准强制覆盖 ctx(离线回放场景专用) 与 ContextFrom 的区别:ContextFrom 尊重 ctx 已有 trace 不覆盖,本方法以消息原始 trace 优先 离线消息在用户上线时推送,ctx 来自上线连接(带连接自身的 trace_id) 强制覆盖为消息原始发送链路的 trace_id,保证同一条消息从 NotifySend 入口 → 离线暂存 → 上线推送 → 最终投递全程同一个 trace_id 可查

func (*HubMessage) WithAllContentExtra

func (m *HubMessage) WithAllContentExtra(contentExtra any) *HubMessage

WithAllContentExtra 批量设置 content_extra(接受任意类型,自动序列化,nil 安全)

func (*HubMessage) WithAllMetadata

func (m *HubMessage) WithAllMetadata(metadata any) *HubMessage

WithAllMetadata 批量设置 metadata(接受任意类型,自动序列化,nil 安全)

func (*HubMessage) WithClassification

func (m *HubMessage) WithClassification(classification *MessageClassification) *HubMessage

WithClassification 设置消息综合分类(链式) 分类评分会被 ResolveGuarantee 消费(≥90 提升为必达级),也保留在 Data 中随消息序列化

func (*HubMessage) WithContentExtra

func (m *HubMessage) WithContentExtra(key string, value any) *HubMessage

WithContentExtra 设置单个 content_extra 字段

func (*HubMessage) WithGuarantee

func (m *HubMessage) WithGuarantee(g DeliveryGuarantee) *HubMessage

WithGuarantee 链式设置显式送达分级(优先级最高,覆盖决策树推导)

func (*HubMessage) WithMediaInfo

func (m *HubMessage) WithMediaInfo(mediaInfo any) *HubMessage

WithMediaInfo 设置媒体信息(接受任意类型,自动序列化为 JSON 字符串,nil 安全)

func (*HubMessage) WithMetadata

func (m *HubMessage) WithMetadata(key string, value string) *HubMessage

WithMetadata 设置单个 metadata 字段

func (*HubMessage) WithOption

func (m *HubMessage) WithOption(key string, value interface{}) *HubMessage

WithOption 设置扩展数据选项

type HubStats

type HubStats struct {
	// 连接统计
	TotalClients     int64 `json:"total_clients"`     // 总客户端数
	WebSocketClients int64 `json:"websocket_clients"` // WebSocket客户端数
	SSEClients       int64 `json:"sse_clients"`       // SSE客户端数
	AgentConnections int64 `json:"agent_connections"` // 座席连接数

	// 消息统计
	MessagesSent     int64 `json:"messages_sent"`     // 已发送消息数
	MessagesReceived int64 `json:"messages_received"` // 已接收消息数
	BroadcastsSent   int64 `json:"broadcasts_sent"`   // 已发送广播数
	QueuedMessages   int   `json:"queued_messages"`   // 排队消息数

	// 其他统计
	OnlineUsers int   `json:"online_users"` // 在线用户数
	Uptime      int64 `json:"uptime"`       // 运行时间(秒)
}

HubStats Hub统计信息结构体

type IDGenerator

type IDGenerator interface {
	GenerateTraceID() string
	GenerateSpanID() string
	GenerateRequestID() string
	GenerateCorrelationID() string
}

IDGenerator ID生成器接口 用于生成消息ID、请求ID等唯一标识符

type KickUserResult

type KickUserResult struct {
	Success           bool // 是否成功执行(幂等:用户已不在线亦为 true)
	NotificationSent  bool // 踢出通知是否已写入发送通道(sendNotification=true 时)
	KickedConnections int  // 本节点触发注销的连接数(0=用户已不在线,幂等达成)
}

KickUserResult 统一踢人结果(Hub.KickUser 单一入口返回)

幂等语义:踢人目标是"用户离线",用户已无连接(收集数 0)即目标达成—— Success=true 且 KickedConnections=0 表示"本来就不在线",调用方据此与"真踢到"(KickedConnections>0)区分(跨节点分发为异步,本结果仅反映本节点信封内踢出数)

type MessageClassification

type MessageClassification struct {
	Type             MessageType      `json:"type"`
	Priority         MessagePriority  `json:"priority"`
	VIPLevel         VIPLevel         `json:"vip_level"`
	UrgencyLevel     UrgencyLevel     `json:"urgency_level"`
	BusinessCategory BusinessCategory `json:"business_category"`
	Timestamp        int64            `json:"timestamp"`
	UserID           string           `json:"user_id,omitempty"`
}

MessageClassification 消息综合分类

func (*MessageClassification) GetDisplayName

func (mc *MessageClassification) GetDisplayName() string

GetDisplayName 获取分类显示名称

func (*MessageClassification) GetFinalPriority

func (mc *MessageClassification) GetFinalPriority() int

GetFinalPriority 获取综合优先级分数 (0-100)

func (*MessageClassification) IsCriticalMessage

func (mc *MessageClassification) IsCriticalMessage() bool

IsCriticalMessage 判断是否为关键消息

func (*MessageClassification) IsHighPriorityMessage

func (mc *MessageClassification) IsHighPriorityMessage() bool

IsHighPriorityMessage 判断是否为高优先级消息

type MessagePriority

type MessagePriority int

MessagePriority 消息优先级

const (
	// 低优先级:一般信息、日志、统计数据等
	MessagePriorityLow MessagePriority = iota + 1
	// 普通优先级:常规用户消息、文件传输等
	MessagePriorityNormal
	// 高优先级:重要通知、系统消息等
	MessagePriorityHigh
	// 紧急优先级:安全警告、错误信息等
	MessagePriorityUrgent
	// 关键优先级:系统故障、紧急维护等
	MessagePriorityCritical
)

func (MessagePriority) GetWeight

func (p MessagePriority) GetWeight() int

GetWeight 获取优先级权重值(数值越大优先级越高)

func (MessagePriority) IsHigherThan

func (p MessagePriority) IsHigherThan(other MessagePriority) bool

IsHigherThan 判断是否比另一个优先级高

func (MessagePriority) String

func (p MessagePriority) String() string

String 实现Stringer接口

type MessageRecordFilter

type MessageRecordFilter struct {
	// Status 按状态查询(可选)
	Status *MessageSendStatus
	// Sender 按发送者查询(可选)
	Sender string
	// Receiver 按接收者查询(可选)
	Receiver string
	// NodeIP 按节点IP查询(可选)
	NodeIP string
	// ClientIP 按客户端IP查询(可选)
	ClientIP string
	// Limit 查询数量限制
	Limit int
	// OrderDesc 是否降序排序(默认降序,false为升序)
	OrderDesc bool
}

MessageRecordFilter 消息记录查询过滤器

type MessageRecordKey

type MessageRecordKey struct {
	// MessageID 消息ID(全局唯一,群发场景同一消息的所有成员记录共享)
	MessageID string
	// Receiver 接收者ID(广播类记录为空串)
	Receiver string
}

MessageRecordKey 消息记录复合键(MessageSink 定位/更新/认领的统一寻址维度)

P2P 场景下同一 message_id 会为每个 receiver 各建一条记录(群组可靠投递 per-member 落库),仅按 message_id 寻址会造成多 receiver 状态相互覆盖 (如 A 的 success 覆盖 B 的 failed)。所有按记录粒度的状态操作 (UpdateStatus/ClaimStaleSending/IncrementRetry/FindByMessageID)均以复合键精确定位。 广播类记录 Receiver 为空串,同样满足复合键等值匹配。

type MessageSendRecord

type MessageSendRecord struct {
	ID            uint                           `gorm:"primaryKey;autoIncrement;comment:主键,唯一标识每条发送记录" json:"id"`                                        // 主键(唯一)
	SessionID     string                         `gorm:"column:session_id;size:255;not null;index;comment:会话ID" json:"session_id"`                        // 会话ID
	MessageID     string                         `gorm:"index:idx_message_id_receiver;size:255;not null;comment:业务消息ID,用于关联业务系统" json:"message_id"`       // 业务消息ID(可重复,支持多次发送记录)
	HubID         string                         `gorm:"index;size:255;not null;comment:Hub内部消息ID,用于ACK确认和日志追踪" json:"hub_id"`                            // Hub内部消息ID(可重复)
	Namespace     string                         `gorm:"index;size:128;not null;default:'';comment:命名空间,用于多租户/业务隔离" json:"namespace"`                     // 命名空间
	GroupID       string                         `gorm:"index;size:255;not null;default:'';comment:业务群组ID,P2P消息为空" json:"group_id"`                       // 业务群组ID
	MessageData   string                         `gorm:"type:text;comment:原始消息数据,类型为文本" json:"message_data"`                                              // 原始消息数据
	Sender        string                         `gorm:"index;size:255;comment:发送者ID" json:"sender"`                                                      // 发送者ID
	Receiver      string                         `gorm:"index:idx_message_id_receiver;size:255;comment:接收者ID" json:"receiver"`                            // 接收者ID
	MessageType   MessageType                    `gorm:"index;size:50;comment:消息类型" json:"message_type"`                                                  // 消息类型
	Source        MessageSource                  `gorm:"index;size:20;not null;default:'online';comment:消息来源(online/offline)" json:"source"`              // 消息来源
	NodeIP        string                         `gorm:"index;size:100;comment:服务器节点IP地址" json:"node_ip"`                                                 // 服务器节点IP地址
	ClientIP      string                         `gorm:"index;size:100;comment:客户端IP地址" json:"client_ip"`                                                 // 客户端IP地址
	Status        MessageSendStatus              `gorm:"index:idx_status_create_time;size:50;not null;default:'pending';comment:当前状态,不能为空" json:"status"` // 当前状态
	CreateTime    time.Time                      `gorm:"index:idx_status_create_time;not null;comment:创建时间" json:"create_time"`                           // 创建时间
	FirstSendTime *time.Time                     `gorm:"index;comment:首次发送时间" json:"first_send_time"`                                                     // 首次发送时间
	LastSendTime  *time.Time                     `gorm:"index;comment:最后发送时间" json:"last_send_time"`                                                      // 最后发送时间
	SuccessTime   *time.Time                     `gorm:"index;comment:成功时间" json:"success_time"`                                                          // 成功时间
	RetryCount    int                            `gorm:"default:0;comment:重试次数" json:"retry_count"`                                                       // 重试次数
	MaxRetry      int                            `gorm:"default:3;comment:最大重试次数" json:"max_retry"`                                                       // 最大重试次数
	FailureReason FailureReason                  `gorm:"size:50;comment:失败原因" json:"failure_reason"`                                                      // 失败原因
	ErrorMessage  string                         `gorm:"type:text;comment:错误信息,类型为文本" json:"error_message"`                                               // 错误信息
	RetryHistory  sqlbuilder.Slice[RetryAttempt] `gorm:"type:json;comment:重试历史,类型为JSON" json:"retry_history"`                                             // 重试历史
	ExpiresAt     *time.Time                     `gorm:"index;comment:过期时间" json:"expires_at"`                                                            // 过期时间
	Metadata      sqlbuilder.MapAny              `gorm:"type:json;comment:扩展元数据,类型为JSON" json:"metadata"`                                                 // 扩展元数据
	CustomFields  sqlbuilder.MapAny              `gorm:"type:json;comment:用户自定义字段,类型为JSON" json:"custom_fields"`                                          // 用户自定义字段
	Tags          sqlbuilder.StringSlice         `gorm:"type:json;comment:标签,类型为JSON" json:"tags"`                                                        // 标签
	ExtraData     string                         `gorm:"type:text;comment:额外数据,类型为文本" json:"extra_data"`                                                  // 额外数据
	CreatedAt     time.Time                      `gorm:"comment:记录创建时间" json:"created_at"`                                                                // 创建时间
	UpdatedAt     time.Time                      `gorm:"comment:记录最后更新时间" json:"updated_at"`                                                              // 最后更新时间
}

MessageSendRecord 消息发送记录(GORM 模型)

func (*MessageSendRecord) BeforeCreate

func (m *MessageSendRecord) BeforeCreate(tx *gorm.DB) error

BeforeCreate GORM 钩子:创建前

func (*MessageSendRecord) GetMessage

func (m *MessageSendRecord) GetMessage() (*HubMessage, error)

GetMessage 获取消息数据(反序列化 HubMessage)

func (*MessageSendRecord) SetMessage

func (m *MessageSendRecord) SetMessage(msg *HubMessage, payload []byte) error

SetMessage 设置消息数据(将 HubMessage 序列化存储)

payload 为调用方已序列化的消息体编码字节,可靠投递热路径与本地投递共享同一份编码,消除逐成员二次 Marshal nil 时内部自行序列化

func (MessageSendRecord) TableComment

func (MessageSendRecord) TableComment() string

TableComment 表注释

func (MessageSendRecord) TableName

func (MessageSendRecord) TableName() string

TableName 指定表名

type MessageSendStatus

type MessageSendStatus string

MessageSendStatus 消息发送状态

const (
	MessageSendStatusPending     MessageSendStatus = "pending"      // 待发送
	MessageSendStatusSending     MessageSendStatus = "sending"      // 发送中
	MessageSendStatusSuccess     MessageSendStatus = "success"      // 发送成功
	MessageSendStatusFailed      MessageSendStatus = "failed"       // 发送失败
	MessageSendStatusRetrying    MessageSendStatus = "retrying"     // 重试中
	MessageSendStatusAckTimeout  MessageSendStatus = "ack_timeout"  // ACK超时
	MessageSendStatusUserOffline MessageSendStatus = "user_offline" // 用户离线
	MessageSendStatusExpired     MessageSendStatus = "expired"      // 已过期
)

type MessageSource

type MessageSource string

MessageSource 消息来源类型

const (
	MessageSourceOnline  MessageSource = "online"  // 实时在线发送
	MessageSourceOffline MessageSource = "offline" // 离线消息推送
)

type MessageStatus

type MessageStatus string

MessageStatus 消息状态

const (
	MessageStatusPending   MessageStatus = "pending"   // 待发送
	MessageStatusSent      MessageStatus = "sent"      // 已发送
	MessageStatusDelivered MessageStatus = "delivered" // 已送达
	MessageStatusRead      MessageStatus = "read"      // 已读
	MessageStatusFailed    MessageStatus = "failed"    // 发送失败
)

func (MessageStatus) IsValid

func (s MessageStatus) IsValid() bool

IsValid 检查消息状态是否有效

func (MessageStatus) String

func (s MessageStatus) String() string

String 实现Stringer接口

type MessageType

type MessageType string

MessageType 消息类型

const (
	MessageTypeText                 MessageType = "text"                   // 文本消息
	MessageTypeImage                MessageType = "image"                  // 图片消息
	MessageTypeFile                 MessageType = "file"                   // 文件消息
	MessageTypeAudio                MessageType = "audio"                  // 音频消息
	MessageTypeVideo                MessageType = "video"                  // 视频消息
	MessageTypeSystem               MessageType = "system"                 // 系统消息
	MessageTypeNotice               MessageType = "notice"                 // 通知消息
	MessageTypeEvent                MessageType = "event"                  // 事件消息
	MessageTypeAck                  MessageType = "ack"                    // ACK确认消息
	MessageTypeLocation             MessageType = "location"               // 位置消息
	MessageTypeCard                 MessageType = "card"                   // 卡片消息
	MessageTypeEmoji                MessageType = "emoji"                  // 表情消息
	MessageTypeSticker              MessageType = "sticker"                // 贴纸消息
	MessageTypeLink                 MessageType = "link"                   // 链接消息
	MessageTypeQuote                MessageType = "quote"                  // 引用回复消息
	MessageTypeForward              MessageType = "forward"                // 转发消息
	MessageTypeCommand              MessageType = "command"                // 命令消息
	MessageTypeMarkdown             MessageType = "markdown"               // Markdown格式消息
	MessageTypeRichText             MessageType = "rich_text"              // 富文本消息
	MessageTypeCode                 MessageType = "code"                   // 代码消息
	MessageTypeJSON                 MessageType = "json"                   // JSON数据消息
	MessageTypeXML                  MessageType = "xml"                    // XML数据消息
	MessageTypeBinary               MessageType = "binary"                 // 二进制数据消息
	MessageTypeVoice                MessageType = "voice"                  // 语音消息
	MessageTypeGIF                  MessageType = "gif"                    // GIF动图消息
	MessageTypeDocument             MessageType = "document"               // 文档消息
	MessageTypeSpreadsheet          MessageType = "spreadsheet"            // 电子表格消息
	MessageTypePresentation         MessageType = "presentation"           // 演示文稿消息
	MessageTypeContact              MessageType = "contact"                // 联系人卡片消息
	MessageTypeCalendar             MessageType = "calendar"               // 日历事件消息
	MessageTypeTask                 MessageType = "task"                   // 任务消息
	MessageTypePoll                 MessageType = "poll"                   // 投票消息
	MessageTypeForm                 MessageType = "form"                   // 表单消息
	MessageTypePayment              MessageType = "payment"                // 支付消息
	MessageTypeOrder                MessageType = "order"                  // 订单消息
	MessageTypeProduct              MessageType = "product"                // 产品消息
	MessageTypeInvite               MessageType = "invite"                 // 邀请消息
	MessageTypeAnnouncement         MessageType = "announcement"           // 公告消息
	MessageTypeAlert                MessageType = "alert"                  // 警告消息
	MessageTypeError                MessageType = "error"                  // 错误消息
	MessageTypeInfo                 MessageType = "info"                   // 信息消息
	MessageTypeSuccess              MessageType = "success"                // 成功消息
	MessageTypeWarning              MessageType = "warning"                // 警告消息
	MessageTypeHeartbeat            MessageType = "heartbeat"              // 心跳消息
	MessageTypePing                 MessageType = "ping"                   // Ping消息
	MessageTypePong                 MessageType = "pong"                   // Pong消息
	MessageTypeTyping               MessageType = "typing"                 // 正在输入状态消息
	MessageTypeRead                 MessageType = "read"                   // 已读消息
	MessageTypeDelivered            MessageType = "delivered"              // 已送达消息
	MessageTypeRecall               MessageType = "recall"                 // 消息撤回
	MessageTypeEdit                 MessageType = "edit"                   // 消息编辑
	MessageTypeReaction             MessageType = "reaction"               // 消息反应/表态
	MessageTypeThread               MessageType = "thread"                 // 线程消息
	MessageTypeReply                MessageType = "reply"                  // 回复消息
	MessageTypeMention              MessageType = "mention"                // @提及消息
	MessageTypeCustom               MessageType = "custom"                 // 自定义类型消息
	MessageTypeUnknown              MessageType = "unknown"                // 未知类型消息
	MessageTypeTicketCreated        MessageType = "ticket_created"         // 工单创建消息
	MessageTypeTicketAssigned       MessageType = "ticket_assigned"        // 分配工单消息
	MessageTypeTicketClosed         MessageType = "ticket_closed"          // 手动关闭工单消息
	MessageTypeTicketTimeoutClosed  MessageType = "ticket_timeout_closed"  // 超时关闭工单消息
	MessageTypeTicketTransfer       MessageType = "ticket_transfer"        // 转移工单消息
	MessageTypeTicketActive         MessageType = "ticket_active"          // 活跃工单列表
	MessageTypeTest                 MessageType = "test"                   // 测试消息
	MessageTypeWelcome              MessageType = "welcome"                // 欢迎消息
	MessageTypeTerminate            MessageType = "terminate"              // 结束消息
	MessageTypeTransferred          MessageType = "transferred"            // 转发消息
	MessageTypeSessionCreated       MessageType = "session_created"        // 会话创建消息
	MessageTypeSessionClosed        MessageType = "session_closed"         // 会话关闭消息
	MessageTypeSessionQueued        MessageType = "session_queued"         // 会话排队消息
	MessageTypeSessionTimeout       MessageType = "session_timeout"        // 会话超时消息
	MessageTypeSessionPaused        MessageType = "session_paused"         // 会话暂停消息
	MessageTypeSessionResumed       MessageType = "session_resumed"        // 会话恢复消息
	MessageTypeSessionTransferred   MessageType = "session_transferred"    // 会话转接消息
	MessageTypeSessionMemberJoined  MessageType = "session_member_joined"  // 会话成员加入消息
	MessageTypeSessionMemberLeft    MessageType = "session_member_left"    // 会话成员离开消息
	MessageTypeSessionStatusChanged MessageType = "session_status_changed" // 会话状态变更消息
	MessageTypeCheckUserStatus      MessageType = "check_user_status"      // 检查用户状态消息
	MessageTypeUserStatusResponse   MessageType = "user_status_response"   // 用户状态响应消息
	MessageTypeGetOnlineUsers       MessageType = "get_online_users"       // 获取在线用户列表
	MessageTypeOnlineUsersList      MessageType = "online_users_list"      // 在线用户列表响应
	MessageTypeGetUserInfo          MessageType = "get_user_info"          // 获取用户信息
	MessageTypeUserInfoResponse     MessageType = "user_info_response"     // 用户信息响应
	MessageTypeSystemQuery          MessageType = "system_query"           // 系统查询消息
	MessageTypeSystemResponse       MessageType = "system_response"        // 系统响应消息
	MessageTypeUserJoined           MessageType = "user_joined"            // 用户加入通知
	MessageTypeUserLeft             MessageType = "user_left"              // 用户离开通知
	MessageTypeUserStatusChanged    MessageType = "user_status_changed"    // 用户状态变更通知
	MessageTypeServerStatus         MessageType = "server_status"          // 服务器状态消息
	MessageTypeServerStats          MessageType = "server_stats"           // 服务器统计消息
	MessageTypeClientConfig         MessageType = "client_config"          // 客户端配置消息
	MessageTypeConfigUpdate         MessageType = "config_update"          // 配置更新通知
	MessageTypeHealthCheck          MessageType = "health_check"           // 健康检查消息
	MessageTypeHealthResponse       MessageType = "health_response"        // 健康检查响应
	MessageTypeConnected            MessageType = "connected"              // 连接成功消息(发给刚连接的客户端)
	MessageTypeClientRegistered     MessageType = "client_registered"      // 客户端注册成功消息(连接建立并注册完成)
	MessageTypeConnectionRejected   MessageType = "connection_rejected"    // 连接被拒绝消息(参数验证失败)
	MessageTypeDisconnected         MessageType = "disconnected"           // 用户断线通知(广播给会话其他成员)
	MessageTypeReconnected          MessageType = "reconnected"            // 重连成功消息(发给重连的客户端)
	MessageTypeConnectionError      MessageType = "connection_error"       // 连接错误消息(发给连接失败的客户端)
	MessageTypeConnectionTimeout    MessageType = "connection_timeout"     // 连接超时消息(发给超时的客户端)
	MessageTypeKickOut              MessageType = "kick_out"               // 被踢出消息(发给被踢的客户端,之后断开)
	MessageTypeForceOffline         MessageType = "force_offline"          // 强制下线通知(异地登录等,发给被下线的客户端)
	MessageTypeOpenWindow           MessageType = "open_window"            // 打开窗口消息
	MessageTypeCloseWindow          MessageType = "close_window"           // 关闭窗口消息
	MessageTypeBStatusReminder      MessageType = "b_status_reminder"      // 业务状态提醒消息
	MessageTypeGroupChanged         MessageType = "group_changed"          // 群组变更通知(成员被 move 时下发,告知客户端已切换群组)
	MessageTypeStream               MessageType = "stream"                 // 高频流式聚合消息(弹幕/跑马灯/滚动通知;发送方用此类型声明消息可聚合,窗口攒批 N 条合 1 条数组帧,content 为 JSON 数组)
)

func GetAllMessageTypes

func GetAllMessageTypes() []MessageType

GetAllMessageTypes 返回所有可用的消息类型

func GetMessageTypesByCategory

func GetMessageTypesByCategory(category string) []MessageType

GetMessageTypesByCategory 根据分类返回消息类型

func GetMessageTypesByPriority

func GetMessageTypesByPriority(priority MessagePriority) []MessageType

GetMessageTypesByPriority 根据优先级获取消息类型列表

func (MessageType) GetCategory

func (t MessageType) GetCategory() string

GetCategory 获取消息类型分类(扩展)

func (MessageType) GetDefaultPriority

func (t MessageType) GetDefaultPriority() MessagePriority

GetDefaultPriority 根据消息类型获取默认优先级

func (MessageType) GetEmoji

func (t MessageType) GetEmoji() string

GetEmoji 获取消息类型对应的 emoji,未找到返回默认值

func (MessageType) IsBusinessType

func (t MessageType) IsBusinessType() bool

IsBusinessType 检查是否为业务相关类型消息

func (MessageType) IsConfigType

func (t MessageType) IsConfigType() bool

IsConfigType 检查是否为配置相关类型消息

func (MessageType) IsConnectionType

func (t MessageType) IsConnectionType() bool

IsConnectionType 检查是否为连接相关类型消息

func (MessageType) IsCustomType

func (t MessageType) IsCustomType() bool

IsCustomType 检查是否为自定义/未知类型

func (MessageType) IsForwardableType

func (t MessageType) IsForwardableType() bool

IsForwardableType 检查是否为需要转发的消息类型 可转发消息:客户端发送的消息需要直接转发给接收者,不需要经过业务处理 包括:窗口消息(打开/关闭)、状态消息(输入状态、已读、送达、确认、反应、编辑、撤回)等需要实时转发的消息

func (MessageType) IsHealthType

func (t MessageType) IsHealthType() bool

IsHealthType 检查是否为健康检查相关类型消息

func (MessageType) IsHeartbeatType

func (t MessageType) IsHeartbeatType() bool

IsHeartbeatType 检查是否为心跳类型消息

func (MessageType) IsInteractiveType

func (t MessageType) IsInteractiveType() bool

IsInteractiveType 检查是否为交互类型消息

func (MessageType) IsMediaType

func (t MessageType) IsMediaType() bool

IsMediaType 检查是否为媒体类型消息

func (MessageType) IsRecallType

func (t MessageType) IsRecallType() bool

IsRecallType 检查是否为消息撤回/编辑/反应等类型

func (MessageType) IsServerType

func (t MessageType) IsServerType() bool

IsServerType 检查是否为服务器相关类型消息

func (MessageType) IsSessionType

func (t MessageType) IsSessionType() bool

IsSessionType 检查是否为会话类型消息

func (MessageType) IsStatusType

func (t MessageType) IsStatusType() bool

IsStatusType 检查是否为状态类型消息

func (MessageType) IsSystemType

func (t MessageType) IsSystemType() bool

IsSystemType 检查是否为系统类型消息

func (MessageType) IsTextType

func (t MessageType) IsTextType() bool

IsTextType 检查是否为文本类型消息

func (MessageType) IsThreadType

func (t MessageType) IsThreadType() bool

IsThreadType 检查是否为线程/回复相关类型

func (MessageType) IsUserType

func (t MessageType) IsUserType() bool

IsUserType 检查是否为用户相关类型消息

func (MessageType) IsValid

func (t MessageType) IsValid() bool

IsValid 检查消息类型是否有效

func (MessageType) IsWindowType

func (t MessageType) IsWindowType() bool

IsWindowType 检查是否为窗口相关类型消息

func (MessageType) ShouldSkipDatabaseRecord

func (t MessageType) ShouldSkipDatabaseRecord() bool

ShouldSkipDatabaseRecord 判断是否应该跳过数据库记录 某些消息类型(如心跳消息、状态消息、系统消息)频繁发送但不需要持久化,跳过记录可以减轻数据库压力

func (MessageType) String

func (t MessageType) String() string

String 实现Stringer接口

type NamespaceObserverStats

type NamespaceObserverStats struct {
	Namespace     string               `json:"namespace"`      // 命名空间
	TotalUsers    int                  `json:"total_users"`    // 该命名空间下观察者用户数
	TotalDevices  int                  `json:"total_devices"`  // 该命名空间下观察者设备数
	GroupStats    []GroupObserverStats `json:"group_stats"`    // 按群组分组的观察者统计
	ObserverStats []*ObserverStats     `json:"observer_stats"` // 命名空间级(非群组级)观察者详细统计
}

NamespaceObserverStats 命名空间级观察者统计

type NodeConnectionStats

type NodeConnectionStats struct {
	NodeID                string  `json:"node_id"`                 // 节点ID
	NodeIP                string  `json:"node_ip"`                 // 节点IP
	NodePort              int     `json:"node_port"`               // 节点端口
	TotalConnections      int64   `json:"total_connections"`       // 总连接数
	ActiveConnections     int64   `json:"active_connections"`      // 活跃连接数
	DisconnectedCount     int64   `json:"disconnected_count"`      // 已断开连接数
	AbnormalCount         int64   `json:"abnormal_count"`          // 异常断开数
	AbnormalRate          float64 `json:"abnormal_rate"`           // 异常断开率(%)
	TotalMessagesSent     int64   `json:"total_messages_sent"`     // 拆表后零填充
	TotalMessagesReceived int64   `json:"total_messages_received"` // 拆表后零填充
	TotalBytesSent        int64   `json:"total_bytes_sent"`        // 拆表后零填充
	TotalBytesReceived    int64   `json:"total_bytes_received"`    // 拆表后零填充
	TotalErrors           int64   `json:"total_errors"`            // 拆表后零填充
	ErrorRate             float64 `json:"error_rate"`              // 拆表后零填充
	AveragePingMs         float64 `json:"average_ping_ms"`         // 拆表后零填充
	MaxPingMs             float64 `json:"max_ping_ms"`             // 拆表后零填充
	MinPingMs             float64 `json:"min_ping_ms"`             // 拆表后零填充
	AverageDuration       float64 `json:"average_duration"`        // 平均连接时长(秒)
	TotalReconnects       int64   `json:"total_reconnects"`        // 拆表后零填充
	AverageReconnectCount float64 `json:"average_reconnect_count"` // 拆表后零填充
	ConnectionQuality     float64 `json:"connection_quality"`      // 拆表后零填充
}

NodeConnectionStats 节点连接统计 质量维度字段(TotalMessages*/TotalBytes*/TotalErrors/ErrorRate/AveragePingMs/MaxPingMs/MinPingMs/ TotalReconnects/AverageReconnectCount/ConnectionQuality)拆表后由本方法零填充,跨表补充由调用方按需从 qualityRepo 取

type NodeInfo

type NodeInfo struct {
	ID          string     `json:"id"`
	IPAddress   string     `json:"ip_address"`
	Port        int        `json:"port"`
	Status      NodeStatus `json:"status"`
	LoadScore   float64    `json:"load_score"`
	LastSeen    time.Time  `json:"last_seen"`
	Connections int64      `json:"connections"`
}

NodeInfo 节点信息

type NodeStats

type NodeStats struct {
	NodeID            string    `json:"node_id" redis:"-"`
	TotalConnections  int64     `json:"total_connections" redis:"total_connections"`
	ActiveConnections int64     `json:"active_connections" redis:"active_connections"`
	MessagesSent      int64     `json:"messages_sent" redis:"messages_sent"`
	MessagesReceived  int64     `json:"messages_received" redis:"messages_received"`
	BroadcastsSent    int64     `json:"broadcasts_sent" redis:"broadcasts_sent"`
	StartTime         int64     `json:"start_time" redis:"start_time"`
	LastHeartbeat     time.Time `json:"last_heartbeat" redis:"-"`
	Uptime            int64     `json:"uptime" redis:"-"` // 运行时间(秒),计算字段
}

NodeStats 节点统计信息

type NodeStatus

type NodeStatus string

NodeStatus 节点状态

const (
	NodeStatusActive   NodeStatus = "active"   // 活跃
	NodeStatusInactive NodeStatus = "inactive" // 非活跃
	NodeStatusOffline  NodeStatus = "offline"  // 离线
)

func (NodeStatus) IsValid

func (s NodeStatus) IsValid() bool

IsValid 检查节点状态是否有效

func (NodeStatus) String

func (s NodeStatus) String() string

String 实现Stringer接口

type ObserverManagerStats

type ObserverManagerStats struct {
	TotalObservers      int                                `json:"total_observers"`      // 观察者用户总数
	TotalDevices        int                                `json:"total_devices"`        // 观察者设备总数
	TotalNotifications  int64                              `json:"total_notifications"`  // 总通知次数
	FailedNotifications int64                              `json:"failed_notifications"` // 失败通知次数
	DroppedMessages     int64                              `json:"dropped_messages"`     // 丢弃消息数(缓冲区满)
	NamespaceStats      map[string]*NamespaceObserverStats `json:"namespace_stats"`      // 按命名空间分组的观察者统计
}

ObserverManagerStats 观察者管理器统计信息

type ObserverStats

type ObserverStats struct {
	ObserverID  string    `json:"observer_id"`
	ClientID    string    `json:"client_id"`
	Namespace   string    `json:"namespace"`          // 命名空间
	GroupID     string    `json:"group_id,omitempty"` // 群组ID(空=命名空间级观察者)
	ConnectedAt time.Time `json:"connected_at"`
	BufferSize  int       `json:"buffer_size"`
	BufferUsage int       `json:"buffer_usage"`
	IsConnected bool      `json:"is_connected"`
	UserType    string    `json:"user_type"`
	ClientType  string    `json:"client_type"`
}

ObserverStats 观察者统计信息

type OfflineMessageRecord

type OfflineMessageRecord struct {
	ID             uint              `gorm:"primaryKey;autoIncrement;comment:主键,唯一标识离线消息记录" json:"id"`
	MessageID      string            `gorm:"column:message_id;size:64;not null;uniqueIndex:idx_message_app_ns_receiver;comment:业务消息ID" json:"message_id"`
	Sender         string            `gorm:"index;size:255;comment:发送者ID" json:"sender"`
	Receiver       string            `` /* 132-byte string literal not displayed */
	AppID          string            `` /* 219-byte string literal not displayed */
	Namespace      string            `` /* 172-byte string literal not displayed */
	GroupID        string            `` /* 136-byte string literal not displayed */
	SessionID      string            `gorm:"column:session_id;size:64;not null;index;comment:会话ID" json:"session_id"`
	CompressedData []byte            `gorm:"not null;comment:压缩完整的HubMessage JSON数据" json:"-"`
	Status         MessageSendStatus `gorm:"column:status;size:20;not null;default:'user_offline';index;comment:消息状态,复用MessageSendStatus" json:"status"`
	RetryCount     int               `gorm:"column:retry_count;default:0;comment:推送重试次数" json:"retry_count"`
	MaxRetry       int               `gorm:"column:max_retry;default:3;comment:最大重试次数" json:"max_retry"`
	ScheduledAt    time.Time         `gorm:"column:scheduled_at;not null;comment:消息计划发送的时间" json:"scheduled_at"`
	ExpireAt       time.Time         `gorm:"column:expire_at;not null;index;comment:消息过期时间" json:"expire_at"`
	FirstPushAt    *time.Time        `gorm:"column:first_push_at;index;comment:首次推送时间" json:"first_push_at,omitempty"`
	LastPushAt     *time.Time        `gorm:"column:last_push_at;index;comment:最后推送时间(成功或失败都更新)" json:"last_push_at,omitempty"`
	ErrorMessage   string            `gorm:"column:error_message;type:text;comment:推送失败的错误信息" json:"error_message,omitempty"`
	CreatedAt      time.Time         `gorm:"autoCreateTime;comment:记录创建时间" json:"created_at"`
	UpdatedAt      time.Time         `gorm:"autoUpdateTime;comment:记录最后更新时间" json:"updated_at"`
}

OfflineMessageRecord 离线消息记录 用于存储用户离线时接收到的消息信息,包含发送者、接收者及消息内容等关键数据 支持应用(app_id)+命名空间(namespace)+群组(group_id)三层隔离,类似 k8s 设计

func (*OfflineMessageRecord) GetAppID

func (r *OfflineMessageRecord) GetAppID() string

GetAppID 获取应用ID,空值返回默认应用ID

func (OfflineMessageRecord) TableComment

func (OfflineMessageRecord) TableComment() string

TableComment 表注释

func (OfflineMessageRecord) TableName

func (OfflineMessageRecord) TableName() string

TableName 指定表名

type OperationType

type OperationType string

OperationType 操作类型

const (
	OperationTypeJoin            OperationType = "join"             // 加入
	OperationTypeLeave           OperationType = "leave"            // 离开
	OperationTypeMessage         OperationType = "message"          // 消息
	OperationTypeBroadcast       OperationType = "broadcast"        // 广播
	OperationTypeNotify          OperationType = "notify"           // 通知
	OperationTypeHeartbeat       OperationType = "heartbeat"        // 心跳
	OperationTypeAuth            OperationType = "auth"             // 认证
	OperationTypeSync            OperationType = "sync"             // 同步
	OperationTypeSendMessage     OperationType = "send_message"     // 跨节点发送消息
	OperationTypeKickUser        OperationType = "kick_user"        // 跨节点踢人
	OperationTypeNodeRegister    OperationType = "node_register"    // 节点注册
	OperationTypeObserverNotify  OperationType = "observer_notify"  // 跨节点观察者通知
	OperationTypeGroupBroadcast  OperationType = "group_broadcast"  // 跨节点单群组广播
	OperationTypeGroupsBroadcast OperationType = "groups_broadcast" // 跨节点批量群组广播
	OperationTypeUserNotFound    OperationType = "user_not_found"   // 目标节点回告发送节点:用户不在该节点(索引死条目自愈信号)
	OperationTypeClientReclaim   OperationType = "client_reclaim"   // 新节点回收旧节点同 clientID 幽灵连接(断线重连跨节点迁移)
	OperationTypeGroupInvalidate OperationType = "group_invalidate" // 群拓扑失效广播:拓扑写路径经 100ms 聚合窗口批量发布,各节点逐出本地群成员缓存
)

func (OperationType) IsValid

func (o OperationType) IsValid() bool

IsValid 检查操作类型是否有效

func (OperationType) String

func (o OperationType) String() string

String 实现Stringer接口

type Priority

type Priority string

Priority 优先级类型

const (
	PriorityLow      Priority = "low"      // 低优先级
	PriorityNormal   Priority = "normal"   // 普通优先级
	PriorityHigh     Priority = "high"     // 高优先级
	PriorityUrgent   Priority = "urgent"   // 紧急优先级
	PriorityCritical Priority = "critical" // 关键优先级
)

func (Priority) IsValid

func (p Priority) IsValid() bool

IsValid 检查优先级是否有效

func (Priority) String

func (p Priority) String() string

String 实现Stringer接口

type PriorityStats

type PriorityStats struct {
	Critical int `json:"critical"` // 关键优先级数量
	Urgent   int `json:"urgent"`   // 紧急优先级数量
	High     int `json:"high"`     // 高优先级数量
	Normal   int `json:"normal"`   // 普通优先级数量
	Low      int `json:"low"`      // 低优先级数量
	Total    int `json:"total"`    // 总数量
}

PriorityStats 优先级统计

func GetPriorityStats

func GetPriorityStats() *PriorityStats

GetPriorityStats 获取所有消息类型的优先级统计

type PushType

type PushType string

PushType 推送类型

const (
	PushTypeNone    PushType = "none"    // 不推送 (仅保存/处理业务)
	PushTypeDirect  PushType = "direct"  // 在线直推 (尝试WebSocket,失败转离线)
	PushTypeQueue   PushType = "queue"   // 队列推送 (推送到Redis Stream)
	PushTypeOffline PushType = "offline" // 离线推送 (仅离线用户)
	PushTypeUnicast PushType = "unicast" // 单播推送 (仅在线用户)
)

func (PushType) IsValid

func (p PushType) IsValid() bool

IsValid 检查推送类型是否有效

func (PushType) String

func (p PushType) String() string

String 实现Stringer接口

type QueueType

type QueueType string

QueueType 队列类型

const (
	QueueTypeBroadcast    QueueType = "broadcast_full" // 广播队列
	QueueTypePending      QueueType = "pending_full"   // 待发送队列
	QueueTypeAllQueues    QueueType = "all_queues"     // 所有队列
	QueueTypeMessageQueue QueueType = "message_queue"  // 消息队列
	QueueTypeClientBuffer QueueType = "client_buffer"  // 客户端缓冲区
)

func (QueueType) IsValid

func (q QueueType) IsValid() bool

IsValid 检查队列类型是否有效

func (QueueType) String

func (q QueueType) String() string

String 实现Stringer接口

type RetryAttempt

type RetryAttempt struct {
	AttemptNumber int           `json:"attempt_number"` // 第几次重试
	Timestamp     time.Time     `json:"timestamp"`      // 重试时间
	Duration      time.Duration `json:"duration"`       // 本次重试耗时
	Error         string        `json:"error"`          // 错误信息
	Success       bool          `json:"success"`        // 是否成功
}

RetryAttempt 重试记录

type SendAttempt

type SendAttempt struct {
	AttemptNumber int
	StartTime     time.Time
	Duration      time.Duration
	Error         error
	Success       bool
}

SendAttempt 发送尝试记录

type SendResult

type SendResult struct {
	Success       bool
	Attempts      []SendAttempt
	TotalRetries  int
	TotalDuration time.Duration
	FinalError    error
	DeliveredAt   time.Time
	StoredOffline bool // 消息因用户离线而存储到离线队列(群组可靠投递据此分类在线/离线,避免预检查 N 次 Redis)
}

SendResult 发送结果

type Skill

type Skill string

Skill 技能标签类型

const (
	SkillTechnical  Skill = "technical" // 技术支持
	SkillSales      Skill = "sales"     // 销售
	SkillBilling    Skill = "billing"   // 计费
	SkillGeneral    Skill = "general"   // 一般客服
	SkillLanguageEN Skill = "english"   // 英语
	SkillLanguageZH Skill = "chinese"   // 中文
	SkillVIP        Skill = "vip"       // VIP客服
)

func (Skill) IsValid

func (s Skill) IsValid() bool

IsValid 检查技能是否有效

func (Skill) String

func (s Skill) String() string

String 实现Stringer接口

type StatsIncrementEntry

type StatsIncrementEntry struct {
	ConnectionID     string
	MessagesSent     int64
	MessagesReceived int64
	BytesSent        int64
	BytesReceived    int64
}

StatsIncrementEntry 统计递增条目(由连接质量仓库批量消费)

type TicketAssignedEvent

type TicketAssignedEvent struct {
	TicketID   string    `json:"ticket_id"`
	UserID     string    `json:"user_id"`
	AgentID    string    `json:"agent_id"`
	SessionID  string    `json:"session_id"`
	Priority   int       `json:"priority"`
	AssignedAt time.Time `json:"assigned_at"`
	NodeID     string    `json:"node_id"`
}

TicketAssignedEvent 工单分配事件

type TicketAssignedEventHandler

type TicketAssignedEventHandler func(event *TicketAssignedEvent) error

TicketAssignedEventHandler 工单分配事件处理器

type TicketAssignmentFailedEvent

type TicketAssignmentFailedEvent struct {
	TicketID  string    `json:"ticket_id"`
	UserID    string    `json:"user_id"`
	SessionID string    `json:"session_id"`
	Reason    string    `json:"reason"`
	Timestamp time.Time `json:"timestamp"`
	NodeID    string    `json:"node_id"`
}

TicketAssignmentFailedEvent 工单分配失败事件

type TicketAssignmentFailedEventHandler

type TicketAssignmentFailedEventHandler func(event *TicketAssignmentFailedEvent) error

TicketAssignmentFailedEventHandler 工单分配失败事件处理器

type TicketQueueEvent

type TicketQueueEvent struct {
	TicketID  string    `json:"ticket_id"`
	UserID    string    `json:"user_id"`
	SessionID string    `json:"session_id"`
	Priority  int       `json:"priority"`
	Timestamp time.Time `json:"timestamp"`
	NodeID    string    `json:"node_id"`
}

TicketQueueEvent 工单入队事件

type TicketQueueEventHandler

type TicketQueueEventHandler func(event *TicketQueueEvent) error

TicketQueueEventHandler 工单入队事件处理器

type UrgencyLevel

type UrgencyLevel string

UrgencyLevel 紧急等级

const (
	UrgencyLevelLow    UrgencyLevel = "low"    // 低紧急
	UrgencyLevelNormal UrgencyLevel = "normal" // 正常
	UrgencyLevelHigh   UrgencyLevel = "high"   // 高紧急
)

func GetAllUrgencyLevels

func GetAllUrgencyLevels() []UrgencyLevel

GetAllUrgencyLevels 获取所有紧急等级

func (UrgencyLevel) GetLevel

func (u UrgencyLevel) GetLevel() int

GetLevel 获取紧急等级数值 (0-2)

func (UrgencyLevel) IsMoreUrgentThan

func (u UrgencyLevel) IsMoreUrgentThan(other UrgencyLevel) bool

IsMoreUrgentThan 比较紧急程度

func (UrgencyLevel) IsValid

func (u UrgencyLevel) IsValid() bool

IsValid 检查紧急等级是否有效

func (UrgencyLevel) String

func (u UrgencyLevel) String() string

String 实现Stringer接口

type UserConnectionStats

type UserConnectionStats struct {
	UserID            string     `json:"user_id"`
	IsActive          bool       `json:"is_active"`
	ConnectedAt       time.Time  `json:"connected_at"`
	DisconnectedAt    *time.Time `json:"disconnected_at,omitempty"`
	Duration          int64      `json:"duration"`
	ReconnectCount    int        `json:"reconnect_count"`    // 拆表后零填充
	ErrorCount        int        `json:"error_count"`        // 拆表后零填充
	MessagesSent      int64      `json:"messages_sent"`      // 拆表后零填充
	MessagesReceived  int64      `json:"messages_received"`  // 拆表后零填充
	AveragePingMs     float64    `json:"average_ping_ms"`    // 拆表后零填充
	ConnectionQuality float64    `json:"connection_quality"` // 拆表后零填充
}

UserConnectionStats 用户连接统计 质量维度字段(ReconnectCount/ErrorCount/MessagesSent/MessagesReceived/AveragePingMs/ConnectionQuality) 拆表后由本方法零填充,跨表补充由调用方按需从 qualityRepo 取

type UserRole

type UserRole string

UserRole 用户角色类型

const (
	UserRoleCustomer UserRole = "customer" // 客户
	UserRoleAgent    UserRole = "agent"    // 客服
	UserRoleAdmin    UserRole = "admin"    // 管理员
)

func (UserRole) IsValid

func (r UserRole) IsValid() bool

IsValid 检查角色是否有效

func (UserRole) String

func (r UserRole) String() string

String 实现Stringer接口

type UserStatus

type UserStatus string

UserStatus 用户状态

const (
	UserStatusOnline    UserStatus = "online"    // 在线
	UserStatusOffline   UserStatus = "offline"   // 离线
	UserStatusBusy      UserStatus = "busy"      // 忙碌
	UserStatusAway      UserStatus = "away"      // 离开
	UserStatusInvisible UserStatus = "invisible" // 隐身
)

func UserStatusFromInt

func UserStatusFromInt(val int) UserStatus

UserStatusFromInt 将数值转换为 UserStatus(0-4 → online/offline/busy/away/invisible,越界返回 offline)

func (UserStatus) IsValid

func (s UserStatus) IsValid() bool

IsValid 检查用户状态是否有效

func (UserStatus) String

func (s UserStatus) String() string

String 实现Stringer接口

func (UserStatus) ToInt

func (s UserStatus) ToInt() int

ToInt 将 UserStatus 转换为数值(用于原子存储)

type UserStatusEvent

type UserStatusEvent struct {
	UserID    string      `json:"user_id"`
	UserType  UserType    `json:"user_type"`
	EventType EventStatus `json:"event_type"`
	Timestamp time.Time   `json:"timestamp"`
	NodeID    string      `json:"node_id"`
	ClientID  string      `json:"client_id,omitempty"` // 可选:客户端ID
}

UserStatusEvent 用户状态事件(统一的上下线事件)

type UserStatusEventHandler

type UserStatusEventHandler func(event *UserStatusEvent) error

UserStatusEventHandler 用户状态事件处理器

type UserType

type UserType string

UserType 用户类型(与角色可能不同,提供更细分的分类)

const (
	UserTypeVisitor  UserType = "visitor"  // 访客用户
	UserTypeCustomer UserType = "customer" // 普通客户
	UserTypeAgent    UserType = "agent"    // 人工客服
	UserTypeAdmin    UserType = "admin"    // 系统管理员
	UserTypeBot      UserType = "bot"      // 机器人客服
	UserTypeVIP      UserType = "vip"      // VIP客户
	UserTypeSystem   UserType = "system"   // 系统用户
	UserTypeObserver UserType = "observer" // 观察者用户(可接收所有消息)
)

func (UserType) IsAgentType

func (t UserType) IsAgentType() bool

IsAgentType 检查是否为客服类型

func (UserType) IsCustomerType

func (t UserType) IsCustomerType() bool

IsCustomerType 检查是否为客户类型(访客、普通客户、VIP客户)

func (UserType) IsHumanType

func (t UserType) IsHumanType() bool

IsHumanType 检查是否为人类用户(排除机器人和系统)

func (UserType) IsSystemType

func (t UserType) IsSystemType() bool

IsSystemType 检查是否为系统类型(系统用户、管理员)

func (UserType) IsVIPType

func (t UserType) IsVIPType() bool

IsVIPType 检查是否为VIP用户

func (UserType) IsValid

func (t UserType) IsValid() bool

IsValid 检查用户类型是否有效

func (UserType) String

func (t UserType) String() string

String 实现Stringer接口

type VIPLevel

type VIPLevel string

VIPLevel VIP等级

const (
	VIPLevelV0 VIPLevel = "v0" // 普通用户
	VIPLevelV1 VIPLevel = "v1" // VIP1级
	VIPLevelV2 VIPLevel = "v2" // VIP2级
	VIPLevelV3 VIPLevel = "v3" // VIP3级
	VIPLevelV4 VIPLevel = "v4" // VIP4级
	VIPLevelV5 VIPLevel = "v5" // VIP5级
	VIPLevelV6 VIPLevel = "v6" // VIP6级
	VIPLevelV7 VIPLevel = "v7" // VIP7级
	VIPLevelV8 VIPLevel = "v8" // VIP8级(最高级)
)

func GetAllVIPLevels

func GetAllVIPLevels() []VIPLevel

GetAllVIPLevels 获取所有VIP等级

func VIPLevelFromLevel

func VIPLevelFromLevel(level int) VIPLevel

VIPLevelFromLevel 将数值等级转换为 VIPLevel(0-8 → v0-v8,越界返回 v0)

func (VIPLevel) GetLevel

func (v VIPLevel) GetLevel() int

GetLevel 获取VIP等级数值 (0-8)

func (VIPLevel) IsHigherThan

func (v VIPLevel) IsHigherThan(other VIPLevel) bool

IsHigherThan 比较VIP等级

func (VIPLevel) IsValid

func (v VIPLevel) IsValid() bool

IsValid 检查VIP等级是否有效

func (VIPLevel) String

func (v VIPLevel) String() string

String 实现Stringer接口

type WelcomeMessage

type WelcomeMessage struct {
	Title    string                 `json:"title"`    // 欢迎标题
	Content  string                 `json:"content"`  // 欢迎内容
	Data     map[string]interface{} `json:"data"`     // 扩展数据
	Priority Priority               `json:"priority"` // 消息优先级
}

WelcomeMessage 欢迎消息

type WelcomeMessageProvider

type WelcomeMessageProvider interface {
	// GetWelcomeMessage 获取欢迎消息
	// 参数: userID 用户ID, userRole 用户角色, userType 用户类型, extraData 扩展数据
	// 返回: 欢迎消息内容, 是否启用欢迎消息, 错误信息
	GetWelcomeMessage(userID string, userRole UserRole, userType UserType, extraData map[string]interface{}) (*WelcomeMessage, bool, error)

	// RefreshConfig 刷新配置 - 当数据库配置更新时调用
	RefreshConfig() error
}

WelcomeMessageProvider 欢迎消息提供者接口

type WelcomeTemplate

type WelcomeTemplate struct {
	Title       string                 `json:"title"`        // 欢迎标题
	Content     string                 `json:"content"`      // 欢迎内容
	MessageType MessageType            `json:"message_type"` // 消息类型
	Data        map[string]interface{} `json:"data"`         // 扩展数据
	Enabled     bool                   `json:"enabled"`      // 是否启用
	Variables   []string               `json:"variables"`    // 支持的变量列表,如: {user_name}, {time}
}

WelcomeTemplate 欢迎消息模板

func (*WelcomeTemplate) ReplaceVariables

func (wt *WelcomeTemplate) ReplaceVariables(variables map[string]interface{}) WelcomeTemplate

ReplaceVariables 替换模板中的变量

type WorkloadDimension

type WorkloadDimension string

WorkloadDimension 负载统计维度

const (
	WorkloadDimensionRealtime WorkloadDimension = "realtime" // 实时(无过期)
	WorkloadDimensionHourly   WorkloadDimension = "hourly"   // 小时(保留7天)
	WorkloadDimensionDaily    WorkloadDimension = "daily"    // 日(保留90天)
	WorkloadDimensionMonthly  WorkloadDimension = "monthly"  // 月(保留13个月)
	WorkloadDimensionYearly   WorkloadDimension = "yearly"   // 年(保留5年)
)

func (WorkloadDimension) FormatKeyPart

func (d WorkloadDimension) FormatKeyPart(t time.Time) string

FormatKeyPart 格式化 Redis key 的维度部分(包含冒号分隔符) realtime: "realtime" hourly: "hourly:2026022513" daily: "daily:20260225"

func (WorkloadDimension) FormatTimeKey

func (d WorkloadDimension) FormatTimeKey(t time.Time) string

FormatTimeKey 格式化时间 key

func (WorkloadDimension) GetTTL

func (d WorkloadDimension) GetTTL() time.Duration

GetTTL 获取维度的 TTL(秒)

func (WorkloadDimension) String

func (d WorkloadDimension) String() string

String 返回维度名称

type WorkloadInfo

type WorkloadInfo struct {
	AgentID      string    `json:"agent_id"`    // 客服ID
	Workload     int64     `json:"workload"`    // 当前工作负载
	LastUpdateAt time.Time `json:"last_update"` // 最后更新时间
}

WorkloadInfo 负载信息

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL