diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/carry.go b/modules/claude-licence-manager/cmd/claude-licence-manager/carry.go new file mode 100644 index 000000000..9f0358033 --- /dev/null +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/carry.go @@ -0,0 +1,437 @@ +package main + +// The carry-over (design 39 bus-and-switch §8, §11 to §13, A3, B14): at its first start the manager reads every row of +// today's database, the module's own and read only, and writes it into its kept data on the bus and its grants into +// the vault, then the key `carried.postgres`. Until that key exists it serves its read verbs alone and does nothing +// else. With the key present it never reads the database again. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "time" +) + +// CarriedKey is the key of the state `carried` the carry-over writes at its end (§8). +const CarriedKey = "postgres" + +// NotCarried is the answer of every acting verb and tool until the carry-over has ended (§11). +const NotCarried = "the carry-over has not ended" + +// usageKeptFor is how far back the carry-over reads usage: the log `usage` keeps no older reading (§6, §12). +const usageKeptFor = 35 * 24 * time.Hour + +// The tables of today's database, in the order the carry-over reads them (§11, §12). +var CarriedTables = []string{"licence", "binding", "lease", "offered", "usage", "audit"} + +// LicenceRow is a row of the table `licence`: the licence and its grant at rest, sealed with `grant.key`. +type LicenceRow struct { + Licence + Sealed string +} + +// OfferedRow is a row of the table `offered`. +type OfferedRow struct { + Fingerprint, Node, Account string + Outcome Outcome + Why string + At time.Time +} + +// UsageTableRow is a row of the table `usage`: the reading and the vendor's answer whole. +type UsageTableRow struct { + Licence string + At int64 // milliseconds + Reading UsageReading + Raw map[string]any +} + +// AuditRow is a row of the table `audit`. +type AuditRow struct { + At time.Time + What string + Detail map[string]any +} + +// CarrySource reads the six tables of today's database, each oldest first; a table that does not exist is no rows. +type CarrySource interface { + Licences(ctx context.Context) ([]LicenceRow, error) + Bindings(ctx context.Context) ([]Binding, error) + Leases(ctx context.Context) (int, error) + Offered(ctx context.Context) ([]OfferedRow, error) + Usage(ctx context.Context, since time.Time) ([]UsageTableRow, error) + Audit(ctx context.Context) ([]AuditRow, error) +} + +// CarriedTable is one table in `carried.postgres`: the rows read and the entries written for them. +type CarriedTable struct { + Read int `json:"read"` + Written int `json:"written"` +} + +// CarriedRecord is the value of `carried.postgres` (§8): when, and for each table the rows read and entries written. +type CarriedRecord struct { + At string `json:"at"` + Tables map[string]CarriedTable `json:"tables"` +} + +// Carry is the carry-over's gate: closed until `carried.postgres` exists, and the failure of its last run. +type Carry struct { + State keptState // the state `carried` + // Source opens today's database; called only while the key is absent. + Source func(ctx context.Context) (CarrySource, error) + + mu sync.Mutex + running bool + failure string + done chan struct{} + once sync.Once +} + +func (c *Carry) init() { + c.mu.Lock() + if c.done == nil { + c.done = make(chan struct{}) + } + c.mu.Unlock() +} + +// Ended is whether the carry-over has ended: the key `carried.postgres` exists. +func (c *Carry) Ended() bool { + if c == nil { + return true // a manager built with no carry-over to wait on: the tests' own + } + c.init() + select { + case <-c.done: + return true + default: + } + if e, err := c.State.Get(CarriedKey); err == nil && e != nil && len(e.Value) > 0 { + c.end() + return true + } + return false +} + +func (c *Carry) end() { c.once.Do(func() { close(c.done) }) } + +// Wait is closed once the carry-over has ended. +func (c *Carry) Wait() <-chan struct{} { + c.init() + return c.done +} + +// Acts answers an acting verb or tool: nothing until the carry-over has ended (§11). +func (c *Carry) Acts() error { + if c.Ended() { + return nil + } + return errors.New(NotCarried) +} + +// Failure is what the last run that failed names, empty when none failed. +func (c *Carry) Failure() string { + c.mu.Lock() + defer c.mu.Unlock() + return c.failure +} + +// Run carries over every row when `carried.postgres` is absent, and says nothing else happened when it is present +// (§13). A run that fails names its table and error, and leaves the key absent, so the next is done again whole. +func (c *Carry) Run(ctx context.Context, m *Manager, s *BusStore) (*CarriedRecord, error) { + if c.Ended() { + return nil, nil + } + c.mu.Lock() + if c.running { + c.mu.Unlock() + return nil, errors.New("a carry-over is running") + } + c.running = true + c.mu.Unlock() + rec, err := c.run(ctx, m, s) + c.mu.Lock() + c.running = false + if err != nil { + c.failure = err.Error() + } else { + c.failure = "" + } + c.mu.Unlock() + if err != nil { + m.Log("the carry-over failed: %v", err) + return nil, err + } + c.end() + m.Log("carried over today's database: %v", rec.Tables) + return rec, nil +} + +func (c *Carry) run(ctx context.Context, m *Manager, s *BusStore) (*CarriedRecord, error) { + src, err := c.Source(ctx) + if err != nil { + return nil, fmt.Errorf("the database cannot be opened: %w", err) + } + if closer, ok := src.(interface{ Close() }); ok { + defer closer.Close() + } + co := carryOver{m: m, s: s, src: src, now: m.Now(), tables: map[string]CarriedTable{}} + if err := co.all(ctx); err != nil { + return nil, err + } + rec := CarriedRecord{At: co.now.UTC().Format(time.RFC3339), Tables: co.tables} + detail := map[string]any{"why": "the first start of the manager on the bus", "by": Itself, "tables": rec.Tables} + if err := s.Audit(ctx, "carried", detail); err != nil { + return nil, fmt.Errorf("carried: %w", err) + } + if _, err := c.State.Put(CarriedKey, rec); err != nil { + return nil, fmt.Errorf("carried: the key %s was not written: %w", CarriedKey, err) + } + return &rec, nil +} + +// carryOver is one run of the carry-over. +type carryOver struct { + m *Manager + s *BusStore + src CarrySource + now time.Time + tables map[string]CarriedTable + audit []AuditRow +} + +func (co *carryOver) all(ctx context.Context) error { + // The audit is read first: a licence's adoptedFrom comes from it (B14). + audit, err := co.src.Audit(ctx) + if err != nil { + return fmt.Errorf("audit: %w", err) + } + co.audit = audit + for _, table := range CarriedTables { + var t CarriedTable + var err error + switch table { + case "licence": + t, err = co.licences(ctx) + case "binding": + t, err = co.bindings(ctx) + case "lease": + t.Read, err = co.src.Leases(ctx) // a lease lives two minutes: none taken before the start still holds + case "offered": + t, err = co.offered(ctx) + case "usage": + t, err = co.usage(ctx) + case "audit": + t, err = co.audits() + } + if err != nil { + return fmt.Errorf("%s: %w", table, err) + } + co.tables[table] = t + } + return nil +} + +// adoptedFrom is the node the newest adoption of a licence names in the audit (B14). +func (co *carryOver) adoptedFrom(name string) string { + from := "" + for _, r := range co.audit { + if r.What != "adopted" || r.Detail["licence"] != name { + continue + } + if n, _ := r.Detail["node"].(string); n != "" { + from = n + } else if f, _ := r.Detail["from"].(string); f != "" { + from = f + } + } + return from +} + +// licences writes a key of `licences` per row, its grant opened with `grant.key` and handed to the vault at +// generation 1, the grant first (§12, §15). +func (co *carryOver) licences(ctx context.Context) (CarriedTable, error) { + rows, err := co.src.Licences(ctx) + if err != nil { + return CarriedTable{}, err + } + t := CarriedTable{Read: len(rows)} + for _, r := range rows { + l := r.Licence + l.Generation, l.Usage = 0, nil + l.AdoptedFrom = co.adoptedFrom(l.Name) + if r.Sealed != "" { + if co.m.Crypt == nil { + return t, fmt.Errorf("the grant of %s cannot be opened: no grant.key", l.Name) + } + plain, err := co.m.Crypt.Open(r.Sealed) + if err != nil { + return t, fmt.Errorf("the grant of %s: %w", l.Name, err) + } + grant, access := plain, plain + if l.Kind == "subscription" { + var g FullGrant + if err := json.Unmarshal([]byte(plain), &g); err != nil { + return t, fmt.Errorf("the grant of %s is not a grant", l.Name) + } + access = accessOf(g) + } + if err := co.m.handOver(ctx, &l, grant, access); err != nil { + return t, err + } + } + if err := co.s.SaveLicence(ctx, l); err != nil { + return t, err + } + t.Written++ + } + return t, nil +} + +// bindings writes a key of `bindings` per row, its generation kept, why "carried over", by the manager, since the +// carry-over's time (§12, B14); a key a run before this one wrote for the same row keeps its since, so it is put again +// with the same value (§13). +func (co *carryOver) bindings(ctx context.Context) (CarriedTable, error) { + rows, err := co.src.Bindings(ctx) + if err != nil { + return CarriedTable{}, err + } + kinds := map[string]string{} + if all, err := co.s.Licences(ctx); err == nil { + for _, l := range all { + kinds[l.Name] = l.Kind + } + } else { + return CarriedTable{}, err + } + t := CarriedTable{Read: len(rows)} + for _, r := range rows { + v := BindingState{Licence: r.Licence, Kind: kinds[r.Licence], Generation: r.Generation, + Since: co.now.UTC().Format(time.RFC3339), Why: "carried over", By: Itself} + var held BindingState + if ok, err := co.s.get(co.s.BindingsState, r.Consumer, &held); err != nil { + return t, err + } else if ok && held.Licence == v.Licence && held.Generation == v.Generation && held.Why == v.Why { + v.Since = held.Since + } + if _, err := co.s.BindingsState.Put(r.Consumer, v); err != nil { + return t, err + } + t.Written++ + } + return t, nil +} + +// offered writes a key of `offered` per row, its lifetime counted from the row's `at`; a row past its lifetime is +// gone already and written nowhere (§5, §12). +func (co *carryOver) offered(ctx context.Context) (CarriedTable, error) { + rows, err := co.src.Offered(ctx) + if err != nil { + return CarriedTable{}, err + } + t := CarriedTable{Read: len(rows)} + for _, r := range rows { + left := offeredFor - co.now.Sub(r.At) + if left <= 0 { + continue + } + v := offeredValue{Node: r.Node, Account: r.Account, Outcome: r.Outcome, Why: r.Why, + At: r.At.UTC().Format(time.RFC3339)} + if err := co.s.putOffered(r.Fingerprint, v, left); err != nil { + return t, err + } + t.Written++ + } + return t, nil +} + +// lastAt is the `at` of a log's last entry for a subject, zero when it has none: rows at or before it are in the log +// already (§13). +func lastAt(l keptLog, key string) (time.Time, error) { + e, err := l.Last(key) + if err != nil || e == nil { + return time.Time{}, err + } + var v struct { + At string `json:"at"` + } + if json.Unmarshal(e.Value, &v) != nil { + return time.Time{}, nil + } + at, _ := time.Parse(time.RFC3339Nano, v.At) + return at, nil +} + +// usage appends one entry of the log `usage` per row of the last 35 days, oldest first, its `at` kept (§12). +func (co *carryOver) usage(ctx context.Context) (CarriedTable, error) { + rows, err := co.src.Usage(ctx, co.now.Add(-usageKeptFor)) + if err != nil { + return CarriedTable{}, err + } + t := CarriedTable{Read: len(rows)} + last := map[string]time.Time{} + for _, r := range rows { + key := KeyFor(r.Licence) + before, seen := last[key] + if !seen { + if before, err = lastAt(co.s.UsageLog, key); err != nil { + return t, err + } + last[key] = before + } + at := time.UnixMilli(r.At).UTC() + if !at.After(before) { + t.Written++ // in the log already, from a run before this one + continue + } + if _, err := co.s.UsageLog.Append(key, UsageEntry{At: at.Format(time.RFC3339Nano), Reading: r.Reading, + Raw: r.Raw}); err != nil { + return t, err + } + t.Written++ + } + return t, nil +} + +// audits appends one entry of the log `audit` per row, oldest first, its `at` kept, under the subject its `what` +// names, `drift` for any other (§7, §12). +func (co *carryOver) audits() (CarriedTable, error) { + t := CarriedTable{Read: len(co.audit)} + last := map[string]time.Time{} + for _, r := range co.audit { + kind := "drift" + for _, k := range AuditKinds { + if k == r.What { + kind = k + } + } + before, seen := last[kind] + if !seen { + var err error + if before, err = lastAt(co.s.AuditLog, kind); err != nil { + return t, err + } + last[kind] = before + } + if !r.At.After(before) { + t.Written++ // in the log already, from a run before this one + continue + } + entry := map[string]any{} + for k, v := range r.Detail { + entry[k] = v + } + entry["at"] = r.At.UTC().Format(time.RFC3339Nano) + if kind != r.What { + entry["what"] = r.What + } + if _, err := co.s.AuditLog.Append(kind, entry); err != nil { + return t, fmt.Errorf("the audit did not take the carried %s entry: %w", r.What, err) + } + t.Written++ + } + return t, nil +} diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/carry_test.go b/modules/claude-licence-manager/cmd/claude-licence-manager/carry_test.go new file mode 100644 index 000000000..a4be4be5e --- /dev/null +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/carry_test.go @@ -0,0 +1,276 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "os" + "regexp" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" +) + +// memSource is today's database as the carry-over reads it, from rows given; failAt makes one table's read fail. +type memSource struct { + licences []LicenceRow + bindings []Binding + leases int + offered []OfferedRow + usage []UsageTableRow + audit []AuditRow + failAt string +} + +func (s *memSource) fail(table string) error { + if s.failAt == table { + return errors.New("connection reset by peer") + } + return nil +} +func (s *memSource) Licences(context.Context) ([]LicenceRow, error) { + return s.licences, s.fail("licence") +} +func (s *memSource) Bindings(context.Context) ([]Binding, error) { + return s.bindings, s.fail("binding") +} +func (s *memSource) Leases(context.Context) (int, error) { return s.leases, s.fail("lease") } +func (s *memSource) Offered(context.Context) ([]OfferedRow, error) { + return s.offered, s.fail("offered") +} +func (s *memSource) Audit(context.Context) ([]AuditRow, error) { return s.audit, s.fail("audit") } +func (s *memSource) Usage(_ context.Context, since time.Time) ([]UsageTableRow, error) { + var out []UsageTableRow + for _, u := range s.usage { + if u.At >= since.UnixMilli() { + out = append(out, u) + } + } + return out, s.fail("usage") +} + +// todaysRows is a copy of the shape of today's rows: a subscription and an API key, two bindings, a lease, two offers, +// three readings (one older than 35 days) and four audit rows (one of a kind the log does not name). +func todaysRows(t *testing.T, mm *miniMesh) *memSource { + t.Helper() + grant, _ := json.Marshal(FullGrant{AccessToken: "at-old", RefreshToken: "rt-old", ExpiresAt: 7}) + pct := 42.0 + return &memSource{ + licences: []LicenceRow{ + {Licence: Licence{Name: licence1, Kind: "subscription", AccountUUID: "acct-1", Email: licence1, + RefreshFingerprint: Fingerprint("rt-old"), AccessExpiresAt: 7, AdoptedAt: 1}, Sealed: sealAtRest(t, mm.m.Crypt, string(grant))}, + {Licence: Licence{Name: "api", Kind: "api-key", AdoptedAt: 2}, Sealed: sealAtRest(t, mm.m.Crypt, "sk-ant-api03-x")}, + }, + bindings: []Binding{{Consumer: "laptop", Licence: licence1, Generation: 1013}, {Consumer: "server", Licence: "api", Generation: 1016}}, + leases: 1, + offered: []OfferedRow{{Fingerprint: "sha256:aa", Node: "laptop", Account: "acct-1", Outcome: Adopted, Why: "refreshed", At: t0.Add(-24 * time.Hour)}, + {Fingerprint: "sha256:bb", Node: "server", Outcome: Dead, Why: "400", At: t0.Add(-48 * time.Hour)}}, + usage: []UsageTableRow{ + {Licence: licence1, At: t0.Add(-40 * 24 * time.Hour).UnixMilli(), Reading: UsageReading{SessionPct: &pct}}, + {Licence: licence1, At: t0.Add(-2 * time.Hour).UnixMilli(), Reading: UsageReading{SessionPct: &pct}, Raw: map[string]any{"five_hour": 1.0}}, + {Licence: licence1, At: t0.Add(-1 * time.Hour).UnixMilli(), Reading: UsageReading{SessionPct: &pct}}, + }, + audit: []AuditRow{ + {At: t0.Add(-72 * time.Hour), What: "adopted", Detail: map[string]any{"licence": licence1, "node": "laptop"}}, + {At: t0.Add(-71 * time.Hour), What: "bound", Detail: map[string]any{"consumer": "laptop"}}, + {At: t0.Add(-70 * time.Hour), What: "switched", Detail: map[string]any{"consumer": "server"}}, + {At: t0.Add(-69 * time.Hour), What: "grant-moved", Detail: map[string]any{"licence": "api"}}, + }, + } +} + +func carryFor(mm *miniMesh, carried *memState, src CarrySource) *Carry { + return &Carry{State: carried, Source: func(context.Context) (CarrySource, error) { return src, nil }} +} + +func readCarried(t *testing.T, carried *memState) CarriedRecord { + t.Helper() + var rec CarriedRecord + e, _ := carried.Get(CarriedKey) + if e == nil { + t.Fatal("carried.postgres is not written") + } + _ = json.Unmarshal(e.Value, &rec) + return rec +} + +// design 39 bus-and-switch §11/AC1, §12/AC1: every row is carried over and the key written; the counts in +// carried.postgres equal the rows read for every table but lease, and the vault answers each licence's grant. +func TestEveryRowIsCarriedOver(t *testing.T) { + mm := newMesh(t) + carried := newMemState(func() time.Time { return mm.now }) + src := todaysRows(t, mm) + c := carryFor(mm, carried, src) + if _, err := c.Run(context.Background(), mm.m, mm.store); err != nil { + t.Fatal(err) + } + rec := readCarried(t, carried) + want := map[string]int{"licence": 2, "binding": 2, "lease": 1, "offered": 2, "usage": 2, "audit": 4} + for table, n := range want { + got := rec.Tables[table] + if got.Read != n || (table != "lease" && got.Written != got.Read) { + t.Errorf("%s: read %d written %d, want %d read and as many written", table, got.Read, got.Written, n) + } + } + if g, err := mm.grants.Get(context.Background(), KeyFor(licence1), PartGrant); err != nil || !strings.Contains(g, "rt-old") { + t.Fatalf("the vault answers the grant %q, %v", g, err) + } + if a, _ := mm.grants.Get(context.Background(), KeyFor(licence1), PartAccess); strings.Contains(a, "rt-old") || !strings.Contains(a, "at-old") { + t.Fatalf("the access secret is %q", a) + } + if a, _ := mm.grants.Get(context.Background(), KeyFor("api"), PartAccess); a != "sk-ant-api03-x" { + t.Fatalf("the API key's access is %q", a) + } + l, _ := mm.store.Licence(context.Background(), licence1) + if l == nil || l.Generation != 1 || l.AdoptedFrom != "laptop" || l.AccountUUID != "acct-1" || l.AccessExpiresAt != 7 { + t.Fatalf("the licence is %+v", l) + } + if b := mm.bound("laptop"); b.Generation != 1013 || b.Why != "carried over" || b.By != Itself || b.Kind != "subscription" || b.Since == "" { + t.Fatalf("laptop's binding %+v", b) + } + if u := mm.offered.until[OfferedKey("sha256:aa")]; !u.Equal(t0.Add(-24 * time.Hour).Add(offeredFor)) { + t.Fatalf("the offer lives until %v", u) + } + if got := mm.audit.under("drift"); len(got) != 1 || got[0]["licence"] != "api" { + t.Fatalf("drift holds %v", got) + } + if got := mm.audit.under("adopted"); len(got) != 1 || got[0]["at"] != t0.Add(-72*time.Hour).Format(time.RFC3339Nano) { + t.Fatalf("adopted holds %v", got) + } + if got := mm.usage.under(KeyFor(licence1)); len(got) != 2 || got[0]["raw"] == nil { + t.Fatalf("usage holds %v", got) + } + if len(mm.audit.under("carried")) != 1 || !c.Ended() { + t.Fatal("the carry-over did not say it carried, or did not end") + } +} + +// design 39 bus-and-switch §13/AC1: a second start, with carried.postgres present, never opens the database and +// writes nothing. +func TestASecondStartCarriesNothing(t *testing.T) { + mm := newMesh(t) + carried := newMemState(func() time.Time { return mm.now }) + if _, err := carryFor(mm, carried, todaysRows(t, mm)).Run(context.Background(), mm.m, mm.store); err != nil { + t.Fatal(err) + } + writes := carried.writes + mm.licences.writes + mm.bindings.writes + mm.offered.writes + entries := len(mm.audit.entries) + len(mm.usage.entries) + puts := len(mm.grants.puts) + opened := false + again := &Carry{State: carried, Source: func(context.Context) (CarrySource, error) { opened = true; return nil, errors.New("no") }} + if rec, err := again.Run(context.Background(), mm.m, mm.store); rec != nil || err != nil || opened { + t.Fatalf("a second start answered %v %v, opened the database: %v", rec, err, opened) + } + if carried.writes+mm.licences.writes+mm.bindings.writes+mm.offered.writes != writes || + len(mm.audit.entries)+len(mm.usage.entries) != entries || len(mm.grants.puts) != puts { + t.Fatal("a second start wrote") + } +} + +// design 39 bus-and-switch §13/AC2: a carry-over stopped after bindings leaves no key, names its table, and a start +// again carries over whole: every count equals the rows read, and no log entry is appended twice. +func TestAStoppedCarryOverIsDoneAgainWhole(t *testing.T) { + mm := newMesh(t) + carried := newMemState(func() time.Time { return mm.now }) + src := todaysRows(t, mm) + src.failAt = "offered" + c := carryFor(mm, carried, src) + if _, err := c.Run(context.Background(), mm.m, mm.store); err == nil || !strings.Contains(c.Failure(), "offered") { + t.Fatalf("a stopped carry-over answered %v, its failure %q", err, c.Failure()) + } + if e, _ := carried.Get(CarriedKey); e != nil || c.Ended() { + t.Fatal("a stopped carry-over wrote its key") + } + src.failAt = "" + since := mm.bound("laptop").Since + mm.now = mm.now.Add(time.Minute) + if _, err := c.Run(context.Background(), mm.m, mm.store); err != nil { + t.Fatal(err) + } + for table, got := range readCarried(t, carried).Tables { + if table != "lease" && got.Written != got.Read { + t.Errorf("%s: read %d written %d", table, got.Read, got.Written) + } + } + if b := mm.bound("laptop"); b.Since != since { + t.Fatalf("the binding put again changed its since from %s to %s", since, b.Since) + } + if n := len(mm.audit.under("")); n != 5 { + t.Fatalf("the audit holds %d entries, want the four rows and one carried", n) + } + // the run stopped before the logs; a second pass over a log already written appends nothing. + co := carryOver{m: mm.m, s: mm.store, src: src, now: mm.now, tables: map[string]CarriedTable{}, audit: src.audit} + if _, err := co.usage(context.Background()); err != nil { + t.Fatal(err) + } + if got, err := co.audits(); err != nil || got.Written != 4 || len(mm.audit.under("")) != 5 || len(mm.usage.under("")) != 2 { + t.Fatalf("a log written again appended: %+v %v, %d usage entries", got, err, len(mm.usage.under(""))) + } +} + +// design 39 bus-and-switch §11/AC2: with the database unreachable the carry-over fails naming it, every acting verb +// answers "the carry-over has not ended", a read verb is served, and no vendor call is made. +func TestAnUnreachableDatabaseActsNotAndCallsNoVendor(t *testing.T) { + mm := newMesh(t) + carried := newMemState(func() time.Time { return mm.now }) + mm.m.Carry = &Carry{State: carried, Source: func(context.Context) (CarrySource, error) { + return nil, errors.New("dial tcp 192.0.2.1:5432: connect: connection refused") + }} + if _, err := mm.m.Carry.Run(context.Background(), mm.m, mm.store); err == nil || !strings.Contains(mm.m.Carry.Failure(), "connection refused") { + t.Fatalf("the run answered %v, its failure %q", err, mm.m.Carry.Failure()) + } + built = mm.m + t.Cleanup(func() { built = nil }) + served := map[string]stdTool{} + for _, tool := range tools() { + served[strings.TrimPrefix(tool.Name, Seat+".")] = stdTool(tool.Run) + } + for _, name := range []string{"bind", "switch", "release", "refresh", "adopt"} { + if _, err := served[name](map[string]any{"consumer": "laptop", "licence": licence1, "name": "api", "file": "/none"}); err == nil || err.Error() != NotCarried { + t.Errorf("%s answered %v", name, err) + } + } + for _, name := range []string{"licences", "bindings", "usage", "public-key"} { + if _, err := served[name](map[string]any{}); err != nil { + t.Errorf("the read verb %s answered %v", name, err) + } + } + if len(mm.vendor.exchanged) != 0 { + t.Fatalf("the vendor was called: %v", mm.vendor.exchanged) + } +} + +type stdTool func(map[string]any) (any, error) + +// design 39 bus-and-switch §11/AC3: a table that does not exist (a fresh install) is read as no rows; any other +// failure of the database is one. +func TestAMissingTableIsNoRows(t *testing.T) { + if !noTable(&pgconn.PgError{Code: "42P01", Message: `relation "offered" does not exist`}) { + t.Fatal("a missing table is a failure") + } + for _, err := range []error{&pgconn.PgError{Code: "28P01"}, errors.New("connection refused"), nil} { + if noTable(err) { + t.Fatalf("%v is read as a missing table", err) + } + } +} + +// design 39 bus-and-switch §1/AC3: no code of the manager writes to its database. +func TestNoWriteToTheDatabase(t *testing.T) { + write := regexp.MustCompile(`(?i)\b(insert\s+into|update\s+\w+\s+set|delete\s+from|create\s+(table|sequence|index)|drop\s+(table|sequence|index)|alter\s+table|truncate\s|setval\s*\(|nextval\s*\()`) + files, _ := os.ReadDir(".") + for _, f := range files { + if !strings.HasSuffix(f.Name(), ".go") || strings.HasSuffix(f.Name(), "_test.go") { + continue + } + raw, err := os.ReadFile(f.Name()) + if err != nil { + t.Fatal(err) + } + if w := write.Find(raw); w != nil { + t.Errorf("%s writes to the database: %q", f.Name(), w) + } + } +} diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/main.go b/modules/claude-licence-manager/cmd/claude-licence-manager/main.go index 839899692..dc3a4292e 100644 --- a/modules/claude-licence-manager/cmd/claude-licence-manager/main.go +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/main.go @@ -73,6 +73,10 @@ func manager() (*Manager, error) { AuditLog: stdio.Log("audit"), Now: time.Now}, Vendor: LiveVendor(), Crypt: crypt, + // Today's database is the carry-over's source alone, read once and never written (§11, A3). + Carry: &Carry{State: stdio.State("carried"), Source: func(ctx context.Context) (CarrySource, error) { + return PgSourceFromEnv(ctx) + }}, Grants: VaultGrants{Keys: keys}, Keys: keys, AskGrant: func(_ context.Context, node, publicKey string) (GrantAnswer, error) { @@ -130,7 +134,25 @@ var ( heard = map[string]Holdings{} ) +// carryOverFirst runs the carry-over at the start, before the manager hears an event, refreshes or reads usage (§11), and +// answers once it has ended: at once when `carried.postgres` exists, else when a run writes it, this one or one asked +// by hand later. +func carryOverFirst() { + <-stdio.Served() + m, err := manager() + if err != nil { + say("the carry-over cannot start: %v", err) + return + } + if !m.Carry.Ended() { + bus, _ := m.Store.(*BusStore) + _, _ = m.Carry.Run(context.Background(), m, bus) // a failure is said, and the run is asked again by hand (§44) + } + <-m.Carry.Wait() +} + func daemon() { + carryOverFirst() mu, reports := &heardMu, heard var passing sync.Mutex pass := func() { @@ -234,7 +256,8 @@ func text(a map[string]any, k string) (string, error) { } // verb is one of the seat's verbs: listed as `.`, so the runtime serves it on the seat's subject. -// What it changes is recorded as asked by its caller, as the runtime proved it (novox/hq issues 365 and 625). +// What it changes is recorded as asked by its caller, as the runtime proved it (novox/hq issues 365 and 625). A verb +// that acts answers nothing but "the carry-over has not ended" until it has (§11). func verb(name, description string, input map[string]any, run func(ctx context.Context, m *Manager, a map[string]any) (any, error)) stdio.Tool { return stdio.Tool{Name: Seat + "." + name, Description: description, Input: input, Run: func(a map[string]any) (any, error) { @@ -242,11 +265,19 @@ func verb(name, description string, input map[string]any, run func(ctx context.C if err != nil { return nil, err } + if !reads[name] { + if err := m.Carry.Acts(); err != nil { + return nil, err + } + } ctx := WithCause(context.Background(), Cause{Why: "the " + name + " verb", By: stdio.CallerOf(a)}) return run(ctx, m, a) }} } +// reads are the seat's verbs that only read, served while the carry-over has not ended (§11). +var reads = map[string]bool{"licences": true, "bindings": true, "usage": true, "public-key": true, "current": true} + func str(description string) map[string]any { return map[string]any{"type": "string", "description": description} } diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/manager.go b/modules/claude-licence-manager/cmd/claude-licence-manager/manager.go index eea1ef527..28a541204 100644 --- a/modules/claude-licence-manager/cmd/claude-licence-manager/manager.go +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/manager.go @@ -155,6 +155,8 @@ type Manager struct { Vendor Vendor // Crypt opens the grants today's database keeps at rest, for the carry-over alone (design 39 bus-and-switch §12, A3). Crypt *Crypt + // Carry is the carry-over's gate (§11): no act until it has ended. Nil is a manager with nothing to carry over. + Carry *Carry // Grants is the vault, which holds every licence's grant and access token (§15, §16). Grants Grants Keys KeyPair @@ -457,10 +459,7 @@ func (m *Manager) adoptOne(ctx context.Context, c Candidate, reports []Holdings, // bus-and-switch §15), and keeps on the licence its fingerprint and expiries alone: no token once it returns. func (m *Manager) keep(ctx context.Context, l *Licence, g FullGrant) error { grant, _ := json.Marshal(g) - access, _ := json.Marshal(map[string]any{"accessToken": g.AccessToken, "expiresAt": g.ExpiresAt, - "refreshTokenExpiresAt": g.RefreshTokenExpiresAt, "scopes": g.Scopes, - "subscriptionType": g.SubscriptionType, "rateLimitTier": g.RateLimitTier}) - if err := m.handOver(ctx, l, string(grant), string(access)); err != nil { + if err := m.handOver(ctx, l, string(grant), accessOf(g)); err != nil { return err } l.RefreshFingerprint = Fingerprint(g.RefreshToken) @@ -473,6 +472,14 @@ func (m *Manager) keep(ctx context.Context, l *Licence, g FullGrant) error { return nil } +// accessOf is a subscription's `access` secret: its access token and expiry, never its refresh token (§15). +func accessOf(g FullGrant) string { + access, _ := json.Marshal(map[string]any{"accessToken": g.AccessToken, "expiresAt": g.ExpiresAt, + "refreshTokenExpiresAt": g.RefreshTokenExpiresAt, "scopes": g.Scopes, + "subscriptionType": g.SubscriptionType, "rateLimitTier": g.RateLimitTier}) + return string(access) +} + // handOver puts a licence's two secrets in the vault at its next generation, the grant first (§15). func (m *Manager) handOver(ctx context.Context, l *Licence, grant, access string) error { generation := l.Generation + 1 diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/manager_test.go b/modules/claude-licence-manager/cmd/claude-licence-manager/manager_test.go index 87a501ed5..d31471ac6 100644 --- a/modules/claude-licence-manager/cmd/claude-licence-manager/manager_test.go +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/manager_test.go @@ -63,7 +63,9 @@ func (g *memGrants) Put(_ context.Context, key, part string, generation int64, v g.mu.Lock() defer g.mu.Unlock() name := SecretName(key, part) - if h, ok := g.held[name]; ok && h.generation >= generation { + if h, ok := g.held[name]; ok && h.generation == generation && h.value == value { + return nil // the vault keeps what it holds, unchanged + } else if ok && h.generation >= generation { return fmt.Errorf("%s is held at generation %d, and generation %d is not newer", name, h.generation, generation) } g.held[name] = heldGrant{generation, value} diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/pgstore.go b/modules/claude-licence-manager/cmd/claude-licence-manager/pgstore.go index 7c544d6ab..f6144ddf4 100644 --- a/modules/claude-licence-manager/cmd/claude-licence-manager/pgstore.go +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/pgstore.go @@ -1,71 +1,27 @@ package main -// The store on the mesh's postgres (novox/hq design 39 §1), the database the mesh provisioned for this -// module. The schema is brought to this version's shape by the preparation step (ADR 0135), each -// statement idempotent. +// Today's database, read once by the carry-over and never written (design 39 bus-and-switch §1, §11, A3): the +// module's own, opened with the credential the mesh still provisions for it alone. Every query is a select; a table +// that does not exist (a fresh install) is read as no rows. import ( "context" "encoding/json" "errors" - "fmt" "os" "strings" "time" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" ) -// Schema is what this version needs. -var Schema = []string{ - `create table if not exists 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 if not exists binding_generation`, - `create table if not exists binding ( - consumer text primary key, - licence text not null references licence(name), - generation bigint not null)`, - `create table if not exists lease ( - key text primary key, - holder text not null, - until timestamptz not null)`, - `create table if not exists 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 if not exists usage ( - licence text not null, - at bigint not null, - reading jsonb not null, - raw jsonb not null)`, - `create index if not exists usage_by_licence on usage (licence, at desc)`, - `create table if not exists audit ( - at timestamptz not null default now(), - what text not null, - detail jsonb not null)`, -} +// PgSource reads today's database. +type PgSource struct{ pool *pgxpool.Pool } -// PgStore is the store on postgres. -type PgStore struct{ pool *pgxpool.Pool } - -// PgStoreFromEnv opens the store the mesh provisioned, its URL in the file DATABASE_URL_FILE names. -func PgStoreFromEnv(ctx context.Context) (*PgStore, error) { +// PgSourceFromEnv opens the database the mesh provisioned, its URL in the file DATABASE_URL_FILE names. +func PgSourceFromEnv(ctx context.Context) (*PgSource, error) { file := os.Getenv("DATABASE_URL_FILE") if file == "" { return nil, errors.New("DATABASE_URL_FILE is not set: the manager's database is a requirement the mesh resolves") @@ -74,219 +30,131 @@ func PgStoreFromEnv(ctx context.Context) (*PgStore, error) { if err != nil { return nil, err } - return OpenPgStore(ctx, strings.TrimSpace(string(raw))) + return OpenPgSource(ctx, strings.TrimSpace(string(raw))) } -// OpenPgStore opens a store at a URL. -func OpenPgStore(ctx context.Context, url string) (*PgStore, error) { +// OpenPgSource opens a database at a URL, and reaches it once, so an unreachable one fails here. +func OpenPgSource(ctx context.Context, url string) (*PgSource, error) { pool, err := pgxpool.New(ctx, url) if err != nil { return nil, err } - return &PgStore{pool: pool}, nil + if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, err + } + return &PgSource{pool: pool}, nil } -// Migrate brings the schema to this version's shape. -func (s *PgStore) Migrate(ctx context.Context) error { - for _, q := range Schema { - if _, err := s.pool.Exec(ctx, q); err != nil { - return fmt.Errorf("%s: %w", strings.Fields(q)[0:6], err) +// Close lets the database go. +func (s *PgSource) Close() { s.pool.Close() } + +// noTable is whether err is postgres saying a table does not exist (SQLSTATE 42P01). +func noTable(err error) bool { + var pg *pgconn.PgError + return errors.As(err, &pg) && pg.Code == "42P01" +} + +// each runs a select and scans each row; a table that does not exist is no rows. +func (s *PgSource) each(ctx context.Context, q string, scan func(pgx.Rows) error, args ...any) error { + rows, err := s.pool.Query(ctx, q, args...) + if err != nil { + if noTable(err) { + return nil } + return err + } + defer rows.Close() + for rows.Next() { + if err := scan(rows); err != nil { + return err + } + } + if err := rows.Err(); err != nil && !noTable(err) { + return err } return nil } -const licenceColumns = `name, kind, coalesce(account_uuid,''), coalesce(email,''), coalesce(organization_uuid,''), - coalesce(sealed,''), coalesce(refresh_fingerprint,''), coalesce(access_expires_at,0), coalesce(refresh_expires_at,0), - failures, coalesce(notified_at,0), adopted_at, coalesce(rotated_at,0)` - -func scanLicence(row pgx.Row) (*Licence, error) { - var l Licence - var sealed string // the grant at rest, read by the carry-over alone (design 39 bus-and-switch §12) - err := row.Scan(&l.Name, &l.Kind, &l.AccountUUID, &l.Email, &l.OrganizationUUID, &sealed, &l.RefreshFingerprint, - &l.AccessExpiresAt, &l.RefreshExpiresAt, &l.Failures, &l.NotifiedAt, &l.AdoptedAt, &l.RotatedAt) - if errors.Is(err, pgx.ErrNoRows) { - return nil, nil - } - return &l, err +func (s *PgSource) Licences(ctx context.Context) ([]LicenceRow, error) { + out := []LicenceRow{} + err := s.each(ctx, `select name, kind, coalesce(account_uuid,''), coalesce(email,''), coalesce(organization_uuid,''), + coalesce(sealed,''), coalesce(refresh_fingerprint,''), coalesce(access_expires_at,0), + coalesce(refresh_expires_at,0), failures, coalesce(notified_at,0), adopted_at, coalesce(rotated_at,0) + from licence order by adopted_at, name`, func(r pgx.Rows) error { + var l LicenceRow + err := r.Scan(&l.Name, &l.Kind, &l.AccountUUID, &l.Email, &l.OrganizationUUID, &l.Sealed, &l.RefreshFingerprint, + &l.AccessExpiresAt, &l.RefreshExpiresAt, &l.Failures, &l.NotifiedAt, &l.AdoptedAt, &l.RotatedAt) + out = append(out, l) + return err + }) + return out, err } -func (s *PgStore) Licences(ctx context.Context) ([]Licence, error) { - rows, err := s.pool.Query(ctx, `select `+licenceColumns+` from licence order by name`) - if err != nil { - return nil, err - } - defer rows.Close() - var out []Licence - for rows.Next() { - l, err := scanLicence(rows) - if err != nil { - return nil, err +func (s *PgSource) Bindings(ctx context.Context) ([]Binding, error) { + out := []Binding{} + err := s.each(ctx, `select consumer, licence, generation from binding order by generation, consumer`, + func(r pgx.Rows) error { + var b Binding + err := r.Scan(&b.Consumer, &b.Licence, &b.Generation) + out = append(out, b) + return err + }) + return out, err +} + +func (s *PgSource) Leases(ctx context.Context) (int, error) { + n := 0 + err := s.each(ctx, `select key from lease`, func(r pgx.Rows) error { + n++ + var k string + return r.Scan(&k) + }) + return n, err +} + +func (s *PgSource) Offered(ctx context.Context) ([]OfferedRow, error) { + out := []OfferedRow{} + err := s.each(ctx, `select fingerprint, node, coalesce(account_uuid,''), outcome, why, at from offered + order by at, fingerprint`, func(r pgx.Rows) error { + var o OfferedRow + var outcome string + err := r.Scan(&o.Fingerprint, &o.Node, &o.Account, &outcome, &o.Why, &o.At) + o.Outcome = Outcome(outcome) + out = append(out, o) + return err + }) + return out, err +} + +func (s *PgSource) Usage(ctx context.Context, since time.Time) ([]UsageTableRow, error) { + out := []UsageTableRow{} + err := s.each(ctx, `select licence, at, reading, raw from usage where at >= $1 order by at, licence`, + func(r pgx.Rows) error { + var u UsageTableRow + var reading, raw []byte + if err := r.Scan(&u.Licence, &u.At, &reading, &raw); err != nil { + return err + } + _ = json.Unmarshal(reading, &u.Reading) + _ = json.Unmarshal(raw, &u.Raw) + out = append(out, u) + return nil + }, since.UnixMilli()) + return out, err +} + +func (s *PgSource) Audit(ctx context.Context) ([]AuditRow, error) { + out := []AuditRow{} + err := s.each(ctx, `select at, what, detail from audit order by at`, func(r pgx.Rows) error { + var a AuditRow + var detail []byte + if err := r.Scan(&a.At, &a.What, &detail); err != nil { + return err } - out = append(out, *l) - } - return out, rows.Err() -} - -func (s *PgStore) Licence(ctx context.Context, name string) (*Licence, error) { - return scanLicence(s.pool.QueryRow(ctx, `select `+licenceColumns+` from licence where name = $1`, name)) -} - -func (s *PgStore) LicenceForAccount(ctx context.Context, account string) (*Licence, error) { - return scanLicence(s.pool.QueryRow(ctx, `select `+licenceColumns+` from licence where account_uuid = $1`, account)) -} - -func nullable(s string) any { - if s == "" { + _ = json.Unmarshal(detail, &a.Detail) + out = append(out, a) return nil - } - return s + }) + return out, err } - -func nullableInt(v int64) any { - if v == 0 { - return nil - } - return v -} - -func (s *PgStore) SaveLicence(ctx context.Context, l Licence) error { - _, err := s.pool.Exec(ctx, `insert into licence (name, kind, account_uuid, email, organization_uuid, refresh_fingerprint, - access_expires_at, refresh_expires_at, failures, notified_at, adopted_at, rotated_at) - values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) - on conflict (name) do update set kind = excluded.kind, account_uuid = excluded.account_uuid, email = excluded.email, - organization_uuid = excluded.organization_uuid, refresh_fingerprint = excluded.refresh_fingerprint, - access_expires_at = excluded.access_expires_at, refresh_expires_at = excluded.refresh_expires_at, - failures = excluded.failures, notified_at = excluded.notified_at, adopted_at = excluded.adopted_at, - rotated_at = excluded.rotated_at`, - l.Name, l.Kind, nullable(l.AccountUUID), nullable(l.Email), nullable(l.OrganizationUUID), - nullable(l.RefreshFingerprint), nullableInt(l.AccessExpiresAt), nullableInt(l.RefreshExpiresAt), l.Failures, - nullableInt(l.NotifiedAt), l.AdoptedAt, nullableInt(l.RotatedAt)) - return err -} - -// Lease is taken in the store before a row is read (design 39 §3): a second run started together finds -// it live and does nothing. -func (s *PgStore) Lease(ctx context.Context, key, holder string, d time.Duration) (bool, error) { - tag, err := s.pool.Exec(ctx, `insert into lease (key, holder, until) values ($1, $2, now() + make_interval(secs => $3)) - on conflict (key) do update set holder = excluded.holder, until = excluded.until - where lease.until < now() or lease.holder = excluded.holder`, key, holder, d.Seconds()) - if err != nil { - return false, err - } - return tag.RowsAffected() == 1, nil -} - -func (s *PgStore) Unlease(ctx context.Context, key, holder string) error { - _, err := s.pool.Exec(ctx, `delete from lease where key = $1 and holder = $2`, key, holder) - return err -} - -func (s *PgStore) bindingsWhere(ctx context.Context, q string, args ...any) ([]Binding, error) { - rows, err := s.pool.Query(ctx, q, args...) - if err != nil { - return nil, err - } - defer rows.Close() - var out []Binding - for rows.Next() { - var b Binding - if err := rows.Scan(&b.Consumer, &b.Licence, &b.Generation); err != nil { - return nil, err - } - out = append(out, b) - } - return out, rows.Err() -} - -func (s *PgStore) Bindings(ctx context.Context) ([]Binding, error) { - return s.bindingsWhere(ctx, `select consumer, licence, generation from binding order by consumer`) -} - -func (s *PgStore) Binding(ctx context.Context, consumer string) (*Binding, error) { - out, err := s.bindingsWhere(ctx, `select consumer, licence, generation from binding where consumer = $1`, consumer) - if err != nil || len(out) == 0 { - return nil, err - } - return &out[0], nil -} - -func (s *PgStore) Bind(ctx context.Context, consumer, licence string) (Binding, error) { - out, err := s.bindingsWhere(ctx, `insert into binding (consumer, licence, generation) values ($1, $2, nextval('binding_generation')) - on conflict (consumer) do update set licence = excluded.licence, generation = excluded.generation - returning consumer, licence, generation`, consumer, licence) - if err != nil { - return Binding{}, err - } - return out[0], nil -} - -func (s *PgStore) Unbind(ctx context.Context, consumer string) (bool, error) { - tag, err := s.pool.Exec(ctx, `delete from binding where consumer = $1`, consumer) - return err == nil && tag.RowsAffected() == 1, err -} - -func (s *PgStore) GenerationsAbove(ctx context.Context, generation int64) error { - if generation <= 0 { - return nil - } - // setval with is_called, so the next nextval is generation+1; only ever forward. - _, err := s.pool.Exec(ctx, `select setval('binding_generation', $1, true) from binding_generation - where not is_called or last_value < $1`, generation) - return err -} - -func (s *PgStore) Advance(ctx context.Context, licence string) ([]Binding, error) { - return s.bindingsWhere(ctx, `update binding set generation = nextval('binding_generation') where licence = $1 - returning consumer, licence, generation`, licence) -} - -func (s *PgStore) Outcome(ctx context.Context, fp string) (Outcome, error) { - var o string - err := s.pool.QueryRow(ctx, `select outcome from offered where fingerprint = $1`, fp).Scan(&o) - if errors.Is(err, pgx.ErrNoRows) { - return "", nil - } - return Outcome(o), err -} - -func (s *PgStore) RecordOutcome(ctx context.Context, fp, node, account string, o Outcome, why string) error { - _, err := s.pool.Exec(ctx, `insert into offered (fingerprint, node, account_uuid, outcome, why) values ($1,$2,$3,$4,$5) - on conflict (fingerprint) do update set outcome = excluded.outcome, why = excluded.why, at = now()`, - fp, node, nullable(account), string(o), why) - return err -} - -func (s *PgStore) RecordUsage(ctx context.Context, licence string, at int64, r UsageReading, raw map[string]any) error { - reading, _ := json.Marshal(r) - rawJSON, _ := json.Marshal(raw) - _, err := s.pool.Exec(ctx, `insert into usage (licence, at, reading, raw) values ($1,$2,$3,$4)`, licence, at, reading, rawJSON) - return err -} - -func (s *PgStore) Usage(ctx context.Context, licence string, limit int) ([]UsageRow, error) { - rows, err := s.pool.Query(ctx, `select licence, at, reading from usage where ($1 = '' or licence = $1) order by at desc limit $2`, licence, limit) - if err != nil { - return nil, err - } - defer rows.Close() - var out []UsageRow - for rows.Next() { - var u UsageRow - var reading []byte - if err := rows.Scan(&u.Licence, &u.At, &reading); err != nil { - return nil, err - } - _ = json.Unmarshal(reading, &u.Reading) - out = append(out, u) - } - return out, rows.Err() -} - -func (s *PgStore) Audit(ctx context.Context, what string, detail map[string]any) error { - raw, _ := json.Marshal(detail) - _, err := s.pool.Exec(ctx, `insert into audit (what, detail) values ($1, $2)`, what, raw) - return err -} - -func (s *PgStore) Close() { s.pool.Close() } diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/store_test.go b/modules/claude-licence-manager/cmd/claude-licence-manager/store_test.go index 498935b83..35f99deab 100644 --- a/modules/claude-licence-manager/cmd/claude-licence-manager/store_test.go +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/store_test.go @@ -1,97 +1,12 @@ package main import ( - "context" "crypto/rand" "encoding/base64" - "os" "strings" "testing" - "time" ) -// Against a real postgres, because the questions are the database's: does the schema apply twice, does a -// lease refuse a second holder, does a generation only grow. Skipped unless one is named: -// -// docker run -d --rm --name licmgr-pg -e POSTGRES_PASSWORD=t -p 15498:5432 postgres:16-alpine -// MESH_TEST_POSTGRES=postgres://postgres:t@127.0.0.1:15498/postgres go test ./... -func TestTheStoreOnPostgres(t *testing.T) { - url := os.Getenv("MESH_TEST_POSTGRES") - if url == "" { - t.Skip("MESH_TEST_POSTGRES unset") - } - ctx := context.Background() - s, err := OpenPgStore(ctx, url) - if err != nil { - t.Fatal(err) - } - defer s.Close() - for _, table := range []string{"binding", "licence", "lease", "offered", "usage", "audit"} { - _, _ = s.pool.Exec(ctx, "drop table if exists "+table+" cascade") - } - _, _ = s.pool.Exec(ctx, "drop sequence if exists binding_generation") - for i := 0; i < 2; i++ { - if err := s.Migrate(ctx); err != nil { - t.Fatalf("migration %d: %v", i+1, err) - } - } - l := Licence{Name: "a@example.org", Kind: "subscription", AccountUUID: "u-1", Email: "a@example.org", - RefreshFingerprint: "sha256:1", AccessExpiresAt: 1, RefreshExpiresAt: 2, AdoptedAt: 3} - if err := s.SaveLicence(ctx, l); err != nil { - t.Fatal(err) - } - l.Failures = 2 - _ = s.SaveLicence(ctx, l) - if got, _ := s.LicenceForAccount(ctx, "u-1"); got == nil || got.Failures != 2 || got.OrganizationUUID != "" { - t.Fatalf("%+v", got) - } - if none, err := s.Licence(ctx, "nobody"); none != nil || err != nil { - t.Fatalf("%v %v", none, err) - } - lease := func(holder string) bool { - ok, err := s.Lease(ctx, "licence:a", holder, time.Minute) - if err != nil { - t.Fatal(err) - } - return ok - } - if !lease("one") || lease("two") || !lease("one") { - t.Fatal("a lease did not refuse a second holder, or its own holder could not renew it") - } - _ = s.Unlease(ctx, "licence:a", "one") - if !lease("two") { - t.Fatal("a released lease was not taken") - } - b1, _ := s.Bind(ctx, "laptop", l.Name) - adv, _ := s.Advance(ctx, l.Name) - b3, _ := s.Bind(ctx, "laptop", l.Name) - if len(adv) != 1 || !(b1.Generation < adv[0].Generation && adv[0].Generation < b3.Generation) { - t.Fatalf("generations %d %v %d", b1.Generation, adv, b3.Generation) - } - _ = s.RecordOutcome(ctx, "sha256:x", "laptop", "u-1", Dead, "400") - if o, _ := s.Outcome(ctx, "sha256:x"); o != Dead { - t.Fatalf("outcome %q", o) - } - if o, _ := s.Outcome(ctx, "sha256:none"); o != "" { - t.Fatalf("an unknown login is %q", o) - } - pct := 1.0 - _ = s.RecordUsage(ctx, l.Name, 5, UsageReading{SessionPct: &pct}, map[string]any{}) - if u, _ := s.Usage(ctx, "", 5); len(u) != 1 || *u[0].Reading.SessionPct != 1 { - t.Fatalf("usage %v", u) - } - if err := s.Audit(ctx, "bound", map[string]any{"consumer": "laptop"}); err != nil { - t.Fatal(err) - } - if ok, _ := s.Unbind(ctx, "laptop"); !ok { - t.Fatal("unbind") - } - all, _ := s.Licences(ctx) - if len(all) != 1 { - t.Fatalf("%v", all) - } -} - func TestARefreshThatDoesNotRotateKeepsTheRefreshToken(t *testing.T) { prev := FullGrant{AccessToken: "a", RefreshToken: "r", ExpiresAt: 1} in := int64(60)