// Malformed health and refresh races need controlled HTTP peers. The real
// daemon E2E verifies attachment and lifecycle behavior with valid health.
package daemon

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"
)

const healthyHealth = `{"ok":true,"version":2,"capabilities":["operation_receipts","session_stream_generation","agents_stream_overflow"],"build":"verified-build"}`

func TestAttachValidatesHealthBeforeReturningAConnection(t *testing.T) {
	for name, body := range map[string]string{
		"missing ok":                    `{"version":2,"capabilities":[]}`,
		"not ready":                     `{"ok":false,"version":2,"capabilities":[]}`,
		"fractional version":            `{"ok":true,"version":2.5,"capabilities":[]}`,
		"missing version":               `{"ok":true,"capabilities":[]}`,
		"nonpositive version":           `{"ok":true,"version":0,"capabilities":[]}`,
		"missing capabilities":          `{"ok":true,"version":2}`,
		"null capabilities":             `{"ok":true,"version":2,"capabilities":null}`,
		"null capability":               `{"ok":true,"version":2,"capabilities":[null]}`,
		"invalid capability":            `{"ok":true,"version":2,"capabilities":[false]}`,
		"invalid build":                 `{"ok":true,"version":2,"capabilities":[],"build":false}`,
		"null build":                    `{"ok":true,"version":2,"capabilities":[],"build":null}`,
		"missing required capabilities": `{"ok":true,"version":2,"capabilities":[]}`,
		"unsupported protocol":          `{"ok":true,"version":3,"capabilities":["operation_receipts","session_stream_generation","agents_stream_overflow"]}`,
	} {
		t.Run(name, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = fmt.Fprint(w, body) }))
			defer server.Close()
			attached, err := Attach(t.Context(), ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port, Token: "token"}, nil)
			if attached != nil || err == nil {
				t.Fatalf("invalid health attached: %+v %v", attached, err)
			}
			if name == "missing required capabilities" || name == "unsupported protocol" {
				if _, ok := errors.AsType[*CompatibilityError](err); !ok {
					t.Fatalf("compatibility failure lost type: %v", err)
				}
			} else if _, ok := errors.AsType[*ProtocolError](err); !ok {
				t.Fatalf("malformed health lost type: %v", err)
			}
		})
	}
}

func TestProbeHealthPreservesAuthenticationAndCancellationWithoutRediscovery(t *testing.T) {
	for _, cancelProbe := range []bool{false, true} {
		t.Run(fmt.Sprint(cancelProbe), func(t *testing.T) {
			entered := make(chan struct{})
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				close(entered)
				if cancelProbe {
					<-r.Context().Done()
					return
				}
				w.WriteHeader(http.StatusForbidden)
				_, _ = fmt.Fprint(w, `{"code":"authentication_required","error":"invalid bearer"}`)
			}))
			defer server.Close()
			var rediscoveries atomic.Int32
			conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port, Token: "old"}, func(context.Context) (ConnectionSnapshot, error) {
				rediscoveries.Add(1)
				return ConnectionSnapshot{}, errors.New("must not rediscover")
			})
			ctx, cancel := context.WithCancel(t.Context())
			defer cancel()
			finished := make(chan error, 1)
			go func() { _, err := ProbeHealth(ctx, conn); finished <- err }()
			select {
			case <-entered:
			case <-time.After(time.Second):
				t.Fatal("health request never started")
			}
			if cancelProbe {
				cancel()
			}
			select {
			case err := <-finished:
				if cancelProbe {
					if !errors.Is(err, context.Canceled) {
						t.Fatalf("cancellation lost: %v", err)
					}
				} else {
					failure, ok := errors.AsType[*APIError](err)
					if !ok || failure.StatusCode != http.StatusForbidden || failure.Code != "authentication_required" {
						t.Fatalf("authentication lost: %v", err)
					}
				}
			case <-time.After(time.Second):
				t.Fatal("health request ignored cancellation")
			}
			if rediscoveries.Load() != 0 {
				t.Fatal("health probe attempted rediscovery")
			}
		})
	}
}

