package console

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

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

// statusArmed is enqueueArmed plus a handle on the engine, because every test
// here has to stage what the object's tag says.
func statusArmed(t *testing.T, code string) (*consoleFixture, *objstore.MemScanner, string, string) {
	t.Helper()
	fx, _, token, _ := storeArmed(t, code)
	grantStore(t, fx, token)
	requestScanning(t, fx, token, false)
	sc := objstore.NewMemScanner()
	fx.srv.cfg.StoreScanner = sc
	enableDeploymentScanning(t, fx)
	fx.srv.cfg.StoreOperators = []string{"alice@keymail.dev"}
	if s, o := postJSON(t, scanURL(fx, "titogo")+"/grant", map[string]any{}, token); s != http.StatusOK {
		t.Fatalf("scan grant = %d %v", s, o)
	}
	rec, _, err := objstore.Get(context.Background(), fx.rel, release.OpsAccount, "titogo")
	if err != nil || rec == nil || rec.Provisioned == nil {
		t.Fatalf("no provisioned record to read a bucket from: %v", err)
	}
	return fx, sc, appToken(t, fx, "titobot"), rec.Provisioned.Bucket
}

// getStatus asks the door about one key and decodes the answer.
func getStatus(t *testing.T, fx *consoleFixture, tok, query string) (int, map[string]any) {
	t.Helper()
	status, _, body := rawReq(t, http.MethodGet, enqueueURL(fx, "titogo")+"?"+query, tok, "")
	out := map[string]any{}
	if body != "" {
		if err := json.Unmarshal([]byte(body), &out); err != nil {
			t.Fatalf("decode %q: %v", body, err)
		}
	}
	return status, out
}

