H-193: add durable transfer acceptance (#31)

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

channels/shangwutong/README.md

@@ -75,9 +75,12 @@ http://shangwutong:9100/webhooks/gochat/v1
 GET  /healthz
 GET  /readyz
 GET  /metrics
-POST /internal/reconcile    # 仅 loopback
+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` 同步结果为准。
+
 ```bash
 shangwutong migrate up
 shangwutong migrate status

channels/shangwutong/db/migrations/004_add_accept_transfer_operation.down.sql

@@ -0,0 +1,22 @@
+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 CHECK(operation IN ('end_conversation', 'change_contact_name')),
+  payload TEXT NOT NULL,
+  occurred_at DATETIME NOT NULL,
+  delivery_status TEXT NOT NULL DEFAULT 'pending' CHECK(delivery_status IN ('pending', 'delivering', 'delivered', 'uncertain', 'failed')),
+  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
+);
+-- Preserve the source table if accept_transfer rows make rollback unsafe.
+INSERT INTO outbound_operations_old SELECT * 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';

channels/shangwutong/db/migrations/004_add_accept_transfer_operation.up.sql

@@ -0,0 +1,23 @@
+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,
+  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')),
+  CHECK(delivery_status IN ('pending', 'delivering', 'delivered', 'uncertain', 'failed'))
+);
+INSERT INTO outbound_operations_new SELECT * 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';

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

channels/shangwutong/internal/delivery/outbound.go

