//go:build linux

package host

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

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/rootstate"
)

func servingParkSocket(t *testing.T, handler http.Handler) hostbox.JournalCall {
	t.Helper()
	return rootJournalClient(servingRootSocket(t, handler))
}
func servingRootSocket(t *testing.T, handler http.Handler) string {
	t.Helper()
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "serving-park-")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	path := filepath.Join(dir, "root.sock")
	listener, err := net.Listen("unix", path)
	if err != nil {
		t.Fatal(err)
	}
	server := &instancectl.Server{ClaimStop: handler, PeerOK: func(net.Conn) error { return nil }, Logf: func(string, ...any) {}}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- server.Serve(ctx, listener) }()
	t.Cleanup(func() {
		cancel()
		if err := <-done; err != nil {
			t.Error(err)
		}
	})
	return path
}

func servingTransportRoot(t *testing.T, f *servingRootFixture) ClaimRootHTTP {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	p := &servingParkRuntime{authority: f.p, state: servingBorrowedState{f.p.ledger.state}, ctx: ctx, cancel: cancel, idle: make(chan struct{})}
	t.Cleanup(func() {
		cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := p.Close(cleanup); err != nil {
			t.Error(err)
		}
	})
	h := f.p.root
	h.servingParks = p
	h.Stop.Coordinator.RetainNoProcess = p.retainNoProcess
	return h
}

func TestServingParkSocketRetiresAndCompletes(t *testing.T) {
	for _, legacy := range []bool{false, true} {
		t.Run(map[bool]string{false: "generation", true: "legacy"}[legacy], func(t *testing.T) {
			var f *servingRootFixture
			if legacy {
				f, _ = newServingLegacyRootFixture(t, false)
			} else {
				f = newServingRootFixture(t, true)
			}
			h := servingTransportRoot(t, f)
			call := servingParkSocket(t, h)
			var exact hostbox.JournalRequest
			client := hostbox.ServingParkClient{Call: func(ctx context.Context, q hostbox.JournalRequest) (hostbox.JournalResponse, error) {
				exact = q
				return call(ctx, q)
			}}
			var a servingParkAuthorization
			var err error
			if legacy {
				a, err = client.ParkLegacy(t.Context(), f.current, servingLegacyPredecessor{})
			} else {
				a, err = client.Park(t.Context(), f.current, f.prior)
			}
			if err != nil {
				t.Fatal(err)
			}
			parkRequest := exact
			r, err := f.p.ledger.read(a.ID)
			if err != nil || !r.Parked || r.Complete {
				t.Fatalf("park: %+v %v", r, err)
			}
			if _, err = os.Stat(f.db); !errors.Is(err, os.ErrNotExist) {
				t.Fatal("source survived successful park", err)
			}
			f.tags(f.current.Stamp)
			if err = client.Complete(t.Context(), a.ID, f.current); err != nil {
				t.Fatal(err)
			}
			r, err = f.p.ledger.read(a.ID)
			if err != nil || !r.Complete {
				t.Fatalf("completion: %+v %v", r, err)
			}
			if !legacy {
				again, err := client.Park(t.Context(), f.current, f.prior)
				if err != nil || again != a {
					t.Fatalf("fresh request after completion: %+v %v", again, err)
				}
				if exact.ID == parkRequest.ID {
					t.Fatal("fresh request reused cached transport reply")
				}
			}
			before, err := f.disk.Read("retirement-" + a.ID + ".json")
			if err != nil {
				t.Fatal(err)
			}
			// A replay returns the immutable reply even after restore writes the
			// successor. It must not inspect or park those new bytes.
			if err = os.WriteFile(f.db, []byte("successor"), 0600); err != nil {
				t.Fatal(err)
			}
			out, err := call(t.Context(), parkRequest)
			if err != nil || out.Error != "" || out.ServingPark == nil || *out.ServingPark != a {
				t.Fatalf("replay: %+v %v", out, err)
			}
			after, err := f.disk.Read("retirement-" + a.ID + ".json")
			if err != nil || string(before) != string(after) {
				t.Fatal("replay changed retirement")
			}
			body, _ := os.ReadFile(f.db)
			if string(body) != "successor" {
				t.Fatal("replay touched successor")
			}
		})
	}
}

