//go:build integration

package steprun

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

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

	sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
	"git.novox.be/novox/mesh-sdk/go/kvlimit"
	"git.novox.be/novox/mesh-sdk/go/testbus"
)

func askBus(t *testing.T, ackWait time.Duration) jetstream.JetStream {
	t.Helper()
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	if err := DeclareAskWorkers(context.Background(), js, sdkstep.AskGo, ackWait); err != nil {
		t.Fatal(err)
	}
	return js
}

func publishAsk(t *testing.T, js jetstream.JetStream, a sdkstep.StepAsk) {
	t.Helper()
	raw, err := sdkstep.EncodeStepAsk(a)
	if err != nil {
		t.Fatal(err)
	}
	// A build rides main's subject; any other kind a named asker's (design 47 G §543: main takes only a build and a run).
	subject, _ := sdkstep.AskSubject(a.Language, a.Kind)
	if a.Kind != sdkstep.AskDelivery {
		subject, _ = sdkstep.AskSubjectBy(a.Language, a.Kind, "mesh-delivery-checker")
	}
	if _, err := js.Publish(context.Background(), subject, raw); err != nil {
		t.Fatal(err)
	}
}

// askKey is the key of a's step on node as publishAsk put it: under its asker, main's form for a build (design 47 G §481).
func askKey(a sdkstep.StepAsk, node string) (string, error) {
	if a.Kind == sdkstep.AskDelivery {
		return sdkstep.AskStepKey("", node, a.ID)
	}
	return sdkstep.AskStepKey("mesh-delivery-checker", node, a.ID)
}

// 1887/A2a, 1887/AC-order, 1887/AC-pull on the real server: a check ask published before a delivery ask runs after it;
// with capacity 2 and three asks two run, the third waits in the stream and starts when one ends.
func TestPullOrderAndCapacityOnTheBus(t *testing.T) {
	js := askBus(t, 30*time.Second)
	check, first, second := testAsk(sdkstep.AskCheck), testAsk(sdkstep.AskDelivery), testAsk(sdkstep.AskDelivery)
	publishAsk(t, js, check)
	publishAsk(t, js, first)
	publishAsk(t, js, second)
	release := make(chan struct{})
	var mu sync.Mutex
	var order []string
	st := newSteps()
	p := &Puller{Source: &JetStreamAsks{JS: js, Language: sdkstep.AskGo}, Language: sdkstep.AskGo, Node: "shanks", Put: st.put,
		Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond,
		Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
			mu.Lock()
			order = append(order, a.ID)
			mu.Unlock()
			<-release
			return nil, nil
		}}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go p.Pull(ctx)
	waitFor(t, "two in hand", func() bool { return p.InHand() == 2 })
	time.Sleep(500 * time.Millisecond)
	mu.Lock()
	if len(order) != 2 || order[0] != first.ID || order[1] != second.ID {
		t.Fatalf("ran %v, want the two delivery asks %s %s first", order, first.ID, second.ID)
	}
	mu.Unlock()
	release <- struct{}{}
	waitFor(t, "the check taken", func() bool { mu.Lock(); defer mu.Unlock(); return len(order) == 3 })
	close(release)
	waitFor(t, "every step ended", func() bool {
		for _, a := range []sdkstep.StepAsk{check, first, second} {
			key, _ := askKey(a, "shanks")
			if s, ok := st.get(key); !ok || s.State != sdkstep.StepPassed {
				return false
			}
		}
		return true
	})
	info, err := js.Stream(ctx, "SEAT_NODE_BUILD_GO")
	if err != nil {
		t.Fatal(err)
	}
	waitFor(t, "the stream empty (every ask acknowledged)", func() bool {
		si, err := info.Info(ctx)
		return err == nil && si.State.Msgs == 0
	})
}

