package broker

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"golang.org/x/crypto/bcrypt"
)

// **Only mesh-config's account writes mesh-config's buckets** (novox/hq ADR 0348 §9 and How it is checked, design
// 57 K3, issue 662): on a real server whose users are composed as the controller composes them. mesh-config runs on
// an account of its own (`runs-as`, ADR 0259 §8), so the machine's runtime does not carry it and its own principal
// alone is granted the writes of its four buckets. A put from any other account is refused by the server: a
// module reading the resolved values through `reads`, the runtime of the machine mesh-config runs on, and the
// controller.
func TestNatsOnlyMeshConfigsAccountPutsToItsBuckets(t *testing.T) {
	hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
	if err != nil {
		t.Fatal(err)
	}
	states := []string{"values", "resolved", "predefined", "proposals"}
	var buckets []Bucket
	for _, s := range states {
		buckets = append(buckets, Bucket{Module: "mesh-config", Name: s})
	}
	seat := Seat{Name: "mesh-config", Scope: "mesh", Serves: []string{"config"}}
	users, err := Users(Records{Nodes: []string{"novox", "g14"}, Assigned: map[string][]Declared{
		"novox": {
			{Module: RuntimeModule},
			{Module: "mesh-config", RunsAs: "mesh-config", Holds: []Seat{seat}, State: buckets,
				Emits: []string{"config-changed"},
				Reads: []string{"mesh-controller.assignments", "mesh-controller.declared-config", "mesh-controller.predefined"}},
			{Module: "mesh-issues", State: []Bucket{{Module: "mesh-issues", Name: "issues"}}},
		},
		"g14": {{Module: "dunst", Reads: []string{"mesh-config.resolved"}}},
	}})
	if err != nil {
		t.Fatal(err)
	}
	for i := range users {
		users[i].PasswordHash = string(hash)
	}
	accounts, err := ComposeAccounts(users)
	if err != nil {
		t.Fatal(err)
	}
	conf := filepath.Join(t.TempDir(), "bus.conf")
	if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+
		`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
		t.Fatal(err)
	}
	opts, err := server.ProcessConfigFile(conf)
	if err != nil {
		t.Fatalf("the composed accounts do not parse: %v", err)
	}
	opts.NoLog, opts.NoSigs = true, true
	s, err := server.NewServer(opts)
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(s.Shutdown)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	connect := func(user string) (*nats.Conn, chan error) {
		t.Helper()
		refused := make(chan error, 8)
		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
			nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
		if err != nil {
			t.Fatalf("%s cannot connect: %v", user, err)
		}
		t.Cleanup(nc.Close)
		return nc, refused
	}
	// The controller makes the buckets, as it does from a manifest's `state`.
	controller, controllerRefused := connect(ControllerName)
	cjs, err := jetstream.New(controller)
	if err != nil {
		t.Fatal(err)
	}
	for _, b := range buckets {
		if _, err := cjs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b.Bucket()}); err != nil {
			t.Fatalf("the controller cannot make %s: %v", b.Bucket(), err)
		}
	}

	owner, _ := connect("novox.mesh-config")
	ojs, err := jetstream.New(owner)
	if err != nil {
		t.Fatal(err)
	}
	for _, b := range buckets {
		kv, err := ojs.KeyValue(ctx, b.Bucket())
		if err != nil {
			t.Fatalf("mesh-config cannot bind its own %s: %v", b.Bucket(), err)
		}
		if _, err := kv.Put(ctx, "mesh.font-size", []byte(`{"value":12}`)); err != nil {
			t.Errorf("mesh-config cannot write its own %s: %v", b.Bucket(), err)
		}
	}

	// Every other account: a publish on the bucket's subject is refused by the server, and the bucket holds
	// what mesh-config wrote.
	reader, readerRefused := connect("g14.dunst")
	runtime, runtimeRefused := connect("novox." + RuntimeModule)
	for _, other := range []struct {
		name    string
		nc      *nats.Conn
		refused chan error
	}{{"a module reading the resolved values", reader, readerRefused},
		{"the runtime of mesh-config's machine", runtime, runtimeRefused},
		{"the controller", controller, controllerRefused}} {
		for _, b := range buckets {
			if err := other.nc.Publish("$KV."+b.Bucket()+".mesh.font-size", []byte(`{"value":99}`)); err != nil {
				t.Fatal(err)
			}
			_ = other.nc.Flush()
			select {
			case err := <-other.refused:
				if !errors.Is(err, nats.ErrPermissionViolation) {
					t.Errorf("%s putting to %s: the server said %v, want a permissions violation", other.name, b.Bucket(), err)
				}
			case <-time.After(3 * time.Second):
				t.Errorf("the server did not refuse %s a put to %s", other.name, b.Bucket())
			}
		}
	}
	// The reader still reads what it was granted.
	rjs, err := jetstream.New(reader)
	if err != nil {
		t.Fatal(err)
	}
	resolved, err := rjs.KeyValue(ctx, "mesh-config_resolved")
	if err != nil {
		t.Fatalf("a module reading mesh-config.resolved cannot bind it: %v", err)
	}
	e, err := resolved.Get(ctx, "mesh.font-size")
	if err != nil || string(e.Value()) != `{"value":12}` {
		t.Errorf("the resolved value reads %v (%v), want mesh-config's own write", e, err)
	}
	for _, b := range buckets {
		kv, err := ojs.KeyValue(ctx, b.Bucket())
		if err != nil {
			t.Fatal(err)
		}
		if e, err := kv.Get(ctx, "mesh.font-size"); err != nil || string(e.Value()) != `{"value":12}` {
			t.Errorf("%s holds %v (%v) after the refused puts", b.Bucket(), e, err)
		}
	}
}