//go:build integration

package main

// 1596/AC6: the tool side and the converter over a real bus (the shape of brother's device_integration_test.go,
// ADR 0436 §7): a JetStream server of the test's own, the work-queue stream convert-asks and its consumer, the state
// convert-answers, and a fake document server. The tool side's Convert queues an ask; the converter takes it and
// answers; Convert returns the answer. Compiled by the merge gate's `go vet -tags integration`; never run on this machine.

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// convertAskedSubject is the wire subject the controller composes for the stream's one event:
// "mesh.stream." + module + ".event." + event (mesh-controller internal/broker/modstreams.go).
const convertAskedSubject = "mesh.stream.only-office.event." + EventAsked

// natsBus is the tool side's bus: the ask published on the stream, the answer read from the state's bucket.
type natsBus struct {
	js jetstream.JetStream
	kv jetstream.KeyValue
}

func (b natsBus) Publish(id string, body any) error {
	raw, _ := json.Marshal(body)
	_, err := b.js.Publish(context.Background(), convertAskedSubject, raw, jetstream.WithMsgID(id))
	return err
}

func (b natsBus) Answer(id string) (json.RawMessage, error) {
	e, err := b.kv.Get(context.Background(), id)
	if err != nil {
		return nil, nil
	}
	return e.Value(), nil
}

func TestTheToolSideAndTheConverterConvertOverARealBus(t *testing.T) {
	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}
	s, err := server.NewServer(opts)
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(30 * time.Second) {
		t.Fatal("a bus for this test did not come up within 30s")
	}
	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
	ctx := context.Background()
	nc, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: AsksStream, Subjects: []string{convertAskedSubject},
		Retention: jetstream.WorkQueuePolicy, Discard: jetstream.DiscardNew, MaxAge: 24 * time.Hour, MaxBytes: 8 << 20}); err != nil {
		t.Fatal(err)
	}
	account := process.Account{URL: s.ClientURL(), Node: "novox", Module: "only-office", User: "novox.only-office.converter", Password: "pw"}
	if _, err := js.CreateConsumer(ctx, AsksStream, jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
		AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: 3}); err != nil {
		t.Fatal(err)
	}
	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: account.Module + "_" + AnswersState, TTL: 10 * time.Minute})
	if err != nil {
		t.Fatal(err)
	}

	d, tgt := newFakeDocs(t), newFakeTarget(t)
	w, _ := worker(d)
	runCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	go runConverter(runCtx, account, w)

	svc := &Service{Asks: natsBus{js: js, kv: kv}, Now: time.Now, Wait: 20 * time.Second, Poll: 50 * time.Millisecond}
	got, err := svc.Convert(ctx, "http://novox.internal:20001/b/in.docx?X-Amz-Signature=a", "docx", tgt.srv.URL+"/b/in.pdf?sig=1", "")
	if err != nil {
		t.Fatal(err)
	}
	if got["size"] != float64(len(d.pdf)) || string(tgt.got) != string(d.pdf) {
		t.Fatalf("answer %v, target got %q", got, tgt.got)
	}
	if len(d.requests) != 1 {
		t.Fatalf("%d conversion requests", len(d.requests))
	}
}