// getStatusSafe is getStatus for a NON-TEST goroutine. rawReq calls t.Fatal on
// a transport error, and FailNow from any goroutine but the test's own is
// undefined behaviour that go vet's testinggoroutine check exists to catch.
func getStatusSafe(fx *consoleFixture, tok, query string) (map[string]any, error) {
	req, err := http.NewRequest(http.MethodGet, enqueueURL(fx, "titogo")+"?"+query, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("Authorization", "Bearer "+tok)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	raw, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	out := map[string]any{}
	if len(raw) > 0 {
		if err := json.Unmarshal(raw, &out); err != nil {
			return nil, fmt.Errorf("decode %q: %w", raw, err)
		}
	}
	return out, nil
}

// seedObject puts an object in the engine's world and returns its version, so
// a test can stage a tag or a result against the same bytes.
func seedObject(sc *objstore.MemScanner, bucket, key, etag string) string {
	mod := time.Now().Truncate(time.Second)
	sc.PutObject(bucket, key, etag, mod, 10)
	return objstore.ScanObject{Key: key, ETag: etag, Modified: mod, Size: 10}.Version()
}

// TestStatusDoorReadsTagThenResultObject pins the ORDER, which is the whole of
// §5.2's read rule: the object's own tag is the more recent and more
// authoritative answer, and a result object exists only because a tag could
// not be written.
func TestStatusDoorReadsTagThenResultObject(t *testing.T) {
	ctx := context.Background()
	fx, sc, tok, bucket := statusArmed(t, "code-status-order")

	// (a) a tag and nothing else.
	vTagged := seedObject(sc, bucket, "uploads/tagged.pdf", "etag-t")
	sc.Put(bucket, "uploads/tagged.pdf", vTagged, objstore.VerdictClean)
	if status, out := getStatus(t, fx, tok, "key=uploads/tagged.pdf"); status != http.StatusOK || out["state"] != "clean" {
		t.Fatalf("tagged = %d %v, want 200 clean", status, out)
	}

	// (b) a result object and no tag.
	vResult := seedObject(sc, bucket, "uploads/untaggable.pdf", "etag-u")
	if err := objstore.WriteScanResult(ctx, fx.rel, release.OpsAccount, "titogo", &objstore.ScanResult{
		Key: "uploads/untaggable.pdf", Verdict: "infected", Version: vResult,
		Engine: "clamav", Signature: "Eicar-Test", At: time.Now().UTC(),
	}); err != nil {
		t.Fatal(err)
	}
	status, out := getStatus(t, fx, tok, "key=uploads/untaggable.pdf")
	if status != http.StatusOK || out["state"] != "infected" {
		t.Fatalf("result-object read = %d %v, want 200 infected", status, out)
	}
	if out["engine"] != "clamav" || out["signature"] != "Eicar-Test" {
		t.Errorf("provenance missing from a result-object answer: %v", out)
	}
	// The app must be told the verdict is NOT on the object: an infected
	// object answered this way is not covered by the enforcement Deny, which
	// matches on a tag that is not there.
	if out["recorded"] != false {
		t.Errorf("recorded = %#v, want JSON false — the app cannot tell this verdict is not enforceable", out["recorded"])
	}

	// (c) BOTH, disagreeing. The tag wins.
	vBoth := seedObject(sc, bucket, "uploads/both.pdf", "etag-b")
	sc.Put(bucket, "uploads/both.pdf", vBoth, objstore.VerdictClean)
	if err := objstore.WriteScanResult(ctx, fx.rel, release.OpsAccount, "titogo", &objstore.ScanResult{
		Key: "uploads/both.pdf", Verdict: "infected", Version: vBoth,
		Engine: "clamav", At: time.Now().UTC(),
	}); err != nil {
		t.Fatal(err)
	}
	if status, out := getStatus(t, fx, tok, "key=uploads/both.pdf"); out["state"] != "clean" {
		t.Fatalf("with both present = %d %v, want the TAG's answer (clean)", status, out)
	}
}

// TestStatusDoorRefusesStaleResultObject is the laundering bug as a
// correctness bug: v1 was judged, v2 was uploaded over it, and the recorded
// answer is about bytes that are no longer there.
func TestStatusDoorRefusesStaleResultObject(t *testing.T) {
	ctx := context.Background()
	fx, sc, tok, bucket := statusArmed(t, "code-status-stale")

	if err := objstore.WriteScanResult(ctx, fx.rel, release.OpsAccount, "titogo", &objstore.ScanResult{
		Key: "uploads/x.pdf", Verdict: "clean", Version: "etag-v1@2020-01-01T00:00:00Z",
		Engine: "clamav", At: time.Now().UTC(),
	}); err != nil {
		t.Fatal(err)
	}
	// The object now holds DIFFERENT bytes.
	seedObject(sc, bucket, "uploads/x.pdf", "etag-v2")

	status, out := getStatus(t, fx, tok, "key=uploads/x.pdf")
	if status != http.StatusOK {
		t.Fatalf("status = %d %v", status, out)
	}
	if out["state"] != "pending" {
		t.Fatalf("state = %v, want pending — a verdict about replaced bytes is not an answer about this object", out["state"])
	}
}

// TestStatusDoorReportsFailedAndDenied: as themselves, never as pending.
//
// The landed TAG READER maps both back to unknown so the sweep retries them,
// which is right as retry policy and wrong as an answer to a question. If this
// door applied that rule, a failed scan would read as pending forever — the
// drain deletes the item as handled, and nothing changes the answer until a
// sweep happens to re-fail it.
func TestStatusDoorReportsFailedAndDenied(t *testing.T) {
	fx, sc, tok, bucket := statusArmed(t, "code-status-fd")

	v := seedObject(sc, bucket, "uploads/failed.pdf", "etag-f")
	sc.Put(bucket, "uploads/failed.pdf", v, objstore.VerdictFailed)
	if _, out := getStatus(t, fx, tok, "key=uploads/failed.pdf"); out["state"] != "failed" {
		t.Errorf("state = %v, want failed", out["state"])
	}

	seedObject(sc, bucket, "uploads/denied.pdf", "etag-d")
	sc.DenyHead(bucket, "uploads/denied.pdf")
	if _, out := getStatus(t, fx, tok, "key=uploads/denied.pdf"); out["state"] != "denied" {
		t.Errorf("state = %v, want denied", out["state"])
	}

	// And skipped is not clean. An app treating it as success reproduces the
	// dishonesty the two-number rule exists to prevent.
	vs := seedObject(sc, bucket, "uploads/huge.bin", "etag-h")
	sc.Put(bucket, "uploads/huge.bin", vs, objstore.VerdictSkipped)
	if _, out := getStatus(t, fx, tok, "key=uploads/huge.bin"); out["state"] != "skipped" {
		t.Errorf("state = %v, want skipped", out["state"])
	}
}

// TestStatusDoorGoneForMissingObject: an object that is not there is `gone`,
// which is distinct from both `pending` and `clean`.
func TestStatusDoorGoneForMissingObject(t *testing.T) {
	fx, _, tok, _ := statusArmed(t, "code-status-gone")

	status, out := getStatus(t, fx, tok, "key=uploads/never-existed.pdf")
	if status != http.StatusOK || out["state"] != "gone" {
		t.Fatalf("missing object = %d %v, want 200 gone", status, out)
	}
}

// TestStatusDoorPendingForAnUnjudgedObject is the positive control for
// everything above: without it, a door that answered `pending` to absolutely
// everything would pass the stale test and the long-poll tests.
func TestStatusDoorPendingForAnUnjudgedObject(t *testing.T) {
	fx, sc, tok, bucket := statusArmed(t, "code-status-pending")
	seedObject(sc, bucket, "uploads/new.pdf", "etag-n")

	if _, out := getStatus(t, fx, tok, "key=uploads/new.pdf"); out["state"] != "pending" {
		t.Fatalf("state = %v, want pending", out["state"])
	}
}

// TestStatusDoorSharesTheConsentChain: the same table the enqueue door reads,
// so the two doors cannot answer one refusal differently — a disagreement is
// itself something an app could conclude from.
func TestStatusDoorSharesTheConsentChain(t *testing.T) {
	fx, _, member, _ := storeArmed(t, "code-status-consent")
	grantStore(t, fx, member)
	fx.srv.cfg.StoreScanner = objstore.NewMemScanner()
	enableDeploymentScanning(t, fx)
	tok := appToken(t, fx, "bot")

	// ARMED TO FAIL ON ANY READ. MemScanner counts Scans, not Status calls, so
	// "no bytes were read" cannot be asserted by counting — but an engine that
	// errors on Status turns any read into a 502, which a 409 proves did not
	// happen. That is what makes this a test of ORDER and not just of wording.
	sc := fx.srv.cfg.StoreScanner.(*objstore.MemScanner)
	sc.Fail("Status", errors.New("the engine must not have been consulted"))

	status, out := getStatus(t, fx, tok, "key=uploads/x.pdf")
	if status != http.StatusConflict || out["reason"] != "not_requested" {
		t.Fatalf("status door = %d %v, want the enqueue door's 409 not_requested", status, out)
	}
	if out["state"] != "unavailable" {
		t.Errorf("state = %v, want unavailable", out["state"])
	}

	// And the reversible refusal, which is a different status code — one row
	// proved nothing about the mapping being shared, only about one entry.
	requestScanning(t, fx, member, false)
	fx.srv.cfg.StoreOperators = []string{"alice@keymail.dev"}
	if s, o := postJSON(t, scanURL(fx, "titogo")+"/grant", map[string]any{}, member); s != http.StatusOK {
		t.Fatalf("scan grant = %d %v", s, o)
	}
	setDeploymentScanning(t, fx, false)
	status, out = getStatus(t, fx, tok, "key=uploads/x.pdf")
	if status != http.StatusServiceUnavailable || out["reason"] != "disabled" {
		t.Fatalf("disabled = %d %v, want 503 disabled", status, out)
	}
}

// TestStatusDoorNeedsAKey and rejects a malformed wait.
func TestStatusDoorArgumentErrors(t *testing.T) {
	fx, _, tok, _ := statusArmed(t, "code-status-args")

	if status, _ := getStatus(t, fx, tok, ""); status != http.StatusBadRequest {
		t.Errorf("no key = %d, want 400", status)
	}
	if status, _ := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=soon"); status != http.StatusBadRequest {
		t.Errorf("malformed wait = %d, want 400", status)
	}
	if status, _ := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=-5s"); status != http.StatusBadRequest {
		t.Errorf("negative wait = %d, want 400", status)
	}
}

