diff --git a/backend/internal/handler/api/v1/shangwutong_connector_handler.go b/backend/internal/handler/api/v1/shangwutong_connector_handler.go index aa4ca9d1..c298bb78 100644 --- a/backend/internal/handler/api/v1/shangwutong_connector_handler.go +++ b/backend/internal/handler/api/v1/shangwutong_connector_handler.go @@ -37,7 +37,12 @@ func (h *ShangwutongConnectorHandler) UpdateContactMetadata(c *gin.Context) { return } var request shangwutongContactMetadataRequest - if err := c.ShouldBindJSON(&request); err != nil || strings.TrimSpace(request.CID) == "" { + if err := c.ShouldBindJSON(&request); err != nil { + h.connectorError(c, http.StatusUnprocessableEntity, "invalid_contact_metadata", "cid is required", false) + return + } + request.CID = strings.TrimSpace(request.CID) + if request.CID == "" || len(request.CID) > 255 { h.connectorError(c, http.StatusUnprocessableEntity, "invalid_contact_metadata", "cid is required", false) return } @@ -48,13 +53,16 @@ func (h *ShangwutongConnectorHandler) UpdateContactMetadata(c *gin.Context) { } metadata := map[string]any{} if len(contactInbox.ChannelMetadata) > 0 { - _ = json.Unmarshal(contactInbox.ChannelMetadata, &metadata) + if err := json.Unmarshal(contactInbox.ChannelMetadata, &metadata); err != nil { + h.connectorError(c, http.StatusInternalServerError, "contact_metadata_invalid", "stored contact metadata is invalid", true) + return + } } - if metadata["cid"] == strings.TrimSpace(request.CID) { + if metadata["cid"] == request.CID { c.JSON(http.StatusOK, gin.H{"updated": false}) return } - metadata["cid"] = strings.TrimSpace(request.CID) + metadata["cid"] = request.CID encoded, err := json.Marshal(metadata) if err != nil { h.connectorError(c, http.StatusInternalServerError, "contact_metadata_update_failed", "failed to encode contact metadata", true) diff --git a/backend/internal/service/shangwutong_contact_sync_test.go b/backend/internal/service/shangwutong_contact_sync_test.go new file mode 100644 index 00000000..f708844b --- /dev/null +++ b/backend/internal/service/shangwutong_contact_sync_test.go @@ -0,0 +1,58 @@ +package service + +import ( + "context" + "encoding/json" + "testing" + + "github.com/gochat/gochat/internal/channel" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/repository" + "github.com/gochat/gochat/internal/worker" + "github.com/stretchr/testify/require" + "gorm.io/datatypes" +) + +func TestContactUpdateQueuesDurableShangwutongEvent(t *testing.T) { + db := setupServiceTestDB(t) + account := createTestAccount(t, db) + contact := &model.Contact{AccountID: account.ID, Name: "旧昵称"} + require.NoError(t, db.Create(contact).Error) + svc := NewContactService(repository.NewContactRepo(db), nil, repository.NewNoteRepo(db)) + svc.SetWorkerPool(worker.NewWorkerPool(db)) + + _, err := svc.Update(context.Background(), account.ID, contact.ID, UpdateContactRequest{Name: "新昵称"}) + require.NoError(t, err) + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", channel.TaskTypeEventDispatch).First(&job).Error) + var event channel.ChannelEvent + require.NoError(t, json.Unmarshal(job.Payload, &event)) + require.Equal(t, channel.EventContactUpdated, event.Type) + require.Equal(t, contact.ID, event.ContactID) +} + +func TestShangwutongContactListenerQueuesOnlyCIDBoundInboxes(t *testing.T) { + db := setupServiceTestDB(t) + account := createTestAccount(t, db) + contact := &model.Contact{AccountID: account.ID, Name: "昵称"} + require.NoError(t, db.Create(contact).Error) + swtInbox := &model.Inbox{AccountID: account.ID, Name: "SWT", ChannelType: "shangwutong", Enabled: true} + webInbox := &model.Inbox{AccountID: account.ID, Name: "Widget", ChannelType: "web_widget", Enabled: true} + require.NoError(t, db.Create(swtInbox).Error) + require.NoError(t, db.Create(webInbox).Error) + require.NoError(t, db.Create(&model.ContactInbox{ContactID: contact.ID, InboxID: swtInbox.ID, SourceID: "sid", ChannelMetadata: datatypes.JSON(`{"cid":"cid-1"}`)}).Error) + require.NoError(t, db.Create(&model.ContactInbox{ContactID: contact.ID, InboxID: webInbox.ID, SourceID: "widget"}).Error) + + wp := worker.NewWorkerPool(db) + listener := NewShangwutongContactListener(db, wp) + require.NoError(t, listener.OnEvent(context.Background(), &channel.ChannelEvent{ + Type: channel.EventContactUpdated, AccountID: account.ID, ContactID: contact.ID, + })) + var jobs []model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeShangwutongWebhookDelivery).Find(&jobs).Error) + require.Len(t, jobs, 1) + var job shangwutongWebhookDeliveryJob + require.NoError(t, json.Unmarshal(jobs[0].Payload, &job)) + require.Equal(t, "sid", job.SourceID) + require.Equal(t, "cid-1", job.CID) +} diff --git a/backend/internal/service/widget_service.go b/backend/internal/service/widget_service.go index 994205b1..0426aa6d 100644 --- a/backend/internal/service/widget_service.go +++ b/backend/internal/service/widget_service.go @@ -8,7 +8,6 @@ import ( "encoding/json" "errors" "fmt" - "reflect" "strconv" "strings" "time" @@ -207,7 +206,6 @@ type PublicContactRequest struct { PhoneNumber string CustomAttributes map[string]any AdditionalAttributes map[string]any - ChannelMetadata map[string]any } type PublicContactResponse struct { @@ -780,11 +778,6 @@ func (s *WidgetService) PublicCreateContact(ctx context.Context, inboxIdentifier if err != nil { return nil, err } - if setPublicContactInboxChannelMetadata(existingInbox, req.ChannelMetadata) { - if err := s.contactInboxRepo.Update(ctx, existingInbox); err != nil { - return nil, err - } - } existingInbox.Contact = *contact return &PublicContactResponse{ContactInbox: existingInbox, Contact: contact}, nil } @@ -808,7 +801,6 @@ func (s *WidgetService) PublicCreateContact(ctx context.Context, inboxIdentifier HMACToken: hmacToken, HMACVerified: req.IdentifierHash != "", } - setPublicContactInboxChannelMetadata(contactInbox, req.ChannelMetadata) if err := s.contactInboxRepo.Create(ctx, contactInbox); err != nil { return nil, err } @@ -843,11 +835,6 @@ func (s *WidgetService) PublicUpdateContact(ctx context.Context, inboxIdentifier if err != nil { return nil, err } - if setPublicContactInboxChannelMetadata(contactInbox, req.ChannelMetadata) { - if err := s.contactInboxRepo.Update(ctx, contactInbox); err != nil { - return nil, err - } - } if req.IdentifierHash != "" && !contactInbox.HMACVerified { contactInbox.HMACVerified = true if err := s.contactInboxRepo.Update(ctx, contactInbox); err != nil { @@ -857,26 +844,6 @@ func (s *WidgetService) PublicUpdateContact(ctx context.Context, inboxIdentifier return &PublicContactResponse{ContactInbox: contactInbox, Contact: contact}, nil } -func setPublicContactInboxChannelMetadata(contactInbox *model.ContactInbox, metadata map[string]any) bool { - if contactInbox == nil || len(metadata) == 0 { - return false - } - current := jsonMap(contactInbox.ChannelMetadata) - changed := false - for key, value := range metadata { - if key == "cid" { - continue - } - if !reflect.DeepEqual(current[key], value) { - current[key], changed = value, true - } - } - if changed { - contactInbox.ChannelMetadata = mustJSON(current) - } - return changed -} - func (s *WidgetService) PublicListConversations(ctx context.Context, inboxIdentifier, sourceID string) ([]model.Conversation, error) { _, contactInbox, err := s.resolvePublicContactInbox(ctx, inboxIdentifier, sourceID) if err != nil { diff --git a/channels/shangwutong/internal/delivery/inbound_test.go b/channels/shangwutong/internal/delivery/inbound_test.go index 43be10ef..94ed9bc1 100644 --- a/channels/shangwutong/internal/delivery/inbound_test.go +++ b/channels/shangwutong/internal/delivery/inbound_test.go @@ -47,6 +47,29 @@ func TestInboundVisitorMessageCreatesResourcesAndPersistsMappings(t *testing.T) } } +func TestInboundKind52PersistsCIDWithoutCreatingResources(t *testing.T) { + ctx := context.Background() + database, account := deliveryDatabase(t, ctx) + persistInboundEvent(t, database, account, swt.HeartbeatEvent{ + SessionID: "visitor", Kind: 52, Text: "a|cid%2F123|x|y|z|name", SeqID: 42, + Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 52 history 42 timestamp", + }) + client := &inboundRecorder{} + worker, err := NewInbound(database, client, nil, 1) + if err != nil { + t.Fatal(err) + } + if worked, err := worker.processInbound(ctx); err != nil || !worked { + t.Fatalf("process kind=52 = %v, %v", worked, err) + } + if client.metadataUpdates != 1 || client.lastCID != "cid/123" || client.lastSourceID != "visitor" || client.lastInboxID != account.GochatInboxID { + t.Fatalf("metadata updates = %#v", client) + } + if len(client.imports) != 0 { + t.Fatalf("kind=52 must not import a message: %#v", client.imports) + } +} + func TestInboundKind3ConfirmsUniqueOutboundEcho(t *testing.T) { ctx := context.Background() database, account := deliveryDatabase(t, ctx) @@ -407,6 +430,10 @@ type inboundRecorder struct { importErr error retractions []int64 attachmentCount int + metadataUpdates int + lastInboxID int64 + lastSourceID string + lastCID string } func (r *inboundRecorder) EnsureContact(_ context.Context, _, _ string, request gochat.ContactRequest) (gochat.Contact, error) { @@ -417,7 +444,9 @@ func (r *inboundRecorder) UpdateContact(_ context.Context, _, _, sourceID string return gochat.Contact{ID: 456, SourceID: sourceID, Name: request.Name}, nil } -func (*inboundRecorder) UpdateContactChannelMetadata(context.Context, int64, string, string) error { +func (r *inboundRecorder) UpdateContactChannelMetadata(_ context.Context, inboxID int64, sourceID, cid string) error { + r.metadataUpdates++ + r.lastInboxID, r.lastSourceID, r.lastCID = inboxID, sourceID, cid return nil } diff --git a/channels/shangwutong/internal/delivery/mapping.go b/channels/shangwutong/internal/delivery/mapping.go index 47483ccc..276f5e6b 100644 --- a/channels/shangwutong/internal/delivery/mapping.go +++ b/channels/shangwutong/internal/delivery/mapping.go @@ -195,8 +195,8 @@ func mapInboundEvent(kind int64, seqID int64, text, operator, rawTimestamp, sour func parseHistoryCID(text string) string { for _, record := range strings.Split(text, "#") { fields := strings.Split(record, "|") - if len(fields) > 5 { - if cid, err := url.QueryUnescape(fields[1]); err == nil && strings.TrimSpace(cid) != "" { + if len(fields) > 5 && strings.TrimSpace(fields[1]) != "" { + if cid, err := url.QueryUnescape(strings.TrimSpace(fields[1])); err == nil && strings.TrimSpace(cid) != "" { return strings.TrimSpace(cid) } } diff --git a/channels/shangwutong/internal/gochat/messaging.go b/channels/shangwutong/internal/gochat/messaging.go index 1a65fcd3..f1b372c4 100644 --- a/channels/shangwutong/internal/gochat/messaging.go +++ b/channels/shangwutong/internal/gochat/messaging.go @@ -37,7 +37,7 @@ func (c *Client) UpdateContactChannelMetadata(ctx context.Context, inboxID int64 if inboxID <= 0 || strings.TrimSpace(sourceID) == "" || strings.TrimSpace(cid) == "" { return errors.New("contact channel metadata identifiers are required") } - path := fmt.Sprintf("/api/v1/connector/shangwutong/inboxes/%d/contacts/%s", inboxID, url.PathEscape(sourceID)) + path := fmt.Sprintf("/api/v1/connector/shangwutong/inboxes/%d/contacts/%s", inboxID, url.PathEscape(strings.TrimSpace(sourceID))) body := map[string]any{"cid": strings.TrimSpace(cid)} return c.doJSON(ctx, http.MethodPatch, path, body, nil, fmt.Sprintf("swt-contact-metadata:%d:%s:%s", inboxID, sourceID, cid)) } diff --git a/channels/shangwutong/internal/store/store_test.go b/channels/shangwutong/internal/store/store_test.go index 22684488..efa6ae60 100644 --- a/channels/shangwutong/internal/store/store_test.go +++ b/channels/shangwutong/internal/store/store_test.go @@ -132,6 +132,36 @@ func TestRollbackMigrationRefusesRenameOperations(t *testing.T) { } } +func TestRollbackMigrationKeepsOldOperationsWhenRenameIsAbsent(t *testing.T) { + db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "rollback.db")) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(`CREATE TABLE outbound_operations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, account_id INTEGER NOT NULL, swt_sid TEXT NOT NULL, + event_id TEXT NOT NULL UNIQUE, operation TEXT NOT NULL, payload TEXT NOT NULL, + occurred_at DATETIME NOT NULL, delivery_status TEXT NOT NULL, claimed_at DATETIME, + attempts INTEGER NOT NULL, next_attempt_at DATETIME, last_error TEXT, + created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`INSERT INTO outbound_operations(account_id,swt_sid,event_id,operation,payload,occurred_at,delivery_status,attempts,created_at,updated_at) VALUES (1,'sid','event','end_conversation','{}',CURRENT_TIMESTAMP,'pending',0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)`); err != nil { + t.Fatal(err) + } + rollback, err := os.ReadFile(filepath.Join("..", "..", "db", "migrations", "002_add_rename_operation.down.sql")) + if err != nil { + t.Fatal(err) + } + if _, err := db.Exec(string(rollback)); err != nil { + t.Fatal(err) + } + var operation string + if err := db.QueryRow("SELECT operation FROM outbound_operations").Scan(&operation); err != nil || operation != "end_conversation" { + t.Fatalf("operation = %q, err=%v", operation, err) + } +} + func TestOutboundClaimStopsBehindUncertainMessage(t *testing.T) { ctx := context.Background() store, err := Open(ctx, filepath.Join(t.TempDir(), "connector.db"))