From 76244d66352ab3525fa4d041e62f0a3e27d56a09 Mon Sep 17 00:00:00 2001 From: Rogee Date: Sat, 15 Aug 2026 15:59:22 +0800 Subject: [PATCH] H-198: harden transfer acceptance boundaries (#32) Co-authored-by: Rogee --- channels/shangwutong/README.md | 2 +- .../shangwutong/db/generated/inbound.sql.go | 19 +++ channels/shangwutong/db/queries/inbound.sql | 6 + .../shangwutong/internal/httpapi/server.go | 30 ++-- .../internal/httpapi/server_test.go | 136 ++++++++++++++++-- .../shangwutong/internal/store/outbound.go | 71 ++++++++- 6 files changed, 236 insertions(+), 28 deletions(-) diff --git a/channels/shangwutong/README.md b/channels/shangwutong/README.md index 26482444..1ebf94a3 100644 --- a/channels/shangwutong/README.md +++ b/channels/shangwutong/README.md @@ -79,7 +79,7 @@ POST /internal/reconcile # 仅 loopback POST /internal/operations/accept-transfer # 仅 loopback ``` -接管请求体为 `{"inbox_id":10,"sid":"...","event_id":"...","occurred_at":"2026-08-15T07:00:00Z"}`。相同请求返回同一队列记录；只有 operation 进入 `delivered` 才证明 `oc/accepttransfer.aspx` 返回了 `r=ok`，最终坐席归属继续以现有 `swt_state=transfer_accepted` 和 `swt_assignee_name` 同步结果为准。 +接管请求体为 `{"inbox_id":10,"conversation_id":100,"sid":"...","event_id":"...","occurred_at":"2026-08-15T07:00:00Z"}`，其中 `conversation_id` 是 GoChat internal ID。新请求只接受已映射到该 conversation 且最新商务通状态为“转接中”的 SID；相同请求返回同一队列记录。只有 operation 进入 `delivered` 才证明 `oc/accepttransfer.aspx` 返回了 `r=ok`，最终坐席归属继续以现有 `swt_state=transfer_accepted` 和 `swt_assignee_name` 同步结果为准。 ```bash shangwutong migrate up diff --git a/channels/shangwutong/db/generated/inbound.sql.go b/channels/shangwutong/db/generated/inbound.sql.go index 79179858..67ee10aa 100644 --- a/channels/shangwutong/db/generated/inbound.sql.go +++ b/channels/shangwutong/db/generated/inbound.sql.go @@ -195,6 +195,25 @@ func (q *Queries) GetInboundEventByKey(ctx context.Context, swtEventKey string) return &i, err } +const getLatestConversationState = `-- name: GetLatestConversationState :one +SELECT text FROM inbound_events +WHERE account_id = ? AND swt_sid = ? AND kind = 0 +ORDER BY id DESC +LIMIT 1 +` + +type GetLatestConversationStateParams struct { + AccountID int64 `json:"account_id"` + SwtSid string `json:"swt_sid"` +} + +func (q *Queries) GetLatestConversationState(ctx context.Context, arg GetLatestConversationStateParams) (*string, error) { + row := q.db.QueryRowContext(ctx, getLatestConversationState, arg.AccountID, arg.SwtSid) + var text *string + err := row.Scan(&text) + return text, err +} + const getMessageMapBySWTMessageID = `-- name: GetMessageMapBySWTMessageID :one SELECT account_id, swt_sid, swt_message_id, swt_seq_id, kind, child_index, direction, gochat_message_id, gochat_source_id, content_fingerprint, retracted_at, created_at, updated_at FROM message_maps WHERE account_id = ? AND swt_sid = ? AND swt_message_id = ? diff --git a/channels/shangwutong/db/queries/inbound.sql b/channels/shangwutong/db/queries/inbound.sql index 732c2977..ef02d36c 100644 --- a/channels/shangwutong/db/queries/inbound.sql +++ b/channels/shangwutong/db/queries/inbound.sql @@ -82,6 +82,12 @@ SELECT * FROM conversation_maps WHERE account_id = ? AND swt_sid = ? LIMIT 1; +-- name: GetLatestConversationState :one +SELECT text FROM inbound_events +WHERE account_id = ? AND swt_sid = ? AND kind = 0 +ORDER BY id DESC +LIMIT 1; + -- name: UpsertConversationMap :one INSERT INTO conversation_maps ( account_id, swt_sid, gochat_contact_source_id, gochat_contact_id, diff --git a/channels/shangwutong/internal/httpapi/server.go b/channels/shangwutong/internal/httpapi/server.go index 5c63e60d..b67f2ca2 100644 --- a/channels/shangwutong/internal/httpapi/server.go +++ b/channels/shangwutong/internal/httpapi/server.go @@ -189,10 +189,11 @@ func (s *Server) registerRoutes() { } type acceptTransferRequest struct { - InboxID int64 `json:"inbox_id"` - SID string `json:"sid"` - EventID string `json:"event_id"` - OccurredAt time.Time `json:"occurred_at"` + InboxID int64 `json:"inbox_id"` + ConversationID int64 `json:"conversation_id"` + SID string `json:"sid"` + EventID string `json:"event_id"` + OccurredAt time.Time `json:"occurred_at"` } func (s *Server) acceptTransfer(c fiber.Ctx) error { @@ -200,8 +201,8 @@ func (s *Server) acceptTransfer(c fiber.Ctx) error { return s.writeError(c, http.StatusForbidden, "forbidden", "loopback access required", false) } var request acceptTransferRequest - if err := json.Unmarshal(c.Body(), &request); err != nil || request.InboxID <= 0 || strings.TrimSpace(request.SID) == "" || strings.TrimSpace(request.EventID) == "" || request.OccurredAt.IsZero() { - return s.writeError(c, http.StatusUnprocessableEntity, "invalid_accept_transfer", "inbox_id, sid, event_id and occurred_at are required", false) + if err := json.Unmarshal(c.Body(), &request); err != nil || request.InboxID <= 0 || request.ConversationID <= 0 || strings.TrimSpace(request.SID) == "" || strings.TrimSpace(request.EventID) == "" || request.OccurredAt.IsZero() { + return s.writeError(c, http.StatusUnprocessableEntity, "invalid_accept_transfer", "inbox_id, conversation_id, sid, event_id and occurred_at are required", false) } request.SID, request.EventID = strings.TrimSpace(request.SID), strings.TrimSpace(request.EventID) payload, _ := json.Marshal(request) @@ -212,16 +213,23 @@ func (s *Server) acceptTransfer(c fiber.Ctx) error { if err != nil { return s.writeError(c, http.StatusServiceUnavailable, "sqlite_unavailable", "account lookup failed", true) } - if !store.AccountRunnable(account) { - return s.writeError(c, http.StatusConflict, "account_unavailable", "connector account is disabled or offline", false) - } - queued, duplicate, err := s.store.EnqueueOutboundOperation(c.Context(), store.OutboundOperationInput{ + input := store.OutboundOperationInput{ AccountID: account.ID, SWTSessionID: request.SID, EventID: request.EventID, Operation: "accept_transfer", Payload: string(payload), OccurredAt: request.OccurredAt, - }) + } + queued, duplicate, err := s.store.EnqueueAcceptTransfer(c.Context(), input, request.ConversationID) if errors.Is(err, store.ErrOutboundConflict) { return s.writeError(c, http.StatusConflict, "idempotency_conflict", "accept transfer operation conflicts with an existing request", false) } + if errors.Is(err, store.ErrOutboundAccountUnavailable) { + return s.writeError(c, http.StatusConflict, "account_unavailable", "connector account is disabled or offline", false) + } + if errors.Is(err, store.ErrOutboundSessionNotFound) { + return s.writeError(c, http.StatusConflict, "session_not_found", "sid is not mapped to the requested conversation", false) + } + if errors.Is(err, store.ErrOutboundSessionState) { + return s.writeError(c, http.StatusConflict, "session_not_transferable", "conversation is not awaiting transfer acceptance", false) + } if err != nil { return s.writeError(c, http.StatusServiceUnavailable, "queue_failed", "accept transfer operation persistence failed", true) } diff --git a/channels/shangwutong/internal/httpapi/server_test.go b/channels/shangwutong/internal/httpapi/server_test.go index e1b2c2cc..8fcd9fa2 100644 --- a/channels/shangwutong/internal/httpapi/server_test.go +++ b/channels/shangwutong/internal/httpapi/server_test.go @@ -16,6 +16,7 @@ import ( "testing" "time" + dbgen "github.com/gochat/gochat/channels/shangwutong/db/generated" "github.com/gochat/gochat/channels/shangwutong/internal/account" "github.com/gochat/gochat/channels/shangwutong/internal/gochat" "github.com/gochat/gochat/channels/shangwutong/internal/store" @@ -276,7 +277,14 @@ func TestAcceptTransferRequestQueuesDurablyAndIsIdempotent(t *testing.T) { if response := doWebhook(t, server, lifecycle, now, "secret"); response.StatusCode != http.StatusAccepted { t.Fatalf("bootstrap status = %d", response.StatusCode) } - payload, _ := json.Marshal(acceptTransferRequest{InboxID: 10, SID: "visitor", EventID: "accept-transfer:visitor:1", OccurredAt: now}) + account, err := database.Reader().GetAccountByInboxID(context.Background(), 10) + if err != nil { + t.Fatal(err) + } + seedConversationState(t, database, account.ID, "visitor", 100, "7") + seedConversationState(t, database, account.ID, "other", 200, "7") + seedConversationState(t, database, account.ID, "left", 300, "10") + payload, _ := json.Marshal(acceptTransferRequest{InboxID: 10, ConversationID: 100, SID: "visitor", EventID: "accept-transfer:visitor:1", OccurredAt: now}) listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) @@ -287,32 +295,134 @@ func TestAcceptTransferRequestQueuesDurablyAndIsIdempotent(t *testing.T) { _ = server.Shutdown(context.Background()) <-served }) - request := func(body []byte) *http.Response { + request := func(body []byte) (*http.Response, error) { req, _ := http.NewRequest(http.MethodPost, "http://"+listener.Addr().String()+"/internal/operations/accept-transfer", bytes.NewReader(body)) - response, err := http.DefaultClient.Do(req) - if err != nil { + return http.DefaultClient.Do(req) + } + type result struct { + response *http.Response + err error + } + const concurrentRequests = 12 + start, results := make(chan struct{}), make(chan result, concurrentRequests) + for range concurrentRequests { + go func() { + <-start + response, requestErr := request(payload) + results <- result{response: response, err: requestErr} + }() + } + close(start) + accepted, duplicates, queueID := 0, 0, float64(0) + for range concurrentRequests { + result := <-results + if result.err != nil { + t.Fatal(result.err) + } + var ack map[string]any + if err := json.NewDecoder(result.response.Body).Decode(&ack); err != nil { t.Fatal(err) } - return response + _ = result.response.Body.Close() + if result.response.StatusCode == http.StatusAccepted && ack["duplicate"] == false { + accepted++ + } else if result.response.StatusCode == http.StatusOK && ack["duplicate"] == true { + duplicates++ + } else { + t.Fatalf("concurrent response status=%d ack=%#v", result.response.StatusCode, ack) + } + if queueID == 0 { + queueID, _ = ack["queue_id"].(float64) + } else if ack["queue_id"] != queueID { + t.Fatalf("queue IDs differ: want %.0f, got %#v", queueID, ack["queue_id"]) + } } - first := request(payload) - if body := readBody(first); first.StatusCode != http.StatusAccepted || !strings.Contains(body, `"duplicate":false`) { - t.Fatalf("first status=%d body=%s", first.StatusCode, body) - } - second := request(payload) - if body := readBody(second); second.StatusCode != http.StatusOK || !strings.Contains(body, `"duplicate":true`) { - t.Fatalf("duplicate status=%d body=%s", second.StatusCode, body) + if accepted != 1 || duplicates != concurrentRequests-1 || queueID == 0 { + t.Fatalf("accepted=%d duplicates=%d queue_id=%.0f", accepted, duplicates, queueID) } operation, err := database.Reader().GetOutboundOperationByEventID(context.Background(), "accept-transfer:visitor:1") if err != nil || operation.Operation != "accept_transfer" || operation.SwtSid != "visitor" { t.Fatalf("operation = %#v, %v", operation, err) } - invalid := request([]byte(`{"inbox_id":10,"event_id":"missing-sid","occurred_at":"2026-08-15T07:00:00Z"}`)) + conflictingPayload, _ := json.Marshal(acceptTransferRequest{InboxID: 10, ConversationID: 200, SID: "other", EventID: "accept-transfer:visitor:1", OccurredAt: now}) + conflict, err := request(conflictingPayload) + if err != nil { + t.Fatal(err) + } + if body := readBody(conflict); conflict.StatusCode != http.StatusConflict || !strings.Contains(body, `"code":"idempotency_conflict"`) { + t.Fatalf("idempotency conflict status=%d body=%s", conflict.StatusCode, body) + } + + for name, test := range map[string]struct { + request acceptTransferRequest + code string + }{ + "other_visitor": {acceptTransferRequest{InboxID: 10, ConversationID: 100, SID: "other", EventID: "accept-transfer:other:1", OccurredAt: now}, "session_not_found"}, + "forged_sid": {acceptTransferRequest{InboxID: 10, ConversationID: 100, SID: "missing", EventID: "accept-transfer:missing:1", OccurredAt: now}, "session_not_found"}, + "expired_sid": {acceptTransferRequest{InboxID: 10, ConversationID: 300, SID: "left", EventID: "accept-transfer:left:1", OccurredAt: now}, "session_not_transferable"}, + } { + body, _ := json.Marshal(test.request) + response, requestErr := request(body) + if requestErr != nil { + t.Fatal(requestErr) + } + responseBody := readBody(response) + if response.StatusCode != http.StatusConflict || !strings.Contains(responseBody, `"code":"`+test.code+`"`) { + t.Fatalf("%s status=%d body=%s", name, response.StatusCode, responseBody) + } + } + + if _, err := database.UpsertAccountConfig(context.Background(), store.AccountConfig{ + GoChatAccountID: 1, GoChatInboxID: 10, GoChatInboxIdentifier: "identifier", ConfigVersion: 2, + SessionID: "BYT99917999", Username: "agent", Password: "password", Enabled: false, + DesiredPresence: "offline", GoChatHMACToken: "hmac", GoChatWebhookSecret: "secret", + }); err != nil { + t.Fatal(err) + } + duplicate, err := request(payload) + if err != nil { + t.Fatal(err) + } + var duplicateAck map[string]any + if err := json.NewDecoder(duplicate.Body).Decode(&duplicateAck); err != nil { + t.Fatal(err) + } + _ = duplicate.Body.Close() + if duplicate.StatusCode != http.StatusOK || duplicateAck["duplicate"] != true || duplicateAck["queue_id"] != queueID { + t.Fatalf("offline duplicate status=%d ack=%#v", duplicate.StatusCode, duplicateAck) + } + newPayload, _ := json.Marshal(acceptTransferRequest{InboxID: 10, ConversationID: 100, SID: "visitor", EventID: "accept-transfer:visitor:2", OccurredAt: now}) + newRequest, err := request(newPayload) + if err != nil { + t.Fatal(err) + } + if body := readBody(newRequest); newRequest.StatusCode != http.StatusConflict || !strings.Contains(body, `"code":"account_unavailable"`) { + t.Fatalf("offline new request status=%d body=%s", newRequest.StatusCode, body) + } + invalid, err := request([]byte(`{"inbox_id":10,"conversation_id":100,"event_id":"missing-sid","occurred_at":"2026-08-15T07:00:00Z"}`)) + if err != nil { + t.Fatal(err) + } if body := readBody(invalid); invalid.StatusCode != http.StatusUnprocessableEntity || !strings.Contains(body, `"code":"invalid_accept_transfer"`) { t.Fatalf("invalid status=%d body=%s", invalid.StatusCode, body) } } +func seedConversationState(t *testing.T, database *store.Store, accountID int64, sid string, conversationID int64, state string) { + t.Helper() + if _, err := database.Writer().UpsertConversationMap(context.Background(), dbgen.UpsertConversationMapParams{ + AccountID: accountID, SwtSid: sid, GochatContactSourceID: sid, GochatConversationID: &conversationID, + }); err != nil { + t.Fatal(err) + } + if _, err := database.Writer().InsertInboundEvent(context.Background(), dbgen.InsertInboundEventParams{ + AccountID: accountID, SwtSid: sid, SeqID: 1, Kind: 0, SwtEventKey: "state:" + sid + ":" + state, + Text: &state, RawLine: "test", DeliveryStatus: "delivered", + }); err != nil { + t.Fatal(err) + } +} + func TestHealthMiddlewareAddsRequestIDAndRejectsLargeBodies(t *testing.T) { server, _, _ := newTestServer(t) request, _ := http.NewRequest(http.MethodGet, "/healthz", nil) diff --git a/channels/shangwutong/internal/store/outbound.go b/channels/shangwutong/internal/store/outbound.go index ddfee9ca..280a2af0 100644 --- a/channels/shangwutong/internal/store/outbound.go +++ b/channels/shangwutong/internal/store/outbound.go @@ -11,7 +11,12 @@ import ( dbgen "github.com/gochat/gochat/channels/shangwutong/db/generated" ) -var ErrOutboundConflict = errors.New("outbound message state conflicts with webhook") +var ( + ErrOutboundConflict = errors.New("outbound message state conflicts with webhook") + ErrOutboundAccountUnavailable = errors.New("outbound account is disabled or offline") + ErrOutboundSessionNotFound = errors.New("outbound sid is not mapped to the requested conversation") + ErrOutboundSessionState = errors.New("outbound conversation is not awaiting transfer acceptance") +) type OutboundInput struct { AccountID int64 @@ -137,13 +142,73 @@ func (s *Store) EnqueueOutboundOperation(ctx context.Context, input OutboundOper AccountID: input.AccountID, SwtSid: input.SWTSessionID, EventID: input.EventID, Operation: input.Operation, Payload: input.Payload, OccurredAt: input.OccurredAt, }) - return created, false, createErr + if createErr == nil { + return created, false, nil + } + if !errors.Is(createErr, sql.ErrNoRows) { + return nil, false, createErr + } + existing, err = s.writerQueries.GetOutboundOperationByEventID(ctx, input.EventID) } if err != nil { return nil, false, err } - if existing.AccountID != input.AccountID || existing.SwtSid != input.SWTSessionID || existing.Operation != input.Operation || existing.Payload != input.Payload || !existing.OccurredAt.Equal(input.OccurredAt) { + if outboundOperationConflicts(existing, input) { return nil, false, ErrOutboundConflict } return existing, true, nil } + +func (s *Store) EnqueueAcceptTransfer(ctx context.Context, input OutboundOperationInput, conversationID int64) (*dbgen.OutboundOperation, bool, error) { + if input.AccountID <= 0 || strings.TrimSpace(input.SWTSessionID) == "" || strings.TrimSpace(input.EventID) == "" || + input.Operation != "accept_transfer" || input.OccurredAt.IsZero() || conversationID <= 0 { + return nil, false, errors.New("valid accept transfer fields are required") + } + var queued *dbgen.OutboundOperation + var duplicate bool + err := s.WithTx(ctx, func(queries *dbgen.Queries) error { + existing, err := queries.GetOutboundOperationByEventID(ctx, input.EventID) + if err == nil { + if outboundOperationConflicts(existing, input) { + return ErrOutboundConflict + } + queued, duplicate = existing, true + return nil + } + if !errors.Is(err, sql.ErrNoRows) { + return err + } + account, err := queries.GetAccountByID(ctx, input.AccountID) + if err != nil { + return err + } + if !AccountRunnable(account) { + return ErrOutboundAccountUnavailable + } + mapping, err := queries.GetConversationMap(ctx, dbgen.GetConversationMapParams{AccountID: input.AccountID, SwtSid: input.SWTSessionID}) + if errors.Is(err, sql.ErrNoRows) || err == nil && (mapping.GochatConversationID == nil || *mapping.GochatConversationID != conversationID) { + return ErrOutboundSessionNotFound + } + if err != nil { + return err + } + state, err := queries.GetLatestConversationState(ctx, dbgen.GetLatestConversationStateParams{AccountID: input.AccountID, SwtSid: input.SWTSessionID}) + if errors.Is(err, sql.ErrNoRows) || err == nil && (state == nil || strings.TrimSpace(*state) != "7") { + return ErrOutboundSessionState + } + if err != nil { + return err + } + queued, err = queries.InsertOutboundOperation(ctx, dbgen.InsertOutboundOperationParams{ + AccountID: input.AccountID, SwtSid: input.SWTSessionID, EventID: input.EventID, + Operation: input.Operation, Payload: input.Payload, OccurredAt: input.OccurredAt, + }) + return err + }) + return queued, duplicate, err +} + +func outboundOperationConflicts(existing *dbgen.OutboundOperation, input OutboundOperationInput) bool { + return existing.AccountID != input.AccountID || existing.SwtSid != input.SWTSessionID || existing.Operation != input.Operation || + existing.Payload != input.Payload || !existing.OccurredAt.Equal(input.OccurredAt) +}