//go:build integration

package main

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

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/process"

	"nats-tools/internal/bussnap"
	"nats-tools/internal/keeper"
)

// kvStore is the keeper's own state on the real bus: one bucket per state, as the grants part creates its old sections.
type kvStore struct {
	js jetstream.JetStream
}

func (s kvStore) kv(state string) (jetstream.KeyValue, error) {
	return s.js.KeyValue(context.Background(), "nats_"+state)
}

func (s kvStore) Get(state, key string) ([]byte, uint64, bool, error) {
	kv, err := s.kv(state)
	if err != nil {
		return nil, 0, false, err
	}
	e, err := kv.Get(context.Background(), key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, 0, false, nil
	}
	if err != nil {
		return nil, 0, false, err
	}
	return e.Value(), e.Revision(), true, nil
}

func (s kvStore) Put(state, key string, value []byte) (uint64, error) {
	kv, err := s.kv(state)
	if err != nil {
		return 0, err
	}
	return kv.Put(context.Background(), key, value)
}

func (s kvStore) Delete(state, key string) error {
	kv, err := s.kv(state)
	if err != nil {
		return err
	}
	return kv.Delete(context.Background(), key)
}

func (s kvStore) Create(state, key string, value []byte) (uint64, error) {
	kv, err := s.kv(state)
	if err != nil {
		return 0, err
	}
	rev, err := kv.Create(context.Background(), key, value)
	if errors.Is(err, jetstream.ErrKeyExists) {
		return 0, keeper.ErrExists
	}
	return rev, err
}

func (s kvStore) Keys(state string) ([]string, error) {
	kv, err := s.kv(state)
	if err != nil {
		return nil, err
	}
	keys, err := kv.Keys(context.Background())
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	return keys, err
}

func eventually(t *testing.T, what string, ok func() bool) {
	t.Helper()
	deadline := time.Now().Add(10 * time.Second)
	for !ok() {
		if time.Now().After(deadline) {
			t.Fatalf("not within 10 s: %s", what)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

// The bus module runs whole (design 25 § 12.7, "The bus module runs whole" and "The bus module's first view"; CAT-N10):
// its keeper against the real bus server, hearing manifest-accepted on its durable consumer, creating on the bus,
// saying its events on its own stream BROKER_EVENTS, keeping its state in its own buckets, and acting on a person's
// requests. One server for every case, in order, as the mesh runs them.
func TestTheBusModuleRunsWhole(t *testing.T) {
	dir := t.TempDir()
	s, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: dir, NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(5 * time.Second) {
		t.Fatal("the server did not start")
	}
	defer s.Shutdown()
	nc, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, _ := jetstream.New(nc)
	ctx := context.Background()
	account := process.Account{Node: "anchor", Module: keeper.Module, User: "anchor.nats.keeper"}
	// What the grants part creates for the bus module from its manifest's old sections: EVENTS and the keeper's
	// consumer on it, its own stream, and its two states.
	for _, c := range []jetstream.StreamConfig{
		{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}},
		{Name: BrokerEventsStream, Subjects: []string{"mesh.stream.nats.event.>"}, MaxAge: 30 * 24 * time.Hour, MaxBytes: 256 << 20},
		{Name: "CONDITIONS", Subjects: []string{"mesh.seat.operator-channel.accept.condition.>"}},
	} {
		if _, err := js.CreateStream(ctx, c); err != nil {
			t.Fatal(err)
		}
	}
	// The keeper's consumer hears the indexer's verdicts and the operator channel's decided for the bus module's own
	// asks, as the grants part composes it from its consumes and the seats it uses.
	if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
		FilterSubjects: []string{AcceptedSubject, DecidedSubject}, AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 5}); err != nil {
		t.Fatal(err)
	}
	// The router's own record of the asks, which the keeper reads under its own process account (process.Conn.Ask), never
	// a stand-in: an ask's record is `<asker>.<ask id>`, as the router keeps it.
	asksKV, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: asksBucket, History: 5})
	if err != nil {
		t.Fatal(err)
	}
	pconn, err := process.Connect(process.Account{URL: s.ClientURL(), Node: account.Node, Module: account.Module, User: account.User, Password: "p"})
	if err != nil {
		t.Fatal(err)
	}
	defer pconn.Close()
	for _, st := range []string{keeper.StateAsked, keeper.StateRequests, keeper.StateSpent} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "nats_" + st, History: 5}); err != nil {
			t.Fatal(err)
		}
	}
	store := kvStore{js}
	snapshots := t.TempDir()
	var mu sync.Mutex
	view := keeper.FirstView{Revision: 1}
	var acts []string
	newKeeper := func() *keeper.Keeper {
		return &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, snapshots)}, Store: store,
			Emit: streamEmitter(js, account), Say: sayer(js),
			View: func() (keeper.FirstView, error) { mu.Lock(); defer mu.Unlock(); return view, nil },
			Act: func(what, why, cause string) error {
				mu.Lock()
				defer mu.Unlock()
				acts = append(acts, cause+": "+what)
				return nil
			},
			Warrant: WarrantFrom(pconn.Ask),
			Logf:    t.Logf}
	}
	run := func(k *keeper.Keeper) func() {
		kctx, stop := context.WithCancel(ctx)
		done := make(chan struct{})
		watch := func(c context.Context, act func()) error {
			kv, _ := js.KeyValue(ctx, "nats_"+keeper.StateRequests)
			w, err := kv.WatchAll(c)
			if err != nil {
				return err
			}
			go func() {
				for range w.Updates() {
					act()
				}
			}()
			return nil
		}
		go func() { _ = Keep(kctx, k, hearer(js, account, t.Logf), watch, t.Logf); close(done) }()
		return func() { stop(); <-done }
	}
	accepted := func(module string, revision uint64, entries ...catalogue.BusStateEntry) {
		body, _ := json.Marshal(map[string]any{"v": 1, "module": module, "digest": fmt.Sprintf("sha256:%d", revision),
			"judge": "j", "revision": revision, "bus-state": entries})
		if _, err := js.Publish(ctx, AcceptedSubject, body); err != nil {
			t.Fatal(err)
		}
	}
	// saidOn reads what one stream holds of the bus module's events, by its subjects' prefix: its own stream (its
	// history), or EVENTS on the seat mesh-broker's subjects (what every consumer of mesh-broker.<event> hears).
	saidOn := func(stream, prefix string) []string {
		cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{prefix + ">"}})
		if err != nil {
			t.Fatal(err)
		}
		info, _ := cons.Info(ctx)
		var out []string
		if info.NumPending == 0 {
			return nil
		}
		batch, _ := cons.Fetch(int(info.NumPending), jetstream.FetchMaxWait(2*time.Second))
		for m := range batch.Messages() {
			var e keeper.Event
			_ = json.Unmarshal(m.Data(), &e)
			out = append(out, strings.TrimPrefix(m.Subject(), prefix)+" "+e.Module+" "+e.Local)
		}
		return out
	}
	said := func() []string { return saidOn(BrokerEventsStream, "mesh.stream.nats.event.") }
	saidHas := func(want string) func() bool {
		return func() bool {
			for _, l := range said() {
				if l == want {
					return true
				}
			}
			return false
		}
	}
	held := func(stream string) (*jetstream.StreamInfo, bool) {
		st, err := js.Stream(ctx, stream)
		if err != nil {
			return nil, false
		}
		info, err := st.Info(ctx)
		return info, err == nil
	}
	state := entry("drafts", "key-value", "state", nil)
	log := entry("log", "stream", "log", map[string]string{"max-mib": "1"})

	// Review round 1 (running F1): the fill pass runs from the start, while the first view is still asked, so the store's
	// one guard never waits on the indexer.
	t.Run("the fill pass runs while the first view is asked", func(t *testing.T) {
		down := newKeeper()
		down.View = func() (keeper.FirstView, error) { return keeper.FirstView{}, errors.New("no responders") }
		stopDown := run(down)
		eventually(t, "a fill pass with no first view", func() bool { return !down.Status().Pass.IsZero() && down.Status().FirstView.Answered.IsZero() })
		stopDown()
		// And while its consumer of manifest-accepted is bound again on its backoff (review round 2, running F1).
		was := FillEvery
		FillEvery = 20 * time.Millisecond
		defer func() { FillEvery = was }()
		deaf := newKeeper()
		kctx, stopDeaf := context.WithCancel(ctx)
		done := make(chan struct{})
		go func() {
			_ = Keep(kctx, deaf, func(context.Context, func(string, []byte) error) error { return errors.New("no consumer yet") }, nil, t.Logf)
			close(done)
		}()
		eventually(t, "the first view answered", func() bool { return !deaf.Status().FirstView.Answered.IsZero() })
		first := deaf.Status().Pass
		eventually(t, "a fill pass while the consumer is bound again", func() bool { return deaf.Status().Pass.After(first) })
		stopDeaf()
		<-done
	})
	k := newKeeper()
	stop := run(k)
	t.Run("a verdict, then created, with its binding", func(t *testing.T) {
		accepted("m", 2, state, log)
		eventually(t, "created m drafts and m log", func() bool { return saidHas("created m drafts")() && saidHas("created m log")() })
		// Said on the seat's subject as well, on the shared events stream (the lead's ruling, 13:0x).
		if seat := saidOn("EVENTS", "mesh.seat.mesh-broker.event."); !slices.Contains(seat, "created m drafts") ||
			!slices.Contains(seat, "created m log") {
			t.Errorf("EVENTS holds %v of mesh-broker's events: created m drafts and m log, as on the bus module's own stream", seat)
		}
		if _, ok := held("KV_m_drafts"); !ok {
			t.Error("the bus holds no KV_m_drafts")
		}
		if info, ok := held("m_log"); !ok || info.Config.Discard != jetstream.DiscardNew || info.Config.Subjects[0] != "mesh.bus.m.log.>" {
			t.Errorf("m_log is %+v: a log refuses when full, on the module's own subjects", info)
		}
	})
	t.Run("a restart catches up a verdict given while it was down", func(t *testing.T) {
		stop()
		accepted("n", 3, entry("jobs", "stream", "work-queue", nil))
		k = newKeeper()
		stop = run(k)
		eventually(t, "created n jobs", saidHas("created n jobs"))
	})
	t.Run("a replay older than the first view is not applied; a set-aside module keeps its key", func(t *testing.T) {
		stop()
		mu.Lock()
		view = keeper.FirstView{Revision: 9, Modules: []keeper.Accepted{{Module: "p", Revision: 9, BusState: []catalogue.BusStateEntry{state}}}}
		mu.Unlock()
		accepted("p", 4)
		k = newKeeper()
		stop = run(k)
		eventually(t, "created p drafts", saidHas("created p drafts"))
		eventually(t, "the replay heard", func() bool {
			info, _ := js.Consumer(ctx, "EVENTS", account.QueueConsumerName())
			i, _ := info.Info(ctx)
			return i.NumPending == 0 && i.NumAckPending == 0
		})
		if saidHas("orphaned p drafts")() {
			t.Error("the replay at revision 4 orphaned what the first view at 9 asked")
		}
		if _, _, found, _ := store.Get(keeper.StateAsked, "m"); !found {
			t.Error("m, missing from the first view, lost its key")
		}
	})
	t.Run("drift is said, a growth applied and the drift cleared", func(t *testing.T) {
		accepted("m", 5, state, entry("log", "stream", "log", map[string]string{"max-mib": "1", "max-age-days": "2"}))
		eventually(t, "drifted m log", saidHas("drifted m log"))
		accepted("m", 6, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
		eventually(t, "drift-cleared m log", saidHas("drift-cleared m log"))
		if info, _ := held("m_log"); info.Config.MaxBytes != 4<<20 {
			t.Errorf("m_log's cap is %d: the larger cap applied", info.Config.MaxBytes)
		}
	})
	t.Run("a refused create hands no binding", func(t *testing.T) {
		bare := entry("raw", "stream", "", nil)
		accepted("m", 7, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}), bare)
		eventually(t, "refused m raw", saidHas("refused m raw"))
		if saidHas("created m raw")() {
			t.Error("created m raw")
		}
	})
	t.Run("each template at a small cap refuses, full raised at the raise mark and cleared below the clear mark", func(t *testing.T) {
		for i := 0; ; i++ {
			_, err := js.Publish(ctx, fmt.Sprintf("mesh.bus.m.log.k%d", i), make([]byte, 200<<10))
			if err != nil {
				break
			}
			if i > 40 {
				t.Fatal("a 4 MiB log took more than 8 MiB")
			}
		}
		k.FillPass()
		eventually(t, "full m log", saidHas("full m log"))
		st, _ := js.Stream(ctx, "m_log")
		_ = st.Purge(ctx)
		k.FillPass()
		eventually(t, "full-cleared m log", saidHas("full-cleared m log"))
	})
	t.Run("an orphan is listed, and goes only by drop on the operator's warrant, after a snapshot", func(t *testing.T) {
		accepted("m", 8, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
		eventually(t, "orphaned m drafts", saidHas("orphaned m drafts"))
		put := func(r keeper.Request) {
			raw, _ := json.Marshal(r)
			if _, err := store.Put(keeper.StateRequests, r.ID, raw); err != nil {
				t.Fatal(err)
			}
		}
		get := func(id string) keeper.Request {
			raw, _, _, _ := store.Get(keeper.StateRequests, id)
			var r keeper.Request
			_ = json.Unmarshal(raw, &r)
			return r
		}
		put(keeper.Request{ID: "r1", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused"})
		eventually(t, "the drop with no ask refused", func() bool { return strings.Contains(get("r1").Error, "operator's warrant") })
		r := keeper.Request{ID: "r2", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused", At: time.Now()}
		a, _ := askFor(r, time.Now())
		r.Ask = a.ID
		record := func(w *oc.Warrant) {
			raw, _ := json.Marshal(map[string]any{"asker": keeper.Module, "ask": a, "state": "open", "warrant": w})
			if _, err := asksKV.Put(ctx, keeper.Module+"."+a.ID, raw); err != nil {
				t.Fatal(err)
			}
		}
		record(nil)
		put(r)
		eventually(t, "the request read, waiting for the operator", func() bool { return get("r2").Waiting != "" })
		if _, ok := held("KV_m_drafts"); !ok || !get("r2").Done.IsZero() {
			t.Fatal("dropped before the operator chose")
		}
		// The operator chooses: the router records the warrant and says decided; the keeper acts on decided, on its
		// durable consumer, with no timer between and no call of its pass by the test (seams F1).
		w := &oc.Warrant{Ask: a.ID, Asker: keeper.Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
			Option: "go", Label: "Remove", Level: occ.Destroy, At: occ.FormatTime(time.Now()), AskDigest: occ.Digest(a)}
		record(w)
		decided := time.Now()
		body, _ := json.Marshal(w)
		if _, err := js.Publish(ctx, DecidedSubject, body); err != nil {
			t.Fatal(err)
		}
		eventually(t, "dropped m drafts", saidHas("dropped m drafts"))
		if since := time.Since(decided); since > 5*time.Second {
			t.Errorf("dropped %s after decided: within seconds", since)
		}
		if _, ok := held("KV_m_drafts"); ok {
			t.Error("KV_m_drafts is still on the bus")
		}
		if p := get("r2").Snapshot; p == "" {
			t.Error("no snapshot taken before the drop")
		} else if st, err := os.Stat(p); err != nil || st.Size() == 0 {
			t.Errorf("the snapshot %s is not there whole: %v", p, err)
		}
		mu.Lock()
		recorded := strings.Join(acts, "\n")
		mu.Unlock()
		if !strings.Contains(recorded, "nats_drop: dropped drafts of m") {
			t.Errorf("recorded %s: the drop as a hand act", recorded)
		}
		// The same warrant handed in a second request is refused and deletes nothing (security F2): drafts asked for and
		// orphaned again, then a request naming the spent ask.
		accepted("m", 9, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
		eventually(t, "m drafts made again", func() bool { _, ok := held("KV_m_drafts"); return ok })
		accepted("m", 10, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
		eventually(t, "m drafts orphaned again", func() bool { return k.Held()["m"].Made["drafts"].Orphaned })
		again := keeper.Request{ID: "r2b", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "again", Ask: a.ID, At: time.Now()}
		put(again)
		eventually(t, "the second request on the spent warrant refused", func() bool { return strings.Contains(get("r2b").Error, "already acted") })
		if _, ok := held("KV_m_drafts"); !ok {
			t.Error("the spent warrant deleted KV_m_drafts a second time")
		}
	})
	t.Run("nats_first_view_again asks at once and records the hand act", func(t *testing.T) {
		raw, _ := json.Marshal(keeper.Request{ID: "r3", Verb: keeper.VerbFirstViewAgain, Why: "a verdict thought missed", Caller: "person.jo"})
		_, _ = store.Put(keeper.StateRequests, "r3", raw)
		eventually(t, "the first view asked again", func() bool {
			mu.Lock()
			defer mu.Unlock()
			return strings.Contains(strings.Join(acts, "\n"), "nats_first_view_again")
		})
	})
	stop()
}

func entry(name, storage, template string, options map[string]string) catalogue.BusStateEntry {
	e := catalogue.BusStateEntry{Name: name, Storage: storage, Template: template}
	if len(options) > 0 {
		e.Options = map[string]json.RawMessage{}
		for k, v := range options {
			e.Options[k] = json.RawMessage(v)
		}
	}
	return e
}