From b8ce072c9f8a78a19b84ca141da4f37ec71b7068 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 00:37:54 +0100 Subject: [PATCH] M-16: the manager whole on a real bus and today's database on a real postgres (issue 1409) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The built binary played as its runtime plays it, its kept data in a real nats-server's buckets and streams, today's rows in a real postgres: a start stopped at a table says carry-over-failed and acts on nothing, the next carries every row and a further one writes nothing; an unreachable database calls no vendor; a reading at the threshold moves four consumers in one act and no-room is raised and cleared on the operator channel's subject, read through its client; a new login is adopted, listed and a target. Builds: design 39 bus-and-switch §9, §11, §12, §13, §23, §26, §28, §29, §46, B41 Claude-Session: https://claude.ai/code/session_01NcbgfHb9d4T9nq5Q6WqdiJ --- .../whole_integration_test.go | 1143 +++++++++++++++++ modules/claude-licence-manager/go.mod | 8 +- modules/claude-licence-manager/go.sum | 46 +- 3 files changed, 1179 insertions(+), 18 deletions(-) create mode 100644 modules/claude-licence-manager/cmd/claude-licence-manager/whole_integration_test.go diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/whole_integration_test.go b/modules/claude-licence-manager/cmd/claude-licence-manager/whole_integration_test.go new file mode 100644 index 000000000..103034053 --- /dev/null +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/whole_integration_test.go @@ -0,0 +1,1143 @@ +//go:build integration + +package main + +// The manager whole against a real bus (design 39 bus-and-switch §9, §11 to §13, §23, §26, §28, §29, §46, B41): the +// built binary, played over its real stdin and stdout as the node's tool runtime plays it (tools/call with the caller in +// _meta; mesh/state.* from a real nats-server's key-value buckets, mesh/log.* from the logs' real streams, mesh/publish +// and mesh/seat.publish onto their subjects), and today's database on a real postgres (MESH_TEST_POSTGRES, or a private +// server started here from the machine's initdb on a socket of its own, listening on no port). What the manager asks of +// other parts is played beside it: the vault's licence-put and get, a node's claude_code_grant and its holdings-changed, +// the operator channel's tell, and the vendor's token and usage endpoints. A condition is taken off the operator +// channel's subject and decoded by its client, the contract the router reads (B41); the router itself is not run. + +import ( + "bufio" + "context" + "crypto/rand" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "os/exec" + "path/filepath" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "golang.org/x/crypto/nacl/box" + + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +func TestMain(m *testing.M) { + code := m.Run() + stopPostgres() + os.Exit(code) +} + +// ---- today's database on a real postgres --------------------------------------------------------------------- + +var ( + pgOnce sync.Once + pgBase string + pgDir string + pgWhy string + pgCount int + pgMu sync.Mutex +) + +func startPostgres() { + if u := os.Getenv("MESH_TEST_POSTGRES"); u != "" { + pgBase = u + return + } + initdb, err := exec.LookPath("initdb") + if err != nil { + pgWhy = "MESH_TEST_POSTGRES is unset and the machine has no initdb" + return + } + dir, err := os.MkdirTemp("", "licpg") + if err != nil { + pgWhy = err.Error() + return + } + pgDir = dir + if out, err := exec.Command(initdb, "-D", dir+"/data", "-U", "postgres", "-A", "trust", "-E", "UTF8", "--no-locale").CombinedOutput(); err != nil { + pgWhy = fmt.Sprintf("initdb: %v: %s", err, out) + return + } + // A socket in its own directory and no port: nothing on this machine reaches it but these tests. + if out, err := exec.Command(filepath.Join(filepath.Dir(initdb), "pg_ctl"), "-D", dir+"/data", "-l", dir+"/log", "-w", + "-o", "-k "+dir+" -c listen_addresses='' -F", "start").CombinedOutput(); err != nil { + pgWhy = fmt.Sprintf("pg_ctl start: %v: %s", err, out) + return + } + pgBase = "postgres://postgres@/postgres?host=" + dir +} + +func stopPostgres() { + if pgDir == "" { + return + } + if initdb, err := exec.LookPath("initdb"); err == nil { + _ = exec.Command(filepath.Join(filepath.Dir(initdb), "pg_ctl"), "-D", pgDir+"/data", "-m", "immediate", "stop").Run() + } + _ = os.RemoveAll(pgDir) +} + +// freshDatabase is an empty database of its own for one test, or a skip saying why there is none. +func freshDatabase(t *testing.T) (string, *pgx.Conn) { + t.Helper() + pgOnce.Do(startPostgres) + if pgBase == "" { + t.Skipf("NOT TESTED ON POSTGRES: %s", pgWhy) + } + pgMu.Lock() + pgCount++ + name := fmt.Sprintf("licmgr_%d_%d", os.Getpid(), pgCount) + pgMu.Unlock() + ctx := context.Background() + admin, err := pgx.Connect(ctx, pgBase) + if err != nil { + t.Fatal(err) + } + defer admin.Close(ctx) + if _, err := admin.Exec(ctx, "create database "+name); err != nil { + t.Fatal(err) + } + u, err := url.Parse(pgBase) + if err != nil { + t.Fatal(err) + } + u.Path = "/" + name + conn, err := pgx.Connect(ctx, u.String()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close(ctx) }) + return u.String(), conn +} + +// todaysSchema is the schema trunk's manager made (its PgStore's Schema): what the carry-over reads. +var todaysSchema = []string{ + `create table licence (name text primary key, kind text not null check (kind in ('subscription', 'api-key')), + account_uuid text unique, email text, organization_uuid text, sealed text, refresh_fingerprint text, + access_expires_at bigint, refresh_expires_at bigint, failures integer not null default 0, notified_at bigint, + adopted_at bigint not null, rotated_at bigint)`, + `create sequence binding_generation`, + `create table binding (consumer text primary key, licence text not null references licence(name), generation bigint not null)`, + `create table lease (key text primary key, holder text not null, until timestamptz not null)`, + `create table offered (fingerprint text primary key, node text not null, account_uuid text, outcome text not null, + why text not null, at timestamptz not null default now())`, + `create table usage (licence text not null, at bigint not null, reading jsonb not null, raw jsonb not null)`, + `create table audit (at timestamptz not null default now(), what text not null, detail jsonb not null)`, +} + +const ( + licenceA = "first@example.org" + licenceB = "second@example.org" + grantKey = "a key the vault made" +) + +// consumers are the four nodes bound to licenceA in today's rows. +var consumers = []string{"anchor", "home-server", "laptop", "workstation"} + +// fillTodaysRows writes a copy of the shape of today's rows: two subscriptions, four consumers bound to the first, a +// lease, two offers (one older than the offered state's 90 days), three readings (one older than 35 days) and four +// audit rows (one of a kind the log does not name). +func fillTodaysRows(t *testing.T, conn *pgx.Conn) { + t.Helper() + ctx := context.Background() + for _, s := range todaysSchema { + if _, err := conn.Exec(ctx, s); err != nil { + t.Fatal(err) + } + } + crypt, _ := NewCrypt(grantKey) + now := time.Now() + far := now.Add(30 * 24 * time.Hour).UnixMilli() + for i, l := range []struct{ name, account, refresh, access string }{ + {licenceA, "acct-a", "rt-a", "at-a"}, {licenceB, "acct-b", "rt-b", "at-b"}} { + grant, _ := json.Marshal(FullGrant{AccessToken: l.access, RefreshToken: l.refresh, ExpiresAt: far}) + if _, err := conn.Exec(ctx, `insert into licence (name, kind, account_uuid, email, sealed, refresh_fingerprint, + access_expires_at, refresh_expires_at, adopted_at, rotated_at) values ($1,'subscription',$2,$1,$3,$4,$5,$5,$6,$7)`, + l.name, l.account, sealAtRest(t, crypt, string(grant)), Fingerprint(l.refresh), far, int64(i+1), now.UnixMilli()); err != nil { + t.Fatal(err) + } + } + for i, c := range consumers { + if _, err := conn.Exec(ctx, `insert into binding values ($1, $2, $3)`, c, licenceA, 1013+i); err != nil { + t.Fatal(err) + } + } + reading := `{"sessionPct": 42, "weeklyPct": 10}` + for _, q := range []struct { + sql string + args []any + }{ + {`insert into lease values ('licence:x', 'novox:1', now())`, nil}, + {`insert into offered values ('sha256:aa', 'laptop', 'acct-a', 'adopted', 'refreshed', $1)`, []any{now.Add(-24 * time.Hour)}}, + {`insert into offered values ('sha256:bb', 'anchor', null, 'dead', '400', $1)`, []any{now.Add(-100 * 24 * time.Hour)}}, + {`insert into usage values ($1, $2, $3, '{}')`, []any{licenceA, now.Add(-40 * 24 * time.Hour).UnixMilli(), reading}}, + {`insert into usage values ($1, $2, $3, '{"five_hour": {"utilization": 42}}')`, []any{licenceA, now.Add(-2 * time.Hour).UnixMilli(), reading}}, + {`insert into usage values ($1, $2, $3, '{}')`, []any{licenceB, now.Add(-time.Hour).UnixMilli(), reading}}, + {`insert into audit values ($1, 'adopted', '{"licence": "first@example.org", "node": "laptop"}')`, []any{now.Add(-72 * time.Hour)}}, + {`insert into audit values ($1, 'bound', '{"consumer": "laptop"}')`, []any{now.Add(-71 * time.Hour)}}, + {`insert into audit values ($1, 'switched', '{"consumer": "anchor"}')`, []any{now.Add(-70 * time.Hour)}}, + {`insert into audit values ($1, 'grant-moved', '{"licence": "second@example.org"}')`, []any{now.Add(-69 * time.Hour)}}, + } { + if _, err := conn.Exec(ctx, q.sql, q.args...); err != nil { + t.Fatal(err) + } + } +} + +// ---- the vendor ---------------------------------------------------------------------------------------------- + +// vendor answers the token endpoint by refresh token and the usage endpoint by access token, and counts its calls. +type vendor struct { + mu sync.Mutex + calls int + logins map[string][2]string // refresh token: the next access token and the account + usage map[string][2]float64 // access token: session and weekly percentages + server *httptest.Server +} + +func newVendor(t *testing.T) *vendor { + v := &vendor{logins: map[string][2]string{}, usage: map[string][2]float64{}} + v.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + v.mu.Lock() + defer v.mu.Unlock() + v.calls++ + if r.URL.Path == "/token" { + _ = r.ParseForm() + l, ok := v.logins[r.PostForm.Get("refresh_token")] + if !ok { + http.Error(w, `{"error":"invalid_grant"}`, http.StatusBadRequest) + return + } + in := int64(8 * 3600) + _ = json.NewEncoder(w).Encode(map[string]any{"access_token": l[0], "expires_in": in, + "account": map[string]string{"uuid": l[1]}}) + return + } + u, ok := v.usage[strings.TrimPrefix(r.Header.Get("authorization"), "Bearer ")] + if !ok { + http.Error(w, `{"error":"unknown token"}`, http.StatusUnauthorized) + return + } + resets := time.Now().Add(3 * time.Hour).UTC().Format(time.RFC3339) + _ = json.NewEncoder(w).Encode(map[string]any{"five_hour": map[string]any{"utilization": u[0], "resets_at": resets}, + "seven_day": map[string]any{"utilization": u[1], "resets_at": resets}}) + })) + t.Cleanup(v.server.Close) + return v +} + +func (v *vendor) set(access string, session, weekly float64) { + v.mu.Lock() + v.usage[access] = [2]float64{session, weekly} + v.mu.Unlock() +} + +func (v *vendor) called() int { + v.mu.Lock() + defer v.mu.Unlock() + return v.calls +} + +// ---- the runtime ----------------------------------------------------------------------------------------------- + +type wire struct { + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *struct { + Message string `json:"message"` + } `json:"error,omitempty"` +} + +// said is one event the manager said, or one call it asked of another part. +type said struct { + key string + body json.RawMessage +} + +// mesh is the bus, the state directory, the database and the parts beside the manager, kept across its starts. +type mesh struct { + t *testing.T + ctx context.Context + nc *nats.Conn + js jetstream.JetStream + bin string + dir string + dbURL string + vendor *vendor + + mu sync.Mutex + said []said + news chan struct{} + secrets map[string]string // the vault: a secret's name and its value + calls map[string][32]byte // the vault: licence-put's open calls and the private key made for each + callArgs map[string]map[string]any + holdings []Holdings // the agent modules: what each node says it holds when asked + logins map[string]FullGrant // the agent modules: the login each node hands over, by node + conditions []oc.Condition +} + +const busStates = "licences bindings leases offered carried conditions" +const busLogs = "usage audit" + +func newWholeMesh(t *testing.T, dbURL string) *mesh { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + dir := t.TempDir() + bin := filepath.Join(dir, "claude-licence-manager") + build := exec.Command("go", "build", "-o", bin, ".") + build.Stderr = os.Stderr + if err := build.Run(); err != nil { + t.Fatalf("the manager does not build: %v", err) + } + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + // What the bus module creates from the manifest's templates (§1 to §9) before the manager's want moves (§50). + for _, s := range strings.Fields(busStates) { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: Publisher + "_" + s, History: 8}); err != nil { + t.Fatal(err) + } + } + for _, l := range strings.Fields(busLogs) { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "LOG_" + Publisher + "_" + l, + Subjects: []string{"mesh.log." + Publisher + "." + l + ".>"}}); err != nil { + t.Fatal(err) + } + } + _ = os.MkdirAll(filepath.Join(dir, "state"), 0o700) + for f, v := range map[string]string{"grant.key": grantKey, "database.url": dbURL, "settings.json": "{}"} { + if err := os.WriteFile(filepath.Join(dir, f), []byte(v), 0o600); err != nil { + t.Fatal(err) + } + } + m := &mesh{t: t, ctx: ctx, nc: nc, js: js, bin: bin, dir: dir, dbURL: dbURL, vendor: newVendor(t), + news: make(chan struct{}, 1), secrets: map[string]string{}, calls: map[string][32]byte{}, + callArgs: map[string]map[string]any{}, logins: map[string]FullGrant{}} + sub, err := nc.Subscribe("mesh.seat.operator-channel.accept.condition.>", func(msg *nats.Msg) { + c, err := occ.DecodeCondition(msg.Data) + if err != nil { + t.Errorf("a condition the operator channel's client does not decode: %v: %s", err, msg.Data) + return + } + if want, _ := occ.ConditionSubject(Publisher); msg.Subject != want { + t.Errorf("a condition said on %s, not %s", msg.Subject, want) + } + m.mu.Lock() + m.conditions = append(m.conditions, c) + m.mu.Unlock() + m.notify() + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + return m +} + +func (m *mesh) notify() { + select { + case m.news <- struct{}{}: + default: + } +} + +// until waits for what to hold, woken by each event, call and condition the manager says, never by a timer of its own. +func (m *mesh) until(what string, ok func() bool) { + m.t.Helper() + deadline := time.After(60 * time.Second) + for { + m.mu.Lock() + done := ok() + m.mu.Unlock() + if done { + return + } + select { + case <-m.news: + case <-deadline: + m.t.Fatalf("never: %s", what) + } + } +} + +// count is how many of what was said have key; under mu. +func (m *mesh) count(key string) int { + n := 0 + for _, s := range m.said { + if s.key == key { + n++ + } + } + return n +} + +// of is the bodies said under key; under mu. +func (m *mesh) of(key string) []map[string]any { + var out []map[string]any + for _, s := range m.said { + if s.key == key { + var b map[string]any + _ = json.Unmarshal(s.body, &b) + out = append(out, b) + } + } + return out +} + +// conditionsOf is each condition said with key, in order; under mu. +func (m *mesh) conditionsOf(key string) []oc.Condition { + var out []oc.Condition + for _, c := range m.conditions { + if c.Key == key { + out = append(out, c) + } + } + return out +} + +// started is one start of the binary. +type started struct { + *mesh + in io.WriteCloser + cmd *exec.Cmd + next int + pending map[string]chan wire + pmu sync.Mutex + subs []string +} + +// start starts the manager as its runtime launches it, with the module's words (ADR 0192). +func (m *mesh) start() *started { + m.t.Helper() + cmd := exec.CommandContext(m.ctx, m.bin) + cmd.Env = append(os.Environ(), "MESH_LICENCE_STATE="+filepath.Join(m.dir, "state"), + "MESH_LICENCE_KEY_FILE="+filepath.Join(m.dir, "grant.key"), "DATABASE_URL_FILE="+filepath.Join(m.dir, "database.url"), + "MESH_LICENCE_SETTINGS="+filepath.Join(m.dir, "settings.json"), "MESH_NODE=novox", "MESH_MODULE="+Publisher, + "MESH_ANTHROPIC_TOKEN_ENDPOINT="+m.vendor.server.URL+"/token", "MESH_ANTHROPIC_USAGE_ENDPOINT="+m.vendor.server.URL+"/usage") + in, _ := cmd.StdinPipe() + out, _ := cmd.StdoutPipe() + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + m.t.Fatal(err) + } + p := &started{mesh: m, in: in, cmd: cmd, pending: map[string]chan wire{}} + m.t.Cleanup(p.stop) + go p.serve(out) + return p +} + +// stop ends this start, as the runtime does when the module moves. +func (p *started) stop() { + _ = p.in.Close() + _ = p.cmd.Process.Kill() + _ = p.cmd.Wait() +} + +func (p *started) send(v map[string]any) { + raw, _ := json.Marshal(v) + p.pmu.Lock() + defer p.pmu.Unlock() + _, _ = p.in.Write(append(raw, '\n')) +} + +func (p *started) answer(id json.RawMessage, result any, err error) { + if err != nil { + code := -32000 + if errors.Is(err, jetstream.ErrKeyExists) { + code = -32010 // a lost compare-and-set, as the runtime says it + } + p.send(map[string]any{"id": id, "error": map[string]any{"code": code, "message": err.Error()}}) + return + } + p.send(map[string]any{"id": id, "result": result}) +} + +func (p *started) serve(out io.Reader) { + scanner := bufio.NewScanner(out) + scanner.Buffer(make([]byte, 1<<20), 1<<24) + for scanner.Scan() { + var w wire + if json.Unmarshal(scanner.Bytes(), &w) != nil { + continue + } + if w.Method == "" { + var key string + _ = json.Unmarshal(w.ID, &key) + p.pmu.Lock() + ch := p.pending[key] + delete(p.pending, key) + p.pmu.Unlock() + if ch != nil { + ch <- w + } + continue + } + go p.handle(w) + } +} + +func (p *started) kv(state string) (jetstream.KeyValue, error) { + return p.js.KeyValue(p.ctx, Publisher+"_"+state) +} + +func (p *started) handle(w wire) { + var a struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Revision uint64 `json:"revision"` + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + Pattern string `json:"pattern"` + } + _ = json.Unmarshal(w.Params, &a) + if strings.HasPrefix(w.Method, "mesh/log.") { + p.log(w) + return + } + switch w.Method { + case "mesh/state.get": + kv, err := p.kv(a.State) + if err != nil { + p.answer(w.ID, nil, err) + return + } + e, err := kv.Get(p.ctx, a.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + p.answer(w.ID, nil, nil) + return + } + if err != nil { + p.answer(w.ID, nil, err) + return + } + p.answer(w.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision(), "at": e.Created()}, nil) + case "mesh/state.put", "mesh/state.create", "mesh/state.update": + kv, err := p.kv(a.State) + var rev uint64 + if err == nil { + // A key's own time to live is the bus's; no test here lives as long as one (a lease's two minutes). + switch w.Method { + case "mesh/state.put": + rev, err = kv.Put(p.ctx, a.Key, a.Value) + case "mesh/state.create": + rev, err = kv.Create(p.ctx, a.Key, a.Value) + default: + rev, err = kv.Update(p.ctx, a.Key, a.Value, a.Revision) + if err != nil { + err = fmt.Errorf("%w: %v", jetstream.ErrKeyExists, err) + } + } + } + p.answer(w.ID, map[string]any{"revision": rev}, err) + case "mesh/state.keys": + kv, err := p.kv(a.State) + if err != nil { + p.answer(w.ID, nil, err) + return + } + keys, err := kv.Keys(p.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + keys, err = []string{}, nil + } + sort.Strings(keys) + p.answer(w.ID, keys, err) + case "mesh/state.delete": + kv, err := p.kv(a.State) + if err == nil { + err = kv.Delete(p.ctx, a.Key) + } + p.answer(w.ID, map[string]any{}, err) + case "mesh/state.history": + kv, err := p.kv(a.State) + if err != nil { + p.answer(w.ID, nil, err) + return + } + out := []map[string]any{} + h, err := kv.History(p.ctx, a.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + p.answer(w.ID, out, nil) + return + } + for _, e := range h { + op, v := "put", json.RawMessage(e.Value()) + if e.Operation() != jetstream.KeyValuePut { + op, v = "delete", nil + } + out = append(out, map[string]any{"key": e.Key(), "value": v, "revision": e.Revision(), "op": op, "at": e.Created()}) + } + p.answer(w.ID, out, err) + case "mesh/publish": + p.answer(w.ID, map[string]any{}, p.nc.Publish("mesh.mod."+Publisher+".event."+a.Key, a.Body)) + p.say(a.Key, a.Body) + if a.Key == HoldingsAsked { + p.answerHoldings() + } + case "mesh/seat.publish": + p.answer(w.ID, map[string]any{}, p.nc.Publish(a.Subject, a.Body)) + case "mesh/subscribe": + p.pmu.Lock() + p.subs = append(p.subs, a.Pattern) + p.pmu.Unlock() + p.answer(w.ID, map[string]any{}, nil) + case "mesh/ask": + result, err := p.asked(a.Key, a.Body) + p.say("ask "+a.Key, a.Body) + p.answer(w.ID, result, err) + default: + p.answer(w.ID, nil, errors.New("nats: no responders available for request")) + } +} + +func (p *started) say(key string, body json.RawMessage) { + p.mu.Lock() + p.said = append(p.said, said{key: key, body: append(json.RawMessage(nil), body...)}) + p.mu.Unlock() + p.notify() +} + +// answerHoldings is each node's agent module hearing holdings.asked and saying what it holds (§23). +func (p *started) answerHoldings() { + p.mu.Lock() + reports := append([]Holdings(nil), p.holdings...) + p.mu.Unlock() + for _, r := range reports { + body, _ := json.Marshal(r) + p.mu.Lock() + p.next++ + id := fmt.Sprintf("e-%d", p.next) + p.mu.Unlock() + p.send(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{ + "key": HoldingsChanged, "node": r.Node, "body": json.RawMessage(body)}}}) + } +} + +// asked answers what the manager asks of another part: the vault, a node's agent module, the operator channel. +func (p *started) asked(key string, body json.RawMessage) (any, error) { + var args map[string]any + _ = json.Unmarshal(body, &args) + p.mu.Lock() + defer p.mu.Unlock() + switch { + case key == "seat:mesh-vault.licence-put": + if call, _ := args["call"].(string); call != "" { + priv, ok := p.calls[call] + if !ok { + return nil, errors.New("no such call") + } + sealed, _ := base64.StdEncoding.DecodeString(fmt.Sprint(args["sealed"])) + var pub [32]byte + copy(pub[:], p.callArgs[call]["pub"].([]byte)) + value, ok := box.OpenAnonymous(nil, sealed, &pub, &priv) + if !ok { + return nil, errors.New("the sealed value does not open") + } + a := p.callArgs[call] + p.secrets[SecretName(fmt.Sprint(a["licence"]), fmt.Sprint(a["part"]))] = string(value) + delete(p.calls, call) + return map[string]any{"put": true}, nil + } + pub, priv, _ := box.GenerateKey(rand.Reader) + call := fmt.Sprintf("call-%d", len(p.callArgs)+1) + p.calls[call] = *priv + args["pub"] = pub[:] + p.callArgs[call] = args + return map[string]any{"call": call, "seal_to": base64.StdEncoding.EncodeToString(pub[:])}, nil + case key == "seat:mesh-vault.get": + name, _ := args["name"].(string) + v, ok := p.secrets[name] + if !ok { + return nil, fmt.Errorf("the vault holds no %s", name) + } + sealed, err := Seal(v, fmt.Sprint(args["public_key"])) + return map[string]any{"sealed": sealed}, err + case strings.HasPrefix(key, "claude-code.claude_code_grant@"): + node := strings.TrimPrefix(key, "claude-code.claude_code_grant@") + g, ok := p.logins[node] + if !ok { + return GrantAnswer{}, nil + } + raw, _ := json.Marshal(g) + sealed, err := Seal(string(raw), fmt.Sprint(args["public_key"])) + return GrantAnswer{Sealed: &sealed, Fingerprint: Fingerprint(g.RefreshToken)}, err + case key == "seat:operator-channel.tell": + return map[string]any{"told": true}, nil + } + return nil, errors.New("nats: no responders available for request") +} + +// keyMatches is a log's key against a pattern: `*` one name, `>` or `**` the rest, empty every key. +func keyMatches(pattern, key string) bool { + if pattern == "" { + return true + } + return namesMatch(strings.Split(pattern, "."), strings.Split(key, ".")) +} + +func namesMatch(p, k []string) bool { + if len(p) == 0 { + return len(k) == 0 + } + if p[0] == ">" || p[0] == "**" { + return len(k) > 0 + } + if len(k) == 0 || (p[0] != "*" && p[0] != k[0]) { + return false + } + return namesMatch(p[1:], k[1:]) +} + +// log answers the runtime's log verbs from the log's real stream, LOG__ on mesh.log.... +func (p *started) log(w wire) { + var a struct { + Log string `json:"log"` + Key string `json:"key"` + Keys string `json:"keys"` + Value json.RawMessage `json:"value"` + From uint64 `json:"from"` + Since string `json:"since"` + Limit int `json:"limit"` + } + _ = json.Unmarshal(w.Params, &a) + subject := "mesh.log." + Publisher + "." + a.Log + s, err := p.js.Stream(p.ctx, "LOG_"+Publisher+"_"+a.Log) + if err != nil { + p.answer(w.ID, nil, err) + return + } + info, err := s.Info(p.ctx) + if err != nil { + p.answer(w.ID, nil, err) + return + } + entry := func(r *jetstream.RawStreamMsg) map[string]any { + return map[string]any{"sequence": r.Sequence, "key": strings.TrimPrefix(r.Subject, subject+"."), + "value": json.RawMessage(r.Data), "at": r.Time} + } + all := func(keys string, since time.Time) []map[string]any { + out := []map[string]any{} + for seq := max(a.From, info.State.FirstSeq, 1); seq <= info.State.LastSeq; seq++ { + r, err := s.GetMsg(p.ctx, seq) + if err != nil || !keyMatches(keys, strings.TrimPrefix(r.Subject, subject+".")) || r.Time.Before(since) { + continue + } + out = append(out, entry(r)) + } + return out + } + switch w.Method { + case "mesh/log.append": + ack, err := p.js.Publish(p.ctx, subject+"."+a.Key, a.Value) + if err != nil { + p.answer(w.ID, nil, err) + return + } + p.answer(w.ID, map[string]any{"sequence": ack.Sequence}, nil) + case "mesh/log.last": + r, err := s.GetLastMsgForSubject(p.ctx, subject+"."+a.Key) + if errors.Is(err, jetstream.ErrMsgNotFound) { + p.answer(w.ID, nil, nil) + return + } + if err != nil { + p.answer(w.ID, nil, err) + return + } + p.answer(w.ID, entry(r), nil) + case "mesh/log.read": + var since time.Time + if a.Since != "" { + since, _ = time.Parse(time.RFC3339Nano, a.Since) + } + entries := all(a.Keys, since) + limit := a.Limit + if limit <= 0 { + limit = 100 + } + next := uint64(0) + if len(entries) > limit { + next = entries[limit]["sequence"].(uint64) + entries = entries[:limit] + } + p.answer(w.ID, map[string]any{"entries": entries, "next": next, "last": info.State.LastSeq}, nil) + case "mesh/log.newest": + newest := map[string]int{} + var out []map[string]any + for _, e := range all(a.Keys, time.Time{}) { + k := e["key"].(string) + if i, ok := newest[k]; ok { + out[i] = e + continue + } + newest[k] = len(out) + out = append(out, e) + } + if out == nil { + out = []map[string]any{} + } + p.answer(w.ID, out, nil) + case "mesh/log.info": + p.answer(w.ID, map[string]any{"entries": info.State.Msgs, "bytes": info.State.Bytes, + "max_bytes": info.Config.MaxBytes, "first": info.State.FirstSeq, "last": info.State.LastSeq}, nil) + default: + p.answer(w.ID, nil, fmt.Errorf("no log verb %s", w.Method)) + } +} + +// call is a tools/call of the manager by a person, answered as the tool's JSON. +func (p *started) call(name string, args map[string]any) (json.RawMessage, error) { + p.mu.Lock() + p.next++ + id := fmt.Sprintf("c-%d", p.next) + p.mu.Unlock() + ch := make(chan wire, 1) + p.pmu.Lock() + p.pending[id] = ch + p.pmu.Unlock() + p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args, + "_meta": map[string]any{"mesh/caller": "person.operator"}}}) + select { + case w := <-ch: + var r struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + IsError bool `json:"isError"` + } + if w.Error != nil { + return nil, errors.New(w.Error.Message) + } + if err := json.Unmarshal(w.Result, &r); err != nil || len(r.Content) != 1 { + return nil, fmt.Errorf("%s answered %s", name, w.Result) + } + if r.IsError { + return nil, errors.New(r.Content[0].Text) + } + return json.RawMessage(r.Content[0].Text), nil + case <-time.After(60 * time.Second): + return nil, fmt.Errorf("%s did not answer in 60s", name) + } +} + +func (p *started) must(name string, args map[string]any) json.RawMessage { + p.t.Helper() + raw, err := p.call(name, args) + if err != nil { + p.t.Fatalf("%s: %v", name, err) + } + return raw +} + +// carried is the manager's carry-over record, as licences_carried answers it. +func (p *started) carried() CarriedRecord { + p.t.Helper() + var a struct { + Carried *CarriedRecord `json:"carried"` + Failure string `json:"failure"` + } + if err := json.Unmarshal(p.must("licences_carried", nil), &a); err != nil || a.Carried == nil { + p.t.Fatalf("licences_carried answers no record (%v): %s", err, a.Failure) + } + return *a.Carried +} + +// bindings are the consumers' bindings, as the seat's verb answers them. +func (p *started) bindings() map[string]Binding { + p.t.Helper() + var list []Binding + if err := json.Unmarshal(p.must(Seat+".bindings", nil), &list); err != nil { + p.t.Fatal(err) + } + out := map[string]Binding{} + for _, b := range list { + out[b.Consumer] = b + } + return out +} + +// written is every revision of the states and every entry of the logs the carry-over writes: what a second start +// must leave as it is (§13). +func (m *mesh) written() map[string]uint64 { + m.t.Helper() + out := map[string]uint64{} + for _, s := range []string{"licences", "bindings", "offered", "carried"} { + info, err := m.js.Stream(m.ctx, "KV_"+Publisher+"_"+s) + if err != nil { + m.t.Fatal(err) + } + i, _ := info.Info(m.ctx) + out[s] = i.State.LastSeq + } + for _, l := range strings.Fields(busLogs) { + info, err := m.js.Stream(m.ctx, "LOG_"+Publisher+"_"+l) + if err != nil { + m.t.Fatal(err) + } + i, _ := info.Info(m.ctx) + out[l] = i.State.LastSeq + } + return out +} + +// ---- the tests ------------------------------------------------------------------------------------------------- + +// design 39 bus-and-switch §11/AC1, §12/AC1, §13/AC1, §13/AC2: a start stopped at a table after bindings says +// carry-over-failed naming it and acts on nothing; started again it carries every row whole, the counts in +// carried.postgres equal to the rows read, the vault answering each licence's grant, and clears the condition; a +// further start writes nothing. +func TestTodaysRowsAreCarriedOverOnceAndWhole(t *testing.T) { + dbURL, conn := freshDatabase(t) + fillTodaysRows(t, conn) + ctx := context.Background() + // The table after bindings cannot be read: the start stops there. + if _, err := conn.Exec(ctx, `alter table offered rename column why to reason`); err != nil { + t.Fatal(err) + } + m := newWholeMesh(t, dbURL) + first := m.start() + m.until("carry-over-failed is said", func() bool { return len(m.conditionsOf(CarryOverFailed)) > 0 }) + m.mu.Lock() + raised := m.conditionsOf(CarryOverFailed)[0] + m.mu.Unlock() + if raised.State != "raised" || raised.Severity != "urgent" || !strings.Contains(raised.Explanation+raised.Summary, "offered") { + t.Errorf("carry-over-failed said %+v", raised) + } + if _, err := first.call(Seat+".bind", map[string]any{"consumer": "laptop", "licence": licenceB}); err == nil || + !strings.Contains(err.Error(), "the carry-over has not ended") { + t.Errorf("bind before the carry-over ended answered %v", err) + } + if n := m.vendor.called(); n != 0 { + t.Errorf("the vendor was called %d times before the carry-over ended", n) + } + first.stop() + + if _, err := conn.Exec(ctx, `alter table offered rename column reason to why`); err != nil { + t.Fatal(err) + } + second := m.start() + m.until("holdings.asked after the carry-over", func() bool { return m.count(HoldingsAsked) > 0 }) + rec := second.carried() + want := map[string][2]int{"licence": {2, 2}, "binding": {4, 4}, "lease": {1, 0}, "offered": {2, 1}, "usage": {2, 2}, "audit": {4, 4}} + for table, n := range want { + if got := rec.Tables[table]; got.Read != n[0] || got.Written != n[1] { + t.Errorf("%s: read %d written %d, want read %d written %d", table, got.Read, got.Written, n[0], n[1]) + } + } + m.mu.Lock() + for _, l := range []struct{ name, refresh string }{{licenceA, "rt-a"}, {licenceB, "rt-b"}} { + if g := m.secrets[SecretName(KeyFor(l.name), PartGrant)]; !strings.Contains(g, l.refresh) { + t.Errorf("the vault holds %q as the grant of %s", g, l.name) + } + } + m.mu.Unlock() + if b := second.bindings(); len(b) != 4 || b["anchor"].Licence != licenceA || b["anchor"].Generation != 1013 || b["anchor"].Why != "carried over" { + t.Errorf("bindings answers %+v", b) + } + m.until("carry-over-failed is cleared", func() bool { + c := m.conditionsOf(CarryOverFailed) + return len(c) > 1 && c[len(c)-1].State == "cleared" + }) + second.stop() + + before := m.written() + asked := m.count(HoldingsAsked) + third := m.start() + m.until("holdings.asked at the third start", func() bool { return m.count(HoldingsAsked) > asked }) + if again := third.carried(); again.At != rec.At { + t.Errorf("the third start carried over again at %s (first at %s)", again.At, rec.At) + } + if after := m.written(); fmt.Sprint(after) != fmt.Sprint(before) { + t.Errorf("a further start wrote %v, before it %v", after, before) + } +} + +// design 39 bus-and-switch §11/AC2: with the database unreachable the condition is raised, no act is taken and the +// vendor is never called. +func TestAnUnreachableDatabaseRaisesAndCallsNoVendor(t *testing.T) { + m := newWholeMesh(t, "postgres://postgres@127.0.0.1:1/claude_licences?connect_timeout=2") + p := m.start() + m.until("carry-over-failed is said", func() bool { return len(m.conditionsOf(CarryOverFailed)) > 0 }) + m.mu.Lock() + c := m.conditionsOf(CarryOverFailed)[0] + m.mu.Unlock() + if c.State != "raised" || c.Severity != "urgent" || c.Subject.ID != Publisher { + t.Errorf("carry-over-failed said %+v", c) + } + if _, err := p.call("licences_read_usage", nil); err == nil || !strings.Contains(err.Error(), "the carry-over has not ended") { + t.Errorf("licences_read_usage before the carry-over answered %v", err) + } + if n := m.vendor.called(); n != 0 { + t.Errorf("the vendor was called %d times", n) + } + m.mu.Lock() + defer m.mu.Unlock() + if n := m.count(HoldingsAsked); n != 0 { + t.Errorf("holdings.asked was said %d times before the carry-over", n) + } +} + +// design 39 bus-and-switch §26/AC2, §28, §29/AC2, §9/AC3, B41: a reading of 99 % moves the four consumers bound to a +// licence to the one with the most room in one act, saying one binding.changed each, one licence.switched and one tell; +// with every licence at its threshold no-room is raised on the operator channel's subject, and a reading with room +// clears it. +func TestAReadingAtTheThresholdMovesEveryConsumer(t *testing.T) { + dbURL, conn := freshDatabase(t) + fillTodaysRows(t, conn) + m := newWholeMesh(t, dbURL) + p := m.start() + m.until("holdings.asked after the carry-over", func() bool { return m.count(HoldingsAsked) > 0 }) + m.vendor.set("at-b", 10, 20) + m.vendor.set("at-a", 99, 50) + p.must("licences_read_usage", map[string]any{"licence": licenceB}) + p.must("licences_read_usage", map[string]any{"licence": licenceA}) + m.until("the four consumers moved", func() bool { return m.count("binding.changed") >= 4 }) + m.mu.Lock() + switched, told := m.of("licence.switched"), m.of("ask seat:operator-channel.tell") + changed := m.of("binding.changed") + m.mu.Unlock() + if len(switched) != 1 || switched[0]["from"] != licenceA || switched[0]["to"] != licenceB || switched[0]["by"] != "usage" || + len(switched[0]["consumers"].([]any)) != 4 { + t.Errorf("licence.switched said %v", switched) + } + if len(told) != 1 { + t.Errorf("the operator was told %d times: %v", len(told), told) + } + moved := map[string]bool{} + for _, c := range changed { + if c["licence"] == licenceB && c["from"] == licenceA && c["by"] == "usage" { + moved[fmt.Sprint(c["consumer"])] = true + } + } + if len(moved) != 4 || len(changed) != 4 { + t.Errorf("binding.changed said %v", changed) + } + for c, b := range p.bindings() { + if b.Licence != licenceB || b.By != "usage" { + t.Errorf("%s is bound %+v", c, b) + } + } + + // Every licence at its threshold: the consumers have nowhere to go. + noRoom := "licence." + KeyFor(licenceB) + ".no-room" + m.vendor.set("at-a", 99, 50) + m.vendor.set("at-b", 99, 60) + p.must("licences_read_usage", map[string]any{"licence": licenceA}) + p.must("licences_read_usage", map[string]any{"licence": licenceB}) + m.until("no-room is raised", func() bool { return len(m.conditionsOf(noRoom)) > 0 }) + m.mu.Lock() + r := m.conditionsOf(noRoom)[0] + m.mu.Unlock() + if r.State != "raised" || r.Severity != "urgent" || r.Subject.ID != Publisher { + t.Errorf("no-room said %+v", r) + } + // A reading with room: the next reading of the full licence moves its consumers, and no-room is cleared. + m.vendor.set("at-a", 5, 30) + p.must("licences_read_usage", map[string]any{"licence": licenceA}) + p.must("licences_read_usage", map[string]any{"licence": licenceB}) + m.until("no-room is cleared", func() bool { + c := m.conditionsOf(noRoom) + return len(c) > 1 && c[len(c)-1].State == "cleared" + }) + m.mu.Lock() + defer m.mu.Unlock() + if c := m.conditionsOf(noRoom); len(c) != 2 { + t.Errorf("no-room was said %d times, not once raised and once cleared: %+v", len(c), c) + } + if n := m.count("licence.switched"); n != 2 { + t.Errorf("licence.switched was said %d times, not once per act", n) + } +} + +// design 39 bus-and-switch §23/AC1, §46/AC1, §11/AC3: on a fresh install (no table) the carry-over carries nothing and +// the manager says holdings.asked; each node's agent module answers with one holdings-changed; a report carrying an +// unknown refresh token's fingerprint makes a candidate, which is adopted, listed by licences with its organisation, +// and chosen as a target at its first reading with room. +func TestANewLoginIsAdoptedListedAndATarget(t *testing.T) { + dbURL, _ := freshDatabase(t) + m := newWholeMesh(t, dbURL) + changed := time.Now().UTC().Format(time.RFC3339Nano) + report := func(node, account, email, org, refresh string) Holdings { + h := Holdings{Node: node, Identity: &Identity{AccountUUID: account, EmailAddress: email, OrganizationUUID: org}, + Kind: "subscription", ChangedAt: changed} + h.Refresh.Present, h.Refresh.Fingerprint = true, Fingerprint(refresh) + return h + } + m.holdings = []Holdings{report("laptop", "acct-old", licenceA, "", "rt-old"), report("workstation", "acct-team", "team@example.org", "org-1", "rt-team")} + m.logins["laptop"] = FullGrant{AccessToken: "at-old-0", RefreshToken: "rt-old"} + m.logins["workstation"] = FullGrant{AccessToken: "at-team-0", RefreshToken: "rt-team"} + m.vendor.logins["rt-old"] = [2]string{"at-old", "acct-old"} + m.vendor.logins["rt-team"] = [2]string{"at-team", "acct-team"} + p := m.start() + m.until("both logins adopted", func() bool { return m.count("licence.adopted") >= 2 }) + m.mu.Lock() + asked := m.count("ask claude-code.claude_code_grant@laptop") + m.count("ask claude-code.claude_code_grant@workstation") + m.mu.Unlock() + if asked != 2 { + t.Errorf("the nodes were asked for their logins %d times, not once each", asked) + } + if rec := p.carried(); rec.Tables["licence"].Read != 0 || rec.Tables["audit"].Read != 0 { + t.Errorf("a fresh install carried %+v", rec.Tables) + } + + // The full licence is read first: nothing has room yet, so its consumer stays. + m.vendor.set("at-old", 99, 40) + m.vendor.set("at-team", 10, 5) + p.must("licences_read_usage", map[string]any{"licence": licenceA}) + p.must("licences_read_usage", map[string]any{"licence": "team@example.org"}) + var listed []map[string]any + if err := json.Unmarshal(p.must(Seat+".licences", nil), &listed); err != nil { + t.Fatal(err) + } + var team map[string]any + for _, l := range listed { + if l["name"] == "team@example.org" { + team = l + } + } + if team == nil || team["organisation"] != "org-1" || team["target"] != true || team["usage"] == nil { + t.Fatalf("licences lists the new login as %v (all: %v)", team, listed) + } + p.must("licences_read_usage", map[string]any{"licence": licenceA}) + m.until("the full licence's consumer moved to the new one", func() bool { + for _, s := range m.of("licence.switched") { + if s["to"] == "team@example.org" && s["from"] == licenceA { + return true + } + } + return false + }) + if b := p.bindings()["laptop"]; b.Licence != "team@example.org" { + t.Errorf("laptop is bound %+v", b) + } +} diff --git a/modules/claude-licence-manager/go.mod b/modules/claude-licence-manager/go.mod index 77296baf3..dc60fbeab 100644 --- a/modules/claude-licence-manager/go.mod +++ b/modules/claude-licence-manager/go.mod @@ -7,18 +7,24 @@ require ( git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 github.com/jackc/pgx/v5 v5.11.0 + github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 ) require ( + github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/compress v1.20.1 // indirect - github.com/nats-io/nats.go v1.54.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.2 // indirect + github.com/nats-io/nats-server/v2 v2.11.17 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/sync v0.23.0 // indirect golang.org/x/sys v0.48.0 // indirect golang.org/x/text v0.42.0 // indirect + golang.org/x/time v0.16.0 // indirect ) diff --git a/modules/claude-licence-manager/go.sum b/modules/claude-licence-manager/go.sum index 7475ceb70..96a5b2dcb 100644 --- a/modules/claude-licence-manager/go.sum +++ b/modules/claude-licence-manager/go.sum @@ -1,38 +1,50 @@ +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= -github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= -github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= -github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= -github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= -github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc= +github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc= github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= -github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= -github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= -github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= -golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= -golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= -golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=