From e61f5b252f7a300f5cbfc13e0f068c213b13d7cf Mon Sep 17 00:00:00 2001 From: Rogee Date: Sat, 15 Aug 2026 19:02:52 +0800 Subject: [PATCH] H-162: add PC conversation transfer flow (#33) Co-authored-by: Rogee --- channels/shangwutong/README.md | 3 + ...d_transfer_conversation_operation.down.sql | 22 +++++ ...add_transfer_conversation_operation.up.sql | 23 +++++ .../shangwutong/internal/command/root_test.go | 2 +- .../shangwutong/internal/delivery/outbound.go | 12 +++ .../internal/delivery/outbound_test.go | 95 +++++++++++++++++-- .../shangwutong/internal/httpapi/server.go | 47 +++++++++ .../internal/httpapi/server_test.go | 78 ++++++++++++++- .../shangwutong/internal/store/outbound.go | 72 ++++++++++++++ .../shangwutong/internal/store/store_test.go | 32 ++++++- 10 files changed, 375 insertions(+), 11 deletions(-) create mode 100644 channels/shangwutong/db/migrations/005_add_transfer_conversation_operation.down.sql create mode 100644 channels/shangwutong/db/migrations/005_add_transfer_conversation_operation.up.sql diff --git a/channels/shangwutong/README.md b/channels/shangwutong/README.md index 1ebf94a3..3092a3e3 100644 --- a/channels/shangwutong/README.md +++ b/channels/shangwutong/README.md @@ -77,10 +77,13 @@ GET /readyz GET /metrics POST /internal/reconcile # 仅 loopback POST /internal/operations/accept-transfer # 仅 loopback +POST /internal/operations/transfer-conversation # 仅 loopback ``` 接管请求体为 `{"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` 同步结果为准。 +发起请求体字段相同。它只接受当前由其他坐席持有、最新状态为 `5` 的 SID，目标坐席从已同步的 `swt_assignee_name` 推导；Connector 以当前登录坐席为 `oname`、原坐席为 `oname1` 调用 `oc/Transfer0.aspx`。成功后仍须等待状态 `7` 再调用接受入口；接受后的归属以状态 `8` 及后续 `swt_assignee_name` 回执为准。 + ```bash shangwutong migrate up shangwutong migrate status diff --git a/channels/shangwutong/db/migrations/005_add_transfer_conversation_operation.down.sql b/channels/shangwutong/db/migrations/005_add_transfer_conversation_operation.down.sql new file mode 100644 index 00000000..37328ade --- /dev/null +++ b/channels/shangwutong/db/migrations/005_add_transfer_conversation_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', 'accept_transfer')), + 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 transfer_conversation 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/005_add_transfer_conversation_operation.up.sql b/channels/shangwutong/db/migrations/005_add_transfer_conversation_operation.up.sql new file mode 100644 index 00000000..3ed916aa --- /dev/null +++ b/channels/shangwutong/db/migrations/005_add_transfer_conversation_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', 'transfer_conversation')), + 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 6c5f483f..7428397f 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 != 4 { + if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 5 { 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 d896d33d..a25ae3a0 100644 --- a/channels/shangwutong/internal/delivery/outbound.go +++ b/channels/shangwutong/internal/delivery/outbound.go @@ -38,6 +38,7 @@ type MessageSender interface { EndConversation(context.Context, swt.Session, string) error ChangeContactName(context.Context, swt.Session, string, string, string) error AcceptTransfer(context.Context, swt.Session, string) error + TransferConversation(context.Context, swt.Session, string, string) error } type ResultClient interface { @@ -413,6 +414,17 @@ func (o *Outbound) processOperation(ctx context.Context) (bool, error) { return o.sender.EndConversation(ctx, session, operation.SwtSid) case "accept_transfer": return o.sender.AcceptTransfer(ctx, session, operation.SwtSid) + case "transfer_conversation": + var payload struct { + OtherLoginName string `json:"other_login_name"` + } + if err := json.Unmarshal([]byte(operation.Payload), &payload); err != nil || strings.TrimSpace(payload.OtherLoginName) == "" { + return &swt.Error{Operation: operation.Operation, Code: "invalid_payload", Err: errors.New("other_login_name is required")} + } + if !session.AllowsJoiningOtherOperatorDialogue() { + return &swt.Error{Operation: operation.Operation, Code: "permission_denied", Err: errors.New("operator may not join another operator's dialogue")} + } + return o.sender.TransferConversation(ctx, session, operation.SwtSid, payload.OtherLoginName) 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 e385bab2..930c8aa6 100644 --- a/channels/shangwutong/internal/delivery/outbound_test.go +++ b/channels/shangwutong/internal/delivery/outbound_test.go @@ -196,6 +196,67 @@ func TestOutboundAcceptTransferPersistsSuccessAndRetry(t *testing.T) { } } +func TestAcceptedTransferUpdatesOwnershipAndKeepsOutboundSendable(t *testing.T) { + ctx := context.Background() + database, account := deliveryDatabase(t, ctx) + conversationID := int64(100) + other := "other-agent" + if _, err := database.Writer().UpsertConversationMap(ctx, dbgen.UpsertConversationMapParams{ + AccountID: account.ID, SwtSid: "visitor", GochatContactSourceID: "visitor", + GochatConversationID: &conversationID, SwtAssigneeName: &other, + }); err != nil { + t.Fatal(err) + } + persistInboundEvent(t, database, account, swt.HeartbeatEvent{SessionID: "visitor", Kind: 0, Text: "5", SeqID: 1, RawLine: "visitor 0 5 1"}) + transfer, duplicate, err := database.EnqueueTransferConversation(ctx, store.OutboundOperationInput{ + AccountID: account.ID, SWTSessionID: "visitor", EventID: "transfer-conversation:visitor:1", + Operation: "transfer_conversation", OccurredAt: time.Now(), + }, conversationID) + if err != nil || duplicate { + t.Fatalf("enqueue transfer=%#v duplicate=%v err=%v", transfer, duplicate, err) + } + + sender := &operationSender{} + worker, _ := NewOutbound(database, allowedSessionStub{}, sender, &resultRecorder{}, nil, 1) + if worked, err := worker.processOperation(ctx); err != nil || !worked || sender.transferSID != "visitor" || sender.otherLoginName != other { + t.Fatalf("transfer worked=%v sender=%#v err=%v", worked, sender, err) + } + + persistInboundEvent(t, database, account, swt.HeartbeatEvent{SessionID: "visitor", Kind: 0, Text: "7", SeqID: 2, RawLine: "visitor 0 7 2"}) + accept, duplicate, err := database.EnqueueAcceptTransfer(ctx, store.OutboundOperationInput{ + AccountID: account.ID, SWTSessionID: "visitor", EventID: "accept-transfer:visitor:1", + Operation: "accept_transfer", Payload: `{}`, OccurredAt: time.Now().Add(time.Second), + }, conversationID) + if err != nil || duplicate { + t.Fatalf("enqueue accept=%#v duplicate=%v err=%v", accept, duplicate, err) + } + if worked, err := worker.processOperation(ctx); err != nil || !worked || sender.acceptSID != "visitor" { + t.Fatalf("accept worked=%v sender=%#v err=%v", worked, sender, err) + } + + persistInboundEvent(t, database, account, swt.HeartbeatEvent{SessionID: "visitor", Kind: 0, Text: "8", SeqID: 3, RawLine: "visitor 0 8 3"}) + persistInboundEvent(t, database, account, swt.HeartbeatEvent{SessionID: "visitor", Kind: 31, Text: "distribute_chat|agent", SeqID: 4, RawLine: "visitor 31 distribute_chat|agent 4"}) + inbound, _ := NewInbound(database, &inboundRecorder{}, nil, 1) + for range 4 { + if worked, err := inbound.processInbound(ctx); err != nil || !worked { + t.Fatalf("inbound worked=%v err=%v", worked, err) + } + } + mapping, err := database.Reader().GetConversationMap(ctx, dbgen.GetConversationMapParams{AccountID: account.ID, SwtSid: "visitor"}) + if err != nil || mapping.SwtAssigneeName == nil || *mapping.SwtAssigneeName != "agent" { + t.Fatalf("mapping=%#v err=%v", mapping, err) + } + if _, _, err := database.EnqueueOutbound(ctx, store.OutboundInput{ + AccountID: account.ID, SWTSessionID: "visitor", EventID: "message:77:created", OccurredAt: time.Now().Add(2 * time.Second), + GoChatMessageID: 77, MessageType: "text", Content: stringPointer("hello"), Payload: deliveryPayload(t, nil), + }, false); err != nil { + t.Fatal(err) + } + if worked, err := worker.processOutbound(ctx); err != nil || !worked || sender.sentSID != "visitor" { + t.Fatalf("outbound worked=%v sender=%#v err=%v", worked, sender, err) + } +} + func TestOutboundRetrySkipsAlreadyDeliveredParts(t *testing.T) { ctx := context.Background() database, account := deliveryDatabase(t, ctx) @@ -319,6 +380,13 @@ func (sessionStub) WithSession(ctx context.Context, _ int64, operation func(swt. func (sessionStub) InvalidateSession(context.Context, int64) error { return nil } +type allowedSessionStub struct{ sessionStub } + +func (allowedSessionStub) WithSession(ctx context.Context, _ int64, operation func(swt.Session) error) error { + purview := uint64(0) + return operation(swt.Session{BaseURL: "http://example.test/", SiteID: "site", LoginName: "agent", MAToken: "token", Purview: &purview}) +} + type sessionRecorder struct{ invalidations int } func (*sessionRecorder) WithSession(ctx context.Context, _ int64, operation func(swt.Session) error) error { @@ -354,14 +422,21 @@ func (s senderStub) ChangeContactName(context.Context, swt.Session, string, stri } func (s senderStub) AcceptTransfer(context.Context, swt.Session, string) error { return s.err } - -type operationSender struct { - sid string - acceptSID string - err error +func (s senderStub) TransferConversation(context.Context, swt.Session, string, string) error { + return s.err } -func (*operationSender) SendText(context.Context, swt.Session, string, string) (swt.SendResult, error) { +type operationSender struct { + sid string + acceptSID string + transferSID string + otherLoginName string + sentSID string + err error +} + +func (s *operationSender) SendText(_ context.Context, _ swt.Session, sid, _ string) (swt.SendResult, error) { + s.sentSID = sid return swt.SendResult{}, nil } @@ -390,6 +465,11 @@ func (s *operationSender) AcceptTransfer(_ context.Context, _ swt.Session, sid s return s.err } +func (s *operationSender) TransferConversation(_ context.Context, _ swt.Session, sid, otherLoginName string) error { + s.transferSID, s.otherLoginName = sid, otherLoginName + return s.err +} + type partialSender struct { textCalls int imageCalls int @@ -423,6 +503,9 @@ func (*partialSender) ChangeContactName(context.Context, swt.Session, string, st } func (*partialSender) AcceptTransfer(context.Context, swt.Session, string) error { return nil } +func (*partialSender) TransferConversation(context.Context, swt.Session, string, string) error { + return nil +} func testUploadFetcher(t *testing.T) mediaFetcher { t.Helper() diff --git a/channels/shangwutong/internal/httpapi/server.go b/channels/shangwutong/internal/httpapi/server.go index b67f2ca2..1911f6d2 100644 --- a/channels/shangwutong/internal/httpapi/server.go +++ b/channels/shangwutong/internal/httpapi/server.go @@ -185,6 +185,7 @@ 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("/internal/operations/transfer-conversation", s.transferConversation) s.app.Post("/webhooks/gochat/v1", s.handleWebhook) } @@ -196,6 +197,52 @@ type acceptTransferRequest struct { OccurredAt time.Time `json:"occurred_at"` } +type transferConversationRequest = acceptTransferRequest + +func (s *Server) transferConversation(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 transferConversationRequest + 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_transfer_conversation", "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) + 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) + } + queued, duplicate, err := s.store.EnqueueTransferConversation(c.Context(), store.OutboundOperationInput{ + AccountID: account.ID, SWTSessionID: request.SID, EventID: request.EventID, + Operation: "transfer_conversation", OccurredAt: request.OccurredAt, + }, request.ConversationID) + if errors.Is(err, store.ErrOutboundConflict) { + return s.writeError(c, http.StatusConflict, "idempotency_conflict", "transfer conversation 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 held by another operator in a transferable state", false) + } + if err != nil { + return s.writeError(c, http.StatusServiceUnavailable, "queue_failed", "transfer conversation 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) 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) diff --git a/channels/shangwutong/internal/httpapi/server_test.go b/channels/shangwutong/internal/httpapi/server_test.go index 8fcd9fa2..2f52e3f4 100644 --- a/channels/shangwutong/internal/httpapi/server_test.go +++ b/channels/shangwutong/internal/httpapi/server_test.go @@ -408,11 +408,83 @@ func TestAcceptTransferRequestQueuesDurablyAndIsIdempotent(t *testing.T) { } } -func seedConversationState(t *testing.T, database *store.Store, accountID int64, sid string, conversationID int64, state string) { +func TestTransferConversationRequestQueuesOtherOperatorsChat(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) + } + account, err := database.Reader().GetAccountByInboxID(context.Background(), 10) + if err != nil { + t.Fatal(err) + } + seedConversationState(t, database, account.ID, "visitor", 100, "5", "other-agent") + seedConversationState(t, database, account.ID, "already-transferring", 200, "7", "other-agent") + seedConversationState(t, database, account.ID, "owned", 300, "5", "agent") + + 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(value transferConversationRequest) *http.Response { + body, _ := json.Marshal(value) + response, requestErr := http.Post("http://"+listener.Addr().String()+"/internal/operations/transfer-conversation", "application/json", bytes.NewReader(body)) + if requestErr != nil { + t.Fatal(requestErr) + } + return response + } + + value := transferConversationRequest{InboxID: 10, ConversationID: 100, SID: "visitor", EventID: "transfer-conversation:visitor:1", OccurredAt: now} + first := request(value) + var firstAck map[string]any + if err := json.NewDecoder(first.Body).Decode(&firstAck); err != nil { + t.Fatal(err) + } + _ = first.Body.Close() + if first.StatusCode != http.StatusAccepted || firstAck["duplicate"] != false { + t.Fatalf("first status=%d ack=%#v", first.StatusCode, firstAck) + } + duplicate := request(value) + if body := readBody(duplicate); duplicate.StatusCode != http.StatusOK || !strings.Contains(body, `"duplicate":true`) { + t.Fatalf("duplicate status=%d body=%s", duplicate.StatusCode, body) + } + operation, err := database.Reader().GetOutboundOperationByEventID(context.Background(), value.EventID) + if err != nil || operation.Operation != "transfer_conversation" || !strings.Contains(operation.Payload, `"other_login_name":"other-agent"`) { + t.Fatalf("operation=%#v err=%v", operation, err) + } + + for name, test := range map[string]struct { + request transferConversationRequest + code string + }{ + "wrong_state": {transferConversationRequest{InboxID: 10, ConversationID: 200, SID: "already-transferring", EventID: "transfer-conversation:state", OccurredAt: now}, "session_not_transferable"}, + "already_owned": {transferConversationRequest{InboxID: 10, ConversationID: 300, SID: "owned", EventID: "transfer-conversation:owned", OccurredAt: now}, "session_not_transferable"}, + "forged_sid": {transferConversationRequest{InboxID: 10, ConversationID: 100, SID: "missing", EventID: "transfer-conversation:missing", OccurredAt: now}, "session_not_found"}, + } { + response := request(test.request) + body := readBody(response) + if response.StatusCode != http.StatusConflict || !strings.Contains(body, `"code":"`+test.code+`"`) { + t.Fatalf("%s status=%d body=%s", name, response.StatusCode, body) + } + } +} + +func seedConversationState(t *testing.T, database *store.Store, accountID int64, sid string, conversationID int64, state string, assignee ...string) { t.Helper() - if _, err := database.Writer().UpsertConversationMap(context.Background(), dbgen.UpsertConversationMapParams{ + params := dbgen.UpsertConversationMapParams{ AccountID: accountID, SwtSid: sid, GochatContactSourceID: sid, GochatConversationID: &conversationID, - }); err != nil { + } + if len(assignee) > 0 { + params.SwtAssigneeName = &assignee[0] + } + if _, err := database.Writer().UpsertConversationMap(context.Background(), params); err != nil { t.Fatal(err) } if _, err := database.Writer().InsertInboundEvent(context.Background(), dbgen.InsertInboundEventParams{ diff --git a/channels/shangwutong/internal/store/outbound.go b/channels/shangwutong/internal/store/outbound.go index 280a2af0..d98e7d8b 100644 --- a/channels/shangwutong/internal/store/outbound.go +++ b/channels/shangwutong/internal/store/outbound.go @@ -3,6 +3,7 @@ package store import ( "context" "database/sql" + "encoding/json" "errors" "fmt" "strings" @@ -208,6 +209,77 @@ func (s *Store) EnqueueAcceptTransfer(ctx context.Context, input OutboundOperati return queued, duplicate, err } +func (s *Store) EnqueueTransferConversation(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 != "transfer_conversation" || input.OccurredAt.IsZero() || conversationID <= 0 { + return nil, false, errors.New("valid transfer conversation 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 { + var payload struct { + ConversationID int64 `json:"conversation_id"` + } + if existing.AccountID != input.AccountID || existing.SwtSid != input.SWTSessionID || existing.Operation != input.Operation || + !existing.OccurredAt.Equal(input.OccurredAt) || json.Unmarshal([]byte(existing.Payload), &payload) != nil || payload.ConversationID != conversationID { + 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 + } + otherLoginName := "" + if mapping.SwtAssigneeName != nil { + otherLoginName = strings.TrimSpace(*mapping.SwtAssigneeName) + } + currentLoginName := account.Username + if account.LoginName != nil && strings.TrimSpace(*account.LoginName) != "" { + currentLoginName = strings.TrimSpace(*account.LoginName) + } + if otherLoginName == "" || strings.EqualFold(otherLoginName, currentLoginName) { + return ErrOutboundSessionState + } + 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) != "5") { + return ErrOutboundSessionState + } + if err != nil { + return err + } + payload, err := json.Marshal(struct { + ConversationID int64 `json:"conversation_id"` + OtherLoginName string `json:"other_login_name"` + }{conversationID, otherLoginName}) + 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: string(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) diff --git a/channels/shangwutong/internal/store/store_test.go b/channels/shangwutong/internal/store/store_test.go index 674e3354..91cb7856 100644 --- a/channels/shangwutong/internal/store/store_test.go +++ b/channels/shangwutong/internal/store/store_test.go @@ -211,6 +211,36 @@ func TestRollbackMigrationRefusesAcceptTransferOperations(t *testing.T) { } } +func TestRollbackMigrationRefusesTransferConversationOperations(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','transfer_conversation','{}',CURRENT_TIMESTAMP,'pending',0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)`); err != nil { + t.Fatal(err) + } + rollback, err := os.ReadFile(filepath.Join("..", "..", "db", "migrations", "005_add_transfer_conversation_operation.down.sql")) + if err != nil { + t.Fatal(err) + } + if _, err := db.Exec(string(rollback)); err == nil { + t.Fatal("expected rollback to reject transfer_conversation") + } + 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 { @@ -590,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 != 4 { + if err != nil || version != 5 { t.Fatalf("backup version = %d, %v", version, err) } }