@@ -37,6 +37,7 @@ type MessageSender interface {
 	SendVoice(context.Context, swt.Session, string, swt.Upload) (swt.SendResult, error)
 	EndConversation(context.Context, swt.Session, string) error
 	ChangeContactName(context.Context, swt.Session, string, string, string) error
+	AcceptTransfer(context.Context, swt.Session, string) error
 }
 
 type ResultClient interface {
@@ -410,6 +411,8 @@ func (o *Outbound) processOperation(ctx context.Context) (bool, error) {
 		switch operation.Operation {
 		case "end_conversation":
 			return o.sender.EndConversation(ctx, session, operation.SwtSid)
+		case "accept_transfer":
+			return o.sender.AcceptTransfer(ctx, session, operation.SwtSid)
 		case "change_contact_name":
 			var envelope struct {
 				Data struct {

channels/shangwutong/internal/delivery/outbound_test.go

@@ -160,6 +160,42 @@ func TestOutboundConversationEndOperationUsesSameDurableWorker(t *testing.T) {
 	}
 }
 
+func TestOutboundAcceptTransferPersistsSuccessAndRetry(t *testing.T) {
+	ctx := context.Background()
+	database, account := deliveryDatabase(t, ctx)
+	first, _, err := database.EnqueueOutboundOperation(ctx, store.OutboundOperationInput{
+		AccountID: account.ID, SWTSessionID: "visitor", EventID: "accept-transfer:visitor:1",
+		Operation: "accept_transfer", Payload: `{}`, OccurredAt: time.Now(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	sender := &operationSender{}
+	worker, _ := NewOutbound(database, sessionStub{}, sender, &resultRecorder{}, nil, 1)
+	if worked, err := worker.processOperation(ctx); err != nil || !worked {
+		t.Fatalf("success = %v, %v", worked, err)
+	}
+	loaded, err := database.Reader().GetOutboundOperationByEventID(ctx, first.EventID)
+	if err != nil || loaded.DeliveryStatus != "delivered" || sender.acceptSID != "visitor" {
+		t.Fatalf("loaded=%#v sender=%#v err=%v", loaded, sender, err)
+	}
+	second, _, err := database.EnqueueOutboundOperation(ctx, store.OutboundOperationInput{
+		AccountID: account.ID, SWTSessionID: "visitor", EventID: "accept-transfer:visitor:2",
+		Operation: "accept_transfer", Payload: `{}`, OccurredAt: time.Now().Add(time.Second),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	sender.err = &swt.Error{Operation: "accept_transfer", Code: "server_err", Retryable: true, Err: errors.New("retry")}
+	if worked, err := worker.processOperation(ctx); err != nil || !worked {
+		t.Fatalf("retry = %v, %v", worked, err)
+	}
+	loaded, err = database.Reader().GetOutboundOperationByEventID(ctx, second.EventID)
+	if err != nil || loaded.DeliveryStatus != "pending" || loaded.Attempts != 1 || loaded.NextAttemptAt == nil || loaded.LastError == nil {
+		t.Fatalf("retried = %#v, %v", loaded, err)
+	}
+}
+
 func TestOutboundRetrySkipsAlreadyDeliveredParts(t *testing.T) {
 	ctx := context.Background()
 	database, account := deliveryDatabase(t, ctx)
@@ -317,7 +353,13 @@ func (s senderStub) ChangeContactName(context.Context, swt.Session, string, stri
 	return s.err
 }
 
-type operationSender struct{ sid string }
+func (s senderStub) AcceptTransfer(context.Context, swt.Session, string) error { return s.err }
+
+type operationSender struct {
+	sid       string
+	acceptSID string
+	err       error
+}
 
 func (*operationSender) SendText(context.Context, swt.Session, string, string) (swt.SendResult, error) {
 	return swt.SendResult{}, nil
@@ -343,6 +385,11 @@ func (*operationSender) ChangeContactName(context.Context, swt.Session, string,
 	return nil
 }
 
+func (s *operationSender) AcceptTransfer(_ context.Context, _ swt.Session, sid string) error {
+	s.acceptSID = sid
+	return s.err
+}
+
 type partialSender struct {
 	textCalls  int
 	imageCalls int
@@ -375,6 +422,8 @@ func (*partialSender) ChangeContactName(context.Context, swt.Session, string, st
 	return nil
 }
 
+func (*partialSender) AcceptTransfer(context.Context, swt.Session, string) error { return nil }
+
 func testUploadFetcher(t *testing.T) mediaFetcher {
 	t.Helper()
 	path := filepath.Join(t.TempDir(), "image.png")

channels/shangwutong/internal/httpapi/server.go

@@ -184,9 +184,56 @@ func (s *Server) registerRoutes() {
 		}
 		return c.Status(http.StatusAccepted).JSON(fiber.Map{"accepted": true})
 	})
+	s.app.Post("/internal/operations/accept-transfer", s.acceptTransfer)
 	s.app.Post("/webhooks/gochat/v1", s.handleWebhook)
 }
 
+type acceptTransferRequest struct {
+	InboxID    int64     `json:"inbox_id"`
+	SID        string    `json:"sid"`
+	EventID    string    `json:"event_id"`
+	OccurredAt time.Time `json:"occurred_at"`
+}
+
+func (s *Server) acceptTransfer(c fiber.Ctx) error {
+	if ip := net.ParseIP(c.IP()); ip == nil || !ip.IsLoopback() {
+		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)
+	}
+	request.SID, request.EventID = strings.TrimSpace(request.SID), strings.TrimSpace(request.EventID)
+	payload, _ := json.Marshal(request)
+	account, err := s.store.Reader().GetAccountByInboxID(c.Context(), request.InboxID)
+	if errors.Is(err, sql.ErrNoRows) {
+		return s.writeError(c, http.StatusNotFound, "account_not_found", "connector account was not found", false)
+	}
+	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{
+		AccountID: account.ID, SWTSessionID: request.SID, EventID: request.EventID,
+		Operation: "accept_transfer", Payload: string(payload), OccurredAt: request.OccurredAt,
+	})
+	if errors.Is(err, store.ErrOutboundConflict) {
+		return s.writeError(c, http.StatusConflict, "idempotency_conflict", "accept transfer operation conflicts with an existing request", false)
+	}
+	if err != nil {
+		return s.writeError(c, http.StatusServiceUnavailable, "queue_failed", "accept transfer operation persistence failed", true)
+	}
+	status := http.StatusAccepted
+	if duplicate {
+		status = http.StatusOK
+	}
+	return c.Status(status).JSON(fiber.Map{
+		"accepted": true, "duplicate": duplicate, "event_id": request.EventID, "queue_id": queued.ID,
+	})
+}
+
 func (s *Server) handleWebhook(c fiber.Ctx) error {
 	body := append([]byte(nil), c.Body()...)
 	envelope, err := gochat.DecodeWebhook(body)

channels/shangwutong/internal/httpapi/server_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"io"
+	"net"
 	"net/http"
 	"path/filepath"
 	"strconv"
@@ -269,6 +270,49 @@ func TestConversationResolveWebhookQueuesDurableEndOperation(t *testing.T) {
 	}
 }
 
+func TestAcceptTransferRequestQueuesDurablyAndIsIdempotent(t *testing.T) {
+	server, database, now := newTestServer(t)
+	lifecycle := webhookBody(t, now, "inbox_created", map[string]any{"channel_type": "shangwutong", "config_version": 1})
+	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})
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	served := make(chan error, 1)
+	go func() { served <- server.App().Listener(listener) }()
+	t.Cleanup(func() {
+		_ = server.Shutdown(context.Background())
+		<-served
+	})
+	request := func(body []byte) *http.Response {
+		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 {
+			t.Fatal(err)
+		}
+		return response
+	}
+	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)
+	}
+	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"}`))
+	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 TestHealthMiddlewareAddsRequestIDAndRejectsLargeBodies(t *testing.T) {
 	server, _, _ := newTestServer(t)
 	request, _ := http.NewRequest(http.MethodGet, "/healthz", nil)

channels/shangwutong/internal/store/outbound.go

@@ -127,7 +127,8 @@ 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.OccurredAt.IsZero() {
+	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() {
 		return nil, false, errors.New("valid outbound operation fields are required")
 	}
 	existing, err := s.writerQueries.GetOutboundOperationByEventID(ctx, input.EventID)

channels/shangwutong/internal/store/store_test.go

@@ -181,6 +181,36 @@ func TestRollbackMigrationRefusesRenameOperations(t *testing.T) {
 	}
 }
 
+func TestRollbackMigrationRefusesAcceptTransferOperations(t *testing.T) {
+	db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "rollback.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`CREATE TABLE outbound_operations (
+		id INTEGER PRIMARY KEY AUTOINCREMENT, account_id INTEGER NOT NULL, 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, claimed_at DATETIME,
+		attempts INTEGER NOT NULL, next_attempt_at DATETIME, last_error TEXT,
+		created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`INSERT INTO outbound_operations(account_id,swt_sid,event_id,operation,payload,occurred_at,delivery_status,attempts,created_at,updated_at) VALUES (1,'sid','event','accept_transfer','{}',CURRENT_TIMESTAMP,'pending',0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)`); err != nil {
+		t.Fatal(err)
+	}
+	rollback, err := os.ReadFile(filepath.Join("..", "..", "db", "migrations", "004_add_accept_transfer_operation.down.sql"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(string(rollback)); err == nil {
+		t.Fatal("expected rollback to reject accept_transfer")
+	}
+	var count int
+	if err := db.QueryRow("SELECT COUNT(*) FROM outbound_operations").Scan(&count); err != nil || count != 1 {
+		t.Fatalf("operation count = %d, err=%v", count, err)
+	}
+}
+
 func TestRollbackMigrationKeepsOldOperationsWhenRenameIsAbsent(t *testing.T) {
 	db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "rollback.db"))
 	if err != nil {
@@ -560,7 +590,7 @@ func TestOnlineBackupCanBeOpenedReadOnly(t *testing.T) {
 		t.Fatalf("backup mode = %v", info.Mode().Perm())
 	}
 	version, err := InspectDatabase(ctx, backup)
-	if err != nil || version != 3 {
+	if err != nil || version != 4 {
 		t.Fatalf("backup version = %d, %v", version, err)
 	}
 }