package console

import (
	"context"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/objstore"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

type callbackRoundTripper func(*http.Request) (*http.Response, error)

func (f callbackRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }

func callbackResponse(code int) *http.Response {
	return &http.Response{StatusCode: code, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("ignored"))}
}

// deliveryArmed is a drain that can actually deliver: a ready route so the
// platform host resolves, a signer so a request may be built at all, and a
// transport the test controls. Anything less and the delivery path is skipped
// rather than exercised, which is how a test of delivery ends up proving
// nothing.
func deliveryArmed(t *testing.T, code string, rt http.RoundTripper) (*consoleFixture, *objstore.MemScanner) {
	t.Helper()
	return deliveryArmedWith(t, code, rt, callbackRoute(fixtureApp+"."+fixtureAcct+".carlosapp.dev"))
}

// callbackRoute is a ready platform route of the fixture app, as the domains
// reconciler leaves it.
func callbackRoute(host string) registry.Route {
	return registry.Route{Host: host, Kind: "instance", Status: "ready", App: fixtureApp, Account: fixtureAcct, Platform: true}
}

// deliveryArmedWith is deliveryArmed with the platform routes chosen by the
// caller, added on top of the fixture's custom kass route.
func deliveryArmedWith(t *testing.T, code string, rt http.RoundTripper, routes ...registry.Route) (*consoleFixture, *objstore.MemScanner) {
	t.Helper()
	fx, sc, _ := drainFixture(t)
	for _, r := range routes {
		fx.addRoute(t, r)
	}
	enableStore(t, fx, fixtureAcct, fixtureApp, true)
	fx.srv.cfg.AppsDomain = "carlosapp.dev"
	fx.srv.cfg.CallbackSigner = callbackTestSigner(t, "master-for-delivery", "")
	fx.srv.cfg.CallbackTransport = rt
	sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-x", time.Now(), 10)
	if _, err := objstore.Enqueue(context.Background(), fx.rel, fx.srv.scanQueue,
		fixtureAcct, fixtureApp,
		&objstore.QueueItem{Key: "uploads/x.pdf", By: "test", CallbackPath: "/hook"},
		time.Now()); err != nil {
		t.Fatal(err)
	}
	return fx, sc
}

// TestDeliveryDoesNotBlockTheTick drives the REAL tick against an endpoint
// that never answers.
//
// The first version of this test called p.enqueue directly and asserted that
// the call returned — which proves the channel has a buffer, and would pass
// just as well if drainScanApp delivered synchronously somewhere else, because
// it never ran a tick at all. The claim in the name is about the TICK, so the
// tick is what has to be timed.
//
// The failure being ruled out is a tenant pointing a callback at a black hole
// and thereby holding up every other tenant's verdict: 50 items times a retry
// schedule inside a pass that sweeps apps one at a time.
func TestDeliveryDoesNotBlockTheTick(t *testing.T) {
	park := make(chan struct{})
	defer close(park)
	reached := make(chan struct{}, 1)
	fx, _ := deliveryArmed(t, "code-deliver-block", callbackRoundTripper(func(*http.Request) (*http.Response, error) {
		select {
		case reached <- struct{}{}:
		default:
		}
		<-park
		return callbackResponse(204), nil
	}))

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv)

	start := time.Now()
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	if took := time.Since(start); took > 3*time.Second {
		t.Fatalf("the tick took %v with one callback endpoint hanging — delivery is in the drain's path", took)
	}
	// And the delivery really was attempted, or the timing above measured a
	// path that never made a request.
	select {
	case <-reached:
	case <-time.After(5 * time.Second):
		t.Fatal("no callback was attempted, so this test timed nothing")
	}
}

