mesh-sdk/go/provider/harness/moved_test.go

package harness

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"reflect"
	"sync"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/provider"
)

const (
	ProvisionName = "postgres-database"
	servedPort    = 5432
)

// postgresSpec is what a postgres-like provider says: a login per consumer, and the grant file's logins.
var postgresSpec = MovedSpec{Module: "postgres", Provision: ProvisionName,
	Offer: provider.Offer{Credential: provider.PerConsumer, Serves: map[string]any{"port": servedPort}},
	Login: func(e provider.Entry) string { return e.User },
	Noun:  "login", Made: "login", Backend: "the backend", WithGrants: true}

// bucketBus is the helper's bus over in-memory buckets.
type bucketBus struct {
	mu      sync.Mutex
	buckets map[string]map[string][]byte
}

func (b *bucketBus) Watch(_ context.Context, bucket string, handle func(string, []byte)) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	for k, v := range b.buckets[bucket] {
		handle(k, v)
	}
	return nil
}

func (b *bucketBus) Put(_ context.Context, bucket, key string, value []byte) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.buckets[bucket] == nil {
		b.buckets[bucket] = map[string][]byte{}
	}
	b.buckets[bucket][key] = append([]byte(nil), value...)
	return nil
}

func (b *bucketBus) Delete(_ context.Context, bucket, key string) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	delete(b.buckets[bucket], key)
	return nil
}

// bind is one consumer's assignment, whose plan names this machine's postgres as the provider and carries what the
// consumer contributes (issue 776).
func (b *bucketBus) bind(node, module string, contributes map[string]any) {
	b.bindNamed("postgres", node, module, contributes, nil)
}

// bindNamed is bind to a provider module, with the names the controller composed for the consumer (issue 792).
func (b *bucketBus) bindNamed(provides, node, module string, contributes map[string]any, names map[string]string) {
	plan, _ := json.Marshal(map[string]any{"node": node, "module": module, "provisioning": map[string]any{
		"provisions": []any{map[string]any{"provision": ProvisionName, "provider": "novox", "provider-module": provides,
			"contributes": contributes, "names": names}}}})
	if err := b.Put(context.Background(), provider.AssignmentsBucket, node+"."+module, plan); err != nil {
		panic(err)
	}
}

// consumersFile writes the file the machine's client fills: one credential per consumer.
func consumersFile(t *testing.T, bound map[[2]string]string) string {
	var list []provider.ConsumerCredential
	for k, password := range bound {
		list = append(list, provider.ConsumerCredential{Consumer: k[0], Module: k[1],
			Credential: provider.PairName("novox", ProvisionName, k[0], k[1], ""), Value: password})
	}
	raw, _ := json.Marshal(list)
	at := filepath.Join(t.TempDir(), "consumers.json")
	if err := os.WriteFile(at, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	return at
}

// movedWorld is postgres on the helper over an in-memory bus, with a recorder for a server and a clock to move.
type movedWorld struct {
	t    *testing.T
	a    *recorder
	bus  *bucketBus
	h    *Harness
	spec MovedSpec
	run  *Moved
	p    *provider.Provider
	now  time.Time
	file string
}

func newMovedWorld(t *testing.T, bound map[[2]string]string) *movedWorld {
	return newMovedWorldOf(t, bound, postgresSpec)
}

func newMovedWorldOf(t *testing.T, bound map[[2]string]string, spec MovedSpec) *movedWorld {
	w := &movedWorld{t: t, spec: spec, a: &recorder{}, bus: &bucketBus{buckets: map[string]map[string][]byte{}},
		now: time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC)}
	w.file = consumersFile(t, bound)
	w.h = &Harness{Resource: ProvisionName, Adapter: w.a, Node: "novox", Now: func() time.Time { return w.now },
		Log: func(string, ...any) {}}
	var bus provider.Bus
	w.run, w.p, bus, _ = NewMoved(w.file, w.h, spec)
	_ = bus
	if err := w.run.watch(ctx, w.bus); err != nil {
		t.Fatal(err)
	}
	return w
}

// pass is one helper pass and one retirement pass, five seconds after the last.
func (w *movedWorld) pass() []provider.Outcome {
	w.now = w.now.Add(5 * time.Second)
	// the plans as the bus holds them now, as the watch would hand them
	w.run.assignments = map[string][]byte{}
	for k, v := range w.bus.buckets[provider.AssignmentsBucket] {
		w.run.assignments[k] = v
	}
	// the helper reads its buckets from the in-memory bus
	w.p = provider.New(provider.Options{Node: "novox", Module: w.spec.Module, Bus: w.bus,
		Vault:  provider.ConsumersFile{Paths: []string{w.file}},
		Offers: map[string]provider.Offer{ProvisionName: w.spec.Offer},
		Make:   w.run.Make, Gone: w.run.Gone, Logf: func(string, ...any) {}})
	out, err := w.p.Pass(ctx)
	if err != nil {
		w.t.Fatal(err)
	}
	w.run.retire(ctx)
	return out
}

func bindings(n int) (map[[2]string]string, []string) {
	bound := map[[2]string]string{}
	var logins []string
	for i := 1; i <= n; i++ {
		module := fmt.Sprintf("app%d", i)
		bound[[2]string{"ace", module}] = "pw-" + module
		logins = append(logins, provider.ConsumerIdentity("ace", module))
	}
	return bound, logins
}

// The consumers' file makes the same calls on the adapter as the equivalent grant files do (the harness).
func TestTheConsumersFileMakesWhatTheGrantFilesMake(t *testing.T) {
	letta := provider.ConsumerIdentity("ace", "letta")
	plain := provider.ConsumerIdentity("ace", "plain")

	// Today: the grant files, through the harness.
	g := newWorld(t)
	g.give(
		map[string]any{"as": letta, "node": "ace", "values": map[string]any{"name": "letta", "extensions": []any{"vector"}}},
		map[string]any{"as": plain, "node": "ace", "values": map[string]any{"name": "plain"}},
	)
	g.h.Reconcile(ctx)

	// The consumers' file, through the provider helper.
	w := newMovedWorld(t, map[[2]string]string{{"ace", "letta"}: "pw-" + letta, {"ace", "plain"}: "pw-" + plain})
	w.bus.bind("ace", "letta", map[string]any{"name": "letta", "extensions": []any{"vector"}})
	w.bus.bind("ace", "plain", map[string]any{"name": "plain"})
	for _, o := range w.pass() {
		if o.Status != provider.Answered {
			t.Fatalf("%+v", o)
		}
	}
	byAs := map[string]Provision{}
	for _, c := range w.a.created {
		byAs[c.As] = c
	}
	if len(w.a.created) != 2 {
		t.Fatalf("created %d, want 2", len(w.a.created))
	}
	for _, want := range g.a.created {
		c := byAs[want.As]
		if c.As != want.As || c.Password != want.Password || c.Consumer != want.Consumer ||
			!reflect.DeepEqual(orEmpty(c.Values), orEmpty(want.Values)) {
			t.Fatalf("made %+v, the grant files make %+v", c, want)
		}
	}
	// Unchanged, and held: nothing is made again, within a minute or after it.
	w.a.held = true
	w.pass()
	w.now = w.now.Add(2 * time.Minute)
	w.pass()
	if len(w.a.created) != 2 {
		t.Fatalf("an unchanged pass created again: %d", len(w.a.created))
	}
	// No longer held: made again.
	w.a.held = false
	w.now = w.now.Add(2 * time.Minute)
	w.pass()
	if len(w.a.created) != 4 {
		t.Fatalf("not made again: %d", len(w.a.created))
	}
}

