From 9b39a574bd3a7541ffdf1513e3e4d401324a1566 Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 11 Sep 2026 23:09:02 +0800 Subject: [PATCH] fix(shangwutong): harden classification operations --- .../api/v1/shangwutong_connector_handler.go | 10 +- .../v1/shangwutong_connector_handler_test.go | 33 ++- channels/shangwutong/db/generated/models.go | 35 ++- .../db/generated/operations.sql.go | 268 +++++++++++++++++- ..._classification_operation_results.down.sql | 38 +++ ...dd_classification_operation_results.up.sql | 48 ++++ .../shangwutong/db/queries/operations.sql | 149 ++++++++++ .../shangwutong/internal/account/manager.go | 12 +- .../internal/account/manager_test.go | 54 ++++ channels/shangwutong/internal/command/root.go | 3 + .../shangwutong/internal/command/root_test.go | 2 +- .../shangwutong/internal/delivery/outbound.go | 142 +++++++--- .../internal/delivery/outbound_test.go | 173 +++++++++++ .../shangwutong/internal/store/outbound.go | 2 +- .../shangwutong/internal/store/store_test.go | 2 +- channels/shangwutong/sqlc.yaml | 1 + 16 files changed, 904 insertions(+), 68 deletions(-) create mode 100644 channels/shangwutong/db/migrations/008_add_classification_operation_results.down.sql create mode 100644 channels/shangwutong/db/migrations/008_add_classification_operation_results.up.sql diff --git a/backend/internal/handler/api/v1/shangwutong_connector_handler.go b/backend/internal/handler/api/v1/shangwutong_connector_handler.go index 2bb1e22e..ac539578 100644 --- a/backend/internal/handler/api/v1/shangwutong_connector_handler.go +++ b/backend/internal/handler/api/v1/shangwutong_connector_handler.go @@ -389,7 +389,15 @@ func (h *ShangwutongConnectorHandler) UpdateConversationClassification(c *gin.Co return } var conversation model.Conversation - if err := h.db.WithContext(c.Request.Context()).Where("id = ? AND account_id = ?", uint(conversationID), uint(accountID)).First(&conversation).Error; err != nil { + lookup := h.db.WithContext(c.Request.Context()).Where( + "id = ? AND account_id = ? AND (display_id IS NULL OR display_id = 0)", uint(conversationID), uint(accountID), + ).First(&conversation) + if errors.Is(lookup.Error, gorm.ErrRecordNotFound) { + lookup = h.db.WithContext(c.Request.Context()).Where( + "display_id = ? AND account_id = ?", uint(conversationID), uint(accountID), + ).First(&conversation) + } + if lookup.Error != nil { h.connectorError(c, http.StatusNotFound, "not_found", "conversation not found", false) return } diff --git a/backend/internal/handler/api/v1/shangwutong_connector_handler_test.go b/backend/internal/handler/api/v1/shangwutong_connector_handler_test.go index 2820c626..bc154af4 100644 --- a/backend/internal/handler/api/v1/shangwutong_connector_handler_test.go +++ b/backend/internal/handler/api/v1/shangwutong_connector_handler_test.go @@ -21,6 +21,7 @@ import ( "github.com/gochat/gochat/internal/service" "github.com/gochat/gochat/internal/worker" "github.com/stretchr/testify/require" + "gorm.io/datatypes" "gorm.io/driver/sqlite" "gorm.io/gorm" ) @@ -156,6 +157,32 @@ func TestShangwutongConnectorContactMetadataIsScopedAndIdempotent(t *testing.T) require.Contains(t, string(job.Payload), `"contact_name":"Renamed"`) } +func TestShangwutongClassificationUpdateAcceptsConversationDisplayID(t *testing.T) { + router, db, _, inbox, _ := setupShangwutongConnectorAPI(t) + contact := &model.Contact{AccountID: inbox.AccountID, Name: "Visitor"} + require.NoError(t, db.Create(contact).Error) + conversation := &model.Conversation{ + AccountID: inbox.AccountID, InboxID: inbox.ID, ContactID: contact.ID, Status: "open", + DisplayID: uintPointer(9001), + } + require.NoError(t, db.Create(conversation).Error) + require.NoError(t, db.Create(&model.ContactInbox{ + ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor", ChannelMetadata: datatypes.JSON([]byte(`{"cid":"cid-1"}`)), + }).Error) + require.NoError(t, db.Create(&model.ShangwutongClassificationCache{ + InboxID: inbox.ID, ConversationKinds: datatypes.JSON([]byte(`[{"id":"kind-1","name":"Normal"}]`)), + CustomerColorKinds: datatypes.JSON([]byte(`[]`)), SyncStatus: "succeeded", + }).Error) + path := fmt.Sprintf("/api/v1/accounts/%d/conversations/%d/shangwutong-classifications", inbox.AccountID, *conversation.DisplayID) + response := connectorRequest(t, router, "", http.MethodPatch, path, map[string]string{"chat_kind_id": "kind-1"}) + require.Equal(t, http.StatusAccepted, response.Code, response.Body.String()) + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", service.TaskTypeShangwutongWebhookDelivery).Order("id DESC").First(&job).Error) + require.Contains(t, string(job.Payload), fmt.Sprintf(`"conversation_id":%d`, conversation.ID)) +} + +func uintPointer(value uint) *uint { return &value } + func setupShangwutongConnectorAPI(t *testing.T) (*gin.Engine, *gorm.DB, string, *model.Inbox, *model.Inbox) { t.Helper() gin.SetMode(gin.TestMode) @@ -165,7 +192,7 @@ func setupShangwutongConnectorAPI(t *testing.T) (*gin.Engine, *gorm.DB, string, &model.Account{}, &model.Inbox{}, &model.ChannelShangwutongConfig{}, &channelmodel.ChannelAPI{}, &model.PlatformApp{}, &model.AccessToken{}, &model.Permissible{}, &model.Contact{}, &model.Conversation{}, &model.ContactInbox{}, - &model.Message{}, &model.Attachment{}, &model.BackgroundJob{}, + &model.Message{}, &model.Attachment{}, &model.BackgroundJob{}, &model.ShangwutongClassificationCache{}, )) active := true app := &model.PlatformApp{Name: "SWT Connector", Type: "integration", Status: "active", Active: &active, Config: json.RawMessage(`{"connector":"shangwutong"}`)} @@ -198,7 +225,8 @@ func setupShangwutongConnectorAPI(t *testing.T) (*gin.Engine, *gorm.DB, string, inboxes = append(inboxes, inbox) } messageSvc := service.NewMessageService(repository.NewMessageRepo(db), channel.NewDispatcher(), nil) - handler := NewShangwutongConnectorHandler(db, messageSvc) + workers := worker.NewWorkerPool(db) + handler := NewShangwutongConnectorHandler(db, messageSvc, workers) router := gin.New() group := router.Group("/api/v1/connector/shangwutong") group.Use(middleware.ConnectorServiceAuth(db)) @@ -207,6 +235,7 @@ func setupShangwutongConnectorAPI(t *testing.T) (*gin.Engine, *gorm.DB, string, group.PUT("/inboxes/:inbox_id/status", handler.UpdateInboxStatus) group.PUT("/inboxes/:inbox_id/messages/:message_id/status", handler.UpdateMessageStatus) group.PATCH("/inboxes/:inbox_id/contacts/:source_id", handler.UpdateContactMetadata) + router.PATCH("/api/v1/accounts/:account_id/conversations/:conversation_id/shangwutong-classifications", handler.UpdateConversationClassification) return router, db, token, inboxes[0], inboxes[1] } diff --git a/channels/shangwutong/db/generated/models.go b/channels/shangwutong/db/generated/models.go index 1b51bf31..59e2bcf8 100644 --- a/channels/shangwutong/db/generated/models.go +++ b/channels/shangwutong/db/generated/models.go @@ -131,20 +131,27 @@ type OutboundMessage struct { } type OutboundOperation struct { - ID int64 `json:"id"` - AccountID int64 `json:"account_id"` - SwtSid string `json:"swt_sid"` - EventID string `json:"event_id"` - Operation string `json:"operation"` - Payload string `json:"payload"` - OccurredAt time.Time `json:"occurred_at"` - DeliveryStatus string `json:"delivery_status"` - ClaimedAt *time.Time `json:"claimed_at"` - Attempts int64 `json:"attempts"` - NextAttemptAt *time.Time `json:"next_attempt_at"` - LastError *string `json:"last_error"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID int64 `json:"id"` + AccountID int64 `json:"account_id"` + SwtSid string `json:"swt_sid"` + EventID string `json:"event_id"` + Operation string `json:"operation"` + Payload string `json:"payload"` + OccurredAt time.Time `json:"occurred_at"` + DeliveryStatus string `json:"delivery_status"` + ClaimedAt *time.Time `json:"claimed_at"` + Attempts int64 `json:"attempts"` + NextAttemptAt *time.Time `json:"next_attempt_at"` + LastError *string `json:"last_error"` + ResultSyncStatus string `json:"result_sync_status"` + ResultSyncAttempts int64 `json:"result_sync_attempts"` + ResultSyncNextAt *time.Time `json:"result_sync_next_at"` + ResultReportedAt *time.Time `json:"result_reported_at"` + ResultStatus *string `json:"result_status"` + ResultErrorCode *string `json:"result_error_code"` + ResultErrorMessage *string `json:"result_error_message"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } type OutboundPart struct { diff --git a/channels/shangwutong/db/generated/operations.sql.go b/channels/shangwutong/db/generated/operations.sql.go index b0dfbcc6..9a334735 100644 --- a/channels/shangwutong/db/generated/operations.sql.go +++ b/channels/shangwutong/db/generated/operations.sql.go @@ -45,11 +45,17 @@ WHERE id = ( (julianday(earlier_message.occurred_at) = julianday(candidate.occurred_at) AND earlier_message.event_id < candidate.event_id) ) ) + AND NOT EXISTS ( + SELECT 1 FROM outbound_operations AS earlier_result + WHERE earlier_result.account_id = candidate.account_id + AND earlier_result.id < candidate.id + AND earlier_result.result_sync_status IN ('pending', 'syncing') + ) ORDER BY candidate.id LIMIT 1 ) AND delivery_status = 'pending' -RETURNING id, account_id, swt_sid, event_id, operation, payload, occurred_at, delivery_status, claimed_at, attempts, next_attempt_at, last_error, created_at, updated_at +RETURNING id, account_id, swt_sid, event_id, operation, payload, occurred_at, delivery_status, claimed_at, attempts, next_attempt_at, last_error, result_sync_status, result_sync_attempts, result_sync_next_at, result_reported_at, result_status, result_error_code, result_error_message, created_at, updated_at ` func (q *Queries) ClaimOutboundOperation(ctx context.Context) (*OutboundOperation, error) { @@ -68,6 +74,68 @@ func (q *Queries) ClaimOutboundOperation(ctx context.Context) (*OutboundOperatio &i.Attempts, &i.NextAttemptAt, &i.LastError, + &i.ResultSyncStatus, + &i.ResultSyncAttempts, + &i.ResultSyncNextAt, + &i.ResultReportedAt, + &i.ResultStatus, + &i.ResultErrorCode, + &i.ResultErrorMessage, + &i.CreatedAt, + &i.UpdatedAt, + ) + return &i, err +} + +const claimOutboundOperationResult = `-- name: ClaimOutboundOperationResult :one +UPDATE outbound_operations SET + result_sync_status = 'syncing', + result_sync_attempts = result_sync_attempts + 1, + result_sync_next_at = NULL, + updated_at = CURRENT_TIMESTAMP +WHERE id = ( + SELECT candidate.id + FROM outbound_operations AS candidate + WHERE candidate.operation IN ('set_chat_kind', 'set_customer_color') + AND candidate.delivery_status IN ('delivered', 'uncertain', 'failed') + AND candidate.result_sync_status = 'pending' + AND (candidate.result_sync_next_at IS NULL OR julianday(candidate.result_sync_next_at) IS NULL OR julianday(candidate.result_sync_next_at) <= julianday('now')) + AND NOT EXISTS ( + SELECT 1 FROM outbound_operations AS earlier + WHERE earlier.account_id = candidate.account_id + AND earlier.id < candidate.id + AND earlier.result_sync_status IN ('pending', 'syncing') + ) + ORDER BY candidate.id + LIMIT 1 +) +AND result_sync_status = 'pending' +RETURNING id, account_id, swt_sid, event_id, operation, payload, occurred_at, delivery_status, claimed_at, attempts, next_attempt_at, last_error, result_sync_status, result_sync_attempts, result_sync_next_at, result_reported_at, result_status, result_error_code, result_error_message, created_at, updated_at +` + +func (q *Queries) ClaimOutboundOperationResult(ctx context.Context) (*OutboundOperation, error) { + row := q.db.QueryRowContext(ctx, claimOutboundOperationResult) + var i OutboundOperation + err := row.Scan( + &i.ID, + &i.AccountID, + &i.SwtSid, + &i.EventID, + &i.Operation, + &i.Payload, + &i.OccurredAt, + &i.DeliveryStatus, + &i.ClaimedAt, + &i.Attempts, + &i.NextAttemptAt, + &i.LastError, + &i.ResultSyncStatus, + &i.ResultSyncAttempts, + &i.ResultSyncNextAt, + &i.ResultReportedAt, + &i.ResultStatus, + &i.ResultErrorCode, + &i.ResultErrorMessage, &i.CreatedAt, &i.UpdatedAt, ) @@ -79,6 +147,23 @@ UPDATE outbound_operations SET delivery_status = 'delivered', claimed_at = NULL, last_error = NULL, + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'succeeded' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN NULL + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN NULL + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND delivery_status = 'delivering' ` @@ -88,13 +173,46 @@ func (q *Queries) CompleteOutboundOperation(ctx context.Context, id int64) error return err } +const completeOutboundOperationResult = `-- name: CompleteOutboundOperationResult :exec +UPDATE outbound_operations SET + result_sync_status = 'synced', + result_reported_at = CURRENT_TIMESTAMP, + result_sync_next_at = NULL, + last_error = NULL, + updated_at = CURRENT_TIMESTAMP +WHERE id = ? AND result_sync_status = 'syncing' +` + +func (q *Queries) CompleteOutboundOperationResult(ctx context.Context, id int64) error { + _, err := q.db.ExecContext(ctx, completeOutboundOperationResult, id) + return err +} + const failExpiredUncertainOperations = `-- name: FailExpiredUncertainOperations :execrows UPDATE outbound_operations SET delivery_status = 'failed', claimed_at = NULL, last_error = 'operation result remained uncertain beyond the observation window', + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'failed' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'uncertain_timeout' + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'operation result remained uncertain beyond the observation window' + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE delivery_status = 'uncertain' + AND result_sync_status <> 'syncing' AND julianday(updated_at) <= julianday(?) ` @@ -111,22 +229,65 @@ UPDATE outbound_operations SET delivery_status = 'failed', claimed_at = NULL, last_error = ?, + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'failed' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN ? + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN ? + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND delivery_status IN ('delivering', 'uncertain') ` type FailOutboundOperationParams struct { + LastError *string `json:"last_error"` + ResultErrorCode *string `json:"result_error_code"` + ResultErrorMessage *string `json:"result_error_message"` + ID int64 `json:"id"` +} + +func (q *Queries) FailOutboundOperation(ctx context.Context, arg FailOutboundOperationParams) error { + _, err := q.db.ExecContext(ctx, failOutboundOperation, + arg.LastError, + arg.ResultErrorCode, + arg.ResultErrorMessage, + arg.ID, + ) + return err +} + +const failOutboundOperationResult = `-- name: FailOutboundOperationResult :exec +UPDATE outbound_operations SET + result_sync_status = 'failed', + result_sync_next_at = NULL, + last_error = ?, + updated_at = CURRENT_TIMESTAMP +WHERE id = ? AND result_sync_status = 'syncing' +` + +type FailOutboundOperationResultParams struct { LastError *string `json:"last_error"` ID int64 `json:"id"` } -func (q *Queries) FailOutboundOperation(ctx context.Context, arg FailOutboundOperationParams) error { - _, err := q.db.ExecContext(ctx, failOutboundOperation, arg.LastError, arg.ID) +func (q *Queries) FailOutboundOperationResult(ctx context.Context, arg FailOutboundOperationResultParams) error { + _, err := q.db.ExecContext(ctx, failOutboundOperationResult, arg.LastError, arg.ID) return err } const getOutboundOperationByEventID = `-- name: GetOutboundOperationByEventID :one -SELECT id, account_id, swt_sid, event_id, operation, payload, occurred_at, delivery_status, claimed_at, attempts, next_attempt_at, last_error, created_at, updated_at FROM outbound_operations +SELECT id, account_id, swt_sid, event_id, operation, payload, occurred_at, delivery_status, claimed_at, attempts, next_attempt_at, last_error, result_sync_status, result_sync_attempts, result_sync_next_at, result_reported_at, result_status, result_error_code, result_error_message, created_at, updated_at FROM outbound_operations WHERE event_id = ? LIMIT 1 ` @@ -147,6 +308,13 @@ func (q *Queries) GetOutboundOperationByEventID(ctx context.Context, eventID str &i.Attempts, &i.NextAttemptAt, &i.LastError, + &i.ResultSyncStatus, + &i.ResultSyncAttempts, + &i.ResultSyncNextAt, + &i.ResultReportedAt, + &i.ResultStatus, + &i.ResultErrorCode, + &i.ResultErrorMessage, &i.CreatedAt, &i.UpdatedAt, ) @@ -158,7 +326,7 @@ INSERT INTO outbound_operations ( account_id, swt_sid, event_id, operation, payload, occurred_at ) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(event_id) DO NOTHING -RETURNING id, account_id, swt_sid, event_id, operation, payload, occurred_at, delivery_status, claimed_at, attempts, next_attempt_at, last_error, created_at, updated_at +RETURNING id, account_id, swt_sid, event_id, operation, payload, occurred_at, delivery_status, claimed_at, attempts, next_attempt_at, last_error, result_sync_status, result_sync_attempts, result_sync_next_at, result_reported_at, result_status, result_error_code, result_error_message, created_at, updated_at ` type InsertOutboundOperationParams struct { @@ -193,6 +361,13 @@ func (q *Queries) InsertOutboundOperation(ctx context.Context, arg InsertOutboun &i.Attempts, &i.NextAttemptAt, &i.LastError, + &i.ResultSyncStatus, + &i.ResultSyncAttempts, + &i.ResultSyncNextAt, + &i.ResultReportedAt, + &i.ResultStatus, + &i.ResultErrorCode, + &i.ResultErrorMessage, &i.CreatedAt, &i.UpdatedAt, ) @@ -204,25 +379,82 @@ UPDATE outbound_operations SET delivery_status = 'uncertain', claimed_at = NULL, last_error = ?, + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'uncertain' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN ? + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN ? + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND delivery_status = 'delivering' ` type MarkOutboundOperationUncertainParams struct { - LastError *string `json:"last_error"` - ID int64 `json:"id"` + LastError *string `json:"last_error"` + ResultErrorCode *string `json:"result_error_code"` + ResultErrorMessage *string `json:"result_error_message"` + ID int64 `json:"id"` } func (q *Queries) MarkOutboundOperationUncertain(ctx context.Context, arg MarkOutboundOperationUncertainParams) error { - _, err := q.db.ExecContext(ctx, markOutboundOperationUncertain, arg.LastError, arg.ID) + _, err := q.db.ExecContext(ctx, markOutboundOperationUncertain, + arg.LastError, + arg.ResultErrorCode, + arg.ResultErrorMessage, + arg.ID, + ) return err } +const recoverOutboundOperationResults = `-- name: RecoverOutboundOperationResults :execrows +UPDATE outbound_operations SET + result_sync_status = 'pending', + result_sync_next_at = NULL, + updated_at = CURRENT_TIMESTAMP +WHERE result_sync_status = 'syncing' +` + +func (q *Queries) RecoverOutboundOperationResults(ctx context.Context) (int64, error) { + result, err := q.db.ExecContext(ctx, recoverOutboundOperationResults) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + const recoverOutboundOperationsAsUncertain = `-- name: RecoverOutboundOperationsAsUncertain :execrows UPDATE outbound_operations SET delivery_status = 'uncertain', claimed_at = NULL, last_error = 'connector restarted while operation was delivering', + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'uncertain' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'connector_restart_uncertain' + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'connector restarted while operation was delivering' + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE delivery_status = 'delivering' ` @@ -255,3 +487,23 @@ func (q *Queries) RetryOutboundOperation(ctx context.Context, arg RetryOutboundO _, err := q.db.ExecContext(ctx, retryOutboundOperation, arg.NextAttemptAt, arg.LastError, arg.ID) return err } + +const retryOutboundOperationResult = `-- name: RetryOutboundOperationResult :exec +UPDATE outbound_operations SET + result_sync_status = 'pending', + result_sync_next_at = ?, + last_error = ?, + updated_at = CURRENT_TIMESTAMP +WHERE id = ? AND result_sync_status = 'syncing' +` + +type RetryOutboundOperationResultParams struct { + ResultSyncNextAt *time.Time `json:"result_sync_next_at"` + LastError *string `json:"last_error"` + ID int64 `json:"id"` +} + +func (q *Queries) RetryOutboundOperationResult(ctx context.Context, arg RetryOutboundOperationResultParams) error { + _, err := q.db.ExecContext(ctx, retryOutboundOperationResult, arg.ResultSyncNextAt, arg.LastError, arg.ID) + return err +} diff --git a/channels/shangwutong/db/migrations/008_add_classification_operation_results.down.sql b/channels/shangwutong/db/migrations/008_add_classification_operation_results.down.sql new file mode 100644 index 00000000..f8d848a2 --- /dev/null +++ b/channels/shangwutong/db/migrations/008_add_classification_operation_results.down.sql @@ -0,0 +1,38 @@ +CREATE TABLE outbound_operations_old ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account_id INTEGER NOT NULL REFERENCES accounts(id), + swt_sid TEXT NOT NULL, + event_id TEXT NOT NULL UNIQUE, + operation TEXT NOT NULL, + payload TEXT NOT NULL, + occurred_at DATETIME NOT NULL, + delivery_status TEXT NOT NULL DEFAULT 'pending', + claimed_at DATETIME, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at DATETIME, + last_error TEXT, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK(operation IN ('end_conversation', 'change_contact_name', 'accept_transfer', 'transfer_conversation')), + CHECK(delivery_status IN ('pending', 'delivering', 'delivered', 'uncertain', 'failed')) +); + +INSERT INTO outbound_operations_old ( + id, account_id, swt_sid, event_id, operation, payload, occurred_at, + delivery_status, claimed_at, attempts, next_attempt_at, last_error, + created_at, updated_at +) +SELECT + id, account_id, swt_sid, event_id, operation, payload, occurred_at, + delivery_status, claimed_at, attempts, next_attempt_at, last_error, + created_at, updated_at +FROM outbound_operations; + +DROP TABLE outbound_operations; +ALTER TABLE outbound_operations_old RENAME TO outbound_operations; + +CREATE INDEX idx_outbound_operations_ready + ON outbound_operations(delivery_status, next_attempt_at, account_id, id); +CREATE UNIQUE INDEX idx_outbound_one_delivering_operation_per_account + ON outbound_operations(account_id) + WHERE delivery_status = 'delivering'; diff --git a/channels/shangwutong/db/migrations/008_add_classification_operation_results.up.sql b/channels/shangwutong/db/migrations/008_add_classification_operation_results.up.sql new file mode 100644 index 00000000..f1778c64 --- /dev/null +++ b/channels/shangwutong/db/migrations/008_add_classification_operation_results.up.sql @@ -0,0 +1,48 @@ +CREATE TABLE outbound_operations_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account_id INTEGER NOT NULL REFERENCES accounts(id), + swt_sid TEXT NOT NULL, + event_id TEXT NOT NULL UNIQUE, + operation TEXT NOT NULL, + payload TEXT NOT NULL, + occurred_at DATETIME NOT NULL, + delivery_status TEXT NOT NULL DEFAULT 'pending', + claimed_at DATETIME, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at DATETIME, + last_error TEXT, + result_sync_status TEXT NOT NULL DEFAULT 'not_required', + result_sync_attempts INTEGER NOT NULL DEFAULT 0, + result_sync_next_at DATETIME, + result_reported_at DATETIME, + result_status TEXT, + result_error_code TEXT, + result_error_message TEXT, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK(operation IN ('end_conversation', 'change_contact_name', 'accept_transfer', 'transfer_conversation', 'set_chat_kind', 'set_customer_color')), + CHECK(delivery_status IN ('pending', 'delivering', 'delivered', 'uncertain', 'failed')), + CHECK(result_sync_status IN ('not_required', 'pending', 'syncing', 'synced', 'failed')) +); + +INSERT INTO outbound_operations_new ( + id, account_id, swt_sid, event_id, operation, payload, occurred_at, + delivery_status, claimed_at, attempts, next_attempt_at, last_error, + created_at, updated_at +) +SELECT + id, account_id, swt_sid, event_id, operation, payload, occurred_at, + delivery_status, claimed_at, attempts, next_attempt_at, last_error, + created_at, updated_at +FROM outbound_operations; + +DROP TABLE outbound_operations; +ALTER TABLE outbound_operations_new RENAME TO outbound_operations; + +CREATE INDEX idx_outbound_operations_ready + ON outbound_operations(delivery_status, next_attempt_at, account_id, id); +CREATE UNIQUE INDEX idx_outbound_one_delivering_operation_per_account + ON outbound_operations(account_id) + WHERE delivery_status = 'delivering'; +CREATE INDEX idx_outbound_operation_result_sync_ready + ON outbound_operations(result_sync_status, result_sync_next_at, account_id, id); diff --git a/channels/shangwutong/db/queries/operations.sql b/channels/shangwutong/db/queries/operations.sql index f66a639e..d408131c 100644 --- a/channels/shangwutong/db/queries/operations.sql +++ b/channels/shangwutong/db/queries/operations.sql @@ -45,6 +45,12 @@ WHERE id = ( (julianday(earlier_message.occurred_at) = julianday(candidate.occurred_at) AND earlier_message.event_id < candidate.event_id) ) ) + AND NOT EXISTS ( + SELECT 1 FROM outbound_operations AS earlier_result + WHERE earlier_result.account_id = candidate.account_id + AND earlier_result.id < candidate.id + AND earlier_result.result_sync_status IN ('pending', 'syncing') + ) ORDER BY candidate.id LIMIT 1 ) @@ -56,6 +62,23 @@ UPDATE outbound_operations SET delivery_status = 'delivered', claimed_at = NULL, last_error = NULL, + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'succeeded' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN NULL + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN NULL + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND delivery_status = 'delivering'; @@ -68,11 +91,85 @@ UPDATE outbound_operations SET updated_at = CURRENT_TIMESTAMP WHERE id = ? AND delivery_status = 'delivering'; +-- name: ClaimOutboundOperationResult :one +UPDATE outbound_operations SET + result_sync_status = 'syncing', + result_sync_attempts = result_sync_attempts + 1, + result_sync_next_at = NULL, + updated_at = CURRENT_TIMESTAMP +WHERE id = ( + SELECT candidate.id + FROM outbound_operations AS candidate + WHERE candidate.operation IN ('set_chat_kind', 'set_customer_color') + AND candidate.delivery_status IN ('delivered', 'uncertain', 'failed') + AND candidate.result_sync_status = 'pending' + AND (candidate.result_sync_next_at IS NULL OR julianday(candidate.result_sync_next_at) IS NULL OR julianday(candidate.result_sync_next_at) <= julianday('now')) + AND NOT EXISTS ( + SELECT 1 FROM outbound_operations AS earlier + WHERE earlier.account_id = candidate.account_id + AND earlier.id < candidate.id + AND earlier.result_sync_status IN ('pending', 'syncing') + ) + ORDER BY candidate.id + LIMIT 1 +) +AND result_sync_status = 'pending' +RETURNING *; + +-- name: CompleteOutboundOperationResult :exec +UPDATE outbound_operations SET + result_sync_status = 'synced', + result_reported_at = CURRENT_TIMESTAMP, + result_sync_next_at = NULL, + last_error = NULL, + updated_at = CURRENT_TIMESTAMP +WHERE id = ? AND result_sync_status = 'syncing'; + +-- name: RetryOutboundOperationResult :exec +UPDATE outbound_operations SET + result_sync_status = 'pending', + result_sync_next_at = ?, + last_error = ?, + updated_at = CURRENT_TIMESTAMP +WHERE id = ? AND result_sync_status = 'syncing'; + +-- name: FailOutboundOperationResult :exec +UPDATE outbound_operations SET + result_sync_status = 'failed', + result_sync_next_at = NULL, + last_error = ?, + updated_at = CURRENT_TIMESTAMP +WHERE id = ? AND result_sync_status = 'syncing'; + +-- name: RecoverOutboundOperationResults :execrows +UPDATE outbound_operations SET + result_sync_status = 'pending', + result_sync_next_at = NULL, + updated_at = CURRENT_TIMESTAMP +WHERE result_sync_status = 'syncing'; + -- name: MarkOutboundOperationUncertain :exec UPDATE outbound_operations SET delivery_status = 'uncertain', claimed_at = NULL, last_error = ?, + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'uncertain' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN ? + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN ? + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND delivery_status = 'delivering'; @@ -81,6 +178,23 @@ UPDATE outbound_operations SET delivery_status = 'failed', claimed_at = NULL, last_error = ?, + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'failed' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN ? + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN ? + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND delivery_status IN ('delivering', 'uncertain'); @@ -89,6 +203,23 @@ UPDATE outbound_operations SET delivery_status = 'uncertain', claimed_at = NULL, last_error = 'connector restarted while operation was delivering', + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'uncertain' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'connector_restart_uncertain' + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'connector restarted while operation was delivering' + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE delivery_status = 'delivering'; @@ -97,6 +228,24 @@ UPDATE outbound_operations SET delivery_status = 'failed', claimed_at = NULL, last_error = 'operation result remained uncertain beyond the observation window', + result_sync_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'pending' + ELSE result_sync_status + END, + result_sync_next_at = NULL, + result_status = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'failed' + ELSE result_status + END, + result_error_code = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'uncertain_timeout' + ELSE result_error_code + END, + result_error_message = CASE + WHEN operation IN ('set_chat_kind', 'set_customer_color') THEN 'operation result remained uncertain beyond the observation window' + ELSE result_error_message + END, updated_at = CURRENT_TIMESTAMP WHERE delivery_status = 'uncertain' + AND result_sync_status <> 'syncing' AND julianday(updated_at) <= julianday(?); diff --git a/channels/shangwutong/internal/account/manager.go b/channels/shangwutong/internal/account/manager.go index 339d5724..b735d29e 100644 --- a/channels/shangwutong/internal/account/manager.go +++ b/channels/shangwutong/internal/account/manager.go @@ -167,6 +167,14 @@ func (m *Manager) SyncClassifications(ctx context.Context, accountID int64, even if !ok { return errors.New("classification reporter is unavailable") } + account, err := m.store.Reader().GetAccountByID(ctx, accountID) + if err != nil { + return fmt.Errorf("load connector account: %w", err) + } + if account.GochatInboxID <= 0 { + return errors.New("GoChat inbox mapping is invalid") + } + inboxID := account.GochatInboxID var catalog swt.ClassificationCatalog if err := m.WithSession(ctx, accountID, func(session swt.Session) error { var err error @@ -179,11 +187,11 @@ func (m *Manager) SyncClassifications(ctx context.Context, accountID int64, even if errors.As(err, &protocolErr) && protocolErr.Code != "" { code = protocolErr.Code } - _ = statusReporter.UpdateClassificationSyncStatus(ctx, accountID, eventID, "failed", code, err.Error()) + _ = statusReporter.UpdateClassificationSyncStatus(ctx, inboxID, eventID, "failed", code, err.Error()) } return err } - return reporter.UpdateClassificationCatalog(ctx, accountID, eventID, gochat.ClassificationCatalog{ + return reporter.UpdateClassificationCatalog(ctx, inboxID, eventID, gochat.ClassificationCatalog{ ConversationKinds: convertConversationKinds(catalog.ConversationKinds), CustomerColors: convertCustomerColors(catalog.CustomerColors), }) diff --git a/channels/shangwutong/internal/account/manager_test.go b/channels/shangwutong/internal/account/manager_test.go index 2d338ad7..4a50687e 100644 --- a/channels/shangwutong/internal/account/manager_test.go +++ b/channels/shangwutong/internal/account/manager_test.go @@ -9,6 +9,7 @@ import ( "time" dbgen "github.com/gochat/gochat/channels/shangwutong/db/generated" + "github.com/gochat/gochat/channels/shangwutong/internal/gochat" "github.com/gochat/gochat/channels/shangwutong/internal/store" "github.com/gochat/gochat/channels/shangwutong/internal/swt" ) @@ -119,6 +120,38 @@ func TestSupervisorRejectsCandidatePasswordWithoutReplacingWorkingSession(t *tes manager.Wait() } +func TestManagerClassificationCallbacksUseGoChatInboxID(t *testing.T) { + ctx := context.Background() + database, err := store.Open(ctx, filepath.Join(t.TempDir(), "connector.db")) + if err != nil { + t.Fatal(err) + } + defer database.Close() + account := createRunnableAccount(t, ctx, database) + protocol := newFakeProtocol() + reporter := &classificationReporter{} + manager, err := NewManager(database, protocol, reporter, nil, 1) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + manager.Stop() + manager.Wait() + }) + manager.initialDelay = func(time.Duration) time.Duration { return 0 } + manager.heartbeatInterval = time.Hour + if err := manager.Start(ctx); err != nil { + t.Fatal(err) + } + waitSignal(t, protocol.logins, "classification login") + if err := manager.SyncClassifications(ctx, account.ID, "classification:1"); err != nil { + t.Fatal(err) + } + if reporter.catalogInboxID != account.GochatInboxID { + t.Fatalf("catalog inbox ID = %d, want GoChat inbox ID %d", reporter.catalogInboxID, account.GochatInboxID) + } +} + func TestManagerMergesTypingIntoNextHeartbeat(t *testing.T) { ctx := context.Background() database, err := store.Open(ctx, filepath.Join(t.TempDir(), "connector.db")) @@ -258,6 +291,10 @@ func (f *fakeProtocol) Heartbeat(_ context.Context, session swt.Session, cursor return swt.HeartbeatResult{Status: swt.HeartbeatOK}, nil } +func (f *fakeProtocol) FetchClassificationCatalog(context.Context, swt.Session) (swt.ClassificationCatalog, error) { + return swt.ClassificationCatalog{ConversationKinds: []swt.ConversationKind{{ID: "kind-1", Name: "Normal"}}}, nil +} + func (f *fakeProtocol) SetPresence(_ context.Context, _ swt.Session, presence swt.Presence) error { f.mu.Lock() if presence == swt.PresenceOffline { @@ -274,6 +311,23 @@ func (f *fakeProtocol) offlineCalls() int { return f.offline } +type classificationReporter struct { + catalogInboxID int64 +} + +func (r *classificationReporter) UpdateInboxStatus(context.Context, int64, gochat.InboxStatus) error { + return nil +} + +func (r *classificationReporter) UpdateClassificationCatalog(_ context.Context, inboxID int64, _ string, _ gochat.ClassificationCatalog) error { + r.catalogInboxID = inboxID + return nil +} + +func (r *classificationReporter) UpdateClassificationSyncStatus(context.Context, int64, string, string, string, string) error { + return nil +} + func waitSignal[T any](t *testing.T, signal <-chan T, name string) T { t.Helper() select { diff --git a/channels/shangwutong/internal/command/root.go b/channels/shangwutong/internal/command/root.go index 3b99558b..3e6d7173 100644 --- a/channels/shangwutong/internal/command/root.go +++ b/channels/shangwutong/internal/command/root.go @@ -72,6 +72,9 @@ func serve(ctx context.Context) error { if _, err := database.Writer().RecoverOutboundOperationsAsUncertain(ctx); err != nil { return fmt.Errorf("recover outbound operation queue: %w", err) } + if _, err := database.Writer().RecoverOutboundOperationResults(ctx); err != nil { + return fmt.Errorf("recover outbound operation result queue: %w", err) + } if _, err := database.Writer().RecoverOutboundStatusSyncs(ctx); err != nil { return fmt.Errorf("recover outbound status sync queue: %w", err) } diff --git a/channels/shangwutong/internal/command/root_test.go b/channels/shangwutong/internal/command/root_test.go index 18d6b6ec..91ae1fa9 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 != 7 { + if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 8 { t.Fatalf("backup version = %d, %v", version, err) } diff --git a/channels/shangwutong/internal/delivery/outbound.go b/channels/shangwutong/internal/delivery/outbound.go index 62cc7f81..c2687a7c 100644 --- a/channels/shangwutong/internal/delivery/outbound.go +++ b/channels/shangwutong/internal/delivery/outbound.go @@ -102,6 +102,11 @@ func (o *Outbound) Start(ctx context.Context) { o.wg.Add(1) go o.operationLoop(ctx) } + resultWorkers := min(o.workers, 4) + for range resultWorkers { + o.wg.Add(1) + go o.operationResultLoop(ctx) + } o.wg.Add(1) go o.uncertainLoop(ctx) } @@ -154,6 +159,21 @@ func (o *Outbound) operationLoop(ctx context.Context) { } } +func (o *Outbound) operationResultLoop(ctx context.Context) { + defer o.wg.Done() + for ctx.Err() == nil { + worked, err := o.processOperationResult(ctx) + if err != nil && ctx.Err() == nil { + o.logger.WithFields(logrus.Fields{ + "component": "outbound_operation_result_worker", "operation": "sync", "result": "failed", + }).WithError(err).Warn("outbound operation result sync failed") + } + if !worked && !wait(ctx, 200*time.Millisecond) { + return + } + } +} + func (o *Outbound) Wait() { o.wg.Wait() } func (o *Outbound) SetMetrics(metrics *observability.Metrics) { o.metrics = metrics } @@ -605,10 +625,8 @@ func (o *Outbound) processOperation(ctx context.Context) (bool, error) { if !classificationSupported { return &swt.Error{Operation: operation.Operation, Code: "unsupported_operation", Err: errors.New("classification sender is unavailable")} } - var payload struct { - ChatKindID string `json:"chat_kind_id"` - } - if err := json.Unmarshal([]byte(operation.Payload), &payload); err != nil || strings.TrimSpace(payload.ChatKindID) == "" { + payload, err := decodeClassificationOperation(operation.Payload) + if err != nil || strings.TrimSpace(payload.ChatKindID) == "" { return &swt.Error{Operation: operation.Operation, Code: "invalid_payload", Err: errors.New("chat_kind_id is required")} } return classificationSender.SetConversationKind(ctx, session, operation.SwtSid, payload.ChatKindID) @@ -616,12 +634,8 @@ func (o *Outbound) processOperation(ctx context.Context) (bool, error) { if !classificationSupported { return &swt.Error{Operation: operation.Operation, Code: "unsupported_operation", Err: errors.New("classification sender is unavailable")} } - var payload struct { - CustomerColorID string `json:"customer_color_id"` - CustomerColorName string `json:"customer_color_name"` - CID string `json:"cid"` - } - if err := json.Unmarshal([]byte(operation.Payload), &payload); err != nil || strings.TrimSpace(payload.CustomerColorID) == "" || strings.TrimSpace(payload.CID) == "" { + payload, err := decodeClassificationOperation(operation.Payload) + if err != nil || strings.TrimSpace(payload.CustomerColorID) == "" || strings.TrimSpace(payload.CID) == "" { return &swt.Error{Operation: operation.Operation, Code: "invalid_payload", Err: errors.New("customer_color_id and cid are required")} } return classificationSender.ChangeCustomerColor(ctx, session, operation.SwtSid, payload.CustomerColorID, payload.CustomerColorName, payload.CID) @@ -658,17 +672,15 @@ func (o *Outbound) processOperation(ctx context.Context) (bool, error) { } }) if err == nil { - if reportErr := o.reportClassificationStatus(ctx, operation, "succeeded", nil); reportErr != nil { - return true, o.retryOperation(operation, "classification result callback failed: "+reportErr.Error()) - } return true, o.store.Writer().CompleteOutboundOperation(ctx, operation.ID) } var protocolErr *swt.Error if errors.As(err, &protocolErr) { if protocolErr.Uncertain { - detail := protocolErr.Error() - _ = o.reportClassificationStatus(ctx, operation, "uncertain", protocolErr) - return true, o.store.Writer().MarkOutboundOperationUncertain(ctx, dbgen.MarkOutboundOperationUncertainParams{LastError: &detail, ID: operation.ID}) + detail, code := protocolErr.Error(), protocolErr.Code + return true, o.store.Writer().MarkOutboundOperationUncertain(ctx, dbgen.MarkOutboundOperationUncertainParams{ + LastError: &detail, ResultErrorCode: &code, ResultErrorMessage: &detail, ID: operation.ID, + }) } if protocolErr.Code == "tickint_reset" || protocolErr.Code == "cache_null" { _ = o.sessions.InvalidateSession(ctx, operation.AccountID) @@ -677,38 +689,79 @@ func (o *Outbound) processOperation(ctx context.Context) (bool, error) { if protocolErr.Retryable && operation.Attempts < 10 { return true, o.retryOperation(operation, protocolErr.Error()) } - return true, o.failOperation(operation, protocolErr.Error()) + return true, o.failOperation(operation, protocolErr.Code, protocolErr.Error()) } if operation.Attempts < 10 { return true, o.retryOperation(operation, err.Error()) } - return true, o.failOperation(operation, err.Error()) + return true, o.failOperation(operation, "delivery_exhausted", err.Error()) } -func (o *Outbound) reportClassificationStatus(ctx context.Context, operation *dbgen.OutboundOperation, status string, cause error) error { +type classificationOperationData struct { + ConversationID uint `json:"conversation_id"` + ChatKindID string `json:"chat_kind_id"` + CustomerColorID string `json:"customer_color_id"` + CustomerColorName string `json:"customer_color_name"` + CID string `json:"cid"` +} + +type classificationOperationEnvelope struct { + Data classificationOperationData `json:"data"` +} + +func decodeClassificationOperation(payload string) (classificationOperationData, error) { + var envelope classificationOperationEnvelope + if err := json.Unmarshal([]byte(payload), &envelope); err != nil { + return classificationOperationData{}, err + } + return envelope.Data, nil +} + +func (o *Outbound) processOperationResult(ctx context.Context) (bool, error) { + operation, err := o.store.Writer().ClaimOutboundOperationResult(ctx) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, err + } + if err := o.reportClassificationStatus(ctx, operation); err != nil { + detail := "classification result callback failed: " + err.Error() + if operation.ResultSyncAttempts < 10 { + return true, o.retryOperationResult(operation, detail) + } + return true, o.failOperationResult(operation, detail) + } + return true, o.store.Writer().CompleteOutboundOperationResult(ctx, operation.ID) +} + +func (o *Outbound) reportClassificationStatus(ctx context.Context, operation *dbgen.OutboundOperation) error { reporter, ok := o.results.(ClassificationResultClient) - if !ok || (operation.Operation != "set_chat_kind" && operation.Operation != "set_customer_color") { - return nil + if !ok { + return errors.New("classification result reporter is unavailable") } - var envelope struct { - Data struct { - ConversationID uint `json:"conversation_id"` - ChatKindID string `json:"chat_kind_id"` - CustomerColorID string `json:"customer_color_id"` - } `json:"data"` + status := value(operation.ResultStatus) + if status != "succeeded" && status != "failed" && status != "uncertain" { + return errors.New("classification result status is invalid") } - if err := json.Unmarshal([]byte(operation.Payload), &envelope); err != nil || envelope.Data.ConversationID == 0 { + data, err := decodeClassificationOperation(operation.Payload) + if err != nil || data.ConversationID == 0 { return errors.New("classification result payload is invalid") } - errorCode, errorMessage := "", "" - if cause != nil { - errorMessage = cause.Error() - var protocolErr *swt.Error - if errors.As(cause, &protocolErr) { - errorCode = protocolErr.Code - } + if operation.Operation == "set_chat_kind" && strings.TrimSpace(data.ChatKindID) == "" { + return errors.New("chat_kind_id is required") } - return reporter.UpdateClassificationStatus(ctx, operation.AccountID, envelope.Data.ConversationID, operation.EventID, operation.Operation, status, envelope.Data.ChatKindID, envelope.Data.CustomerColorID, errorCode, errorMessage) + if operation.Operation == "set_customer_color" && (strings.TrimSpace(data.CustomerColorID) == "" || strings.TrimSpace(data.CID) == "") { + return errors.New("customer_color_id and cid are required") + } + account, err := o.store.Reader().GetAccountByID(ctx, operation.AccountID) + if err != nil { + return fmt.Errorf("load GoChat inbox mapping: %w", err) + } + if account.GochatInboxID <= 0 { + return errors.New("GoChat inbox mapping is invalid") + } + return reporter.UpdateClassificationStatus(ctx, account.GochatInboxID, data.ConversationID, operation.EventID, operation.Operation, status, data.ChatKindID, data.CustomerColorID, value(operation.ResultErrorCode), value(operation.ResultErrorMessage)) } func (o *Outbound) retry(message *dbgen.OutboundMessage, detail string) error { @@ -769,8 +822,21 @@ func (o *Outbound) retryOperation(operation *dbgen.OutboundOperation, detail str }) } -func (o *Outbound) failOperation(operation *dbgen.OutboundOperation, detail string) error { - return o.store.Writer().FailOutboundOperation(context.Background(), dbgen.FailOutboundOperationParams{LastError: &detail, ID: operation.ID}) +func (o *Outbound) failOperation(operation *dbgen.OutboundOperation, code, detail string) error { + return o.store.Writer().FailOutboundOperation(context.Background(), dbgen.FailOutboundOperationParams{ + LastError: &detail, ResultErrorCode: &code, ResultErrorMessage: &detail, ID: operation.ID, + }) +} + +func (o *Outbound) retryOperationResult(operation *dbgen.OutboundOperation, detail string) error { + next := time.Now().Add(backoff(operation.ResultSyncAttempts, 5*time.Minute)) + return o.store.Writer().RetryOutboundOperationResult(context.Background(), dbgen.RetryOutboundOperationResultParams{ + ResultSyncNextAt: &next, LastError: &detail, ID: operation.ID, + }) +} + +func (o *Outbound) failOperationResult(operation *dbgen.OutboundOperation, detail string) error { + return o.store.Writer().FailOutboundOperationResult(context.Background(), dbgen.FailOutboundOperationResultParams{LastError: &detail, ID: operation.ID}) } func backoff(attempts int64, maximum time.Duration) time.Duration { diff --git a/channels/shangwutong/internal/delivery/outbound_test.go b/channels/shangwutong/internal/delivery/outbound_test.go index 3ea014ca..9724d7bb 100644 --- a/channels/shangwutong/internal/delivery/outbound_test.go +++ b/channels/shangwutong/internal/delivery/outbound_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "os" "path/filepath" "strings" @@ -622,6 +623,137 @@ func TestOutboundConversationEndOperationUsesSameDurableWorker(t *testing.T) { } } +func TestClassificationOperationsUseNestedPayloadAndReportWithInboxID(t *testing.T) { + ctx := context.Background() + database, account := deliveryDatabase(t, ctx) + tests := []struct { + name string + operation string + payload string + }{ + { + name: "chat kind", + operation: "set_chat_kind", + payload: `{"schema_version":1,"event":"conversation_classification_changed","event_id":"classification:kind","occurred_at":"2026-01-01T00:00:00Z","account_id":1,"inbox_id":10,"data":{"conversation_id":88,"sid":"visitor","chat_kind_id":"kind-1"}}`, + }, + { + name: "customer color", + operation: "set_customer_color", + payload: `{"schema_version":1,"event":"conversation_classification_changed","event_id":"classification:color","occurred_at":"2026-01-01T00:00:01Z","account_id":1,"inbox_id":10,"data":{"conversation_id":89,"sid":"visitor","cid":"cid-1","customer_color_id":"color-1","customer_color_name":"VIP"}}`, + }, + } + for index, test := range tests { + t.Run(test.name, func(t *testing.T) { + operation, _, err := database.EnqueueOutboundOperation(ctx, store.OutboundOperationInput{ + AccountID: account.ID, SWTSessionID: "visitor", EventID: fmt.Sprintf("classification:%d", index), + Operation: test.operation, Payload: test.payload, OccurredAt: time.Now().Add(time.Duration(index) * time.Second), + }) + if err != nil { + t.Fatal(err) + } + sender := &operationSender{} + results := &classificationResultRecorder{} + worker, err := NewOutbound(database, sessionStub{}, sender, results, nil, 1) + if err != nil { + t.Fatal(err) + } + if worked, err := worker.processOperation(ctx); err != nil || !worked { + t.Fatalf("operation = %v, %v", worked, err) + } + loaded, err := database.Reader().GetOutboundOperationByEventID(ctx, operation.EventID) + if err != nil || loaded.DeliveryStatus != "delivered" || loaded.ResultSyncStatus != "pending" || value(loaded.ResultStatus) != "succeeded" { + t.Fatalf("loaded before callback = %#v, %v", loaded, err) + } + if worked, err := worker.processOperationResult(ctx); err != nil || !worked { + t.Fatalf("result = %v, %v", worked, err) + } + if len(results.calls) != 1 || results.calls[0].inboxID != account.GochatInboxID || results.calls[0].conversationID != 88+uint(index) || results.calls[0].operation != test.operation || results.calls[0].status != "succeeded" { + t.Fatalf("classification result = %#v", results.calls) + } + if test.operation == "set_chat_kind" && (sender.kindCalls != 1 || sender.kindID != "kind-1") { + t.Fatalf("kind sender = %#v", sender) + } + if test.operation == "set_customer_color" && (sender.colorCalls != 1 || sender.colorID != "color-1" || sender.colorCID != "cid-1") { + t.Fatalf("color sender = %#v", sender) + } + loaded, err = database.Reader().GetOutboundOperationByEventID(ctx, operation.EventID) + if err != nil || loaded.ResultSyncStatus != "synced" { + t.Fatalf("loaded after callback = %#v, %v", loaded, err) + } + }) + } +} + +func TestClassificationResultRetryDoesNotRepeatSWTOperation(t *testing.T) { + ctx := context.Background() + database, account := deliveryDatabase(t, ctx) + operation, _, err := database.EnqueueOutboundOperation(ctx, store.OutboundOperationInput{ + AccountID: account.ID, SWTSessionID: "visitor", EventID: "classification:retry", + Operation: "set_chat_kind", Payload: `{"data":{"conversation_id":88,"sid":"visitor","chat_kind_id":"kind-1"}}`, OccurredAt: time.Now(), + }) + if err != nil { + t.Fatal(err) + } + sender := &operationSender{} + results := &classificationResultRecorder{err: errors.New("GoChat unavailable")} + worker, _ := NewOutbound(database, sessionStub{}, sender, results, nil, 1) + if worked, err := worker.processOperation(ctx); err != nil || !worked { + t.Fatalf("operation = %v, %v", worked, err) + } + claimed, err := database.Writer().ClaimOutboundOperationResult(ctx) + if err != nil { + t.Fatal(err) + } + if err := worker.reportClassificationStatus(ctx, claimed); err == nil { + t.Fatal("expected result callback failure") + } + past := time.Now().Add(-time.Second) + if err := database.Writer().RetryOutboundOperationResult(ctx, dbgen.RetryOutboundOperationResultParams{ + ResultSyncNextAt: &past, LastError: stringPointer("GoChat unavailable"), ID: claimed.ID, + }); err != nil { + t.Fatal(err) + } + results.err = nil + if worked, err := worker.processOperationResult(ctx); err != nil || !worked { + t.Fatalf("retried result = %v, %v", worked, err) + } + if sender.kindCalls != 1 || len(results.calls) != 1 { + t.Fatalf("external calls repeated: sender=%#v results=%#v", sender, results.calls) + } + loaded, err := database.Reader().GetOutboundOperationByEventID(ctx, operation.EventID) + if err != nil || loaded.DeliveryStatus != "delivered" || loaded.ResultSyncStatus != "synced" { + t.Fatalf("loaded = %#v, %v", loaded, err) + } +} + +func TestClassificationFailureIsReportedAfterExternalFailure(t *testing.T) { + ctx := context.Background() + database, account := deliveryDatabase(t, ctx) + operation, _, err := database.EnqueueOutboundOperation(ctx, store.OutboundOperationInput{ + AccountID: account.ID, SWTSessionID: "visitor", EventID: "classification:failed", + Operation: "set_customer_color", Payload: `{"data":{"conversation_id":88,"sid":"visitor","cid":"cid-1","customer_color_id":"color-1"}}`, OccurredAt: time.Now(), + }) + if err != nil { + t.Fatal(err) + } + sender := &operationSender{err: &swt.Error{Operation: "set_customer_color", Code: "rejected", Retryable: false, Err: errors.New("rejected")}} + results := &classificationResultRecorder{} + worker, _ := NewOutbound(database, sessionStub{}, sender, results, nil, 1) + if worked, err := worker.processOperation(ctx); err != nil || !worked { + t.Fatalf("operation = %v, %v", worked, err) + } + loaded, err := database.Reader().GetOutboundOperationByEventID(ctx, operation.EventID) + if err != nil || loaded.DeliveryStatus != "failed" || loaded.ResultSyncStatus != "pending" || value(loaded.ResultStatus) != "failed" || value(loaded.ResultErrorCode) != "rejected" { + t.Fatalf("failed operation = %#v, %v", loaded, err) + } + if worked, err := worker.processOperationResult(ctx); err != nil || !worked { + t.Fatalf("failure result = %v, %v", worked, err) + } + if len(results.calls) != 1 || results.calls[0].inboxID != account.GochatInboxID || results.calls[0].status != "failed" || results.calls[0].errorCode != "rejected" { + t.Fatalf("failure callback = %#v", results.calls) + } +} + func TestOutboundAcceptTransferPersistsSuccessAndRetry(t *testing.T) { ctx := context.Background() database, account := deliveryDatabase(t, ctx) @@ -944,6 +1076,11 @@ type operationSender struct { transferSID string otherLoginName string sentSID string + kindCalls int + kindID string + colorCalls int + colorID string + colorCID string err error } @@ -990,6 +1127,16 @@ func (s *operationSender) TransferConversation(_ context.Context, _ swt.Session, return s.err } +func (s *operationSender) SetConversationKind(_ context.Context, _ swt.Session, sid, kindID string) error { + s.kindCalls, s.sentSID, s.kindID = s.kindCalls+1, sid, kindID + return s.err +} + +func (s *operationSender) ChangeCustomerColor(_ context.Context, _ swt.Session, sid, colorID, _, cid string) error { + s.colorCalls, s.sentSID, s.colorID, s.colorCID = s.colorCalls+1, sid, colorID, cid + return s.err +} + type partialSender struct { textCalls int imageCalls int @@ -1063,4 +1210,30 @@ func (r *resultRecorder) UpdateMessageStatus(_ context.Context, inboxID, message return nil } +type classificationResultCall struct { + inboxID int64 + conversationID uint + eventID string + operation string + status string + errorCode string +} + +type classificationResultRecorder struct { + resultRecorder + calls []classificationResultCall + err error +} + +func (r *classificationResultRecorder) UpdateClassificationStatus(_ context.Context, inboxID int64, conversationID uint, eventID, operation, status, _, _, errorCode, _ string) error { + if r.err != nil { + return r.err + } + r.calls = append(r.calls, classificationResultCall{ + inboxID: inboxID, conversationID: conversationID, eventID: eventID, + operation: operation, status: status, errorCode: errorCode, + }) + return nil +} + func stringPointer(value string) *string { return &value } diff --git a/channels/shangwutong/internal/store/outbound.go b/channels/shangwutong/internal/store/outbound.go index d98e7d8b..f80b76cd 100644 --- a/channels/shangwutong/internal/store/outbound.go +++ b/channels/shangwutong/internal/store/outbound.go @@ -134,7 +134,7 @@ type OutboundOperationInput struct { func (s *Store) EnqueueOutboundOperation(ctx context.Context, input OutboundOperationInput) (*dbgen.OutboundOperation, bool, error) { if input.AccountID <= 0 || strings.TrimSpace(input.SWTSessionID) == "" || strings.TrimSpace(input.EventID) == "" || - (input.Operation != "end_conversation" && input.Operation != "change_contact_name" && input.Operation != "accept_transfer") || input.OccurredAt.IsZero() { + (input.Operation != "end_conversation" && input.Operation != "change_contact_name" && input.Operation != "accept_transfer" && input.Operation != "transfer_conversation" && input.Operation != "set_chat_kind" && input.Operation != "set_customer_color") || input.OccurredAt.IsZero() { return nil, false, errors.New("valid outbound operation fields are required") } existing, err := s.writerQueries.GetOutboundOperationByEventID(ctx, input.EventID) diff --git a/channels/shangwutong/internal/store/store_test.go b/channels/shangwutong/internal/store/store_test.go index 20d50ab0..54d031e2 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 != 7 { + if err != nil || version != 8 { t.Fatalf("backup version = %d, %v", version, err) } } diff --git a/channels/shangwutong/sqlc.yaml b/channels/shangwutong/sqlc.yaml index d14e4be1..f3165468 100644 --- a/channels/shangwutong/sqlc.yaml +++ b/channels/shangwutong/sqlc.yaml @@ -5,6 +5,7 @@ sql: - "db/migrations/001_init.up.sql" - "db/migrations/006_add_xst_outbound_stages.up.sql" - "db/migrations/007_add_xst_reception_sync.up.sql" + - "db/migrations/008_add_classification_operation_results.up.sql" queries: "db/queries" gen: go: