package broker

import (
	"context"
	"os"
	"testing"
	"time"

	paho "github.com/eclipse/paho.mqtt.golang"
)

// Runs against a real Mosquitto with the dynamic-security plugin, because a
// fake broker would only prove the JSON matches what I believe the plugin
// wants. Gated on the environment like the store's live tests:
//
//	DYNSEC_TEST_URL=tcp://127.0.0.1:51884 DYNSEC_TEST_USER=behavision-backend \
//	DYNSEC_TEST_PASS=pw-backend go test ./internal/broker -run Live -v
func TestLiveEnsureAndDeleteSite(t *testing.T) {
	url, user, pass := os.Getenv("DYNSEC_TEST_URL"), os.Getenv("DYNSEC_TEST_USER"), os.Getenv("DYNSEC_TEST_PASS")
	if url == "" {
		t.Skip("DYNSEC_TEST_URL not set")
	}
	d := New(url, user, pass, nil)
	defer d.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	site := "livetest.shop1"
	if err := d.EnsureSite(ctx, site, "first-pw"); err != nil {
		t.Fatalf("EnsureSite: %v", err)
	}
	// Idempotent, and the password becomes the one given THIS time.
	if err := d.EnsureSite(ctx, site, "second-pw"); err != nil {
		t.Fatalf("EnsureSite again: %v", err)
	}
	if err := canConnect(url, site, "first-pw"); err == nil {
		t.Fatalf("old password still accepted after re-ensure")
	}
	if err := canConnect(url, site, "second-pw"); err != nil {
		t.Fatalf("new password refused: %v", err)
	}
	// Own topic allowed, another site's refused. A denied publish at QoS 1
	// still gets a PUBACK, so this is observed through the backend's inbox.
	got := make(chan string, 4)
	backend := connect(t, url, user, pass)
	defer backend.Disconnect(100)
	backend.Subscribe("bv/#", 1, func(_ paho.Client, m paho.Message) { got <- m.Topic() }).Wait()
	shop := connect(t, url, site, "second-pw")
	defer shop.Disconnect(100)
	shop.Publish("bv/other.shop/visit", 1, false, "leak").Wait()
	shop.Publish("bv/"+site+"/visit", 1, false, "ok").Wait()
	select {
	case topic := <-got:
		if topic != "bv/"+site+"/visit" {
			t.Fatalf("first delivered topic was %s", topic)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("own-topic publish never arrived")
	}
	select {
	case topic := <-got:
		t.Fatalf("unexpected second delivery: %s", topic)
	case <-time.After(1500 * time.Millisecond):
	}

	if err := d.DeleteSite(ctx, site); err != nil {
		t.Fatalf("DeleteSite: %v", err)
	}
	if err := d.DeleteSite(ctx, site); err != nil {
		t.Fatalf("DeleteSite twice: %v", err)
	}
	if err := canConnect(url, site, "second-pw"); err == nil {
		t.Fatal("deleted site can still connect")
	}
}

func connect(t *testing.T, url, user, pass string) paho.Client {
	t.Helper()
	c := paho.NewClient(paho.NewClientOptions().AddBroker(url).SetUsername(user).SetPassword(pass).SetConnectTimeout(5 * time.Second))
	tok := c.Connect()
	tok.Wait()
	if tok.Error() != nil {
		t.Fatalf("connect as %s: %v", user, tok.Error())
	}
	return c
}

func canConnect(url, user, pass string) error {
	c := paho.NewClient(paho.NewClientOptions().AddBroker(url).SetUsername(user).SetPassword(pass).SetConnectTimeout(5 * time.Second))
	tok := c.Connect()
	tok.Wait()
	if tok.Error() == nil {
		c.Disconnect(50)
	}
	return tok.Error()
}