// 1887/A2c, ADR 0525 §6: a killed puller's ask is redelivered by the stream's acknowledgement wait to the other node,
// whose step key names the reference.
func TestPullRedeliversAKilledPullersAsk(t *testing.T) {
	js := askBus(t, 2*time.Second)
	a := testAsk(sdkstep.AskDelivery)
	publishAsk(t, js, a)
	st := newSteps()
	taken := make(chan struct{})
	dying := &Puller{Source: &JetStreamAsks{JS: js, Language: sdkstep.AskGo}, Language: sdkstep.AskGo, Node: "ace", Put: st.put,
		Capacity: func() int { return 1 }, PullWait: 200 * time.Millisecond, Heartbeat: time.Hour,
		Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
			close(taken)
			<-ctx.Done()
			return nil, ctx.Err()
		}}
	kill, killed := context.WithCancel(context.Background())
	go dying.Pull(kill)
	<-taken
	killed()
	other := &Puller{Source: &JetStreamAsks{JS: js, Language: sdkstep.AskGo}, Language: sdkstep.AskGo, Node: "shanks", Put: st.put,
		Capacity: func() int { return 1 }, PullWait: 200 * time.Millisecond,
		Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
			return sdkstep.BuildAnswer{Artifacts: []sdkstep.AskBuilt{{Artifact: "bundle", Digest: "sha256:00", Reference: "artifact-store://x"}}}, nil
		}}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go other.Pull(ctx)
	key, _ := askKey(a, "shanks")
	deadline := time.Now().Add(10 * time.Second)
	for {
		if s, ok := st.get(key); ok && s.State == sdkstep.StepPassed {
			if want := `artifact-store://x`; !contains(string(s.Result), want) {
				t.Fatalf("the step's result %s names no %s", s.Result, want)
			}
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("the killed puller's ask was not taken by shanks within 10 s")
		}
		time.Sleep(20 * time.Millisecond)
	}
}

// ADR 0664 §1, §5 (issue 1748) on the real server: a run that ends on a transport fault is written again and the bus gives the
// ask again after the delay, not before, as the second run (its delivery count one higher), and that run's end is the ask's;
// a step that runs when its ask is dropped ends dropped, naming who dropped it, and the ask is not given again.
func TestAnAskThatMetTheNetworkIsGivenAgainAndADroppedOneIsEnded(t *testing.T) {
	js := askBus(t, 30*time.Second)
	a, doomed := testAsk(sdkstep.AskDelivery), testAsk(sdkstep.AskDelivery)
	publishAsk(t, js, a)
	st, kv := newSteps(), &pauseKV{}
	const delay = 600 * time.Millisecond
	var mu sync.Mutex
	var runs []time.Time
	hold := make(chan struct{})
	p := &Puller{Source: &JetStreamAsks{JS: js, Language: sdkstep.AskGo}, Language: sdkstep.AskGo, Node: "ace", Put: st.put, PausePut: kv.put, Load: kv.load,
		ForgetPause: kv.forget, PauseKeys: kv.keys, Capacity: func() int { return 1 }, PullWait: 100 * time.Millisecond,
		Delay: func(int) time.Duration { return delay },
		Run: func(ctx context.Context, got sdkstep.StepAsk) (any, error) {
			if got.ID == doomed.ID {
				close(hold)
				<-ctx.Done()
				return nil, ctx.Err()
			}
			mu.Lock()
			runs = append(runs, time.Now())
			n := len(runs)
			mu.Unlock()
			if n == 1 {
				return nil, &sdkstep.Passing{Err: errors.New("lookup git.novox.be: i/o timeout")}
			}
			return nil, nil
		}}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go p.Pull(ctx)
	key, _ := askKey(a, "ace")
	waitFor(t, "the ask's second run ended", func() bool { s, ok := st.get(key); return ok && s.State == sdkstep.StepPassed })
	mu.Lock()
	gap := runs[1].Sub(runs[0])
	mu.Unlock()
	if s, _ := st.get(key); len(runs) != 2 || gap < delay || s.Run != 2 {
		t.Errorf("runs %d, %s apart (delay %s), the step's run %d: want two runs, the second after the delay, as run 2", len(runs), gap, delay, s.Run)
	}
	publishAsk(t, js, doomed)
	<-hold
	if err := p.Drop(doomed.ID, "mesh-delivery-builder", "superseded"); err != nil {
		t.Fatal(err)
	}
	dkey, _ := askKey(doomed, "ace")
	waitFor(t, "the dropped step dropped", func() bool { s, ok := st.get(dkey); return ok && s.State == sdkstep.StepDropped })
	if s, _ := st.get(dkey); !contains(s.Reason, "dropped by mesh-delivery-builder: superseded") {
		t.Errorf("the dropped step says %q", s.Reason)
	}
}

func contains(s, sub string) bool { return strings.Contains(s, sub) }

