diff --git a/channels/shangwutong/db/generated/inbound.sql.go b/channels/shangwutong/db/generated/inbound.sql.go index 67ee10aa..c195950f 100644 --- a/channels/shangwutong/db/generated/inbound.sql.go +++ b/channels/shangwutong/db/generated/inbound.sql.go @@ -134,7 +134,7 @@ func (q *Queries) FailInboundEvent(ctx context.Context, arg FailInboundEventPara } const getConversationMap = `-- name: GetConversationMap :one -SELECT account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, gochat_conversation_id, gochat_display_id, swt_assignee_name, created_at, updated_at 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 } diff --git a/channels/shangwutong/db/generated/models.go b/channels/shangwutong/db/generated/models.go index 4e1940a2..ea2bfb90 100644 --- a/channels/shangwutong/db/generated/models.go +++ b/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 { diff --git a/channels/shangwutong/db/generated/outbound.sql.go b/channels/shangwutong/db/generated/outbound.sql.go index 2b95f10a..ae794be4 100644 --- a/channels/shangwutong/db/generated/outbound.sql.go +++ b/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', diff --git a/channels/shangwutong/db/migrations/006_add_xst_outbound_stages.down.sql b/channels/shangwutong/db/migrations/006_add_xst_outbound_stages.down.sql new file mode 100644 index 00000000..f4570cc4 --- /dev/null +++ b/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; diff --git a/channels/shangwutong/db/migrations/006_add_xst_outbound_stages.up.sql b/channels/shangwutong/db/migrations/006_add_xst_outbound_stages.up.sql new file mode 100644 index 00000000..4e94ee48 --- /dev/null +++ b/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')); diff --git a/channels/shangwutong/db/queries/inbound.sql b/channels/shangwutong/db/queries/inbound.sql index ef02d36c..cb14fd6d 100644 --- a/channels/shangwutong/db/queries/inbound.sql +++ b/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, diff --git a/channels/shangwutong/db/queries/outbound.sql b/channels/shangwutong/db/queries/outbound.sql index 293f0245..f2dba60d 100644 --- a/channels/shangwutong/db/queries/outbound.sql +++ b/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 diff --git a/channels/shangwutong/internal/command/root_test.go b/channels/shangwutong/internal/command/root_test.go index 7428397f..ca19eff9 100644 --- a/channels/shangwutong/internal/command/root_test.go +++ b/channels/shangwutong/internal/command/root_test.go @@ -87,7 +87,7 @@ func TestOperationalCommands(t *testing.T) { t.Fatalf("%v produced no output", args) } } - if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 5 { + if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 6 { t.Fatalf("backup version = %d, %v", version, err) } diff --git a/channels/shangwutong/internal/delivery/inbound.go b/channels/shangwutong/internal/delivery/inbound.go index 610a44f6..d8054879 100644 --- a/channels/shangwutong/internal/delivery/inbound.go +++ b/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) } diff --git a/channels/shangwutong/internal/delivery/outbound.go b/channels/shangwutong/internal/delivery/outbound.go index b720dd98..7ae04181 100644 --- a/channels/shangwutong/internal/delivery/outbound.go +++ b/channels/shangwutong/internal/delivery/outbound.go @@ -32,6 +32,7 @@ type SessionProvider interface { type MessageSender interface { SendText(context.Context, swt.Session, string, string) (swt.SendResult, error) + 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 { diff --git a/channels/shangwutong/internal/delivery/outbound_test.go b/channels/shangwutong/internal/delivery/outbound_test.go index d5fd5e8f..d3d854f7 100644 --- a/channels/shangwutong/internal/delivery/outbound_test.go +++ b/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 diff --git a/channels/shangwutong/internal/store/heartbeat.go b/channels/shangwutong/internal/store/heartbeat.go index 72d1de03..fc79926d 100644 --- a/channels/shangwutong/internal/store/heartbeat.go +++ b/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: diff --git a/channels/shangwutong/internal/store/heartbeat_test.go b/channels/shangwutong/internal/store/heartbeat_test.go index 5bf05a02..358821ab 100644 --- a/channels/shangwutong/internal/store/heartbeat_test.go +++ b/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) { diff --git a/channels/shangwutong/internal/store/store_test.go b/channels/shangwutong/internal/store/store_test.go index 91cb7856..0ec86cab 100644 --- a/channels/shangwutong/internal/store/store_test.go +++ b/channels/shangwutong/internal/store/store_test.go @@ -620,7 +620,7 @@ func TestOnlineBackupCanBeOpenedReadOnly(t *testing.T) { t.Fatalf("backup mode = %v", info.Mode().Perm()) } version, err := InspectDatabase(ctx, backup) - if err != nil || version != 5 { + if err != nil || version != 6 { t.Fatalf("backup version = %d, %v", version, err) } } diff --git a/channels/shangwutong/internal/swt/client.go b/channels/shangwutong/internal/swt/client.go index db59b374..b1c2b54f 100644 --- a/channels/shangwutong/internal/swt/client.go +++ b/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 { diff --git a/channels/shangwutong/internal/swt/client_test.go b/channels/shangwutong/internal/swt/client_test.go index cf1a8b44..ec7a92d4 100644 --- a/channels/shangwutong/internal/swt/client_test.go +++ b/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") diff --git a/channels/shangwutong/internal/swt/types.go b/channels/shangwutong/internal/swt/types.go index 383d4db8..56e19dae 100644 --- a/channels/shangwutong/internal/swt/types.go +++ b/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") diff --git a/channels/shangwutong/sqlc.yaml b/channels/shangwutong/sqlc.yaml index b02585b9..12dd67a8 100644 --- a/channels/shangwutong/sqlc.yaml +++ b/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: