From 67eb80cb2b3e2dc50e0e92c2385451f407767936 Mon Sep 17 00:00:00 2001 From: Rogee Date: Sat, 15 Aug 2026 15:22:05 +0800 Subject: [PATCH] H-193: add durable transfer acceptance (#31) Co-authored-by: Rogee --- channels/shangwutong/README.md | 5 +- ...004_add_accept_transfer_operation.down.sql | 22 ++++++++ .../004_add_accept_transfer_operation.up.sql | 23 +++++++++ .../shangwutong/internal/command/root_test.go | 2 +- .../shangwutong/internal/delivery/outbound.go | 3 ++ .../internal/delivery/outbound_test.go | 51 ++++++++++++++++++- .../shangwutong/internal/httpapi/server.go | 47 +++++++++++++++++ .../internal/httpapi/server_test.go | 44 ++++++++++++++++ .../shangwutong/internal/store/outbound.go | 3 +- .../shangwutong/internal/store/store_test.go | 32 +++++++++++- 10 files changed, 227 insertions(+), 5 deletions(-) create mode 100644 channels/shangwutong/db/migrations/004_add_accept_transfer_operation.down.sql create mode 100644 channels/shangwutong/db/migrations/004_add_accept_transfer_operation.up.sql diff --git a/channels/shangwutong/README.md b/channels/shangwutong/README.md index c26a0fb5..26482444 100644 --- a/channels/shangwutong/README.md +++ b/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 diff --git a/channels/shangwutong/db/migrations/004_add_accept_transfer_operation.down.sql b/channels/shangwutong/db/migrations/004_add_accept_transfer_operation.down.sql new file mode 100644 index 00000000..c28edfcb --- /dev/null +++ b/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'; diff --git a/channels/shangwutong/db/migrations/004_add_accept_transfer_operation.up.sql b/channels/shangwutong/db/migrations/004_add_accept_transfer_operation.up.sql new file mode 100644 index 00000000..59ca72bb --- /dev/null +++ b/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'; diff --git a/channels/shangwutong/internal/command/root_test.go b/channels/shangwutong/internal/command/root_test.go index 0fe516f0..6c5f483f 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 != 3 { + if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 4 { 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 b21662df..d896d33d 100644 --- a/channels/shangwutong/internal/delivery/outbound.go +++ b/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 { diff --git a/channels/shangwutong/internal/delivery/outbound_test.go b/channels/shangwutong/internal/delivery/outbound_test.go index 51f34030..e385bab2 100644 --- a/channels/shangwutong/internal/delivery/outbound_test.go +++ b/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") diff --git a/channels/shangwutong/internal/httpapi/server.go b/channels/shangwutong/internal/httpapi/server.go index edf9b150..5c63e60d 100644 --- a/channels/shangwutong/internal/httpapi/server.go +++ b/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) diff --git a/channels/shangwutong/internal/httpapi/server_test.go b/channels/shangwutong/internal/httpapi/server_test.go index 2c9a974f..e1b2c2cc 100644 --- a/channels/shangwutong/internal/httpapi/server_test.go +++ b/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) diff --git a/channels/shangwutong/internal/store/outbound.go b/channels/shangwutong/internal/store/outbound.go index 77b60d14..ddfee9ca 100644 --- a/channels/shangwutong/internal/store/outbound.go +++ b/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) diff --git a/channels/shangwutong/internal/store/store_test.go b/channels/shangwutong/internal/store/store_test.go index 93eb065e..674e3354 100644 --- a/channels/shangwutong/internal/store/store_test.go +++ b/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) } }