func TestServingParkSocketLostReplyReplaysExactRequest(t *testing.T) {
	f := newServingRootFixture(t, true)
	h := servingTransportRoot(t, f)
	var requests atomic.Int32
	call := servingParkSocket(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if requests.Add(1) == 1 {
			// Execute against the real connection context, then drop the reply.
			h.ServeHTTP(discardServingReply{HeaderMap: make(http.Header)}, r)
			conn, _, err := w.(http.Hijacker).Hijack()
			if err == nil {
				conn.Close()
			}
			return
		}
		h.ServeHTTP(w, r)
	}))
	client := hostbox.ServingParkClient{Call: call}
	a, err := client.Park(t.Context(), f.current, f.prior)
	if err != nil || requests.Load() != 2 {
		t.Fatalf("lost reply: %+v %v requests=%d", a, err, requests.Load())
	}
	r, err := f.p.ledger.read(a.ID)
	if err != nil || !r.Parked {
		t.Fatalf("park: %+v %v", r, err)
	}
}

type discardServingReply struct{ HeaderMap http.Header }

func (w discardServingReply) Header() http.Header       { return w.HeaderMap }
func (discardServingReply) WriteHeader(int)             {}
func (discardServingReply) Write(b []byte) (int, error) { return len(b), nil }

func TestServingParkSocketAuthorityGates(t *testing.T) {
	for _, gate := range []string{"peer", "census", "enrollment", "revision", "box", "quarantine"} {
		t.Run(gate, func(t *testing.T) {
			f := newServingRootFixture(t, true)
			h := servingTransportRoot(t, f)
			var refuse atomic.Bool
			auth := h.Stop.Authenticate
			h.Stop.Authenticate = func(ctx context.Context, c net.Conn) (StopPeer, error) {
				if refuse.Load() && gate == "peer" {
					return StopPeer{}, ErrClaimStopIdentity
				}
				return auth(ctx, c)
			}
			h.RequireNewWork = func(context.Context) error {
				if refuse.Load() && gate == "census" {
					return hostbox.ErrChanged
				}
				return nil
			}
			h.InventoryActor = func(context.Context) (hostbox.Actor, error) {
				a := f.actor
				if refuse.Load() {
					switch gate {
					case "enrollment":
						a.Enrollment = "other"
					case "revision":
						a.Revision = 0
					case "box":
						a.Box = "other"
					case "quarantine":
						a.Quarantined = true
					}
				}
				return a, nil
			}
			call := servingParkSocket(t, h)
			var q hostbox.JournalRequest
			client := hostbox.ServingParkClient{Call: func(ctx context.Context, in hostbox.JournalRequest) (hostbox.JournalResponse, error) {
				q = in
				return call(ctx, in)
			}}
			a, err := client.Park(t.Context(), f.current, f.prior)
			if err != nil {
				t.Fatal(err)
			}
			before, _ := f.disk.Read("retirement-" + a.ID + ".json")
			refuse.Store(true)
			out, err := call(t.Context(), q)
			if err == nil && out.Error == "" {
				t.Fatal("replay bypassed authority", gate)
			}
			after, _ := f.disk.Read("retirement-" + a.ID + ".json")
			if !reflect.DeepEqual(before, after) {
				t.Fatal("refusal mutated record")
			}
		})
	}
}

type servingCloseStore struct {
	rootstate.Store
	closed atomic.Bool
}

func (s *servingCloseStore) Close() error { s.closed.Store(true); return nil }