// TestScanWaitForClamps: an over-long wait is the client asking for the
// longest it can have, not an error to special-case.
func TestScanWaitForClamps(t *testing.T) {
	for _, tc := range []struct {
		in   string
		want time.Duration
		bad  bool
	}{
		{in: "", want: 0},
		{in: "5s", want: 5 * time.Second},
		{in: "30s", want: scanWaitMax},
		{in: "5m", want: scanWaitMax},
		{in: "24h", want: scanWaitMax},
		{in: "soon", bad: true},
		{in: "-1s", bad: true},
	} {
		got, err := scanWaitFor(tc.in)
		if tc.bad {
			if err == nil {
				t.Errorf("scanWaitFor(%q) = %v, want an error", tc.in, got)
			}
			continue
		}
		if err != nil {
			t.Errorf("scanWaitFor(%q): %v", tc.in, err)
			continue
		}
		if got != tc.want {
			t.Errorf("scanWaitFor(%q) = %v, want %v", tc.in, got, tc.want)
		}
	}
}

// TestLongPollWakesOnDrain: a request parked on `wait` returns as soon as the
// drain records a verdict, rather than sitting out its window.
//
// The wait is long and the assertion is that the answer arrives EARLY, which
// is the only formulation that fails if the wake is never delivered — a test
// that merely checked the final answer would pass against a door that always
// slept the full duration.
func TestLongPollWakesOnDrain(t *testing.T) {
	ctx := context.Background()
	fx, sc, tok, bucket := statusArmed(t, "code-status-poll")
	seedObject(sc, bucket, "uploads/x.pdf", "etag-x")
	if _, err := objstore.Enqueue(ctx, fx.rel, fx.srv.scanQueue, release.OpsAccount, "titogo",
		&objstore.QueueItem{Key: "uploads/x.pdf", By: "test"}, time.Now()); err != nil {
		t.Fatal(err)
	}

	type reply struct {
		out   map[string]any
		took  time.Duration
		state string
	}
	done := make(chan reply, 1)
	errs := make(chan error, 1)
	go func() {
		start := time.Now()
		out, err := getStatusSafe(fx, tok, "key=uploads/x.pdf&wait=25s")
		if err != nil {
			errs <- err
			return
		}
		st, _ := out["state"].(string)
		done <- reply{out: out, took: time.Since(start), state: st}
	}()

	// Give the request time to park before the drain runs, so the wake is
	// what returns it rather than the first read.
	time.Sleep(150 * time.Millisecond)
	if err := fx.srv.scanDrainTick(ctx); err != nil {
		t.Fatal(err)
	}

	select {
	case got := <-done:
		if got.state != "clean" {
			t.Fatalf("state = %v (%v), want clean", got.state, got.out)
		}
		if got.took > 20*time.Second {
			t.Errorf("took %v against a 25s wait — it sat out the window rather than being woken", got.took)
		}
	case err := <-errs:
		t.Fatalf("long-poll request: %v", err)
	case <-time.After(20 * time.Second):
		t.Fatal("the parked request was never woken by the drain")
	}
}

