diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 58b99e42..6d07b7f7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,6 +43,8 @@ jobs: run: deploy/docker/observability_test.sh - name: Test quality gate failure contracts working-directory: backend + env: + GOCHAT_QUALITY_DIR: ${{ github.workspace }}/backend/quality/sqlite/failure-injection run: python3 scripts/quality_gate_test.py - name: Test working-directory: backend @@ -106,6 +108,15 @@ jobs: --health-interval 5s --health-timeout 5s --health-retries 5 + redis: + image: redis:7.4.2-alpine@sha256:02419de7eddf55aa5bcf49efb74e88fa8d931b4d77c07eff8a6b2144472b6952 + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 5s + --health-retries 5 steps: - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5 @@ -242,6 +253,20 @@ jobs: if empty: raise SystemExit(f"required E2E suites ran no passing subtests: {', '.join(empty)}") PY + - name: Test multi-instance WebSocket fanout + working-directory: backend + env: + GOCHAT_MULTI_INSTANCE_E2E: "1" + GOCHAT_MULTI_INSTANCE_DATABASE_DSN: postgres://postgres:postgres@localhost:5432/gochat_test?sslmode=disable + GOCHAT_REDIS_DSN: redis://localhost:6379 + GOCHAT_WS_E2E_EVIDENCE_DIR: ${{ github.workspace }}/backend/quality/postgres/websocket-multi-instance + run: | + set -euo pipefail + mkdir -p "$GOCHAT_WS_E2E_EVIDENCE_DIR" + go test -json -count=1 -timeout 2m ./tests/e2e -run '^TestWebSocketMultiInstanceFanout$' \ + | tee "$GOCHAT_WS_E2E_EVIDENCE_DIR/test.jsonl" + grep -Eq '"Action":"pass".*"Test":"TestWebSocketMultiInstanceFanout"' \ + "$GOCHAT_WS_E2E_EVIDENCE_DIR/test.jsonl" - name: Upload PostgreSQL quality evidence if: always() uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 diff --git a/backend/scripts/quality_gate_test.py b/backend/scripts/quality_gate_test.py index 66a1e7a5..7b0b74f2 100755 --- a/backend/scripts/quality_gate_test.py +++ b/backend/scripts/quality_gate_test.py @@ -2,6 +2,7 @@ """Failure-injection checks for CI quality gates.""" import json +import os import subprocess import sys import tempfile @@ -60,6 +61,22 @@ def gate(directory, budget=5, exit_code=0): ) +def assert_auditable_failure(result, directory, name, expected): + report = json.loads((directory / "report.json").read_text(encoding="utf-8")) + assert result.returncode != 0, f"{name} must fail" + assert report["status"] == "failed" + assert expected in report["failures"] + assert expected in result.stdout + print(result.stdout.strip()) + + if evidence := os.getenv("GOCHAT_QUALITY_DIR"): + target = Path(evidence) + target.mkdir(parents=True, exist_ok=True) + (target / f"{name}.json").write_text( + json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + + def main(): with tempfile.TemporaryDirectory() as temp: directory = Path(temp) @@ -97,10 +114,14 @@ def main(): {"Action": "fail", "Package": "github.com/gochat/gochat/internal/ws", "Test": "TestRelay"} ], ) - assert gate(directory).returncode != 0, "failed test evidence must fail" + assert_auditable_failure( + gate(directory), directory, "action-fail", "test evidence contains 1 failed test events" + ) write_fixtures(directory) - assert gate(directory, exit_code=2).returncode != 0, "non-zero go test status must fail" + assert_auditable_failure( + gate(directory, exit_code=2), directory, "nonzero-test-exit", "go test exited with 2" + ) health = directory / "health.json" health.write_text( diff --git a/backend/tests/e2e/websocket_multi_instance_e2e_test.go b/backend/tests/e2e/websocket_multi_instance_e2e_test.go new file mode 100644 index 00000000..bacf05d5 --- /dev/null +++ b/backend/tests/e2e/websocket_multi_instance_e2e_test.go @@ -0,0 +1,378 @@ +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 +}