wlw/ai-agent
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(notification): add dashboard notification center

Browse Source
This commit is contained in:
mlogclub
2026-04-28 19:20:38 +08:00
parent f287727c59
commit 909854cd03
22 changed files with 1230 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/services/event_handlers/notification_event_handler.go
+84
View File
@@ -0,0 +1,84 @@
package event_handlers
import (
"context"
"fmt"
"log/slog"
"strings"
"cs-agent/internal/events"
"cs-agent/internal/pkg/dto/request"
"cs-agent/internal/pkg/eventbus"
"cs-agent/internal/services"
"github.com/mlogclub/simple/common/strs"
)
func init() {
eventbus.
Register[events.TicketAssignedEvent]().
Subscribe(handleTicketAssignedInAppNotification)
eventbus.
Register[events.ConversationAssignedEvent]().
Subscribe(handleConversationAssignedInAppNotification)
}
func handleTicketAssignedInAppNotification(ctx context.Context, event events.TicketAssignedEvent) error {
if event.TicketID <= 0 || event.ToUserID <= 0 {
return nil
}
ticket := services.TicketService.Get(event.TicketID)
if ticket == nil {
return nil
}
content := fmt.Sprintf("工单 %s 已指派给你", strs.DefaultIfBlank(ticket.TicketNo, fmt.Sprintf("#%d", ticket.ID)))
if title := strings.TrimSpace(ticket.Title); title != "" {
content = content + "\n" + title
}
if reason := strings.TrimSpace(event.Reason); reason != "" {
content = content + "\n指派原因: " + reason
}
_, err := services.NotificationService.CreateAndPush(request.CreateNotificationRequest{
RecipientUserID: event.ToUserID,
Title: "工单指派提醒",
Content: content,
NotificationType: "ticket_assigned",
BizType: "ticket",
BizID: ticket.ID,
ActionURL: fmt.Sprintf("/dashboard/tickets/%d", ticket.ID),
})
if err != nil {
slog.Error("create ticket assigned in-app notification failed", "error", err, "ticketId", event.TicketID, "toUserId", event.ToUserID)
}
return nil
}
func handleConversationAssignedInAppNotification(ctx context.Context, event events.ConversationAssignedEvent) error {
if event.ConversationID <= 0 || event.ToUserID <= 0 {
return nil
}
conversation := services.ConversationService.Get(event.ConversationID)
if conversation == nil {
return nil
}
content := fmt.Sprintf("会话 #%d 已分配给你", conversation.ID)
if summary := strings.TrimSpace(services.ConversationService.BuildConversationSummary(conversation)); summary != "" {
content = content + "\n" + summary
}
if reason := strings.TrimSpace(event.Reason); reason != "" {
content = content + "\n分配原因: " + reason
}
_, err := services.NotificationService.CreateAndPush(request.CreateNotificationRequest{
RecipientUserID: event.ToUserID,
Title: conversationAssignedNotifyTitle(event.AssignType),
Content: content,
NotificationType: "conversation_assigned",
BizType: "conversation",
BizID: conversation.ID,
ActionURL: fmt.Sprintf("/dashboard/conversations?conversationId=%d", conversation.ID),
})
if err != nil {
slog.Error("create conversation assigned in-app notification failed", "error", err, "conversationId", event.ConversationID, "toUserId", event.ToUserID)
}
return nil
}
internal/services/event_handlers/notification_event_handler_test.go
+123
View File
@@ -0,0 +1,123 @@
package event_handlers
import (
"context"
"testing"
"time"
"cs-agent/internal/events"
"cs-agent/internal/models"
"cs-agent/internal/pkg/enums"
"cs-agent/internal/repositories"
"github.com/glebarez/sqlite"
"github.com/mlogclub/simple/sqls"
"gorm.io/gorm"
"gorm.io/gorm/schema"
)
func TestTicketAssignedInAppNotification(t *testing.T) {
setupNotificationEventHandlerTestDB(t)
ticket := &models.Ticket{
TicketNo: "TK202604280001",
Title: "退款处理",
Source: enums.TicketSourceManual,
Status: enums.TicketStatusOpen,
CurrentAssigneeID: 11,
AuditFields: models.AuditFields{
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
},
}
if err := repositories.TicketRepository.Create(sqls.DB(), ticket); err != nil {
t.Fatalf("create ticket error = %v", err)
}
if err := handleTicketAssignedInAppNotification(context.Background(), events.TicketAssignedEvent{
TicketID: ticket.ID,
FromUserID: 0,
ToUserID: 11,
OperatorID: 1,
Reason: "需要人工跟进",
}); err != nil {
t.Fatalf("handler error = %v", err)
}
list := repositories.NotificationRepository.Find(sqls.DB(), sqls.NewCnd().Eq("recipient_user_id", 11))
if len(list) != 1 {
t.Fatalf("expected 1 notification, got %d", len(list))
}
got := list[0]
if got.NotificationType != "ticket_assigned" || got.BizType != "ticket" || got.BizID != ticket.ID {
t.Fatalf("unexpected notification: %+v", got)
}
if got.ActionURL != "/dashboard/tickets/1" {
t.Fatalf("unexpected action url: %q", got.ActionURL)
}
}
func TestConversationAssignedInAppNotification(t *testing.T) {
setupNotificationEventHandlerTestDB(t)
conversation := &models.Conversation{
CustomerName: "张三",
Status: enums.IMConversationStatusActive,
CurrentAssigneeID: 22,
AuditFields: models.AuditFields{
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
},
}
if err := repositories.ConversationRepository.Create(sqls.DB(), conversation); err != nil {
t.Fatalf("create conversation error = %v", err)
}
if err := handleConversationAssignedInAppNotification(context.Background(), events.ConversationAssignedEvent{
ConversationID: conversation.ID,
FromUserID: 0,
ToUserID: 22,
OperatorID: 1,
Reason: "自动分配",
AssignType: events.ConversationAssignTypeAutoAssign,
}); err != nil {
t.Fatalf("handler error = %v", err)
}
list := repositories.NotificationRepository.Find(sqls.DB(), sqls.NewCnd().Eq("recipient_user_id", 22))
if len(list) != 1 {
t.Fatalf("expected 1 notification, got %d", len(list))
}
got := list[0]
if got.NotificationType != "conversation_assigned" || got.BizType != "conversation" || got.BizID != conversation.ID {
t.Fatalf("unexpected notification: %+v", got)
}
if got.ActionURL != "/dashboard/conversations?conversationId=1" {
t.Fatalf("unexpected action url: %q", got.ActionURL)
}
}
func setupNotificationEventHandlerTestDB(t *testing.T) *gorm.DB {
t.Helper()
db, err := gorm.Open(sqlite.Open("file::memory:?cache=shared"), &gorm.Config{
NamingStrategy: schema.NamingStrategy{
TablePrefix: "t_",
SingularTable: true,
},
})
if err != nil {
t.Fatalf("open sqlite error = %v", err)
}
t.Cleanup(func() {
sqlDB, err := db.DB()
if err == nil {
_ = sqlDB.Close()
}
})
if err := db.AutoMigrate(&models.Notification{}, &models.Ticket{}, &models.Conversation{}); err != nil {
t.Fatalf("auto migrate error = %v", err)
}
sqls.SetDB(db)
return db
}
internal/services/notification_service.go
+105
View File
@@ -0,0 +1,105 @@
package services
import (
"strings"
"time"
"cs-agent/internal/models"
"cs-agent/internal/pkg/dto/request"
"cs-agent/internal/pkg/dto/response"
"cs-agent/internal/pkg/enums"
"cs-agent/internal/pkg/errorsx"
"cs-agent/internal/pkg/utils"
"cs-agent/internal/repositories"
"github.com/mlogclub/simple/sqls"
)
var NotificationService = newNotificationService()
func newNotificationService() *notificationService {
return &notificationService{}
}
type notificationService struct {
}
func (s *notificationService) Create(req request.CreateNotificationRequest) (*models.Notification, error) {
if req.RecipientUserID <= 0 {
return nil, errorsx.InvalidParam("接收人不能为空")
}
now := time.Now()
item := &models.Notification{
RecipientUserID: req.RecipientUserID,
Title: strings.TrimSpace(req.Title),
Content: strings.TrimSpace(req.Content),
NotificationType: strings.TrimSpace(req.NotificationType),
BizType: strings.TrimSpace(req.BizType),
BizID: req.BizID,
ActionURL: strings.TrimSpace(req.ActionURL),
Status: enums.StatusOk,
CreatedAt: now,
}
if err := repositories.NotificationRepository.Create(sqls.DB(), item); err != nil {
return nil, err
}
return item, nil
}
func (s *notificationService) CreateAndPush(req request.CreateNotificationRequest) (*models.Notification, error) {
item, err := s.Create(req)
if err != nil {
return nil, err
}
WsService.PublishNotificationCreated(item.RecipientUserID, response.NotificationResponse{
ID: item.ID,
RecipientUserID: item.RecipientUserID,
Title: item.Title,
Content: item.Content,
NotificationType: item.NotificationType,
BizType: item.BizType,
BizID: item.BizID,
ActionURL: item.ActionURL,
ReadAt: utils.FormatTimePtr(item.ReadAt),
CreatedAt: utils.FormatTime(item.CreatedAt),
})
return item, nil
}
func (s *notificationService) FindPageByCnd(cnd *sqls.Cnd) ([]models.Notification, *sqls.Paging) {
return repositories.NotificationRepository.FindPageByCnd(sqls.DB(), cnd)
}
func (s *notificationService) CountUnread(userID int64) int64 {
if userID <= 0 {
return 0
}
return repositories.NotificationRepository.Count(sqls.DB(), sqls.NewCnd().
Eq("recipient_user_id", userID).
Eq("status", enums.StatusOk).
Where("read_at IS NULL"))
}
func (s *notificationService) MarkRead(id int64, userID int64) error {
if id <= 0 {
return errorsx.InvalidParam("通知不存在")
}
item := repositories.NotificationRepository.Get(sqls.DB(), id)
if item == nil || item.RecipientUserID != userID {
return errorsx.InvalidParam("通知不存在")
}
if item.ReadAt != nil {
return nil
}
now := time.Now()
return repositories.NotificationRepository.Updates(sqls.DB(), id, map[string]any{
"read_at": now,
})
}
func (s *notificationService) MarkAllRead(userID int64) error {
if userID <= 0 {
return errorsx.InvalidParam("接收人不能为空")
}
return repositories.NotificationRepository.MarkAllRead(sqls.DB(), userID, time.Now())
}
internal/services/notification_service_test.go
+126
View File
@@ -0,0 +1,126 @@
package services_test
import (
"testing"
"cs-agent/internal/models"
"cs-agent/internal/pkg/dto/request"
"cs-agent/internal/services"
"github.com/glebarez/sqlite"
"github.com/mlogclub/simple/sqls"
"gorm.io/gorm"
"gorm.io/gorm/schema"
)
func TestNotificationServiceCreateAndUnreadCount(t *testing.T) {
setupNotificationTestDB(t)
item, err := services.NotificationService.Create(request.CreateNotificationRequest{
RecipientUserID: 101,
Title: "工单指派提醒",
Content: "工单 TK-1 已指派给你",
NotificationType: "ticket_assigned",
BizType: "ticket",
BizID: 1,
ActionURL: "/dashboard/tickets/1",
})
if err != nil {
t.Fatalf("Create() error = %v", err)
}
if item.ID == 0 {
t.Fatalf("expected notification id to be assigned")
}
if item.RecipientUserID != 101 || item.ReadAt != nil {
t.Fatalf("unexpected notification: %+v", item)
}
if got := services.NotificationService.CountUnread(101); got != 1 {
t.Fatalf("expected unread count 1, got %d", got)
}
if got := services.NotificationService.CountUnread(102); got != 0 {
t.Fatalf("expected unread count 0 for another user, got %d", got)
}
}
func TestNotificationServiceMarkReadRequiresOwner(t *testing.T) {
setupNotificationTestDB(t)
item, err := services.NotificationService.Create(request.CreateNotificationRequest{
RecipientUserID: 201,
Title: "会话分配提醒",
Content: "会话 #9 已分配给你",
NotificationType: "conversation_assigned",
BizType: "conversation",
BizID: 9,
ActionURL: "/dashboard/conversations?conversationId=9",
})
if err != nil {
t.Fatalf("Create() error = %v", err)
}
if err := services.NotificationService.MarkRead(item.ID, 202); err == nil {
t.Fatalf("expected foreign user mark read to fail")
}
if got := services.NotificationService.CountUnread(201); got != 1 {
t.Fatalf("expected notification to remain unread, got %d", got)
}
if err := services.NotificationService.MarkRead(item.ID, 201); err != nil {
t.Fatalf("MarkRead() owner error = %v", err)
}
if got := services.NotificationService.CountUnread(201); got != 0 {
t.Fatalf("expected unread count 0 after mark read, got %d", got)
}
}
func TestNotificationServiceMarkAllReadOnlyCurrentUser(t *testing.T) {
setupNotificationTestDB(t)
for _, userID := range []int64{301, 301, 302} {
if _, err := services.NotificationService.Create(request.CreateNotificationRequest{
RecipientUserID: userID,
Title: "工单指派提醒",
Content: "工单已指派给你",
NotificationType: "ticket_assigned",
BizType: "ticket",
BizID: userID,
ActionURL: "/dashboard/tickets/1",
}); err != nil {
t.Fatalf("Create() error = %v", err)
}
}
if err := services.NotificationService.MarkAllRead(301); err != nil {
t.Fatalf("MarkAllRead() error = %v", err)
}
if got := services.NotificationService.CountUnread(301); got != 0 {
t.Fatalf("expected user 301 unread count 0, got %d", got)
}
if got := services.NotificationService.CountUnread(302); got != 1 {
t.Fatalf("expected user 302 unread count 1, got %d", got)
}
}
func setupNotificationTestDB(t *testing.T) *gorm.DB {
t.Helper()
db, err := gorm.Open(sqlite.Open("file::memory:?cache=shared"), &gorm.Config{
NamingStrategy: schema.NamingStrategy{
TablePrefix: "t_",
SingularTable: true,
},
})
if err != nil {
t.Fatalf("open sqlite error = %v", err)
}
t.Cleanup(func() {
sqlDB, err := db.DB()
if err == nil {
_ = sqlDB.Close()
}
})
if err := db.AutoMigrate(&models.Notification{}); err != nil {
t.Fatalf("auto migrate error = %v", err)
}
sqls.SetDB(db)
return db
}
internal/services/ws_realtime_types.go
+22 -2
View File
@@ -22,8 +22,9 @@ const (
)
const (
realtimeRoleUser = "user"
realtimeRoleAdmin = "admin"
realtimeRoleUser = "user"
realtimeRoleAdmin = "admin"
realtimeRoleNotification = "notification"
)
const (
@@ -31,6 +32,7 @@ const (
realtimeTopicGuestPrefix = "guest:"
realtimeTopicAdminPrefix = "admin:"
realtimeTopicConversationPrefix = "conversation:"
realtimeTopicNotificationPrefix = "notification:"
realtimeTopicAdminAll = "admin:all"
)
@@ -219,6 +221,24 @@ func (e RealtimeConversationChangedEvent) EventPayload() RealtimeEventPayload {
return e.Payload
}
type RealtimeNotificationCreatedPayload struct {
Notification response.NotificationResponse `json:"notification"`
}
func (RealtimeNotificationCreatedPayload) realtimeEventPayload() {}
type RealtimeNotificationCreatedEvent struct {
Payload RealtimeNotificationCreatedPayload
}
func (e RealtimeNotificationCreatedEvent) EventType() string {
return enums.IMRealtimeEventNotificationCreated
}
func (e RealtimeNotificationCreatedEvent) EventPayload() RealtimeEventPayload {
return e.Payload
}
type realtimeClientMessage struct {
Type string `json:"type"`
Topics []string `json:"topics,omitempty"`
internal/services/ws_service.go
+39
View File
@@ -54,6 +54,19 @@ func (s *wsService) HandleDashboardWS(ctx iris.Context) {
}
}
func (s *wsService) HandleDashboardNotificationWS(ctx iris.Context) {
principal := AuthService.GetAuthPrincipal(ctx)
if principal == nil {
_ = ctx.StopWithJSON(iris.StatusUnauthorized, web.JsonError(errorsx.Unauthorized("未登录或登录已过期")))
return
}
if err := s.upgradeConnection(ctx, principal, nil, realtimeRoleNotification); err != nil {
slog.Error("upgrade dashboard notification websocket failed", "error", err, "path", ctx.Path())
ctx.StopExecution()
return
}
}
func (s *wsService) HandleOpenWS(ctx iris.Context) {
channel := ChannelService.GetEnabledChannel(ctx)
if channel == nil {
@@ -420,6 +433,19 @@ func (s *wsService) PublishConversationChanged(conversation *models.Conversation
s.PublishToTopics(s.routeConversationTopics(conversation), event)
}
func (s *wsService) PublishNotificationCreated(userID int64, notification response.NotificationResponse) {
if userID <= 0 || notification.ID <= 0 {
return
}
topic := s.notificationTopic(userID)
event := s.newEvent(topic, RealtimeNotificationCreatedEvent{
Payload: RealtimeNotificationCreatedPayload{
Notification: notification,
},
})
s.PublishToTopic(topic, event)
}
func (s *wsService) PublishResyncRequired(topics []string, reason string) {
reason = strings.TrimSpace(reason)
if reason == "" {
@@ -531,6 +557,11 @@ func (s *wsService) defaultTopics(session *ClientSession) []string {
}
switch session.Role {
case realtimeRoleNotification:
if session.Principal == nil || session.Principal.UserID <= 0 {
return nil
}
return []string{s.notificationTopic(session.Principal.UserID)}
case realtimeRoleAdmin:
if session.Principal == nil || session.Principal.UserID <= 0 {
return []string{realtimeTopicAdminAll}
@@ -554,6 +585,10 @@ func (s *wsService) filterAllowedTopics(session *ClientSession, topics []string)
return nil
}
switch session.Role {
case realtimeRoleNotification:
if session.Principal == nil {
return nil
}
case realtimeRoleAdmin:
if session.Principal == nil {
return nil
@@ -645,6 +680,10 @@ func (s *wsService) adminTopic(userID int64) string {
return realtimeTopicAdminPrefix + strconv.FormatInt(userID, 10)
}
func (s *wsService) notificationTopic(userID int64) string {
return realtimeTopicNotificationPrefix + strconv.FormatInt(userID, 10)
}
func (s *wsService) conversationTopic(conversationID int64) string {
return realtimeTopicConversationPrefix + strconv.FormatInt(conversationID, 10)
}
internal/services/ws_service_test.go
+28
View File
@@ -0,0 +1,28 @@
package services
import (
"testing"
"cs-agent/internal/pkg/dto/response"
)
func TestWsNotificationTopic(t *testing.T) {
svc := newWsService()
if got := svc.notificationTopic(123); got != "notification:123" {
t.Fatalf("expected notification:123, got %q", got)
}
}
func TestWsNotificationCreatedEventType(t *testing.T) {
event := RealtimeNotificationCreatedEvent{
Payload: RealtimeNotificationCreatedPayload{
Notification: response.NotificationResponse{ID: 1},
},
}
if got := event.EventType(); got != "notification.created" {
t.Fatalf("expected notification.created, got %q", got)
}
if payload := event.EventPayload(); payload == nil {
t.Fatalf("expected payload")
}
}