mesh-sdk/go/process/buckets.go

package process

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

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

// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the delivery
// builder exited on a refused read of the gate's source and crash-looped). A grant reaches the bus with a delivery, so a
// bucket, a watch or a consumer the bus refuses, or times out on, at start is often only not granted yet: exiting for it
// stops the part until a person's hand. A part opens what it cannot do its job at all without with EssentialBucket, which
// stops its start with an error naming the bucket and its grant, and everything else with OptionalBucket, which says what
// is not heard and goes on; a watch the bus refuses on an optional bucket is said with NotWatched. Here, beside State,
// because this is where a process opens what it keeps and reads on the bus; the parts that open buckets with their own
// JetStream (the delivery parts) take the same three.

// BucketOpener is the part of the bus a start opens buckets with: a jetstream.JetStream, or a test's fake.
type BucketOpener interface {
	KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error)
}

// EssentialBucket opens a bucket part cannot work without (it cannot `does` without it); its error is EssentialErr's.
func EssentialBucket(ctx context.Context, js BucketOpener, part, name, does string) (jetstream.KeyValue, error) {
	kv, err := js.KeyValue(ctx, name)
	if err != nil {
		return nil, EssentialErr(part, name, does, err)
	}
	return kv, nil
}

// EssentialErr is the error of a bucket, consumer or stream named name that part cannot work without: it says what part
// cannot do, that it stops, and the grant to look at.
func EssentialErr(part, name, does string, err error) error {
	return fmt.Errorf("%s cannot be read (%v): the %s cannot %s without it, so it stops; where the bus refused it or did not "+
		"answer, this module's grant of %s (its manifest's reads, state or consumes, issued with its credential) has not "+
		"reached the bus yet", name, err, part, does, name)
}