// One read without a consumer retires nothing: it takes the harness's stable passes and time, and then the entry
// goes and the login is retired, never dropped.
func TestAConsumerMissingFromOneReadRetiresNothing(t *testing.T) {
	bound, logins := bindings(2)
	w := newMovedWorld(t, bound)
	w.bus.bind("ace", "app1", nil)
	w.bus.bind("ace", "app2", nil)
	w.pass()

	// One read without app2.
	saved := w.bus.buckets[provider.AssignmentsBucket]["ace.app2"]
	w.bus.Delete(ctx, provider.AssignmentsBucket, "ace.app2")
	w.pass()
	if len(w.a.removed) != 0 {
		t.Fatalf("retired %v on one read", w.a.removed)
	}
	if _, kept := w.bus.buckets[provider.BucketOf("postgres")]["ace.app2.postgres-database"]; !kept {
		t.Fatal("the entry went on one read")
	}
	// It is back: nothing was lost, and the entry never left the bucket.
	w.bus.Put(ctx, provider.AssignmentsBucket, "ace.app2", saved)
	for i := 0; i < 8; i++ {
		w.pass()
	}
	w.now = w.now.Add(15 * time.Minute)
	w.pass()
	if len(w.a.removed) != 0 || len(w.bus.buckets[provider.BucketOf("postgres")]) != 2 {
		t.Fatalf("retired %v, entries %d", w.a.removed, len(w.bus.buckets[provider.BucketOf("postgres")]))
	}

	// Gone for good: not retired before StablePasses and StableFor, then retired and its entry removed.
	w.bus.Delete(ctx, provider.AssignmentsBucket, "ace.app2")
	for i := 0; i < 6; i++ {
		w.pass()
	}
	if len(w.a.removed) != 0 {
		t.Fatalf("retired %v before the time was out", w.a.removed)
	}
	w.now = w.now.Add(11 * time.Minute)
	w.pass()
	w.pass()
	if !reflect.DeepEqual(w.a.removed, []string{logins[1]}) || len(w.a.deleted) != 0 {
		t.Fatalf("retired %v, deleted %v", w.a.removed, w.a.deleted)
	}
	if _, kept := w.bus.buckets[provider.BucketOf("postgres")]["ace.app2.postgres-database"]; kept {
		t.Fatal("the entry stayed after the login was retired")
	}
}

// More than three at once waits for a person, as the harness's bound says.
func TestMoreThanTheBoundWaitsForAPerson(t *testing.T) {
	bound, _ := bindings(5)
	w := newMovedWorld(t, bound)
	for i := 1; i <= 5; i++ {
		w.bus.bind("ace", fmt.Sprintf("app%d", i), nil)
	}
	w.pass()
	for i := 1; i <= 4; i++ {
		w.bus.Delete(ctx, provider.AssignmentsBucket, fmt.Sprintf("ace.app%d", i))
	}
	for i := 0; i < 6; i++ {
		w.pass()
	}
	w.now = w.now.Add(11 * time.Minute)
	w.pass()
	w.pass()
	if len(w.a.removed) != 0 || w.h.r.waiting == nil {
		t.Fatalf("retired %v, waiting %v", w.a.removed, w.h.r.waiting)
	}
}

// A plan list that was never read is not nobody asking.
func TestNoPlansReadRetiresNothing(t *testing.T) {
	bound, _ := bindings(1)
	w := newMovedWorld(t, bound)
	w.bus.bind("ace", "app1", nil)
	w.pass()
	w.bus.Delete(ctx, provider.AssignmentsBucket, "ace.app1")
	for i := 0; i < 8; i++ {
		w.pass()
	}
	w.now = w.now.Add(20 * time.Minute)
	w.pass()
	if len(w.a.removed) != 0 {
		t.Fatalf("retired %v with no plans read", w.a.removed)
	}
}

// A consumer whose credential is not in the file yet is waited for: nothing is made for it.
func TestAConsumerWithoutACredentialIsWaitedFor(t *testing.T) {
	w := newMovedWorld(t, map[[2]string]string{})
	w.bus.bind("ace", "letta", map[string]any{"name": "letta"})
	outcomes := w.pass()
	if len(outcomes) != 1 || outcomes[0].Status != provider.Waiting || len(w.a.created) != 0 {
		t.Fatalf("%+v, created %v", outcomes, w.a.created)
	}
}