// TestLongPollReturnsPendingWhenNothingHappens: the window is a ceiling, not a
// promise, and it must end.
func TestLongPollReturnsPendingWhenNothingHappens(t *testing.T) {
	fx, sc, tok, bucket := statusArmed(t, "code-status-timeout")
	seedObject(sc, bucket, "uploads/quiet.pdf", "etag-q")

	start := time.Now()
	status, out := getStatus(t, fx, tok, "key=uploads/quiet.pdf&wait=300ms")
	if status != http.StatusOK || out["state"] != "pending" {
		t.Fatalf("quiet wait = %d %v, want 200 pending", status, out)
	}
	if took := time.Since(start); took < 250*time.Millisecond {
		t.Errorf("returned after %v — it did not wait the window it was given", took)
	}
}

// TestLongPollCapsWaiters covers BOTH caps, and the per-app one is not implied
// by the global one: without it a single tenant fills the global budget and
// every other tenant's wait degrades to polling.
func TestLongPollCapsWaiters(t *testing.T) {
	fx, _, _, _ := statusArmed(t, "code-status-caps")

	// Per app: one app's own keys, up to its cap and then no further.
	ref := objstore.AppRef{Account: release.OpsAccount, App: "titogo"}
	var release1 []func()
	for i := 0; i < scanWaitersPerApp; i++ {
		_, rel, ok := fx.srv.scanWait(ref, string(rune('a'+i%26))+"/x")
		if !ok {
			t.Fatalf("refused waiter %d, below the per-app cap of %d", i, scanWaitersPerApp)
		}
		release1 = append(release1, rel)
	}
	if _, _, ok := fx.srv.scanWait(ref, "one/too/many"); ok {
		t.Errorf("parked a waiter past the per-app cap of %d", scanWaitersPerApp)
	}
	// A DIFFERENT app is unaffected — which is the whole point of the per-app
	// cap existing beside the global one.
	other := objstore.AppRef{Account: release.OpsAccount, App: "otherapp"}
	if _, rel, ok := fx.srv.scanWait(other, "x"); !ok {
		t.Errorf("one app at its cap blocked another app entirely")
	} else {
		rel()
	}
	for _, rel := range release1 {
		rel()
	}
	// Releasing gives the capacity back rather than burning it.
	if _, rel, ok := fx.srv.scanWait(ref, "after/release"); !ok {
		t.Error("capacity was not returned when the waiters were released")
	} else {
		rel()
	}

	// Global: enough distinct apps that no per-app cap is ever reached.
	var release2 []func()
	for i := 0; ; i++ {
		app := "app" + string(rune('A'+i%26)) + string(rune('A'+(i/26)%26))
		r := objstore.AppRef{Account: release.OpsAccount, App: app}
		ch, rel, ok := fx.srv.scanWait(r, "k")
		if !ok {
			break
		}
		_ = ch
		release2 = append(release2, rel)
		if i > scanWaitersGlobal+8 {
			t.Fatalf("parked %d waiters with a global cap of %d", i, scanWaitersGlobal)
		}
	}
	if len(release2) != scanWaitersGlobal {
		t.Errorf("global cap admitted %d waiters, want %d", len(release2), scanWaitersGlobal)
	}
	for _, rel := range release2 {
		rel()
	}
}

