//go:build integration

package main

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

	sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/link"
	"github.com/novox/mesh-controller/internal/testbus"
)

type sigOf struct{}

func (sigOf) Sign(context.Context, []byte) ([]byte, error) { return make([]byte, 64), nil }

// 1696/AC7 (ADR 0486 §1): on a real bus with its default largest message of 1 MiB, a declaration whose files add up to more than
// 3 MiB is sent as references, its objects are in the bucket, each fetched back whole and verified against the digest and size the
// declaration names; the same declaration inline is refused by the controller, naming size and limit. A changed byte in an object
// fails the verification.
func TestADeclarationOfMoreThanThreeMiBIsSentByReferenceOnARealBus(t *testing.T) { // 1696/AC7
	s := testbus.Start(t)
	conn, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	nc, err := conn.JetStream()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := nc.AddStream(&nats.StreamConfig{Name: "NODES", Subjects: []string{"mesh.node.*.declare"},
		MaxMsgsPerSubject: 1}); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	api, err := jetstream.New(conn)
	if err != nil {
		t.Fatal(err)
	}
	if err := broker.EnsureDeclarationFiles(ctx, api, "novox"); err != nil {
		t.Fatal(err)
	}
	// Run twice: an assertion of an existing bucket is no error and no change.
	if err := broker.EnsureDeclarationFiles(ctx, api, "novox"); err != nil {
		t.Fatalf("asserting the bucket again: %v", err)
	}

	var resources []map[string]any
	for i := 0; i < 3; i++ {
		resources = append(resources, aFileResource("big-"+string(rune('a'+i)), contentOf("big"+string(rune('a'+i)), 1200<<10)))
	}
	resources = append(resources, aFileResource("small", "tiny\n"))
	inline := sendable{Resources: resources}
	byDigest := sendable{fileReferences: true, Resources: resources}

	bus := link.OverNATS{Conn: conn, JS: nc}
	body, err := inline.Body()
	if err != nil {
		t.Fatal(err)
	}
	if err := link.Declare(ctx, bus, sigOf{}, "novox", body, 5*time.Second); err == nil ||
		!strings.Contains(err.Error(), "1048576") || strings.Contains(err.Error(), "maximum payload exceeded") {
		t.Fatalf("the inline declaration of more than 3 MiB: %v", err)
	}

	bucket, err := openBucket(ctx, conn, "novox")
	if err != nil {
		t.Fatal(err)
	}
	objects, err := byDigest.objects()
	if err != nil || len(objects) != 3 {
		t.Fatalf("objects: %d %v", len(objects), err)
	}
	if err := putObjects(ctx, bucket, objects); err != nil {
		t.Fatal(err)
	}
	if err := putObjects(ctx, bucket, objects); err != nil {
		t.Fatalf("putting them again: %v", err)
	}
	refBody, err := byDigest.Body()
	if err != nil {
		t.Fatal(err)
	}
	if err := link.Declare(ctx, bus, sigOf{}, "novox", refBody, 5*time.Second); err != nil {
		t.Fatalf("the declaration by reference: %v", err)
	}

	// What the stream holds for the node is the reference form, and it names exactly the three objects.
	latest, err := latestDeclarations(ctx, conn, []string{"novox", "ace"})
	if err != nil || len(latest) != 1 {
		t.Fatalf("the latest declarations: %v %v", len(latest), err)
	}
	named, err := namedByLatest("novox", latest["novox"])
	if err != nil || len(named) != 3 {
		t.Fatalf("what the latest declaration names: %v %v", named, err)
	}
	novoxBucket, err := broker.DeclarationFilesBucketOf("novox")
	if err != nil {
		t.Fatal(err)
	}
	store, err := api.ObjectStore(ctx, novoxBucket)
	if err != nil {
		t.Fatal(err)
	}
	var envelope struct {
		Resources []sdkdecl.File `json:"resources"`
	}
	if err := json.Unmarshal(refBody, &envelope); err != nil {
		t.Fatal(err)
	}
	for _, f := range envelope.Resources {
		if !f.IsReference() {
			continue
		}
		name, _ := sdkdecl.ObjectName(f.Digest)
		got, err := store.GetBytes(ctx, name)
		if err != nil {
			t.Fatalf("the object of %s: %v", f.ID, err)
		}
		if err := f.Verify(got); err != nil {
			t.Errorf("the object does not verify: %v", err)
		}
		got[len(got)/2] ^= 1
		if err := f.Verify(got); err == nil {
			t.Errorf("a changed byte in the object of %s passed the verification", f.ID)
		}
	}

	// The sweep against the real bucket: all named, nothing marked; a fourth object named by none is marked, kept a day less a
	// minute, and removed at the day.
	extra := bytes.Repeat([]byte("x"), 100)
	extraName := strings.Repeat("e", 64)
	if err := putObjects(ctx, bucket, map[string][]byte{extraName: extra}); err != nil {
		t.Fatal(err)
	}
	b := busSweptBucket{store}
	t0 := time.Now()
	if r, err := sweepDeclarationFiles(ctx, b, named, t0); err != nil || r.Marked != 1 || r.Removed != 0 || r.Kept != 3 {
		t.Fatalf("first sweep: %+v %v", r, err)
	}
	if r, err := sweepDeclarationFiles(ctx, b, named, t0.Add(24*time.Hour-time.Minute)); err != nil || r.Removed != 0 {
		t.Fatalf("a minute short of a day: %+v %v", r, err)
	}
	if r, err := sweepDeclarationFiles(ctx, b, named, t0.Add(24*time.Hour+time.Second)); err != nil || r.Removed != 1 {
		t.Fatalf("a day after: %+v %v", r, err)
	}
	if infos, err := store.List(ctx); err != nil || len(infos) != 3 {
		t.Fatalf("the bucket holds %d objects (%v), want the 3 the declaration names", len(infos), err)
	}
}

