rogee/gochat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

H-28: harden Shangwutong CID sync (#5)

Browse Source 
* fix(shangwutong): close contact sync review gaps

* fix(shangwutong): harden CID sync boundaries

---------

Co-authored-by: Rogee <rogee@ipao.vip>
This commit is contained in:
Rogee
2026-08-13 01:26:56 +08:00
committed by GitHub
co-authored by rogee
parent 0e9bf8dc14
commit d0995798f4
7 changed files with 133 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/handler/api/v1/shangwutong_connector_handler.go
+12 -4
View File
@@ -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)
backend/internal/service/shangwutong_contact_sync_test.go
+58
View File
@@ -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)
}
backend/internal/service/widget_service.go
-33
View File
@@ -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 {
channels/shangwutong/internal/delivery/inbound_test.go
+30 -1
View File
@@ -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
}
channels/shangwutong/internal/delivery/mapping.go
+2 -2
View File
@@ -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)
}
}
channels/shangwutong/internal/gochat/messaging.go
+1 -1
View File
@@ -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))
}
channels/shangwutong/internal/store/store_test.go
+30
View File
@@ -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"))