// TestReleaseIsIdempotent: the door defers the release and a caller may call it
// early. A second call that decremented again would leak capacity downwards
// until the console stopped parking anything at all.
func TestReleaseIsIdempotent(t *testing.T) {
	fx, _, _, _ := statusArmed(t, "code-status-idem")
	ref := objstore.AppRef{Account: release.OpsAccount, App: "titogo"}

	_, rel, ok := fx.srv.scanWait(ref, "k")
	if !ok {
		t.Fatal("idle console refused a waiter")
	}
	rel()
	rel()
	rel()

	fx.srv.scanWaits.mu.Lock()
	total := fx.srv.scanWaits.total
	fx.srv.scanWaits.mu.Unlock()
	if total != 0 {
		t.Errorf("total = %d after releasing one waiter three times, want 0", total)
	}
}

// TestStatusDoorAnswersWhenTheWaitCapIsFull: over the cap nothing is refused.
// The read still happens and still answers truthfully; only the waiting is
// skipped, so a full console degrades to polling rather than to errors.
func TestStatusDoorAnswersWhenTheWaitCapIsFull(t *testing.T) {
	fx, sc, tok, bucket := statusArmed(t, "code-status-full")
	// PENDING, not clean. The first version staged a CLEAN object, so the door
	// answered on its first read and returned before the wait mattered at all
	// — it passed against a door that ignored the caps entirely AND against
	// one that parked forever when refused. The object has to be one the door
	// would WANT to wait for.
	seedObject(sc, bucket, "uploads/x.pdf", "etag-x")

	ref := objstore.AppRef{Account: release.OpsAccount, App: "titogo"}
	var rels []func()
	for i := 0; i < scanWaitersPerApp; i++ {
		if _, rel, ok := fx.srv.scanWait(ref, "filler/"+string(rune('a'+i%26))); ok {
			rels = append(rels, rel)
		}
	}
	defer func() {
		for _, rel := range rels {
			rel()
		}
	}()

	start := time.Now()
	status, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=10s")
	if status != http.StatusOK || out["state"] != "pending" {
		t.Fatalf("over the cap = %d %v, want the truthful 200 pending", status, out)
	}
	if took := time.Since(start); took > 3*time.Second {
		t.Fatalf("took %v against a 10s wait — a request that cannot park must answer at once, not park anyway", took)
	}

	// THE CONTROL, without which "answered immediately" proves nothing: under
	// the cap, the same request on the same pending object waits.
	for _, rel := range rels {
		rel()
	}
	rels = nil
	start = time.Now()
	if _, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=400ms"); out["state"] != "pending" {
		t.Fatalf("under the cap = %v, want pending", out)
	}
	if took := time.Since(start); took < 350*time.Millisecond {
		t.Errorf("under the cap it returned after %v — it did not park, so the test above says nothing", took)
	}
}

