package node_test

import (
	"context"
	"errors"
	"net"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	toki "git.toki-labs.com/toki/proto-socket/go"
	"go.uber.org/zap"
	"google.golang.org/protobuf/proto"

	"iop/apps/node/internal/transport"
	runtime "iop/packages/go/execution"
	iop "iop/proto/gen/iop"
)

// --- test doubles used by run/cancel/terminal tests ---

// countingAdapter reports one execute call per invocation and records the last spec.
type countingAdapter struct {
	executeCalls int32
	lastSpec     runtime.ExecutionSpec
}

func (a *countingAdapter) Name() string { return "test" }
func (a *countingAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{AdapterName: "test", Targets: []string{"v1"}, MaxConcurrency: 1}, nil
}
func (a *countingAdapter) Execute(_ context.Context, spec runtime.ExecutionSpec, _ runtime.EventSink) error {
	atomic.AddInt32(&a.executeCalls, 1)
	a.lastSpec = spec
	return nil
}

// failingAdapter returns a fixed error from Execute.
type failingAdapter struct{ err error }

func (a *failingAdapter) Name() string { return "failing" }
func (a *failingAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{AdapterName: "failing"}, nil
}
func (a *failingAdapter) Execute(_ context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
	return a.err
}

// blockingAdapter blocks until its context is cancelled, then returns ErrRunCancelled.
type blockingAdapter struct {
	started chan struct{}
	done    chan struct{}
}

func newBlockingAdapter() *blockingAdapter {
	return &blockingAdapter{started: make(chan struct{}), done: make(chan struct{})}
}

func (a *blockingAdapter) Name() string { return "blocking" }
func (a *blockingAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{AdapterName: "blocking"}, nil
}
func (a *blockingAdapter) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
	close(a.started)
	<-ctx.Done()
	close(a.done)
	return runtime.ErrRunCancelled
}

// countingAdapterNoTerminal returns success without emitting any terminal event.
type countingAdapterNoTerminal struct {
	executeCalls int32
}

func (a *countingAdapterNoTerminal) Name() string { return "no-terminal" }
func (a *countingAdapterNoTerminal) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{AdapterName: "no-terminal", MaxConcurrency: 1}, nil
}
func (a *countingAdapterNoTerminal) Execute(_ context.Context, spec runtime.ExecutionSpec, _ runtime.EventSink) error {
	atomic.AddInt32(&a.executeCalls, 1)
	return nil
}

// failingAdapterNoTerminal returns an error without emitting terminal events.
type failingAdapterNoTerminal struct{ err error }

func (a *failingAdapterNoTerminal) Name() string { return "no-term-err" }
func (a *failingAdapterNoTerminal) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{AdapterName: "no-term-err", MaxConcurrency: 1}, nil
}
func (a *failingAdapterNoTerminal) Execute(_ context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
	return a.err
}

// --- run tests ---

func TestOnRunRequest_RouterError(t *testing.T) {
	n, _ := makeNode(t, &errorRouter{err: errors.New("boom")})

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-1"})
	if err == nil {
		t.Fatal("expected error")
	}
	if !strings.Contains(err.Error(), "node: resolve:") {
		t.Fatalf("expected resolve prefix, got %v", err)
	}
}

func TestOnRunRequest_AdapterNotFound(t *testing.T) {
	router := &fixedRouter{adapterName: "missing", adapters: make(map[string]runtime.Provider)}
	n, _ := makeNode(t, router)

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{RunId: "run-1"})
	if err == nil {
		t.Fatal("expected error")
	}
	if !strings.Contains(err.Error(), "not found") {
		t.Fatalf("expected adapter lookup error, got %v", err)
	}
}

func TestOnRunRequest_Success(t *testing.T) {
	adapter := &countingAdapter{}
	router := &fixedRouter{adapterName: "test", adapters: make(map[string]runtime.Provider)}
	router.adapters["test"] = adapter
	n, st := makeNode(t, router)

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId:   "run-1",
		Adapter: "test",
		Target:  "v1",
	})
	if err != nil {
		t.Fatalf("run request: %v", err)
	}
	if got := atomic.LoadInt32(&adapter.executeCalls); got != 1 {
		t.Fatalf("expected 1 execute call, got %d", got)
	}
	run, err := st.GetRun(context.Background(), "run-1")
	if err != nil {
		t.Fatalf("get run: %v", err)
	}
	if run == nil {
		t.Fatal("expected persisted run record")
	}
	if run.Status != "completed" {
		t.Fatalf("expected completed status, got %q", run.Status)
	}
	if adapter.lastSpec.ResponseStallTimeoutMS != runtime.DefaultResponseStallTimeoutMS {
		t.Fatalf("default response stall timeout = %d, want %d", adapter.lastSpec.ResponseStallTimeoutMS, runtime.DefaultResponseStallTimeoutMS)
	}
}

func TestOnRunRequestRetainsValidatedStallTimeout(t *testing.T) {
	cases := []struct {
		name      string
		raw       int64
		want      int64
		wantError bool
	}{
		{name: "positive override", raw: 45000, want: 45000},
		{name: "negative rejected before adapter", raw: -1, wantError: true},
		{name: "overflow rejected before adapter", raw: 99999999999999, wantError: true},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			adapter := &countingAdapter{}
			rtr := &fixedRouter{adapterName: "test", adapters: map[string]runtime.Provider{"test": adapter}}
			n, _ := makeNode(t, rtr)
			err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
				RunId:                  "run-stall-" + tc.name,
				Adapter:                "test",
				Target:                 "v1",
				TimeoutSec:             17,
				ResponseStallTimeoutMs: tc.raw,
			})
			if (err != nil) != tc.wantError {
				t.Fatalf("OnRunRequest error = %v, want error=%t", err, tc.wantError)
			}
			if tc.wantError {
				if got := atomic.LoadInt32(&adapter.executeCalls); got != 0 {
					t.Fatalf("adapter execute calls = %d, want 0", got)
				}
				return
			}
			if adapter.lastSpec.ResponseStallTimeoutMS != tc.want {
				t.Errorf("response stall timeout = %d, want %d", adapter.lastSpec.ResponseStallTimeoutMS, tc.want)
			}
			if adapter.lastSpec.TimeoutSec != 17 {
				t.Errorf("hard timeout = %d, want 17", adapter.lastSpec.TimeoutSec)
			}
		})
	}
}

func TestOnRunRequest_ForegroundAdapterErrorReturned(t *testing.T) {
	adapter := &failingAdapter{err: errors.New("adapter boom")}
	router := &fixedRouter{adapterName: "failing", adapters: make(map[string]runtime.Provider)}
	router.adapters["failing"] = adapter
	n, st := makeNode(t, router)

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId:   "run-fail",
		Adapter: "failing",
		Target:  "v1",
	})
	if err == nil {
		t.Fatal("expected error from OnRunRequest")
	}
	if !strings.Contains(err.Error(), "adapter boom") {
		t.Fatalf("expected adapter boom in error, got %v", err)
	}

	run, err := st.GetRun(context.Background(), "run-fail")
	if err != nil {
		t.Fatalf("get run: %v", err)
	}
	if run == nil {
		t.Fatal("expected persisted run record")
	}
	if run.Status != "failed" {
		t.Fatalf("expected failed status, got %q", run.Status)
	}
	if !strings.Contains(run.Error, "adapter boom") {
		t.Fatalf("expected adapter boom in stored error, got %q", run.Error)
	}
}

func TestOnRunRequest_ForegroundCancelReturnedAndStored(t *testing.T) {
	adapter := &failingAdapter{err: runtime.ErrRunCancelled}
	router := &fixedRouter{adapterName: "failing", adapters: make(map[string]runtime.Provider)}
	router.adapters["failing"] = adapter
	n, st := makeNode(t, router)

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId:   "run-cancel-fg",
		Adapter: "failing",
		Target:  "v1",
	})
	if !errors.Is(err, runtime.ErrRunCancelled) {
		t.Fatalf("expected ErrRunCancelled, got %v", err)
	}
	run, err := st.GetRun(context.Background(), "run-cancel-fg")
	if err != nil {
		t.Fatalf("get run: %v", err)
	}
	if run == nil || run.Status != "cancelled" {
		t.Fatalf("expected cancelled status, got %+v", run)
	}
}

