package bus

import (
	"encoding/json"
	"fmt"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
)

// novox/hq issue 222: a subscription the bus refused is asked for again. A push sends the machine
// that needs a grant and the bus's machine together; the runtime may subscribe the moment before the
// bus reloads, and a refusal must not leave the module unreachable until some later membership.
func TestARefusedSubscriptionIsAskedForAgain(t *testing.T) {
	url := os.Getenv("MESH_TEST_NATS")
	if url == "" {
		t.Skip("MESH_TEST_NATS unset")
	}
	c, err := Connect(Credential{URL: url, Module: "alpha", Node: "anchor"})
	if err != nil {
		t.Fatal(err)
	}
	defer c.Close()
	var mu sync.Mutex
	var said []string
	c.Logf = func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() }
	c.RetryAfter = []time.Duration{50 * time.Millisecond, 50 * time.Millisecond}

	subject := "mesh.mod.alpha.tool.ping.anchor"
	stop, err := c.answerOn(subject, "", func(json.RawMessage) (any, error) { return "pong", nil })
	if err != nil {
		t.Fatal(err)
	}
	first := c.answering[answeringKey(subject, "")].sub

	// What the bus says when the grant is not there yet.
	c.refused(fmt.Errorf("%w: Permissions Violation for Subscription to %q", nats.ErrPermissionViolation, subject))
	deadline := time.Now().Add(2 * time.Second)
	for {
		c.mu.Lock()
		again := c.answering[answeringKey(subject, "")].sub
		c.mu.Unlock()
		if again != first {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the refused subscription was not asked for again")
		}
		time.Sleep(10 * time.Millisecond)
	}
	asker, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer asker.Close()
	reply, err := asker.Request(subject, []byte("{}"), 2*time.Second)
	if err != nil {
		t.Fatalf("the subject asked for again does not answer: %v", err)
	}
	if !strings.Contains(string(reply.Data), "pong") {
		t.Fatalf("the subject asked for again answered %s", reply.Data)
	}

	// Past its attempts it is given up, and said.
	for i := 0; i < 3; i++ {
		c.refused(fmt.Errorf("%w: Permissions Violation for Subscription to %q", nats.ErrPermissionViolation, subject))
		time.Sleep(80 * time.Millisecond)
	}
	mu.Lock()
	gaveUp := strings.Contains(strings.Join(said, "\n"), "still refuses "+subject)
	mu.Unlock()
	if !gaveUp {
		t.Errorf("never gave up on a subject that stays refused:\n%s", strings.Join(said, "\n"))
	}

	// A subject the runtime stopped answering is not asked for again.
	stop()
	c.refused(fmt.Errorf("%w: Permissions Violation for Subscription to %q", nats.ErrPermissionViolation, subject))
	if _, still := c.answering[answeringKey(subject, "")]; still {
		t.Error("a stopped subject is still tracked")
	}
}

// Issue 1966: a refused publish of an answer (to a caller's inbox) is said, once per subject per minute, never dropped
// in silence while the caller waits out its time.
func TestARefusedAnswerIsSaidOncePerMinute(t *testing.T) {
	c := &Conn{}
	var said []string
	c.Logf = func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
	refuse := func(subject string) bool {
		return c.answerRefused(fmt.Errorf("%w: Permissions Violation for Publish to %q", nats.ErrPermissionViolation, subject))
	}
	// A reply subject ends in a token new to every request: a burst under one prefix is one line.
	if !refuse("_INBOX.user.abc.t1") || !refuse("_INBOX.user.abc.t2") || !refuse("_INBOX.user.def.t1") {
		t.Fatal("a refused answer was not taken for one")
	}
	if len(said) != 2 {
		t.Fatalf("want one line per caller, got %v", said)
	}
	c.inboxSaid["_INBOX.user.abc"] = time.Now().Add(-2 * time.Minute)
	c.inboxSaid["_INBOX.user.old"] = time.Now().Add(-2 * time.Minute)
	refuse("_INBOX.user.abc.t3")
	if len(said) != 3 || !strings.Contains(said[2], "_INBOX.user.abc") {
		t.Fatalf("a minute later it is said again, got %v", said)
	}
	if _, kept := c.inboxSaid["_INBOX.user.old"]; kept || len(c.inboxSaid) != 2 {
		t.Fatalf("entries older than a minute are dropped on write, kept %v", c.inboxSaid)
	}
	if refuse("mesh.mod.x.tool.y") {
		t.Fatal("a refused publish that is no answer was taken for one")
	}
}