feat(distribution): 分销站点消息中心与钱包提现多租户独立分支隔离

- 新增 000070 数据库迁移:支持 chat_conversations、wallet_accounts、wallet_ledger、withdrawal_requests 按 distribution_site_id 隔离
- 消息中心:会话列表、未读数统计、站点专属客服及订单群聊与主站实现双向完全隔离
- 资金账户与提现:独立按 (user_id, distribution_site_id) 建账,分销订单自动结算至对应分站钱包,提现与流水相互独立
- 前台界面:Axios 自动透传 X-Distribution-Site-Code,钱包、提现与消息页面呈现分站专属标牌与独立说明
This commit is contained in:
yml2213
2026-09-07 18:11:27 +08:00
parent 72b805c499
commit e024319860
35 changed files with 1109 additions and 298 deletions
+118 -19
View File
@@ -3,28 +3,72 @@ package chat
import (
"context"
"errors"
"fmt"
"strings"
"time"
"hfb_sys/backend/internal/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"hfb_sys/backend/internal/model"
"time"
)
func (r *Repository) ListConversations(ctx context.Context, principal Principal, page, pageSize int) (*PaginatedResult, error) {
func (r *Repository) resolveSite(ctx context.Context, siteCode string) (*model.DistributionSite, error) {
siteCode = strings.TrimSpace(siteCode)
if siteCode == "" {
return nil, nil
}
var site model.DistributionSite
if err := r.db.WithContext(ctx).Where("site_code = ? AND status = 'active'", siteCode).First(&site).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, errors.New("distribution site unavailable")
}
return nil, err
}
return &site, nil
}
func (r *Repository) ListConversations(ctx context.Context, principal Principal, page, pageSize int, siteCode ...string) (*PaginatedResult, error) {
page, pageSize = normalizePagination(page, pageSize)
db := r.db.WithContext(ctx)
var total int64
var code string
if len(siteCode) > 0 {
code = siteCode[0]
}
site, err := r.resolveSite(ctx, code)
if err != nil {
return nil, err
}
countDB := db.Table("chat_conversations AS c").
Joins("JOIN chat_participants AS cp ON cp.conversation_id = c.id").
Where("cp.participant_type = ? AND cp.participant_id = ?", principal.Type, principal.ID).
Where("c.status = ?", "active")
dataQuery := r.conversationQuery(ctx, principal).
Where("c.status = ?", "active")
if principal.Type == "user" {
if site != nil {
siteFilter := "(c.distribution_site_id = ? OR (c.order_id IS NOT NULL AND c.order_id IN (SELECT ro.id FROM rental_orders ro WHERE ro.distribution_site_id = ?)))"
countDB = countDB.Where(siteFilter, site.ID, site.ID)
dataQuery = dataQuery.Where(siteFilter, site.ID, site.ID)
} else {
mainFilter := "(c.distribution_site_id IS NULL OR c.distribution_site_id = 0) AND (c.order_id IS NULL OR c.order_id NOT IN (SELECT ro.id FROM rental_orders ro WHERE ro.distribution_site_id IS NOT NULL AND ro.distribution_site_id > 0))"
countDB = countDB.Where(mainFilter)
dataQuery = dataQuery.Where(mainFilter)
}
}
var total int64
if err := countDB.Count(&total).Error; err != nil {
return nil, err
}
var rows []conversationRow
offset := (page - 1) * pageSize
err := r.conversationQuery(ctx, principal).
Where("c.status = ?", "active").
err = dataQuery.
Order("COALESCE(c.last_message_at, c.created_at) DESC, c.id DESC").
Offset(offset).
Limit(pageSize).
@@ -38,6 +82,7 @@ func (r *Repository) ListConversations(ctx context.Context, principal Principal,
}
return &PaginatedResult{Items: items, Total: total, Page: page, PageSize: pageSize}, nil
}
func (r *Repository) FindConversation(ctx context.Context, principal Principal, id uint64) (*ConversationDTO, error) {
db := r.db.WithContext(ctx)
// 管理员可以查看任意会话,无需是 participant
@@ -69,6 +114,7 @@ func (r *Repository) FindConversation(ctx context.Context, principal Principal,
ID: conversation.ID,
OrderID: conversation.OrderID,
ListingID: conversation.ListingID,
DistributionSiteID: conversation.DistributionSiteID,
Type: conversation.Type,
SupportScene: conversation.SupportScene,
Title: conversation.Title,
@@ -102,6 +148,7 @@ func (r *Repository) FindConversation(ctx context.Context, principal Principal,
dto := row.toDTO(participants)
return &dto, nil
}
func (r *Repository) FindOrderConversation(ctx context.Context, userID uint64, orderID uint64) (*ConversationDTO, error) {
principal := Principal{Type: "user", ID: userID}
var order model.RentalOrder
@@ -112,6 +159,22 @@ func (r *Repository) FindOrderConversation(ctx context.Context, userID uint64, o
return nil, err
}
if order.DistributionSiteID != nil && *order.DistributionSiteID > 0 {
var row conversationRow
err := r.conversationQuery(ctx, principal).Where("c.order_id = ?", orderID).First(&row).Error
if err == nil {
participants, err := r.participants(ctx, row.ID)
if err != nil {
return nil, err
}
dto := row.toDTO(participants)
return &dto, nil
}
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
}
if order.ListingID > 0 {
var row conversationRow
err := r.conversationQuery(ctx, principal).
@@ -145,31 +208,52 @@ func (r *Repository) FindOrderConversation(ctx context.Context, userID uint64, o
dto := row.toDTO(participants)
return &dto, nil
}
func (r *Repository) EnsureSupportConversation(ctx context.Context, userID uint64, scene string) (*ConversationDTO, error) {
func (r *Repository) EnsureSupportConversation(ctx context.Context, userID uint64, scene string, siteCode ...string) (*ConversationDTO, error) {
var code string
if len(siteCode) > 0 {
code = siteCode[0]
}
site, err := r.resolveSite(ctx, code)
if err != nil {
return nil, err
}
scene, title, groupCode, welcome := resolveSupportScene(scene)
if site != nil {
title = fmt.Sprintf("【%s】%s", site.SiteName, title)
welcome = fmt.Sprintf("您好,欢迎咨询【%s】在线客服!请直接描述您遇到的问题,客服会尽快回复。", site.SiteName)
}
var conversationID uint64
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
err = r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var existing model.ChatConversation
err := tx.Table("chat_conversations AS c").
existingQuery := tx.Table("chat_conversations AS c").
Select("c.*").
Joins("JOIN chat_participants AS cp ON cp.conversation_id = c.id").
Where(
"c.type = ? AND c.support_scene = ? AND cp.participant_type = ? AND cp.participant_id = ?",
ConversationTypeGeneralSupport, scene, "user", userID,
).
Order("c.id ASC").
Limit(1).
Find(&existing).Error
)
if site != nil {
existingQuery = existingQuery.Where("c.distribution_site_id = ?", site.ID)
} else {
existingQuery = existingQuery.Where("c.distribution_site_id IS NULL OR c.distribution_site_id = 0")
}
err := existingQuery.Order("c.id ASC").Limit(1).Find(&existing).Error
if err != nil {
return err
}
// 兼容旧数据:scene=general 时也匹配未回填 support_scene 的历史会话
if existing.ID == 0 && scene == SupportSceneGeneral {
// 兼容旧数据:主站 scene=general 时也匹配未回填 support_scene 的历史会话
if site == nil && existing.ID == 0 && scene == SupportSceneGeneral {
err = tx.Table("chat_conversations AS c").
Select("c.*").
Joins("JOIN chat_participants AS cp ON cp.conversation_id = c.id").
Where(
"c.type = ? AND (c.support_scene = '' OR c.support_scene IS NULL) AND cp.participant_type = ? AND cp.participant_id = ?",
"c.type = ? AND (c.support_scene = '' OR c.support_scene IS NULL) AND cp.participant_type = ? AND cp.participant_id = ? AND (c.distribution_site_id IS NULL OR c.distribution_site_id = 0)",
ConversationTypeGeneralSupport, "user", userID,
).
Order("c.id ASC").
@@ -183,12 +267,12 @@ func (r *Repository) EnsureSupportConversation(ctx context.Context, userID uint6
}
}
// 兼容旧摸大红场景码 mohong → crash
if existing.ID == 0 && scene == SupportSceneCrash {
if site == nil && existing.ID == 0 && scene == SupportSceneCrash {
err = tx.Table("chat_conversations AS c").
Select("c.*").
Joins("JOIN chat_participants AS cp ON cp.conversation_id = c.id").
Where(
"c.type = ? AND c.support_scene = ? AND cp.participant_type = ? AND cp.participant_id = ?",
"c.type = ? AND c.support_scene = ? AND cp.participant_type = ? AND cp.participant_id = ? AND (c.distribution_site_id IS NULL OR c.distribution_site_id = 0)",
ConversationTypeGeneralSupport, SupportSceneMohong, "user", userID,
).
Order("c.id ASC").
@@ -210,6 +294,9 @@ func (r *Repository) EnsureSupportConversation(ctx context.Context, userID uint6
Title: title,
Status: "active",
}
if site != nil {
conversation.DistributionSiteID = &site.ID
}
if err := tx.Create(&conversation).Error; err != nil {
return err
}
@@ -223,7 +310,19 @@ func (r *Repository) EnsureSupportConversation(ctx context.Context, userID uint6
JoinedAt: now,
},
}
if supportID := pickSupportAdminForScene(tx, groupCode); supportID > 0 {
var supportID uint64
if site != nil && site.DefaultSupportGroupID != nil && *site.DefaultSupportGroupID > 0 {
var member model.ChatSupportGroupMember
if err := tx.Where("group_id = ?", *site.DefaultSupportGroupID).Order("id ASC").First(&member).Error; err == nil && member.AdminID > 0 {
supportID = member.AdminID
}
}
if supportID == 0 {
supportID = pickSupportAdminForScene(tx, groupCode)
}
if supportID > 0 {
participants = append(participants, model.ChatParticipant{
ConversationID: conversation.ID,
ParticipantType: "admin",
@@ -26,6 +26,7 @@ func setupConversationTestDB(t *testing.T) *gorm.DB {
&model.ChatParticipant{},
&model.ChatAdminConversationState{},
&model.ChatMessage{},
&model.DistributionSite{},
); err != nil {
t.Fatalf("数据库迁移失败: %v", err)
}
@@ -151,3 +152,127 @@ func TestFindOrderConversationFallsBackToOrderGroup(t *testing.T) {
t.Fatalf("会话类型 = %q, want %q", dto.Type, ConversationTypeOrderGroup)
}
}
func TestChatTenantIsolation(t *testing.T) {
db := setupConversationTestDB(t)
repo := NewRepository(db, nil, nil)
now := time.Now()
user := model.User{Phone: "13900002001", Nickname: "分销用户"}
if err := db.Create(&user).Error; err != nil {
t.Fatalf("创建用户失败: %v", err)
}
site := model.DistributionSite{
ID: 20,
SiteCode: "site_beta",
SiteName: "贝塔分销",
Domain: "beta.hfb.com",
Status: "active",
}
if err := db.Create(&site).Error; err != nil {
t.Fatalf("创建分销站点失败: %v", err)
}
// 1. 创建主站客服会话
mainConv := model.ChatConversation{
Type: ConversationTypeGeneralSupport,
SupportScene: SupportSceneGeneral,
Title: "主站客服",
Status: "active",
}
if err := db.Create(&mainConv).Error; err != nil {
t.Fatalf("创建主站客服会话失败: %v", err)
}
if err := db.Create(&model.ChatParticipant{
ConversationID: mainConv.ID,
ParticipantType: "user",
ParticipantID: user.ID,
Role: "customer",
JoinedAt: now,
}).Error; err != nil {
t.Fatalf("添加主站会话成员失败: %v", err)
}
// 2. 创建分销站客服会话
siteConv := model.ChatConversation{
DistributionSiteID: &site.ID,
Type: ConversationTypeGeneralSupport,
SupportScene: SupportSceneGeneral,
Title: "【贝塔分销】平台客服",
Status: "active",
}
if err := db.Create(&siteConv).Error; err != nil {
t.Fatalf("创建分站客服会话失败: %v", err)
}
if err := db.Create(&model.ChatParticipant{
ConversationID: siteConv.ID,
ParticipantType: "user",
ParticipantID: user.ID,
Role: "customer",
JoinedAt: now,
}).Error; err != nil {
t.Fatalf("添加分站会话成员失败: %v", err)
}
principal := Principal{Type: "user", ID: user.ID}
// 3. 校验主站查询会话列表:严格只看到主站会话
mainList, err := repo.ListConversations(t.Context(), principal, 1, 20)
if err != nil {
t.Fatalf("查询主站会话列表失败: %v", err)
}
if mainList.Total != 1 {
t.Fatalf("主站会话数量 = %d, want 1", mainList.Total)
}
mainItems := mainList.Items.([]ConversationDTO)
if mainItems[0].ID != mainConv.ID {
t.Fatalf("主站会话 ID = %d, want %d", mainItems[0].ID, mainConv.ID)
}
if mainItems[0].DistributionSiteID != nil {
t.Fatalf("主站会话 DistributionSiteID = %v, want nil", mainItems[0].DistributionSiteID)
}
// 4. 校验分站查询会话列表:严格只看到分站会话
siteList, err := repo.ListConversations(t.Context(), principal, 1, 20, site.SiteCode)
if err != nil {
t.Fatalf("查询分站会话列表失败: %v", err)
}
if siteList.Total != 1 {
t.Fatalf("分站会话数量 = %d, want 1", siteList.Total)
}
siteItems := siteList.Items.([]ConversationDTO)
if siteItems[0].ID != siteConv.ID {
t.Fatalf("分站会话 ID = %d, want %d", siteItems[0].ID, siteConv.ID)
}
if siteItems[0].DistributionSiteID == nil || *siteItems[0].DistributionSiteID != site.ID {
t.Fatalf("分站会话 DistributionSiteID = %v, want %d", siteItems[0].DistributionSiteID, site.ID)
}
// 5. 测试 EnsureSupportConversation 分销站点场景
ensuredSiteConv, err := repo.EnsureSupportConversation(t.Context(), user.ID, "crash", site.SiteCode)
if err != nil {
t.Fatalf("分站创建撞车客服失败: %v", err)
}
if ensuredSiteConv.DistributionSiteID == nil || *ensuredSiteConv.DistributionSiteID != site.ID {
t.Fatalf("新建分站会话 DistributionSiteID = %v, want %d", ensuredSiteConv.DistributionSiteID, site.ID)
}
// 主站再次查询,仍然只有 1 个主站会话
mainListAfter, err := repo.ListConversations(t.Context(), principal, 1, 20)
if err != nil {
t.Fatalf("主站列表查询失败: %v", err)
}
if mainListAfter.Total != 1 {
t.Fatalf("主站会话数量 = %d, want 1 (分站新增会话不应出现在主站)", mainListAfter.Total)
}
// 分站再次查询,有 2 个分站会话
siteListAfter, err := repo.ListConversations(t.Context(), principal, 1, 20, site.SiteCode)
if err != nil {
t.Fatalf("分站列表查询失败: %v", err)
}
if siteListAfter.Total != 2 {
t.Fatalf("分站会话数量 = %d, want 2", siteListAfter.Total)
}
}
+3 -1
View File
@@ -11,6 +11,7 @@ type ConversationDTO struct {
ID uint64 `json:"id"`
OrderID *uint64 `json:"order_id"`
ListingID *uint64 `json:"listing_id"`
DistributionSiteID *uint64 `json:"distribution_site_id,omitempty"`
LatestOrderID *uint64 `json:"latest_order_id,omitempty"`
LatestOrderNo string `json:"latest_order_no,omitempty"`
LatestOrderStatus string `json:"latest_order_status,omitempty"`
@@ -38,7 +39,8 @@ type ConversationDTO struct {
// EnsureSupportRequest 创建/复用客服会话。
type EnsureSupportRequest struct {
// Scene 业务场景:general(默认)/ crash(撞车,兼容旧值 mohong)
Scene string `json:"scene"`
Scene string `json:"scene"`
DistributionSiteCode string `json:"distribution_site_code"`
}
type ParticipantDTO struct {
@@ -1,14 +1,31 @@
package chat
import (
"strings"
"hfb_sys/backend/pkg/response"
"github.com/gin-gonic/gin"
)
func parseSiteCode(c *gin.Context) string {
for _, val := range []string{
c.Query("distribution_site_code"),
c.Query("site_code"),
c.GetHeader("X-Distribution-Site-Code"),
c.GetHeader("X-Site-Code"),
} {
if trimmed := strings.TrimSpace(val); trimmed != "" {
return trimmed
}
}
return ""
}
func (h *Handler) list(c *gin.Context, principal Principal) {
page, pageSize := parsePagination(c)
result, err := h.service.ListConversations(c.Request.Context(), principal, page, pageSize)
siteCode := parseSiteCode(c)
result, err := h.service.ListConversations(c.Request.Context(), principal, page, pageSize, siteCode)
if err != nil {
writeChatError(c, err)
return
@@ -21,7 +21,8 @@ func (h *Handler) UnreadCount(c *gin.Context) {
response.Unauthorized(c, "缺少用户上下文")
return
}
result, err := h.service.CountUnreadMessages(c.Request.Context(), Principal{Type: "user", ID: userID})
siteCode := parseSiteCode(c)
result, err := h.service.CountUnreadMessages(c.Request.Context(), Principal{Type: "user", ID: userID}, siteCode)
if err != nil {
writeChatError(c, err)
return
@@ -65,7 +66,11 @@ func (h *Handler) EnsureSupportConversation(c *gin.Context) {
var req EnsureSupportRequest
// scene 可选;空 body 仍按平台客服处理
_ = c.ShouldBindJSON(&req)
item, err := h.service.EnsureSupportConversation(c.Request.Context(), userID, req.Scene)
siteCode := parseSiteCode(c)
if siteCode == "" && req.DistributionSiteCode != "" {
siteCode = req.DistributionSiteCode
}
item, err := h.service.EnsureSupportConversation(c.Request.Context(), userID, req.Scene, siteCode)
if err != nil {
writeChatError(c, err)
return
+28 -6
View File
@@ -15,6 +15,7 @@ type conversationRow struct {
ID uint64
OrderID *uint64
ListingID *uint64
DistributionSiteID *uint64
LatestOrderID *uint64
LatestOrderNo string
LatestOrderStatus string
@@ -40,7 +41,7 @@ type conversationRow struct {
func (r *Repository) conversationQuery(ctx context.Context, principal Principal) *gorm.DB {
return r.db.WithContext(ctx).Table("chat_conversations AS c").
Select(`c.id, c.order_id, c.listing_id, c.type, c.support_scene, c.title, c.status, c.last_message_id,
Select(`c.id, c.order_id, c.listing_id, c.distribution_site_id, c.type, c.support_scene, c.title, c.status, c.last_message_id,
c.last_message_preview, c.last_message_at, c.created_at, c.updated_at, cp.role,
(
SELECT COUNT(1)
@@ -54,17 +55,37 @@ func (r *Repository) conversationQuery(ctx context.Context, principal Principal)
Where("cp.participant_type = ? AND cp.participant_id = ?", principal.Type, principal.ID)
}
func (r *Repository) CountUnreadMessages(ctx context.Context, principal Principal) (int64, error) {
var total int64
err := r.db.WithContext(ctx).Table("chat_messages AS cm").
func (r *Repository) CountUnreadMessages(ctx context.Context, principal Principal, siteCode ...string) (int64, error) {
var code string
if len(siteCode) > 0 {
code = siteCode[0]
}
site, err := r.resolveSite(ctx, code)
if err != nil {
return 0, err
}
query := r.db.WithContext(ctx).Table("chat_messages AS cm").
Joins("JOIN chat_participants AS cp ON cp.conversation_id = cm.conversation_id").
Joins("JOIN chat_conversations AS c ON c.id = cm.conversation_id").
Where("cp.participant_type = ? AND cp.participant_id = ?", principal.Type, principal.ID).
Where("c.status = ?", "active").
Where("NOT (cm.sender_type = ? AND cm.sender_id = ?)", principal.Type, principal.ID).
Where("cm.sender_type <> ?", "system").
Where("(cp.last_read_at IS NULL OR cm.created_at > cp.last_read_at)").
Count(&total).Error
Where("(cp.last_read_at IS NULL OR cm.created_at > cp.last_read_at)")
if principal.Type == "user" {
if site != nil {
siteFilter := "(c.distribution_site_id = ? OR (c.order_id IS NOT NULL AND c.order_id IN (SELECT ro.id FROM rental_orders ro WHERE ro.distribution_site_id = ?)))"
query = query.Where(siteFilter, site.ID, site.ID)
} else {
mainFilter := "(c.distribution_site_id IS NULL OR c.distribution_site_id = 0) AND (c.order_id IS NULL OR c.order_id NOT IN (SELECT ro.id FROM rental_orders ro WHERE ro.distribution_site_id IS NOT NULL AND ro.distribution_site_id > 0))"
query = query.Where(mainFilter)
}
}
var total int64
err = query.Count(&total).Error
return total, err
}
@@ -315,6 +336,7 @@ func (row conversationRow) toDTO(participants []ParticipantDTO) ConversationDTO
ID: row.ID,
OrderID: row.OrderID,
ListingID: row.ListingID,
DistributionSiteID: row.DistributionSiteID,
LatestOrderID: row.LatestOrderID,
LatestOrderNo: row.LatestOrderNo,
LatestOrderStatus: row.LatestOrderStatus,
+5 -4
View File
@@ -89,10 +89,11 @@ func EnsureOrderConversation(tx *gorm.DB, order model.RentalOrder) (*model.ChatC
now := time.Now()
conversation := model.ChatConversation{
OrderID: &order.ID,
Type: "order_group",
Title: orderConversationTitle(order),
Status: "active",
OrderID: &order.ID,
DistributionSiteID: order.DistributionSiteID,
Type: "order_group",
Title: orderConversationTitle(order),
Status: "active",
}
if err := tx.Create(&conversation).Error; err != nil {
return nil, err
+6 -6
View File
@@ -22,18 +22,18 @@ func NewService(repo *Repository) *Service {
return &Service{repo: repo}
}
func (s *Service) ListConversations(ctx context.Context, principal Principal, page, pageSize int) (*PaginatedResult, error) {
func (s *Service) ListConversations(ctx context.Context, principal Principal, page, pageSize int, siteCode ...string) (*PaginatedResult, error) {
if s.repo == nil {
return nil, ErrDependencyUnavailable
}
return s.repo.ListConversations(ctx, principal, page, pageSize)
return s.repo.ListConversations(ctx, principal, page, pageSize, siteCode...)
}
func (s *Service) CountUnreadMessages(ctx context.Context, principal Principal) (*UnreadCountDTO, error) {
func (s *Service) CountUnreadMessages(ctx context.Context, principal Principal, siteCode ...string) (*UnreadCountDTO, error) {
if s.repo == nil {
return nil, ErrDependencyUnavailable
}
total, err := s.repo.CountUnreadMessages(ctx, principal)
total, err := s.repo.CountUnreadMessages(ctx, principal, siteCode...)
if err != nil {
return nil, err
}
@@ -54,14 +54,14 @@ func (s *Service) FindOrderConversation(ctx context.Context, userID uint64, orde
return s.repo.FindOrderConversation(ctx, userID, orderID)
}
func (s *Service) EnsureSupportConversation(ctx context.Context, userID uint64, scene string) (*ConversationDTO, error) {
func (s *Service) EnsureSupportConversation(ctx context.Context, userID uint64, scene string, siteCode ...string) (*ConversationDTO, error) {
if s.repo == nil {
return nil, ErrDependencyUnavailable
}
if userID == 0 {
return nil, ErrPermissionDenied
}
return s.repo.EnsureSupportConversation(ctx, userID, scene)
return s.repo.EnsureSupportConversation(ctx, userID, scene, siteCode...)
}
func (s *Service) Messages(ctx context.Context, principal Principal, conversationID uint64, page, pageSize int) (*PaginatedResult, error) {