gochat/backend/tests/e2e/websocket_multi_instance_e2e_test.go

package e2e

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"mime/multipart"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/gorilla/websocket"
	"github.com/stretchr/testify/require"
)

type fanoutSeed struct {
	AccountID             uint   `json:"account_id"`
	ConversationID        uint   `json:"conversation_id"`
	ConversationDisplayID uint   `json:"conversation_display_id"`
	CaptainAssistantID    uint   `json:"captain_assistant_id"`
	AdminEmail            string `json:"admin_email"`
	AdminPassword         string `json:"admin_password"`
	WidgetToken           string `json:"widget_token"`
}

type testProcess struct {
	cancel context.CancelFunc
	cmd    *exec.Cmd
	done   chan error
	log    *os.File
}

func TestWebSocketMultiInstanceFanout(t *testing.T) {
	if os.Getenv("GOCHAT_MULTI_INSTANCE_E2E") != "1" {
		t.Skip("set GOCHAT_MULTI_INSTANCE_E2E=1 with PostgreSQL and Redis DSNs")
	}

	databaseDSN := os.Getenv("GOCHAT_MULTI_INSTANCE_DATABASE_DSN")
	if databaseDSN == "" {
		databaseDSN = os.Getenv("GOCHAT_TEST_DB_URL")
	}
	require.NotEmpty(t, databaseDSN, "GOCHAT_MULTI_INSTANCE_DATABASE_DSN or GOCHAT_TEST_DB_URL is required")
	redisDSN := os.Getenv("GOCHAT_REDIS_DSN")
	require.NotEmpty(t, redisDSN, "GOCHAT_REDIS_DSN is required")

	root := moduleRoot(t)
	stamp := strconv.FormatInt(time.Now().UnixNano(), 10)
	evidenceDir := os.Getenv("GOCHAT_WS_E2E_EVIDENCE_DIR")
	if evidenceDir == "" {
		evidenceDir = t.TempDir()
	} else {
		evidenceDir = filepath.Join(evidenceDir, stamp)
	}
	require.NoError(t, os.MkdirAll(evidenceDir, 0o755))

	report := map[string]any{"status": "failed", "instances": 2}
	defer func() {
		data, err := json.MarshalIndent(report, "", "  ")
		if err == nil {
			err = os.WriteFile(filepath.Join(evidenceDir, "fanout.json"), append(data, '\n'), 0o644)
		}
		if err != nil {
			t.Errorf("write fanout evidence: %v", err)
		}
	}()

	binary := filepath.Join(t.TempDir(), "gochat")
	build := exec.Command("go", "build", "-o", binary, "./cmd/gochat")
	build.Dir = root
	if output, err := build.CombinedOutput(); err != nil {
		t.Fatalf("build GoChat: %v\n%s", err, output)
	}

	baseEnv := map[string]string{
		"GOCHAT_ENV":                      "development",
		"GOCHAT_DATABASE_DSN":             databaseDSN,
		"GOCHAT_DATABASE_MIGRATIONS_PATH": filepath.Join(root, "migrations"),
		"GOCHAT_REDIS_DSN":                redisDSN,
		"GOCHAT_JWT_SECRET":               "ws-e2e-jwt-secret-at-least-32-characters",
		"GOCHAT_SEARCH_ENGINE":            "db",
		"GOCHAT_RATE_LIMIT_ENABLED":       "false",
		"GOCHAT_LOG_LEVEL":                "info",
		"GOCHAT_SEED_ADMIN_EMAIL":         "ws-e2e-" + stamp + "@gochat.local",
		"GOCHAT_SEED_ADMIN_PASSWORD":      "ws-e2e-password",
		"GOCHAT_SEED_ACCOUNT_NAME":        "WebSocket E2E " + stamp,
		"GOCHAT_SEED_INBOX_NAME":          "WebSocket E2E Inbox " + stamp,
	}

	seedEnv := cloneMap(baseEnv)
	seedEnv["GOCHAT_DATABASE_RUN_MIGRATIONS"] = "true"
	seed := exec.Command(binary, "seed")
	seed.Dir = root
	seed.Env = mergedEnv(seedEnv)
	seedOutput, err := seed.CombinedOutput()
	require.NoError(t, os.WriteFile(filepath.Join(evidenceDir, "seed.log"), seedOutput, 0o644))
	require.NoError(t, err, "seed GoChat; see seed.log")
	seedData := decodeFanoutSeed(t, seedOutput)

	serverEnv := cloneMap(baseEnv)
	serverEnv["GOCHAT_DATABASE_RUN_MIGRATIONS"] = "false"
	serverEnv["GOCHAT_SERVER_HOST"] = "127.0.0.1"
	instanceA, baseURLA := startHealthyTestProcess(t, root, binary, serverEnv, filepath.Join(evidenceDir, "instance-a.log"))
	t.Cleanup(instanceA.stop)
	instanceB, baseURLB := startHealthyTestProcess(t, root, binary, serverEnv, filepath.Join(evidenceDir, "instance-b.log"))
	t.Cleanup(instanceB.stop)

	authHeaders := signIn(t, baseURLA, seedData.AdminEmail, seedData.AdminPassword)
	connA, identifier := connectAccountCable(t, baseURLA, issueWSTicket(t, baseURLA, authHeaders), seedData.AccountID)
	defer connA.Close()
	connB, identifierB := connectAccountCable(t, baseURLB, issueWSTicket(t, baseURLB, authHeaders), seedData.AccountID)
	defer connB.Close()
	require.JSONEq(t, identifier, identifierB)

	agentBotID := startAITakeover(t, baseURLA, authHeaders, seedData)
	senderCases := []struct {
		name             string
		senderType       string
		senderObjectType string
		senderID         uint
		widget           bool
		attachment       bool
	}{
		{name: "agent_bot", senderType: "AgentBot", senderObjectType: "agent_bot", senderID: agentBotID},
		{name: "contact", senderType: "Contact", senderObjectType: "contact", widget: true},
		{name: "captain_assistant", senderType: "Captain::Assistant", senderObjectType: "captain_assistant", senderID: seedData.CaptainAssistantID},
		{name: "user", senderType: "User", senderObjectType: "user", attachment: true},
	}
	frames := make(map[string]any, len(senderCases))
	for _, senderCase := range senderCases {
		content := "multi-instance " + senderCase.name + " " + stamp
		var created map[string]any
		if senderCase.widget {
			createWidgetMessage(t, baseURLA, seedData, content)
		} else {
			created = createDashboardMessage(t, baseURLA, authHeaders, seedData, content, senderCase.senderType, senderCase.senderID, senderCase.attachment)
		}
		frameA := readMessageCreated(t, connA, content)
		frameB := readMessageCreated(t, connB, content)
		if created == nil {
			created = fetchHTTPMessage(t, baseURLA, authHeaders, seedData, content)
		}

		require.Equal(t, frameA, frameB, "%s sender must fan out identically", senderCase.name)
		assertFanoutContract(t, frameA, identifier, created, seedData, content, senderCase.senderType, senderCase.senderObjectType, !senderCase.widget, senderCase.attachment)
		assertFanoutContract(t, frameB, identifier, created, seedData, content, senderCase.senderType, senderCase.senderObjectType, !senderCase.widget, senderCase.attachment)
		frames[senderCase.name+"_a"] = frameA
		frames[senderCase.name+"_b"] = frameB
	}

	report["status"] = "passed"
	report["instance_a"] = baseURLA
	report["instance_b"] = baseURLB
	report["frames"] = frames
}

func moduleRoot(t *testing.T) string {
	t.Helper()
	root, err := filepath.Abs(filepath.Join("..", ".."))
	require.NoError(t, err)
	return root
}

func cloneMap(input map[string]string) map[string]string {
	result := make(map[string]string, len(input))
	for key, value := range input {
		result[key] = value
	}
	return result
}

func mergedEnv(overrides map[string]string) []string {
	env := make([]string, 0, len(os.Environ())+len(overrides))
	for _, item := range os.Environ() {
		key, _, _ := strings.Cut(item, "=")
		if _, replaced := overrides[key]; !replaced {
			env = append(env, item)
		}
	}
	for key, value := range overrides {
		env = append(env, key+"="+value)
	}
	return env
}

func decodeFanoutSeed(t *testing.T, output []byte) fanoutSeed {
	t.Helper()
	for start, value := range output {
		if value != '{' {
			continue
		}
		var seed fanoutSeed
		if json.NewDecoder(bytes.NewReader(output[start:])).Decode(&seed) == nil && seed.AccountID != 0 && seed.ConversationDisplayID != 0 {
			return seed
		}
	}
	t.Fatalf("seed output did not contain the expected summary; see seed.log")
	return fanoutSeed{}
}

func freePort(t *testing.T) int {
	t.Helper()
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	require.NoError(t, err)
	defer listener.Close()
	return listener.Addr().(*net.TCPAddr).Port
}

