package callscore

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"sync"
	"testing"
	"time"

	"amadan.net/oficina/calls/protocol"
)

func TestChatOutcomeWaitsForServer(t *testing.T) {
	for _, status := range []int{http.StatusOK, http.StatusBadRequest, http.StatusRequestEntityTooLarge} {
		t.Run(fmt.Sprint(status), func(t *testing.T) {
			received := make(chan protocol.Signal, 1)
			release := make(chan struct{})
			events := make(chan Event, 1)
			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
				var signal protocol.Signal
				if err := json.NewDecoder(req.Body).Decode(&signal); err != nil {
					t.Error(err)
				}
				received <- signal
				select {
				case <-release:
					w.WriteHeader(status)
				case <-req.Context().Done():
				}
			}))
			defer srv.Close()
			r, err := New(srv.URL, "room", srv.Client(), func(e Event) { events <- e })
			if err != nil {
				t.Fatal(err)
			}
			r.started = true
			r.state.Peer = "self"
			defer r.Close()
			const payload = `{"text":"hello"}`
			if err := r.Send(protocol.TypeChat, "", 0, payload); err != nil {
				t.Fatal(err)
			}
			select {
			case signal := <-received:
				if signal.Type != protocol.TypeChat || signal.To != "" || signal.Seq != 0 || signal.Gen != 0 {
					t.Fatalf("invalid broadcast: %+v", signal)
				}
			case <-time.After(time.Second):
				t.Fatal("chat was not posted")
			}
			select {
			case e := <-events:
				t.Fatalf("outcome before server response: %+v", e)
			default:
			}
			close(release)
			select {
			case e := <-events:
				want := "send-failed"
				if status == http.StatusOK {
					want = "sent"
				}
				if e.Type != want || e.Signal == nil || string(e.Signal.Payload) != payload {
					t.Fatalf("outcome = %+v, want %s with original text", e, want)
				}
			case <-time.After(time.Second):
				t.Fatal("missing chat outcome")
			}
		})
	}
}

func TestReregisterReportsEveryQueuedChatAsFailed(t *testing.T) {
	received := make(chan struct{}, 1)
	events := make(chan Event, 8)
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		// Hold the first POST so the remaining chats must still be queued.
		var signal protocol.Signal
		if err := json.NewDecoder(req.Body).Decode(&signal); err != nil {
			t.Error(err)
		}
		received <- struct{}{}
		<-req.Context().Done()
	}))
	defer server.Close()
	r, err := New(server.URL, "room", server.Client(), func(e Event) {
		if e.Signal != nil {
			events <- e
		}
	})
	if err != nil {
		t.Fatal(err)
	}
	defer r.Close()
	r.started = true
	r.state.Peer = "old"
	for i := range 3 {
		if err := r.Send(protocol.TypeChat, "", 0, fmt.Sprintf(`{"text":"message %d"}`, i)); err != nil {
			t.Fatal(err)
		}
		if i == 0 {
			select {
			case <-received:
			case <-time.After(time.Second):
				t.Fatal("first chat was not posted")
			}
		}
	}
	r.fresh()
	r.Close() // Wait for every cancellation callback, including the queue drain.
	if len(events) != 3 {
		t.Fatalf("got %d outcomes, want all three accepted chats", len(events))
	}
	for i := range 3 {
		e := <-events
		if e.Type != "send-failed" || string(e.Signal.Payload) != fmt.Sprintf(`{"text":"message %d"}`, i) {
			t.Fatalf("outcome %d lost its text or order: %+v", i, e)
		}
	}
}

