package bus_test

import (
	"encoding/json"
	"errors"
	"strings"
	"sync"
	"testing"
	"time"

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

	"github.com/novox/mesh-tools/node-tools/internal/bus"
	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)

// The seats of novox/hq ADR 0259 §3 as the controller issues them: an asker, the router, and a channel of
// kind telegram, each with the seat traffic its membership lists.
func seatMesh(t *testing.T) (*mt.Mesh, *bus.Conn, nats.JetStreamContext) {
	t.Helper()
	mesh := mt.New(t)
	nc, err := nats.Connect(mt.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, _ := nc.JetStream()
	for _, s := range []*nats.StreamConfig{
		{Name: "SEAT_OPERATOR_CHANNEL", Subjects: []string{"mesh.seat.operator-channel.accept.>"}, Retention: nats.WorkQueuePolicy},
		{Name: "SEAT_CHANNEL", Subjects: []string{"mesh.seat.channel.accept.>"}, Retention: nats.WorkQueuePolicy},
		{Name: "SEAT_EVENTS", Subjects: []string{"mesh.seat.*.event.>"}},
	} {
		if _, err := js.AddStream(s); err != nil {
			t.Fatal(err)
		}
	}
	for stream, c := range map[string]*nats.ConsumerConfig{
		"SEAT_OPERATOR_CHANNEL": {Durable: "SEAT_OPERATOR_CHANNEL_worker", AckPolicy: nats.AckExplicitPolicy,
			FilterSubject: "mesh.seat.operator-channel.accept.>", AckWait: 2 * time.Second},
		"SEAT_CHANNEL": {Durable: "SEAT_CHANNEL_TELEGRAM_worker", AckPolicy: nats.AckExplicitPolicy,
			FilterSubject: "mesh.seat.channel.accept.*.telegram", AckWait: 2 * time.Second},
	} {
		if _, err := js.AddConsumer(stream, c); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "messenger_asks"}); err != nil {
		t.Fatal(err)
	}

	asker := mt.MembershipOf("mesh-delivery", "anchor", false, nil)
	asker.SeatTraffic = &bus.SeatTraffic{
		Publish:   []string{"mesh.seat.operator-channel.accept.ask.mesh-delivery"},
		Subscribe: []string{"mesh.seat.operator-channel.event.decided.mesh-delivery"},
		Records:   []string{"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.mesh-delivery.>"},
	}
	router := mt.MembershipOf("messenger", "anchor", false, nil)
	router.SeatTraffic = &bus.SeatTraffic{
		Publish: []string{"mesh.seat.operator-channel.event.decided.*", "mesh.seat.channel.accept.show.*"},
		Answers: []string{"mesh.seat.intake.proof.code.*"},
		Workers: []bus.SeatWorker{{Stream: "SEAT_OPERATOR_CHANNEL", Consumer: "SEAT_OPERATOR_CHANNEL_worker",
			Filter: "mesh.seat.operator-channel.accept.>"}},
	}
	router.State = []bus.StateIssued{{Name: "asks", Bucket: "messenger_asks", Writes: true}}
	telegram := mt.MembershipOf("telegram", "anchor", false, nil)
	telegram.SeatTraffic = &bus.SeatTraffic{
		Publish: []string{"mesh.seat.intake.event.choice.telegram", "mesh.seat.intake.proof.code.telegram"},
		Workers: []bus.SeatWorker{{Stream: "SEAT_CHANNEL", Consumer: "SEAT_CHANNEL_TELEGRAM_worker",
			Filter: "mesh.seat.channel.accept.*.telegram"}},
	}
	for _, m := range []bus.Membership{asker, router, telegram} {
		mesh.Issue(t, m)
	}
	conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
	if err != nil {
		t.Fatal(err)
	}
	conn.Logf = func(string, ...any) {}
	t.Cleanup(conn.Close)
	for _, m := range []string{"mesh-delivery", "messenger", "telegram"} {
		conn.Follow(m)
	}
	mt.Until(t, func() error {
		for _, m := range []string{"mesh-delivery", "messenger", "telegram"} {
			if conn.Membership(m) == nil {
				return errors.New(m + " not issued yet")
			}
		}
		return nil
	})
	return mesh, conn, js
}

func TestAModuleSubmitsAndSaysOnlyUnderItsOwnNameAndKind(t *testing.T) {
	_, conn, _ := seatMesh(t)
	body := json.RawMessage(`{"id":"a1"}`)
	if _, err := conn.SeatPublish("mesh-delivery", "mesh.seat.operator-channel.accept.ask.mesh-delivery", body, "a1"); err != nil {
		t.Fatalf("an asker could not ask under its own name: %v", err)
	}
	for module, subject := range map[string]string{
		"mesh-delivery": "mesh.seat.operator-channel.accept.ask.mesh-controller",
		"telegram":      "mesh.seat.intake.event.choice.desktop",
		"messenger":     "mesh.seat.intake.event.choice.telegram",
	} {
		if _, err := conn.SeatPublish(module, subject, body, ""); err == nil || !strings.Contains(err.Error(), "may not publish") {
			t.Errorf("%s published %s: %v", module, subject, err)
		}
	}
	if _, err := conn.SeatPublish("messenger", "mesh.seat.operator-channel.event.decided.*", body, ""); err == nil {
		t.Error("a wildcard was published")
	}
	if _, err := conn.SeatPublish("telegram", "mesh.seat.intake.proof.code.telegram", body, ""); err == nil ||
		!strings.Contains(err.Error(), "never kept") {
		t.Errorf("a proof was published onto a stream: %v", err)
	}
}