// Design 47 G §544: main's worker takes an ask put on main's subject and the named worker one put on the asker's, the two
// filters not overlapping.
func TestTheTwoWorkersTakeTheTwoFormsOfTheAcceptSubject(t *testing.T) {
	js := askBus(t, 30*time.Second)
	stream, _ := sdkstep.AskStream(sdkstep.AskGo)
	worker, _ := sdkstep.AskWorker(sdkstep.AskGo, sdkstep.AskDelivery)
	named, _ := sdkstep.AskNamedWorker(sdkstep.AskGo, sdkstep.AskDelivery)
	main, _ := sdkstep.AskSubject(sdkstep.AskGo, sdkstep.AskDelivery)
	for _, subject := range []string{main, main + ".mesh-builds"} {
		if _, err := js.Publish(context.Background(), subject, []byte("{}")); err != nil {
			t.Fatal(err)
		}
	}
	for name, want := range map[string]string{worker: main, named: main + ".mesh-builds"} {
		c, err := js.Consumer(context.Background(), stream, name)
		if err != nil {
			t.Fatal(err)
		}
		batch, err := c.Fetch(2, jetstream.FetchMaxWait(time.Second))
		if err != nil {
			t.Fatal(err)
		}
		var got []string
		for m := range batch.Messages() {
			got = append(got, m.Subject())
			_ = m.Ack()
		}
		if len(got) != 1 || got[0] != want {
			t.Errorf("%s took %v, want only %s", name, got, want)
		}
	}
}

// Design 47 G A-290 on the real server: a check's answer over what the steps state holds per value (a bucket made with
// kvlimit.ModuleStateValueBytes, as every module's is) is kept as the object output.<ask> in the outputs store, the step's key
// names it and ends passed, and the answer reads back whole through the seat's verb output (Step.Whole).
func TestACheckAnswerOverTheStepsStateIsKeptAndReadBackWhole(t *testing.T) {
	js := askBus(t, 30*time.Second)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "steps-a290", MaxValueSize: kvlimit.ModuleStateValueBytes})
	if err != nil {
		t.Fatal(err)
	}
	objects, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "outputs-a290"})
	if err != nil {
		t.Fatal(err)
	}
	put := func(key string, value any) (uint64, error) {
		raw, err := json.Marshal(value)
		if err != nil {
			return 0, err
		}
		return kv.Put(ctx, key, raw)
	}
	answer := map[string]any{"fetch": map[string]any{"workspace": "/w"}, "levels": []any{map[string]any{"level": "unit", "said": strings.Repeat("z", 2*kvlimit.ModuleStateValueBytes)}}}
	a := testAsk(sdkstep.AskCheck)
	publishAsk(t, js, a)
	p := &Puller{Source: &JetStreamAsks{JS: js, Language: sdkstep.AskGo}, Language: sdkstep.AskGo, Node: "ace", Put: put,
		Capacity: func() int { return 1 }, PullWait: 100 * time.Millisecond,
		KeepOutput: func(object string, body []byte) error { _, err := objects.PutBytes(ctx, object, body); return err },
		Run:        func(context.Context, sdkstep.StepAsk) (any, error) { return answer, nil }}
	go p.Pull(ctx)
	key, _ := askKey(a, "ace")
	var step sdkstep.Step
	waitFor(t, "the check's step ended", func() bool {
		e, err := kv.Get(ctx, key)
		if err != nil {
			return false
		}
		step, err = sdkstep.DecodeStep(e.Value())
		return err == nil && sdkstep.Ended(step)
	})
	if step.State != sdkstep.StepPassed || step.OutputObject != sdkstep.OutputObjectName(a.ID) || len(step.Result) != 0 {
		t.Fatalf("the key ended %s (%s) naming %q with %d bytes of result, want passed, naming the object and no result", step.State, step.Reason, step.OutputObject, len(step.Result))
	}
	verb := func(k string, body any) (json.RawMessage, error) {
		if k != "seat:node-build-go.output@ace" {
			return nil, fmt.Errorf("asked %s", k)
		}
		data, err := objects.GetBytes(ctx, sdkstep.OutputObjectName(body.(map[string]any)["ask"].(string)))
		if err != nil {
			return nil, err
		}
		return json.Marshal(map[string]any{"ask": a.ID, "file": base64.StdEncoding.EncodeToString(data)})
	}
	whole, err := step.Whole(sdkstep.AskGo, false, verb)
	if err != nil {
		t.Fatal(err)
	}
	want, _ := json.Marshal(answer)
	if string(whole.Result) != string(want) {
		t.Fatalf("read back %d bytes, want the whole answer of %d", len(whole.Result), len(want))
	}
}