func TestResumeRetainsParticipationAndDedupe(t *testing.T) {
	var mu sync.Mutex
	opens, joins := 0, 0
	messages := make(chan Event, 64)
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		switch {
		case req.URL.Path == "/m/room/ice":
			fmt.Fprint(w, `{"iceServers":[],"expiresAt":0,"renewAt":0}`)
		case req.URL.Path == "/m/room/signal":
			var s protocol.Signal
			_ = json.NewDecoder(req.Body).Decode(&s)
			mu.Lock()
			if s.Type == "join" {
				joins++
			}
			mu.Unlock()
			w.WriteHeader(200)
		case req.URL.Path == "/m/room/stream":
			mu.Lock()
			opens++
			n := opens
			mu.Unlock()
			if req.Header.Get("Accept") != "text/event-stream" {
				t.Error("missing SSE Accept")
			}
			if n == 2 && (req.URL.Query().Get("peer") != "self" || req.URL.Query().Get("last") != `{"other":1}`) {
				t.Error("resume lost peer or cursor")
			}
			w.Header().Set("Content-Type", "text/event-stream")
			fmt.Fprint(w, "data: {\"type\":\"hello\",\"peer\":\"self\",\"boot\":\"boot\"}\n\n")
			fmt.Fprint(w, "data: {\"type\":\"roster\",\"roster\":{\"peers\":[{\"peer\":\"self\",\"kind\":\"member\"},{\"peer\":\"other\",\"gen\":1}]}}\n\n")
			fmt.Fprint(w, "data: {\"type\":\"offer\",\"from\":\"other\",\"seq\":1,\"gen\":1,\"payload\":{}}\n\n")
			w.(http.Flusher).Flush()
			if n == 1 {
				time.Sleep(50 * time.Millisecond)
				return
			}
			fmt.Fprint(w, "data: {\"type\":\"candidate\",\"from\":\"other\",\"seq\":2,\"gen\":1,\"payload\":{}}\n\n")
			w.(http.Flusher).Flush()
			<-req.Context().Done()
		default:
			w.WriteHeader(404)
		}
	}))
	defer server.Close()
	r, err := New(server.URL, "room", server.Client(), func(e Event) { messages <- e })
	if err != nil {
		t.Fatal(err)
	}
	if err = r.Start(); err != nil {
		t.Fatal(err)
	}
	defer r.Close()
	timer := time.NewTimer(5 * time.Second)
	defer timer.Stop()
	offers := 0
	for {
		select {
		case e := <-messages:
			if e.Message != nil {
				if e.Message.Type == "offer" {
					offers++
				}
				if e.Message.Type == "candidate" {
					mu.Lock()
					j := joins
					mu.Unlock()
					if offers != 1 || j != 1 {
						t.Fatalf("offers=%d joins=%d", offers, j)
					}
					return
				}
			}
		case <-timer.C:
			t.Fatal("resume never delivered")
		}
	}
}
func TestGenerationFenceAndMonotonicSequence(t *testing.T) {
	received := make(chan protocol.Signal, 10)
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		var s protocol.Signal
		_ = json.NewDecoder(req.Body).Decode(&s)
		received <- s
	}))
	defer srv.Close()
	r, _ := New(srv.URL, "room", srv.Client(), func(Event) {})
	// Seed a live participant without a stream to isolate outgoing generation
	// fencing, including callbacks that arrive after the native PC was replaced.
	r.started = true
	r.state.Peer = "self"
	r.state.Peers = []protocol.RosterPeer{{Peer: "other", Gen: 1}}
	defer r.Close()
	if err := r.Send("offer", "other", 1, `{}`); err != nil {
		t.Fatal(err)
	}
	select {
	case s := <-received:
		if s.Seq != 1 {
			t.Fatal(s.Seq)
		}
	case <-time.After(time.Second):
		t.Fatal("missing offer")
	}
	if err := r.receive(protocol.Message{Type: "pair", Peer: "other", Gen: 2}); err != nil {
		t.Fatal(err)
	}
	if err := r.Send("candidate", "other", 1, `{}`); err == nil {
		t.Fatal("accepted callback from replaced PC")
	}
	if err := r.Send("offer", "other", 2, `{}`); err != nil {
		t.Fatal(err)
	}
	select {
	case s := <-received:
		if s.Seq != 2 || s.Gen != 2 {
			t.Fatal(s)
		}
	case <-time.After(time.Second):
		t.Fatal("missing new offer")
	}
}
func TestTerminalLossClearsMediaAndDoesNotRejoin(t *testing.T) {
	r, _ := New("https://calls.example.com", "room", http.DefaultClient, func(Event) {})
	r.started = true
	r.state.Peer = "self"
	r.state.Peers = []protocol.RosterPeer{{Peer: "other", Gen: 1}}
	_ = r.receive(protocol.Message{Type: "lost", Reason: protocol.ReasonRemoved})
	s := r.Snapshot()
	if s.Phase != "ended" || s.Reason != "removed" || len(s.Peers) != 0 {
		t.Fatal(s)
	}
	if err := r.Send("join", "", 0, ""); err == nil {
		t.Fatal("terminal participant rejoined")
	}
	r.Close()
}

