feat(shangwutong): sync classifications and update conversations

This commit is contained in:
2026-09-11 17:08:18 +08:00
parent a06d4f4199
commit 31f157a4f5
28 changed files with 2541 additions and 232 deletions
@@ -15,12 +15,15 @@ import (
"github.com/gochat/gochat/internal/model"
channelmodel "github.com/gochat/gochat/internal/model/channel"
"github.com/gochat/gochat/internal/service"
"github.com/gochat/gochat/internal/worker"
"gorm.io/datatypes"
"gorm.io/gorm"
)
type ShangwutongConnectorHandler struct {
db *gorm.DB
messageSvc *service.MessageService
worker *worker.WorkerPool
}
type shangwutongContactMetadataRequest struct {
@@ -75,8 +78,12 @@ func (h *ShangwutongConnectorHandler) UpdateContactMetadata(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"updated": true})
}
func NewShangwutongConnectorHandler(db *gorm.DB, messageSvc *service.MessageService) *ShangwutongConnectorHandler {
return &ShangwutongConnectorHandler{db: db, messageSvc: messageSvc}
func NewShangwutongConnectorHandler(db *gorm.DB, messageSvc *service.MessageService, pools ...*worker.WorkerPool) *ShangwutongConnectorHandler {
var pool *worker.WorkerPool
if len(pools) > 0 {
pool = pools[0]
}
return &ShangwutongConnectorHandler{db: db, messageSvc: messageSvc, worker: pool}
}
type shangwutongConnectorInbox struct {
@@ -270,6 +277,454 @@ func (h *ShangwutongConnectorHandler) UpdateMessageStatus(c *gin.Context) {
})
}
type shangwutongClassificationCacheResponse struct {
InboxID uint `json:"inbox_id"`
ConversationKinds json.RawMessage `json:"conversation_kinds"`
CustomerColorKinds json.RawMessage `json:"customer_color_kinds"`
SyncStatus string `json:"sync_status"`
SyncedAt *time.Time `json:"synced_at,omitempty"`
LastErrorCode *string `json:"last_error_code,omitempty"`
LastErrorMessage *string `json:"last_error_message,omitempty"`
}
type shangwutongClassificationCallbackRequest struct {
EventID string `json:"event_id"`
ConversationKinds []shangwutongConversationKind `json:"conversation_kinds"`
CustomerColorKinds []shangwutongCustomerColorKind `json:"customer_colors"`
}
type shangwutongConversationKind struct {
ID string `json:"id"`
Name string `json:"name"`
IconIndex int `json:"icon_index"`
}
type shangwutongCustomerColorKind struct {
ID string `json:"id"`
Name string `json:"name"`
}
func (h *ShangwutongConnectorHandler) ListClassifications(c *gin.Context) {
inbox, ok := h.userInbox(c)
if !ok {
return
}
var cache model.ShangwutongClassificationCache
err := h.db.WithContext(c.Request.Context()).Where("inbox_id = ?", inbox.ID).First(&cache).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusOK, shangwutongClassificationCacheResponse{
InboxID: inbox.ID, ConversationKinds: json.RawMessage(`[]`), CustomerColorKinds: json.RawMessage(`[]`), SyncStatus: "never",
})
return
}
if err != nil {
h.connectorError(c, http.StatusInternalServerError, "classification_cache_lookup_failed", "failed to load classifications", true)
return
}
c.JSON(http.StatusOK, classificationCacheResponse(&cache))
}
func (h *ShangwutongConnectorHandler) SyncClassifications(c *gin.Context) {
inbox, ok := h.userInbox(c)
if !ok {
return
}
if h.worker == nil {
h.connectorError(c, http.StatusServiceUnavailable, "classification_sync_unavailable", "classification sync is unavailable", true)
return
}
pending := map[string]any{"sync_status": "pending", "last_error_code": nil, "last_error_message": nil}
var cache model.ShangwutongClassificationCache
err := h.db.WithContext(c.Request.Context()).Where("inbox_id = ?", inbox.ID).First(&cache).Error
switch {
case errors.Is(err, gorm.ErrRecordNotFound):
cache = model.ShangwutongClassificationCache{InboxID: inbox.ID, ConversationKinds: datatypes.JSON([]byte(`[]`)), CustomerColorKinds: datatypes.JSON([]byte(`[]`)), SyncStatus: "pending"}
if err := h.db.WithContext(c.Request.Context()).Create(&cache).Error; err != nil {
h.connectorError(c, http.StatusInternalServerError, "classification_sync_state_failed", "failed to create sync state", true)
return
}
case err != nil:
h.connectorError(c, http.StatusInternalServerError, "classification_sync_state_failed", "failed to load sync state", true)
return
default:
if err := h.db.WithContext(c.Request.Context()).Model(&cache).Updates(pending).Error; err != nil {
h.connectorError(c, http.StatusInternalServerError, "classification_sync_state_failed", "failed to update sync state", true)
return
}
}
eventID, err := service.EnqueueShangwutongClassificationSync(c.Request.Context(), h.worker, inbox)
if err != nil {
code, message := "classification_sync_queue_failed", err.Error()
_ = h.db.WithContext(c.Request.Context()).Model(&cache).Updates(map[string]any{"sync_status": "failed", "last_error_code": code, "last_error_message": message})
h.connectorError(c, http.StatusServiceUnavailable, code, "failed to queue classification sync", true)
return
}
c.JSON(http.StatusAccepted, gin.H{"sync_id": eventID, "sync_status": "pending"})
}
type shangwutongConversationClassificationRequest struct {
ChatKindID string `json:"chat_kind_id"`
CustomerColorID string `json:"customer_color_id"`
}
func (h *ShangwutongConnectorHandler) UpdateConversationClassification(c *gin.Context) {
accountID, err := strconv.ParseUint(c.Param("account_id"), 10, 64)
if err != nil || accountID == 0 {
h.connectorError(c, http.StatusNotFound, "not_found", "conversation not found", false)
return
}
conversationID, err := strconv.ParseUint(c.Param("conversation_id"), 10, 64)
if err != nil || conversationID == 0 {
h.connectorError(c, http.StatusNotFound, "not_found", "conversation not found", false)
return
}
var request shangwutongConversationClassificationRequest
if err := c.ShouldBindJSON(&request); err != nil {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classification_change", "classification change payload is invalid", false)
return
}
request.ChatKindID, request.CustomerColorID = strings.TrimSpace(request.ChatKindID), strings.TrimSpace(request.CustomerColorID)
if (request.ChatKindID == "") == (request.CustomerColorID == "") {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classification_change", "exactly one classification is required", false)
return
}
var conversation model.Conversation
if err := h.db.WithContext(c.Request.Context()).Where("id = ? AND account_id = ?", uint(conversationID), uint(accountID)).First(&conversation).Error; err != nil {
h.connectorError(c, http.StatusNotFound, "not_found", "conversation not found", false)
return
}
var inbox model.Inbox
if err := h.db.WithContext(c.Request.Context()).Where("id = ? AND account_id = ? AND channel_type = ?", conversation.InboxID, uint(accountID), "shangwutong").First(&inbox).Error; err != nil {
h.connectorError(c, http.StatusUnprocessableEntity, "not_shangwutong_conversation", "conversation is not a Shangwutong conversation", false)
return
}
var contactInbox model.ContactInbox
query := h.db.WithContext(c.Request.Context()).Where("inbox_id = ? AND contact_id = ?", inbox.ID, conversation.ContactID)
if conversation.ContactInboxID != nil && *conversation.ContactInboxID != 0 {
query = h.db.WithContext(c.Request.Context()).Where("id = ? AND inbox_id = ?", *conversation.ContactInboxID, inbox.ID)
}
if err := query.First(&contactInbox).Error; err != nil || strings.TrimSpace(contactInbox.SourceID) == "" {
h.connectorError(c, http.StatusUnprocessableEntity, "conversation_target_unavailable", "conversation has no Shangwutong session", false)
return
}
var cache model.ShangwutongClassificationCache
if err := h.db.WithContext(c.Request.Context()).Where("inbox_id = ?", inbox.ID).First(&cache).Error; err != nil {
h.connectorError(c, http.StatusConflict, "classification_not_synced", "sync Shangwutong classifications first", false)
return
}
if cache.SyncStatus != "succeeded" {
h.connectorError(c, http.StatusConflict, "classification_not_synced", "sync Shangwutong classifications first", false)
return
}
var conversationKinds []shangwutongConversationKind
var customerColors []shangwutongCustomerColorKind
if err := json.Unmarshal(cache.ConversationKinds, &conversationKinds); err != nil || json.Unmarshal(cache.CustomerColorKinds, &customerColors) != nil {
h.connectorError(c, http.StatusInternalServerError, "classification_cache_invalid", "stored classifications are invalid", true)
return
}
colorName := ""
if request.ChatKindID != "" {
if !containsConversationKind(conversationKinds, request.ChatKindID) {
h.connectorError(c, http.StatusUnprocessableEntity, "classification_not_found", "conversation classification is not in the synced catalog", false)
return
}
} else {
for _, color := range customerColors {
if color.ID == request.CustomerColorID {
colorName = color.Name
break
}
}
if colorName == "" {
h.connectorError(c, http.StatusUnprocessableEntity, "classification_not_found", "customer classification is not in the synced catalog", false)
return
}
var metadata struct {
CID string `json:"cid"`
}
if err := json.Unmarshal(contactInbox.ChannelMetadata, &metadata); err != nil || strings.TrimSpace(metadata.CID) == "" {
h.connectorError(c, http.StatusUnprocessableEntity, "conversation_target_unavailable", "conversation has no customer cid", false)
return
}
request.CustomerColorID = strings.TrimSpace(request.CustomerColorID)
if h.worker == nil {
h.connectorError(c, http.StatusServiceUnavailable, "classification_queue_unavailable", "classification queue is unavailable", true)
return
}
eventID, err := service.EnqueueShangwutongClassificationChange(c.Request.Context(), h.worker, &inbox, conversation.ID, contactInbox.SourceID, metadata.CID, "", request.CustomerColorID, colorName)
if err != nil {
h.connectorError(c, http.StatusServiceUnavailable, "classification_queue_failed", "failed to queue classification change", true)
return
}
c.JSON(http.StatusAccepted, gin.H{"sync_id": eventID, "status": "pending", "customer_color_id": request.CustomerColorID})
return
}
if h.worker == nil {
h.connectorError(c, http.StatusServiceUnavailable, "classification_queue_unavailable", "classification queue is unavailable", true)
return
}
eventID, err := service.EnqueueShangwutongClassificationChange(c.Request.Context(), h.worker, &inbox, conversation.ID, contactInbox.SourceID, "", request.ChatKindID, "", "")
if err != nil {
h.connectorError(c, http.StatusServiceUnavailable, "classification_queue_failed", "failed to queue classification change", true)
return
}
c.JSON(http.StatusAccepted, gin.H{"sync_id": eventID, "status": "pending", "chat_kind_id": request.ChatKindID})
}
func containsConversationKind(kinds []shangwutongConversationKind, id string) bool {
for _, kind := range kinds {
if kind.ID == id {
return true
}
}
return false
}
func (h *ShangwutongConnectorHandler) UpdateClassificationCatalog(c *gin.Context) {
inbox, ok := h.authorizedInbox(c)
if !ok {
return
}
var request shangwutongClassificationCallbackRequest
if err := c.ShouldBindJSON(&request); err != nil || strings.TrimSpace(request.EventID) == "" || len(request.EventID) > 128 {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classifications", "classification payload is invalid", false)
return
}
expectedKey := fmt.Sprintf("swt-classification-sync:%d:%s", inbox.ID, request.EventID)
if c.GetHeader("Idempotency-Key") != expectedKey {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_idempotency_key", "Idempotency-Key does not match event_id", false)
return
}
if !validClassificationCatalog(request) {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classifications", "classification payload is invalid", false)
return
}
conversationKinds, err := json.Marshal(request.ConversationKinds)
if err != nil {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classifications", "classification payload is invalid", false)
return
}
customerColors, err := json.Marshal(request.CustomerColorKinds)
if err != nil {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classifications", "classification payload is invalid", false)
return
}
var cache model.ShangwutongClassificationCache
err = h.db.WithContext(c.Request.Context()).Where("inbox_id = ?", inbox.ID).First(&cache).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
cache = model.ShangwutongClassificationCache{InboxID: inbox.ID}
} else if err != nil {
h.connectorError(c, http.StatusInternalServerError, "classification_cache_update_failed", "failed to load classification cache", true)
return
} else if cache.LastSyncEventID == request.EventID {
c.JSON(http.StatusOK, gin.H{"updated": false, "sync_status": cache.SyncStatus})
return
}
now := time.Now().UTC()
cache.ConversationKinds = conversationKinds
cache.CustomerColorKinds = customerColors
cache.SyncStatus = "succeeded"
cache.SyncedAt = &now
cache.LastErrorCode = nil
cache.LastErrorMessage = nil
cache.LastSyncEventID = request.EventID
if err := h.db.WithContext(c.Request.Context()).Save(&cache).Error; err != nil {
h.connectorError(c, http.StatusInternalServerError, "classification_cache_update_failed", "failed to save classifications", true)
return
}
c.JSON(http.StatusOK, gin.H{"updated": true, "sync_status": cache.SyncStatus, "synced_at": cache.SyncedAt})
}
type shangwutongClassificationSyncStatusRequest struct {
EventID string `json:"event_id"`
Status string `json:"status"`
ErrorCode string `json:"error_code"`
ErrorMessage string `json:"error_message"`
}
func (h *ShangwutongConnectorHandler) UpdateClassificationSyncStatus(c *gin.Context) {
inbox, ok := h.authorizedInbox(c)
if !ok {
return
}
var request shangwutongClassificationSyncStatusRequest
if err := c.ShouldBindJSON(&request); err != nil || strings.TrimSpace(request.EventID) == "" || request.Status != "failed" {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classification_sync_status", "classification sync status is invalid", false)
return
}
expectedKey := fmt.Sprintf("swt-classification-sync:%d:%s", inbox.ID, request.EventID)
if c.GetHeader("Idempotency-Key") != expectedKey {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_idempotency_key", "Idempotency-Key does not match event_id", false)
return
}
var cache model.ShangwutongClassificationCache
if err := h.db.WithContext(c.Request.Context()).Where("inbox_id = ?", inbox.ID).First(&cache).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
cache = model.ShangwutongClassificationCache{InboxID: inbox.ID, ConversationKinds: datatypes.JSON([]byte(`[]`)), CustomerColorKinds: datatypes.JSON([]byte(`[]`))}
} else {
h.connectorError(c, http.StatusInternalServerError, "classification_cache_update_failed", "failed to load classification cache", true)
return
}
}
cache.SyncStatus = request.Status
cache.LastErrorCode = stringPointer(request.ErrorCode)
cache.LastErrorMessage = stringPointer(request.ErrorMessage)
if err := h.db.WithContext(c.Request.Context()).Save(&cache).Error; err != nil {
h.connectorError(c, http.StatusInternalServerError, "classification_cache_update_failed", "failed to save classification sync status", true)
return
}
c.JSON(http.StatusOK, gin.H{"updated": true, "sync_status": cache.SyncStatus})
}
func stringPointer(value string) *string {
value = strings.TrimSpace(value)
if value == "" {
return nil
}
return &value
}
type shangwutongClassificationStatusRequest struct {
EventID string `json:"event_id"`
Operation string `json:"operation"`
Status string `json:"status"`
ChatKindID string `json:"chat_kind_id"`
CustomerColorID string `json:"customer_color_id"`
ErrorCode string `json:"error_code"`
ErrorMessage string `json:"error_message"`
}
func (h *ShangwutongConnectorHandler) UpdateClassificationStatus(c *gin.Context) {
inbox, ok := h.authorizedInbox(c)
if !ok {
return
}
conversationID, err := strconv.ParseUint(c.Params.ByName("conversation_id"), 10, 64)
if err != nil || conversationID == 0 {
h.connectorError(c, http.StatusNotFound, "not_found", "conversation not found", false)
return
}
var request shangwutongClassificationStatusRequest
if err := c.ShouldBindJSON(&request); err != nil || strings.TrimSpace(request.EventID) == "" || request.Operation == "" || (request.Status != "succeeded" && request.Status != "failed" && request.Status != "uncertain") {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classification_status", "classification status is invalid", false)
return
}
expectedKey := fmt.Sprintf("swt-classification-operation:%d:%s", inbox.ID, request.EventID)
if c.GetHeader("Idempotency-Key") != expectedKey {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_idempotency_key", "Idempotency-Key does not match event_id", false)
return
}
request.Operation = strings.TrimSpace(request.Operation)
request.ChatKindID, request.CustomerColorID = strings.TrimSpace(request.ChatKindID), strings.TrimSpace(request.CustomerColorID)
if request.Operation != "set_chat_kind" && request.Operation != "set_customer_color" || (request.ChatKindID == "") == (request.CustomerColorID == "") {
h.connectorError(c, http.StatusUnprocessableEntity, "invalid_classification_status", "classification operation and target are invalid", false)
return
}
var conversation model.Conversation
if err := h.db.WithContext(c.Request.Context()).Where("id = ? AND inbox_id = ?", uint(conversationID), inbox.ID).First(&conversation).Error; err != nil {
h.connectorError(c, http.StatusNotFound, "not_found", "conversation not found", false)
return
}
attributes := map[string]any{}
if len(conversation.AdditionalAttributes) > 0 {
if err := json.Unmarshal(conversation.AdditionalAttributes, &attributes); err != nil {
h.connectorError(c, http.StatusInternalServerError, "conversation_attributes_invalid", "conversation attributes are invalid", true)
return
}
}
if request.Status == "succeeded" {
if request.ChatKindID != "" {
attributes["swt_chat_kind"] = request.ChatKindID
} else {
attributes["swt_label_color"] = request.CustomerColorID
}
}
attributes["swt_classification_status"] = request.Status
attributes["swt_classification_event_id"] = request.EventID
if request.ErrorCode != "" {
attributes["swt_classification_error_code"] = request.ErrorCode
} else {
delete(attributes, "swt_classification_error_code")
}
if request.ErrorMessage != "" {
attributes["swt_classification_error"] = request.ErrorMessage
} else {
delete(attributes, "swt_classification_error")
}
updatedAttributes, err := json.Marshal(attributes)
if err != nil {
h.connectorError(c, http.StatusInternalServerError, "conversation_attributes_invalid", "conversation attributes are invalid", true)
return
}
if err := h.db.WithContext(c.Request.Context()).Model(&conversation).Update("additional_attributes", updatedAttributes).Error; err != nil {
h.connectorError(c, http.StatusInternalServerError, "conversation_update_failed", "failed to update conversation classification", true)
return
}
c.JSON(http.StatusOK, gin.H{"updated": true, "status": request.Status})
}
func (h *ShangwutongConnectorHandler) userInbox(c *gin.Context) (*model.Inbox, bool) {
accountID, err := strconv.ParseUint(c.Param("account_id"), 10, 64)
if err != nil || accountID == 0 {
h.connectorError(c, http.StatusNotFound, "not_found", "inbox not found", false)
return nil, false
}
inboxID, err := strconv.ParseUint(c.Param("inbox_id"), 10, 64)
if err != nil || inboxID == 0 {
h.connectorError(c, http.StatusNotFound, "not_found", "inbox not found", false)
return nil, false
}
var inbox model.Inbox
if err := h.db.WithContext(c.Request.Context()).Where("id = ? AND account_id = ? AND channel_type = ?", uint(inboxID), uint(accountID), "shangwutong").First(&inbox).Error; err != nil {
h.connectorError(c, http.StatusNotFound, "not_found", "inbox not found", false)
return nil, false
}
return &inbox, true
}
func classificationCacheResponse(cache *model.ShangwutongClassificationCache) shangwutongClassificationCacheResponse {
conversationKinds := json.RawMessage(cache.ConversationKinds)
if len(conversationKinds) == 0 {
conversationKinds = json.RawMessage(`[]`)
}
customerColors := json.RawMessage(cache.CustomerColorKinds)
if len(customerColors) == 0 {
customerColors = json.RawMessage(`[]`)
}
return shangwutongClassificationCacheResponse{
InboxID: cache.InboxID, ConversationKinds: conversationKinds, CustomerColorKinds: customerColors,
SyncStatus: cache.SyncStatus, SyncedAt: cache.SyncedAt, LastErrorCode: cache.LastErrorCode, LastErrorMessage: cache.LastErrorMessage,
}
}
func validClassificationCatalog(request shangwutongClassificationCallbackRequest) bool {
if len(request.ConversationKinds) > 256 || len(request.CustomerColorKinds) > 256 {
return false
}
seen := make(map[string]struct{}, len(request.ConversationKinds)+len(request.CustomerColorKinds))
for _, kind := range request.ConversationKinds {
if strings.TrimSpace(kind.ID) == "" || strings.TrimSpace(kind.Name) == "" || len(kind.ID) > 128 || len(kind.Name) > 255 {
return false
}
key := "conversation:" + kind.ID
if _, ok := seen[key]; ok {
return false
}
seen[key] = struct{}{}
}
for _, color := range request.CustomerColorKinds {
if strings.TrimSpace(color.ID) == "" || strings.TrimSpace(color.Name) == "" || len(color.ID) > 128 || len(color.Name) > 255 {
return false
}
key := "color:" + color.ID
if _, ok := seen[key]; ok {
return false
}
seen[key] = struct{}{}
}
return true
}
func (h *ShangwutongConnectorHandler) inboxItem(c *gin.Context, inbox *model.Inbox) (shangwutongConnectorInbox, error) {
var config model.ChannelShangwutongConfig
if err := h.db.WithContext(c.Request.Context()).Where("inbox_id = ?", inbox.ID).First(&config).Error; err != nil {