package host

import (
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/registry"
)

type startupBackend struct {
	fakeWakeBackend
	recovered *atomic.Bool
	started   *atomic.Bool
	attempts  atomic.Int32
	gate      func() error
}

func (b *startupBackend) Acquire(ctx context.Context, rt registry.Route) (activator.Lease, error) {
	b.attempts.Add(1)
	if !b.recovered.Load() || !b.started.Load() {
		return nil, errors.New("wake preceded recovery or Start")
	}
	if b.gate != nil {
		if err := b.gate(); err != nil {
			return nil, err
		}
	}
	return b.fakeWakeBackend.Acquire(ctx, rt)
}

type observingActivator struct {
	*activator.Service
	entered chan struct{}
}

func (a observingActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
	a.entered <- struct{}{}
	return a.Service.ServeHTTP(w, r, rt, proxy)
}

func TestStartupServesWhileSweepAndRecoveryWait(t *testing.T) {
	t.Setenv("CARLOS_STEERING_ZONE_ID", "Z1234567890")
	t.Setenv("PLATFORM_APPS_DOMAIN", "apps.example.com")
	for _, fail := range []bool{false, true} {
		t.Run(map[bool]string{false: "slow sweep and recovery", true: "unreachable instance control"}[fail], func(t *testing.T) {
			dd, err := os.MkdirTemp(os.TempDir(), "startup-")
			if err != nil {
				t.Fatal(err)
			}
			t.Cleanup(func() { os.RemoveAll(dd) })
			reg, err := registry.Open(filepath.Join(dd, "registry.db"))
			if err != nil {
				t.Fatal(err)
			}
			defer reg.Close()
			inst := &fakeInstance{socket: filepath.Join(dd, "app.sock"), handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "awake") })}
			rt := registry.Route{Host: "shop.example.com", Kind: "instance", Hibernate: true, Status: "ready", Socket: inst.socket, DBPath: filepath.Join(dd, "app.db")}
			if err := reg.Upsert(rt); err != nil {
				t.Fatal(err)
			}
			console := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }))
			defer console.Close()
			if err := reg.Upsert(registry.Route{Host: "console.example.com", Addr: console.Listener.Addr().String()}); err != nil {
				t.Fatal(err)
			}
			var recovered, startedService atomic.Bool
			backend := &startupBackend{fakeWakeBackend: fakeWakeBackend{inst: inst}, recovered: &recovered, started: &startedService}
			control := instancectl.NewClient(filepath.Join(dd, "unreachable.sock"))
			if fail {
				backend.gate = nestedGate(func() error { return nil }, func(string) (bool, error) { return true, nil }, control.Ready)
			}
			svc := activator.New(activator.Config{Bucket: "unused", Idle: time.Hour}, backend, reg)
			ready := make(chan struct{})
			svc.Startup = ready
			ctx, cancel := context.WithCancel(t.Context())
			defer cancel()
			defer func() { svc.Shutdown(context.Background()) }()
			sweepEntered, recoverEntered := make(chan struct{}), make(chan struct{})
			sweepRelease, recoverRelease := make(chan struct{}), make(chan struct{})
			done := activatorStart{
				Sweep: func(ctx context.Context) error {
					close(sweepEntered)
					select {
					case <-sweepRelease:
					case <-ctx.Done():
						return ctx.Err()
					}
					if fail {
						_, err := control.Sweep(ctx)
						return err
					}
					return nil
				},
				Recover: func(ctx context.Context, swept bool) *layout.Pass {
					if swept == fail {
						t.Error("recovery received the wrong sweep result")
					}
					close(recoverEntered)
					select {
					case <-recoverRelease:
					case <-ctx.Done():
						return nil
					}
					recovered.Store(true)
					return nil
				},
				Publish: func(*layout.Pass) {}, Start: func() { svc.Start(); startedService.Store(true) }, Ready: func() { close(ready) },
				Routes: func() ([]registry.Route, error) { return nil, nil },
			}.Run(ctx)
			defer func() { cancel(); <-done }()
			<-sweepEntered
			entered := make(chan struct{}, 1)
			e, err := edge.New(reg, observingActivator{svc, entered})
			if err != nil {
				t.Fatal(err)
			}
			server := httptest.NewServer(e)
			defer server.Close()
			client := &http.Client{Timeout: time.Second}
			get := func(host string) (int, error) {
				r, _ := http.NewRequest(http.MethodGet, server.URL+"/", nil)
				r.Host = host
				resp, err := client.Do(r)
				if err != nil {
					return 0, err
				}
				io.Copy(io.Discard, resp.Body)
				resp.Body.Close()
				return resp.StatusCode, nil
			}
			for host, want := range map[string]int{"127.0.0.1": http.StatusNotFound, "console.example.com": http.StatusNoContent, "edge.apps.example.com": http.StatusOK} {
				began := time.Now()
				if got, err := get(host); err != nil || got != want {
					t.Fatalf("%s during sweep: %d, %v", host, got, err)
				}
				t.Logf("%s answered in %s while sweep was blocked", host, time.Since(began))
			}
			answer := make(chan int, 1)
			go func() { status, _ := get(rt.Host); answer <- status }()
			select {
			case <-entered:
			case <-time.After(time.Second):
				t.Fatal("request never reached the activator")
			}
			assertHeld := func() {
				t.Helper()
				select {
				case status := <-answer:
					t.Fatalf("request answered before startup: %d", status)
				case <-time.After(25 * time.Millisecond):
				}
				if backend.attempts.Load() != 0 {
					t.Fatal("wake attempted before sweep and recovery finished")
				}
			}
			assertHeld()
			close(sweepRelease)
			<-recoverEntered
			assertHeld()
			if got, err := get("console.example.com"); err != nil || got != http.StatusNoContent {
				t.Fatalf("console during recovery: %d, %v", got, err)
			}
			close(recoverRelease)
			want := http.StatusOK
			if fail {
				want = http.StatusServiceUnavailable
			}
			select {
			case got := <-answer:
				if got != want {
					t.Fatalf("instance after Start: %d, want %d", got, want)
				}
			case <-time.After(time.Second):
				t.Fatal("request did not finish after Start")
			}
			if backend.attempts.Load() != 1 || inst.running() == fail {
				t.Fatal("unexpected wake outcome")
			}
		})
	}
}

func TestStartupCancellationAndBoundedJoin(t *testing.T) {
	t.Parallel()
	ctx, cancel := context.WithCancel(context.Background())
	entered := make(chan struct{})
	var started atomic.Bool
	done := activatorStart{
		Sweep:   func(ctx context.Context) error { close(entered); <-ctx.Done(); return ctx.Err() },
		Recover: func(context.Context, bool) *layout.Pass { t.Error("recovery after cancellation"); return nil },
		Start:   func() { started.Store(true) },
	}.Run(ctx)
	<-entered
	cancel()
	join, stop := context.WithTimeout(t.Context(), time.Second)
	defer stop()
	if !joinAgentStartup(join, done) || started.Load() {
		t.Fatal("cancelled startup was not joined, or started")
	}
	blocked := make(chan struct{})
	expired, expire := context.WithTimeout(t.Context(), 20*time.Millisecond)
	defer expire()
	if joinAgentStartup(expired, blocked) {
		t.Fatal("join ignored its bound")
	}
}

func TestStartupJoinsKeyPassOnAgentCancellation(t *testing.T) {
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	entered := make(chan struct{})
	rt := registry.Route{Host: "shop.example.com", Kind: "instance", Socket: "/unused.sock", Status: "ready"}
	done := activatorStart{
		Sweep:   func(context.Context) error { return nil },
		Recover: func(context.Context, bool) *layout.Pass { return nil },
		Publish: func(*layout.Pass) {}, Start: func() {},
		Routes: func() ([]registry.Route, error) { return []registry.Route{rt}, nil },
		Route:  routeOf(rt), Slot: openSlot,
		Keys: keyBackfillFunc(func(ctx context.Context, _ registry.Route) (activator.InstanceKeyOutcome, error) {
			close(entered)
			<-ctx.Done()
			return activator.InstanceKeyFailed, ctx.Err()
		}),
	}.Run(ctx)
	select {
	case <-entered:
	case <-time.After(time.Second):
		t.Fatal("key pass did not start")
	}
	cancel()
	join, stop := context.WithTimeout(t.Context(), time.Second)
	defer stop()
	if !joinAgentStartup(join, done) {
		t.Fatal("key pass outlived agent cancellation")
	}
}

func TestStartupCancellationDuringRecoveryPreservesAgentLifetime(t *testing.T) {
	agentCtx, cancelAgent := context.WithCancel(t.Context())
	defer cancelAgent()
	startupCtx, cancelStartup := context.WithCancel(agentCtx)
	defer cancelStartup()
	recovering := make(chan struct{})
	svc := activator.New(activator.Config{}, &fakeWakeBackend{}, nil)
	ready := make(chan struct{})
	svc.Startup = ready
	done := activatorStart{
		Configure: func(context.Context) {
			if agentCtx.Err() != nil {
				t.Error("agent lifetime ended before configuration")
			}
		},
		Sweep: func(context.Context) error { return nil },
		Recover: func(ctx context.Context, _ bool) *layout.Pass {
			close(recovering)
			<-ctx.Done()
			return nil
		},
		Publish: func(*layout.Pass) {},
		Start:   func() { t.Error("Start after recovery cancellation") },
		Ready:   func() { close(ready) },
	}.Run(startupCtx)
	<-recovering
	cancelStartup()
	join, stop := context.WithTimeout(t.Context(), time.Second)
	defer stop()
	if !joinAgentStartup(join, done) {
		t.Fatal("recovery did not join")
	}
	svc.Shutdown(join)
	if agentCtx.Err() != nil {
		t.Fatal("startup cancellation stopped rollout's agent lifetime before children drained")
	}
	select {
	case <-ready:
		t.Fatal("cancelled recovery released admission")
	default:
	}
}