H-198: harden transfer acceptance boundaries (#32)

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

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

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 = ?

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,

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

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)

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