//go:build integration

package broker

import (
	"context"
	"errors"
	"time"

	"testing"

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

func TestABucketIsAssertedInPlace(t *testing.T) {
	js := aLiveBus(t)
	b := Bucket{Module: "statetest", Name: "servers"}
	if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
		t.Fatalf("a real server refused a module's bucket: %v", err)
	}
	kv, err := js.Context().KeyValue(b.Bucket())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := kv.Put("all.one", []byte(`{"kept":true}`)); err != nil {
		t.Fatal(err)
	}
	b.History = 3
	if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
		t.Fatalf("asserting the bucket again failed, so a restart would: %v", err)
	}
	got, err := kv.Get("all.one")
	if err != nil || string(got.Value()) != `{"kept":true}` {
		t.Fatalf("asserting again lost what the bucket held: %v %v", got, err)
	}
	status, err := kv.Status()
	if err != nil {
		t.Fatal(err)
	}
	if status.History() != 3 {
		t.Fatalf("history is %d, the owner declared 3", status.History())
	}
	if s, ok := status.(*nats.KeyValueBucketStatus); ok {
		if c := s.StreamInfo().Config; c.MaxMsgSize != StateMaxValueBytes || c.MaxBytes != StateMaxBytes {
			t.Fatalf("the mesh's caps are not on the bucket: value %d, bucket %d", c.MaxMsgSize, c.MaxBytes)
		}
	}
}

func TestABucketIsAssertedOverAHeldConnection(t *testing.T) {
	js := aLiveBus(t)
	held := OnConn(js.Conn())
	if _, err := RaiseBuckets(held, []Bucket{{Module: "statetest", Name: "held"}}); err != nil {
		t.Fatalf("asserting over a held connection failed: %v", err)
	}
	if _, err := js.Context().KeyValue("statetest_held"); err != nil {
		t.Fatalf("the bucket is not there: %v", err)
	}
}

// 1154/AC4: on the pinned server a bucket declared with key-ttl takes a put with a time to live of a second, the key is
// gone after it and the server has written a delete marker; and the bucket asserted again without key-ttl is not refused.
func TestAKeyTTLBucketExpiresAKeyAndIsAssertedAgain(t *testing.T) { // 1154/AC4
	js := aLiveBus(t)
	b := Bucket{Module: "statetest", Name: "pulse", KeyTTL: true}
	if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
		t.Fatalf("a real server refused a key-ttl bucket: %v", err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	kv, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	bucket, err := kv.KeyValue(ctx, b.Bucket())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := bucket.Create(ctx, "presence.alpha", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil {
		t.Fatalf("the bucket refused a key with a time to live: %v", err)
	}
	deadline := time.Now().Add(10 * time.Second)
	for {
		if _, err := bucket.Get(ctx, "presence.alpha"); errors.Is(err, jetstream.ErrKeyNotFound) {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the key is still there after its time to live")
		}
		time.Sleep(200 * time.Millisecond)
	}
	hist, err := bucket.History(ctx, "presence.alpha")
	if err != nil {
		t.Fatal(err)
	}
	if last := hist[len(hist)-1]; last.Operation() == jetstream.KeyValuePut {
		t.Fatalf("no delete marker was left: last entry is %v", last.Operation())
	}
	// The owner no longer says key-ttl: the server cannot turn the age off, so the bucket keeps it and is not refused.
	b.KeyTTL = false
	if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
		t.Fatalf("asserting the bucket without key-ttl was refused, so a restart would fail: %v", err)
	}
}