func TestServingParkRuntimeJoinBeforePrivateStateClose(t *testing.T) {
	s := &servingCloseStore{}
	ctx, cancel := context.WithCancel(context.Background())
	p := &servingParkRuntime{state: s, ctx: ctx, cancel: cancel, idle: make(chan struct{})}
	call, done, err := p.begin(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	deadline, stop := context.WithCancel(context.Background())
	stop()
	if err = p.Close(deadline); !errors.Is(err, context.Canceled) || s.closed.Load() {
		t.Fatal("closed state before join", err)
	}
	select {
	case <-call.Done():
	case <-time.After(time.Second):
		t.Fatal("active call not canceled")
	}
	if _, _, err = p.begin(t.Context()); err == nil {
		t.Fatal("new call admitted during close")
	}
	done()
	if err = p.Close(t.Context()); err != nil || !s.closed.Load() {
		t.Fatal("join did not close store", err)
	}
	if err = p.Close(t.Context()); err != nil {
		t.Fatal(err)
	}
}

func TestServingParkRuntimeRetainsNoProcess(t *testing.T) {
	// The production lifecycle wrapper must retain the same root observation
	// as the direct coordinator fixture, and refuse after shutdown.
	f, _ := newServingLegacyRootFixture(t, false)
	h := servingTransportRoot(t, f)
	saved, err := f.p.ledger.legacyStop(f.current)
	if err != nil {
		t.Fatal(err)
	}
	// Retention rechecks the original claim, so it refuses after admission.
	if err = h.Stop.Coordinator.RetainNoProcess(t.Context(), saved.Observed, saved.Result); err == nil {
		t.Fatal("retained stale preadmission observation")
	}
	if err = h.servingParks.Close(t.Context()); err != nil {
		t.Fatal(err)
	}
	if err = h.Stop.Coordinator.RetainNoProcess(t.Context(), hostclaim.Observed{}, StopResult{}); err == nil {
		t.Fatal("closed runtime retained evidence")
	}
}

func TestServingParkProductionConstructorRequiresPrivateDependencies(t *testing.T) {
	f := newServingRootFixture(t, true)
	if _, err := newServingParkRuntime(f.p.root, nil, nil); err == nil {
		t.Fatal("missing private dependencies accepted")
	}
	// Production FS construction fixes root ownership and the nosymfollow
	// check. This test does not claim to open a privileged rootstate store.
	p, err := newServingParkRuntime(f.p.root, f.disk, plan9D5Fixture(t).owners)
	if err != nil {
		t.Fatal(err)
	}
	defer p.Close(t.Context())
	fs := p.authority.(*servingParkRoot).ledger.fs.(*servingParkFS)
	if fs.owner != 0 || fs.mount == nil || p.authority.(*servingParkRoot).ledger.journal != f.p.root.Journals.Launches {
		t.Fatal("production constructor replaced root authority")
	}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	if err = p.Close(ctx); err != nil {
		t.Fatal(err)
	}
}

func TestServingParkProductionBindingRetainsSessionZero(t *testing.T) {
	old := openClaimServingParkAuthority
	t.Cleanup(func() { openClaimServingParkAuthority = old })
	var closed *servingCloseStore
	f, _ := newServingLegacyRootFixture(t, false, func(f *servingRootFixture, coordinator *ClaimStopCoordinator, request StopRequest) {
		closed = &servingCloseStore{Store: f.disk}
		openClaimServingParkAuthority = func() (rootstate.Store, error) { return closed, nil }
		h, err := withServingParkRoot(f.p.root, f.p.empty)
		if err != nil {
			t.Fatal(err)
		}
		t.Cleanup(func() {
			if err := h.servingParks.Close(context.Background()); err != nil {
				t.Error(err)
			}
		})
		// This repeats the real preadmission coordinator path through production
		// binding. The original private no-process record must remain identical.
		key := servingLegacyKey("legacy-stop", servingArtifactIdentity{Host: request.Host, Box: request.Box, Incarnation: request.Incarnation})
		before, err := f.disk.Read(key)
		if err != nil {
			t.Fatal(err)
		}
		coordinator.RetainNoProcess = h.Stop.Coordinator.RetainNoProcess
		if _, err = coordinator.FenceAndStop(t.Context(), request); err != nil {
			t.Fatal(err)
		}
		after, err := f.disk.Read(key)
		if err != nil || !reflect.DeepEqual(before, after) {
			t.Fatal("production retention changed original evidence", err)
		}
		if closed.closed.Load() {
			t.Fatal("private store closed during service")
		}
		if err = h.servingParks.Close(t.Context()); err != nil {
			t.Fatal(err)
		}
	})
	if f == nil || closed == nil || !closed.closed.Load() {
		t.Fatal("production private store leaked")
	}
	bad := &servingCloseStore{Store: f.disk}
	openClaimServingParkAuthority = func() (rootstate.Store, error) { return bad, nil }
	if _, err := withServingParkRoot(ClaimRootHTTP{}, nil); err == nil || !bad.closed.Load() {
		t.Fatal("failed binding leaked private store", err)
	}
}

type servingBorrowedState struct{ rootstate.Store }

func (servingBorrowedState) Close() error { return nil }