func startTestProcess(t *testing.T, root, binary string, port int, baseEnv map[string]string, logPath string) *testProcess {
	t.Helper()
	logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
	require.NoError(t, err)
	_, err = fmt.Fprintf(logFile, "\n=== start attempt on port %d ===\n", port)
	require.NoError(t, err)
	ctx, cancel := context.WithCancel(context.Background())
	cmd := exec.CommandContext(ctx, binary, "serve")
	cmd.Dir = root
	env := cloneMap(baseEnv)
	env["GOCHAT_SERVER_PORT"] = strconv.Itoa(port)
	cmd.Env = mergedEnv(env)
	cmd.Stdout, cmd.Stderr = logFile, logFile
	require.NoError(t, cmd.Start())
	done := make(chan error, 1)
	go func() { done <- cmd.Wait() }()
	return &testProcess{cancel: cancel, cmd: cmd, done: done, log: logFile}
}

func startHealthyTestProcess(t *testing.T, root, binary string, baseEnv map[string]string, logPath string) (*testProcess, string) {
	t.Helper()
	for attempt := 1; attempt <= 5; attempt++ {
		port := freePort(t)
		baseURL := fmt.Sprintf("http://127.0.0.1:%d", port)
		process := startTestProcess(t, root, binary, port, baseEnv, logPath)
		if waitForHealth(baseURL, process, 45*time.Second) {
			return process, baseURL
		}
		process.stop()
	}
	t.Fatalf("GoChat failed to claim a free port after 5 attempts; see %s", logPath)
	return nil, ""
}

func (process *testProcess) stop() {
	process.cancel()
	select {
	case <-process.done:
	case <-time.After(5 * time.Second):
		_ = process.cmd.Process.Kill()
		<-process.done
	}
	_ = process.log.Close()
}

func waitForHealth(baseURL string, process *testProcess, timeout time.Duration) bool {
	client := &http.Client{Timeout: time.Second}
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		response, err := client.Get(baseURL + "/health")
		if err == nil {
			response.Body.Close()
			if response.StatusCode == http.StatusOK {
				return true
			}
		}
		select {
		case processErr := <-process.done:
			process.done <- processErr
			return false
		default:
		}
		time.Sleep(250 * time.Millisecond)
	}
	return false
}

func signIn(t *testing.T, baseURL, email, password string) http.Header {
	t.Helper()
	body, err := json.Marshal(map[string]string{"email": email, "password": password})
	require.NoError(t, err)
	request, err := http.NewRequest(http.MethodPost, baseURL+"/auth/sign_in", bytes.NewReader(body))
	require.NoError(t, err)
	request.Header.Set("Content-Type", "application/json")
	response, err := http.DefaultClient.Do(request)
	require.NoError(t, err)
	defer response.Body.Close()
	require.Equal(t, http.StatusOK, response.StatusCode)
	for _, name := range []string{"access-token", "client", "uid"} {
		require.NotEmpty(t, response.Header.Get(name), "missing auth header %s", name)
	}
	return response.Header.Clone()
}