// 1696/AC8 on a real bucket: the sweep removes an object only if it still carries the mark it judged it by. An object marked
// unnamed, then put again by a send (which erases the mark), is kept when the sweep deletes it by the old mark.
func TestTheSweepsDeleteLeavesAnObjectASendPutAgainAfterItWasMarked(t *testing.T) { // 1696/AC8
	s := testbus.Start(t)
	conn, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	bucket, err := openBucket(ctx, conn, "novox")
	if err != nil {
		t.Fatal(err)
	}
	name := strings.Repeat("a", 64)
	content := map[string][]byte{name: bytes.Repeat([]byte("y"), 100)}
	if err := putObjects(ctx, bucket, content); err != nil {
		t.Fatal(err)
	}
	b := busSweptBucket{bucket.store}
	judged := time.Now().Add(-25 * time.Hour).Truncate(time.Second)
	if err := b.MarkUnnamed(ctx, name, judged); err != nil {
		t.Fatal(err)
	}
	if _, marked, err := bucket.Has(ctx, name); err != nil || !marked {
		t.Fatalf("the object was not marked: %v %v", marked, err)
	}
	// The sweep has read the mark; a send now puts the object again, which erases it.
	if err := putObjects(ctx, bucket, content); err != nil {
		t.Fatal(err)
	}
	if have, marked, err := bucket.Has(ctx, name); err != nil || !have || marked {
		t.Fatalf("a send did not put the marked object again: have %v marked %v %v", have, marked, err)
	}
	gone, err := b.Delete(ctx, name, judged)
	if err != nil || gone {
		t.Fatalf("the sweep deleted by an old mark: gone %v, %v", gone, err)
	}
	if have, _, err := bucket.Has(ctx, name); err != nil || !have {
		t.Fatalf("the object a send relies on is gone: %v %v", have, err)
	}
	// And the mark still on it is honoured: marked again and deleted by that mark, it goes.
	if err := b.MarkUnnamed(ctx, name, judged); err != nil {
		t.Fatal(err)
	}
	if gone, err := b.Delete(ctx, name, judged); err != nil || !gone {
		t.Fatalf("an object still carrying the mark it was judged by is kept: gone %v, %v", gone, err)
	}
}

// 1696/AC9 on a real bus: the sweep of a node's bucket takes the node's lock first, so it waits for a send that has put its objects
// and not yet published the declaration that names them.
func TestTheSweepOfANodesBucketHoldsTheNodesLock(t *testing.T) { // 1696/AC9
	s := testbus.Start(t)
	conn, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	const node = "sweep-lock-node"
	if _, err := openBucket(ctx, conn, node); err != nil { // the bucket exists, or the sweep would return before the lock
		t.Fatal(err)
	}
	release := lockFilesOf(node) // a send is between its put and its publication
	done := make(chan struct{})
	go func() {
		_, _ = sweepNode(ctx, conn, node)
		close(done)
	}()
	select {
	case <-done:
		t.Fatal("the sweep judged the node's bucket while a send held it")
	case <-time.After(300 * time.Millisecond):
	}
	release()
	select {
	case <-done:
	case <-time.After(20 * time.Second):
		t.Fatal("the sweep did not go on once the send let go")
	}
}