From d43c0e43022b02ad4deeee75082d114e5a2cc170 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 26 Aug 2026 15:16:28 +1000 Subject: [PATCH] feat: add kvstore typed store and convert the ochrevector job store (#888) --- spinifex/handlers/ochrevector/ingest_test.go | 23 +-- spinifex/handlers/ochrevector/jobs.go | 170 +++-------------- spinifex/handlers/ochrevector/jobs_test.go | 6 +- spinifex/kvstore/bucket.go | 68 +++++++ spinifex/kvstore/store.go | 159 ++++++++++++++++ spinifex/kvstore/store_test.go | 186 +++++++++++++++++++ 6 files changed, 449 insertions(+), 163 deletions(-) create mode 100644 spinifex/kvstore/bucket.go create mode 100644 spinifex/kvstore/store.go create mode 100644 spinifex/kvstore/store_test.go diff --git a/spinifex/handlers/ochrevector/ingest_test.go b/spinifex/handlers/ochrevector/ingest_test.go index 1642b5044..a339f797b 100644 --- a/spinifex/handlers/ochrevector/ingest_test.go +++ b/spinifex/handlers/ochrevector/ingest_test.go @@ -6,7 +6,6 @@ package handlers_ochrevector import ( "context" - "encoding/json" "errors" "math" "strings" @@ -401,23 +400,15 @@ func TestRunJob_EmptyDocumentClearsExistingRows(t *testing.T) { assert.Empty(t, backend.documentRows(ingestAccountA, "idx-one", key)) } -// forceJobUpdatedAt seeds jobID's UpdatedAt directly via the underlying KV -// bucket, bypassing JobStore.Update (which always refreshes UpdatedAt to -// now), so a test can simulate a record a crashed worker left stale -- -// mirroring appliance_test.go's raw-KV seeding for the same reason. +// forceJobUpdatedAt seeds jobID's UpdatedAt directly, bypassing +// JobStore.Update (which always refreshes UpdatedAt to now), so a test can +// simulate a record a crashed worker left stale. func forceJobUpdatedAt(t *testing.T, store *JobStore, accountID, jobID string, when time.Time) { t.Helper() - ctx := context.Background() - kv, err := store.bucket(ctx) - require.NoError(t, err) - entry, err := kv.Get(ctx, jobKey(accountID, jobID)) - require.NoError(t, err) - var rec JobRecord - require.NoError(t, json.Unmarshal(entry.Value(), &rec)) - rec.UpdatedAt = when - data, err := json.Marshal(rec) - require.NoError(t, err) - _, err = kv.Update(ctx, jobKey(accountID, jobID), data, entry.Revision()) + err := store.store.Mutate(context.Background(), jobKey(accountID, jobID), func(rec *JobRecord) (bool, error) { + rec.UpdatedAt = when + return true, nil + }) require.NoError(t, err) } diff --git a/spinifex/handlers/ochrevector/jobs.go b/spinifex/handlers/ochrevector/jobs.go index d191f89e4..884984f5f 100644 --- a/spinifex/handlers/ochrevector/jobs.go +++ b/spinifex/handlers/ochrevector/jobs.go @@ -2,14 +2,11 @@ package handlers_ochrevector import ( "context" - "encoding/json" "errors" "fmt" - "strings" - "sync" "time" - "github.com/mulgadc/spinifex/spinifex/kvutil" + "github.com/mulgadc/spinifex/spinifex/kvstore" "github.com/nats-io/nats.go/jetstream" ) @@ -66,36 +63,18 @@ type JobRecord struct { } // JobStore persists JobRecords in the ochre-vector-jobs JetStream KV bucket, -// mirroring Registry's lazy get-or-create bucket, key "accountID/id", -// per-account prefix-scan list. +// key "accountID/id", per-account prefix-scan list. type JobStore struct { - js jetstream.JetStream - - mu sync.Mutex - kv jetstream.KeyValue + store *kvstore.Store[JobRecord] } // NewJobStore constructs a JobStore over js. func NewJobStore(js jetstream.JetStream) *JobStore { - return &JobStore{js: js} -} - -// bucket lazily opens (or creates) the jobs KV bucket, caching the handle. -func (s *JobStore) bucket(ctx context.Context) (jetstream.KeyValue, error) { - if s.js == nil { - return nil, errors.New("ochrevector: job store has no JetStream client configured") - } - s.mu.Lock() - defer s.mu.Unlock() - if s.kv != nil { - return s.kv, nil - } - kv, err := kvutil.GetOrCreateBucket(ctx, s.js, jobsBucket, jobsBucketHistory) - if err != nil { - return nil, err - } - s.kv = kv - return kv, nil + return &JobStore{store: kvstore.New[JobRecord](js, kvstore.Config{ + Name: jobsBucket, + History: jobsBucketHistory, + Missing: "ochrevector: job store has no JetStream client configured", + })} } // jobKey scopes every record to its owning account, so a foreign account's @@ -109,17 +88,9 @@ func jobKey(accountID, jobID string) string { // starts of the same job id race safely and exactly one wins. rec.AccountID // is stamped from accountID, overriding whatever the caller set. func (s *JobStore) Reserve(ctx context.Context, accountID string, rec JobRecord) error { - kv, err := s.bucket(ctx) - if err != nil { - return err - } rec.AccountID = accountID - data, err := json.Marshal(rec) - if err != nil { - return fmt.Errorf("ochrevector: encode job record %s: %w", rec.ID, err) - } - if _, err := kv.Create(ctx, jobKey(accountID, rec.ID), data); err != nil { - if errors.Is(err, jetstream.ErrKeyExists) { + if err := s.store.Create(ctx, jobKey(accountID, rec.ID), &rec); err != nil { + if errors.Is(err, kvstore.ErrExists) { return ErrJobExists } return fmt.Errorf("ochrevector: reserve job %s: %w", rec.ID, err) @@ -129,74 +100,23 @@ func (s *JobStore) Reserve(ctx context.Context, accountID string, rec JobRecord) // Get reads accountID's record for jobID, returning (nil, nil) when absent. func (s *JobStore) Get(ctx context.Context, accountID, jobID string) (*JobRecord, error) { - kv, err := s.bucket(ctx) - if err != nil { - return nil, err + rec, _, err := s.store.Get(ctx, jobKey(accountID, jobID)) + if errors.Is(err, kvstore.ErrNotFound) { + return nil, nil } - return getJobRecord(ctx, kv, jobKey(accountID, jobID)) -} - -// getJobRecord reads and decodes one record, returning (nil, nil) when absent. -func getJobRecord(ctx context.Context, kv jetstream.KeyValue, key string) (*JobRecord, error) { - entry, err := kv.Get(ctx, key) - if err != nil { - if errors.Is(err, jetstream.ErrKeyNotFound) { - return nil, nil - } - return nil, fmt.Errorf("ochrevector: kv get %s: %w", key, err) - } - var rec JobRecord - if err := json.Unmarshal(entry.Value(), &rec); err != nil { - return nil, fmt.Errorf("ochrevector: decode %s: %w", key, err) - } - return &rec, nil + return rec, err } // List returns every job record owned by accountID, so one tenant's listing // never surfaces another's. func (s *JobStore) List(ctx context.Context, accountID string) ([]JobRecord, error) { - kv, err := s.bucket(ctx) - if err != nil { - return nil, err - } - return listJobRecords(ctx, kv, accountID+"/") + return s.store.List(ctx, accountID+"/") } // ListAll returns every job record across every account, for the ingestion // reconciler's crash-recovery sweep. func (s *JobStore) ListAll(ctx context.Context) ([]JobRecord, error) { - kv, err := s.bucket(ctx) - if err != nil { - return nil, err - } - return listJobRecords(ctx, kv, "") -} - -// listJobRecords walks every key with the given prefix ("" matches -// everything) and decodes each into a JobRecord, skipping any key that -// disappears between the key listing and the read. -func listJobRecords(ctx context.Context, kv jetstream.KeyValue, prefix string) ([]JobRecord, error) { - keys, err := kv.Keys(ctx) - if err != nil { - if errors.Is(err, jetstream.ErrNoKeysFound) { - return nil, nil - } - return nil, fmt.Errorf("ochrevector: list keys: %w", err) - } - var out []JobRecord - for _, key := range keys { - if !strings.HasPrefix(key, prefix) { - continue - } - rec, err := getJobRecord(ctx, kv, key) - if err != nil { - return nil, err - } - if rec != nil { - out = append(out, *rec) - } - } - return out, nil + return s.store.List(ctx, "") } // Update reads accountID's jobID record, applies fn to mutate it in place, @@ -204,29 +124,15 @@ func listJobRecords(ctx context.Context, kv jetstream.KeyValue, prefix string) ( // update, mirroring Registry.SetState's pattern for the general case of an // arbitrary field mutation (progress counters, failed-document list, error). func (s *JobStore) Update(ctx context.Context, accountID, jobID string, fn func(rec *JobRecord)) error { - kv, err := s.bucket(ctx) + err := s.store.Mutate(ctx, jobKey(accountID, jobID), func(rec *JobRecord) (bool, error) { + fn(rec) + rec.UpdatedAt = time.Now().UTC() + return true, nil + }) + if errors.Is(err, kvstore.ErrNotFound) { + return ErrJobNotFound + } if err != nil { - return err - } - key := jobKey(accountID, jobID) - entry, err := kv.Get(ctx, key) - if err != nil { - if errors.Is(err, jetstream.ErrKeyNotFound) { - return ErrJobNotFound - } - return fmt.Errorf("ochrevector: kv get %s: %w", key, err) - } - var rec JobRecord - if err := json.Unmarshal(entry.Value(), &rec); err != nil { - return fmt.Errorf("ochrevector: decode %s: %w", key, err) - } - fn(&rec) - rec.UpdatedAt = time.Now().UTC() - data, err := json.Marshal(rec) - if err != nil { - return fmt.Errorf("ochrevector: encode job record %s: %w", rec.ID, err) - } - if _, err := kv.Update(ctx, key, data, entry.Revision()); err != nil { return fmt.Errorf("ochrevector: update job %s: %w", jobID, err) } return nil @@ -244,35 +150,11 @@ func (s *JobStore) SetState(ctx context.Context, accountID, jobID, state string) // appliance's job history never references data that no longer exists. // Idempotent: an empty bucket is a no-op success. func (s *JobStore) PurgeAll(ctx context.Context) error { - kv, err := s.bucket(ctx) - if err != nil { - return err - } - keys, err := kv.Keys(ctx) - if err != nil { - if errors.Is(err, jetstream.ErrNoKeysFound) { - return nil - } - return fmt.Errorf("ochrevector: list keys: %w", err) - } - for _, key := range keys { - if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { - return fmt.Errorf("ochrevector: purge job %s: %w", key, err) - } - } - return nil + return s.store.DeletePrefix(ctx, "") } // Delete removes accountID's jobID record. Idempotent: deleting an // already-absent record is a no-op success. func (s *JobStore) Delete(ctx context.Context, accountID, jobID string) error { - kv, err := s.bucket(ctx) - if err != nil { - return err - } - key := jobKey(accountID, jobID) - if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { - return fmt.Errorf("ochrevector: delete job %s: %w", jobID, err) - } - return nil + return s.store.Delete(ctx, jobKey(accountID, jobID)) } diff --git a/spinifex/handlers/ochrevector/jobs_test.go b/spinifex/handlers/ochrevector/jobs_test.go index 494f2a219..a027a1150 100644 --- a/spinifex/handlers/ochrevector/jobs_test.go +++ b/spinifex/handlers/ochrevector/jobs_test.go @@ -196,7 +196,7 @@ func TestJobStore_ListAll(t *testing.T) { } func TestJobStore_BucketRequiresJetStream(t *testing.T) { - store := &JobStore{} - _, err := store.bucket(context.Background()) - assert.Error(t, err) + store := NewJobStore(nil) + _, err := store.Get(context.Background(), "acct-a", "job-1") + require.ErrorContains(t, err, "job store has no JetStream client configured") } diff --git a/spinifex/kvstore/bucket.go b/spinifex/kvstore/bucket.go new file mode 100644 index 000000000..b9c053e29 --- /dev/null +++ b/spinifex/kvstore/bucket.go @@ -0,0 +1,68 @@ +// Package kvstore provides a typed store over a single JetStream KV bucket: +// one memoised get-or-create accessor and a JSON codec for the record type, +// replacing the accessor each caller used to hand-roll. +package kvstore + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/mulgadc/spinifex/spinifex/kvutil" + "github.com/nats-io/nats.go/jetstream" +) + +// Config describes the bucket that a Store or Bucket sits over. +type Config struct { + Name string + History int + Replicas int + TTL time.Duration + Missing string +} + +// Bucket memoises a lazily created KV bucket. Construct it with NewBucket. +type Bucket struct { + js jetstream.JetStream + cfg Config + + mu sync.Mutex + kv jetstream.KeyValue +} + +// NewBucket returns a Bucket over js. A nil js is permitted: every call to KV then +// fails with cfg.Missing rather than panicking. +func NewBucket(js jetstream.JetStream, cfg Config) *Bucket { + return &Bucket{js: js, cfg: cfg} +} + +// KV opens the bucket on first use and returns the cached handle thereafter. +func (b *Bucket) KV(ctx context.Context) (jetstream.KeyValue, error) { + if b.js == nil { + return nil, errors.New(b.cfg.Missing) + } + b.mu.Lock() + defer b.mu.Unlock() + if b.kv != nil { + return b.kv, nil + } + kv, err := b.open(ctx) + if err != nil { + return nil, err + } + b.kv = kv + return kv, nil +} + +// open picks the kvutil helper matching the configured TTL and replica count. +func (b *Bucket) open(ctx context.Context) (jetstream.KeyValue, error) { + switch { + case b.cfg.TTL > 0: + return kvutil.GetOrCreateBucketWithTTL(ctx, b.js, b.cfg.Name, b.cfg.History, b.cfg.TTL) + case b.cfg.Replicas > 0: + return kvutil.GetOrCreateBucketWithReplicas(ctx, b.js, b.cfg.Name, b.cfg.History, b.cfg.Replicas) + default: + return kvutil.GetOrCreateBucket(ctx, b.js, b.cfg.Name, b.cfg.History) + } +} diff --git a/spinifex/kvstore/store.go b/spinifex/kvstore/store.go new file mode 100644 index 000000000..803d79941 --- /dev/null +++ b/spinifex/kvstore/store.go @@ -0,0 +1,159 @@ +package kvstore + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "github.com/mulgadc/spinifex/spinifex/kvutil" + "github.com/nats-io/nats.go/jetstream" +) + +// Sentinels for the two conditions every caller distinguishes. Callers map +// these onto their own package errors. +var ( + ErrNotFound = errors.New("kvstore: key not found") + ErrExists = errors.New("kvstore: key already exists") +) + +// Store is a Bucket plus a JSON codec for T. +type Store[T any] struct { + *Bucket +} + +// New returns a Store over js for records of type T. +func New[T any](js jetstream.JetStream, cfg Config) *Store[T] { + return &Store[T]{Bucket: NewBucket(js, cfg)} +} + +// Get reads and decodes one record, returning ErrNotFound when the key is +// absent. The revision is returned for callers doing their own CAS. +func (s *Store[T]) Get(ctx context.Context, key string) (*T, uint64, error) { + kv, err := s.KV(ctx) + if err != nil { + return nil, 0, err + } + return s.get(ctx, kv, key) +} + +// Create writes a record only if the key is absent, returning ErrExists when +// it is not. The create-only write is the single-writer claim. +func (s *Store[T]) Create(ctx context.Context, key string, v *T) error { + kv, err := s.KV(ctx) + if err != nil { + return err + } + data, err := json.Marshal(v) + if err != nil { + return fmt.Errorf("kvstore: encode %s: %w", key, err) + } + if _, err := kv.Create(ctx, key, data); err != nil { + if errors.Is(err, jetstream.ErrKeyExists) { + return fmt.Errorf("%w: %s", ErrExists, key) + } + return fmt.Errorf("kvstore: create %s: %w", key, err) + } + return nil +} + +// Mutate applies fn under a revision-guarded retry loop. fn reports whether it +// changed anything; false commits no write. An absent key is ErrNotFound. +func (s *Store[T]) Mutate(ctx context.Context, key string, fn func(*T) (bool, error)) error { + kv, err := s.KV(ctx) + if err != nil { + return err + } + _, err = kvutil.Update(ctx, kv, key, kvutil.CASConfig{ + NotFound: fmt.Errorf("%w: %s", ErrNotFound, key), + }, fn) + return err +} + +// Delete removes a key. Idempotent: an already-absent key is success. +func (s *Store[T]) Delete(ctx context.Context, key string) error { + kv, err := s.KV(ctx) + if err != nil { + return err + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("kvstore: delete %s: %w", key, err) + } + return nil +} + +// List decodes every record whose key carries the given prefix. An empty +// prefix matches everything and an empty bucket is not an error. A key that +// disappears between the listing and the read is skipped. +func (s *Store[T]) List(ctx context.Context, prefix string) ([]T, error) { + kv, keys, err := s.keys(ctx) + if err != nil { + return nil, err + } + var out []T + for _, key := range keys { + if !strings.HasPrefix(key, prefix) { + continue + } + v, _, err := s.get(ctx, kv, key) + if errors.Is(err, ErrNotFound) { + continue + } + if err != nil { + return nil, err + } + out = append(out, *v) + } + return out, nil +} + +// DeletePrefix removes every key carrying the given prefix. An empty prefix +// purges the bucket. Idempotent, like Delete. +func (s *Store[T]) DeletePrefix(ctx context.Context, prefix string) error { + kv, keys, err := s.keys(ctx) + if err != nil { + return err + } + for _, key := range keys { + if !strings.HasPrefix(key, prefix) { + continue + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("kvstore: delete %s: %w", key, err) + } + } + return nil +} + +// keys lists the bucket, treating an empty bucket as an empty listing. +func (s *Store[T]) keys(ctx context.Context) (jetstream.KeyValue, []string, error) { + kv, err := s.KV(ctx) + if err != nil { + return nil, nil, err + } + keys, err := kv.Keys(ctx) + if err != nil { + if errors.Is(err, jetstream.ErrNoKeysFound) { + return kv, nil, nil + } + return nil, nil, fmt.Errorf("kvstore: list keys: %w", err) + } + return kv, keys, nil +} + +// get is Get against an already-opened bucket, for loops that opened it once. +func (s *Store[T]) get(ctx context.Context, kv jetstream.KeyValue, key string) (*T, uint64, error) { + entry, err := kv.Get(ctx, key) + if err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, 0, fmt.Errorf("%w: %s", ErrNotFound, key) + } + return nil, 0, fmt.Errorf("kvstore: get %s: %w", key, err) + } + var v T + if err := json.Unmarshal(entry.Value(), &v); err != nil { + return nil, 0, fmt.Errorf("kvstore: decode %s: %w", key, err) + } + return &v, entry.Revision(), nil +} diff --git a/spinifex/kvstore/store_test.go b/spinifex/kvstore/store_test.go new file mode 100644 index 000000000..4020c9d20 --- /dev/null +++ b/spinifex/kvstore/store_test.go @@ -0,0 +1,186 @@ +package kvstore_test + +import ( + "encoding/json" + "testing" + + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// record is a stand-in for any caller's JSON document. +type record struct { + Name string `json:"name"` + Count int `json:"count"` +} + +// newStore starts an embedded JetStream server and returns a Store over a +// bucket on it. Each test gets its own server, so bucket names need not differ. +func newStore(t *testing.T) *kvstore.Store[record] { + t.Helper() + _, nc, _ := testutil.StartTestJetStream(t) + return kvstore.New[record](testutil.NewJetStream(t, nc), kvstore.Config{ + Name: "kvstore-test", + History: 1, + Missing: "kvstore test: no JetStream client configured", + }) +} + +// seedRaw writes a record straight to the bucket, bypassing Store, so a test +// can move the revision underneath an in-flight Mutate. +func seedRaw(t *testing.T, store *kvstore.Store[record], key string, rec record) { + t.Helper() + kv, err := store.KV(t.Context()) + require.NoError(t, err) + data, err := json.Marshal(rec) + require.NoError(t, err) + _, err = kv.Put(t.Context(), key, data) + require.NoError(t, err) +} + +func TestStore_GetAbsentKeyIsNotFound(t *testing.T) { + store := newStore(t) + + _, _, err := store.Get(t.Context(), "nobody") + require.ErrorIs(t, err, kvstore.ErrNotFound) + assert.ErrorContains(t, err, "nobody", "the sentinel should name the key it could not find") +} + +func TestStore_CreateRoundTripsAndRejectsDuplicate(t *testing.T) { + store := newStore(t) + + require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one", Count: 3})) + + got, rev, err := store.Get(t.Context(), "acct-a/one") + require.NoError(t, err) + assert.Equal(t, record{Name: "one", Count: 3}, *got) + assert.NotZero(t, rev, "a committed record has a revision") + + err = store.Create(t.Context(), "acct-a/one", &record{Name: "impostor"}) + require.ErrorIs(t, err, kvstore.ErrExists) + + // The loser of the create must not have overwritten the winner. + got, _, err = store.Get(t.Context(), "acct-a/one") + require.NoError(t, err) + assert.Equal(t, "one", got.Name) +} + +func TestStore_DeleteIsIdempotent(t *testing.T) { + store := newStore(t) + require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one"})) + + require.NoError(t, store.Delete(t.Context(), "acct-a/one")) + require.NoError(t, store.Delete(t.Context(), "acct-a/one"), "deleting an absent key is success") + require.NoError(t, store.Delete(t.Context(), "never-existed")) + + _, _, err := store.Get(t.Context(), "acct-a/one") + require.ErrorIs(t, err, kvstore.ErrNotFound) +} + +func TestStore_ListOnEmptyBucketIsNotAnError(t *testing.T) { + store := newStore(t) + + out, err := store.List(t.Context(), "") + require.NoError(t, err, "an empty bucket is an empty listing, not a failure") + assert.Empty(t, out) +} + +func TestStore_ListFiltersByPrefix(t *testing.T) { + store := newStore(t) + require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one"})) + require.NoError(t, store.Create(t.Context(), "acct-a/two", &record{Name: "two"})) + require.NoError(t, store.Create(t.Context(), "acct-b/three", &record{Name: "three"})) + + mine, err := store.List(t.Context(), "acct-a/") + require.NoError(t, err) + assert.Len(t, mine, 2, "one tenant's listing must not surface another's") + + all, err := store.List(t.Context(), "") + require.NoError(t, err) + assert.Len(t, all, 3, "an empty prefix matches everything") +} + +// TestStore_MutateRetriesARevisionConflict forces the conflict deterministically +// rather than racing two goroutines: the first mutate attempt writes a competing +// value out of band, so the commit that follows it must lose and re-read. +func TestStore_MutateRetriesARevisionConflict(t *testing.T) { + store := newStore(t) + require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one", Count: 1})) + + attempts := 0 + err := store.Mutate(t.Context(), "acct-a/one", func(rec *record) (bool, error) { + attempts++ + if attempts == 1 { + seedRaw(t, store, "acct-a/one", record{Name: "one", Count: 99}) + } + rec.Count++ + return true, nil + }) + require.NoError(t, err) + assert.Equal(t, 2, attempts, "the first commit loses the CAS and the loop re-runs mutate") + + got, _, err := store.Get(t.Context(), "acct-a/one") + require.NoError(t, err) + assert.Equal(t, 100, got.Count, "the retry must re-read the competing write, not reapply to a stale copy") +} + +func TestStore_MutateReportingNoChangeCommitsNoWrite(t *testing.T) { + store := newStore(t) + require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one", Count: 1})) + _, before, err := store.Get(t.Context(), "acct-a/one") + require.NoError(t, err) + + err = store.Mutate(t.Context(), "acct-a/one", func(rec *record) (bool, error) { + rec.Count = 42 + return false, nil + }) + require.NoError(t, err) + + got, after, err := store.Get(t.Context(), "acct-a/one") + require.NoError(t, err) + assert.Equal(t, 1, got.Count, "a mutate reporting no change must not commit its edits") + assert.Equal(t, before, after, "no write means no new revision") +} + +func TestStore_MutateOnAbsentKeyIsNotFound(t *testing.T) { + store := newStore(t) + + err := store.Mutate(t.Context(), "nobody", func(*record) (bool, error) { + t.Error("mutate must not run against an absent key") + return false, nil + }) + require.ErrorIs(t, err, kvstore.ErrNotFound) +} + +func TestStore_DeletePrefixLeavesOtherPrefixes(t *testing.T) { + store := newStore(t) + require.NoError(t, store.Create(t.Context(), "acct-a/one", &record{Name: "one"})) + require.NoError(t, store.Create(t.Context(), "acct-b/two", &record{Name: "two"})) + + require.NoError(t, store.DeletePrefix(t.Context(), "acct-a/")) + + remaining, err := store.List(t.Context(), "") + require.NoError(t, err) + require.Len(t, remaining, 1) + assert.Equal(t, "two", remaining[0].Name) + + require.NoError(t, store.DeletePrefix(t.Context(), "acct-a/"), "purging an already-empty prefix is success") +} + +func TestStore_NilJetStreamReportsTheConfiguredMessage(t *testing.T) { + store := kvstore.New[record](nil, kvstore.Config{ + Name: "kvstore-test", + Missing: "kvstore test: no JetStream client configured", + }) + + _, _, err := store.Get(t.Context(), "acct-a/one") + require.ErrorContains(t, err, "no JetStream client configured") + + err = store.Create(t.Context(), "acct-a/one", &record{Name: "one"}) + require.ErrorContains(t, err, "no JetStream client configured") + + _, err = store.List(t.Context(), "") + require.ErrorContains(t, err, "no JetStream client configured") +}