func issueWSTicket(t *testing.T, baseURL string, authHeaders http.Header) string {
	t.Helper()
	request, err := http.NewRequest(http.MethodPost, baseURL+"/api/v1/auth/ws_ticket", nil)
	require.NoError(t, err)
	for _, name := range []string{"access-token", "client", "uid", "token-type"} {
		if value := authHeaders.Get(name); value != "" {
			request.Header.Set(name, value)
		}
	}
	response, err := http.DefaultClient.Do(request)
	require.NoError(t, err)
	defer response.Body.Close()
	require.Equal(t, http.StatusOK, response.StatusCode)
	var body struct {
		Data struct {
			Ticket string `json:"ticket"`
		} `json:"data"`
	}
	require.NoError(t, json.NewDecoder(response.Body).Decode(&body))
	require.NotEmpty(t, body.Data.Ticket)
	return body.Data.Ticket
}

func connectAccountCable(t *testing.T, baseURL, ticket string, accountID uint) (*websocket.Conn, string) {
	t.Helper()
	wsURL := "ws" + strings.TrimPrefix(baseURL, "http") + "/cable?ticket=" + url.QueryEscape(ticket)
	conn, response, err := websocket.DefaultDialer.Dial(wsURL, nil)
	if err != nil {
		var body []byte
		if response != nil {
			body, _ = io.ReadAll(response.Body)
			response.Body.Close()
		}
		require.NoError(t, err, "websocket handshake status=%v body=%s", responseStatus(response), body)
	}
	require.NoError(t, conn.SetReadDeadline(time.Now().Add(10*time.Second)))
	var welcome map[string]any
	require.NoError(t, conn.ReadJSON(&welcome))
	require.Equal(t, "welcome", welcome["type"])

	identifierBytes, err := json.Marshal(map[string]any{"channel": "RoomChannel", "account_id": accountID})
	require.NoError(t, err)
	identifier := string(identifierBytes)
	require.NoError(t, conn.WriteJSON(map[string]any{"command": "subscribe", "identifier": identifier}))
	var confirmation map[string]any
	require.NoError(t, conn.ReadJSON(&confirmation))
	require.Equal(t, "confirm_subscription", confirmation["type"])
	require.JSONEq(t, identifier, confirmation["identifier"].(string))
	return conn, identifier
}

func responseStatus(response *http.Response) any {
	if response == nil {
		return nil
	}
	return response.StatusCode
}

func startAITakeover(t *testing.T, baseURL string, authHeaders http.Header, seed fanoutSeed) uint {
	t.Helper()
	endpoint := fmt.Sprintf("%s/api/v1/accounts/%d/conversations/%d/ai_takeover", baseURL, seed.AccountID, seed.ConversationDisplayID)
	request, err := http.NewRequest(http.MethodPost, endpoint, nil)
	require.NoError(t, err)
	setAuthHeaders(request, authHeaders)
	response, err := http.DefaultClient.Do(request)
	require.NoError(t, err)
	defer response.Body.Close()
	responseBody, err := io.ReadAll(response.Body)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.StatusCode, "start AI takeover response: %s", responseBody)
	var conversation map[string]any
	require.NoError(t, json.Unmarshal(responseBody, &conversation))
	assignee := conversation["meta"].(map[string]any)["assignee"].(map[string]any)
	return uint(assignee["id"].(float64))
}

func createDashboardMessage(t *testing.T, baseURL string, authHeaders http.Header, seed fanoutSeed, content, senderType string, senderID uint, attachment bool) map[string]any {
	t.Helper()
	var body bytes.Buffer
	writer := multipart.NewWriter(&body)
	for key, value := range map[string]string{
		"content":               content,
		"message_type":          "outgoing",
		"private":               "false",
		"source_id":             "ws-e2e-source",
		"echo_id":               "ws-e2e-echo",
		"content_attributes":    `{"e2e":{"enabled":true,"levels":[1,"two",{"deep":null}]}}`,
		"additional_attributes": `{"e2e":{"score":2.5,"tags":["fanout",7]}}`,
		"external_source_ids":   `{"e2e":{"source":"fanout","ids":["a",2]}}`,
	} {
		require.NoError(t, writer.WriteField(key, value))
	}
	if senderType != "User" {
		require.NoError(t, writer.WriteField("sender_type", senderType))
		require.NoError(t, writer.WriteField("sender_id", strconv.FormatUint(uint64(senderID), 10)))
	}
	if attachment {
		part, err := writer.CreateFormFile("attachments[]", "fanout.txt")
		require.NoError(t, err)
		_, err = part.Write([]byte("fanout attachment"))
		require.NoError(t, err)
	}
	require.NoError(t, writer.Close())
	endpoint := fmt.Sprintf("%s/api/v1/accounts/%d/conversations/%d/messages", baseURL, seed.AccountID, seed.ConversationDisplayID)
	request, err := http.NewRequest(http.MethodPost, endpoint, &body)
	require.NoError(t, err)
	request.Header.Set("Content-Type", writer.FormDataContentType())
	setAuthHeaders(request, authHeaders)
	response, err := http.DefaultClient.Do(request)
	require.NoError(t, err)
	defer response.Body.Close()
	responseBody, err := io.ReadAll(response.Body)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.StatusCode, "create message response: %s", responseBody)
	var created map[string]any
	require.NoError(t, json.Unmarshal(responseBody, &created))
	return created
}

func createWidgetMessage(t *testing.T, baseURL string, seed fanoutSeed, content string) {
	t.Helper()
	body, err := json.Marshal(map[string]any{"content": content, "conversation_id": seed.ConversationID})
	require.NoError(t, err)
	request, err := http.NewRequest(http.MethodPost, baseURL+"/widget/messages", bytes.NewReader(body))
	require.NoError(t, err)
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("X-Widget-Token", seed.WidgetToken)
	response, err := http.DefaultClient.Do(request)
	require.NoError(t, err)
	defer response.Body.Close()
	responseBody, err := io.ReadAll(response.Body)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.StatusCode, "create widget message response: %s", responseBody)
}

func fetchHTTPMessage(t *testing.T, baseURL string, authHeaders http.Header, seed fanoutSeed, content string) map[string]any {
	t.Helper()
	endpoint := fmt.Sprintf("%s/api/v1/accounts/%d/conversations/%d/messages", baseURL, seed.AccountID, seed.ConversationDisplayID)
	request, err := http.NewRequest(http.MethodGet, endpoint, nil)
	require.NoError(t, err)
	setAuthHeaders(request, authHeaders)
	response, err := http.DefaultClient.Do(request)
	require.NoError(t, err)
	defer response.Body.Close()
	responseBody, err := io.ReadAll(response.Body)
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.StatusCode, "list messages response: %s", responseBody)
	var result struct {
		Payload []map[string]any `json:"payload"`
	}
	require.NoError(t, json.Unmarshal(responseBody, &result))
	for _, message := range result.Payload {
		if message["content"] == content {
			return message
		}
	}
	t.Fatalf("HTTP message response did not contain %q", content)
	return nil
}

func setAuthHeaders(request *http.Request, authHeaders http.Header) {
	for _, name := range []string{"access-token", "client", "uid", "token-type"} {
		if value := authHeaders.Get(name); value != "" {
			request.Header.Set(name, value)
		}
	}
}

func readMessageCreated(t *testing.T, conn *websocket.Conn, content string) map[string]any {
	t.Helper()
	require.NoError(t, conn.SetReadDeadline(time.Now().Add(10*time.Second)))
	for {
		_, raw, err := conn.ReadMessage()
		require.NoError(t, err)
		var frame map[string]any
		require.NoError(t, json.Unmarshal(raw, &frame))
		message, _ := frame["message"].(map[string]any)
		data, _ := message["data"].(map[string]any)
		if message["event"] == "message.created" && data["content"] == content {
			return frame
		}
	}
}

func assertFanoutContract(t *testing.T, frame map[string]any, identifier string, created map[string]any, seed fanoutSeed, content, senderType, senderObjectType string, structured, attachment bool) {
	t.Helper()
	require.ElementsMatch(t, []string{"identifier", "message"}, mapKeys(frame))
	require.JSONEq(t, identifier, frame["identifier"].(string))
	message := frame["message"].(map[string]any)
	require.ElementsMatch(t, []string{"account_id", "data", "event"}, mapKeys(message))
	require.Equal(t, "message.created", message["event"])
	require.Equal(t, float64(seed.AccountID), message["account_id"])
	payload := message["data"].(map[string]any)
	assertWSTransportFields(t, payload, created, senderType)
	require.Equal(t, normalizedHTTPMessage(t, created), normalizedWSMessage(t, payload), "normalized HTTP and WebSocket message contracts must match completely")
	require.Equal(t, content, payload["content"])
	assertNestedFrontendFields(t, payload, senderObjectType, structured, attachment)
}

func assertWSTransportFields(t *testing.T, payload, created map[string]any, senderType string) {
	t.Helper()
	require.Equal(t, senderType, payload["sender_type"])
	httpSender := created["sender"].(map[string]any)
	require.Equal(t, httpSender["id"], payload["sender_id"])
	require.Nil(t, payload["deleted_at"])
	_, err := time.Parse(time.RFC3339Nano, payload["updated_at"].(string))
	require.NoError(t, err)
	conversation := payload["conversation"].(map[string]any)
	require.ElementsMatch(t, []string{"assignee_id", "contact_inbox", "last_activity_at"}, mapKeys(conversation))
	require.IsType(t, float64(0), conversation["last_activity_at"])
	contactInbox := conversation["contact_inbox"].(map[string]any)
	require.IsType(t, "", contactInbox["source_id"])
}

func normalizedHTTPMessage(t *testing.T, message map[string]any) map[string]any {
	t.Helper()
	return cloneJSONMap(t, message)
}

func normalizedWSMessage(t *testing.T, message map[string]any) map[string]any {
	t.Helper()
	normalized := cloneJSONMap(t, message)
	for _, key := range []string{"conversation", "deleted_at", "sender_id", "sender_type", "updated_at"} {
		delete(normalized, key)
	}
	return normalized
}

func cloneJSONMap(t *testing.T, value map[string]any) map[string]any {
	t.Helper()
	raw, err := json.Marshal(value)
	require.NoError(t, err)
	var clone map[string]any
	require.NoError(t, json.Unmarshal(raw, &clone))
	return clone
}

func assertNestedFrontendFields(t *testing.T, payload map[string]any, senderObjectType string, structured, attachment bool) {
	t.Helper()
	sender := payload["sender"].(map[string]any)
	require.NotEmpty(t, sender["name"].(string))
	require.Equal(t, senderObjectType, sender["type"])
	if structured {
		contentAttributes := payload["content_attributes"].(map[string]any)
		require.IsType(t, []any{}, contentAttributes["e2e"].(map[string]any)["levels"])
		additionalAttributes := payload["additional_attributes"].(map[string]any)
		require.IsType(t, []any{}, additionalAttributes["e2e"].(map[string]any)["tags"])
		externalSourceIDs := payload["external_source_ids"].(map[string]any)
		require.IsType(t, []any{}, externalSourceIDs["e2e"].(map[string]any)["ids"])
	}
	if attachment {
		attachments := payload["attachments"].([]any)
		require.Len(t, attachments, 1)
		item := attachments[0].(map[string]any)
		require.Equal(t, "file", item["file_type"])
		require.Equal(t, "txt", item["extension"])
		require.IsType(t, float64(0), item["file_size"])
	}
	require.NotContains(t, payload, "call", "non-voice messages must not expose a call object")
}

func mapKeys(value map[string]any) []string {
	result := make([]string, 0, len(value))
	for key := range value {
		result = append(result, key)
	}
	return result
}