// OptionalBucket opens a bucket the part works without: any error is said with what is not heard (without), and nil
// answered.
func OptionalBucket(ctx context.Context, js BucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue {
	kv, err := js.KeyValue(ctx, name)
	if err != nil {
		logf("the bucket %s cannot be read (%v): %s", name, err, without)
		return nil
	}
	return kv
}

// NotWatched says that an optional bucket opened but its watch was refused or not answered: what is not heard.
func NotWatched(logf func(string, ...any), name string, err error, without string) {
	logf("the bucket %s cannot be watched (%v): %s", name, err, without)
}

// ErrNotYet is a write to an awaited bucket that is not on the bus yet: the value is kept, the latest per key, and
// written once the bucket appears.
var ErrNotYet = errors.New("the bucket is not on the bus yet: the value is kept and written when it appears")

// Subscriber is the part of a connection that hears what the bus says: a *nats.Conn, or a test's fake.
type Subscriber interface {
	Subscribe(subject string, handle nats.MsgHandler) (*nats.Subscription, error)
}

// createdAdvisory is the subject the bus says a bucket's stream was created on: the server's own advisory, which a
// principal is granted for each bucket it may bind (mesh-sdk buscompose StreamCreatedAdvisory, stateCreatedHeard).
func createdAdvisory(bucket string) string { return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket }

// bindWithin bounds one bind of an awaited bucket: one STREAM.INFO.
const bindWithin = 10 * time.Second

// AwaitedBucket is a bucket a part keeps or reads that may not be on the bus at its start, bound once it is (novox/hq
// issue 1790, met 2026-10-26: since the grants part creates a manifest's buckets after a delivery, ADR 0547, a module
// restarted by that delivery opened its new state `hearing` before it existed, got "bucket not found", and never looked
// again). It binds at AwaitBucket when it can; otherwise when the bus says the bucket's stream was created (heard on
// nc), and at each use before then (one STREAM.INFO, the cheap check), never on a timer. It says once that it waits and
// once that the bucket appeared.
type AwaitedBucket struct {
	ctx     context.Context
	js      BucketOpener
	name    string
	without string
	logf    func(string, ...any)

	nc Subscriber

	mu          sync.Mutex
	kv          jetstream.KeyValue
	refusedSaid bool
	stopStatus  func()
	pending     map[string][]byte
	order       []string
	ready       []func(jetstream.KeyValue)
	sub         *nats.Subscription
}

// AwaitBucket opens the bucket name now if the bus holds it, and otherwise waits for it as AwaitedBucket says; without
// is what is not done while it is missing. nc may be nil: then only a use binds it. ctx is the part's life: a bind
// after it ends is not tried.
func AwaitBucket(ctx context.Context, js BucketOpener, nc Subscriber, name, without string, logf func(string, ...any)) *AwaitedBucket {
	if logf == nil {
		logf = func(string, ...any) {}
	}
	b := &AwaitedBucket{ctx: ctx, js: js, name: name, without: without, logf: logf, pending: map[string][]byte{}}
	// Heard before the first try, so a creation between the try and the subscription is not missed.
	if nc != nil {
		b.nc = nc
		b.hear()
		b.followReconnects()
	}
	ctxB, cancel := context.WithTimeout(ctx, bindWithin)
	kv, err := js.KeyValue(ctxB, name)
	cancel()
	if err == nil {
		b.bound(kv, false)
		return b
	}
	logf("the bucket %s cannot be read yet (%v): %s until it appears; it is bound when the bus says it was created, or at "+
		"its next use", name, err, without)
	return b
}

// KV is the bound bucket, or nil while it is not on the bus; while nil, the call tries to bind it once.
func (b *AwaitedBucket) KV() jetstream.KeyValue {
	b.mu.Lock()
	kv := b.kv
	b.mu.Unlock()
	if kv != nil {
		return kv
	}
	return b.bind()
}

// Ready runs f with the bucket: now when it is bound, otherwise once it appears (on the goroutine that bound it).
func (b *AwaitedBucket) Ready(f func(jetstream.KeyValue)) {
	b.mu.Lock()
	kv := b.kv
	if kv == nil {
		b.ready = append(b.ready, f)
	}
	b.mu.Unlock()
	if kv != nil {
		f(kv)
	}
}

// Put writes key, as jetstream.KeyValue's Put. While the bucket is missing it keeps value as key's latest and answers
// ErrNotYet; the kept values are written, in the order their keys were first kept, when it appears.
func (b *AwaitedBucket) Put(ctx context.Context, key string, value []byte) (uint64, error) {
	kv := b.KV()
	if kv == nil {
		b.mu.Lock()
		if b.kv == nil {
			if _, ok := b.pending[key]; !ok {
				b.order = append(b.order, key)
			}
			b.pending[key] = append([]byte(nil), value...)
			b.mu.Unlock()
			return 0, fmt.Errorf("%s: %w", b.name, ErrNotYet)
		}
		kv = b.kv
		b.mu.Unlock()
	}
	return kvlimit.Put(ctx, kv, key, value)
}

// bind tries one STREAM.INFO of the bucket and, when it is there, binds it; it answers the bucket or nil.
func (b *AwaitedBucket) bind() jetstream.KeyValue {
	if b.ctx.Err() != nil {
		return nil
	}
	ctx, cancel := context.WithTimeout(b.ctx, bindWithin)
	defer cancel()
	kv, err := b.js.KeyValue(ctx, b.name)
	if err != nil {
		return nil
	}
	return b.bound(kv, true)
}

// bound keeps kv as the bucket (unless another bind did first), writes what was kept and runs what waited on it.
func (b *AwaitedBucket) bound(kv jetstream.KeyValue, appeared bool) jetstream.KeyValue {
	b.mu.Lock()
	if b.kv != nil {
		kv = b.kv
		b.mu.Unlock()
		return kv
	}
	// The kept values are written before the bucket is handed to anyone, so no later write is overtaken by an earlier.
	written, failed := 0, 0
	for _, key := range b.order {
		ctx, cancel := context.WithTimeout(b.ctx, bindWithin)
		_, err := kvlimit.Put(ctx, kv, key, b.pending[key])
		cancel()
		if err != nil {
			failed++
			b.logf("the bucket %s appeared but the value kept for %s could not be written: %v", b.name, key, err)
			continue
		}
		written++
	}
	b.kv, b.pending, b.order = kv, nil, nil
	ready, sub, stop := b.ready, b.sub, b.stopStatus
	b.ready, b.sub, b.stopStatus = nil, nil, nil
	b.mu.Unlock()
	if sub != nil {
		_ = sub.Unsubscribe()
	}
	if stop != nil {
		stop()
	}
	if appeared {
		b.logf("the bucket %s appeared and is bound: %d kept value(s) written, %d not", b.name, written, failed)
	}
	for _, f := range ready {
		f(kv)
	}
	return kv
}

// refusalReader is the part of a *nats.Conn that says whether the bus refused a subscription: the server's refusal comes
// back after a round trip as the connection's last error, naming the subject.
type refusalReader interface {
	FlushTimeout(time.Duration) error
	LastError() error
}

// statusFollower is the part of a *nats.Conn that says when it connected again, without taking its reconnect handler.
type statusFollower interface {
	StatusChanged(statuses ...nats.Status) chan nats.Status
	RemoveStatusListener(ch chan nats.Status)
}

// hear subscribes to the bucket's creation (again), and says once, loudly, when the bus refuses it: then only a use or
// a reconnect binds the bucket.
func (b *AwaitedBucket) hear() {
	subject := createdAdvisory(b.name)
	b.mu.Lock()
	old, bound := b.sub, b.kv != nil
	b.sub = nil
	b.mu.Unlock()
	if old != nil {
		_ = old.Unsubscribe()
	}
	if bound {
		return
	}
	checks, _ := b.nc.(refusalReader)
	var before error
	if checks != nil {
		before = checks.LastError()
	}
	sub, err := b.nc.Subscribe(subject, func(*nats.Msg) { go b.bind() })
	if err == nil && checks != nil && checks.FlushTimeout(bindWithin) == nil {
		if after := checks.LastError(); after != nil && after != before && errors.Is(after, nats.ErrPermissionViolation) &&
			strings.Contains(after.Error(), subject) {
			err = after
		}
	}
	if err != nil {
		if sub != nil {
			_ = sub.Unsubscribe()
		}
		b.mu.Lock()
		said := b.refusedSaid
		b.refusedSaid = true
		b.mu.Unlock()
		if !said {
			b.logf("cannot hear the creation of %s: %v; it binds at its next use, or when the connection reconnects", b.name, err)
		}
		return
	}
	b.mu.Lock()
	if b.kv != nil {
		b.mu.Unlock()
		if sub != nil {
			_ = sub.Unsubscribe()
		}
		return
	}
	b.sub, b.refusedSaid = sub, false
	b.mu.Unlock()
}

// followReconnects hears the bucket's creation again and tries a bind each time the connection connects again (a grant
// reaches the bus with a new user list; a lost subscription is made again), until the bucket is bound or the part ends.
func (b *AwaitedBucket) followReconnects() {
	w, ok := b.nc.(statusFollower)
	if !ok {
		return
	}
	ch := w.StatusChanged(nats.CONNECTED)
	b.mu.Lock()
	if b.kv != nil {
		b.mu.Unlock()
		w.RemoveStatusListener(ch)
		return
	}
	b.stopStatus = func() { w.RemoveStatusListener(ch) }
	b.mu.Unlock()
	go func() {
		for {
			select {
			case <-b.ctx.Done():
				w.RemoveStatusListener(ch)
				return
			case _, open := <-ch:
				if !open {
					return
				}
				b.hear()
				b.KV()
			}
		}
	}()
}