func TestOnRunRequest_BackgroundReturnsBeforeAdapterCompletes(t *testing.T) {
	ba := newBlockingAdapter()
	router := &fixedRouter{adapterName: "blocking", adapters: make(map[string]runtime.Provider)}
	router.adapters["blocking"] = ba
	n, st := makeNode(t, router)

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId:      "run-bg",
		Adapter:    "blocking",
		Target:     "v1",
		Background: true,
	})
	if err != nil {
		t.Fatalf("OnRunRequest: %v", err)
	}

	// OnRunRequest should return before adapter finishes.
	select {
	case <-ba.started:
	case <-time.After(2 * time.Second):
		t.Fatal("adapter never started")
	}

	// Cancel via OnCancel so the blocking adapter finishes.
	if err := n.OnCancel(context.Background(), &transport.Session{}, &iop.CancelRequest{RunId: "run-bg"}); err != nil {
		t.Fatalf("OnCancel: %v", err)
	}

	select {
	case <-ba.done:
	case <-time.After(2 * time.Second):
		t.Fatal("adapter did not finish after cancel")
	}

	// Wait briefly for completeRun goroutine to update the store.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		run, err := st.GetRun(context.Background(), "run-bg")
		if err != nil {
			t.Fatalf("get run: %v", err)
		}
		if run != nil && run.Status == "cancelled" {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	run, _ := st.GetRun(context.Background(), "run-bg")
	t.Fatalf("expected cancelled status, got %q", run.Status)
}

func TestOnCancel_CancelsRunViaRunManager(t *testing.T) {
	ba := newBlockingAdapter()
	router := &fixedRouter{adapterName: "blocking", adapters: make(map[string]runtime.Provider)}
	router.adapters["blocking"] = ba
	n, _ := makeNode(t, router)

	go func() {
		_ = n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
			RunId:   "run-cancel",
			Adapter: "blocking",
			Target:  "v1",
		})
	}()

	<-ba.started

	if err := n.OnCancel(context.Background(), &transport.Session{}, &iop.CancelRequest{RunId: "run-cancel"}); err != nil {
		t.Fatalf("OnCancel: %v", err)
	}

	select {
	case <-ba.done:
	case <-time.After(2 * time.Second):
		t.Fatal("adapter did not finish after cancel")
	}
}

// --- terminal event synthesis tests ---

// TestOnRunRequestEmitsCompleteWhenAdapterReturnsWithoutTerminal verifies that when an
// adapter returns nil (success) without emitting any terminal event, Node synthesizes
// a complete event so Edge can release the in_flight slot.
func TestOnRunRequestEmitsCompleteWhenAdapterReturnsWithoutTerminal(t *testing.T) {
	adapter := &countingAdapterNoTerminal{}
	router := &fixedRouter{adapterName: "no-terminal", adapters: make(map[string]runtime.Provider)}
	router.adapters["no-terminal"] = adapter
	n, st := makeNode(t, router)

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId:   "run-no-term",
		Adapter: "no-terminal",
		Target:  "v1",
	})
	if err != nil {
		t.Fatalf("run request: %v", err)
	}

	// Store should show completed (synthesized terminal event processed).
	run, err := st.GetRun(context.Background(), "run-no-term")
	if err != nil {
		t.Fatalf("get run: %v", err)
	}
	if run == nil || run.Status != "completed" {
		t.Fatalf("expected completed status, got %q", run.Status)
	}
	if atomic.LoadInt32(&adapter.executeCalls) != 1 {
		t.Fatalf("expected 1 execute call, got %d", adapter.executeCalls)
	}
}

// TestOnRunRequestEmitsErrorWhenAdapterReturnsErrorWithoutTerminal verifies that when an
// adapter returns a non-cancel error without emitting a terminal event, Node synthesizes
// an error event.
func TestOnRunRequestEmitsErrorWhenAdapterReturnsErrorWithoutTerminal(t *testing.T) {
	failing := &failingAdapterNoTerminal{err: errors.New("stream closed")}
	router := &fixedRouter{adapterName: "no-term-err", adapters: make(map[string]runtime.Provider)}
	router.adapters["no-term-err"] = failing
	n, st := makeNode(t, router)

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId:   "run-no-term-err",
		Adapter: "no-term-err",
		Target:  "v1",
	})
	if err == nil {
		t.Fatal("expected error from OnRunRequest")
	}
	if !strings.Contains(err.Error(), "stream closed") {
		t.Fatalf("expected 'stream closed' in error, got %v", err)
	}

	// Store should show failed.
	run, err := st.GetRun(context.Background(), "run-no-term-err")
	if err != nil {
		t.Fatalf("get run: %v", err)
	}
	if run == nil || run.Status != "failed" {
		t.Fatalf("expected failed status, got %q", run.Status)
	}
}

// TestResolveAdapterErrorObservedByEdge verifies that when ResolveAdapter fails,
// Node returns an error. The Edge-observable RunEvent delivery is validated by
// the integration test TestIntegration_ResolveAdapterErrorObservedByEdge in
// node_concurrency_integration_test.go.
func TestResolveAdapterErrorObservedByEdge(t *testing.T) {
	router := &errorRouter{err: errors.New("adapter not found")}
	n, _ := makeNode(t, router)

	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId:   "run-resolve-fail",
		Adapter: "nonexistent",
		Target:  "v1",
	})
	if err == nil {
		t.Fatal("expected error from OnRunRequest on ResolveAdapter failure")
	}
	if !strings.Contains(err.Error(), "node: resolve:") {
		t.Fatalf("expected resolve prefix, got %v", err)
	}
}

// TestOnRunRequestSuccessTerminalCarriesNoHealthEvidence proves the normalized
// health evidence and connection-scoped observation sequence are confined to the
// stall terminal: a successful run over a bound session emits a completion event
// with no stall/health metadata.
func TestOnRunRequestSuccessTerminalCarriesNoHealthEvidence(t *testing.T) {
	adapter := &countingAdapter{}
	router := &fixedRouter{adapterName: "test", adapters: map[string]runtime.Provider{"test": adapter}}
	n, _ := makeNode(t, router)

	edgeConn, nodeConn := net.Pipe()
	edge := toki.NewTcpClient(edgeConn, 0, 0, toki.ParserMap{
		toki.TypeNameOf(&iop.RunEvent{}): func(b []byte) (proto.Message, error) {
			m := &iop.RunEvent{}
			return m, proto.Unmarshal(b, m)
		},
	})
	nodeSide := toki.NewTcpClient(nodeConn, 0, 0, toki.ParserMap{})
	t.Cleanup(func() { _ = edge.Close(); _ = nodeSide.Close() })
	events := make(chan *iop.RunEvent, 8)
	toki.AddListenerTyped[*iop.RunEvent](&edge.Communicator, func(e *iop.RunEvent) {
		events <- proto.Clone(e).(*iop.RunEvent)
	})
	sess := transport.ExportNewSession(nodeSide, zap.NewNop(), "node-id-1", "alias-1")

	if err := n.OnRunRequest(context.Background(), sess, &iop.RunRequest{RunId: "run-health-scope", Adapter: "test", Target: "v1"}); err != nil {
		t.Fatalf("run request: %v", err)
	}

	stallKeys := []string{"provider_health", "liveness_classification", "health_observation_seq", "attempt_fence", "failure_code"}
	select {
	case ev := <-events:
		if ev.GetType() != string(runtime.EventTypeComplete) {
			t.Fatalf("terminal type = %q, want complete", ev.GetType())
		}
		meta := ev.GetMetadata()
		for _, key := range stallKeys {
			if _, present := meta[key]; present {
				t.Fatalf("successful run terminal leaked stall/health key %q: %#v", key, meta)
			}
		}
	case <-time.After(2 * time.Second):
		t.Fatal("no run terminal emitted")
	}
}
