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
}