H-162: sync XST reception before outbound delivery (#39)

Co-authored-by: Rogee <rogee@ipao.vip>

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
 }

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 {

channels/shangwutong/db/migrations/007_add_xst_reception_sync.down.sql

@@ -0,0 +1 @@
+ALTER TABLE conversation_maps DROP COLUMN xst_sync_key;

channels/shangwutong/db/migrations/007_add_xst_reception_sync.up.sql

@@ -0,0 +1 @@
+ALTER TABLE conversation_maps ADD COLUMN xst_sync_key TEXT;

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,

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)
 	}
 

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
 	}

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
 }

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)
 	}
 }

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

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) {

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: