mesh-delivery/modules/mesh-grants/cmd/mesh-grants/create.go

package main

import (
	"context"
	"errors"
	"fmt"
	"slices"

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

	"mesh-grants/internal/grants"
)

// creating creates a stream a manifest declares that the bus lacks (issue 1790; the operator, 2026-10-26 08:54): the
// create alone, which the account is granted by the stream's name, never an update or a delete. The one place the part
// creates on the bus.
type creating struct {
	ctx context.Context
	js  jetstream.JetStream
}

// CreateConsumer creates one durable consumer an account needs, by its stream and name (ADR 0561 §6): the create alone;
// one the bus holds answers ErrHeld and is left as it is.
func (c creating) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error {
	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
	defer cancel()
	if _, err := c.js.Consumer(ctx, stream, cfg.Durable); err == nil {
		return grants.ErrHeld
	}
	_, err := c.js.CreateConsumer(ctx, stream, cfg)
	switch {
	case err == nil:
		return nil
	case errors.Is(err, jetstream.ErrConsumerExists):
		return grants.ErrHeld
	case errors.Is(err, context.DeadlineExceeded):
		return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err)
	}
	return err
}

// MatchConsumer brings a held consumer to cfg on a person's ask (grants_consumer_match), keeping where it delivers from:
// the server refuses a changed deliver policy, and a consumer moved would replay or lose what it holds.
func (c creating) MatchConsumer(stream string, cfg jetstream.ConsumerConfig) error {
	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
	defer cancel()
	held, err := c.js.Consumer(ctx, stream, cfg.Durable)
	if err != nil {
		return err
	}
	have := held.CachedInfo().Config
	cfg.DeliverPolicy, cfg.OptStartSeq, cfg.OptStartTime = have.DeliverPolicy, have.OptStartSeq, have.OptStartTime
	_, err = c.js.UpdateConsumer(ctx, stream, cfg)
	return err
}

func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error {
	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
	defer cancel()
	_, err := c.js.CreateStream(ctx, cfg)
	switch {
	case err == nil:
		return nil
	case errors.Is(err, jetstream.ErrStreamNameAlreadyInUse):
		return grants.ErrHeld
	case errors.Is(err, context.DeadlineExceeded):
		return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err)
	}
	return err
}

// AddSubjects adds subjects to a held stream and changes nothing else: it reads the stream's configuration, sets its
// subjects to the union and updates with the rest as it was.
func (c creating) AddSubjects(stream string, add []string) error {
	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
	defer cancel()
	s, err := c.js.Stream(ctx, stream)
	if err != nil {
		return err
	}
	cfg := s.CachedInfo().Config
	for _, a := range add {
		if !slices.Contains(cfg.Subjects, a) {
			cfg.Subjects = append(cfg.Subjects, a)
		}
	}
	_, err = c.js.UpdateStream(ctx, cfg)
	return err
}

// RaiseMaxBytes raises a held stream's size cap to `to` when that is larger than the cap it holds, and changes nothing
// else: it reads the stream's configuration and updates with the rest as it was. An unbounded stream (-1) and a cap that
// is larger or equal are left as they are.
func (c creating) RaiseMaxBytes(stream string, to int64) (int64, bool, error) {
	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
	defer cancel()
	s, err := c.js.Stream(ctx, stream)
	if err != nil {
		return 0, false, err
	}
	cfg := s.CachedInfo().Config
	if cfg.MaxBytes <= 0 || cfg.MaxBytes >= to {
		return cfg.MaxBytes, false, nil
	}
	from := cfg.MaxBytes
	cfg.MaxBytes = to
	if _, err := c.js.UpdateStream(ctx, cfg); err != nil {
		return from, false, err
	}
	return from, true, nil
}

// AddFilters adds filter subjects to a held consumer and changes nothing else.
func (c creating) AddFilters(stream, name string, add []string) error {
	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
	defer cancel()
	held, err := c.js.Consumer(ctx, stream, name)
	if err != nil {
		return err
	}
	cfg := held.CachedInfo().Config
	filters := slices.Clone(cfg.FilterSubjects)
	if cfg.FilterSubject != "" {
		filters = append(filters, cfg.FilterSubject)
	}
	for _, a := range add {
		if !slices.Contains(filters, a) {
			filters = append(filters, a)
		}
	}
	cfg.FilterSubject, cfg.FilterSubjects = "", filters
	_, err = c.js.UpdateConsumer(ctx, stream, cfg)
	return err
}

// PurgeSubject purges the messages of one subject of a stream (design 47 G §544): the named asks of a build seat's queue.
func (c creating) PurgeSubject(stream, subject string) error {
	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
	defer cancel()
	st, err := c.js.Stream(ctx, stream)
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	return st.Purge(ctx, jetstream.WithPurgeSubject(subject))
}

// DeleteConsumer deletes one consumer of a stream (design 47 G §544): a build seat's named worker; one the bus does not hold
// is no error.
func (c creating) DeleteConsumer(stream, name string) error {
	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
	defer cancel()
	err := c.js.DeleteConsumer(ctx, stream, name)
	if errors.Is(err, jetstream.ErrConsumerNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil
	}
	return err
}