//go:build integration

package broker

import (
	"errors"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"golang.org/x/crypto/bcrypt"
)

// 1181/AC3, the release's acceptance test, on a real bus: the account module issue delivers for mesh-mail's
// process "send" does exactly what its bus declares — pull and acknowledge its one work-queue consumer,
// publish its event, read and write its state, and call operator-channel's tell in its own name — and
// nothing it does not: no tool call or seat verb answered, no warrant published. Modelled on
// callers_live_test.go's TestAServerComposedFromTheGrantsRefusesACallNamingAnother.
//
//	go test -tags integration ./internal/broker/ -run TestAProcessAccountOnARealBus
func TestAProcessAccountOnARealBus(t *testing.T) {
	hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
	if err != nil {
		t.Fatal(err)
	}
	controller := Principal{Kind: KindController, PasswordHash: string(hash)}
	messenger := Principal{Kind: KindModule, Node: "novox", Module: "messenger", PasswordHash: string(hash),
		Holds: []Seat{operatorChannel()}}
	process := Principal{Kind: KindProcess, Node: "novox", Module: "mesh-mail", Process: "send",
		PasswordHash: string(hash), Emits: []string{"mail.send.sent", "mail.send.failed"},
		Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}},
		State:  []string{"sends"}, Asks: true, Uses: []Seat{operatorChannel()},
		Invokes: []string{"seat:operator-channel.tell"},
	}
	accounts, err := ComposeAccounts([]Principal{controller, messenger, process})
	if err != nil {
		t.Fatal(err)
	}
	conf := filepath.Join(t.TempDir(), "bus.conf")
	if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+
		`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
		t.Fatal(err)
	}
	opts, err := server.ProcessConfigFile(conf)
	if err != nil {
		t.Fatalf("the composed accounts do not parse: %v", err)
	}
	opts.NoLog, opts.NoSigs = true, true
	s, err := server.NewServer(opts)
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	defer s.Shutdown()

	// The controller sets up the stream, the one consumer, and the process's state bucket — exactly as
	// `module issue` and the controller's raise would, before anything connects as the process.
	admin, err := Dial(s.ClientURL(), nats.UserInfo("controller", "pw"))
	if err != nil {
		t.Fatal(err)
	}
	defer admin.Close()
	mailSends := ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
		MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1}
	if err := admin.EnsureStream(mailSends.Stream()); err != nil {
		t.Fatal(err)
	}
	if err := admin.EnsureConsumer(QueueConsumer(process, process.Queues[0])); err != nil {
		t.Fatal(err)
	}
	if err := admin.EnsureBucket(Bucket{Module: "mesh-mail", Name: "sends"}); err != nil {
		t.Fatal(err)
	}

	// The router's own records bucket operator-channel names (novox/hq ADR 0259 §3): present so the
	// process's direct-get of its own module's asks is a real subject, not merely one it is permitted.
	js, err := jetstream.New(admin.Conn())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: "messenger_asks"}); err != nil {
		t.Fatal(err)
	}

	// messenger, the router, answers operator-channel's tell in the caller's own name.
	holder, err := nats.Connect(s.ClientURL(), nats.UserInfo("novox.messenger", "pw"))
	if err != nil {
		t.Fatal(err)
	}
	defer holder.Close()
	heard := make(chan string, 1)
	if _, err := holder.Subscribe("mesh.seat.operator-channel.tool.>", func(m *nats.Msg) {
		heard <- m.Subject
		_ = m.Respond([]byte(`{"result":{}}`))
	}); err != nil {
		t.Fatal(err)
	}
	_ = holder.Flush()

	refusals := make(chan error, 8)
	sender, err := nats.Connect(s.ClientURL(), nats.UserInfo("novox.mesh-mail.send", "pw"),
		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err }))
	if err != nil {
		t.Fatal(err)
	}
	defer sender.Close()

	// 1. Its event.
	if err := sender.Publish("mesh.mod.mesh-mail.event.mail.send.sent", []byte("{}")); err != nil {
		t.Fatal(err)
	}

	// 2. Its state: writes and reads its own bucket.
	kv, err := jetstream.New(sender)
	if err != nil {
		t.Fatal(err)
	}
	bucket, err := kv.KeyValue(t.Context(), "mesh-mail_sends")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := bucket.Put(t.Context(), "key", []byte("value")); err != nil {
		t.Fatalf("the process may not write its own state: %v", err)
	}
	if got, err := bucket.Get(t.Context(), "key"); err != nil || string(got.Value()) != "value" {
		t.Fatalf("the process may not read its own state back: %v, %v", got, err)
	}

	// 3. Its queue: pulls and acknowledges MAIL_SENDS through its own consumer.
	js2, err := jetstream.New(admin.Conn())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js2.Publish(t.Context(), ModuleStreamSubject("mesh-mail", "mail.send.approved"), []byte("{}")); err != nil {
		t.Fatal(err)
	}
	reading, err := jetstream.New(sender)
	if err != nil {
		t.Fatal(err)
	}
	consumer, err := reading.Consumer(t.Context(), "MAIL_SENDS", QueueConsumer(process, process.Queues[0]).Name)
	if err != nil {
		t.Fatalf("the process may not bind its own queue consumer: %v", err)
	}
	batch, err := consumer.Fetch(1, jetstream.FetchMaxWait(3*time.Second))
	if err != nil {
		t.Fatal(err)
	}
	got := false
	for msg := range batch.Messages() {
		got = true
		if err := msg.Ack(); err != nil {
			t.Fatalf("the process may not acknowledge its own queue: %v", err)
		}
	}
	if !got {
		t.Fatal("the process did not pull the message its own queue holds")
	}

	// 4. Its invoked seat verb, in its own name.
	if _, err := sender.Request("mesh.seat.operator-channel.tool.tell", []byte(`{}`), 3*time.Second); err != nil {
		t.Fatalf("the process may not call operator-channel.tell: %v", err)
	}
	select {
	case s := <-heard:
		if s != "mesh.seat.operator-channel.tool.tell" {
			t.Fatalf("messenger heard %s", s)
		}
	case <-time.After(time.Second):
		t.Fatal("messenger never heard the call")
	}

	// 5. Refused: answering a tool call (it holds no seat, serves no tool).
	if _, err := sender.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil {
		t.Fatal(err)
	}
	_ = sender.Flush()
	// 6. Refused: publishing a warrant (only the holder, messenger, may say operator-channel decided).
	if err := sender.Publish("mesh.seat.operator-channel.event.decided.mesh-mail", []byte("{}")); err != nil {
		t.Fatal(err)
	}
	_ = sender.Flush()
	seen := 0
	timeout := time.After(3 * time.Second)
	for seen < 2 {
		select {
		case err := <-refusals:
			if !errors.Is(err, nats.ErrPermissionViolation) {
				t.Errorf("a refusal was not a permissions violation: %v", err)
			}
			seen++
		case <-timeout:
			t.Fatalf("saw %d of 2 expected permission violations", seen)
		}
	}
}