// TestLongPollSurvivesADrainDuringItsFirstRead runs a whole drain INSIDE the
// door's first Status call and requires a prompt, correct answer.
//
// WHAT IT DOES NOT PROVE, stated because the obvious reading of it is wrong: it
// does not pin the register-before-read ORDER. That was the intent, and a
// mutation that moves registration after the first read SURVIVES it. The
// reason is that the hook fires at the top of Status, so by the time the read
// returns the drain has already tagged the object and the door answers "clean"
// from the read itself — in both orderings, without the wake mattering at all.
//
// The ordering therefore remains unasserted, and the comment on the door says
// so. What this does pin is that a drain landing mid-read cannot make the door
// lose the answer or hang, which is worth having on its own.
func TestLongPollSurvivesADrainDuringItsFirstRead(t *testing.T) {
	ctx := context.Background()
	fx, sc, tok, bucket := statusArmed(t, "code-status-order2")
	seedObject(sc, bucket, "uploads/x.pdf", "etag-x")
	if _, err := objstore.Enqueue(ctx, fx.rel, fx.srv.scanQueue, release.OpsAccount, "titogo",
		&objstore.QueueItem{Key: "uploads/x.pdf", By: "test"}, time.Now()); err != nil {
		t.Fatal(err)
	}

	// FIRE ONCE, and with a compare-and-swap rather than sync.Once: the drain
	// this hook starts calls Status itself, so the hook re-enters — and
	// Once.Do is not reentrant, it DEADLOCKS the second caller against the
	// first. A CAS lets the inner call see "already fired" and return.
	var fired atomic.Bool
	sc.OnStatus(func() {
		if !fired.CompareAndSwap(false, true) {
			return
		}
		if err := fx.srv.scanDrainTick(ctx); err != nil {
			t.Errorf("drain inside the door's read: %v", err)
		}
	})

	start := time.Now()
	status, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=20s")
	took := time.Since(start)
	if status != http.StatusOK {
		t.Fatalf("status = %d %v", status, out)
	}
	if out["state"] != "clean" {
		t.Fatalf("state = %v, want clean — the drain ran during the first read", out["state"])
	}
	if took > 15*time.Second {
		t.Errorf("took %v against a 20s wait — a drain during the read must not cost the whole window", took)
	}
}