package process import ( "context" "encoding/json" "errors" "fmt" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" ) // stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is // caught at once by Conn's own ErrorHandler (conn.go withRefusal/publishRefused), never waited out, so // this stays the ordinary request bound and is not exposed for a test to shrink. const stateTimeout = 30 * time.Second // State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436 // clause 4), in the bucket _ (mesh-controller internal/broker/state.go:49) — only as the // account's grants allow. It satisfies go/asker's Store. type State struct { Module string Name string // kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a // unit test, a fake that never reaches a bus. kv func(ctx context.Context) (jetstream.KeyValue, error) } // bucket is this state's name on the bus. func (s State) bucket() string { return s.Module + "_" + s.Name } // State is the named state this connection's module keeps, bound lazily to its bucket on each call so a // bucket the account is not granted fails at the call that needed it, naming the bucket. The bind itself // (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_, so it is what withRefusal watches: a bucket // not granted is refused there, promptly, as nats.ErrPermissionViolation. func (c *Conn) State(name string) State { bucket := c.account.Module + "_" + name return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) { js, err := jetstream.New(c.nc) if err != nil { return nil, err } subject := "$JS.API.STREAM.INFO.KV_" + bucket ctx, stop := c.withRefusal(ctx, subject) defer stop() kv, err := js.KeyValue(ctx, bucket) if err != nil { if errors.Is(context.Cause(ctx), errPermissionRefused) { return nil, fmt.Errorf("%w: publishing %s", nats.ErrPermissionViolation, subject) } return nil, err } return kv, nil }} } func (s State) store(ctx context.Context) (jetstream.KeyValue, error) { kv, err := s.kv(ctx) if err != nil { return nil, fmt.Errorf("%s's state %s (bucket %s): %w", s.Module, s.Name, s.bucket(), err) } return kv, nil } // Get is a key's current value, or nil when it has none. func (s State) Get(key string) (*stdio.StateEntry, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) if err != nil { return nil, err } entry, err := kv.Get(ctx, key) if errors.Is(err, jetstream.ErrKeyNotFound) { return nil, nil } if err != nil { return nil, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) } return &stdio.StateEntry{Key: key, Value: json.RawMessage(entry.Value()), Revision: entry.Revision()}, nil } // Create writes a key only when it has none, and returns the revision; a key that has one is refused as // stdio.ErrChanged. func (s State) Create(key string, value any) (uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) if err != nil { return 0, err } data, err := json.Marshal(value) if err != nil { return 0, err } rev, err := kv.Create(ctx, key, data) if errors.Is(err, jetstream.ErrKeyExists) { return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) } if err != nil { return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) } return rev, nil } // Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile // is refused as stdio.ErrChanged. func (s State) Update(key string, value any, revision uint64) (uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() kv, err := s.store(ctx) if err != nil { return 0, err } data, err := json.Marshal(value) if err != nil { return 0, err } rev, err := kv.Update(ctx, key, data, revision) if errors.Is(err, jetstream.ErrKeyRevisionMismatch) { return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) } if err != nil { return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) } return rev, nil }