From f374fb9856325428b31af3577f57a98bc68b0051 Mon Sep 17 00:00:00 2001 From: Rogee Date: Sun, 16 Aug 2026 21:47:46 +0800 Subject: [PATCH] H-162: sync XST reception before outbound delivery (#39) Co-authored-by: Rogee --- .../shangwutong/db/generated/inbound.sql.go | 60 ++++++++++- channels/shangwutong/db/generated/models.go | 1 + .../007_add_xst_reception_sync.down.sql | 1 + .../007_add_xst_reception_sync.up.sql | 1 + channels/shangwutong/db/queries/inbound.sql | 12 +++ .../shangwutong/internal/command/root_test.go | 2 +- .../shangwutong/internal/delivery/outbound.go | 45 ++++++++- .../internal/delivery/outbound_test.go | 99 ++++++++++++++++--- .../shangwutong/internal/store/store_test.go | 2 +- channels/shangwutong/internal/swt/client.go | 90 +++++++++++++++-- .../shangwutong/internal/swt/client_test.go | 45 ++++++++- channels/shangwutong/sqlc.yaml | 1 + 12 files changed, 329 insertions(+), 30 deletions(-) create mode 100644 channels/shangwutong/db/migrations/007_add_xst_reception_sync.down.sql create mode 100644 channels/shangwutong/db/migrations/007_add_xst_reception_sync.up.sql diff --git a/channels/shangwutong/db/generated/inbound.sql.go b/channels/shangwutong/db/generated/inbound.sql.go index c195950f..0d3801bc 100644 --- a/channels/shangwutong/db/generated/inbound.sql.go +++ b/channels/shangwutong/db/generated/inbound.sql.go @@ -134,7 +134,7 @@ func (q *Queries) FailInboundEvent(ctx context.Context, arg FailInboundEventPara } const getConversationMap = `-- name: GetConversationMap :one -SELECT account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, gochat_conversation_id, gochat_display_id, swt_assignee_name, created_at, updated_at, xst_required, xst_token, xst_cid, xst_kfname FROM conversation_maps +SELECT account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, gochat_conversation_id, gochat_display_id, swt_assignee_name, created_at, updated_at, xst_required, xst_token, xst_cid, xst_kfname, xst_sync_key FROM conversation_maps WHERE account_id = ? AND swt_sid = ? LIMIT 1 ` @@ -161,6 +161,7 @@ func (q *Queries) GetConversationMap(ctx context.Context, arg GetConversationMap &i.XstToken, &i.XstCid, &i.XstKfname, + &i.XstSyncKey, ) return &i, err } @@ -218,6 +219,30 @@ func (q *Queries) GetLatestConversationState(ctx context.Context, arg GetLatestC return text, err } +const getLatestXSTOperator = `-- name: GetLatestXSTOperator :one +SELECT text, seq_id FROM inbound_events +WHERE account_id = ? AND swt_sid = ? AND kind = 6 +ORDER BY id DESC +LIMIT 1 +` + +type GetLatestXSTOperatorParams struct { + AccountID int64 `json:"account_id"` + SwtSid string `json:"swt_sid"` +} + +type GetLatestXSTOperatorRow struct { + Text *string `json:"text"` + SeqID int64 `json:"seq_id"` +} + +func (q *Queries) GetLatestXSTOperator(ctx context.Context, arg GetLatestXSTOperatorParams) (*GetLatestXSTOperatorRow, error) { + row := q.db.QueryRowContext(ctx, getLatestXSTOperator, arg.AccountID, arg.SwtSid) + var i GetLatestXSTOperatorRow + err := row.Scan(&i.Text, &i.SeqID) + return &i, err +} + const getMessageMapBySWTMessageID = `-- name: GetMessageMapBySWTMessageID :one SELECT account_id, swt_sid, swt_message_id, swt_seq_id, kind, child_index, direction, gochat_message_id, gochat_source_id, content_fingerprint, retracted_at, created_at, updated_at FROM message_maps WHERE account_id = ? AND swt_sid = ? AND swt_message_id = ? @@ -361,6 +386,33 @@ func (q *Queries) InsertMessageMap(ctx context.Context, arg InsertMessageMapPara return err } +const markConversationXSTSynced = `-- name: MarkConversationXSTSynced :execrows +UPDATE conversation_maps SET + xst_sync_key = ?, + updated_at = CURRENT_TIMESTAMP +WHERE account_id = ? AND swt_sid = ? AND xst_token = ? +` + +type MarkConversationXSTSyncedParams struct { + XstSyncKey *string `json:"xst_sync_key"` + AccountID int64 `json:"account_id"` + SwtSid string `json:"swt_sid"` + XstToken *string `json:"xst_token"` +} + +func (q *Queries) MarkConversationXSTSynced(ctx context.Context, arg MarkConversationXSTSyncedParams) (int64, error) { + result, err := q.db.ExecContext(ctx, markConversationXSTSynced, + arg.XstSyncKey, + arg.AccountID, + arg.SwtSid, + arg.XstToken, + ) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + const markMessageMapRetracted = `-- name: MarkMessageMapRetracted :exec UPDATE message_maps SET retracted_at = CURRENT_TIMESTAMP, @@ -426,7 +478,7 @@ ON CONFLICT(account_id, swt_sid) DO UPDATE SET gochat_display_id = COALESCE(excluded.gochat_display_id, conversation_maps.gochat_display_id), swt_assignee_name = COALESCE(excluded.swt_assignee_name, conversation_maps.swt_assignee_name), updated_at = CURRENT_TIMESTAMP -RETURNING account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, gochat_conversation_id, gochat_display_id, swt_assignee_name, created_at, updated_at, xst_required, xst_token, xst_cid, xst_kfname +RETURNING account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, gochat_conversation_id, gochat_display_id, swt_assignee_name, created_at, updated_at, xst_required, xst_token, xst_cid, xst_kfname, xst_sync_key ` type UpsertConversationMapParams struct { @@ -464,6 +516,7 @@ func (q *Queries) UpsertConversationMap(ctx context.Context, arg UpsertConversat &i.XstToken, &i.XstCid, &i.XstKfname, + &i.XstSyncKey, ) return &i, err } @@ -479,7 +532,7 @@ ON CONFLICT(account_id, swt_sid) DO UPDATE SET xst_cid = COALESCE(excluded.xst_cid, conversation_maps.xst_cid), xst_kfname = COALESCE(excluded.xst_kfname, conversation_maps.xst_kfname), updated_at = CURRENT_TIMESTAMP -RETURNING account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, gochat_conversation_id, gochat_display_id, swt_assignee_name, created_at, updated_at, xst_required, xst_token, xst_cid, xst_kfname +RETURNING account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, gochat_conversation_id, gochat_display_id, swt_assignee_name, created_at, updated_at, xst_required, xst_token, xst_cid, xst_kfname, xst_sync_key ` type UpsertConversationXSTRouteParams struct { @@ -517,6 +570,7 @@ func (q *Queries) UpsertConversationXSTRoute(ctx context.Context, arg UpsertConv &i.XstToken, &i.XstCid, &i.XstKfname, + &i.XstSyncKey, ) return &i, err } diff --git a/channels/shangwutong/db/generated/models.go b/channels/shangwutong/db/generated/models.go index ea2bfb90..1b51bf31 100644 --- a/channels/shangwutong/db/generated/models.go +++ b/channels/shangwutong/db/generated/models.go @@ -59,6 +59,7 @@ type ConversationMap struct { XstToken *string `json:"xst_token"` XstCid *string `json:"xst_cid"` XstKfname *string `json:"xst_kfname"` + XstSyncKey *string `json:"xst_sync_key"` } type InboundEvent struct { diff --git a/channels/shangwutong/db/migrations/007_add_xst_reception_sync.down.sql b/channels/shangwutong/db/migrations/007_add_xst_reception_sync.down.sql new file mode 100644 index 00000000..34d9438f --- /dev/null +++ b/channels/shangwutong/db/migrations/007_add_xst_reception_sync.down.sql @@ -0,0 +1 @@ +ALTER TABLE conversation_maps DROP COLUMN xst_sync_key; diff --git a/channels/shangwutong/db/migrations/007_add_xst_reception_sync.up.sql b/channels/shangwutong/db/migrations/007_add_xst_reception_sync.up.sql new file mode 100644 index 00000000..3c94b00b --- /dev/null +++ b/channels/shangwutong/db/migrations/007_add_xst_reception_sync.up.sql @@ -0,0 +1 @@ +ALTER TABLE conversation_maps ADD COLUMN xst_sync_key TEXT; diff --git a/channels/shangwutong/db/queries/inbound.sql b/channels/shangwutong/db/queries/inbound.sql index cb14fd6d..e8606406 100644 --- a/channels/shangwutong/db/queries/inbound.sql +++ b/channels/shangwutong/db/queries/inbound.sql @@ -88,6 +88,12 @@ WHERE account_id = ? AND swt_sid = ? AND kind = 0 ORDER BY id DESC LIMIT 1; +-- name: GetLatestXSTOperator :one +SELECT text, seq_id FROM inbound_events +WHERE account_id = ? AND swt_sid = ? AND kind = 6 +ORDER BY id DESC +LIMIT 1; + -- name: UpsertConversationXSTRoute :one INSERT INTO conversation_maps ( account_id, swt_sid, gochat_contact_source_id, @@ -101,6 +107,12 @@ ON CONFLICT(account_id, swt_sid) DO UPDATE SET updated_at = CURRENT_TIMESTAMP RETURNING *; +-- name: MarkConversationXSTSynced :execrows +UPDATE conversation_maps SET + xst_sync_key = ?, + updated_at = CURRENT_TIMESTAMP +WHERE account_id = ? AND swt_sid = ? AND xst_token = ?; + -- name: UpsertConversationMap :one INSERT INTO conversation_maps ( account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, diff --git a/channels/shangwutong/internal/command/root_test.go b/channels/shangwutong/internal/command/root_test.go index ca19eff9..18d6b6ec 100644 --- a/channels/shangwutong/internal/command/root_test.go +++ b/channels/shangwutong/internal/command/root_test.go @@ -87,7 +87,7 @@ func TestOperationalCommands(t *testing.T) { t.Fatalf("%v produced no output", args) } } - if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 6 { + if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 7 { t.Fatalf("backup version = %d, %v", version, err) } diff --git a/channels/shangwutong/internal/delivery/outbound.go b/channels/shangwutong/internal/delivery/outbound.go index 43ecf4a0..676d8211 100644 --- a/channels/shangwutong/internal/delivery/outbound.go +++ b/channels/shangwutong/internal/delivery/outbound.go @@ -32,6 +32,7 @@ type SessionProvider interface { type MessageSender interface { SendText(context.Context, swt.Session, string, string) (swt.SendResult, error) + SyncXSTReception(context.Context, swt.Session, swt.XSTRoute) (swt.SendResult, error) SendXSTText(context.Context, swt.Session, swt.XSTRoute, string) (swt.SendResult, error) SendImage(context.Context, swt.Session, string, swt.Upload) (swt.SendResult, error) SendFile(context.Context, swt.Session, string, swt.Upload) (swt.SendResult, error) @@ -46,6 +47,12 @@ type ResultClient interface { UpdateMessageStatus(context.Context, int64, int64, gochat.MessageResult) error } +type xstDeliveryRoute struct { + swt.XSTRoute + syncKey string + synced bool +} + type Outbound struct { store *store.Store sessions SessionProvider @@ -235,7 +242,7 @@ func (o *Outbound) processOutbound(ctx context.Context) (bool, error) { return true, err } -func (o *Outbound) sendPart(ctx context.Context, message *dbgen.OutboundMessage, part *dbgen.OutboundPart, xstRoute *swt.XSTRoute) error { +func (o *Outbound) sendPart(ctx context.Context, message *dbgen.OutboundMessage, part *dbgen.OutboundPart, xstRoute *xstDeliveryRoute) error { if part.PartType == "video" || part.PartType == "unsupported" { return &permanentDeliveryError{code: "unsupported_outbound_content", err: fmt.Errorf("outbound %s is not supported", part.PartType)} } @@ -265,7 +272,18 @@ func (o *Outbound) sendPart(ctx context.Context, message *dbgen.OutboundMessage, if err != nil || rows != 1 { return errors.Join(err, errors.New("XST stage was not claimable")) } - if _, err := o.sender.SendXSTText(ctx, session, *xstRoute, content); err != nil { + if !xstRoute.synced { + if _, err := o.sender.SyncXSTReception(ctx, session, xstRoute.XSTRoute); err != nil { + return &xstDeliveryError{err: err} + } + rows, err = o.store.Writer().MarkConversationXSTSynced(ctx, dbgen.MarkConversationXSTSyncedParams{ + XstSyncKey: &xstRoute.syncKey, AccountID: message.AccountID, SwtSid: message.SwtSid, XstToken: &xstRoute.Token, + }) + if err != nil || rows != 1 { + return &xstDeliveryError{err: &swt.Error{Operation: "persist_xst_reception_sync", Code: "persistence_error", Retryable: true, Err: errors.Join(err, errors.New("XST reception sync was not persisted"))}} + } + } + if _, err := o.sender.SendXSTText(ctx, session, xstRoute.XSTRoute, content); err != nil { return &xstDeliveryError{err: err} } rows, err = o.store.Writer().CompleteOutboundXST(ctx, message.ID) @@ -315,7 +333,7 @@ func (o *Outbound) sendPart(ctx context.Context, message *dbgen.OutboundMessage, }) } -func (o *Outbound) xstRoute(ctx context.Context, message *dbgen.OutboundMessage) (*swt.XSTRoute, error) { +func (o *Outbound) xstRoute(ctx context.Context, message *dbgen.OutboundMessage) (*xstDeliveryRoute, error) { mapping, err := o.store.Reader().GetConversationMap(ctx, dbgen.GetConversationMapParams{AccountID: message.AccountID, SwtSid: message.SwtSid}) if errors.Is(err, sql.ErrNoRows) { return nil, nil @@ -327,7 +345,26 @@ func (o *Outbound) xstRoute(ctx context.Context, message *dbgen.OutboundMessage) if err != nil { return nil, fmt.Errorf("load XST conversation state: %w", err) } - route := &swt.XSTRoute{SID: message.SwtSid, CID: value(mapping.XstCid), Token: value(mapping.XstToken), State: value(state), KFName: firstText(value(mapping.XstKfname), value(mapping.SwtAssigneeName))} + operator, err := o.store.Reader().GetLatestXSTOperator(ctx, dbgen.GetLatestXSTOperatorParams{AccountID: message.AccountID, SwtSid: message.SwtSid}) + if err != nil { + return nil, fmt.Errorf("load XST operator assignment: %w", err) + } + account, err := o.store.Reader().GetAccountByID(ctx, message.AccountID) + if err != nil { + return nil, fmt.Errorf("load XST account: %w", err) + } + if !strings.EqualFold(strings.TrimSpace(value(operator.Text)), strings.TrimSpace(account.Username)) { + return nil, errors.New("XST conversation is not assigned to the current operator") + } + syncKey := fmt.Sprintf("%s\x1a%s\x1a%d", value(mapping.XstToken), account.Username, operator.SeqID) + route := &xstDeliveryRoute{ + XSTRoute: swt.XSTRoute{ + SID: message.SwtSid, CID: value(mapping.XstCid), Token: value(mapping.XstToken), State: value(state), + KFName: firstText(value(mapping.XstKfname), value(mapping.SwtAssigneeName), value(operator.Text)), + }, + syncKey: syncKey, + synced: mapping.XstSyncKey != nil && *mapping.XstSyncKey == syncKey, + } if err := route.Validate(); err != nil { return nil, err } diff --git a/channels/shangwutong/internal/delivery/outbound_test.go b/channels/shangwutong/internal/delivery/outbound_test.go index 8e2aca26..f106f81a 100644 --- a/channels/shangwutong/internal/delivery/outbound_test.go +++ b/channels/shangwutong/internal/delivery/outbound_test.go @@ -6,6 +6,7 @@ import ( "errors" "os" "path/filepath" + "strings" "testing" "time" @@ -60,11 +61,21 @@ func TestXSTOutboundRequiresBothStagesAndEcho(t *testing.T) { }, false); err != nil { t.Fatal(err) } - primaryCalls, xstCalls := 0, 0 + primaryCalls, syncCalls, xstCalls := 0, 0, 0 + order := []string{} sender := senderStub{ - beforeReturn: func() { primaryCalls++ }, + beforeReturn: func() { primaryCalls++; order = append(order, "swt") }, + syncXST: func(route swt.XSTRoute) (swt.SendResult, error) { + syncCalls++ + order = append(order, "sync") + if route.SID != "visitor" || route.CID != "cid-1" || route.State != "5" { + t.Fatalf("XST sync route = %#v", route) + } + return swt.SendResult{Status: "ok"}, nil + }, xst: func(route swt.XSTRoute, _ string) (swt.SendResult, error) { xstCalls++ + order = append(order, "xst") if route.KFName != "口腔客服2" { t.Fatalf("XST KFName = %q", route.KFName) } @@ -77,8 +88,8 @@ func TestXSTOutboundRequiresBothStagesAndEcho(t *testing.T) { t.Fatalf("outbound = %v, %v", worked, err) } queued, err := database.Reader().GetOutboundByGoChatMessageID(ctx, 77) - if err != nil || primaryCalls != 1 || xstCalls != 1 || queued.DeliveryStatus != "uncertain" || queued.XstDeliveryStatus != "delivered" { - t.Fatalf("queued=%#v primary=%d xst=%d err=%v", queued, primaryCalls, xstCalls, err) + if err != nil || primaryCalls != 1 || syncCalls != 1 || xstCalls != 1 || strings.Join(order, ",") != "swt,sync,xst" || queued.DeliveryStatus != "uncertain" || queued.XstDeliveryStatus != "delivered" { + t.Fatalf("queued=%#v primary=%d sync=%d xst=%d order=%v err=%v", queued, primaryCalls, syncCalls, xstCalls, order, err) } if worked, err := outbound.processStatus(ctx); err != nil || worked || results.result.Status != "" { t.Fatalf("premature status = %v, %v, %#v", worked, err, results) @@ -143,6 +154,51 @@ func TestXSTRetryOnlyRepeatsIncompleteStage(t *testing.T) { } } +func TestXSTReceptionSyncRetryDoesNotRepeatPrimarySend(t *testing.T) { + ctx := context.Background() + database, account := deliveryDatabase(t, ctx) + seedXSTRoute(t, database, account) + if _, _, err := database.EnqueueOutbound(ctx, store.OutboundInput{ + AccountID: account.ID, SWTSessionID: "visitor", EventID: "message:77:created", OccurredAt: time.Now(), GoChatMessageID: 77, + MessageType: "text", Content: stringPointer("hello"), Payload: deliveryPayload(t, nil), + }, false); err != nil { + t.Fatal(err) + } + primaryCalls, syncCalls, xstCalls := 0, 0, 0 + sender := senderStub{ + beforeReturn: func() { primaryCalls++ }, + syncXST: func(swt.XSTRoute) (swt.SendResult, error) { + syncCalls++ + if syncCalls == 1 { + return swt.SendResult{}, &swt.Error{Operation: "sync_xst_reception", Code: "network_error", Retryable: true} + } + return swt.SendResult{Status: "ok"}, nil + }, + xst: func(swt.XSTRoute, string) (swt.SendResult, error) { + xstCalls++ + return swt.SendResult{Status: "ok"}, nil + }, + } + outbound, _ := NewOutbound(database, sessionStub{}, sender, &resultRecorder{}, nil, 1) + if worked, err := outbound.processOutbound(ctx); err != nil || !worked { + t.Fatalf("first delivery = %v, %v", worked, err) + } + queued, err := database.Reader().GetOutboundByGoChatMessageID(ctx, 77) + if err != nil || queued.SwtSentAt == nil || queued.DeliveryStatus != "pending" || primaryCalls != 1 || syncCalls != 1 || xstCalls != 0 { + t.Fatalf("first queued=%#v primary=%d sync=%d xst=%d err=%v", queued, primaryCalls, syncCalls, xstCalls, err) + } + if delay := time.Until(*queued.NextAttemptAt); delay > 0 { + time.Sleep(delay) + } + if worked, err := outbound.processOutbound(ctx); err != nil || !worked { + t.Fatalf("retry delivery = %v, %v", worked, err) + } + mapping, err := database.Reader().GetConversationMap(ctx, dbgen.GetConversationMapParams{AccountID: account.ID, SwtSid: "visitor"}) + if err != nil || primaryCalls != 1 || syncCalls != 2 || xstCalls != 1 || mapping.XstSyncKey == nil { + t.Fatalf("mapping=%#v primary=%d sync=%d xst=%d err=%v", mapping, primaryCalls, syncCalls, xstCalls, err) + } +} + func TestXSTRejectThenEchoRetriesOnlyXST(t *testing.T) { ctx := context.Background() database, account := deliveryDatabase(t, ctx) @@ -681,7 +737,7 @@ func deliveryDatabase(t *testing.T, ctx context.Context) (*store.Store, *dbgen.A t.Cleanup(func() { _ = database.Close() }) account, err := database.Writer().CreateAccount(ctx, dbgen.CreateAccountParams{ GochatAccountID: 1, GochatInboxID: 10, GochatInboxIdentifier: "identifier", ConfigVersion: 1, - SessionID: "BYT99917999", Username: "agent", Password: "password", Enabled: 1, + SessionID: "BYT99917999", Username: "口腔客服2", Password: "password", Enabled: 1, DesiredPresence: "online", GochatHmacToken: "hmac", GochatWebhookSecret: "secret", }) if err != nil { @@ -694,11 +750,12 @@ func seedXSTRoute(t *testing.T, database *store.Store, account *dbgen.Account) { t.Helper() ctx := context.Background() events, err := swt.ParseHeartbeatBody( - "visitor 65 %7Cxst%7Csbox%1Axst-route-token%1Aquery%1Aword%1Acity%1A48989266 41 639183484622475271\r\n" + - "visitor 31 %7Cdistribute_lastoname%7C%E5%8F%A3%E8%85%94%E5%AE%A2%E6%9C%8D2 42 639183484622475272\r\n" + - "visitor 31 %7CACT_XST%7CNotShow%7CQuDaoVisitorInfoMsg%7Cprivate 43 639183484622475273\r\n" + - "visitor 0 %7C5 44 639183484622475274\r\n" + - "visitor 24 cid-1 45 639183484622475275\r\n") + "visitor 24 cid-1 40 639183484622475270\r\n" + + "visitor 65 %7Cxst%7Csbox%1Axst-route-token%1Aquery%1Aword%1Acity%1A48989266 41 639183484622475271\r\n" + + "visitor 6 %7C%E5%8F%A3%E8%85%94%E5%AE%A2%E6%9C%8D2 42 639183484622475272\r\n" + + "visitor 0 %7C5 43 639183484622475273\r\n" + + "visitor 31 %7Cdistribute_lastoname%7C%E5%8F%A3%E8%85%94%E5%AE%A2%E6%9C%8D2 44 639183484622475274\r\n" + + "visitor 31 %7CACT_XST%7CNotShow%7CQuDaoVisitorInfoMsg%7Cprivate 45 639183484622475275\r\n") if err != nil { t.Fatal(err) } @@ -736,7 +793,7 @@ func deliveryPayload(t *testing.T, attachments []gochat.WebhookAttachment) strin type sessionStub struct{} func (sessionStub) WithSession(ctx context.Context, _ int64, operation func(swt.Session) error) error { - return operation(swt.Session{BaseURL: "http://example.test/", SiteID: "site", LoginName: "agent", MAToken: "token"}) + return operation(swt.Session{BaseURL: "http://example.test/", SiteID: "site", LoginName: "口腔客服2", MAToken: "token"}) } func (sessionStub) InvalidateSession(context.Context, int64) error { return nil } @@ -745,13 +802,13 @@ type allowedSessionStub struct{ sessionStub } func (allowedSessionStub) WithSession(ctx context.Context, _ int64, operation func(swt.Session) error) error { purview := uint64(0) - return operation(swt.Session{BaseURL: "http://example.test/", SiteID: "site", LoginName: "agent", MAToken: "token", Purview: &purview}) + return operation(swt.Session{BaseURL: "http://example.test/", SiteID: "site", LoginName: "口腔客服2", MAToken: "token", Purview: &purview}) } type sessionRecorder struct{ invalidations int } func (*sessionRecorder) WithSession(ctx context.Context, _ int64, operation func(swt.Session) error) error { - return operation(swt.Session{BaseURL: "http://example.test/", SiteID: "site", LoginName: "agent", MAToken: "token"}) + return operation(swt.Session{BaseURL: "http://example.test/", SiteID: "site", LoginName: "口腔客服2", MAToken: "token"}) } func (s *sessionRecorder) InvalidateSession(context.Context, int64) error { @@ -762,6 +819,7 @@ func (s *sessionRecorder) InvalidateSession(context.Context, int64) error { type senderStub struct { err error beforeReturn func() + syncXST func(swt.XSTRoute) (swt.SendResult, error) xst func(swt.XSTRoute, string) (swt.SendResult, error) } @@ -784,6 +842,13 @@ func (s senderStub) SendVoice(context.Context, swt.Session, string, swt.Upload) return swt.SendResult{Status: "ok"}, s.err } +func (s senderStub) SyncXSTReception(_ context.Context, _ swt.Session, route swt.XSTRoute) (swt.SendResult, error) { + if s.syncXST != nil { + return s.syncXST(route) + } + return swt.SendResult{Status: "ok"}, s.err +} + func (s senderStub) SendXSTText(_ context.Context, _ swt.Session, route swt.XSTRoute, content string) (swt.SendResult, error) { if s.xst != nil { return s.xst(route, content) @@ -827,6 +892,10 @@ func (*operationSender) SendVoice(context.Context, swt.Session, string, swt.Uplo return swt.SendResult{}, nil } +func (*operationSender) SyncXSTReception(context.Context, swt.Session, swt.XSTRoute) (swt.SendResult, error) { + return swt.SendResult{}, nil +} + func (*operationSender) SendXSTText(context.Context, swt.Session, swt.XSTRoute, string) (swt.SendResult, error) { return swt.SendResult{}, nil } @@ -876,6 +945,10 @@ func (*partialSender) SendVoice(context.Context, swt.Session, string, swt.Upload return swt.SendResult{Status: "ok"}, nil } +func (*partialSender) SyncXSTReception(context.Context, swt.Session, swt.XSTRoute) (swt.SendResult, error) { + return swt.SendResult{Status: "ok"}, nil +} + func (*partialSender) SendXSTText(context.Context, swt.Session, swt.XSTRoute, string) (swt.SendResult, error) { return swt.SendResult{Status: "ok"}, nil } diff --git a/channels/shangwutong/internal/store/store_test.go b/channels/shangwutong/internal/store/store_test.go index 0ec86cab..20d50ab0 100644 --- a/channels/shangwutong/internal/store/store_test.go +++ b/channels/shangwutong/internal/store/store_test.go @@ -620,7 +620,7 @@ func TestOnlineBackupCanBeOpenedReadOnly(t *testing.T) { t.Fatalf("backup mode = %v", info.Mode().Perm()) } version, err := InspectDatabase(ctx, backup) - if err != nil || version != 6 { + if err != nil || version != 7 { t.Fatalf("backup version = %d, %v", version, err) } } diff --git a/channels/shangwutong/internal/swt/client.go b/channels/shangwutong/internal/swt/client.go index b1c2b54f..190d82a9 100644 --- a/channels/shangwutong/internal/swt/client.go +++ b/channels/shangwutong/internal/swt/client.go @@ -4,6 +4,7 @@ import ( "context" "crypto/md5" "crypto/rand" + "crypto/sha1" "encoding/hex" "encoding/json" "errors" @@ -222,27 +223,72 @@ func (c *Client) SendXSTText(ctx context.Context, session Session, route XSTRout "msgkind": {"1"}, "kfversion": {xstClientVersion}, } - body, response, requestWritten, err := c.postFormTargetTracked(ctx, target+"?"+query.Encode(), form) + return c.sendXSTForm(ctx, target+"?"+query.Encode(), form, "send_xst_text", "xst_rejected", false, session, route) +} + +func (c *Client) SyncXSTReception(ctx context.Context, session Session, route XSTRoute) (SendResult, error) { + if err := session.Validate(); err != nil { + return SendResult{}, err + } + if err := route.Validate(); err != nil { + return SendResult{}, err + } + deviceID := xstDeviceID(session) + ticks := nativeXSTTicks(c.now()) + query := url.Values{ + "t": {strconv.FormatInt(ticks, 10)}, + "sign": {xstReceptionSignature(ticks, session, route, deviceID, xstClientVersion)}, + "swtidhead": {session.SiteID}, + "onamehead": {session.LoginName}, + "kfversionhead": {xstClientVersion}, + } + target, err := resolveEndpoint(xstBaseURL, "api/SwtSyncKfInfo.ashx") + if err != nil { + return SendResult{}, err + } + form := url.Values{ + "RESET": {""}, + "cid": {route.CID}, + "sid": {route.SID}, + "clienttype": {"2"}, + "clientkind": {"3"}, + "imeiaddress": {deviceID}, + "accounttype": {"2"}, + "state": {route.State}, + "oname": {session.LoginName}, + "siteid": {session.SiteID}, + "kfversion": {xstClientVersion}, + } + return c.sendXSTForm(ctx, target+"?"+query.Encode(), form, "sync_xst_reception", "xst_sync_rejected", true, session, route) +} + +func (c *Client) sendXSTForm(ctx context.Context, target string, form url.Values, operation, rejectionCode string, idempotent bool, session Session, route XSTRoute) (SendResult, error) { + body, response, requestWritten, err := c.postFormTargetTracked(ctx, target, form) if err != nil { if response != nil && (response.StatusCode < 200 || response.StatusCode >= 300) { - return SendResult{}, &Error{Operation: "send_xst_text", Code: "http_error", Retryable: response.StatusCode >= 500, Err: err} + return SendResult{}, &Error{Operation: operation, Code: "http_error", Retryable: response.StatusCode >= 500, Err: err} } - if requestWritten { - return SendResult{}, &Error{Operation: "send_xst_text", Code: "network_result_uncertain", Uncertain: true, Err: err} + if requestWritten && !idempotent { + return SendResult{}, &Error{Operation: operation, Code: "network_result_uncertain", Uncertain: true, Err: err} } - return SendResult{}, &Error{Operation: "send_xst_text", Code: "network_error", Retryable: true, Err: err} + return SendResult{}, &Error{Operation: operation, Code: "network_error", Retryable: true, Err: err} } var payload struct { Message string `json:"msg"` } if err := json.Unmarshal([]byte(body), &payload); err != nil { - return SendResult{Body: body}, &Error{Operation: "send_xst_text", Code: "invalid_response", Retryable: true, Err: err} + return SendResult{Body: body}, &Error{Operation: operation, Code: "invalid_response", Retryable: true, Err: err} } result := SendResult{Status: strings.TrimSpace(payload.Message), Body: body} if strings.EqualFold(result.Status, "ok") { return result, nil } - return result, &Error{Operation: "send_xst_text", Code: "xst_rejected", Retryable: true} + detail := sanitizeXSTMessage(payload.Message, session.MAToken, route.Token, route.SID, route.CID, session.SiteID, session.LoginName) + var detailErr error + if detail != "" { + detailErr = errors.New(detail) + } + return result, &Error{Operation: operation, Code: rejectionCode, Retryable: true, Err: detailErr} } func javaFormEscape(value string) string { @@ -271,6 +317,36 @@ func xstSignature(ticks int64, session Session, route XSTRoute, content, version return hex.EncodeToString(digest[:]) } +func xstReceptionSignature(ticks int64, session Session, route XSTRoute, deviceID, version string) string { + separator := string(rune(26)) + plain := strings.Join([]string{ + strconv.FormatInt(ticks+10, 10), session.SiteID, "2", "3", deviceID, route.State, + "2", session.LoginName, route.CID, route.SID, version, "", + }, separator) + digest := md5.Sum([]byte(plain)) + return hex.EncodeToString(digest[:]) +} + +func xstDeviceID(session Session) string { + digest := sha1.Sum([]byte("gochat-shangwutong\x1a" + session.SiteID + "\x1a" + session.LoginName)) + return strings.ToUpper(hex.EncodeToString(digest[:])) +} + +func sanitizeXSTMessage(message string, sensitive ...string) string { + message = strings.ToValidUTF8(message, "") + for _, value := range sensitive { + if strings.TrimSpace(value) != "" { + message = strings.ReplaceAll(message, value, "[redacted]") + } + } + message = strings.Join(strings.Fields(message), " ") + runes := []rune(message) + if len(runes) > 256 { + message = string(runes[:256]) + } + return message +} + func (c *Client) sendHTML(ctx context.Context, session Session, sid, content, operation string) (SendResult, error) { if err := session.Validate(); err != nil { return SendResult{}, err diff --git a/channels/shangwutong/internal/swt/client_test.go b/channels/shangwutong/internal/swt/client_test.go index ec7a92d4..8720633b 100644 --- a/channels/shangwutong/internal/swt/client_test.go +++ b/channels/shangwutong/internal/swt/client_test.go @@ -43,6 +43,17 @@ func TestXSTSignatureKnownVector(t *testing.T) { } } +func TestXSTReceptionSignatureKnownVector(t *testing.T) { + route := XSTRoute{SID: "sid-1", CID: "cid-1", Token: "xst-1", State: "5", KFName: "客服"} + deviceID := xstDeviceID(testSession()) + if deviceID != "3352A6B5AE0FA4ED75745D33D454E3E61A1E860B" { + t.Fatalf("device ID = %q", deviceID) + } + if got := xstReceptionSignature(638900000000000000, testSession(), route, deviceID, xstClientVersion); got != "81a5ace9d65e85f519af28482ba6a200" { + t.Fatalf("signature = %q", got) + } +} + func TestNativeXSTTicksKnownVector(t *testing.T) { location, err := time.LoadLocation("Asia/Shanghai") if err != nil { @@ -81,9 +92,38 @@ func TestClientSendXSTTextMatchesNativeWireProtocol(t *testing.T) { } } +func TestClientSyncXSTReceptionMatchesNativeWireProtocol(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + query := request.URL.Query() + if request.URL.Path != "/api/SwtSyncKfInfo.ashx" || query.Get("sign") == "" || query.Get("t") == "" || + query.Get("swtidhead") != "99917999" || query.Get("onamehead") != "agent" || query.Get("kfversionhead") != xstClientVersion { + t.Fatalf("XST reception sync URL = %s", request.URL.String()) + } + if err := request.ParseForm(); err != nil { + t.Fatal(err) + } + for key, want := range map[string]string{ + "cid": "cid-1", "sid": "sid-1", "clienttype": "2", "clientkind": "3", + "imeiaddress": "3352A6B5AE0FA4ED75745D33D454E3E61A1E860B", "accounttype": "2", + "state": "5", "oname": "agent", "siteid": "99917999", "kfversion": xstClientVersion, "RESET": "", + } { + if got := request.Form.Get(key); got != want { + t.Fatalf("form[%s] = %q, want %q", key, got, want) + } + } + _, _ = response.Write([]byte(`{"msg":"ok"}`)) + })) + defer server.Close() + client := NewClient(rewriteTransportClient(server.URL)) + client.now = func() time.Time { return time.Date(2026, time.August, 16, 12, 0, 0, 0, time.FixedZone("CST", 8*60*60)) } + if _, err := client.SyncXSTReception(context.Background(), testSession(), XSTRoute{SID: "sid-1", CID: "cid-1", Token: "xst-1", State: "5", KFName: "客服"}); err != nil { + t.Fatal(err) + } +} + func TestClientSendXSTTextRejectsNonOK(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) { - _, _ = response.Write([]byte(`{"msg":"failed"}`)) + _, _ = response.Write([]byte(`{"msg":"failed for sid-1 xst-1\nplease sync reception"}`)) })) defer server.Close() _, err := NewClient(rewriteTransportClient(server.URL)).SendXSTText(context.Background(), testSession(), XSTRoute{ @@ -93,6 +133,9 @@ func TestClientSendXSTTextRejectsNonOK(t *testing.T) { if !errors.As(err, &protocolErr) || protocolErr.Code != "xst_rejected" || !protocolErr.Retryable || protocolErr.Uncertain { t.Fatalf("error = %#v", err) } + if detail := protocolErr.Error(); !strings.Contains(detail, "please sync reception") || strings.Contains(detail, "sid-1") || strings.Contains(detail, "xst-1") || strings.Contains(detail, "\n") { + t.Fatalf("error detail = %q", detail) + } } func TestClientHeartbeatParsesEvents(t *testing.T) { diff --git a/channels/shangwutong/sqlc.yaml b/channels/shangwutong/sqlc.yaml index 12dd67a8..d14e4be1 100644 --- a/channels/shangwutong/sqlc.yaml +++ b/channels/shangwutong/sqlc.yaml @@ -4,6 +4,7 @@ sql: schema: - "db/migrations/001_init.up.sql" - "db/migrations/006_add_xst_outbound_stages.up.sql" + - "db/migrations/007_add_xst_reception_sync.up.sql" queries: "db/queries" gen: go: