creator-hub/internal/controlplane/api/private_messages_worker_test.go

package api

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"os"
	"sync/atomic"
	"testing"
	"time"

	"git.ipao.vip/rogee/creator-hub/internal/account"
	"git.ipao.vip/rogee/creator-hub/internal/creator"
	hub "git.ipao.vip/rogee/creator-hub/internal/environment"
)

func privateWorkerFixture(t *testing.T) (*creator.Store, creator.ListenerState) {
	t.Helper()
	databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL")
	if databaseURL == "" {
		t.Skip("requires CREATORHUB_POSTGRES_TEST_URL")
	}
	store, accounts, ctx := openCreatorIntegrationStoreForAPITest(t, databaseURL)
	owner := account.Account{ID: "dm-owner", Platform: creator.PlatformDouyin, Name: "本人", PlatformAccountKey: "123", CredentialReference: account.CredentialReference{ID: "dm-credential", Provider: "os_keyring"}, CredentialKey: "creatorhub/dm-owner/cookies"}
	if err := accounts.CreateAccount(ctx, owner, &testCredentialBridge{values: make(map[string]string)}); err != nil {
		t.Fatal(err)
	}
	if _, err := store.RecordVerifiedLoginResult(ctx, owner.ID, "123"); err != nil {
		t.Fatal(err)
	}
	state, err := store.SetListenerEnabled(ctx, owner.ID, true)
	if err != nil {
		t.Fatal(err)
	}
	return store, state
}
func TestPrivateGatewaySendAndSync(t *testing.T) {
	store, state := privateWorkerFixture(t)
	ctx := context.Background()
	var mode atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if mode.Load() == 1 {
			w.WriteHeader(502)
			return
		}
		var payload map[string]any
		json.NewDecoder(r.Body).Decode(&payload)
		if payload["expected_uid"] != "123" {
			t.Errorf("identity=%v", payload)
		}
		if r.URL.Path == "/v1/browsers/browser/douyin/action" {
			w.Write([]byte(`{"status":"succeeded","success":true,"message":{"server_id":"100"}}`))
			return
		}
		if r.URL.Path != "/v1/browsers/browser/douyin/inbox" {
			t.Errorf("path=%s", r.URL.Path)
		}
		if mode.Load() == 2 {
			w.Write([]byte(`{"status":"succeeded","account_uid":"123","messages":[]}`))
			return
		}
		if mode.Load() == 3 {
			w.Write([]byte(`{"status":"failed","code":"IM_SDK_NOT_READY"}`))
			return
		}
		if mode.Load() == 4 {
			w.Write([]byte(`{"status":"succeeded","account_uid":"123","messages":[],"peers":[{"peer_uid":"456","peer_name":"改名后昵称","error":""}],"checkpoints":{"456":"101"}}`))
			return
		}
		if mode.Load() == 5 {
			w.Write([]byte(`{"status":"succeeded","account_uid":"123","messages":[],"peers":[{"peer_uid":"456","peer_name":"","error":"昵称读取失败"}],"checkpoints":{"456":"101"}}`))
			return
		}
		w.Write([]byte(`{"status":"succeeded","account_uid":"123","peers":[{"peer_uid":"456","peer_name":"真实昵称","error":""}],"messages":[{"peer_uid":"456","server_id":"101","sender_uid":"456","message_type":"7","content":"{\"aweType\":700,\"text\":\"你好\"}","created_at":"1700000000000"}],"checkpoints":{"456":"101"}}`))
	}))
	defer server.Close()
	accounts := listenerAccountsStub{gateway: hub.Gateway{Endpoint: server.URL, Token: "token"}, environment: hub.EnvironmentContext{Env: hub.Env{Alias: "browser", Gateway: "gateway"}, RuntimeID: "runtime", RuntimeNetworkID: "network", BindingVersion: 7}}
	input := creator.PrivateMessageSendInput{RequestID: "send-one", AccountID: state.AccountID, PeerUID: "456", Text: "你好"}
	reservation, err := store.BeginPrivateMessage(ctx, input)
	if err != nil {
		t.Fatal(err)
	}
	result := gatewayPrivateMessageSender(accounts, store)(ctx, input, reservation)
	if result.State != "succeeded" || result.ServerID != "100" {
		t.Fatalf("send=%+v", result)
	}
	checkpoints, err := syncPrivateInbox(ctx, accounts, store, state, map[string]string{"456": "99"})
	if err != nil || checkpoints["456"] != "101" {
		t.Fatalf("sync=%v %v", checkpoints, err)
	}
	page, err := store.ListPrivateMessages(ctx, state.AccountID, "456", 1, 25)
	if err != nil || page.Total != 2 {
		t.Fatalf("page=%+v %v", page, err)
	}
	for _, m := range []int32{1, 2, 3} {
		mode.Store(m)
		if _, err := syncPrivateInbox(ctx, accounts, store, state, checkpoints); err == nil {
			t.Fatalf("accepted mode %d", m)
		}
	}
	mode.Store(4)
	refreshed, err := syncPrivateInbox(ctx, accounts, store, state, checkpoints)
	if err != nil || refreshed["456"] != "101" {
		t.Fatalf("nickname-only sync=%+v %v", refreshed, err)
	}
	conversations, err := store.ListPrivateConversations(ctx, state.AccountID, 1, 25)
	if err != nil || conversations.Data[0].PeerName != "改名后昵称" || conversations.Data[0].MessageCount != 2 {
		t.Fatalf("nickname refresh without new messages=%+v %v", conversations, err)
	}
	mode.Store(5)
	if refreshed, err = syncPrivateInbox(ctx, accounts, store, state, checkpoints); err != nil || refreshed["456"] != "101" {
		t.Fatalf("profile error must not block checkpoint=%+v %v", refreshed, err)
	}
	statuses, err := store.ListPrivateSyncStatus(ctx)
	if err != nil || statuses[0].Error != "昵称读取失败" {
		t.Fatalf("profile error persistence=%+v %v", statuses, err)
	}
	mode.Store(1)
	result = gatewayPrivateMessageSender(accounts, store)(ctx, input, reservation)
	if result.State != "unknown" {
		t.Fatalf("HTTP error=%+v", result)
	}
	invalid := accounts
	invalid.environmentErr = errors.New("missing environment")
	result = gatewayPrivateMessageSender(invalid, store)(ctx, input, reservation)
	if result.State != "failed" {
		t.Fatalf("missing environment=%+v", result)
	}
	if _, err := syncPrivateInbox(ctx, invalid, store, state, nil); err == nil {
		t.Fatal("missing environment accepted")
	}
	if _, err := store.SetListenerEnabled(ctx, state.AccountID, false); err != nil {
		t.Fatal(err)
	}
	result = gatewayPrivateMessageSender(accounts, store)(ctx, input, reservation)
	if result.State != "failed" {
		t.Fatalf("disabled=%+v", result)
	}
	if _, err := privateListenerState(ctx, store, "missing"); !errors.Is(err, creator.ErrNotFound) {
		t.Fatalf("missing=%v", err)
	}
}
func TestPrivateSyncWorkerRecoveryAndErrors(t *testing.T) {
	for _, failure := range []bool{false, true} {
		t.Run(map[bool]string{false: "success", true: "failure"}[failure], func(t *testing.T) {
			store, state := privateWorkerFixture(t)
			input := creator.PrivateMessageSendInput{RequestID: "interrupted", AccountID: state.AccountID, PeerUID: "456", Text: "你好"}
			if _, err := store.BeginPrivateMessage(context.Background(), input); err != nil {
				t.Fatal(err)
			}
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if failure {
					w.WriteHeader(502)
					return
				}
				w.Write([]byte(`{"status":"succeeded","account_uid":"123","messages":[],"peers":[],"checkpoints":{}}`))
			}))
			defer server.Close()
			accounts := listenerAccountsStub{gateway: hub.Gateway{Endpoint: server.URL}, environment: hub.EnvironmentContext{Env: hub.Env{Alias: "browser", Gateway: "gateway"}, RuntimeID: "runtime", RuntimeNetworkID: "network", BindingVersion: 7}}
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			done := make(chan struct{})
			go func() { defer close(done); RunPrivateMessageSync(ctx, accounts, store) }()
			deadline := time.Now().Add(3 * time.Second)
			matched := false
			for time.Now().Before(deadline) {
				statuses, err := store.ListPrivateSyncStatus(context.Background())
				if err != nil {
					t.Fatal(err)
				}
				if len(statuses) == 1 && (failure && statuses[0].Error != "" || !failure && statuses[0].SyncedAt != nil) {
					matched = true
					break
				}
				time.Sleep(10 * time.Millisecond)
			}
			if !matched {
				t.Fatal("sync result not persisted")
			}
			cancel()
			select {
			case <-done:
			case <-time.After(time.Second):
				t.Fatal("worker did not stop")
			}
			page, err := store.ListPrivateMessages(context.Background(), state.AccountID, "456", 1, 25)
			if err != nil || page.Data[0].State != "unknown" {
				t.Fatalf("interrupted=%+v %v", page, err)
			}
		})
	}
}