func TestTheHolderTakesItsWorkOnceAndAFailureComesBack(t *testing.T) {
	_, conn, _ := seatMesh(t)
	if _, err := conn.SeatPublish("mesh-delivery", "mesh.seat.operator-channel.accept.ask.mesh-delivery",
		json.RawMessage(`{"id":"a1"}`), "a1"); err != nil {
		t.Fatal(err)
	}
	// The same id again is one ask: a retry after a lost answer does not ask twice.
	if _, err := conn.SeatPublish("mesh-delivery", "mesh.seat.operator-channel.accept.ask.mesh-delivery",
		json.RawMessage(`{"id":"a1"}`), "a1"); err != nil {
		t.Fatal(err)
	}
	var mu sync.Mutex
	var got []string
	fail := true
	stop, err := conn.SeatTake("messenger", "SEAT_OPERATOR_CHANNEL_worker", func(w bus.Work) error {
		mu.Lock()
		defer mu.Unlock()
		got = append(got, w.Subject+" from "+w.Headers["x-source"])
		if fail {
			fail = false
			return errors.New("not recorded")
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	defer stop()
	deadline := time.Now().Add(15 * time.Second)
	for {
		mu.Lock()
		n := len(got)
		mu.Unlock()
		if n >= 2 || time.Now().After(deadline) {
			break
		}
		time.Sleep(50 * time.Millisecond)
	}
	time.Sleep(500 * time.Millisecond)
	mu.Lock()
	defer mu.Unlock()
	if len(got) != 2 || got[0] != "mesh.seat.operator-channel.accept.ask.mesh-delivery from mesh-delivery" || got[0] != got[1] {
		t.Fatalf("want the one ask, offered again after the failure: %v", got)
	}
	if _, err := conn.SeatTake("telegram", "SEAT_OPERATOR_CHANNEL_worker", func(bus.Work) error { return nil }); err == nil {
		t.Error("a channel took the router's work")
	}
}

func TestAProofIsAskedAndAnsweredAndNeverKept(t *testing.T) {
	_, conn, js := seatMesh(t)
	stop, err := conn.SeatAnswer("messenger", "mesh.seat.intake.proof.code.*", func(subject string, body json.RawMessage) (json.RawMessage, error) {
		if !strings.HasSuffix(subject, ".telegram") {
			return nil, errors.New("asked by another kind")
		}
		return json.RawMessage(`{"linked":true}`), nil
	})
	if err != nil {
		t.Fatal(err)
	}
	defer stop()
	var answer json.RawMessage
	mt.Until(t, func() error {
		answer, err = conn.SeatProve("telegram", "mesh.seat.intake.proof.code.telegram", json.RawMessage(`{"code":"123456"}`))
		return err
	})
	if string(answer) != `{"linked":true}` {
		t.Fatalf("the proof was answered %s", answer)
	}
	if _, err := conn.SeatProve("telegram", "mesh.seat.intake.proof.code.desktop", json.RawMessage(`{}`)); err == nil {
		t.Error("a channel asked a proof as another kind")
	}
	if _, err := conn.SeatAnswer("telegram", "mesh.seat.intake.proof.code.*", nil); err == nil {
		t.Error("a channel answers proofs")
	}
	for name := range js.StreamNames() {
		info, _ := js.StreamInfo(name)
		for _, s := range info.Config.Subjects {
			if bus.SubjectMatches(s, "mesh.seat.intake.proof.code.telegram") {
				t.Errorf("%s keeps proofs", name)
			}
		}
	}
}

func TestAnAskerReadsItsOwnRecordAndNoOther(t *testing.T) {
	_, conn, js := seatMesh(t)
	kv, _ := js.KeyValue("messenger_asks")
	if _, err := kv.Put("mesh-delivery.a1", []byte(`{"state":"open"}`)); err != nil {
		t.Fatal(err)
	}
	if _, err := kv.Put("mesh-controller.a1", []byte(`{"state":"open"}`)); err != nil {
		t.Fatal(err)
	}
	got, err := conn.SeatRecord("mesh-delivery", "messenger_asks", "a1")
	if err != nil || got == nil || string(got.Value) != `{"state":"open"}` {
		t.Fatalf("the asker could not read its own ask: %v %v", got, err)
	}
	if missing, err := conn.SeatRecord("mesh-delivery", "messenger_asks", "a2"); err != nil || missing != nil {
		t.Fatalf("a missing ask read as %v, %v", missing, err)
	}
	if _, err := conn.SeatRecord("telegram", "messenger_asks", "a1"); err == nil {
		t.Error("a module read records it is not given")
	}
}

func TestOfTwoAnswersToOneAskOnlyTheFirstIsWritten(t *testing.T) {
	_, conn, _ := seatMesh(t)
	rev, err := conn.StateCreate("messenger", "asks", "mesh-delivery.a1", json.RawMessage(`{"state":"open"}`))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := conn.StateCreate("messenger", "asks", "mesh-delivery.a1", json.RawMessage(`{"state":"open"}`)); !errors.Is(err, bus.ErrStateChanged) {
		t.Errorf("a second create stood: %v", err)
	}
	if _, err := conn.StateUpdate("messenger", "asks", "mesh-delivery.a1", json.RawMessage(`{"state":"answered"}`), rev); err != nil {
		t.Fatal(err)
	}
	if _, err := conn.StateUpdate("messenger", "asks", "mesh-delivery.a1", json.RawMessage(`{"state":"answered"}`), rev); !errors.Is(err, bus.ErrStateChanged) {
		t.Errorf("a second answer from the same revision stood: %v", err)
	}
}