// Where the file is not there, postgres takes the grant directory, as before; where it is, the helper.
// The grant path restarts itself when the consumers file is first filled.
// A consumer whose module has a slug keeps the login its plan names, the slug first (issue 788): the adapter makes it
// under that login, and the retirement counts it as asked for, so a pass retires nothing.
func TestASluggedConsumerKeepsTheLoginItsPlanNames(t *testing.T) {
	agent := provider.ConsumerIdentity("ace", "agent")
	w := newMovedWorld(t, map[[2]string]string{{"ace", "claude-code"}: "pw-agent"})
	plan, _ := json.Marshal(map[string]any{"node": "ace", "module": "claude-code", "provisioning": map[string]any{
		"provisions": []any{map[string]any{"provision": ProvisionName, "provider": "novox",
			"provider-module": "postgres", "as": agent}}}})
	if err := w.bus.Put(ctx, provider.AssignmentsBucket, "ace.claude-code", plan); err != nil {
		t.Fatal(err)
	}
	for _, o := range w.pass() {
		if o.Status != provider.Answered {
			t.Fatalf("%+v", o)
		}
	}
	if len(w.a.created) != 1 || w.a.created[0].As != agent {
		t.Fatalf("made %+v, want one under %s", w.a.created, agent)
	}
	for i := 0; i < 8; i++ {
		w.pass()
	}
	w.now = w.now.Add(15 * time.Minute)
	w.pass()
	w.pass()
	if len(w.a.removed) != 0 {
		t.Fatalf("retired %v: the plan's login was counted unasked", w.a.removed)
	}
}

// A provider whose spec names the plan's `as` as the login and no grant file (the keycloak shape) makes the consumer
// under that name with the values its spec composes, and a grant file that cannot be read does not hold up its
// retirement pass.
func TestASpecWithoutGrantsMakesUnderThePlansAsAndIgnoresTheGrantFile(t *testing.T) {
	a := &recorder{}
	bus := &bucketBus{buckets: map[string]map[string][]byte{}}
	now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC)
	h := &Harness{Resource: ProvisionName, Adapter: a, Node: "novox", Now: func() time.Time { return now },
		Receives: filepath.Join(t.TempDir(), "unreadable-or-missing.json"), Log: func(string, ...any) {}}
	spec := MovedSpec{Module: "keycloak", Provision: ProvisionName,
		Offer: provider.Offer{Credential: provider.PerConsumer},
		Login: func(e provider.Entry) string { return e.As }, Noun: "client id", Made: "client", Backend: "Keycloak",
		Values: func(e provider.Entry) map[string]any { return map[string]any{"label": e.Contributes["label"]} }}
	file := consumersFile(t, map[[2]string]string{{"ace", "wiki"}: "pw-wiki"})
	run, _, _, err := NewMoved(file, h, spec)
	if err != nil {
		t.Fatal(err)
	}
	plan, _ := json.Marshal(map[string]any{"node": "ace", "module": "wiki", "provisioning": map[string]any{
		"provisions": []any{map[string]any{"provision": ProvisionName, "provider": "novox",
			"provider-module": "keycloak", "as": "wiki-client", "contributes": map[string]any{"label": "Wiki"}}}}})
	_ = bus.Put(ctx, provider.AssignmentsBucket, "ace.wiki", plan)
	if err := run.watch(ctx, bus); err != nil {
		t.Fatal(err)
	}
	p := provider.New(provider.Options{Node: "novox", Module: "keycloak", Bus: bus,
		Vault:  provider.ConsumersFile{Paths: []string{file}},
		Offers: map[string]provider.Offer{ProvisionName: spec.Offer}, Make: run.Make, Gone: run.Gone,
		Logf: func(string, ...any) {}})
	out, err := p.Pass(ctx)
	if err != nil || len(out) != 1 || out[0].Status != provider.Answered {
		t.Fatalf("%v %+v", err, out)
	}
	run.retire(ctx)
	if len(a.created) != 1 || a.created[0].As != "wiki-client" || a.created[0].Password != "pw-wiki" ||
		a.created[0].Values["label"] != "Wiki" {
		t.Fatalf("made %+v", a.created)
	}
}

const testIssuer = "https://id.example.org/realms/Novox"

var (
	grafanaContributes = map[string]any{"label": "grafana", "endpoint": "web", "callback": "/login/generic_oauth"}
	grafanaNames       = map[string]string{"name": "grafana.example.org", "internal-name": "grafana.home.internal"}
)

// keycloakSpec is what a keycloak-like provider says: the client id is the plan's `as`, the values are what the
// consumer contributes with the controller's names over them, and no grant file is consulted.
var keycloakSpec = MovedSpec{Module: "keycloak", Provision: ProvisionName,
	Offer: provider.Offer{Credential: provider.PerConsumer, Serves: map[string]any{"issuer": testIssuer,
		"token-path": "/protocol/openid-connect/token"}},
	Login: func(e provider.Entry) string { return e.As },
	Values: func(e provider.Entry) map[string]any {
		values := map[string]any{}
		for k, v := range e.Contributes {
			values[k] = v
		}
		for k, name := range e.Names {
			values[k] = name
		}
		return values
	},
	Noun: "client id", Made: "client", Backend: "Keycloak"}