// TestCallbackGoesToPlatformHostOnly asserts where the DELIVERY actually goes.
//
// The first version called callbackTarget and checked its return, which is
// already covered by TestCallbackTargetIsBuiltStructurally in scanhook_test.go
// — it re-tested the helper rather than the caller. What matters here is that
// the pool builds its URL from the app's platform-assigned name and nothing
// from the queue item, so the request itself is what gets inspected.
func TestCallbackGoesToPlatformHostOnly(t *testing.T) {
	got := make(chan string, 4)
	fx, _ := deliveryArmed(t, "code-deliver-host", callbackRoundTripper(func(r *http.Request) (*http.Response, error) {
		got <- r.URL.String()
		return callbackResponse(204), nil
	}))

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv)
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	select {
	case u := <-got:
		want := "https://" + fixtureApp + "." + fixtureAcct + ".carlosapp.dev/hook"
		if u != want {
			t.Fatalf("posted to %q, want %q", u, want)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("no callback was delivered")
	}
}

func TestCallbackRefusesRedirects(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	p := newScanDeliveryPool(ctx, &Server{})
	defer func() { cancel(); p.stop() }()
	req, _ := http.NewRequest(http.MethodPost, "https://app.example/hook", nil)
	if err := p.client.CheckRedirect(req, nil); !errors.Is(err, http.ErrUseLastResponse) {
		t.Fatalf("redirect error = %v", err)
	}
}

func TestTransportDoesNotSkipTLSVerification(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	p := newScanDeliveryPool(ctx, &Server{})
	cancel()
	p.stop()
	tr, ok := p.client.Transport.(*http.Transport)
	if !ok {
		t.Fatalf("production transport = %T", p.client.Transport)
	}
	if tr.TLSClientConfig != nil && tr.TLSClientConfig.InsecureSkipVerify {
		t.Fatal("production callback transport disables TLS verification")
	}
}

// TestNeverPostsUnsigned. The first version used a bare &Server{}, which has
// no AppsDomain — so delivery returned at "no platform host" long before the
// signer mattered, and the test passed with the signer guard deleted. Worse,
// with the guard gone AND a real host it would nil-deref and panic a worker.
//
// The fixture therefore has a resolvable host and a real queued item, exactly
// as a delivering deployment does, and only the signer is missing.
func TestNeverPostsUnsigned(t *testing.T) {
	var calls atomic.Int32
	fx, _ := deliveryArmed(t, "code-deliver-unsigned", callbackRoundTripper(func(*http.Request) (*http.Response, error) {
		calls.Add(1)
		return callbackResponse(204), nil
	}))
	fx.srv.cfg.CallbackSigner = nil // the one thing under test

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv)
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	// Give any wrongly-made request time to arrive before concluding none did.
	time.Sleep(300 * time.Millisecond)
	if n := calls.Load(); n != 0 {
		t.Fatalf("made %d request(s) with no signer — an unsigned callback is the hole §6.3 exists to close", n)
	}
}

// TestDeliveredSecretVerifiesTheHeaderItReceives is the round trip nothing
// covered, and the one an app author's first hour depends on.
//
// The secret handed to an app is HEX (EnvironmentFile values are text), while
// the MAC is keyed on the RAW 32 bytes. An author who feeds the environment
// string straight into HMAC gets a failure on every callback, and the natural
// next move is to stop verifying. So this asserts the whole path: the key as
// delivered, hex-decoded, verifies the header the pool actually sent.
func TestDeliveredSecretVerifiesTheHeaderItReceives(t *testing.T) {
	type sent struct {
		header string
		body   []byte
	}
	got := make(chan sent, 4)
	fx, _ := deliveryArmed(t, "code-deliver-roundtrip", callbackRoundTripper(func(r *http.Request) (*http.Response, error) {
		body, _ := io.ReadAll(r.Body)
		got <- sent{header: r.Header.Get("Carlos-Signature"), body: body}
		return callbackResponse(204), nil
	}))

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv)
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	select {
	case s := <-got:
		if s.header == "" {
			t.Fatal("no Carlos-Signature header")
		}
		// Exactly what the app is given, and exactly what it must do with it.
		delivered := fx.srv.cfg.CallbackSigner.DerivedKey(fixtureAcct, fixtureApp, 0)
		key, err := hex.DecodeString(string(delivered))
		if err != nil {
			t.Fatalf("the delivered secret is not hex: %v", err)
		}
		if err := VerifyCallback(key, s.header, s.body, time.Now()); err != nil {
			t.Fatalf("the delivered secret does not verify the header we sent: %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("no callback was delivered")
	}
}

func TestCallbackRetriesAreBounded(t *testing.T) {
	old := scanDeliveryRetryDelays
	scanDeliveryRetryDelays = []time.Duration{0, 0, 0}
	defer func() { scanDeliveryRetryDelays = old }()
	var calls atomic.Int32
	signer := callbackTestSigner(t, "master", "")
	p := &scanDeliveryPool{s: &Server{cfg: Config{CallbackSigner: signer}}, ctx: context.Background(), client: &http.Client{
		Transport: callbackRoundTripper(func(*http.Request) (*http.Response, error) {
			calls.Add(1)
			return callbackResponse(500), nil
		}),
	}}
	job := scanDeliveryJob{ref: objstore.AppRef{Account: "bab", App: "app"}}
	body := []byte(`{"verdict":"clean"}`)
	for attempt := 0; ; attempt++ {
		retry, _ := p.post(context.Background(), job, "https://app.bab.example/hook", body, 1)
		if !retry || attempt == len(scanDeliveryRetryDelays) {
			break
		}
	}
	if calls.Load() != 4 {
		t.Fatalf("attempts = %d, want initial plus three retries", calls.Load())
	}
}

func TestDroppedCallbackStillReadableFromStatusDoor(t *testing.T) {
	ctx := context.Background()
	store := release.NewDirStore(t.TempDir())
	want := &objstore.ScanResult{Key: "uploads/x", Version: "opaque", Verdict: string(objstore.VerdictInfected)}
	if err := objstore.WriteScanResult(ctx, store, "bab", "app", want); err != nil {
		t.Fatal(err)
	}
	// A full queue takes the documented drop branch.
	p := &scanDeliveryPool{ctx: ctx, jobs: make(chan scanDeliveryJob, 1)}
	p.jobs <- scanDeliveryJob{}
	p.enqueue(scanDeliveryJob{ref: objstore.AppRef{Account: "bab", App: "app"}, path: "/hook"})
	got, err := objstore.ReadScanResult(ctx, store, "bab", "app", "uploads/x", "opaque")
	if err != nil || got == nil || got.Verdict != string(objstore.VerdictInfected) {
		t.Fatalf("durable status after callback drop = %+v, err = %v", got, err)
	}
}

// TestCallbackBodyCarriesRecorded inspects the body the pool BUILDS, not one
// the test marshals itself. The first version constructed a literal and
// checked its JSON, which passes just as well against a deliver() that
// hard-codes Recorded: true — and Recorded is the field that tells an app
// whether an infected verdict is actually backed by the enforcement deny.
func TestCallbackBodyCarriesRecorded(t *testing.T) {
	got := make(chan map[string]any, 4)
	fx, sc := deliveryArmed(t, "code-deliver-recorded", callbackRoundTripper(func(r *http.Request) (*http.Response, error) {
		raw, _ := io.ReadAll(r.Body)
		var body map[string]any
		_ = json.Unmarshal(raw, &body)
		got <- body
		return callbackResponse(204), nil
	}))
	// The ten-tag case: a verdict is reached and cannot be written onto the
	// object, so the app must be told the deny is not in force.
	sc.Unrecordable("uploads/x.pdf")

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv)
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}
	select {
	case body := <-got:
		if rec, ok := body["recorded"].(bool); !ok || rec {
			t.Fatalf("recorded = %v, want false — the verdict is not on the object, so no deny matches it", body["recorded"])
		}
		if _, ok := body["version"]; ok {
			t.Errorf("body carries the opaque version token: %v", body)
		}
		if body["etag"] == "" || body["etag"] == nil {
			t.Errorf("body carries no etag, which is what an app matches on: %v", body)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("no callback was delivered")
	}
}

func TestCallbackBodyOmitsTheVersionToken(t *testing.T) {
	b, err := json.Marshal(scanCallbackBody{Key: "x", ETag: "etag"})
	if err != nil {
		t.Fatal(err)
	}
	if strings.Contains(string(b), "version") {
		t.Fatalf("body exposes version token: %s", b)
	}
}