func TestRefreshGateCancellationAndSuccessfulAttachment(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path == "/health" {
			_, _ = fmt.Fprint(w, healthyHealth)
			return
		}
		if r.Header.Get("Authorization") != "Bearer new" {
			t.Errorf("request used stale credential: %s", r.Header.Get("Authorization"))
		}
		_, _ = fmt.Fprint(w, `{"ok":true}`)
	}))
	defer server.Close()
	entered, release := make(chan struct{}), make(chan struct{})
	var calls atomic.Int32
	candidate := ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port, Token: "new", Version: 99, Build: "unverified"}
	conn := NewConnection(ConnectionSnapshot{Port: candidate.Port, Token: "old"}, func(ctx context.Context) (ConnectionSnapshot, error) {
		calls.Add(1)
		close(entered)
		select {
		case <-release:
			return candidate, nil
		case <-ctx.Done():
			return ConnectionSnapshot{}, ctx.Err()
		}
	})
	defer conn.HTTPClient().CloseIdleConnections()
	finished := make(chan error, 1)
	go func() { finished <- conn.Refresh(t.Context()) }()
	<-entered
	waiting, cancel := context.WithCancel(t.Context())
	cancel()
	if err := conn.Refresh(waiting); !errors.Is(err, context.Canceled) {
		t.Fatalf("refresh wait cancellation lost: %v", err)
	}
	close(release)
	if err := <-finished; err != nil {
		t.Fatal(err)
	}
	if calls.Load() != 1 {
		t.Fatalf("canceled waiter rediscovered: %d", calls.Load())
	}
	if conn.Version() != ProtocolVersion || conn.Build() != "verified-build" {
		t.Fatalf("attachment trusted unverified discovery: %+v", conn.Snapshot())
	}
	if _, err := requestBytes(t.Context(), conn, operation{Name: "read", Method: http.MethodGet, Path: "/read", Policy: readRecovery}, responseLimits{bodyBytes: 50 * 1024 * 1024, errorBytes: 64 * 1024}); err != nil {
		t.Fatal(err)
	}
}

func TestOperationSurfacesRefreshFailureWithoutReplaying(t *testing.T) {
	for _, malformed := range []bool{false, true} {
		t.Run(fmt.Sprint(malformed), func(t *testing.T) {
			var operations, rediscoveries atomic.Int32
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if r.URL.Path == "/health" {
					_, _ = fmt.Fprint(w, `{"ok":true,"version":2,"capabilities":null}`)
					return
				}
				operations.Add(1)
				w.WriteHeader(http.StatusForbidden)
				_, _ = fmt.Fprint(w, `{"code":"authentication_required"}`)
			}))
			defer server.Close()
			cause := errors.New("discovery inaccessible")
			conn := NewConnection(ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port, Token: "old"}, func(context.Context) (ConnectionSnapshot, error) {
				rediscoveries.Add(1)
				if !malformed {
					return ConnectionSnapshot{}, cause
				}
				return ConnectionSnapshot{Port: server.Listener.Addr().(*net.TCPAddr).Port, Token: "new"}, nil
			})
			_, err := requestBytes(t.Context(), conn, operation{Name: "change", Method: http.MethodPost, Path: "/mutation", Policy: authRecovery}, responseLimits{bodyBytes: 50 * 1024 * 1024, errorBytes: 64 * 1024})
			if malformed {
				if _, ok := errors.AsType[*ProtocolError](err); !ok {
					t.Fatalf("health failure hidden: %v", err)
				}
			} else if !errors.Is(err, cause) {
				t.Fatalf("discovery failure hidden: %v", err)
			}
			if operations.Load() != 1 || rediscoveries.Load() != 1 || conn.Token() != "old" {
				t.Fatalf("failed refresh changed endpoint or replayed: operations=%d rediscoveries=%d token=%s", operations.Load(), rediscoveries.Load(), conn.Token())
			}
		})
	}
}