// The consumers' file makes the same call on the adapter as the equivalent grant file does (the harness), for a
// provider of the keycloak shape: the same client id, secret and values; and the entry tells the consumer what the
// provider serves.
func TestTheConsumersFileMakesWhatTheGrantFileMakesForTheKeycloakShape(t *testing.T) {
	grafana := provider.ConsumerIdentity("home", "grafana")
	values := map[string]any{"name": grafanaNames["name"], "internal-name": grafanaNames["internal-name"]}
	for k, v := range grafanaContributes {
		values[k] = v
	}
	g := newWorld(t)
	g.give(map[string]any{"as": grafana, "node": "home", "values": values})
	g.h.Reconcile(ctx)

	w := newMovedWorldOf(t, map[[2]string]string{{"home", "grafana"}: "pw-" + grafana}, keycloakSpec)
	w.bus.bindNamed("keycloak", "home", "grafana", grafanaContributes, grafanaNames)
	for _, o := range w.pass() {
		if o.Status != provider.Answered {
			t.Fatalf("%+v", o)
		}
	}
	if len(w.a.created) != 1 || len(g.a.created) != 1 {
		t.Fatalf("created %d through the helper, %d through the harness, want 1 each", len(w.a.created), len(g.a.created))
	}
	c, want := w.a.created[0], g.a.created[0]
	if c.As != want.As || c.Password != want.Password || c.Consumer != want.Consumer ||
		!reflect.DeepEqual(orEmpty(c.Values), orEmpty(want.Values)) {
		t.Fatalf("made %+v, the grant file makes %+v", c, want)
	}
	var entry provider.Entry
	if err := json.Unmarshal(w.bus.buckets[provider.BucketOf("keycloak")]["home.grafana."+ProvisionName], &entry); err != nil {
		t.Fatal(err)
	}
	if entry.Values["issuer"] != testIssuer || entry.Values["token-path"] != "/protocol/openid-connect/token" {
		t.Fatalf("served %v", entry.Values)
	}
	w.a.held = true
	w.pass()
	w.now = w.now.Add(2 * time.Minute)
	w.pass()
	if len(w.a.created) != 1 {
		t.Fatalf("an unchanged pass created again: %d", len(w.a.created))
	}
	w.a.held = false
	w.now = w.now.Add(2 * time.Minute)
	w.pass()
	if len(w.a.created) != 2 {
		t.Fatalf("not made again: %d", len(w.a.created))
	}
}

// A consumer whose plan names its client id (a slugged module) keeps it, and the retirement counts it as asked for.
func TestAConsumerNamedByItsPlanKeepsItsClientId(t *testing.T) {
	agent := provider.ConsumerIdentity("home", "dashboard")
	w := newMovedWorldOf(t, map[[2]string]string{{"home", "grafana-2"}: "pw"}, keycloakSpec)
	plan, _ := json.Marshal(map[string]any{"node": "home", "module": "grafana-2", "provisioning": map[string]any{
		"provisions": []any{map[string]any{"provision": ProvisionName, "provider": "novox", "provider-module": "keycloak",
			"as": agent, "contributes": grafanaContributes, "names": grafanaNames}}}})
	if err := w.bus.Put(ctx, provider.AssignmentsBucket, "home.grafana-2", plan); err != nil {
		t.Fatal(err)
	}
	for _, o := range w.pass() {
		if o.Status != provider.Answered {
			t.Fatalf("%+v", o)
		}
	}
	if len(w.a.created) != 1 || w.a.created[0].As != agent {
		t.Fatalf("made %+v, want one under %s", w.a.created, agent)
	}
	for i := 0; i < 8; i++ {
		w.pass()
	}
	w.now = w.now.Add(15 * time.Minute)
	w.pass()
	w.pass()
	if len(w.a.removed) != 0 {
		t.Fatalf("retired %v: the plan's client id was counted unasked", w.a.removed)
	}
}