func TestNoPlatformHostIsNotAnError(t *testing.T) {
	host, err := (&Server{}).scanCallbackHost(objstore.AppRef{Account: "bab", App: "app"})
	if err != nil || host != "" {
		t.Fatalf("host = %q, err = %v", host, err)
	}
}

type callbackHostCase struct {
	name   string
	routes []registry.Route
	want   string
}

// callbackHostCases is shared by the helper test and the delivered POST so
// both are held to the same routes. The fixture already holds a ready custom
// kass route (kass.carlosapp.dev, not Platform), which must never win.
func callbackHostCases() []callbackHostCase {
	const canon = fixtureApp + "." + fixtureAcct + ".carlosapp.dev"
	with := func(mod func(*registry.Route)) registry.Route {
		r := callbackRoute(canon)
		mod(&r)
		return r
	}
	return []callbackHostCase{
		{"canonical wins over an alphabetically earlier alternate",
			[]registry.Route{callbackRoute("a." + canon), callbackRoute(canon)}, canon},
		{"only an alternate platform apex uses its actual hostname",
			[]registry.Route{callbackRoute("kass.alt.example.test")}, "kass.alt.example.test"},
		{"alternates resolve lexicographically",
			[]registry.Route{callbackRoute("kass.zeta.example.test"), callbackRoute("kass.beta.example.test")}, "kass.beta.example.test"},
		{"custom-only has no host", nil, ""},
		{"wrong account cannot win",
			[]registry.Route{with(func(r *registry.Route) { r.Account = "zzzz" })}, ""},
		{"wrong app cannot win",
			[]registry.Route{with(func(r *registry.Route) { r.App = "other" })}, ""},
		{"pending cannot win",
			[]registry.Route{with(func(r *registry.Route) { r.Status = "pending" })}, ""},
		{"canary cannot win",
			[]registry.Route{with(func(r *registry.Route) { r.Canary = true })}, ""},
		{"sidecar cannot win",
			[]registry.Route{with(func(r *registry.Route) { r.Kind = "sidecar" })}, ""},
		{"ineligible canonical yields to an eligible alternate",
			[]registry.Route{with(func(r *registry.Route) { r.Canary = true }), callbackRoute("kass.alt.example.test")}, "kass.alt.example.test"},
		{"canonical-looking custom route is not platform ownership",
			[]registry.Route{with(func(r *registry.Route) { r.Platform = false })}, ""},
	}
}

func TestCallbackHostSelection(t *testing.T) {
	for _, tc := range callbackHostCases() {
		t.Run(tc.name, func(t *testing.T) {
			fx := newConsoleFixture(t)
			fx.srv.cfg.AppsDomain = "carlosapp.dev"
			for _, r := range tc.routes {
				fx.addRoute(t, r)
			}
			host, err := fx.srv.scanCallbackHost(objstore.AppRef{Account: fixtureAcct, App: fixtureApp})
			if err != nil || host != tc.want {
				t.Fatalf("host = %q, err = %v, want %q", host, err, tc.want)
			}
		})
	}
}

func TestCallbackPostsToSelectedHost(t *testing.T) {
	for _, tc := range callbackHostCases() {
		t.Run(tc.name, func(t *testing.T) {
			got := make(chan string, 4)
			fx, _ := deliveryArmedWith(t, "code-deliver-select", callbackRoundTripper(func(r *http.Request) (*http.Response, error) {
				got <- r.URL.String()
				return callbackResponse(204), nil
			}), tc.routes...)
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv)
			if err := fx.srv.scanDrainTick(ctx); err != nil {
				t.Fatal(err)
			}
			if tc.want == "" {
				select {
				case u := <-got:
					t.Fatalf("posted to %q with no eligible platform route", u)
				case <-time.After(500 * time.Millisecond):
				}
				return
			}
			select {
			case u := <-got:
				if want := "https://" + tc.want + "/hook"; u != want {
					t.Fatalf("posted to %q, want %q", u, want)
				}
			case <-time.After(5 * time.Second):
				t.Fatal("no callback was delivered")
			}
		})
	}
}