H-162: add recoverable XST outbound delivery (#37)

* H-162: add recoverable XST outbound delivery

* H-162: close XST echo delivery races

* H-162: match echoes during XST retry

---------

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 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 FROM conversation_maps
 WHERE account_id = ? AND swt_sid = ?
 LIMIT 1
 `
@@ -157,6 +157,10 @@ func (q *Queries) GetConversationMap(ctx context.Context, arg GetConversationMap
 		&i.SwtAssigneeName,
 		&i.CreatedAt,
 		&i.UpdatedAt,
+		&i.XstRequired,
+		&i.XstToken,
+		&i.XstCid,
+		&i.XstKfname,
 	)
 	return &i, err
 }
@@ -422,7 +426,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
+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
 `
 
 type UpsertConversationMapParams struct {
@@ -456,6 +460,63 @@ func (q *Queries) UpsertConversationMap(ctx context.Context, arg UpsertConversat
 		&i.SwtAssigneeName,
 		&i.CreatedAt,
 		&i.UpdatedAt,
+		&i.XstRequired,
+		&i.XstToken,
+		&i.XstCid,
+		&i.XstKfname,
+	)
+	return &i, err
+}
+
+const upsertConversationXSTRoute = `-- name: UpsertConversationXSTRoute :one
+INSERT INTO conversation_maps (
+  account_id, swt_sid, gochat_contact_source_id,
+  xst_required, xst_token, xst_cid, xst_kfname
+) VALUES (?, ?, ?, ?, ?, ?, ?)
+ON CONFLICT(account_id, swt_sid) DO UPDATE SET
+  xst_required = CASE WHEN excluded.xst_required = 1 THEN 1 ELSE conversation_maps.xst_required END,
+  xst_token = COALESCE(excluded.xst_token, conversation_maps.xst_token),
+  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
+`
+
+type UpsertConversationXSTRouteParams struct {
+	AccountID             int64   `json:"account_id"`
+	SwtSid                string  `json:"swt_sid"`
+	GochatContactSourceID string  `json:"gochat_contact_source_id"`
+	XstRequired           int64   `json:"xst_required"`
+	XstToken              *string `json:"xst_token"`
+	XstCid                *string `json:"xst_cid"`
+	XstKfname             *string `json:"xst_kfname"`
+}
+
+func (q *Queries) UpsertConversationXSTRoute(ctx context.Context, arg UpsertConversationXSTRouteParams) (*ConversationMap, error) {
+	row := q.db.QueryRowContext(ctx, upsertConversationXSTRoute,
+		arg.AccountID,
+		arg.SwtSid,
+		arg.GochatContactSourceID,
+		arg.XstRequired,
+		arg.XstToken,
+		arg.XstCid,
+		arg.XstKfname,
+	)
+	var i ConversationMap
+	err := row.Scan(
+		&i.AccountID,
+		&i.SwtSid,
+		&i.GochatContactSourceID,
+		&i.GochatContactID,
+		&i.GochatConversationID,
+		&i.GochatDisplayID,
+		&i.SwtAssigneeName,
+		&i.CreatedAt,
+		&i.UpdatedAt,
+		&i.XstRequired,
+		&i.XstToken,
+		&i.XstCid,
+		&i.XstKfname,
 	)
 	return &i, err
 }

channels/shangwutong/db/generated/models.go

@@ -55,6 +55,10 @@ type ConversationMap struct {
 	SwtAssigneeName       *string   `json:"swt_assignee_name"`
 	CreatedAt             time.Time `json:"created_at"`
 	UpdatedAt             time.Time `json:"updated_at"`
+	XstRequired           int64     `json:"xst_required"`
+	XstToken              *string   `json:"xst_token"`
+	XstCid                *string   `json:"xst_cid"`
+	XstKfname             *string   `json:"xst_kfname"`
 }
 
 type InboundEvent struct {
@@ -121,6 +125,8 @@ type OutboundMessage struct {
 	LastError          *string    `json:"last_error"`
 	CreatedAt          time.Time  `json:"created_at"`
 	UpdatedAt          time.Time  `json:"updated_at"`
+	SwtSentAt          *time.Time `json:"swt_sent_at"`
+	XstDeliveryStatus  string     `json:"xst_delivery_status"`
 }
 
 type OutboundOperation struct {

channels/shangwutong/db/generated/outbound.sql.go

@@ -85,7 +85,7 @@ WHERE id = (
   LIMIT 1
 )
 AND delivery_status = 'pending'
-RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at
+RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at, swt_sent_at, xst_delivery_status
 `
 
 func (q *Queries) ClaimOutboundMessage(ctx context.Context) (*OutboundMessage, error) {
@@ -116,10 +116,29 @@ func (q *Queries) ClaimOutboundMessage(ctx context.Context) (*OutboundMessage, e
 		&i.LastError,
 		&i.CreatedAt,
 		&i.UpdatedAt,
+		&i.SwtSentAt,
+		&i.XstDeliveryStatus,
 	)
 	return &i, err
 }
 
+const claimOutboundXST = `-- name: ClaimOutboundXST :execrows
+UPDATE outbound_messages SET
+  xst_delivery_status = 'delivering',
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND xst_delivery_status = 'pending'
+`
+
+func (q *Queries) ClaimOutboundXST(ctx context.Context, id int64) (int64, error) {
+	result, err := q.db.ExecContext(ctx, claimOutboundXST, id)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
 const claimStatusSync = `-- name: ClaimStatusSync :one
 UPDATE outbound_messages SET
   status_sync_status = 'syncing',
@@ -133,7 +152,7 @@ WHERE id = (
   LIMIT 1
 )
 AND status_sync_status = 'pending'
-RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at
+RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at, swt_sent_at, xst_delivery_status
 `
 
 func (q *Queries) ClaimStatusSync(ctx context.Context) (*OutboundMessage, error) {
@@ -164,6 +183,8 @@ func (q *Queries) ClaimStatusSync(ctx context.Context) (*OutboundMessage, error)
 		&i.LastError,
 		&i.CreatedAt,
 		&i.UpdatedAt,
+		&i.SwtSentAt,
+		&i.XstDeliveryStatus,
 	)
 	return &i, err
 }
@@ -210,6 +231,54 @@ func (q *Queries) CompleteOutboundPart(ctx context.Context, arg CompleteOutbound
 	return err
 }
 
+const completeOutboundXST = `-- name: CompleteOutboundXST :execrows
+UPDATE outbound_messages SET
+  xst_delivery_status = 'delivered',
+  delivery_status = CASE
+    WHEN NOT EXISTS (
+      SELECT 1 FROM outbound_parts
+      WHERE outbound_message_id = outbound_messages.id
+        AND delivery_status <> 'delivered'
+    ) THEN 'delivered'
+    ELSE delivery_status
+  END,
+  result_version = result_version + CASE
+    WHEN delivery_status <> 'delivered' AND NOT EXISTS (
+      SELECT 1 FROM outbound_parts
+      WHERE outbound_message_id = outbound_messages.id
+        AND delivery_status <> 'delivered'
+    ) THEN 1 ELSE 0 END,
+  status_sync_status = CASE
+    WHEN NOT EXISTS (
+      SELECT 1 FROM outbound_parts
+      WHERE outbound_message_id = outbound_messages.id
+        AND delivery_status <> 'delivered'
+    ) THEN 'pending'
+    ELSE status_sync_status
+  END,
+  status_sync_next_at = NULL,
+  external_error_code = NULL,
+  last_error = NULL,
+  claimed_at = CASE
+    WHEN NOT EXISTS (
+      SELECT 1 FROM outbound_parts
+      WHERE outbound_message_id = outbound_messages.id
+        AND delivery_status <> 'delivered'
+    ) THEN NULL ELSE claimed_at END,
+  updated_at = CURRENT_TIMESTAMP
+WHERE outbound_messages.id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND xst_delivery_status = 'delivering'
+`
+
+func (q *Queries) CompleteOutboundXST(ctx context.Context, id int64) (int64, error) {
+	result, err := q.db.ExecContext(ctx, completeOutboundXST, id)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
 const completeStatusSync = `-- name: CompleteStatusSync :exec
 UPDATE outbound_messages SET
   status_sync_status = 'synced',
@@ -226,18 +295,18 @@ func (q *Queries) CompleteStatusSync(ctx context.Context, id int64) error {
 const confirmOutboundEcho = `-- name: ConfirmOutboundEcho :execrows
 UPDATE outbound_messages SET
   delivery_status = CASE
-    WHEN EXISTS (
+    WHEN xst_delivery_status NOT IN ('not_required', 'delivered') OR EXISTS (
       SELECT 1 FROM outbound_parts
       WHERE outbound_message_id = outbound_messages.id
         AND delivery_status <> 'delivered'
-    ) THEN 'uncertain'
+    ) THEN CASE WHEN delivery_status = 'pending' THEN 'pending' ELSE 'uncertain' END
     ELSE 'delivered'
   END,
   result_version = result_version + 1,
   external_id = ?1,
   external_error_code = NULL,
   status_sync_status = CASE
-    WHEN EXISTS (
+    WHEN xst_delivery_status NOT IN ('not_required', 'delivered') OR EXISTS (
       SELECT 1 FROM outbound_parts
       WHERE outbound_message_id = outbound_messages.id
         AND delivery_status <> 'delivered'
@@ -249,7 +318,10 @@ UPDATE outbound_messages SET
   last_error = NULL,
   updated_at = CURRENT_TIMESTAMP
 WHERE outbound_messages.id = ?2
-  AND delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
+  AND (
+    delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
+    OR (delivery_status = 'pending' AND swt_sent_at IS NOT NULL AND xst_delivery_status = 'pending')
+  )
   AND (delivery_status <> 'failed' OR external_error_code = 'uncertain_timeout')
 `
 
@@ -352,8 +424,36 @@ func (q *Queries) FailOutboundPart(ctx context.Context, arg FailOutboundPartPara
 	return err
 }
 
+const failOutboundXST = `-- name: FailOutboundXST :execrows
+UPDATE outbound_messages SET
+  delivery_status = 'failed',
+  xst_delivery_status = 'failed',
+  result_version = result_version + 1,
+  external_error_code = ?,
+  status_sync_status = 'pending',
+  claimed_at = NULL,
+  last_error = ?,
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+`
+
+type FailOutboundXSTParams struct {
+	ExternalErrorCode *string `json:"external_error_code"`
+	LastError         *string `json:"last_error"`
+	ID                int64   `json:"id"`
+}
+
+func (q *Queries) FailOutboundXST(ctx context.Context, arg FailOutboundXSTParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, failOutboundXST, arg.ExternalErrorCode, arg.LastError, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
 const getOutboundByGoChatMessageID = `-- name: GetOutboundByGoChatMessageID :one
-SELECT id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at FROM outbound_messages WHERE gochat_message_id = ? LIMIT 1
+SELECT id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at, swt_sent_at, xst_delivery_status FROM outbound_messages WHERE gochat_message_id = ? LIMIT 1
 `
 
 func (q *Queries) GetOutboundByGoChatMessageID(ctx context.Context, gochatMessageID int64) (*OutboundMessage, error) {
@@ -384,6 +484,8 @@ func (q *Queries) GetOutboundByGoChatMessageID(ctx context.Context, gochatMessag
 		&i.LastError,
 		&i.CreatedAt,
 		&i.UpdatedAt,
+		&i.SwtSentAt,
+		&i.XstDeliveryStatus,
 	)
 	return &i, err
 }
@@ -394,7 +496,7 @@ INSERT INTO outbound_messages (
   message_type, content, payload
 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
 ON CONFLICT(gochat_message_id) DO NOTHING
-RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at
+RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at, swt_sent_at, xst_delivery_status
 `
 
 type InsertOutboundMessageParams struct {
@@ -447,6 +549,8 @@ func (q *Queries) InsertOutboundMessage(ctx context.Context, arg InsertOutboundM
 		&i.LastError,
 		&i.CreatedAt,
 		&i.UpdatedAt,
+		&i.SwtSentAt,
+		&i.XstDeliveryStatus,
 	)
 	return &i, err
 }
@@ -505,10 +609,13 @@ func (q *Queries) InsertOutboundPart(ctx context.Context, arg InsertOutboundPart
 }
 
 const listOutboundEchoCandidates = `-- name: ListOutboundEchoCandidates :many
-SELECT id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at FROM outbound_messages
+SELECT id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at, swt_sent_at, xst_delivery_status FROM outbound_messages
 WHERE account_id = ?
   AND swt_sid = ?
-  AND delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
+  AND (
+    delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
+    OR (delivery_status = 'pending' AND swt_sent_at IS NOT NULL AND xst_delivery_status = 'pending')
+  )
   AND (delivery_status <> 'failed' OR external_error_code = 'uncertain_timeout')
   AND EXISTS (
     SELECT 1 FROM outbound_parts
@@ -559,6 +666,8 @@ func (q *Queries) ListOutboundEchoCandidates(ctx context.Context, arg ListOutbou
 			&i.LastError,
 			&i.CreatedAt,
 			&i.UpdatedAt,
+			&i.SwtSentAt,
+			&i.XstDeliveryStatus,
 		); err != nil {
 			return nil, err
 		}
@@ -636,6 +745,23 @@ func (q *Queries) MarkOutboundPartUncertain(ctx context.Context, arg MarkOutboun
 	return err
 }
 
+const markOutboundSWTSent = `-- name: MarkOutboundSWTSent :execrows
+UPDATE outbound_messages SET
+  swt_sent_at = CURRENT_TIMESTAMP,
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND swt_sent_at IS NULL
+`
+
+func (q *Queries) MarkOutboundSWTSent(ctx context.Context, id int64) (int64, error) {
+	result, err := q.db.ExecContext(ctx, markOutboundSWTSent, id)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
 const markOutboundUncertain = `-- name: MarkOutboundUncertain :exec
 UPDATE outbound_messages SET
   delivery_status = 'uncertain',
@@ -659,9 +785,42 @@ func (q *Queries) MarkOutboundUncertain(ctx context.Context, arg MarkOutboundUnc
 	return err
 }
 
+const markOutboundXSTUncertain = `-- name: MarkOutboundXSTUncertain :execrows
+UPDATE outbound_messages SET
+  delivery_status = 'uncertain',
+  xst_delivery_status = 'uncertain',
+  result_version = result_version + 1,
+  external_error_code = ?,
+  status_sync_status = 'pending',
+  claimed_at = NULL,
+  last_error = ?,
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND xst_delivery_status = 'delivering'
+`
+
+type MarkOutboundXSTUncertainParams struct {
+	ExternalErrorCode *string `json:"external_error_code"`
+	LastError         *string `json:"last_error"`
+	ID                int64   `json:"id"`
+}
+
+func (q *Queries) MarkOutboundXSTUncertain(ctx context.Context, arg MarkOutboundXSTUncertainParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, markOutboundXSTUncertain, arg.ExternalErrorCode, arg.LastError, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
 const recoverOutboundDeliveriesAsUncertain = `-- name: RecoverOutboundDeliveriesAsUncertain :execrows
 UPDATE outbound_messages SET
   delivery_status = 'uncertain',
+  xst_delivery_status = CASE
+    WHEN xst_delivery_status = 'delivering' THEN 'uncertain'
+    ELSE xst_delivery_status
+  END,
   result_version = result_version + 1,
   external_error_code = 'connector_restart_uncertain',
   status_sync_status = 'pending',
@@ -723,6 +882,23 @@ func (q *Queries) RecoverOutboundStatusSyncs(ctx context.Context) (int64, error)
 	return result.RowsAffected()
 }
 
+const requireOutboundXST = `-- name: RequireOutboundXST :execrows
+UPDATE outbound_messages SET
+  xst_delivery_status = 'pending',
+  updated_at = CURRENT_TIMESTAMP
+WHERE outbound_messages.id = ?
+  AND delivery_status = 'delivering'
+  AND xst_delivery_status = 'not_required'
+`
+
+func (q *Queries) RequireOutboundXST(ctx context.Context, id int64) (int64, error) {
+	result, err := q.db.ExecContext(ctx, requireOutboundXST, id)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
 const resetUndeliveredOutboundParts = `-- name: ResetUndeliveredOutboundParts :exec
 UPDATE outbound_parts SET
   delivery_status = 'pending',
@@ -768,6 +944,10 @@ UPDATE outbound_messages SET
   content = ?5,
   payload = ?6,
   delivery_status = 'pending',
+  xst_delivery_status = CASE
+    WHEN xst_delivery_status IN ('failed', 'uncertain') THEN 'pending'
+    ELSE xst_delivery_status
+  END,
   status_sync_status = 'not_required',
   next_attempt_at = NULL,
   claimed_at = NULL,
@@ -804,6 +984,33 @@ func (q *Queries) RetryOutboundMessage(ctx context.Context, arg RetryOutboundMes
 	return result.RowsAffected()
 }
 
+const retryOutboundXST = `-- name: RetryOutboundXST :execrows
+UPDATE outbound_messages SET
+  delivery_status = 'pending',
+  xst_delivery_status = 'pending',
+  claimed_at = NULL,
+  next_attempt_at = ?,
+  last_error = ?,
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND xst_delivery_status = 'delivering'
+`
+
+type RetryOutboundXSTParams struct {
+	NextAttemptAt *time.Time `json:"next_attempt_at"`
+	LastError     *string    `json:"last_error"`
+	ID            int64      `json:"id"`
+}
+
+func (q *Queries) RetryOutboundXST(ctx context.Context, arg RetryOutboundXSTParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, retryOutboundXST, arg.NextAttemptAt, arg.LastError, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
 const retryStatusSync = `-- name: RetryStatusSync :exec
 UPDATE outbound_messages SET
   status_sync_status = 'pending',

channels/shangwutong/db/migrations/006_add_xst_outbound_stages.down.sql

@@ -0,0 +1,7 @@
+ALTER TABLE outbound_messages DROP COLUMN xst_delivery_status;
+ALTER TABLE outbound_messages DROP COLUMN swt_sent_at;
+
+ALTER TABLE conversation_maps DROP COLUMN xst_kfname;
+ALTER TABLE conversation_maps DROP COLUMN xst_cid;
+ALTER TABLE conversation_maps DROP COLUMN xst_token;
+ALTER TABLE conversation_maps DROP COLUMN xst_required;

channels/shangwutong/db/migrations/006_add_xst_outbound_stages.up.sql

@@ -0,0 +1,8 @@
+ALTER TABLE conversation_maps ADD COLUMN xst_required INTEGER NOT NULL DEFAULT 0 CHECK(xst_required IN (0, 1));
+ALTER TABLE conversation_maps ADD COLUMN xst_token TEXT;
+ALTER TABLE conversation_maps ADD COLUMN xst_cid TEXT;
+ALTER TABLE conversation_maps ADD COLUMN xst_kfname TEXT;
+
+ALTER TABLE outbound_messages ADD COLUMN swt_sent_at DATETIME;
+ALTER TABLE outbound_messages ADD COLUMN xst_delivery_status TEXT NOT NULL DEFAULT 'not_required'
+  CHECK(xst_delivery_status IN ('not_required', 'pending', 'delivering', 'delivered', 'uncertain', 'failed'));

channels/shangwutong/db/queries/inbound.sql

@@ -88,6 +88,19 @@ WHERE account_id = ? AND swt_sid = ? AND kind = 0
 ORDER BY id DESC
 LIMIT 1;
 
+-- name: UpsertConversationXSTRoute :one
+INSERT INTO conversation_maps (
+  account_id, swt_sid, gochat_contact_source_id,
+  xst_required, xst_token, xst_cid, xst_kfname
+) VALUES (?, ?, ?, ?, ?, ?, ?)
+ON CONFLICT(account_id, swt_sid) DO UPDATE SET
+  xst_required = CASE WHEN excluded.xst_required = 1 THEN 1 ELSE conversation_maps.xst_required END,
+  xst_token = COALESCE(excluded.xst_token, conversation_maps.xst_token),
+  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 *;
+
 -- name: UpsertConversationMap :one
 INSERT INTO conversation_maps (
   account_id, swt_sid, gochat_contact_source_id, gochat_contact_id,

channels/shangwutong/db/queries/outbound.sql

@@ -129,6 +129,108 @@ UPDATE outbound_messages SET
   updated_at = CURRENT_TIMESTAMP
 WHERE id = ? AND delivery_status = 'delivering';
 
+-- name: RequireOutboundXST :execrows
+UPDATE outbound_messages SET
+  xst_delivery_status = 'pending',
+  updated_at = CURRENT_TIMESTAMP
+WHERE outbound_messages.id = ?
+  AND delivery_status = 'delivering'
+  AND xst_delivery_status = 'not_required';
+
+-- name: MarkOutboundSWTSent :execrows
+UPDATE outbound_messages SET
+  swt_sent_at = CURRENT_TIMESTAMP,
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND swt_sent_at IS NULL;
+
+-- name: ClaimOutboundXST :execrows
+UPDATE outbound_messages SET
+  xst_delivery_status = 'delivering',
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND xst_delivery_status = 'pending';
+
+-- name: CompleteOutboundXST :execrows
+UPDATE outbound_messages SET
+  xst_delivery_status = 'delivered',
+  delivery_status = CASE
+    WHEN NOT EXISTS (
+      SELECT 1 FROM outbound_parts
+      WHERE outbound_message_id = outbound_messages.id
+        AND delivery_status <> 'delivered'
+    ) THEN 'delivered'
+    ELSE delivery_status
+  END,
+  result_version = result_version + CASE
+    WHEN delivery_status <> 'delivered' AND NOT EXISTS (
+      SELECT 1 FROM outbound_parts
+      WHERE outbound_message_id = outbound_messages.id
+        AND delivery_status <> 'delivered'
+    ) THEN 1 ELSE 0 END,
+  status_sync_status = CASE
+    WHEN NOT EXISTS (
+      SELECT 1 FROM outbound_parts
+      WHERE outbound_message_id = outbound_messages.id
+        AND delivery_status <> 'delivered'
+    ) THEN 'pending'
+    ELSE status_sync_status
+  END,
+  status_sync_next_at = NULL,
+  external_error_code = NULL,
+  last_error = NULL,
+  claimed_at = CASE
+    WHEN NOT EXISTS (
+      SELECT 1 FROM outbound_parts
+      WHERE outbound_message_id = outbound_messages.id
+        AND delivery_status <> 'delivered'
+    ) THEN NULL ELSE claimed_at END,
+  updated_at = CURRENT_TIMESTAMP
+WHERE outbound_messages.id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND xst_delivery_status = 'delivering';
+
+-- name: RetryOutboundXST :execrows
+UPDATE outbound_messages SET
+  delivery_status = 'pending',
+  xst_delivery_status = 'pending',
+  claimed_at = NULL,
+  next_attempt_at = ?,
+  last_error = ?,
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND xst_delivery_status = 'delivering';
+
+-- name: MarkOutboundXSTUncertain :execrows
+UPDATE outbound_messages SET
+  delivery_status = 'uncertain',
+  xst_delivery_status = 'uncertain',
+  result_version = result_version + 1,
+  external_error_code = ?,
+  status_sync_status = 'pending',
+  claimed_at = NULL,
+  last_error = ?,
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain')
+  AND xst_delivery_status = 'delivering';
+
+-- name: FailOutboundXST :execrows
+UPDATE outbound_messages SET
+  delivery_status = 'failed',
+  xst_delivery_status = 'failed',
+  result_version = result_version + 1,
+  external_error_code = ?,
+  status_sync_status = 'pending',
+  claimed_at = NULL,
+  last_error = ?,
+  updated_at = CURRENT_TIMESTAMP
+WHERE id = ?
+  AND delivery_status IN ('delivering', 'uncertain');
+
 -- name: MarkOutboundUncertain :exec
 UPDATE outbound_messages SET
   delivery_status = 'uncertain',
@@ -169,6 +271,10 @@ UPDATE outbound_messages SET
   content = sqlc.narg(content),
   payload = sqlc.arg(payload),
   delivery_status = 'pending',
+  xst_delivery_status = CASE
+    WHEN xst_delivery_status IN ('failed', 'uncertain') THEN 'pending'
+    ELSE xst_delivery_status
+  END,
   status_sync_status = 'not_required',
   next_attempt_at = NULL,
   claimed_at = NULL,
@@ -181,6 +287,10 @@ WHERE gochat_message_id = sqlc.arg(gochat_message_id)
 -- name: RecoverOutboundDeliveriesAsUncertain :execrows
 UPDATE outbound_messages SET
   delivery_status = 'uncertain',
+  xst_delivery_status = CASE
+    WHEN xst_delivery_status = 'delivering' THEN 'uncertain'
+    ELSE xst_delivery_status
+  END,
   result_version = result_version + 1,
   external_error_code = 'connector_restart_uncertain',
   status_sync_status = 'pending',
@@ -249,7 +359,10 @@ WHERE id = ? AND status_sync_status = 'syncing';
 SELECT * FROM outbound_messages
 WHERE account_id = ?
   AND swt_sid = ?
-  AND delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
+  AND (
+    delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
+    OR (delivery_status = 'pending' AND swt_sent_at IS NOT NULL AND xst_delivery_status = 'pending')
+  )
   AND (delivery_status <> 'failed' OR external_error_code = 'uncertain_timeout')
   AND EXISTS (
     SELECT 1 FROM outbound_parts
@@ -263,18 +376,18 @@ LIMIT 20;
 -- name: ConfirmOutboundEcho :execrows
 UPDATE outbound_messages SET
   delivery_status = CASE
-    WHEN EXISTS (
+    WHEN xst_delivery_status NOT IN ('not_required', 'delivered') OR EXISTS (
       SELECT 1 FROM outbound_parts
       WHERE outbound_message_id = outbound_messages.id
         AND delivery_status <> 'delivered'
-    ) THEN 'uncertain'
+    ) THEN CASE WHEN delivery_status = 'pending' THEN 'pending' ELSE 'uncertain' END
     ELSE 'delivered'
   END,
   result_version = result_version + 1,
   external_id = sqlc.narg(external_id),
   external_error_code = NULL,
   status_sync_status = CASE
-    WHEN EXISTS (
+    WHEN xst_delivery_status NOT IN ('not_required', 'delivered') OR EXISTS (
       SELECT 1 FROM outbound_parts
       WHERE outbound_message_id = outbound_messages.id
         AND delivery_status <> 'delivered'
@@ -286,7 +399,10 @@ UPDATE outbound_messages SET
   last_error = NULL,
   updated_at = CURRENT_TIMESTAMP
 WHERE outbound_messages.id = sqlc.arg(id)
-  AND delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
+  AND (
+    delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
+    OR (delivery_status = 'pending' AND swt_sent_at IS NOT NULL AND xst_delivery_status = 'pending')
+  )
   AND (delivery_status <> 'failed' OR external_error_code = 'uncertain_timeout');
 
 -- name: ConfirmOutboundPartEcho :execrows

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 != 5 {
+	if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 6 {
 		t.Fatalf("backup version = %d, %v", version, err)
 	}
 

channels/shangwutong/internal/delivery/inbound.go

@@ -108,6 +108,12 @@ func (i *Inbound) processInbound(ctx context.Context) (bool, error) {
 		return true, i.deliverRetraction(ctx, account, event, mapped)
 	}
 	if mapped.ContactCID != "" {
+		if _, err := i.store.Writer().UpsertConversationXSTRoute(ctx, dbgen.UpsertConversationXSTRouteParams{
+			AccountID: account.ID, SwtSid: event.SwtSid, GochatContactSourceID: event.SwtSid,
+			XstCid: &mapped.ContactCID,
+		}); err != nil {
+			return true, i.retryOrFail(event, mapped, err)
+		}
 		if err := i.client.UpdateContactChannelMetadata(ctx, account.GochatInboxID, event.SwtSid, mapped.ContactCID); err != nil {
 			return true, i.retryOrFail(event, mapped, 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)
+	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)
 	SendVoice(context.Context, swt.Session, string, swt.Upload) (swt.SendResult, error)
@@ -191,18 +192,40 @@ func (o *Outbound) processOutbound(ctx context.Context) (bool, error) {
 			return true, o.failPart(message, part, "unsupported_outbound_content", fmt.Sprintf("outbound %s is not supported", part.PartType))
 		}
 	}
+	xstRoute, err := o.xstRoute(ctx, message)
+	if err != nil {
+		return true, o.retry(message, err.Error())
+	}
+	if xstRoute != nil {
+		if len(parts) != 1 || parts[0].PartType != "text" {
+			return true, o.fail(message, "unsupported_xst_content", "XST delivery currently requires one text part")
+		}
+		rows, err := o.store.Writer().RequireOutboundXST(ctx, message.ID)
+		if err != nil {
+			return true, err
+		}
+		if rows == 1 {
+			message, err = o.store.Reader().GetOutboundByGoChatMessageID(ctx, message.GochatMessageID)
+			if err != nil {
+				return true, err
+			}
+		}
+	}
 	for _, part := range parts {
-		if part.DeliveryStatus == "delivered" {
+		needsXST := xstRoute != nil && message.XstDeliveryStatus != "delivered"
+		if part.DeliveryStatus == "delivered" && !needsXST {
 			continue
 		}
-		if part.DeliveryStatus != "pending" {
+		if part.DeliveryStatus != "pending" && part.DeliveryStatus != "delivered" {
 			return true, o.fail(message, "outbound_part_state_conflict", "outbound part is not pending")
 		}
-		if err := o.sendPart(ctx, message, part); err != nil {
+		if err := o.sendPart(ctx, message, part, xstRoute); err != nil {
 			return true, o.handlePartError(message, part, err)
 		}
-		if err := o.store.Writer().AwaitOutboundPartEcho(ctx, part.ID); err != nil {
-			return true, err
+		if part.DeliveryStatus == "pending" {
+			if err := o.store.Writer().AwaitOutboundPartEcho(ctx, part.ID); err != nil {
+				return true, err
+			}
 		}
 	}
 	err = o.store.Writer().AwaitOutboundEcho(ctx, message.ID)
@@ -212,7 +235,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) error {
+func (o *Outbound) sendPart(ctx context.Context, message *dbgen.OutboundMessage, part *dbgen.OutboundPart, xstRoute *swt.XSTRoute) error {
 	if part.PartType == "video" || part.PartType == "unsupported" {
 		return &permanentDeliveryError{code: "unsupported_outbound_content", err: fmt.Errorf("outbound %s is not supported", part.PartType)}
 	}
@@ -222,8 +245,34 @@ func (o *Outbound) sendPart(ctx context.Context, message *dbgen.OutboundMessage,
 			content = *part.Content
 		}
 		return o.sessions.WithSession(ctx, message.AccountID, func(session swt.Session) error {
-			_, err := o.sender.SendText(ctx, session, message.SwtSid, content)
-			return err
+			if xstRoute == nil {
+				_, err := o.sender.SendText(ctx, session, message.SwtSid, content)
+				return err
+			}
+			if message.SwtSentAt == nil {
+				if _, err := o.sender.SendText(ctx, session, message.SwtSid, content); err != nil {
+					return err
+				}
+				rows, err := o.store.Writer().MarkOutboundSWTSent(ctx, message.ID)
+				if err != nil || rows != 1 {
+					return &swt.Error{Operation: "persist_swt_stage", Code: "network_result_uncertain", Uncertain: true, Err: errors.Join(err, errors.New("SWT accepted stage was not persisted"))}
+				}
+			}
+			if message.XstDeliveryStatus == "delivered" {
+				return nil
+			}
+			rows, err := o.store.Writer().ClaimOutboundXST(ctx, message.ID)
+			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 {
+				return &xstDeliveryError{err: err}
+			}
+			rows, err = o.store.Writer().CompleteOutboundXST(ctx, message.ID)
+			if err != nil || rows != 1 {
+				return &xstDeliveryError{err: &swt.Error{Operation: "persist_xst_stage", Code: "network_result_uncertain", Uncertain: true, Err: errors.Join(err, errors.New("XST accepted stage was not persisted"))}}
+			}
+			return nil
 		})
 	}
 	if part.DataUrl == nil || strings.TrimSpace(*part.DataUrl) == "" {
@@ -266,7 +315,30 @@ func (o *Outbound) sendPart(ctx context.Context, message *dbgen.OutboundMessage,
 	})
 }
 
+func (o *Outbound) xstRoute(ctx context.Context, message *dbgen.OutboundMessage) (*swt.XSTRoute, 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
+	}
+	if err != nil || mapping.XstRequired == 0 {
+		return nil, err
+	}
+	state, err := o.store.Reader().GetLatestConversationState(ctx, dbgen.GetLatestConversationStateParams{AccountID: message.AccountID, SwtSid: message.SwtSid})
+	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: value(mapping.XstKfname)}
+	if err := route.Validate(); err != nil {
+		return nil, err
+	}
+	return route, nil
+}
+
 func (o *Outbound) handlePartError(message *dbgen.OutboundMessage, part *dbgen.OutboundPart, deliveryErr error) error {
+	var xstErr *xstDeliveryError
+	if errors.As(deliveryErr, &xstErr) {
+		return o.handleXSTError(message, part, xstErr.err)
+	}
 	var permanent *permanentDeliveryError
 	if errors.As(deliveryErr, &permanent) {
 		return o.failPart(message, part, permanent.code, permanent.Error())
@@ -294,6 +366,78 @@ func (o *Outbound) handlePartError(message *dbgen.OutboundMessage, part *dbgen.O
 	return o.failPart(message, part, "delivery_exhausted", deliveryErr.Error())
 }
 
+type xstDeliveryError struct{ err error }
+
+func (e *xstDeliveryError) Error() string { return e.err.Error() }
+func (e *xstDeliveryError) Unwrap() error { return e.err }
+
+func (o *Outbound) handleXSTError(message *dbgen.OutboundMessage, part *dbgen.OutboundPart, deliveryErr error) error {
+	var protocolErr *swt.Error
+	if errors.As(deliveryErr, &protocolErr) {
+		if protocolErr.Uncertain {
+			return o.uncertainXST(message, part, protocolErr.Code, protocolErr.Error())
+		}
+		if protocolErr.Retryable && message.Attempts < 10 {
+			return o.retryXST(message, protocolErr.Error())
+		}
+		return o.failXST(message, part, protocolErr.Code, protocolErr.Error())
+	}
+	if message.Attempts < 10 {
+		return o.retryXST(message, deliveryErr.Error())
+	}
+	return o.failXST(message, part, "delivery_exhausted", deliveryErr.Error())
+}
+
+func (o *Outbound) retryXST(message *dbgen.OutboundMessage, detail string) error {
+	next := time.Now().Add(backoff(message.Attempts, 5*time.Minute))
+	rows, err := o.store.Writer().RetryOutboundXST(context.Background(), dbgen.RetryOutboundXSTParams{
+		NextAttemptAt: &next, LastError: &detail, ID: message.ID,
+	})
+	if err == nil && rows == 1 {
+		o.metrics.Delivery("outbound", "retry")
+		return nil
+	}
+	return errors.Join(err, errors.New("XST retry stage was not persisted"))
+}
+
+func (o *Outbound) uncertainXST(message *dbgen.OutboundMessage, part *dbgen.OutboundPart, code, detail string) error {
+	err := o.store.WithTx(context.Background(), func(queries *dbgen.Queries) error {
+		if err := queries.MarkOutboundPartUncertain(context.Background(), dbgen.MarkOutboundPartUncertainParams{LastError: &detail, ID: part.ID}); err != nil {
+			return err
+		}
+		rows, err := queries.MarkOutboundXSTUncertain(context.Background(), dbgen.MarkOutboundXSTUncertainParams{
+			ExternalErrorCode: &code, LastError: &detail, ID: message.ID,
+		})
+		if err != nil || rows != 1 {
+			return errors.Join(err, errors.New("XST uncertain stage was not persisted"))
+		}
+		return nil
+	})
+	if err == nil {
+		o.metrics.Delivery("outbound", "uncertain")
+	}
+	return err
+}
+
+func (o *Outbound) failXST(message *dbgen.OutboundMessage, part *dbgen.OutboundPart, code, detail string) error {
+	err := o.store.WithTx(context.Background(), func(queries *dbgen.Queries) error {
+		if err := queries.FailOutboundPart(context.Background(), dbgen.FailOutboundPartParams{LastError: &detail, ID: part.ID}); err != nil {
+			return err
+		}
+		rows, err := queries.FailOutboundXST(context.Background(), dbgen.FailOutboundXSTParams{
+			ExternalErrorCode: &code, LastError: &detail, ID: message.ID,
+		})
+		if err != nil || rows != 1 {
+			return errors.Join(err, errors.New("XST failed stage was not persisted"))
+		}
+		return nil
+	})
+	if err == nil {
+		o.metrics.Delivery("outbound", "failed")
+	}
+	return err
+}
+
 func (o *Outbound) uncertainPart(message *dbgen.OutboundMessage, part *dbgen.OutboundPart, code, detail string) error {
 	err := o.store.WithTx(context.Background(), func(queries *dbgen.Queries) error {
 		if err := queries.MarkOutboundPartUncertain(context.Background(), dbgen.MarkOutboundPartUncertainParams{LastError: &detail, ID: part.ID}); err != nil {

channels/shangwutong/internal/delivery/outbound_test.go

@@ -50,6 +50,301 @@ func TestAcceptedOutboundTimesOutThroughExistingFailurePath(t *testing.T) {
 	}
 }
 
+func TestXSTOutboundRequiresBothStagesAndEcho(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, xstCalls := 0, 0
+	sender := senderStub{
+		beforeReturn: func() { primaryCalls++ },
+		xst: func(swt.XSTRoute, string) (swt.SendResult, error) {
+			xstCalls++
+			return swt.SendResult{Status: "ok"}, nil
+		},
+	}
+	results := &resultRecorder{}
+	outbound, _ := NewOutbound(database, sessionStub{}, sender, results, nil, 1)
+	if worked, err := outbound.processOutbound(ctx); err != nil || !worked {
+		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 worked, err := outbound.processStatus(ctx); err != nil || worked || results.result.Status != "" {
+		t.Fatalf("premature status = %v, %v, %#v", worked, err, results)
+	}
+	persistInboundEvent(t, database, account, swt.HeartbeatEvent{
+		SessionID: "visitor", Kind: 3, OpName: "agent", Text: "hello", SeqID: 51,
+		Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 3 agent|hello 51 timestamp",
+	})
+	inbound, _ := NewInbound(database, &inboundRecorder{}, nil, 1)
+	if worked, err := inbound.processInbound(ctx); err != nil || !worked {
+		t.Fatalf("echo = %v, %v", worked, err)
+	}
+	if worked, err := outbound.processStatus(ctx); err != nil || !worked || results.result.Status != "sent" {
+		t.Fatalf("final status = %v, %v, %#v", worked, err, results)
+	}
+}
+
+func TestXSTRetryOnlyRepeatsIncompleteStage(t *testing.T) {
+	ctx := context.Background()
+	database, account := deliveryDatabase(t, ctx)
+	seedXSTRoute(t, database, account)
+	input := 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),
+	}
+	if _, _, err := database.EnqueueOutbound(ctx, input, false); err != nil {
+		t.Fatal(err)
+	}
+	primaryCalls, xstCalls := 0, 0
+	xstUncertain := true
+	sender := senderStub{
+		beforeReturn: func() { primaryCalls++ },
+		xst: func(swt.XSTRoute, string) (swt.SendResult, error) {
+			xstCalls++
+			if xstUncertain {
+				return swt.SendResult{}, &swt.Error{Operation: "send_xst_text", Code: "network_result_uncertain", Uncertain: true}
+			}
+			return swt.SendResult{Status: "ok"}, nil
+		},
+	}
+	results := &resultRecorder{}
+	outbound, _ := NewOutbound(database, sessionStub{}, sender, results, nil, 1)
+	if worked, err := outbound.processOutbound(ctx); err != nil || !worked {
+		t.Fatalf("first delivery = %v, %v", worked, err)
+	}
+	if worked, err := outbound.processStatus(ctx); err != nil || !worked || results.result.Status != "uncertain" {
+		t.Fatalf("xst timeout status = %v, %v, %#v", worked, err, results)
+	}
+	if messages, _, err := outbound.expireUncertain(ctx, time.Now().Add(uncertainObservationWindow+time.Second)); err != nil || messages != 1 {
+		t.Fatalf("expire = %d, %v", messages, err)
+	}
+	input.EventID, input.RetryVersion, input.OccurredAt, input.Payload = "message:77:retry:1", 1, time.Now(), `{"retry":1}`
+	if _, duplicate, err := database.EnqueueOutbound(ctx, input, true); err != nil || duplicate {
+		t.Fatalf("retry enqueue = %v, %v", duplicate, err)
+	}
+	xstUncertain = false
+	if worked, err := outbound.processOutbound(ctx); err != nil || !worked {
+		t.Fatalf("retry delivery = %v, %v", worked, err)
+	}
+	if primaryCalls != 1 || xstCalls != 2 {
+		t.Fatalf("primary calls=%d xst calls=%d", primaryCalls, xstCalls)
+	}
+}
+
+func TestXSTRejectThenEchoRetriesOnlyXST(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)
+	}
+	results := &resultRecorder{}
+	client := &inboundRecorder{}
+	inbound, _ := NewInbound(database, client, nil, 1)
+	primaryCalls, xstCalls := 0, 0
+	sender := senderStub{xst: func(swt.XSTRoute, string) (swt.SendResult, error) {
+		xstCalls++
+		if xstCalls == 1 {
+			return swt.SendResult{Status: "failed"}, &swt.Error{Operation: "send_xst_text", Code: "xst_rejected", Retryable: true}
+		}
+		return swt.SendResult{Status: "ok"}, nil
+	}, beforeReturn: func() { primaryCalls++ }}
+	outbound, _ := NewOutbound(database, sessionStub{}, sender, results, nil, 1)
+	if worked, err := outbound.processOutbound(ctx); err != nil || !worked {
+		t.Fatalf("delivery = %v, %v", worked, err)
+	}
+	queued, err := database.Reader().GetOutboundByGoChatMessageID(ctx, 77)
+	if err != nil || queued.DeliveryStatus != "pending" || queued.XstDeliveryStatus != "pending" || queued.SwtSentAt == nil {
+		t.Fatalf("queued = %#v, %v", queued, err)
+	}
+	if worked, err := outbound.processStatus(ctx); err != nil || worked || results.result.Status != "" {
+		t.Fatalf("status = %v, %v, %#v", worked, err, results)
+	}
+	persistInboundEvent(t, database, account, swt.HeartbeatEvent{
+		SessionID: "visitor", Kind: 3, OpName: "agent", Text: "hello", SeqID: 51,
+		Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 3 agent|hello 51 timestamp",
+	})
+	if worked, err := inbound.processInbound(ctx); err != nil || !worked {
+		t.Fatalf("echo after XST retry persisted = %v, %v", worked, err)
+	}
+	queued, err = database.Reader().GetOutboundByGoChatMessageID(ctx, 77)
+	if err != nil || queued.DeliveryStatus != "pending" || queued.XstDeliveryStatus != "pending" || value(queued.ExternalID) != "51" || len(client.imports) != 0 {
+		t.Fatalf("queued after echo = %#v, imports=%#v, %v", queued, client.imports, 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)
+	}
+	queued, err = database.Reader().GetOutboundByGoChatMessageID(ctx, 77)
+	if err != nil || queued.DeliveryStatus != "delivered" || queued.XstDeliveryStatus != "delivered" || value(queued.ExternalID) != "51" || primaryCalls != 1 || xstCalls != 2 || len(client.imports) != 0 {
+		t.Fatalf("queued=%#v primary=%d xst=%d imports=%#v err=%v", queued, primaryCalls, xstCalls, client.imports, err)
+	}
+	if worked, err := outbound.processStatus(ctx); err != nil || !worked || results.result.Status != "sent" {
+		t.Fatalf("final status = %v, %v, %#v", worked, err, results)
+	}
+}
+
+func TestXSTEchoBeforeSWTReturnStillDeliversXST(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)
+	}
+	client := &inboundRecorder{}
+	inbound, _ := NewInbound(database, client, nil, 1)
+	primaryCalls, xstCalls := 0, 0
+	sender := senderStub{
+		beforeReturn: func() {
+			primaryCalls++
+			persistInboundEvent(t, database, account, swt.HeartbeatEvent{
+				SessionID: "visitor", Kind: 3, OpName: "agent", Text: "hello", SeqID: 51,
+				Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 3 agent|hello 51 timestamp",
+			})
+			if worked, err := inbound.processInbound(ctx); err != nil || !worked {
+				t.Fatalf("process echo during SWT send = %v, %v", worked, err)
+			}
+		},
+		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("delivery = %v, %v", worked, err)
+	}
+	queued, err := database.Reader().GetOutboundByGoChatMessageID(ctx, 77)
+	if err != nil || queued.DeliveryStatus != "delivered" || queued.XstDeliveryStatus != "delivered" || value(queued.ExternalID) != "51" || primaryCalls != 1 || xstCalls != 1 || len(client.imports) != 0 {
+		t.Fatalf("queued=%#v primary=%d xst=%d imports=%#v err=%v", queued, primaryCalls, xstCalls, client.imports, err)
+	}
+}
+
+func TestXSTRejectAfterEchoRetriesOnlyXST(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)
+	}
+	client := &inboundRecorder{}
+	inbound, _ := NewInbound(database, client, nil, 1)
+	primaryCalls, xstCalls := 0, 0
+	sender := senderStub{
+		beforeReturn: func() { primaryCalls++ },
+		xst: func(swt.XSTRoute, string) (swt.SendResult, error) {
+			xstCalls++
+			if xstCalls == 1 {
+				persistInboundEvent(t, database, account, swt.HeartbeatEvent{
+					SessionID: "visitor", Kind: 3, OpName: "agent", Text: "hello", SeqID: 51,
+					Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 3 agent|hello 51 timestamp",
+				})
+				if worked, err := inbound.processInbound(ctx); err != nil || !worked {
+					t.Fatalf("process echo during XST send = %v, %v", worked, err)
+				}
+				return swt.SendResult{Status: "failed"}, &swt.Error{Operation: "send_xst_text", Code: "xst_rejected", Retryable: true}
+			}
+			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.DeliveryStatus != "pending" || queued.XstDeliveryStatus != "pending" || queued.NextAttemptAt == nil || value(queued.ExternalID) != "51" {
+		t.Fatalf("queued after rejection = %#v, %v", queued, 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)
+	}
+	queued, err = database.Reader().GetOutboundByGoChatMessageID(ctx, 77)
+	if err != nil || queued.DeliveryStatus != "delivered" || queued.XstDeliveryStatus != "delivered" || value(queued.ExternalID) != "51" || primaryCalls != 1 || xstCalls != 2 || len(client.imports) != 0 {
+		t.Fatalf("queued=%#v primary=%d xst=%d imports=%#v err=%v", queued, primaryCalls, xstCalls, client.imports, err)
+	}
+}
+
+func TestXSTInFlightStageRecoversAsUncertain(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)
+	}
+	message, err := database.Writer().ClaimOutboundMessage(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if rows, err := database.Writer().RequireOutboundXST(ctx, message.ID); err != nil || rows != 1 {
+		t.Fatalf("require XST = %d, %v", rows, err)
+	}
+	if rows, err := database.Writer().MarkOutboundSWTSent(ctx, message.ID); err != nil || rows != 1 {
+		t.Fatalf("complete SWT stage = %d, %v", rows, err)
+	}
+	if rows, err := database.Writer().ClaimOutboundXST(ctx, message.ID); err != nil || rows != 1 {
+		t.Fatalf("claim XST = %d, %v", rows, err)
+	}
+	if _, err := database.Writer().RecoverOutboundPartDeliveriesAsUncertain(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := database.Writer().RecoverOutboundDeliveriesAsUncertain(ctx); err != nil {
+		t.Fatal(err)
+	}
+	message, err = database.Reader().GetOutboundByGoChatMessageID(ctx, 77)
+	if err != nil || message.DeliveryStatus != "uncertain" || message.XstDeliveryStatus != "uncertain" || message.SwtSentAt == nil {
+		t.Fatalf("recovered message = %#v, %v", message, err)
+	}
+}
+
+func TestNonXSTOutboundKeepsSingleStageFlow(t *testing.T) {
+	ctx := context.Background()
+	database, account := deliveryDatabase(t, ctx)
+	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)
+	}
+	xstCalls := 0
+	sender := senderStub{xst: func(swt.XSTRoute, string) (swt.SendResult, error) {
+		xstCalls++
+		return swt.SendResult{}, nil
+	}}
+	outbound, _ := NewOutbound(database, sessionStub{}, sender, &resultRecorder{}, nil, 1)
+	if worked, err := outbound.processOutbound(ctx); err != nil || !worked {
+		t.Fatalf("delivery = %v, %v", worked, err)
+	}
+	if xstCalls != 0 {
+		t.Fatalf("xst calls = %d", xstCalls)
+	}
+}
+
 func TestOutboundEchoBeforeSendReturnsConfirmsDelivery(t *testing.T) {
 	ctx := context.Background()
 	database, account := deliveryDatabase(t, ctx)
@@ -392,6 +687,26 @@ func deliveryDatabase(t *testing.T, ctx context.Context) (*store.Store, *dbgen.A
 	return database, account
 }
 
+func seedXSTRoute(t *testing.T, database *store.Store, account *dbgen.Account) {
+	t.Helper()
+	ctx := context.Background()
+	events := []swt.HeartbeatEvent{
+		{SessionID: "visitor", Kind: 65, Text: "xst|sbox\x1axst-route-token\x1aquery\x1aword\x1acity\x1a48989266", SeqID: 41, RawLine: "xst profile"},
+		{SessionID: "visitor", Kind: 31, OpName: "agent", Text: "ACT_XST|NotShow|QuDaoVisitorInfoMsg|private", SeqID: 42, RawLine: "xst state"},
+		{SessionID: "visitor", Kind: 0, OpName: "agent", Text: "5", SeqID: 43, RawLine: "chatting"},
+		{SessionID: "visitor", Kind: 24, Text: "cid-1", SeqID: 44, RawLine: "cid"},
+	}
+	if _, err := database.PersistHeartbeat(ctx, account, events); err != nil {
+		t.Fatal(err)
+	}
+	inbound, _ := NewInbound(database, &inboundRecorder{}, nil, 1)
+	for range events {
+		if worked, err := inbound.processInbound(ctx); err != nil || !worked {
+			t.Fatalf("seed XST route = %v, %v", worked, err)
+		}
+	}
+}
+
 func deliveryPayload(t *testing.T, attachments []gochat.WebhookAttachment) string {
 	t.Helper()
 	data, _ := json.Marshal(gochat.MessageWebhookData{
@@ -437,6 +752,7 @@ func (s *sessionRecorder) InvalidateSession(context.Context, int64) error {
 type senderStub struct {
 	err          error
 	beforeReturn func()
+	xst          func(swt.XSTRoute, string) (swt.SendResult, error)
 }
 
 func (s senderStub) SendText(context.Context, swt.Session, string, string) (swt.SendResult, error) {
@@ -458,6 +774,13 @@ func (s senderStub) SendVoice(context.Context, swt.Session, string, swt.Upload)
 	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)
+	}
+	return swt.SendResult{Status: "ok"}, s.err
+}
+
 func (s senderStub) EndConversation(context.Context, swt.Session, string) error { return s.err }
 func (s senderStub) ChangeContactName(context.Context, swt.Session, string, string, string) error {
 	return s.err
@@ -494,6 +817,10 @@ func (*operationSender) SendVoice(context.Context, swt.Session, string, swt.Uplo
 	return swt.SendResult{}, nil
 }
 
+func (*operationSender) SendXSTText(context.Context, swt.Session, swt.XSTRoute, string) (swt.SendResult, error) {
+	return swt.SendResult{}, nil
+}
+
 func (s *operationSender) EndConversation(_ context.Context, _ swt.Session, sid string) error {
 	s.sid = sid
 	return nil
@@ -539,6 +866,10 @@ func (*partialSender) SendVoice(context.Context, swt.Session, string, swt.Upload
 	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
+}
+
 func (*partialSender) EndConversation(context.Context, swt.Session, string) error { return nil }
 func (*partialSender) ChangeContactName(context.Context, swt.Session, string, string, string) error {
 	return nil

channels/shangwutong/internal/store/heartbeat.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 
 	dbgen "github.com/gochat/gochat/channels/shangwutong/db/generated"
 	"github.com/gochat/gochat/channels/shangwutong/internal/swt"
@@ -16,6 +17,14 @@ func (s *Store) PersistHeartbeat(ctx context.Context, account *dbgen.Account, ev
 	err := s.WithTx(ctx, func(queries *dbgen.Queries) error {
 		for _, event := range events {
 			cursor.Apply(event.Kind, event.SeqID)
+			if required, token, kfname, ok := heartbeatXSTRoute(event); ok {
+				if _, err := queries.UpsertConversationXSTRoute(ctx, dbgen.UpsertConversationXSTRouteParams{
+					AccountID: account.ID, SwtSid: event.SessionID, GochatContactSourceID: event.SessionID,
+					XstRequired: required, XstToken: token, XstKfname: kfname,
+				}); err != nil {
+					return err
+				}
+			}
 			status, strategy := "pending", (*string)(nil)
 			if ignoredKind(event.Kind) {
 				status = "ignored"
@@ -61,6 +70,21 @@ func (s *Store) PersistHeartbeat(ctx context.Context, account *dbgen.Account, ev
 	return cursor, nil
 }
 
+func heartbeatXSTRoute(event swt.HeartbeatEvent) (int64, *string, *string, bool) {
+	if event.Kind == 31 && strings.HasPrefix(strings.TrimSpace(event.Text), "ACT_XST|") {
+		return 1, nil, optionalString(strings.TrimSpace(event.OpName)), true
+	}
+	if event.Kind != 65 || !strings.Contains(strings.ToLower(event.Text), "xst") {
+		return 0, nil, nil, false
+	}
+	parts := strings.Split(event.Text, "\x1a")
+	var token *string
+	if len(parts) > 1 {
+		token = optionalString(strings.TrimSpace(parts[1]))
+	}
+	return 1, token, optionalString(strings.TrimSpace(event.OpName)), true
+}
+
 func ignoredKind(kind int) bool {
 	switch kind {
 	case -7, 1, 14, 38, 44, 62:

channels/shangwutong/internal/store/heartbeat_test.go

@@ -3,6 +3,7 @@ package store
 import (
 	"context"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	dbgen "github.com/gochat/gochat/channels/shangwutong/db/generated"
@@ -47,6 +48,38 @@ func TestPersistHeartbeatStoresEventsBeforeCursorAndIsIdempotent(t *testing.T) {
 	}
 }
 
+func TestPersistHeartbeatStoresXSTRouteBySID(t *testing.T) {
+	ctx := context.Background()
+	database, err := Open(ctx, filepath.Join(t.TempDir(), "connector.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer database.Close()
+	account, err := database.Writer().CreateAccount(ctx, dbgen.CreateAccountParams{
+		GochatAccountID: 1, GochatInboxID: 22, GochatInboxIdentifier: "id", ConfigVersion: 1,
+		SessionID: "BYT99917999", Username: "agent", Password: "password", Enabled: 1,
+		DesiredPresence: "online", GochatHmacToken: "hmac", GochatWebhookSecret: "secret",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	profile := strings.Join([]string{"xst|sbox|zhinengzx", "xst-route-token", "query", "word", "city", "48989266"}, "\x1a")
+	events := []swt.HeartbeatEvent{
+		{SessionID: "visitor", Kind: 65, Text: profile, SeqID: 41, RawLine: "xst profile"},
+		{SessionID: "visitor", Kind: 31, OpName: "客服甲", Text: "ACT_XST|NotShow|QuDaoVisitorInfoMsg|private", SeqID: 42, RawLine: "xst state"},
+	}
+	if _, err := database.PersistHeartbeat(ctx, account, events); err != nil {
+		t.Fatal(err)
+	}
+	route, err := database.Reader().GetConversationMap(ctx, dbgen.GetConversationMapParams{AccountID: account.ID, SwtSid: "visitor"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if route.XstRequired != 1 || route.XstToken == nil || *route.XstToken != "xst-route-token" || route.XstKfname == nil || *route.XstKfname != "客服甲" {
+		t.Fatalf("xst route = %#v", route)
+	}
+}
+
 func TestIgnoredKindMatrix(t *testing.T) {
 	for _, kind := range []int{-7, 1, 14, 38, 44, 62} {
 		if !ignoredKind(kind) {

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 != 5 {
+	if err != nil || version != 6 {
 		t.Fatalf("backup version = %d, %v", version, err)
 	}
 }

channels/shangwutong/internal/swt/client.go

@@ -2,7 +2,10 @@ package swt
 
 import (
 	"context"
+	"crypto/md5"
 	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"html"
@@ -19,6 +22,8 @@ import (
 const (
 	defaultRequestTimeout = 30 * time.Second
 	maxProtocolBodyBytes  = 8 << 20
+	xstBaseURL            = "http://swttobaidukf.zoosnet.net/"
+	xstClientVersion      = "android:v102019.0906"
 )
 
 type Error struct {
@@ -40,13 +45,14 @@ func (e *Error) Unwrap() error { return e.Err }
 
 type Client struct {
 	httpClient *http.Client
+	now        func() time.Time
 }
 
 func NewClient(httpClient *http.Client) *Client {
 	if httpClient == nil {
 		httpClient = &http.Client{Timeout: defaultRequestTimeout}
 	}
-	return &Client{httpClient: httpClient}
+	return &Client{httpClient: httpClient, now: time.Now}
 }
 
 func (c *Client) Login(ctx context.Context, credentials Credentials, presence Presence, verificationCode string) (Session, error) {
@@ -182,6 +188,89 @@ func (c *Client) SendText(ctx context.Context, session Session, sid, text string
 	return c.sendHTML(ctx, session, sid, textHTML(text), "send_text")
 }
 
+func (c *Client) SendXSTText(ctx context.Context, session Session, route XSTRoute, text string) (SendResult, error) {
+	if err := session.Validate(); err != nil {
+		return SendResult{}, err
+	}
+	if err := route.Validate(); err != nil {
+		return SendResult{}, err
+	}
+	wireContent := textHTML(text)
+	ticks := nativeXSTTicks(c.now())
+	query := url.Values{
+		"t":             {strconv.FormatInt(ticks, 10)},
+		"sign":          {xstSignature(ticks, session, route, wireContent, xstClientVersion)},
+		"swtidhead":     {session.SiteID},
+		"onamehead":     {session.LoginName},
+		"kfversionhead": {xstClientVersion},
+	}
+	target, err := resolveEndpoint(xstBaseURL, "api/swtsynckfmsg.ashx")
+	if err != nil {
+		return SendResult{}, err
+	}
+	form := url.Values{
+		"html":      {wireContent},
+		"RESET":     {""},
+		"siteid":    {session.SiteID},
+		"oname":     {session.LoginName},
+		"cid":       {route.CID},
+		"sid":       {route.SID},
+		"xst":       {route.Token},
+		"state":     {route.State},
+		"kfname":    {route.KFName},
+		"content":   {javaFormEscape(wireContent)},
+		"msgkind":   {"1"},
+		"kfversion": {xstClientVersion},
+	}
+	body, response, requestWritten, err := c.postFormTargetTracked(ctx, target+"?"+query.Encode(), 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}
+		}
+		if requestWritten {
+			return SendResult{}, &Error{Operation: "send_xst_text", Code: "network_result_uncertain", Uncertain: true, Err: err}
+		}
+		return SendResult{}, &Error{Operation: "send_xst_text", 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}
+	}
+	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}
+}
+
+func javaFormEscape(value string) string {
+	encoded := url.QueryEscape(value)
+	encoded = strings.ReplaceAll(encoded, "%2A", "*")
+	return strings.ReplaceAll(encoded, "~", "%7E")
+}
+
+func nativeXSTTicks(now time.Time) int64 {
+	// The Android client passes the one-based month to Calendar.set and uses
+	// 0001-02-03 on GregorianCalendar's Julian side as its epoch.
+	location := time.FixedZone("CST", 8*60*60)
+	now = now.In(location)
+	target := time.Date(now.Year(), now.Month()+1, now.Day(), now.Hour(), now.Minute(), now.Second(), 0, location)
+	base := time.Date(1, time.February, 3, 0, 0, 0, 0, location)
+	return (target.Unix() - base.Unix() + 2*24*60*60) * 10_000_000
+}
+
+func xstSignature(ticks int64, session Session, route XSTRoute, content, version string) string {
+	separator := string(rune(26))
+	plain := strings.Join([]string{
+		strconv.FormatInt(ticks+10, 10), session.SiteID, session.LoginName, route.CID, route.SID,
+		route.Token, route.State, route.KFName, "1", content, version, "",
+	}, separator)
+	digest := md5.Sum([]byte(plain))
+	return hex.EncodeToString(digest[:])
+}
+
 func (c *Client) sendHTML(ctx context.Context, session Session, sid, content, operation string) (SendResult, error) {
 	if err := session.Validate(); err != nil {
 		return SendResult{}, err
@@ -276,6 +365,10 @@ func (c *Client) postFormTracked(ctx context.Context, baseURL, endpoint string,
 	if err != nil {
 		return "", nil, false, err
 	}
+	return c.postFormTargetTracked(ctx, target, form)
+}
+
+func (c *Client) postFormTargetTracked(ctx context.Context, target string, form url.Values) (string, *http.Response, bool, error) {
 	var requestWritten atomic.Bool
 	trace := &httptrace.ClientTrace{WroteRequest: func(info httptrace.WroteRequestInfo) {
 		if info.Err == nil {

channels/shangwutong/internal/swt/client_test.go

@@ -8,6 +8,7 @@ import (
 	"net/url"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestClientLogin(t *testing.T) {
@@ -35,6 +36,65 @@ func TestClientLogin(t *testing.T) {
 	}
 }
 
+func TestXSTSignatureKnownVector(t *testing.T) {
+	route := XSTRoute{SID: "sid-1", CID: "cid-1", Token: "xst-1", State: "5", KFName: "客服"}
+	if got := xstSignature(638900000000000000, testSession(), route, "hello", xstClientVersion); got != "92872732c501789da6bf1c882f3ede40" {
+		t.Fatalf("signature = %q", got)
+	}
+}
+
+func TestNativeXSTTicksKnownVector(t *testing.T) {
+	location, err := time.LoadLocation("Asia/Shanghai")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := nativeXSTTicks(time.Date(2026, time.August, 16, 12, 0, 0, 0, location)); got != 639224784000000000 {
+		t.Fatalf("ticks = %d", got)
+	}
+}
+
+func TestClientSendXSTTextMatchesNativeWireProtocol(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		query := request.URL.Query()
+		if request.URL.Path != "/api/swtsynckfmsg.ashx" || query.Get("sign") == "" || query.Get("t") == "" ||
+			query.Get("swtidhead") != "99917999" || query.Get("onamehead") != "agent" || query.Get("kfversionhead") != xstClientVersion {
+			t.Fatalf("xst request URL = %s", request.URL.String())
+		}
+		if err := request.ParseForm(); err != nil {
+			t.Fatal(err)
+		}
+		for key, want := range map[string]string{
+			"siteid": "99917999", "oname": "agent", "cid": "cid-1", "sid": "sid-1", "xst": "xst-1",
+			"state": "5", "kfname": "客服", "html": "hello *~", "content": "hello+*%7E", "msgkind": "1", "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.SendXSTText(context.Background(), testSession(), XSTRoute{SID: "sid-1", CID: "cid-1", Token: "xst-1", State: "5", KFName: "客服"}, "hello *~"); 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"}`))
+	}))
+	defer server.Close()
+	_, err := NewClient(rewriteTransportClient(server.URL)).SendXSTText(context.Background(), testSession(), XSTRoute{
+		SID: "sid-1", CID: "cid-1", Token: "xst-1", State: "5", KFName: "客服",
+	}, "hello")
+	var protocolErr *Error
+	if !errors.As(err, &protocolErr) || protocolErr.Code != "xst_rejected" || !protocolErr.Retryable || protocolErr.Uncertain {
+		t.Fatalf("error = %#v", err)
+	}
+}
+
 func TestClientHeartbeatParsesEvents(t *testing.T) {
 	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
 		response.Header().Set("r", "ok")

channels/shangwutong/internal/swt/types.go

@@ -91,6 +91,22 @@ type Session struct {
 	Purview   *uint64
 }
 
+type XSTRoute struct {
+	SID    string
+	CID    string
+	Token  string
+	State  string
+	KFName string
+}
+
+func (r XSTRoute) Validate() error {
+	if strings.TrimSpace(r.SID) == "" || strings.TrimSpace(r.CID) == "" || strings.TrimSpace(r.Token) == "" ||
+		strings.TrimSpace(r.State) == "" || strings.TrimSpace(r.KFName) == "" {
+		return errors.New("complete XST route is required")
+	}
+	return nil
+}
+
 func (s Session) Validate() error {
 	if strings.TrimSpace(s.BaseURL) == "" || strings.TrimSpace(s.SiteID) == "" || strings.TrimSpace(s.LoginName) == "" || s.MAToken == "" {
 		return errors.New("session is incomplete")

channels/shangwutong/sqlc.yaml

@@ -3,6 +3,7 @@ sql:
   - engine: "sqlite"
     schema:
       - "db/migrations/001_init.up.sql"
+      - "db/migrations/006_add_xst_outbound_stages.up.sql"
     queries: "db/queries"
     gen:
       go: