379 lines
13 KiB
Go
379 lines
13 KiB
Go
package e2e
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"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"`
|
|
ConversationDisplayID uint `json:"conversation_display_id"`
|
|
AdminEmail string `json:"admin_email"`
|
|
AdminPassword string `json:"admin_password"`
|
|
}
|
|
|
|
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)
|
|
evidenceDir := os.Getenv("GOCHAT_WS_E2E_EVIDENCE_DIR")
|
|
if evidenceDir == "" {
|
|
evidenceDir = t.TempDir()
|
|
}
|
|
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)
|
|
}
|
|
|
|
stamp := strconv.FormatInt(time.Now().UnixNano(), 10)
|
|
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"
|
|
portA, portB := freePort(t), freePort(t)
|
|
for portB == portA {
|
|
portB = freePort(t)
|
|
}
|
|
baseURLA := fmt.Sprintf("http://127.0.0.1:%d", portA)
|
|
baseURLB := fmt.Sprintf("http://127.0.0.1:%d", portB)
|
|
instanceA := startTestProcess(t, root, binary, portA, serverEnv, filepath.Join(evidenceDir, "instance-a.log"))
|
|
instanceB := startTestProcess(t, root, binary, portB, serverEnv, filepath.Join(evidenceDir, "instance-b.log"))
|
|
t.Cleanup(instanceB.stop)
|
|
t.Cleanup(instanceA.stop)
|
|
waitForHealth(t, baseURLA)
|
|
waitForHealth(t, baseURLB)
|
|
|
|
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)
|
|
|
|
content := "multi-instance fanout " + stamp
|
|
created := createMessage(t, baseURLA, authHeaders, seedData, content)
|
|
frameA := readMessageCreated(t, connA, content)
|
|
frameB := readMessageCreated(t, connB, content)
|
|
|
|
require.Equal(t, frameA, frameB, "both processes must deliver the complete identical frame")
|
|
assertFanoutContract(t, frameA, identifier, created, seedData, content)
|
|
|
|
report["status"] = "passed"
|
|
report["instance_a"] = baseURLA
|
|
report["instance_b"] = baseURLB
|
|
report["frame"] = frameA
|
|
}
|
|
|
|
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.Create(logPath)
|
|
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 (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(t *testing.T, baseURL string) {
|
|
t.Helper()
|
|
client := &http.Client{Timeout: time.Second}
|
|
deadline := time.Now().Add(45 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
response, err := client.Get(baseURL + "/health")
|
|
if err == nil {
|
|
response.Body.Close()
|
|
if response.StatusCode == http.StatusOK {
|
|
return
|
|
}
|
|
}
|
|
time.Sleep(250 * time.Millisecond)
|
|
}
|
|
t.Fatalf("GoChat did not become healthy: %s", baseURL)
|
|
}
|
|
|
|
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 createMessage(t *testing.T, baseURL string, authHeaders http.Header, seed fanoutSeed, content string) map[string]any {
|
|
t.Helper()
|
|
body, err := json.Marshal(map[string]any{"content": content, "message_type": "outgoing", "private": false})
|
|
require.NoError(t, err)
|
|
endpoint := fmt.Sprintf("%s/api/v1/accounts/%d/conversations/%d/messages", baseURL, seed.AccountID, seed.ConversationDisplayID)
|
|
request, err := http.NewRequest(http.MethodPost, endpoint, bytes.NewReader(body))
|
|
require.NoError(t, err)
|
|
request.Header.Set("Content-Type", "application/json")
|
|
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 created map[string]any
|
|
require.NoError(t, json.NewDecoder(response.Body).Decode(&created))
|
|
return created
|
|
}
|
|
|
|
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 string) {
|
|
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)
|
|
for _, key := range []string{
|
|
"id", "account_id", "inbox_id", "conversation_id", "content", "message_type",
|
|
"content_type", "status", "private", "external", "sender_type", "created_at", "conversation",
|
|
} {
|
|
require.Contains(t, payload, key)
|
|
}
|
|
for _, key := range []string{"id", "account_id", "inbox_id", "conversation_id", "content", "message_type", "content_type", "status", "private", "external"} {
|
|
require.Equal(t, created[key], payload[key], "payload field %s must match the HTTP contract", key)
|
|
}
|
|
require.Equal(t, content, payload["content"])
|
|
require.Equal(t, "User", payload["sender_type"])
|
|
require.Contains(t, payload["conversation"].(map[string]any), "last_activity_at")
|
|
}
|
|
|
|
func mapKeys(value map[string]any) []string {
|
|
result := make([]string, 0, len(value))
|
|
for key := range value {
|
|
result = append(result, key)
|
|
}
|
|
return result
|
|
}
|