mesh-catalog/modules/postgres/cmd/postgres-provider/moved_test.go

package main

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

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

// 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) {
	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": "postgres",
			"contributes": contributes}}}})
	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
	run  *movedRun
	p    *provider.Provider
	now  time.Time
	file string
}

func newMovedWorld(t *testing.T, bound map[[2]string]string) *movedWorld {
	w := &movedWorld{t: t, 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, _ = moved(w.file, w.h)
	_ = 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: "postgres", Bus: w.bus,
		Vault: provider.ConsumersFile{Paths: []string{w.file}},
		Offers: map[string]provider.Offer{ProvisionName: {Credential: provider.PerConsumer,
			Serves: map[string]any{"port": servedPort}}},
		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.
func TestTheGrantPathIsTakenWhereTheConsumersFileIsAbsent(t *testing.T) {
	dir := t.TempDir()
	at := filepath.Join(dir, "consumers.json")
	env := func(k string) string {
		if k == "MESH_CONSUMERS_FILE" {
			return at
		}
		return ""
	}
	if got := handedConsumersFile(env, os.Stat); got != "" {
		t.Fatalf("took %q for a file that is not there", got)
	}
	if got := handedConsumersFile(func(string) string { return "" }, os.Stat); got != "" {
		t.Fatalf("took %q with no environment", got)
	}
	if err := os.WriteFile(at, []byte("[]"), 0o600); err != nil {
		t.Fatal(err)
	}
	if got := handedConsumersFile(env, os.Stat); got != at {
		t.Fatalf("took %q, want %q", got, at)
	}
}

// The grant path restarts itself when the consumers file is first filled.
func TestTheGrantPathLeavesWhenTheConsumersFileIsFilled(t *testing.T) {
	at := filepath.Join(t.TempDir(), "consumers.json")
	done := make(chan struct{})
	go exitWhenFilled(at, os.Stat, time.Millisecond, func() { close(done) })
	select {
	case <-done:
		t.Fatal("left before the file was there")
	case <-time.After(20 * time.Millisecond):
	}
	if err := os.WriteFile(at, []byte("[]"), 0o600); err != nil {
		t.Fatal(err)
	}
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("did not leave when the file was 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)
	}
}