func TestTargetQueueRetriesSameSequenceBeforeNextMessage(t *testing.T) {
	received := make(chan protocol.Signal, 8)
	attempts := 0
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		var s protocol.Signal
		_ = json.NewDecoder(req.Body).Decode(&s)
		attempts++
		received <- s
		if attempts == 1 {
			w.WriteHeader(503)
			return
		}
		if attempts == 2 {
			w.Header().Set("Retry-After", "1")
			w.WriteHeader(429)
			return
		}
	}))
	defer srv.Close()
	r, _ := New(srv.URL, "room", srv.Client(), func(Event) {})
	r.started = true
	r.state.Peer = "self"
	r.state.Peers = []protocol.RosterPeer{{Peer: "other", Gen: 1}}
	defer r.Close()
	if err := r.Send("offer", "other", 1, `{}`); err != nil {
		t.Fatal(err)
	}
	if err := r.Send("candidate", "other", 1, `{}`); err != nil {
		t.Fatal(err)
	}
	deadline := time.NewTimer(5 * time.Second)
	defer deadline.Stop()
	for i, want := range []uint64{1, 1, 1, 2} {
		select {
		case s := <-received:
			if s.Seq != want || (i < 3 && s.Type != "offer") || (i == 3 && s.Type != "candidate") {
				t.Fatalf("attempt %d: %+v", i, s)
			}
		case <-deadline.C:
			t.Fatal("ordered retry stalled")
		}
	}
}

// A fresh open refused with 409 takes the session's own participation back
// once, and a refused takeover — or a second 409 — is "replaced" as before.
func TestFreshOpenConflictTakesOverOnce(t *testing.T) {
	for _, tc := range []struct {
		name      string
		takeover  int
		location  string
		conflicts int
		want      string
	}{
		{"taken back", 303, "/m/room", 1, "admitted"},
		{"takeover refused", 404, "", 1, "replaced"},
		{"sign-in redirect is a refusal", 303, "/signin", 1, "replaced"},
		{"only once", 303, "/m/room", 2, "replaced"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			var mu sync.Mutex
			opens, takeovers := 0, 0
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
				switch req.URL.Path {
				case "/m/room/ice":
					fmt.Fprint(w, `{"iceServers":[],"expiresAt":0,"renewAt":0}`)
				case "/m/room/signal":
					w.WriteHeader(200)
				case "/m/room/takeover":
					mu.Lock()
					takeovers++
					mu.Unlock()
					if req.Method != http.MethodPost || req.Header.Get("Origin") == "" {
						t.Error("takeover must be a POST carrying Origin")
					}
					if tc.location != "" {
						w.Header().Set("Location", tc.location)
					}
					w.WriteHeader(tc.takeover)
				case "/m/room/stream":
					mu.Lock()
					opens++
					n := opens
					mu.Unlock()
					if n <= tc.conflicts {
						w.Header().Set("Content-Type", "application/json")
						w.WriteHeader(409)
						fmt.Fprint(w, `{"peer":"ghost"}`)
						return
					}
					w.Header().Set("Content-Type", "text/event-stream")
					fmt.Fprint(w, "data: {\"type\":\"hello\",\"peer\":\"self\",\"boot\":\"boot\"}\n\n")
					fmt.Fprint(w, "data: {\"type\":\"roster\",\"roster\":{\"peers\":[{\"peer\":\"self\",\"kind\":\"member\"}]}}\n\n")
					w.(http.Flusher).Flush()
					<-req.Context().Done()
				default:
					w.WriteHeader(404)
				}
			}))
			defer server.Close()
			states := make(chan Snapshot, 64)
			r, err := New(server.URL, "room", server.Client(), func(e Event) {
				if e.State != nil {
					states <- *e.State
				}
			})
			if err != nil {
				t.Fatal(err)
			}
			if err = r.Start(); err != nil {
				t.Fatal(err)
			}
			defer r.Close()
			timer := time.NewTimer(5 * time.Second)
			defer timer.Stop()
			for {
				select {
				case s := <-states:
					got := ""
					if s.Phase == "admitted" {
						got = "admitted"
					} else if s.Phase == "ended" {
						got = s.Reason
					}
					if got == "" {
						continue
					}
					if got != tc.want {
						t.Fatalf("got %q, want %q", got, tc.want)
					}
					mu.Lock()
					defer mu.Unlock()
					if takeovers != 1 {
						t.Fatalf("takeovers = %d, want exactly 1", takeovers)
					}
					return
				case <-timer.C:
					t.Fatal("no outcome")
				}
			}
		})
	}
}