// Copyright 2019-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package server

import (
	"archive/tar"
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"math/big"
	"math/rand"
	"os"
	"path/filepath"
	"reflect"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/klauspost/compress/s2"
	"github.com/nats-io/nats-server/v2/server/gsl"
	"github.com/nats-io/nuid"
)

// StreamConfigRequest is used to create or update a stream.
type StreamConfigRequest struct {
	StreamConfig
	// This is not part of the StreamConfig, because its scoped to request,
	// and not to the stream itself.
	Pedantic bool `json:"pedantic,omitempty"`
}

// StreamConfig will determine the name, subjects and retention policy
// for a given stream. If subjects is empty the name will be used.
type StreamConfig struct {
	Name         string           `json:"name"`
	Description  string           `json:"description,omitempty"`
	Subjects     []string         `json:"subjects,omitempty"`
	Retention    RetentionPolicy  `json:"retention"`
	MaxConsumers int              `json:"max_consumers"`
	MaxMsgs      int64            `json:"max_msgs"`
	MaxBytes     int64            `json:"max_bytes"`
	MaxAge       time.Duration    `json:"max_age"`
	MaxMsgsPer   int64            `json:"max_msgs_per_subject"`
	MaxMsgSize   int32            `json:"max_msg_size,omitempty"`
	Discard      DiscardPolicy    `json:"discard"`
	Storage      StorageType      `json:"storage"`
	Replicas     int              `json:"num_replicas"`
	NoAck        bool             `json:"no_ack,omitempty"`
	Template     string           `json:"template_owner,omitempty"` // Deprecated: stream templates are deprecated and will be removed in a future version.
	Duplicates   time.Duration    `json:"duplicate_window,omitempty"`
	Placement    *Placement       `json:"placement,omitempty"`
	Mirror       *StreamSource    `json:"mirror,omitempty"`
	Sources      []*StreamSource  `json:"sources,omitempty"`
	Compression  StoreCompression `json:"compression"`
	FirstSeq     uint64           `json:"first_seq,omitempty"`

	// Allow applying a subject transform to incoming messages before doing anything else
	SubjectTransform *SubjectTransformConfig `json:"subject_transform,omitempty"`

	// Allow republish of the message after being sequenced and stored.
	RePublish *RePublish `json:"republish,omitempty"`

	// Allow higher performance, direct access to get individual messages. E.g. KeyValue
	AllowDirect bool `json:"allow_direct"`
	// Allow higher performance and unified direct access for mirrors as well.
	MirrorDirect bool `json:"mirror_direct"`

	// Allow KV like semantics to also discard new on a per subject basis
	DiscardNewPer bool `json:"discard_new_per_subject,omitempty"`

	// Optional qualifiers. These can not be modified after set to true.

	// Sealed will seal a stream so no messages can get out or in.
	Sealed bool `json:"sealed"`
	// DenyDelete will restrict the ability to delete messages.
	DenyDelete bool `json:"deny_delete"`
	// DenyPurge will restrict the ability to purge messages.
	DenyPurge bool `json:"deny_purge"`
	// AllowRollup allows messages to be placed into the system and purge
	// all older messages using a special msg header.
	AllowRollup bool `json:"allow_rollup_hdrs"`

	// The following defaults will apply to consumers when created against
	// this stream, unless overridden manually.
	// TODO(nat): Can/should we name these better?
	ConsumerLimits StreamConsumerLimits `json:"consumer_limits"`

	// AllowMsgTTL allows header initiated per-message TTLs. If disabled,
	// then the `NATS-TTL` header will be ignored.
	AllowMsgTTL bool `json:"allow_msg_ttl"`

	// SubjectDeleteMarkerTTL sets the TTL of delete marker messages left behind by
	// subject delete markers.
	SubjectDeleteMarkerTTL time.Duration `json:"subject_delete_marker_ttl,omitempty"`

	// AllowMsgCounter allows a stream to use (only) counter CRDTs.
	AllowMsgCounter bool `json:"allow_msg_counter,omitempty"`

	// AllowAtomicPublish allows atomic batch publishing into the stream.
	AllowAtomicPublish bool `json:"allow_atomic,omitempty"`

	// AllowMsgSchedules allows the scheduling of messages.
	AllowMsgSchedules bool `json:"allow_msg_schedules,omitempty"`

	// PersistMode allows to opt-in to different persistence mode settings.
	PersistMode PersistModeType `json:"persist_mode,omitempty"`

	// Metadata is additional metadata for the Stream.
	Metadata map[string]string `json:"metadata,omitempty"`
}

// clone performs a deep copy of the StreamConfig struct, returning a new clone with
// all values copied.
func (cfg *StreamConfig) clone() *StreamConfig {
	clone := *cfg
	if cfg.Placement != nil {
		placement := *cfg.Placement
		clone.Placement = &placement
	}
	if cfg.Mirror != nil {
		mirror := *cfg.Mirror
		clone.Mirror = &mirror
	}
	if len(cfg.Sources) > 0 {
		clone.Sources = make([]*StreamSource, len(cfg.Sources))
		for i, cfgSource := range cfg.Sources {
			source := *cfgSource
			clone.Sources[i] = &source
		}
	}
	if cfg.SubjectTransform != nil {
		transform := *cfg.SubjectTransform
		clone.SubjectTransform = &transform
	}
	if cfg.RePublish != nil {
		rePublish := *cfg.RePublish
		clone.RePublish = &rePublish
	}
	if cfg.Metadata != nil {
		clone.Metadata = make(map[string]string, len(cfg.Metadata))
		for k, v := range cfg.Metadata {
			clone.Metadata[k] = v
		}
	}
	return &clone
}

type StreamConsumerLimits struct {
	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
	MaxAckPending     int           `json:"max_ack_pending,omitempty"`
}

// SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received
type SubjectTransformConfig struct {
	Source      string `json:"src"`
	Destination string `json:"dest"`
}

// RePublish is for republishing messages once committed to a stream.
type RePublish struct {
	Source      string `json:"src,omitempty"`
	Destination string `json:"dest"`
	HeadersOnly bool   `json:"headers_only,omitempty"`
}

// PersistModeType determines what persistence mode the stream uses.
type PersistModeType int

const (
	// DefaultPersistMode specifies the default persist mode. Writes to the stream will immediately be flushed.
	// The publish acknowledgement will be sent after the persisting completes.
	DefaultPersistMode = PersistModeType(iota)
	// AsyncPersistMode specifies writes to the stream will be flushed asynchronously.
	// The publish acknowledgement may be sent before the persisting completes.
	// This means writes could be lost if they weren't flushed prior to a hard kill of the server.
	AsyncPersistMode
)

const (
	defaultPersistModeJSONString = `"default"`
	asyncPersistModeJSONString   = `"async"`
)

var (
	defaultPersistModeJSONBytes = []byte(defaultPersistModeJSONString)
	asyncPersistModeJSONBytes   = []byte(asyncPersistModeJSONString)
)

func (wc PersistModeType) String() string {
	switch wc {
	case DefaultPersistMode:
		return "Default"
	case AsyncPersistMode:
		return "Async"
	default:
		return "Unknown Persist Mode Type"
	}
}

func (wc PersistModeType) MarshalJSON() ([]byte, error) {
	switch wc {
	case DefaultPersistMode:
		return defaultPersistModeJSONBytes, nil
	case AsyncPersistMode:
		return asyncPersistModeJSONBytes, nil
	default:
		return nil, fmt.Errorf("can not marshal %v", wc)
	}
}

func (wc *PersistModeType) UnmarshalJSON(data []byte) error {
	switch string(data) {
	case defaultPersistModeJSONString, `""`:
		*wc = DefaultPersistMode
	case asyncPersistModeJSONString:
		*wc = AsyncPersistMode
	default:
		return fmt.Errorf("can not unmarshal %q", data)
	}
	return nil
}

// JSPubAckResponse is a formal response to a publish operation.
type JSPubAckResponse struct {
	Error *ApiError `json:"error,omitempty"`
	*PubAck
}

// ToError checks if the response has a error and if it does converts it to an error
// avoiding the pitfalls described by https://yourbasic.org/golang/gotcha-why-nil-error-not-equal-nil/
func (r *JSPubAckResponse) ToError() error {
	if r.Error == nil {
		return nil
	}
	return r.Error
}

// PubAck is the detail you get back from a publish to a stream that was successful.
// e.g. +OK {"stream": "Orders", "seq": 22}
type PubAck struct {
	Stream    string `json:"stream"`
	Sequence  uint64 `json:"seq"`
	Domain    string `json:"domain,omitempty"`
	Duplicate bool   `json:"duplicate,omitempty"`
	Value     string `json:"val,omitempty"`
	BatchId   string `json:"batch,omitempty"`
	BatchSize int    `json:"count,omitempty"`
}

// CounterValue is the body of a message when used as a counter.
// e.g. {"val":"123"}
type CounterValue struct {
	Value string `json:"val"`
}

// CounterSources is the body of the Nats-Counter-Sources header.
// e.g. {"stream":{"subject":"123"}}
type CounterSources map[string]map[string]string

// StreamInfo shows config and current state for this stream.
type StreamInfo struct {
	Config     StreamConfig        `json:"config"`
	Created    time.Time           `json:"created"`
	State      StreamState         `json:"state"`
	Domain     string              `json:"domain,omitempty"`
	Cluster    *ClusterInfo        `json:"cluster,omitempty"`
	Mirror     *StreamSourceInfo   `json:"mirror,omitempty"`
	Sources    []*StreamSourceInfo `json:"sources,omitempty"`
	Alternates []StreamAlternate   `json:"alternates,omitempty"`
	// TimeStamp indicates when the info was gathered
	TimeStamp time.Time `json:"ts"`
}

// streamInfoClusterResponse is a response used in a cluster to communicate the stream info
// back to the meta leader as part of a stream list request.
type streamInfoClusterResponse struct {
	StreamInfo
	OfflineReason string `json:"offline_reason,omitempty"` // Reporting when a stream is offline.
}

type StreamAlternate struct {
	Name    string `json:"name"`
	Domain  string `json:"domain,omitempty"`
	Cluster string `json:"cluster"`
}

// ClusterInfo shows information about the underlying set of servers
// that make up the stream or consumer.
type ClusterInfo struct {
	Name        string      `json:"name,omitempty"`
	RaftGroup   string      `json:"raft_group,omitempty"`
	Leader      string      `json:"leader,omitempty"`
	LeaderSince *time.Time  `json:"leader_since,omitempty"`
	SystemAcc   bool        `json:"system_account,omitempty"`
	TrafficAcc  string      `json:"traffic_account,omitempty"`
	Replicas    []*PeerInfo `json:"replicas,omitempty"`
}

// PeerInfo shows information about all the peers in the cluster that
// are supporting the stream or consumer.
type PeerInfo struct {
	Name    string        `json:"name"`              // Name is the unique name for the peer
	Current bool          `json:"current"`           // Current indicates if it was seen recently and fully caught up
	Offline bool          `json:"offline,omitempty"` // Offline indicates if it has not been seen recently
	Active  time.Duration `json:"active"`            // Active is the timestamp it was last active
	Lag     uint64        `json:"lag,omitempty"`     // Lag is how many operations behind it is
	Peer    string        `json:"peer"`              // Peer is the unique ID for the peer
	// For migrations.
	cluster string
}

// StreamSourceInfo shows information about an upstream stream source.
type StreamSourceInfo struct {
	Name              string                   `json:"name"`
	External          *ExternalStream          `json:"external,omitempty"`
	Lag               uint64                   `json:"lag"`
	Active            time.Duration            `json:"active"`
	Error             *ApiError                `json:"error,omitempty"`
	FilterSubject     string                   `json:"filter_subject,omitempty"`
	SubjectTransforms []SubjectTransformConfig `json:"subject_transforms,omitempty"`
}

// StreamSource dictates how streams can source from other streams.
type StreamSource struct {
	Name              string                   `json:"name"`
	OptStartSeq       uint64                   `json:"opt_start_seq,omitempty"`
	OptStartTime      *time.Time               `json:"opt_start_time,omitempty"`
	FilterSubject     string                   `json:"filter_subject,omitempty"`
	SubjectTransforms []SubjectTransformConfig `json:"subject_transforms,omitempty"`
	External          *ExternalStream          `json:"external,omitempty"`

	// Internal
	iname string // For indexing when stream names are the same for multiple sources.
}

// ExternalStream allows you to qualify access to a stream source in another account or domain.
type ExternalStream struct {
	ApiPrefix     string `json:"api"`
	DeliverPrefix string `json:"deliver"`
}

// Will return the domain for this external stream.
func (ext *ExternalStream) Domain() string {
	if ext == nil || ext.ApiPrefix == _EMPTY_ {
		return _EMPTY_
	}
	return tokenAt(ext.ApiPrefix, 2)
}

// For managing stream ingest.
const (
	streamDefaultMaxQueueMsgs  = 100_000
	streamDefaultMaxQueueBytes = 128 * 1024 * 1024
)

// For managing stream batches.
const (
	streamDefaultMaxBatchInflightPerStream = 50
	streamDefaultMaxBatchInflightTotal     = 1000
	streamDefaultMaxBatchSize              = 1000
	streamDefaultMaxBatchTimeout           = 10 * time.Second
)

var (
	streamMaxBatchInflightPerStream = streamDefaultMaxBatchInflightPerStream
	streamMaxBatchInflightTotal     = streamDefaultMaxBatchInflightTotal
	streamMaxBatchSize              = streamDefaultMaxBatchSize
	streamMaxBatchTimeout           = streamDefaultMaxBatchTimeout
)

// Stream is a jetstream stream of messages. When we receive a message internally destined
// for a Stream we will direct link from the client to this structure.
type stream struct {
	mu     sync.RWMutex // Read/write lock for the stream.
	js     *jetStream   // The internal *jetStream for the account.
	jsa    *jsAccount   // The JetStream account-level information.
	acc    *Account     // The account this stream is defined in.
	srv    *Server      // The server we are running in.
	client *client      // The internal JetStream client.
	sysc   *client      // The internal JetStream system client.

	// The current last subscription ID for the subscriptions through `client`.
	// Those subscriptions are for the subjects filters being listened to and captured by the stream.
	sid atomic.Uint64

	pubAck    []byte                  // The template (prefix) to generate the pubAck responses for this stream quickly.
	outq      *jsOutQ                 // Queue of *jsPubMsg for sending messages.
	msgs      *ipQueue[*inMsg]        // Intra-process queue for the ingress of messages.
	gets      *ipQueue[*directGetReq] // Intra-process queue for the direct get requests.
	store     StreamStore             // The storage for this stream.
	ackq      *ipQueue[uint64]        // Intra-process queue for acks.
	lseq      uint64                  // The sequence number of the last message stored in the stream.
	lmsgId    string                  // The de-duplication message ID of the last message stored in the stream.
	consumers map[string]*consumer    // The consumers for this stream.
	numFilter int                     // The number of filtered consumers.
	cfg       StreamConfig            // The stream's config.
	cfgMu     sync.RWMutex            // Config mutex used to solve some races with consumer code
	created   time.Time               // Time the stream was created.
	stype     StorageType             // The storage type.
	tier      string                  // The tier is the number of replicas for the stream (e.g. "R1" or "R3").
	ddMu      sync.Mutex              // Lock for dedupe state.
	ddmap     map[string]*ddentry     // The dedupe map.
	ddarr     []*ddentry              // The dedupe array.
	ddindex   int                     // The dedupe index.
	ddtmr     *time.Timer             // The dedupe timer.
	qch       chan struct{}           // The quit channel.
	mqch      chan struct{}           // The monitor's quit channel.
	active    bool                    // Indicates that there are active internal subscriptions (for the subject filters)
	// and/or mirror/sources consumers are scheduled to be established or already started.
	closed atomic.Bool // Set to true when stop() is called on the stream.

	// Mirror
	mirror *sourceInfo

	// Sources
	sources              map[string]*sourceInfo
	sourceSetupSchedules map[string]*time.Timer
	sourcesConsumerSetup *time.Timer
	smsgs                *ipQueue[*inMsg] // Intra-process queue for all incoming sourced messages.

	// Indicates we have direct consumers.
	directs int

	// For input subject transform.
	itr *subjectTransform

	// For republishing.
	tr *subjectTransform

	// For processing consumers without main stream lock.
	clsMu sync.RWMutex
	cList []*consumer                    // Consumer list.
	sch   chan struct{}                  // Channel to signal consumers.
	sigq  *ipQueue[*cMsg]                // Intra-process queue for the messages to signal to the consumers.
	csl   *gsl.GenericSublist[*consumer] // Consumer subscription list.

	// Leader will store seq/msgTrace in clustering mode. Used in applyStreamEntries
	// to know if trace event should be sent after processing.
	mt map[uint64]*msgTrace

	// For non limits policy streams when they process an ack before the actual msg.
	// Can happen in stretch clusters, multi-cloud, or during catchup for a restarted server.
	preAcks map[uint64]map[*consumer]struct{}

	// TODO(dlc) - Hide everything below behind two pointers.
	// Clustered mode.
	sa        *streamAssignment // What the meta controller uses to assign streams to peers.
	node      RaftNode          // Our RAFT node for the stream's group.
	catchup   atomic.Bool       // Used to signal we are in catchup mode.
	catchups  map[string]uint64 // The number of messages that need to be caught per peer.
	syncSub   *subscription     // Internal subscription for sync messages (on "$JSC.SYNC").
	infoSub   *subscription     // Internal subscription for stream info requests.
	clMu      sync.Mutex        // The mutex for clseq and clfs.
	clseq     uint64            // The current last seq being proposed to the NRG layer.
	clfs      uint64            // The count (offset) of the number of failed NRG sequences used to compute clseq.
	lqsent    time.Time         // The time at which the last lost quorum advisory was sent. Used to rate limit.
	uch       chan struct{}     // The channel to signal updates to the monitor routine.
	inMonitor bool              // True if the monitor routine has been started.

	inflight                    map[string]*inflightSubjectRunningTotal // Inflight message sizes per subject.
	clusteredCounterTotal       map[string]*msgCounterRunningTotal      // Inflight counter totals.
	expectedPerSubjectSequence  map[uint64]string                       // Inflight 'expected per subject' subjects per clseq.
	expectedPerSubjectInProcess map[string]struct{}                     // Current 'expected per subject' subjects in process.

	// Direct get subscription.
	directLeaderSub *subscription
	directSub       *subscription
	lastBySub       *subscription
	mirrorDirectSub *subscription // Mirrors only.
	mirrorLastBySub *subscription // Mirrors only.

	monitorWg sync.WaitGroup // Wait group for the monitor routine.

	// If standalone/single-server, the offline reason needs to be stored directly in the stream.
	// Otherwise, if clustered it will be part of the stream assignment.
	offlineReason string

	batches    *batching   // Inflight batches prior to committing them.
	batchApply *batchApply // State to check for batch completeness before applying it.
}

// inflightSubjectRunningTotal stores a running total of inflight messages for a specific subject.
type inflightSubjectRunningTotal struct {
	bytes uint64 // Running total of inflight bytes for inflight messages.
	ops   uint64 // Inflight operations, i.e. inflight messages for this subject. If this reaches zero, we can remove the running total.
}

// msgCounterRunningTotal stores a running total and a number of inflight
// but not yet applied clustered proposals/operations for this counter.
type msgCounterRunningTotal struct {
	total   *big.Int       // Running total.
	sources CounterSources // Last seen counter sources.
	ops     uint64         // Inflight operations. If this reaches zero, we can remove the running total.
}

type sourceInfo struct {
	name  string        // The name of the stream being sourced.
	iname string        // The unique index name of this particular source.
	cname string        // The name of the current consumer for this source.
	sub   *subscription // The subscription to the consumer.

	msgs  *ipQueue[*inMsg]    // Intra-process queue for incoming messages.
	sseq  uint64              // Last stream message sequence number seen from the source.
	dseq  uint64              // Last delivery (i.e. consumer's) sequence number.
	lag   uint64              // 0 or number of messages pending (as last reported by the consumer) - 1.
	err   *ApiError           // The API error that caused the last consumer setup to fail.
	fails int                 // The number of times trying to setup the consumer failed.
	last  atomic.Int64        // Time the consumer was created or of last message it received.
	lreq  time.Time           // The last time setupMirrorConsumer/setupSourceConsumer was called.
	qch   chan struct{}       // Quit channel.
	sip   bool                // Setup in progress.
	wg    sync.WaitGroup      // WaitGroup for the consumer's go routine.
	sf    string              // The subject filter.
	sfs   []string            // The subject filters.
	trs   []*subjectTransform // The subject transforms.
}

// For mirrors and direct get
const (
	dgetGroup          = sysGroup
	dgetCaughtUpThresh = 10
)

// Headers for published messages.
const (
	JSMsgId                   = "Nats-Msg-Id"
	JSExpectedStream          = "Nats-Expected-Stream"
	JSExpectedLastSeq         = "Nats-Expected-Last-Sequence"
	JSExpectedLastSubjSeq     = "Nats-Expected-Last-Subject-Sequence"
	JSExpectedLastSubjSeqSubj = "Nats-Expected-Last-Subject-Sequence-Subject"
	JSExpectedLastMsgId       = "Nats-Expected-Last-Msg-Id"
	JSStreamSource            = "Nats-Stream-Source"
	JSLastConsumerSeq         = "Nats-Last-Consumer"
	JSLastStreamSeq           = "Nats-Last-Stream"
	JSConsumerStalled         = "Nats-Consumer-Stalled"
	JSMsgRollup               = "Nats-Rollup"
	JSMsgSize                 = "Nats-Msg-Size"
	JSResponseType            = "Nats-Response-Type"
	JSMessageTTL              = "Nats-TTL"
	JSMarkerReason            = "Nats-Marker-Reason"
	JSMessageIncr             = "Nats-Incr"
	JSMessageCounterSources   = "Nats-Counter-Sources"
	JSBatchId                 = "Nats-Batch-Id"
	JSBatchSeq                = "Nats-Batch-Sequence"
	JSBatchCommit             = "Nats-Batch-Commit"
	JSSchedulePattern         = "Nats-Schedule"
	JSScheduleTTL             = "Nats-Schedule-TTL"
	JSScheduleTarget          = "Nats-Schedule-Target"
)

// Headers for published KV messages.
var (
	KVOperation           = "KV-Operation"
	KVOperationValuePurge = []byte("PURGE")
)

// Headers for scheduled messages.
const (
	JSScheduler         = "Nats-Scheduler"
	JSScheduleNext      = "Nats-Schedule-Next"
	JSScheduleNextPurge = "purge" // If it's a non-repeating/delayed message, the schedule is purged.
)

// Headers for republished messages and direct get responses.
const (
	JSStream       = "Nats-Stream"
	JSSequence     = "Nats-Sequence"
	JSTimeStamp    = "Nats-Time-Stamp"
	JSSubject      = "Nats-Subject"
	JSLastSequence = "Nats-Last-Sequence" // The sequence number of the precedent message either republished or in a batch of responses. 0 in the first message.
	JSNumPending   = "Nats-Num-Pending"   // Number of messages pending in the multi/batched get response
	JSUpToSequence = "Nats-UpTo-Sequence"
)

// Rollups, can be subject only or all messages.
const (
	JSMsgRollupSubject = "sub"
	JSMsgRollupAll     = "all"
)

// Applied limits in the Nats-Applied-Limit header.
const (
	JSMarkerReasonMaxAge = "MaxAge"
	JSMarkerReasonPurge  = "Purge"
	JSMarkerReasonRemove = "Remove"
)

const (
	jsCreateResponse = "create"
)

// Dedupe entry
type ddentry struct {
	id  string // The unique message ID provided by the client.
	seq uint64 // The sequence number of the message.
	ts  int64  // The timestamp of the message.
}

// Replicas Range
const StreamMaxReplicas = 5

// AddStream adds a stream for the given account.
func (a *Account) addStream(config *StreamConfig) (*stream, error) {
	return a.addStreamWithAssignment(config, nil, nil, false)
}

// AddStreamWithStore adds a stream for the given account with custome store config options.
func (a *Account) addStreamWithStore(config *StreamConfig, fsConfig *FileStoreConfig) (*stream, error) {
	return a.addStreamWithAssignment(config, fsConfig, nil, false)
}

func (a *Account) addStreamPedantic(config *StreamConfig, pedantic bool) (*stream, error) {
	return a.addStreamWithAssignment(config, nil, nil, pedantic)
}

func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileStoreConfig, sa *streamAssignment, pedantic bool) (*stream, error) {
	s, jsa, err := a.checkForJetStream()
	if err != nil {
		return nil, err
	}

	// If we do not have the stream currently assigned to us in cluster mode we will proceed but warn.
	// This can happen on startup with restored state where on meta replay we still do not have
	// the assignment. Running in single server mode this always returns true.
	if !jsa.streamAssigned(config.Name) {
		s.Debugf("Stream '%s > %s' does not seem to be assigned to this server", a.Name, config.Name)
	}

	// Sensible defaults.
	ccfg, apiErr := s.checkStreamCfg(config, a, pedantic)
	if apiErr != nil {
		return nil, apiErr
	}
	cfg := &ccfg

	singleServerMode := !s.JetStreamIsClustered() && s.standAloneMode()
	if singleServerMode && cfg.Replicas > 1 {
		return nil, ApiErrors[JSStreamReplicasNotSupportedErr]
	}

	// Make sure we are ok when these are done in parallel.
	// We used to call Add(1) in the "else" clause of the "if loaded"
	// statement. This caused a data race because it was possible
	// that one go routine stores (with count==0) and another routine
	// gets "loaded==true" and calls wg.Wait() while the other routine
	// then calls wg.Add(1). It also could mean that two routines execute
	// the rest of the code concurrently.
	swg := &sync.WaitGroup{}
	swg.Add(1)
	v, loaded := jsa.inflight.LoadOrStore(cfg.Name, swg)
	wg := v.(*sync.WaitGroup)
	if loaded {
		wg.Wait()
		// This waitgroup is "thrown away" (since there was an existing one).
		swg.Done()
	} else {
		defer func() {
			jsa.inflight.Delete(cfg.Name)
			wg.Done()
		}()
	}

	js, isClustered := jsa.jetStreamAndClustered()
	jsa.mu.Lock()
	if mset, ok := jsa.streams[cfg.Name]; ok {
		jsa.mu.Unlock()
		// Check to see if configs are same.
		ocfg := mset.config()

		// set the index name on cfg since it would not contain a value for iname while the return from mset.config() does to ensure the DeepEqual works
		for _, s := range cfg.Sources {
			s.setIndexName()
		}

		// Hold lock, because we'll be reading from and writing to a shared object.
		js.mu.Lock()
		copyStreamMetadata(cfg, &ocfg)
		deepEqual := reflect.DeepEqual(cfg, &ocfg)
		js.mu.Unlock()
		if deepEqual {
			if sa != nil {
				mset.setStreamAssignment(sa)
			}
			return mset, nil
		} else {
			return nil, ApiErrors[JSStreamNameExistErr]
		}
	}
	jsa.usageMu.RLock()
	selected, tier, hasTier := jsa.selectLimits(cfg.Replicas)
	jsa.usageMu.RUnlock()
	reserved := int64(0)
	if !isClustered {
		reserved = jsa.tieredReservation(tier, cfg)
	}
	jsa.mu.Unlock()

	if !hasTier {
		return nil, NewJSNoLimitsError()
	}
	js.mu.RLock()
	if isClustered {
		_, reserved = tieredStreamAndReservationCount(js.cluster.streams[a.Name], tier, cfg)
	}
	if err := js.checkAllLimits(&selected, cfg, reserved, 0); err != nil {
		js.mu.RUnlock()
		return nil, err
	}
	js.mu.RUnlock()
	jsa.mu.Lock()
	// Check for template ownership if present.
	if cfg.Template != _EMPTY_ && jsa.account != nil {
		if !jsa.checkTemplateOwnership(cfg.Template, cfg.Name) {
			jsa.mu.Unlock()
			return nil, fmt.Errorf("stream not owned by template")
		}
	}

	// If mirror, check if the transforms (if any) are valid.
	if cfg.Mirror != nil {
		if len(cfg.Mirror.SubjectTransforms) == 0 {
			if cfg.Mirror.FilterSubject != _EMPTY_ && !IsValidSubject(cfg.Mirror.FilterSubject) {
				jsa.mu.Unlock()
				return nil, fmt.Errorf("subject filter '%s' for the mirror %w", cfg.Mirror.FilterSubject, ErrBadSubject)
			}
		} else {
			for _, st := range cfg.Mirror.SubjectTransforms {
				if st.Source != _EMPTY_ && !IsValidSubject(st.Source) {
					jsa.mu.Unlock()
					return nil, fmt.Errorf("invalid subject transform source '%s' for the mirror: %w", st.Source, ErrBadSubject)
				}
				// check the transform, if any, is valid
				if st.Destination != _EMPTY_ {
					if _, err = NewSubjectTransform(st.Source, st.Destination); err != nil {
						jsa.mu.Unlock()
						return nil, fmt.Errorf("subject transform from '%s' to '%s' for the mirror: %w", st.Source, st.Destination, err)
					}
				}
			}
		}
	}

	// Setup our internal indexed names here for sources and check if the transforms (if any) are valid.
	for _, ssi := range cfg.Sources {
		if len(ssi.SubjectTransforms) == 0 {
			// check the filter, if any, is valid
			if ssi.FilterSubject != _EMPTY_ && !IsValidSubject(ssi.FilterSubject) {
				jsa.mu.Unlock()
				return nil, fmt.Errorf("subject filter '%s' for the source: %w", ssi.FilterSubject, ErrBadSubject)
			}
		} else {
			for _, st := range ssi.SubjectTransforms {
				if st.Source != _EMPTY_ && !IsValidSubject(st.Source) {
					jsa.mu.Unlock()
					return nil, fmt.Errorf("subject filter '%s' for the source: %w", st.Source, ErrBadSubject)
				}
				// check the transform, if any, is valid
				if st.Destination != _EMPTY_ {
					if _, err = NewSubjectTransform(st.Source, st.Destination); err != nil {
						jsa.mu.Unlock()
						return nil, fmt.Errorf("subject transform from '%s' to '%s' for the source: %w", st.Source, st.Destination, err)
					}
				}
			}
		}
	}

	// Check for overlapping subjects with other streams.
	// These are not allowed for now.
	if jsa.subjectsOverlap(cfg.Subjects, nil) {
		jsa.mu.Unlock()
		return nil, NewJSStreamSubjectOverlapError()
	}

	if !hasTier {
		jsa.mu.Unlock()
		return nil, fmt.Errorf("no applicable tier found")
	}

	// Setup the internal clients.
	c := s.createInternalJetStreamClient()
	ic := s.createInternalJetStreamClient()

	// Work out the stream ingest limits.
	mlen := s.opts.StreamMaxBufferedMsgs
	msz := uint64(s.opts.StreamMaxBufferedSize)
	if mlen == 0 {
		mlen = streamDefaultMaxQueueMsgs
	}
	if msz == 0 {
		msz = streamDefaultMaxQueueBytes
	}

	qpfx := fmt.Sprintf("[ACC:%s] stream '%s' ", a.Name, config.Name)
	mset := &stream{
		acc:       a,
		jsa:       jsa,
		cfg:       *cfg,
		js:        js,
		srv:       s,
		client:    c,
		sysc:      ic,
		tier:      tier,
		stype:     cfg.Storage,
		consumers: make(map[string]*consumer),
		msgs: newIPQueue[*inMsg](s, qpfx+"messages",
			ipqSizeCalculation(func(msg *inMsg) uint64 {
				return uint64(len(msg.hdr) + len(msg.msg) + len(msg.rply) + len(msg.subj))
			}),
			ipqLimitByLen[*inMsg](mlen),
			ipqLimitBySize[*inMsg](msz),
		),
		gets: newIPQueue[*directGetReq](s, qpfx+"direct gets"),
		qch:  make(chan struct{}),
		mqch: make(chan struct{}),
		uch:  make(chan struct{}, 4),
		sch:  make(chan struct{}, 1),
	}

	// Start our signaling routine to process consumers.
	mset.sigq = newIPQueue[*cMsg](s, qpfx+"obs") // of *cMsg
	go mset.signalConsumersLoop()

	// For no-ack consumers when we are interest retention.
	if cfg.Retention != LimitsPolicy {
		mset.ackq = newIPQueue[uint64](s, qpfx+"acks")
	}

	// Check for input subject transform
	if cfg.SubjectTransform != nil {
		tr, err := NewSubjectTransform(cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination)
		if err != nil {
			jsa.mu.Unlock()
			return nil, fmt.Errorf("stream subject transform from '%s' to '%s': %w", cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination, err)
		}
		mset.itr = tr
	}

	// Check for RePublish.
	if cfg.RePublish != nil {
		tr, err := NewSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
		if err != nil {
			jsa.mu.Unlock()
			return nil, fmt.Errorf("stream republish transform from '%s' to '%s': %w", cfg.RePublish.Source, cfg.RePublish.Destination, err)
		}
		// Assign our transform for republishing.
		mset.tr = tr
	}
	storeDir := filepath.Join(jsa.storeDir, streamsDir, cfg.Name)
	jsa.mu.Unlock()

	// Bind to the user account.
	c.registerWithAccount(a)
	// Bind to the system account.
	ic.registerWithAccount(s.SystemAccount())

	// Create the appropriate storage
	fsCfg := fsConfig
	if fsCfg == nil {
		fsCfg = &FileStoreConfig{}
		// If we are file based and not explicitly configured
		// we may be able to auto-tune based on max msgs or bytes.
		if cfg.Storage == FileStorage {
			mset.autoTuneFileStorageBlockSize(fsCfg)
		}
	}
	fsCfg.StoreDir = storeDir
	// Grab configured sync interval.
	fsCfg.SyncInterval = s.getOpts().SyncInterval
	fsCfg.SyncAlways = s.getOpts().SyncAlways
	fsCfg.Compression = config.Compression
	// Async flushing is only allowed if the stream has a sync log backing it.
	fsCfg.AsyncFlush = !fsCfg.SyncAlways && config.Replicas > 1

	// Async persist mode opts in to async flushing,
	// sync always would also be disabled if it was configured.
	if config.PersistMode == AsyncPersistMode {
		fsCfg.SyncAlways = false
		fsCfg.AsyncFlush = true
	}

	if err := mset.setupStore(fsCfg); err != nil {
		mset.stop(true, false)
		return nil, NewJSStreamStoreFailedError(err)
	}

	// Create our pubAck template here. Better than json marshal each time on success.
	if domain := s.getOpts().JetStreamDomain; domain != _EMPTY_ {
		mset.pubAck = fmt.Appendf(nil, "{%q:%q,%q:%q,%q:", "stream", cfg.Name, "domain", domain, "seq")
	} else {
		mset.pubAck = fmt.Appendf(nil, "{%q:%q,%q:", "stream", cfg.Name, "seq")
	}
	end := len(mset.pubAck)
	mset.pubAck = mset.pubAck[:end:end]

	// Set our known last sequence.
	var state StreamState
	mset.store.FastState(&state)

	// Possible race with consumer.setLeader during recovery.
	mset.mu.Lock()
	mset.lseq = state.LastSeq

	// Ensure dedupe state is loaded.
	mset.ddMu.Lock()
	mset.rebuildDedupe()
	mset.ddMu.Unlock()
	mset.mu.Unlock()

	// Set our stream assignment if in clustered mode.
	reserveResources := true
	if sa != nil {
		mset.setStreamAssignment(sa)

		// If the stream is resetting we must not double-account resources, they were already accounted for.
		js.mu.Lock()
		if sa.resetting {
			reserveResources, sa.resetting = false, false
		}
		js.mu.Unlock()
	}

	// Setup our internal send go routine.
	mset.setupSendCapabilities()

	// Reserve resources if MaxBytes present.
	if reserveResources {
		mset.js.reserveStreamResources(&mset.cfg)
	}

	// Call directly to set leader if not in clustered mode.
	// This can be called though before we actually setup clustering, so check both.
	if singleServerMode {
		if err := mset.setLeader(true); err != nil {
			mset.stop(true, false)
			return nil, err
		}
	}

	// This is always true in single server mode.
	if mset.IsLeader() {
		// Send advisory.
		var suppress bool
		if !s.standAloneMode() && sa == nil {
			if cfg.Replicas > 1 {
				suppress = true
			}
		} else if sa != nil {
			suppress = sa.responded
		}
		if !suppress {
			mset.sendCreateAdvisory()
		}
	}

	// Register with our account last.
	jsa.mu.Lock()
	jsa.streams[cfg.Name] = mset
	jsa.mu.Unlock()

	return mset, nil
}

// Composes the index name. Contains the stream name, subject filter, and transform destination
// when the stream is external we will use the api prefix as part of the index name
// (as the same stream name could be used in multiple JS domains)
func (ssi *StreamSource) composeIName() string {
	var iName = ssi.Name

	if ssi.External != nil {
		iName = iName + ":" + getHash(ssi.External.ApiPrefix)
	}

	source := ssi.FilterSubject
	destination := fwcs

	if len(ssi.SubjectTransforms) == 0 {
		// normalize filter and destination in case they are empty
		if source == _EMPTY_ {
			source = fwcs
		}
		if destination == _EMPTY_ {
			destination = fwcs
		}
	} else {
		var sources, destinations []string

		for _, tr := range ssi.SubjectTransforms {
			trsrc, trdest := tr.Source, tr.Destination
			if trsrc == _EMPTY_ {
				trsrc = fwcs
			}
			if trdest == _EMPTY_ {
				trdest = fwcs
			}
			sources = append(sources, trsrc)
			destinations = append(destinations, trdest)
		}
		source = strings.Join(sources, "\f")
		destination = strings.Join(destinations, "\f")
	}

	return strings.Join([]string{iName, source, destination}, " ")
}

// Sets the index name.
func (ssi *StreamSource) setIndexName() {
	ssi.iname = ssi.composeIName()
}

func (mset *stream) streamAssignment() *streamAssignment {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.sa
}

func (mset *stream) setStreamAssignment(sa *streamAssignment) {
	var node RaftNode
	var peers []string

	mset.mu.RLock()
	js := mset.js
	mset.mu.RUnlock()

	if js != nil {
		js.mu.RLock()
		if sa.Group != nil {
			node = sa.Group.node
			peers = sa.Group.Peers
		}
		js.mu.RUnlock()
	}

	mset.mu.Lock()
	defer mset.mu.Unlock()

	mset.sa = sa
	if sa == nil {
		return
	}

	// Set our node.
	mset.node = node
	if mset.node != nil {
		mset.node.UpdateKnownPeers(peers)
	}

	// Setup our info sub here as well for all stream members. This is now by design.
	if mset.infoSub == nil {
		isubj := fmt.Sprintf(clusterStreamInfoT, mset.jsa.acc(), mset.cfg.Name)
		// Note below the way we subscribe here is so that we can send requests to ourselves.
		mset.infoSub, _ = mset.srv.systemSubscribe(isubj, _EMPTY_, false, mset.sysc, mset.handleClusterStreamInfoRequest)
	}

	// Trigger update chan.
	select {
	case mset.uch <- struct{}{}:
	default:
	}
}

func (mset *stream) monitorQuitC() <-chan struct{} {
	if mset == nil {
		return nil
	}
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.mqch
}

// signalMonitorQuit signals to exit the monitor loop. If there's no Raft node,
// this will be the only way to stop the monitor goroutine.
func (mset *stream) signalMonitorQuit() {
	mset.mu.Lock()
	defer mset.mu.Unlock()
	if mset.mqch != nil {
		close(mset.mqch)
		mset.mqch = nil
	}
}

func (mset *stream) updateC() <-chan struct{} {
	if mset == nil {
		return nil
	}
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.uch
}

// IsLeader will return if we are the current leader.
func (mset *stream) IsLeader() bool {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.isLeader()
}

// Lock should be held.
func (mset *stream) isLeader() bool {
	if mset.isClustered() {
		return mset.node.Leader()
	}
	return true
}

// isLeaderNodeState should NOT be used normally, use isLeader instead.
// Returns whether the node thinks it is the leader, regardless of whether applies are up-to-date yet
// (unlike isLeader, which requires applies to be caught up).
// May be used to respond to clients after a leader change, when applying entries from a former leader.
// Lock should be held.
func (mset *stream) isLeaderNodeState() bool {
	if mset.isClustered() {
		return mset.node.State() == Leader
	}
	return true
}

// TODO(dlc) - Check to see if we can accept being the leader or we should step down.
func (mset *stream) setLeader(isLeader bool) error {
	mset.mu.Lock()
	// If we are here we have a change in leader status.
	if isLeader {
		// Make sure we are listening for sync requests.
		// TODO(dlc) - Original design was that all in sync members of the group would do DQ.
		if mset.isClustered() {
			mset.startClusterSubs()
		}

		// Setup subscriptions if we were not already the leader.
		if err := mset.subscribeToStream(); err != nil {
			if mset.isClustered() {
				// Stepdown since we have an error.
				mset.node.StepDown()
			}
			mset.mu.Unlock()
			return err
		}
	} else {
		// cancel timer to create the source consumers if not fired yet
		if mset.sourcesConsumerSetup != nil {
			mset.sourcesConsumerSetup.Stop()
			mset.sourcesConsumerSetup = nil
		} else {
			// Stop any source consumers
			mset.stopSourceConsumers()
		}

		// Stop responding to sync requests.
		mset.stopClusterSubs()
		// Unsubscribe from direct stream.
		mset.unsubscribeToStream(false, false)
		// Clear catchup state
		mset.clearAllCatchupPeers()
	}
	mset.store.ResetState()
	mset.mu.Unlock()

	// If we are interest based make sure to check consumers.
	// This is to make sure we process any outstanding acks.
	mset.checkInterestState()

	return nil
}

// Lock should be held.
func (mset *stream) startClusterSubs() {
	if mset.syncSub == nil {
		mset.syncSub, _ = mset.srv.systemSubscribe(mset.sa.Sync, _EMPTY_, false, mset.sysc, mset.handleClusterSyncRequest)
	}
}

// Lock should be held.
func (mset *stream) stopClusterSubs() {
	if mset.syncSub != nil {
		mset.srv.sysUnsubscribe(mset.syncSub)
		mset.syncSub = nil
	}
}

// account gets the account for this stream.
func (mset *stream) account() *Account {
	return mset.accountLocked(true)
}

func (mset *stream) accountLocked(needLock bool) *Account {
	if needLock {
		mset.mu.RLock()
	}
	jsa := mset.jsa
	if needLock {
		mset.mu.RUnlock()
	}
	if jsa == nil {
		return nil
	}
	return jsa.acc()
}

// Helper to determine the max msg size for this stream if file based.
func (mset *stream) maxMsgSize() uint64 {
	maxMsgSize := mset.cfg.MaxMsgSize
	if maxMsgSize <= 0 {
		// Pull from the account.
		if mset.jsa != nil {
			if acc := mset.jsa.acc(); acc != nil {
				acc.mu.RLock()
				maxMsgSize = acc.mpay
				acc.mu.RUnlock()
			}
		}
		// If all else fails use default.
		if maxMsgSize <= 0 {
			maxMsgSize = MAX_PAYLOAD_SIZE
		}
	}
	// Now determine an estimation for the subjects etc.
	maxSubject := -1
	for _, subj := range mset.cfg.Subjects {
		if subjectIsLiteral(subj) {
			if len(subj) > maxSubject {
				maxSubject = len(subj)
			}
		}
	}
	if maxSubject < 0 {
		const defaultMaxSubject = 256
		maxSubject = defaultMaxSubject
	}
	// filestore will add in estimates for record headers, etc.
	return fileStoreMsgSizeEstimate(maxSubject, int(maxMsgSize))
}

// If we are file based and the file storage config was not explicitly set
// we can autotune block sizes to better match. Our target will be to store 125%
// of the theoretical limit. We will round up to nearest 100 bytes as well.
func (mset *stream) autoTuneFileStorageBlockSize(fsCfg *FileStoreConfig) {
	var totalEstSize uint64

	// MaxBytes will take precedence for now.
	if mset.cfg.MaxBytes > 0 {
		totalEstSize = uint64(mset.cfg.MaxBytes)
	} else if mset.cfg.MaxMsgs > 0 {
		// Determine max message size to estimate.
		totalEstSize = mset.maxMsgSize() * uint64(mset.cfg.MaxMsgs)
	} else if mset.cfg.MaxMsgsPer > 0 {
		fsCfg.BlockSize = uint64(defaultKVBlockSize)
		return
	} else {
		// If nothing set will let underlying filestore determine blkSize.
		return
	}

	blkSize := (totalEstSize / 4) + 1 // (25% overhead)
	// Round up to nearest 100
	if m := blkSize % 100; m != 0 {
		blkSize += 100 - m
	}
	if blkSize <= FileStoreMinBlkSize {
		blkSize = FileStoreMinBlkSize
	} else if blkSize >= FileStoreMaxBlkSize {
		blkSize = FileStoreMaxBlkSize
	} else {
		blkSize = defaultMediumBlockSize
	}
	fsCfg.BlockSize = uint64(blkSize)
}

// rebuildDedupe will rebuild any dedupe structures needed after recovery of a stream.
// Will be called lazily to avoid penalizing startup times.
// TODO(dlc) - Might be good to know if this should be checked at all for streams with no
// headers and msgId in them. Would need signaling from the storage layer.
// mset.mu and mset.ddMu locks should be held.
func (mset *stream) rebuildDedupe() {
	// We have some messages. Lookup starting sequence by duplicate time window.
	sseq := mset.store.GetSeqFromTime(time.Now().Add(-mset.cfg.Duplicates))
	if sseq == 0 {
		return
	}

	var smv StoreMsg
	var state StreamState
	mset.store.FastState(&state)

	for seq := sseq; seq <= state.LastSeq; seq++ {
		sm, err := mset.store.LoadMsg(seq, &smv)
		if err != nil {
			continue
		}
		var msgId string
		if len(sm.hdr) > 0 {
			if msgId = getMsgId(sm.hdr); msgId != _EMPTY_ {
				mset.storeMsgIdLocked(&ddentry{msgId, sm.seq, sm.ts})
			}
		}
		if seq == state.LastSeq {
			mset.lmsgId = msgId
		}
	}
}

// Lock should be held.
func (mset *stream) lastSeqAndCLFS() (uint64, uint64) {
	return mset.lseq, mset.getCLFS()
}

func (mset *stream) getCLFS() uint64 {
	if mset == nil {
		return 0
	}
	mset.clMu.Lock()
	defer mset.clMu.Unlock()
	return mset.clfs
}

func (mset *stream) setCLFS(clfs uint64) {
	mset.clMu.Lock()
	mset.clfs = clfs
	mset.clMu.Unlock()
}

func (mset *stream) lastSeq() uint64 {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.lseq
}

// Set last seq.
// Write lock should be held.
func (mset *stream) setLastSeq(lseq uint64) {
	mset.lseq = lseq
}

func (mset *stream) sendCreateAdvisory() {
	mset.mu.RLock()
	name := mset.cfg.Name
	template := mset.cfg.Template
	outq := mset.outq
	srv := mset.srv
	mset.mu.RUnlock()

	if outq == nil {
		return
	}

	// finally send an event that this stream was created
	m := JSStreamActionAdvisory{
		TypedEvent: TypedEvent{
			Type: JSStreamActionAdvisoryType,
			ID:   nuid.Next(),
			Time: time.Now().UTC(),
		},
		Stream:   name,
		Action:   CreateEvent,
		Template: template,
		Domain:   srv.getOpts().JetStreamDomain,
	}

	j, err := json.Marshal(m)
	if err != nil {
		return
	}

	subj := JSAdvisoryStreamCreatedPre + "." + name
	outq.sendMsg(subj, j)
}

func (mset *stream) sendDeleteAdvisoryLocked() {
	if mset.outq == nil {
		return
	}

	m := JSStreamActionAdvisory{
		TypedEvent: TypedEvent{
			Type: JSStreamActionAdvisoryType,
			ID:   nuid.Next(),
			Time: time.Now().UTC(),
		},
		Stream:   mset.cfg.Name,
		Action:   DeleteEvent,
		Template: mset.cfg.Template,
		Domain:   mset.srv.getOpts().JetStreamDomain,
	}

	j, err := json.Marshal(m)
	if err == nil {
		subj := JSAdvisoryStreamDeletedPre + "." + mset.cfg.Name
		mset.outq.sendMsg(subj, j)
	}
}

func (mset *stream) sendUpdateAdvisoryLocked() {
	if mset.outq == nil {
		return
	}

	m := JSStreamActionAdvisory{
		TypedEvent: TypedEvent{
			Type: JSStreamActionAdvisoryType,
			ID:   nuid.Next(),
			Time: time.Now().UTC(),
		},
		Stream: mset.cfg.Name,
		Action: ModifyEvent,
		Domain: mset.srv.getOpts().JetStreamDomain,
	}

	j, err := json.Marshal(m)
	if err == nil {
		subj := JSAdvisoryStreamUpdatedPre + "." + mset.cfg.Name
		mset.outq.sendMsg(subj, j)
	}
}

func (mset *stream) sendStreamBatchAbandonedAdvisory(batchId string, reason BatchAbandonReason) {
	if mset == nil {
		return
	}
	s := mset.srv
	stream, acc := mset.name(), mset.account()
	subj := JSAdvisoryStreamBatchAbandonedPre + "." + stream
	adv := &JSStreamBatchAbandonedAdvisory{
		TypedEvent: TypedEvent{
			Type: JSStreamBatchAbandonedAdvisoryType,
			ID:   nuid.Next(),
			Time: time.Now().UTC(),
		},
		Stream:  stream,
		Domain:  s.getOpts().JetStreamDomain,
		BatchId: batchId,
		Reason:  reason,
	}

	// Send to the user's account if not the system account.
	if acc != s.SystemAccount() {
		s.publishAdvisory(acc, subj, adv)
	}
	// Now do system level one. Place account info in adv, and nil account means system.
	adv.Account = acc.GetName()
	s.publishAdvisory(nil, subj, adv)
}

// Created returns created time.
func (mset *stream) createdTime() time.Time {
	mset.mu.RLock()
	created := mset.created
	mset.mu.RUnlock()
	return created
}

// Internal to allow creation time to be restored.
func (mset *stream) setCreatedTime(created time.Time) {
	mset.mu.Lock()
	mset.created = created
	mset.mu.Unlock()
}

// subjectsOverlap to see if these subjects overlap with existing subjects.
// Use only for non-clustered JetStream
// RLock minimum should be held.
func (jsa *jsAccount) subjectsOverlap(subjects []string, self *stream) bool {
	for _, mset := range jsa.streams {
		if self != nil && mset == self {
			continue
		}
		for _, subj := range mset.cfg.Subjects {
			for _, tsubj := range subjects {
				if SubjectsCollide(tsubj, subj) {
					return true
				}
			}
		}
	}
	return false
}

// StreamDefaultDuplicatesWindow default duplicates window.
const StreamDefaultDuplicatesWindow = 2 * time.Minute

func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account, pedantic bool) (StreamConfig, *ApiError) {
	lim := &s.getOpts().JetStreamLimits

	if config == nil {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration invalid"))
	}
	if !isValidName(config.Name) {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream name is required and can not contain '.', '*', '>'"))
	}
	if len(config.Name) > JSMaxNameLen {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream name is too long, maximum allowed is %d", JSMaxNameLen))
	}
	if len(config.Description) > JSMaxDescriptionLen {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream description is too long, maximum allowed is %d", JSMaxDescriptionLen))
	}

	var metadataLen int
	for k, v := range config.Metadata {
		metadataLen += len(k) + len(v)
	}
	if metadataLen > JSMaxMetadataLen {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream metadata exceeds maximum size of %d bytes", JSMaxMetadataLen))
	}

	cfg := *config

	if _, err := cfg.Retention.MarshalJSON(); err != nil {
		return cfg, NewJSStreamInvalidConfigError(fmt.Errorf("invalid retention"))
	}
	if _, err := cfg.Discard.MarshalJSON(); err != nil {
		return cfg, NewJSStreamInvalidConfigError(fmt.Errorf("invalid discard policy"))
	}
	if _, err := cfg.Compression.MarshalJSON(); err != nil {
		return cfg, NewJSStreamInvalidConfigError(fmt.Errorf("invalid compression"))
	}

	// Make file the default.
	if cfg.Storage == 0 {
		cfg.Storage = FileStorage
	}
	if _, err := cfg.Storage.MarshalJSON(); err != nil {
		return cfg, NewJSStreamInvalidConfigError(fmt.Errorf("invalid storage type"))
	}

	if cfg.Replicas == 0 {
		cfg.Replicas = 1
	}
	if cfg.Replicas > StreamMaxReplicas {
		return cfg, NewJSStreamInvalidConfigError(fmt.Errorf("maximum replicas is %d", StreamMaxReplicas))
	}
	if cfg.Replicas < 0 {
		return cfg, NewJSReplicasCountCannotBeNegativeError()
	}
	if cfg.MaxMsgs == 0 || cfg.MaxMsgs < -1 {
		if pedantic && cfg.MaxMsgs < -1 {
			return StreamConfig{}, NewJSPedanticError(fmt.Errorf("max_msgs must be set to -1"))
		}
		cfg.MaxMsgs = -1
	}
	if cfg.MaxMsgsPer == 0 || cfg.MaxMsgsPer < -1 {
		if pedantic && cfg.MaxMsgsPer < -1 {
			return StreamConfig{}, NewJSPedanticError(fmt.Errorf("max_msgs_per_subject must be set to -1"))
		}
		cfg.MaxMsgsPer = -1
	}
	if cfg.MaxBytes == 0 || cfg.MaxBytes < -1 {
		if pedantic && cfg.MaxBytes < -1 {
			return StreamConfig{}, NewJSPedanticError(fmt.Errorf("max_bytes must be set to -1"))
		}
		cfg.MaxBytes = -1
	}
	if cfg.MaxMsgSize == 0 || cfg.MaxMsgSize < -1 {
		if pedantic && cfg.MaxMsgSize < -1 {
			return StreamConfig{}, NewJSPedanticError(fmt.Errorf("max_msg_size must be set to -1"))
		}
		cfg.MaxMsgSize = -1
	}
	if cfg.MaxConsumers == 0 || cfg.MaxConsumers < -1 {
		if pedantic && cfg.MaxConsumers < -1 {
			return StreamConfig{}, NewJSPedanticError(fmt.Errorf("max_consumers must be set to -1"))
		}
		cfg.MaxConsumers = -1
	}
	if cfg.MaxAge < 0 {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("max age can not be negative"))
	}
	if cfg.MaxAge != 0 && cfg.MaxAge < 100*time.Millisecond {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("max age needs to be >= 100ms"))
	}
	if cfg.Duplicates == 0 && cfg.Mirror == nil {
		maxWindow := StreamDefaultDuplicatesWindow
		if lim.Duplicates > 0 && maxWindow > lim.Duplicates {
			if pedantic {
				return StreamConfig{}, NewJSPedanticError(fmt.Errorf("duplicate window limits are higher than current limits"))
			}
			maxWindow = lim.Duplicates
		}
		if cfg.MaxAge != 0 && cfg.MaxAge < maxWindow {
			if pedantic {
				return StreamConfig{}, NewJSPedanticError(fmt.Errorf("duplicate window cannot be bigger than max age"))
			}
			cfg.Duplicates = cfg.MaxAge
		} else {
			cfg.Duplicates = maxWindow
		}
	}
	if cfg.Duplicates < 0 {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicates window can not be negative"))
	}
	// Check that duplicates is not larger then age if set.
	if cfg.MaxAge != 0 && cfg.Duplicates > cfg.MaxAge {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicates window can not be larger then max age"))
	}
	if lim.Duplicates > 0 && cfg.Duplicates > lim.Duplicates {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicates window can not be larger then server limit of %v",
			lim.Duplicates.String()))
	}
	if cfg.Duplicates > 0 && cfg.Duplicates < 100*time.Millisecond {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicates window needs to be >= 100ms"))
	}

	if cfg.DenyPurge && cfg.AllowRollup {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("roll-ups require the purge permission"))
	}

	// Counter is not compatible with some settings.
	if cfg.AllowMsgCounter {
		if cfg.Discard == DiscardNew {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("counter stream cannot use discard new"))
		}
		if cfg.AllowMsgTTL {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("counter stream cannot use message TTLs"))
		}
		if cfg.Retention != LimitsPolicy {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("counter stream can only use limits retention"))
		}
	}

	// Check for new discard new per subject, we require the discard policy to also be new.
	if cfg.DiscardNewPer {
		if cfg.Discard != DiscardNew {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("discard new per subject requires discard new policy to be set"))
		}
		if cfg.MaxMsgsPer <= 0 {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("discard new per subject requires max msgs per subject > 0"))
		}
	}

	if cfg.SubjectDeleteMarkerTTL > 0 {
		if cfg.SubjectDeleteMarkerTTL < time.Second {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subject delete marker TTL must be at least 1 second"))
		}
		if !cfg.AllowMsgTTL {
			if pedantic {
				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subject delete marker cannot be set if message TTLs are disabled"))
			}
			cfg.AllowMsgTTL = true
		}
		if !cfg.AllowRollup {
			if pedantic {
				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subject delete marker cannot be set if roll-ups are disabled"))
			}
			cfg.AllowRollup, cfg.DenyPurge = true, false
		}
	} else if cfg.SubjectDeleteMarkerTTL < 0 {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subject delete marker TTL must not be negative"))
	}

	if cfg.AllowMsgSchedules {
		if !cfg.AllowRollup {
			if pedantic {
				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("message scheduling cannot be set if roll-ups are disabled"))
			}
			cfg.AllowRollup, cfg.DenyPurge = true, false
		}
	}

	if cfg.PersistMode == AsyncPersistMode {
		if cfg.Storage != FileStorage {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("async persist mode is only supported on file storage"))
		}
		if cfg.Replicas > 1 {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("async persist mode is not supported on replicated streams"))
		}
		if cfg.AllowAtomicPublish {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("async persist mode is not supported with atomic batch publish"))
		}
	}

	getStream := func(streamName string) (bool, StreamConfig) {
		var exists bool
		var cfg StreamConfig
		if s.JetStreamIsClustered() {
			if js, _ := s.getJetStreamCluster(); js != nil {
				js.mu.RLock()
				if sa := js.streamAssignment(acc.Name, streamName); sa != nil {
					cfg = *sa.Config.clone()
					exists = true
				}
				js.mu.RUnlock()
			}
		} else if mset, err := acc.lookupStream(streamName); err == nil {
			cfg = mset.cfg
			exists = true
		}
		return exists, cfg
	}

	hasStream := func(streamName string) (bool, int32, []string) {
		exists, cfg := getStream(streamName)
		return exists, cfg.MaxMsgSize, cfg.Subjects
	}

	var streamSubs []string
	var deliveryPrefixes []string
	var apiPrefixes []string

	// Do some pre-checking for mirror config to avoid cycles in clustered mode.
	if cfg.Mirror != nil {
		if cfg.FirstSeq > 0 {
			return StreamConfig{}, NewJSMirrorWithFirstSeqError()
		}
		if len(cfg.Subjects) > 0 {
			return StreamConfig{}, NewJSMirrorWithSubjectsError()
		}
		if len(cfg.Sources) > 0 {
			return StreamConfig{}, NewJSMirrorWithSourcesError()
		}
		if cfg.AllowMsgCounter {
			return StreamConfig{}, NewJSMirrorWithCountersError()
		}
		if cfg.AllowAtomicPublish {
			return StreamConfig{}, NewJSMirrorWithAtomicPublishError()
		}
		if cfg.AllowMsgSchedules {
			return StreamConfig{}, NewJSMirrorWithMsgSchedulesError()
		}
		if cfg.Mirror.FilterSubject != _EMPTY_ && len(cfg.Mirror.SubjectTransforms) != 0 {
			return StreamConfig{}, NewJSMirrorMultipleFiltersNotAllowedError()
		}
		if cfg.SubjectDeleteMarkerTTL > 0 {
			// Delete markers cannot be configured on a mirror as it would result in new
			// tombstones which would use up sequence numbers, diverging from the origin
			// stream.
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subject delete markers forbidden on mirrors"))
		}
		// Check subject filters overlap.
		for outer, tr := range cfg.Mirror.SubjectTransforms {
			if tr.Source != _EMPTY_ && !IsValidSubject(tr.Source) {
				return StreamConfig{}, NewJSMirrorInvalidSubjectFilterError(fmt.Errorf("%w %s", ErrBadSubject, tr.Source))
			}

			err := ValidateMapping(tr.Source, tr.Destination)
			if err != nil {
				return StreamConfig{}, NewJSMirrorInvalidTransformDestinationError(err)
			}

			for inner, innertr := range cfg.Mirror.SubjectTransforms {
				if inner != outer && SubjectsCollide(tr.Source, innertr.Source) {
					return StreamConfig{}, NewJSMirrorOverlappingSubjectFiltersError()
				}
			}
		}
		// Do not perform checks if External is provided, as it could lead to
		// checking against itself (if sourced stream name is the same on different JetStream)
		if cfg.Mirror.External == nil {
			if !isValidName(cfg.Mirror.Name) {
				return StreamConfig{}, NewJSMirrorInvalidStreamNameError()
			}
			// We do not require other stream to exist anymore, but if we can see it check payloads.
			exists, maxMsgSize, subs := hasStream(cfg.Mirror.Name)
			if len(subs) > 0 {
				streamSubs = append(streamSubs, subs...)
			}
			if exists {
				if cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
					return StreamConfig{}, NewJSMirrorMaxMessageSizeTooBigError()
				}
			}
			// Determine if we are inheriting direct gets.
			if exists, ocfg := getStream(cfg.Mirror.Name); exists {
				if pedantic && cfg.MirrorDirect != ocfg.AllowDirect {
					return StreamConfig{}, NewJSPedanticError(fmt.Errorf("origin stream has direct get set, mirror has it disabled"))
				}
				cfg.MirrorDirect = ocfg.AllowDirect
			} else if js := s.getJetStream(); js != nil && js.isClustered() {
				// Could not find it here. If we are clustered we can look it up.
				js.mu.RLock()
				if cc := js.cluster; cc != nil {
					if as := cc.streams[acc.Name]; as != nil {
						if sa := as[cfg.Mirror.Name]; sa != nil {
							if pedantic && cfg.MirrorDirect != sa.Config.AllowDirect {
								js.mu.RUnlock()
								return StreamConfig{}, NewJSPedanticError(fmt.Errorf("origin stream has direct get set, mirror has it disabled"))
							}
							cfg.MirrorDirect = sa.Config.AllowDirect
						}
					}
				}
				js.mu.RUnlock()
			}
		} else {
			if cfg.Mirror.External.DeliverPrefix != _EMPTY_ {
				deliveryPrefixes = append(deliveryPrefixes, cfg.Mirror.External.DeliverPrefix)
			}

			if cfg.Mirror.External.ApiPrefix != _EMPTY_ {
				apiPrefixes = append(apiPrefixes, cfg.Mirror.External.ApiPrefix)
			}
		}
	}

	if len(cfg.Sources) > 0 {
		if cfg.AllowMsgSchedules {
			return StreamConfig{}, NewJSSourceWithMsgSchedulesError()
		}
	}

	// check for duplicates
	var iNames = make(map[string]struct{})
	for _, src := range cfg.Sources {
		if !isValidName(src.Name) {
			return StreamConfig{}, NewJSSourceInvalidStreamNameError()
		}
		if _, ok := iNames[src.composeIName()]; !ok {
			iNames[src.composeIName()] = struct{}{}
		} else {
			return StreamConfig{}, NewJSSourceDuplicateDetectedError()
		}
		// Do not perform checks if External is provided, as it could lead to
		// checking against itself (if sourced stream name is the same on different JetStream)
		if src.External == nil {
			exists, maxMsgSize, subs := hasStream(src.Name)
			if len(subs) > 0 {
				streamSubs = append(streamSubs, subs...)
			}
			if exists {
				if cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
					return StreamConfig{}, NewJSSourceMaxMessageSizeTooBigError()
				}
			}

			if src.FilterSubject != _EMPTY_ && len(src.SubjectTransforms) != 0 {
				return StreamConfig{}, NewJSSourceMultipleFiltersNotAllowedError()
			}

			for _, tr := range src.SubjectTransforms {
				if tr.Source != _EMPTY_ && !IsValidSubject(tr.Source) {
					return StreamConfig{}, NewJSSourceInvalidSubjectFilterError(fmt.Errorf("%w %s", ErrBadSubject, tr.Source))
				}

				err := ValidateMapping(tr.Source, tr.Destination)
				if err != nil {
					return StreamConfig{}, NewJSSourceInvalidTransformDestinationError(err)
				}
			}

			// Check subject filters overlap.
			for outer, tr := range src.SubjectTransforms {
				for inner, innertr := range src.SubjectTransforms {
					if inner != outer && subjectIsSubsetMatch(tr.Source, innertr.Source) {
						return StreamConfig{}, NewJSSourceOverlappingSubjectFiltersError()
					}
				}
			}
			continue
		} else {
			if src.External.DeliverPrefix != _EMPTY_ {
				deliveryPrefixes = append(deliveryPrefixes, src.External.DeliverPrefix)
			}
			if src.External.ApiPrefix != _EMPTY_ {
				apiPrefixes = append(apiPrefixes, src.External.ApiPrefix)
			}
		}
	}

	// check prefix overlap with subjects
	for _, pfx := range deliveryPrefixes {
		if !IsValidPublishSubject(pfx) {
			return StreamConfig{}, NewJSStreamInvalidExternalDeliverySubjError(pfx)
		}
		for _, sub := range streamSubs {
			if SubjectsCollide(sub, fmt.Sprintf("%s.%s", pfx, sub)) {
				return StreamConfig{}, NewJSStreamExternalDelPrefixOverlapsError(pfx, sub)
			}
		}
	}
	// check if api prefixes overlap
	for _, apiPfx := range apiPrefixes {
		if !IsValidPublishSubject(apiPfx) {
			return StreamConfig{}, NewJSStreamInvalidConfigError(
				fmt.Errorf("stream external api prefix %q must be a valid subject without wildcards", apiPfx))
		}
		if SubjectsCollide(apiPfx, JSApiPrefix) {
			return StreamConfig{}, NewJSStreamExternalApiOverlapError(apiPfx, JSApiPrefix)
		}
	}

	// cycle check for source cycle
	toVisit := []*StreamConfig{&cfg}
	visited := make(map[string]struct{})
	overlaps := func(subjects []string, filter string) bool {
		if filter == _EMPTY_ {
			return true
		}
		for _, subject := range subjects {
			if SubjectsCollide(subject, filter) {
				return true
			}
		}
		return false
	}

	for len(toVisit) > 0 {
		cfg := toVisit[0]
		toVisit = toVisit[1:]
		visited[cfg.Name] = struct{}{}
		for _, src := range cfg.Sources {
			if src.External != nil {
				continue
			}
			// We can detect a cycle between streams, but let's double check that the
			// subjects actually form a cycle.
			if _, ok := visited[src.Name]; ok {
				if overlaps(cfg.Subjects, src.FilterSubject) {
					return StreamConfig{}, NewJSStreamInvalidConfigError(errors.New("detected cycle"))
				}
			} else if exists, cfg := getStream(src.Name); exists {
				toVisit = append(toVisit, &cfg)
			}
		}
		// Avoid cycles hiding behind mirrors
		if m := cfg.Mirror; m != nil {
			if m.External == nil {
				if _, ok := visited[m.Name]; ok {
					return StreamConfig{}, NewJSStreamInvalidConfigError(errors.New("detected cycle"))
				}
				if exists, cfg := getStream(m.Name); exists {
					toVisit = append(toVisit, &cfg)
				}
			}
		}
	}

	if len(cfg.Subjects) == 0 {
		if cfg.Mirror == nil && len(cfg.Sources) == 0 {
			cfg.Subjects = append(cfg.Subjects, cfg.Name)
		}
	} else {
		if cfg.Mirror != nil {
			return StreamConfig{}, NewJSMirrorWithSubjectsError()
		}

		// Check for literal duplication of subject interest in config
		// and no overlap with any JS or SYS API subject space.
		dset := make(map[string]struct{}, len(cfg.Subjects))
		for _, subj := range cfg.Subjects {
			// Make sure the subject is valid. Check this first.
			if !IsValidSubject(subj) {
				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("invalid subject"))
			}
			if _, ok := dset[subj]; ok {
				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicate subjects detected"))
			}
			// Check for trying to capture everything.
			if subj == fwcs {
				if !cfg.NoAck {
					return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("capturing all subjects requires no-ack to be true"))
				}
				// Capturing everything also will require R1.
				if cfg.Replicas != 1 {
					return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("capturing all subjects requires replicas of 1"))
				}
			}
			// Also check to make sure we do not overlap with our $JS API subjects.
			if !cfg.NoAck {
				for _, namespace := range []string{"$JS.>", "$JSC.>", "$NRG.>"} {
					if SubjectsCollide(subj, namespace) {
						// We allow an exception for $JS.EVENT.> since these could have been created in the past.
						if !subjectIsSubsetMatch(subj, "$JS.EVENT.>") {
							return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subjects that overlap with jetstream api require no-ack to be true"))
						}
					}
				}
				if SubjectsCollide(subj, "$SYS.>") {
					if !subjectIsSubsetMatch(subj, "$SYS.ACCOUNT.>") {
						return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subjects that overlap with system api require no-ack to be true"))
					}
				}
			}
			// Mark for duplicate check.
			dset[subj] = struct{}{}
		}
	}

	if len(cfg.Subjects) == 0 && len(cfg.Sources) == 0 && cfg.Mirror == nil {
		return StreamConfig{}, NewJSStreamInvalidConfigError(
			fmt.Errorf("stream needs at least one configured subject or be a source/mirror"))
	}

	// Check for MaxBytes required and it's limit
	if required, limit := acc.maxBytesLimits(&cfg); required && cfg.MaxBytes <= 0 {
		return StreamConfig{}, NewJSStreamMaxBytesRequiredError()
	} else if limit > 0 && cfg.MaxBytes > limit {
		return StreamConfig{}, NewJSStreamMaxStreamBytesExceededError()
	}

	// Now check if we have multiple subjects they we do not overlap ourselves
	// which would cause duplicate entries (assuming no MsgID).
	if len(cfg.Subjects) > 1 {
		for _, subj := range cfg.Subjects {
			for _, tsubj := range cfg.Subjects {
				if tsubj != subj && SubjectsCollide(tsubj, subj) {
					return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subject %q overlaps with %q", subj, tsubj))
				}
			}
		}
	}

	// If we have a republish directive check if we can create a transform here.
	if cfg.RePublish != nil {
		// Check to make sure source is a valid subset of the subjects we have.
		// Also make sure it does not form a cycle.
		// Empty same as all.
		if cfg.RePublish.Source == _EMPTY_ {
			if pedantic {
				return StreamConfig{}, NewJSPedanticError(fmt.Errorf("republish source can not be empty"))
			}
			cfg.RePublish.Source = fwcs
		}
		var formsCycle bool
		for _, subj := range cfg.Subjects {
			if SubjectsCollide(cfg.RePublish.Destination, subj) {
				formsCycle = true
				break
			}
		}
		if formsCycle {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish destination forms a cycle"))
		}
		if _, err := NewSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish with transform from '%s' to '%s' not valid", cfg.RePublish.Source, cfg.RePublish.Destination))
		}
	}

	// Check the subject transform if any
	if cfg.SubjectTransform != nil {
		if cfg.SubjectTransform.Source != _EMPTY_ && !IsValidSubject(cfg.SubjectTransform.Source) {
			return StreamConfig{}, NewJSStreamTransformInvalidSourceError(fmt.Errorf("%w %s", ErrBadSubject, cfg.SubjectTransform.Source))
		}

		err := ValidateMapping(cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination)
		if err != nil {
			return StreamConfig{}, NewJSStreamTransformInvalidDestinationError(err)
		}
	}

	// Remove placement if it's an empty object.
	if cfg.Placement != nil && reflect.DeepEqual(cfg.Placement, &Placement{}) {
		cfg.Placement = nil
	}
	// For now don't allow preferred server in placement.
	if cfg.Placement != nil && cfg.Placement.Preferred != _EMPTY_ {
		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("preferred server not permitted in placement"))
	}

	return cfg, nil
}

// Config returns the stream's configuration.
func (mset *stream) config() StreamConfig {
	mset.cfgMu.RLock()
	defer mset.cfgMu.RUnlock()
	return mset.cfg
}

func (mset *stream) fileStoreConfig() (FileStoreConfig, error) {
	mset.mu.Lock()
	defer mset.mu.Unlock()
	fs, ok := mset.store.(*fileStore)
	if !ok {
		return FileStoreConfig{}, ErrStoreWrongType
	}
	return fs.fileStoreConfig(), nil
}

// Do not hold jsAccount or jetStream lock
func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig, s *Server, pedantic bool) (*StreamConfig, error) {
	cfg, apiErr := s.checkStreamCfg(new, jsa.acc(), pedantic)
	if apiErr != nil {
		return nil, apiErr
	}

	// Name must match.
	if cfg.Name != old.Name {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration name must match original"))
	}
	// Can't change MaxConsumers for now.
	if cfg.MaxConsumers != old.MaxConsumers {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not change MaxConsumers"))
	}
	// Can't change storage types.
	if cfg.Storage != old.Storage {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not change storage type"))
	}
	// Can only change retention from limits to interest or back, not to/from work queue for now.
	if cfg.Retention != old.Retention {
		if old.Retention == WorkQueuePolicy || cfg.Retention == WorkQueuePolicy {
			return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not change retention policy to/from workqueue"))
		}
	}
	// Can not have a template owner for now.
	if old.Template != _EMPTY_ {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update not allowed on template owned stream"))
	}
	if cfg.Template != _EMPTY_ {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not be owned by a template"))
	}
	// Can not change from true to false.
	if !cfg.Sealed && old.Sealed {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not unseal a sealed stream"))
	}
	// Can not change from true to false.
	if !cfg.DenyDelete && old.DenyDelete {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not cancel deny message deletes"))
	}
	// Can not change from true to false.
	if !cfg.DenyPurge && old.DenyPurge {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not cancel deny purge"))
	}
	// Check for mirror changes which are not allowed.
	// We will allow removing the mirror config to "promote" the mirror to a normal stream.
	if cfg.Mirror != nil && !reflect.DeepEqual(cfg.Mirror, old.Mirror) {
		return nil, NewJSStreamMirrorNotUpdatableError()
	}

	// Check on new discard new per subject.
	if cfg.DiscardNewPer {
		if cfg.Discard != DiscardNew {
			return nil, NewJSStreamInvalidConfigError(fmt.Errorf("discard new per subject requires discard new policy to be set"))
		}
		if cfg.MaxMsgsPer <= 0 {
			return nil, NewJSStreamInvalidConfigError(fmt.Errorf("discard new per subject requires max msgs per subject > 0"))
		}
	}

	// Check on the allowed message TTL status.
	if old.AllowMsgTTL && !cfg.AllowMsgTTL {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("message TTL status can not be disabled"))
	}

	// Can't change counter setting.
	if cfg.AllowMsgCounter != old.AllowMsgCounter {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not change message counter setting"))
	}

	// Can't disable message schedules setting.
	if old.AllowMsgSchedules && !cfg.AllowMsgSchedules {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("message schedules can not be disabled"))
	}

	if old.PersistMode != cfg.PersistMode {
		return nil, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration update can not change persist mode"))
	}

	// Do some adjustments for being sealed.
	// Pedantic mode will allow those changes to be made, as they are deterministic and important to get a sealed stream.
	if cfg.Sealed {
		cfg.MaxAge = 0
		cfg.Discard = DiscardNew
		cfg.DenyDelete, cfg.DenyPurge = true, true
		cfg.AllowRollup = false
	}

	// Check limits. We need some extra handling to allow updating MaxBytes.

	// First, let's calculate the difference between the new and old MaxBytes.
	maxBytesDiff := max(cfg.MaxBytes, 0) - max(old.MaxBytes, 0)
	if maxBytesDiff < 0 {
		// If we're updating to a lower MaxBytes (maxBytesDiff is negative),
		// then set to zero so checkBytesLimits doesn't set addBytes to 1.
		maxBytesDiff = 0
	}
	// If maxBytesDiff == 0, then that means MaxBytes didn't change.
	// If maxBytesDiff > 0, then we want to reserve additional bytes.

	// Save the user configured MaxBytes.
	newMaxBytes := cfg.MaxBytes
	maxBytesOffset := int64(0)

	// We temporarily set cfg.MaxBytes to maxBytesDiff because checkAllLimits
	// adds cfg.MaxBytes to the current reserved limit and checks if we've gone
	// over. However, we don't want an addition cfg.MaxBytes, we only want to
	// reserve the difference between the new and the old values.
	cfg.MaxBytes = maxBytesDiff

	// Check limits.
	js, isClustered := jsa.jetStreamAndClustered()
	jsa.mu.RLock()
	acc := jsa.account
	jsa.usageMu.RLock()
	selected, tier, hasTier := jsa.selectLimits(cfg.Replicas)
	if !hasTier && old.Replicas != cfg.Replicas {
		selected, tier, hasTier = jsa.selectLimits(old.Replicas)
	}
	jsa.usageMu.RUnlock()
	reserved := int64(0)
	if !isClustered {
		reserved = jsa.tieredReservation(tier, &cfg)
	}
	jsa.mu.RUnlock()
	if !hasTier {
		return nil, NewJSNoLimitsError()
	}
	js.mu.RLock()
	defer js.mu.RUnlock()
	if isClustered {
		_, reserved = tieredStreamAndReservationCount(js.cluster.streams[acc.Name], tier, &cfg)
	}
	// reservation does not account for this stream, hence add the old value
	if tier == _EMPTY_ && old.Replicas > 1 {
		reserved += old.MaxBytes * int64(old.Replicas)
	} else {
		reserved += old.MaxBytes
	}
	if err := js.checkAllLimits(&selected, &cfg, reserved, maxBytesOffset); err != nil {
		return nil, err
	}
	// Restore the user configured MaxBytes.
	cfg.MaxBytes = newMaxBytes
	return &cfg, nil
}

// Update will allow certain configuration properties of an existing stream to be updated.
func (mset *stream) update(config *StreamConfig) error {
	return mset.updateWithAdvisory(config, true, false)
}

func (mset *stream) updatePedantic(config *StreamConfig, pedantic bool) error {
	return mset.updateWithAdvisory(config, true, pedantic)
}

// Update will allow certain configuration properties of an existing stream to be updated.
func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool, pedantic bool) error {
	_, jsa, err := mset.acc.checkForJetStream()
	if err != nil {
		return err
	}

	mset.mu.RLock()
	ocfg := mset.cfg
	s := mset.srv
	mset.mu.RUnlock()

	cfg, err := mset.jsa.configUpdateCheck(&ocfg, config, s, pedantic)
	if err != nil {
		return NewJSStreamInvalidConfigError(err, Unless(err))
	}

	// In the event that some of the stream-level limits have changed, yell appropriately
	// if any of the consumers exceed that limit.
	updateLimits := ocfg.ConsumerLimits.InactiveThreshold != cfg.ConsumerLimits.InactiveThreshold ||
		ocfg.ConsumerLimits.MaxAckPending != cfg.ConsumerLimits.MaxAckPending
	if updateLimits {
		var errorConsumers []string
		consumers := map[string]*ConsumerConfig{}
		if mset.js.isClustered() {
			for _, c := range mset.sa.consumers {
				consumers[c.Name] = c.Config
			}
		} else {
			for _, c := range mset.consumers {
				consumers[c.name] = &c.cfg
			}
		}
		for name, ccfg := range consumers {
			if ccfg.InactiveThreshold > cfg.ConsumerLimits.InactiveThreshold ||
				ccfg.MaxAckPending > cfg.ConsumerLimits.MaxAckPending {
				errorConsumers = append(errorConsumers, name)
			}
		}
		if len(errorConsumers) > 0 {
			// TODO(nat): Return a parsable error so that we can surface something
			// sensible through the JS API.
			return fmt.Errorf("change to limits violates consumers: %s", strings.Join(errorConsumers, ", "))
		}
	}

	jsa.mu.RLock()
	if jsa.subjectsOverlap(cfg.Subjects, mset) {
		jsa.mu.RUnlock()
		return NewJSStreamSubjectOverlapError()
	}
	jsa.mu.RUnlock()

	mset.mu.Lock()
	if mset.isLeader() {
		// Check for mirror promotion.
		if ocfg.Mirror != nil && cfg.Mirror == nil {
			mset.cancelMirrorConsumer()
			mset.mirror = nil
		}

		// Now check for subject interest differences.
		current := make(map[string]struct{}, len(ocfg.Subjects))
		for _, s := range ocfg.Subjects {
			current[s] = struct{}{}
		}
		// Update config with new values. The store update will enforce any stricter limits.

		// Now walk new subjects. All of these need to be added, but we will check
		// the originals first, since if it is in there we can skip, already added.
		for _, s := range cfg.Subjects {
			if _, ok := current[s]; !ok {
				if _, err := mset.subscribeInternal(s, mset.processInboundJetStreamMsg); err != nil {
					mset.mu.Unlock()
					return err
				}
			}
			delete(current, s)
		}
		// What is left in current needs to be deleted.
		for s := range current {
			if err := mset.unsubscribeInternal(s); err != nil {
				mset.mu.Unlock()
				return err
			}
		}

		// Check for the Duplicates
		mset.ddMu.Lock()
		if cfg.Duplicates != ocfg.Duplicates && mset.ddtmr != nil {
			// Let it fire right away, it will adjust properly on purge.
			mset.ddtmr.Reset(time.Microsecond)
		}
		mset.ddMu.Unlock()

		// Check for Sources.
		if len(cfg.Sources) > 0 || len(ocfg.Sources) > 0 {
			currentIName := make(map[string]struct{})
			needsStartingSeqNum := make(map[string]struct{})

			for _, s := range ocfg.Sources {
				currentIName[s.iname] = struct{}{}
			}
			for _, s := range cfg.Sources {
				s.setIndexName()
				if _, ok := currentIName[s.iname]; !ok {
					// new source
					if mset.sources == nil {
						mset.sources = make(map[string]*sourceInfo)
					}
					mset.cfg.Sources = append(mset.cfg.Sources, s)

					var si *sourceInfo

					if len(s.SubjectTransforms) == 0 {
						si = &sourceInfo{name: s.Name, iname: s.iname, sf: s.FilterSubject}
					} else {
						si = &sourceInfo{name: s.Name, iname: s.iname}
						si.trs = make([]*subjectTransform, len(s.SubjectTransforms))
						si.sfs = make([]string, len(s.SubjectTransforms))
						for i := range s.SubjectTransforms {
							// err can be ignored as already validated in config check
							si.sfs[i] = s.SubjectTransforms[i].Source
							var err error
							si.trs[i], err = NewSubjectTransform(s.SubjectTransforms[i].Source, s.SubjectTransforms[i].Destination)
							if err != nil {
								mset.mu.Unlock()
								return fmt.Errorf("unable to get subject transform for source: %v", err)
							}
						}
					}

					mset.sources[s.iname] = si
					needsStartingSeqNum[s.iname] = struct{}{}
				} else {
					// source already exists
					delete(currentIName, s.iname)
				}
			}
			// What is left in currentIName needs to be deleted.
			for iName := range currentIName {
				mset.cancelSourceConsumer(iName)
				delete(mset.sources, iName)
			}
			neededCopy := make(map[string]struct{}, len(needsStartingSeqNum))
			for iName := range needsStartingSeqNum {
				neededCopy[iName] = struct{}{}
			}
			mset.setStartingSequenceForSources(needsStartingSeqNum)
			for iName := range neededCopy {
				mset.setupSourceConsumer(iName, mset.sources[iName].sseq+1, time.Time{})
			}
		}
	}

	// Check for a change in allow direct/mirror status.
	// These will run on all members, so just update as appropriate here.
	// We do make sure we are caught up under monitorStream() during initial startup.
	if cfg.AllowDirect != ocfg.AllowDirect {
		if cfg.AllowDirect {
			mset.subscribeToDirect()
		} else {
			mset.unsubscribeToDirect()
		}
	}
	if cfg.MirrorDirect != ocfg.MirrorDirect {
		if cfg.MirrorDirect {
			mset.subscribeToMirrorDirect()
		} else {
			mset.unsubscribeToMirrorDirect()
		}
	}

	// Check for changes to RePublish.
	if cfg.RePublish != nil {
		// Empty same as all.
		if cfg.RePublish.Source == _EMPTY_ {
			cfg.RePublish.Source = fwcs
		}
		if cfg.RePublish.Destination == _EMPTY_ {
			cfg.RePublish.Destination = fwcs
		}
		tr, err := NewSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
		if err != nil {
			mset.mu.Unlock()
			return fmt.Errorf("stream configuration for republish from '%s' to '%s': %w", cfg.RePublish.Source, cfg.RePublish.Destination, err)
		}
		// Assign our transform for republishing.
		mset.tr = tr
	} else {
		mset.tr = nil
	}

	// Check for changes to subject transform
	if ocfg.SubjectTransform == nil && cfg.SubjectTransform != nil {
		tr, err := NewSubjectTransform(cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination)
		if err != nil {
			mset.mu.Unlock()
			return fmt.Errorf("stream configuration for subject transform from '%s' to '%s': %w", cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination, err)
		}
		mset.itr = tr
	} else if ocfg.SubjectTransform != nil && cfg.SubjectTransform != nil &&
		(ocfg.SubjectTransform.Source != cfg.SubjectTransform.Source || ocfg.SubjectTransform.Destination != cfg.SubjectTransform.Destination) {
		tr, err := NewSubjectTransform(cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination)
		if err != nil {
			mset.mu.Unlock()
			return fmt.Errorf("stream configuration for subject transform from '%s' to '%s': %w", cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination, err)
		}
		mset.itr = tr
	} else if ocfg.SubjectTransform != nil && cfg.SubjectTransform == nil {
		mset.itr = nil
	}

	js := mset.js

	if targetTier := tierName(cfg.Replicas); mset.tier != targetTier {
		// In cases such as R1->R3, only one update is needed
		jsa.usageMu.RLock()
		_, ok := jsa.limits[targetTier]
		jsa.usageMu.RUnlock()
		if ok {
			// error never set
			_, reported, _ := mset.store.Utilization()
			jsa.updateUsage(mset.tier, mset.stype, -int64(reported))
			jsa.updateUsage(targetTier, mset.stype, int64(reported))
			mset.tier = targetTier
		}
		// else in case the new tier does not exist (say on move), keep the old tier around
		// a subsequent update to an existing tier will then move from existing past tier to existing new tier
	}

	if mset.isLeader() && mset.sa != nil && ocfg.Retention != cfg.Retention && cfg.Retention == InterestPolicy {
		// Before we can update the retention policy for the consumer, we need
		// the replica count of all consumers to match the stream.
		for _, c := range mset.sa.consumers {
			if c.Config.Replicas > 0 && c.Config.Replicas != cfg.Replicas {
				mset.mu.Unlock()
				return fmt.Errorf("consumer %q replica count must be %d", c.Name, cfg.Replicas)
			}
		}
	}

	// If atomic publish is disabled, delete any in-progress batches.
	if !cfg.AllowAtomicPublish {
		mset.deleteInflightBatches(false)
		mset.deleteBatchApplyState()
	}

	// Now update config and store's version of our config.
	// Although we are under the stream write lock, we will also assign the new
	// configuration under mset.cfgMu lock. This is so that in places where
	// mset.mu cannot be acquired (like many cases in consumer.go where code
	// is under the consumer's lock), and the stream's configuration needs to
	// be inspected, one can use mset.cfgMu's read lock to do that safely.
	mset.cfgMu.Lock()
	mset.cfg = *cfg
	mset.cfgMu.Unlock()

	// If we're changing retention and haven't errored because of consumer
	// replicas by now, whip through and update the consumer retention.
	if ocfg.Retention != cfg.Retention {
		toUpdate := make([]*consumer, 0, len(mset.consumers))
		for _, c := range mset.consumers {
			toUpdate = append(toUpdate, c)
		}
		var ss StreamState
		mset.store.FastState(&ss)
		mset.mu.Unlock()
		for _, c := range toUpdate {
			c.mu.Lock()
			c.retention = cfg.Retention
			c.mu.Unlock()
			if c.retention == InterestPolicy {
				// If we're switching to interest, force a check of the
				// interest of existing stream messages.
				c.checkStateForInterestStream(&ss)
			}
		}
		mset.mu.Lock()
	}

	// If we are the leader never suppress update advisory, simply send.
	if mset.isLeader() && sendAdvisory {
		mset.sendUpdateAdvisoryLocked()
	}
	mset.mu.Unlock()

	if js != nil {
		maxBytesDiff := max(cfg.MaxBytes, 0) - max(ocfg.MaxBytes, 0)
		if maxBytesDiff > 0 {
			// Reserve the difference
			js.reserveStreamResources(&StreamConfig{
				MaxBytes: maxBytesDiff,
				Storage:  cfg.Storage,
			})
		} else if maxBytesDiff < 0 {
			// Release the difference
			js.releaseStreamResources(&StreamConfig{
				MaxBytes: -maxBytesDiff,
				Storage:  ocfg.Storage,
			})
		}
	}

	mset.store.UpdateConfig(cfg)

	return nil
}

// Small helper to return the Name field from mset.cfg, protected by
// the mset.cfgMu mutex. This is simply because we have several places
// in consumer.go where we need it.
func (mset *stream) getCfgName() string {
	mset.cfgMu.RLock()
	defer mset.cfgMu.RUnlock()
	return mset.cfg.Name
}

// Purge will remove all messages from the stream and underlying store based on the request.
func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err error) {
	return mset.purgeLocked(preq, true)
}

func (mset *stream) purgeLocked(preq *JSApiStreamPurgeRequest, needLock bool) (purged uint64, err error) {
	if needLock {
		mset.mu.Lock()
		defer mset.mu.Unlock()
	}
	if mset.closed.Load() {
		return 0, errStreamClosed
	}
	if mset.cfg.Sealed {
		return 0, errors.New("sealed stream")
	}
	store, mlseq := mset.store, mset.lseq

	if preq != nil {
		purged, err = mset.store.PurgeEx(preq.Subject, preq.Sequence, preq.Keep)
	} else {
		purged, err = mset.store.Purge()
	}
	if err != nil {
		return purged, err
	}

	// Grab our stream state.
	var state StreamState
	store.FastState(&state)
	fseq, lseq := state.FirstSeq, state.LastSeq

	// Check if our last has moved past what our original last sequence was, if so reset.
	if lseq > mlseq {
		mset.setLastSeq(lseq)
	}

	// Clear any pending acks below first seq.
	mset.clearAllPreAcksBelowFloor(fseq)

	// Purge consumers.
	// Check for filtered purge.
	if preq != nil && preq.Subject != _EMPTY_ {
		ss := store.FilteredState(fseq, preq.Subject)
		fseq = ss.First
	}

	// Take a copy of cList to avoid o.purge() potentially taking the stream lock and
	// violating the lock ordering.
	mset.clsMu.RLock()
	cList := slices.Clone(mset.cList)
	mset.clsMu.RUnlock()
	for _, o := range cList {
		start := fseq
		o.mu.RLock()
		// we update consumer sequences if:
		// no subject was specified, we can purge all consumers sequences
		doPurge := preq == nil ||
			preq.Subject == _EMPTY_ ||
			// consumer filter subject is equal to purged subject
			// or consumer filter subject is subset of purged subject,
			// but not the other way around.
			o.isEqualOrSubsetMatch(preq.Subject)
		// Check if a consumer has a wider subject space than what we purged
		var isWider bool
		if !doPurge && preq != nil && o.isFilteredMatch(preq.Subject) {
			doPurge, isWider = true, true
			start = state.FirstSeq
		}
		o.mu.RUnlock()
		if doPurge {
			o.purge(start, lseq, isWider)
		}
	}

	return purged, nil
}

// RemoveMsg will remove a message from a stream.
// FIXME(dlc) - Should pick one and be consistent.
func (mset *stream) removeMsg(seq uint64) (bool, error) {
	return mset.deleteMsg(seq)
}

// DeleteMsg will remove a message from a stream.
func (mset *stream) deleteMsg(seq uint64) (bool, error) {
	if mset.closed.Load() {
		return false, errStreamClosed
	}
	removed, err := mset.store.RemoveMsg(seq)
	if err != nil {
		return removed, err
	}
	mset.mu.Lock()
	mset.clearAllPreAcks(seq)
	mset.mu.Unlock()
	return removed, err
}

// EraseMsg will securely remove a message and rewrite the data with random data.
func (mset *stream) eraseMsg(seq uint64) (bool, error) {
	if mset.closed.Load() {
		return false, errStreamClosed
	}
	removed, err := mset.store.EraseMsg(seq)
	if err != nil {
		return removed, err
	}
	mset.mu.Lock()
	mset.clearAllPreAcks(seq)
	mset.mu.Unlock()
	return removed, err
}

// Are we a mirror?
func (mset *stream) isMirror() bool {
	mset.cfgMu.RLock()
	defer mset.cfgMu.RUnlock()
	return mset.cfg.Mirror != nil
}

func (mset *stream) sourcesInfo() (sis []*StreamSourceInfo) {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	sis = make([]*StreamSourceInfo, 0, len(mset.sources))
	for _, si := range mset.sources {
		sis = append(sis, mset.sourceInfo(si))
	}
	return sis
}

// Lock should be held
func (mset *stream) sourceInfo(si *sourceInfo) *StreamSourceInfo {
	if si == nil {
		return nil
	}

	var ssi = StreamSourceInfo{Name: si.name, Lag: si.lag, Error: si.err, FilterSubject: si.sf}

	trConfigs := make([]SubjectTransformConfig, len(si.sfs))
	for i := range si.sfs {
		var destination string
		if si.trs[i] != nil {
			destination = si.trs[i].dest
		}
		trConfigs[i] = SubjectTransformConfig{si.sfs[i], destination}
	}

	ssi.SubjectTransforms = trConfigs

	// If we have not heard from the source, set Active to -1.
	if last := si.last.Load(); last == 0 {
		ssi.Active = -1
	} else {
		ssi.Active = time.Since(time.Unix(0, last))
	}

	var ext *ExternalStream
	if mset.cfg.Mirror != nil {
		ext = mset.cfg.Mirror.External
	} else if ss := mset.streamSource(si.iname); ss != nil && ss.External != nil {
		ext = ss.External
	}
	if ext != nil {
		ssi.External = &ExternalStream{
			ApiPrefix:     ext.ApiPrefix,
			DeliverPrefix: ext.DeliverPrefix,
		}
	}
	return &ssi
}

// Return our source info for our mirror.
func (mset *stream) mirrorInfo() *StreamSourceInfo {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.sourceInfo(mset.mirror)
}

// retryDisconnectedSyncConsumers() will check if we have any disconnected
// sync consumers for either mirror or a source and will reset and retry to connect.
func (mset *stream) retryDisconnectedSyncConsumers() {
	mset.mu.Lock()
	defer mset.mu.Unlock()

	// Only applicable if we are the stream leader.
	if !mset.isLeader() {
		return
	}

	shouldRetry := func(si *sourceInfo) bool {
		if si != nil && (si.sip || si.sub == nil || (si.sub.client != nil && si.sub.client.isClosed())) {
			// Need to reset
			si.fails, si.sip = 0, false
			mset.cancelSourceInfo(si)
			return true
		}
		return false
	}

	// Check mirrors first.
	if si := mset.mirror; si != nil {
		if shouldRetry(si) {
			mset.scheduleSetupMirrorConsumerRetry()
		}
	} else {
		for _, si := range mset.sources {
			if shouldRetry(si) {
				mset.setupSourceConsumer(si.iname, si.sseq+1, time.Time{})
			}
		}
	}
}

const (
	// Our consumer HB interval.
	sourceHealthHB = 1 * time.Second
	// How often we check and our stalled interval.
	sourceHealthCheckInterval = 10 * time.Second
)

// Will run as a Go routine to process mirror consumer messages.
func (mset *stream) processMirrorMsgs(mirror *sourceInfo, ready *sync.WaitGroup) {
	s := mset.srv
	defer func() {
		mirror.wg.Done()
		s.grWG.Done()
	}()

	// Grab stream quit channel.
	mset.mu.Lock()
	msgs, qch, siqch := mirror.msgs, mset.qch, mirror.qch
	// Set the last seen as now so that we don't fail at the first check.
	mirror.last.Store(time.Now().UnixNano())
	mset.mu.Unlock()

	// Signal the caller that we have captured the above fields.
	ready.Done()

	// Make sure we have valid ipq for msgs.
	if msgs == nil {
		mset.mu.Lock()
		mset.cancelMirrorConsumer()
		mset.mu.Unlock()
		return
	}

	t := time.NewTicker(sourceHealthCheckInterval)
	defer t.Stop()

	for {
		select {
		case <-s.quitCh:
			return
		case <-qch:
			return
		case <-siqch:
			return
		case <-msgs.ch:
			ims := msgs.pop()
			for _, im := range ims {
				if !mset.processInboundMirrorMsg(im) {
					break
				}
				im.returnToPool()
			}
			msgs.recycle(&ims)
		case <-t.C:
			mset.mu.RLock()
			var stalled bool
			if mset.mirror != nil {
				stalled = time.Since(time.Unix(0, mset.mirror.last.Load())) > sourceHealthCheckInterval
			}
			isLeader := mset.isLeader()
			mset.mu.RUnlock()
			// No longer leader.
			if !isLeader {
				mset.mu.Lock()
				mset.cancelMirrorConsumer()
				mset.mu.Unlock()
				return
			}
			// We are stalled.
			if stalled {
				mset.retryMirrorConsumer()
			}
		}
	}
}

// Checks that the message is from our current direct consumer. We can not depend on sub comparison
// since cross account imports break.
func (si *sourceInfo) isCurrentSub(reply string) bool {
	return si.cname != _EMPTY_ && strings.HasPrefix(reply, jsAckPre) && si.cname == tokenAt(reply, 4)
}

// processInboundMirrorMsg handles processing messages bound for a stream.
func (mset *stream) processInboundMirrorMsg(m *inMsg) bool {
	mset.mu.Lock()
	if mset.mirror == nil {
		mset.mu.Unlock()
		return false
	}
	if !mset.isLeader() {
		mset.cancelMirrorConsumer()
		mset.mu.Unlock()
		return false
	}

	isControl := m.isControlMsg()

	// Ignore from old subscriptions.
	// The reason we can not just compare subs is that on cross account imports they will not match.
	if !mset.mirror.isCurrentSub(m.rply) && !isControl {
		mset.mu.Unlock()
		return false
	}

	// Check for heartbeats and flow control messages.
	if isControl {
		var needsRetry bool
		// Flow controls have reply subjects.
		if m.rply != _EMPTY_ {
			mset.handleFlowControl(m)
		} else {
			// For idle heartbeats make sure we did not miss anything and check if we are considered stalled.
			if ldseq := parseInt64(getHeader(JSLastConsumerSeq, m.hdr)); ldseq > 0 && uint64(ldseq) != mset.mirror.dseq {
				needsRetry = true
			} else if fcReply := getHeader(JSConsumerStalled, m.hdr); len(fcReply) > 0 {
				// Other side thinks we are stalled, so send flow control reply.
				mset.outq.sendMsg(string(fcReply), nil)
			}
		}
		mset.mu.Unlock()
		if needsRetry {
			mset.retryMirrorConsumer()
		}
		return !needsRetry
	}

	sseq, dseq, dc, ts, pending := replyInfo(m.rply)

	if dc > 1 {
		mset.mu.Unlock()
		return false
	}

	// Mirror info tracking.
	olag, osseq, odseq := mset.mirror.lag, mset.mirror.sseq, mset.mirror.dseq
	if sseq == mset.mirror.sseq+1 {
		mset.mirror.dseq = dseq
		mset.mirror.sseq++
	} else if sseq <= mset.mirror.sseq {
		// Ignore older messages.
		mset.mu.Unlock()
		return true
	} else if mset.mirror.cname == _EMPTY_ {
		mset.mirror.cname = tokenAt(m.rply, 4)
		mset.mirror.dseq, mset.mirror.sseq = dseq, sseq
	} else {
		// If the deliver sequence matches then the upstream stream has expired or deleted messages.
		if dseq == mset.mirror.dseq+1 {
			mset.skipMsgs(mset.mirror.sseq+1, sseq-1)
			mset.mirror.dseq++
			mset.mirror.sseq = sseq
		} else {
			mset.mu.Unlock()
			mset.retryMirrorConsumer()
			return false
		}
	}

	if pending == 0 {
		mset.mirror.lag = 0
	} else {
		mset.mirror.lag = pending - 1
	}

	// Check if we allow mirror direct here. If so check they we have mostly caught up.
	// The reason we do not require 0 is if the source is active we may always be slightly behind.
	if mset.cfg.MirrorDirect && mset.mirrorDirectSub == nil && pending < dgetCaughtUpThresh {
		if err := mset.subscribeToMirrorDirect(); err != nil {
			// Disable since we had problems above.
			mset.cfg.MirrorDirect = false
		}
	}

	// Do the subject transform if there's one
	if len(mset.mirror.trs) > 0 {
		for _, tr := range mset.mirror.trs {
			if tr == nil {
				continue
			} else {
				tsubj, err := tr.Match(m.subj)
				if err == nil {
					m.subj = tsubj
					break
				}
			}
		}
	}

	s, js, stype := mset.srv, mset.js, mset.cfg.Storage
	node := mset.node
	mset.mu.Unlock()

	var err error
	if node != nil {
		if js.limitsExceeded(stype) {
			s.resourcesExceededError(stype)
			err = ApiErrors[JSInsufficientResourcesErr]
		} else {
			err = node.Propose(encodeStreamMsg(m.subj, _EMPTY_, m.hdr, m.msg, sseq-1, ts, true))
		}
	} else {
		err = mset.processJetStreamMsg(m.subj, _EMPTY_, m.hdr, m.msg, sseq-1, ts, nil, true, true)
	}
	if err != nil {
		if strings.Contains(err.Error(), "no space left") {
			s.Errorf("JetStream out of space, will be DISABLED")
			s.DisableJetStream()
			return false
		}
		if err != errLastSeqMismatch {
			mset.mu.RLock()
			accName, sname := mset.acc.Name, mset.cfg.Name
			mset.mu.RUnlock()
			s.RateLimitWarnf("Error processing inbound mirror message for '%s' > '%s': %v",
				accName, sname, err)
		} else {
			// We may have missed messages, restart.
			if sseq <= mset.lastSeq() {
				mset.mu.Lock()
				mset.mirror.lag = olag
				mset.mirror.sseq = osseq
				mset.mirror.dseq = odseq
				mset.mu.Unlock()
				return false
			} else {
				mset.mu.Lock()
				mset.mirror.dseq = odseq
				mset.mirror.sseq = osseq
				mset.mu.Unlock()
				mset.retryMirrorConsumer()
			}
		}
	}
	return err == nil
}

func (mset *stream) setMirrorErr(err *ApiError) {
	mset.mu.Lock()
	if mset.mirror != nil {
		mset.mirror.err = err
	}
	mset.mu.Unlock()
}

// Cancels a mirror consumer.
//
// Lock held on entry
func (mset *stream) cancelMirrorConsumer() {
	if mset.mirror == nil {
		return
	}
	mset.cancelSourceInfo(mset.mirror)
}

// Similar to setupMirrorConsumer except that it will print a debug statement
// indicating that there is a retry.
//
// Lock is acquired in this function
func (mset *stream) retryMirrorConsumer() error {
	mset.mu.Lock()
	defer mset.mu.Unlock()
	mset.srv.Debugf("Retrying mirror consumer for '%s > %s'", mset.acc.Name, mset.cfg.Name)
	mset.cancelMirrorConsumer()
	return mset.setupMirrorConsumer()
}

// Lock should be held.
func (mset *stream) skipMsgs(start, end uint64) {
	node, store := mset.node, mset.store
	// If we are not clustered we can short circuit now with store.SkipMsgs
	if node == nil {
		store.SkipMsgs(start, end-start+1)
		mset.lseq = end
		return
	}

	// FIXME (dlc) - We should allow proposals of DeleteRange, but would need to make sure all peers support.
	// With syncRequest was easy to add bool into request.
	var entries []*Entry
	for seq := start; seq <= end; seq++ {
		entries = append(entries, newEntry(EntryNormal, encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq-1, 0, false)))
		// So a single message does not get too big.
		if len(entries) > 10_000 {
			node.ProposeMulti(entries)
			// We need to re-create `entries` because there is a reference
			// to it in the node's pae map.
			entries = entries[:0]
		}
	}
	// Send all at once.
	if len(entries) > 0 {
		node.ProposeMulti(entries)
	}
}

const (
	// Base retry backoff duration.
	retryBackOff = 5 * time.Second
	// Maximum amount we will wait.
	retryMaximum = 2 * time.Minute
)

// Calculate our backoff based on number of failures.
func calculateRetryBackoff(fails int) time.Duration {
	backoff := time.Duration(retryBackOff) * time.Duration(fails*2)
	if backoff > retryMaximum {
		backoff = retryMaximum
	}
	return backoff
}

// This will schedule a call to setupMirrorConsumer, taking into account the last
// time it was retried and determine the soonest setupMirrorConsumer can be called
// without tripping the sourceConsumerRetryThreshold. We will also take into account
// number of failures and will back off our retries.
// The mset.mirror pointer has been verified to be not nil by the caller.
//
// Lock held on entry
func (mset *stream) scheduleSetupMirrorConsumerRetry() {
	// We are trying to figure out how soon we can retry. setupMirrorConsumer will reject
	// a retry if last was done less than "sourceConsumerRetryThreshold" ago.
	next := sourceConsumerRetryThreshold - time.Since(mset.mirror.lreq)
	if next < 0 {
		// It means that we have passed the threshold and so we are ready to go.
		next = 0
	}
	// Take into account failures here.
	next += calculateRetryBackoff(mset.mirror.fails)

	// Add some jitter.
	next += time.Duration(rand.Intn(int(100*time.Millisecond))) + 100*time.Millisecond

	time.AfterFunc(next, func() {
		mset.mu.Lock()
		mset.setupMirrorConsumer()
		mset.mu.Unlock()
	})
}

// How long we wait for a response from a consumer create request for a source or mirror.
var srcConsumerWaitTime = 30 * time.Second

// Setup our mirror consumer.
// Lock should be held.
func (mset *stream) setupMirrorConsumer() error {
	if mset.closed.Load() {
		return errStreamClosed
	}
	if mset.outq == nil {
		return errors.New("outq required")
	}
	// We use to prevent update of a mirror configuration in cluster
	// mode but not in standalone. This is now fixed. However, without
	// rejecting the update, it could be that if the source stream was
	// removed and then later the mirrored stream config changed to
	// remove mirror configuration, this function would panic when
	// accessing mset.cfg.Mirror fields. Adding this protection in case
	// we allow in the future the mirror config to be changed (removed).
	if mset.cfg.Mirror == nil {
		return errors.New("invalid mirror configuration")
	}

	// If this is the first time
	if mset.mirror == nil {
		mset.mirror = &sourceInfo{name: mset.cfg.Mirror.Name}
	} else {
		mset.cancelSourceInfo(mset.mirror)
		mset.mirror.sseq = mset.lseq
	}

	// If we are no longer the leader stop trying.
	if !mset.isLeader() {
		return nil
	}

	mirror := mset.mirror

	// We want to throttle here in terms of how fast we request new consumers,
	// or if the previous is still in progress.
	if last := time.Since(mirror.lreq); last < sourceConsumerRetryThreshold || mirror.sip {
		mset.scheduleSetupMirrorConsumerRetry()
		return nil
	}
	mirror.lreq = time.Now()

	// Determine subjects etc.
	var deliverSubject string
	ext := mset.cfg.Mirror.External

	if ext != nil && ext.DeliverPrefix != _EMPTY_ {
		deliverSubject = strings.ReplaceAll(ext.DeliverPrefix+syncSubject(".M"), "..", ".")
	} else {
		deliverSubject = syncSubject("$JS.M")
	}

	// Now send off request to create/update our consumer. This will be all API based even in single server mode.
	// We calculate durable names apriori so we do not need to save them off.

	var state StreamState
	mset.store.FastState(&state)

	req := &CreateConsumerRequest{
		Stream: mset.cfg.Mirror.Name,
		Config: ConsumerConfig{
			DeliverSubject:    deliverSubject,
			DeliverPolicy:     DeliverByStartSequence,
			OptStartSeq:       state.LastSeq + 1,
			AckPolicy:         AckNone,
			AckWait:           22 * time.Hour,
			MaxDeliver:        1,
			Heartbeat:         sourceHealthHB,
			FlowControl:       true,
			Direct:            true,
			InactiveThreshold: sourceHealthCheckInterval,
		},
	}

	// Only use start optionals on first time.
	if state.Msgs == 0 && state.FirstSeq == 0 {
		req.Config.OptStartSeq = 0
		if mset.cfg.Mirror.OptStartSeq > 0 {
			req.Config.OptStartSeq = mset.cfg.Mirror.OptStartSeq
		} else if mset.cfg.Mirror.OptStartTime != nil {
			req.Config.OptStartTime = mset.cfg.Mirror.OptStartTime
			req.Config.DeliverPolicy = DeliverByStartTime
		}
	}
	if req.Config.OptStartSeq == 0 && req.Config.OptStartTime == nil {
		// If starting out and lastSeq is 0.
		req.Config.DeliverPolicy = DeliverAll
	}

	// Filters
	if mset.cfg.Mirror.FilterSubject != _EMPTY_ {
		req.Config.FilterSubject = mset.cfg.Mirror.FilterSubject
		mirror.sf = mset.cfg.Mirror.FilterSubject
	}

	if lst := len(mset.cfg.Mirror.SubjectTransforms); lst > 0 {
		sfs := make([]string, lst)
		trs := make([]*subjectTransform, lst)

		for i, tr := range mset.cfg.Mirror.SubjectTransforms {
			// will not fail as already checked before that the transform will work
			subjectTransform, err := NewSubjectTransform(tr.Source, tr.Destination)
			if err != nil {
				mset.srv.Errorf("Unable to get transform for mirror consumer: %v", err)
			}
			sfs[i] = tr.Source
			trs[i] = subjectTransform
		}
		mirror.sfs = sfs
		mirror.trs = trs
		// If there was no explicit FilterSubject defined and we have a single
		// subject transform, use Config.FilterSubject instead of FilterSubjects
		// so that we can use the extended consumer create API down below.
		if req.Config.FilterSubject == _EMPTY_ && len(sfs) == 1 {
			req.Config.FilterSubject = sfs[0]
		} else {
			req.Config.FilterSubjects = sfs
		}
	}

	respCh := make(chan *JSApiConsumerCreateResponse, 1)
	reply := infoReplySubject()
	crSub, err := mset.subscribeInternal(reply, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
		mset.unsubscribe(sub)
		_, msg := c.msgParts(rmsg)

		var ccr JSApiConsumerCreateResponse
		if err := json.Unmarshal(msg, &ccr); err != nil {
			c.Warnf("JetStream bad mirror consumer create response: %q", msg)
			mset.setMirrorErr(ApiErrors[JSInvalidJSONErr])
			return
		}
		select {
		case respCh <- &ccr:
		default:
		}
	})
	if err != nil {
		mirror.err = NewJSMirrorConsumerSetupFailedError(err, Unless(err))
		mset.scheduleSetupMirrorConsumerRetry()
		return nil
	}

	var subject string
	if req.Config.FilterSubject != _EMPTY_ {
		req.Config.Name = fmt.Sprintf("mirror-%s", createConsumerName())
		subject = fmt.Sprintf(JSApiConsumerCreateExT, mset.cfg.Mirror.Name, req.Config.Name, req.Config.FilterSubject)
	} else {
		subject = fmt.Sprintf(JSApiConsumerCreateT, mset.cfg.Mirror.Name)
	}
	if ext != nil {
		subject = strings.Replace(subject, JSApiPrefix, ext.ApiPrefix, 1)
		subject = strings.ReplaceAll(subject, "..", ".")
	}

	// Marshal now that we are done with `req`.
	b, _ := json.Marshal(req)

	// Reset
	mirror.msgs = nil
	mirror.err = nil
	mirror.sip = true

	// Send the consumer create request
	mset.outq.send(newJSPubMsg(subject, _EMPTY_, reply, nil, b, nil, 0))

	go func() {

		var retry bool
		defer func() {
			mset.mu.Lock()
			// Check that this is still valid and if so, clear the "setup in progress" flag.
			if mset.mirror != nil {
				mset.mirror.sip = false
				// If we need to retry, schedule now
				// If sub is not nil means we re-established somewhere else so do not re-attempt here.
				if retry && mset.mirror.sub == nil {
					mset.mirror.fails++
					// Cancel here since we can not do anything with this consumer at this point.
					mset.cancelSourceInfo(mset.mirror)
					mset.scheduleSetupMirrorConsumerRetry()
				} else {
					// Clear on success.
					mset.mirror.fails = 0
				}
			}
			mset.mu.Unlock()
		}()

		// Wait for previous processMirrorMsgs go routine to be completely done.
		// If none is running, this will not block.
		mirror.wg.Wait()

		select {
		case ccr := <-respCh:
			mset.mu.Lock()
			// Mirror config has been removed.
			if mset.mirror == nil {
				mset.mu.Unlock()
				return
			}
			ready := sync.WaitGroup{}
			mirror := mset.mirror
			mirror.err = nil
			if ccr.Error != nil || ccr.ConsumerInfo == nil {
				mset.srv.Warnf("JetStream error response for create mirror consumer: %+v", ccr.Error)
				mirror.err = ccr.Error
				// Let's retry as soon as possible, but we are gated by sourceConsumerRetryThreshold
				retry = true
				mset.mu.Unlock()
				return
			} else {
				// Setup actual subscription to process messages from our source.
				qname := fmt.Sprintf("[ACC:%s] stream mirror '%s' of '%s' msgs", mset.acc.Name, mset.cfg.Name, mset.cfg.Mirror.Name)
				// Create a new queue each time
				mirror.msgs = newIPQueue[*inMsg](mset.srv, qname)
				msgs := mirror.msgs
				sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
					hdr, msg := c.msgParts(copyBytes(rmsg)) // Need to copy.
					if len(hdr) > 0 {
						// Remove any Nats-Expected- headers as we don't want to validate them.
						hdr = removeHeaderIfPrefixPresent(hdr, "Nats-Expected-")
						// Remove any Nats-Batch- headers, batching is not supported when mirroring.
						hdr = removeHeaderIfPrefixPresent(hdr, "Nats-Batch-")
					}
					mset.queueInbound(msgs, subject, reply, hdr, msg, nil, nil)
					mirror.last.Store(time.Now().UnixNano())
				})
				if err != nil {
					mirror.err = NewJSMirrorConsumerSetupFailedError(err, Unless(err))
					retry = true
					mset.mu.Unlock()
					return
				}
				// Save our sub.
				mirror.sub = sub

				// When an upstream stream expires messages or in general has messages that we want
				// that are no longer available we need to adjust here.
				var state StreamState
				mset.store.FastState(&state)

				// Check if we need to skip messages.
				if state.LastSeq != ccr.ConsumerInfo.Delivered.Stream {
					// Check to see if delivered is past our last and we have no msgs. This will help the
					// case when mirroring a stream that has a very high starting sequence number.
					if state.Msgs == 0 && ccr.ConsumerInfo.Delivered.Stream > state.LastSeq {
						mset.store.PurgeEx(_EMPTY_, ccr.ConsumerInfo.Delivered.Stream+1, 0)
						mset.lseq = ccr.ConsumerInfo.Delivered.Stream
					} else {
						mset.skipMsgs(state.LastSeq+1, ccr.ConsumerInfo.Delivered.Stream)
					}
				}

				// Capture consumer name.
				mirror.cname = ccr.ConsumerInfo.Name
				mirror.dseq = 0
				mirror.sseq = ccr.ConsumerInfo.Delivered.Stream
				mirror.qch = make(chan struct{})
				mirror.wg.Add(1)
				ready.Add(1)
				if !mset.srv.startGoRoutine(
					func() { mset.processMirrorMsgs(mirror, &ready) },
					pprofLabels{
						"type":     "mirror",
						"account":  mset.acc.Name,
						"stream":   mset.cfg.Name,
						"consumer": mirror.cname,
					},
				) {
					ready.Done()
				}
			}
			mset.mu.Unlock()
			ready.Wait()
		case <-time.After(srcConsumerWaitTime):
			mset.unsubscribe(crSub)
			// We already waited 30 seconds, let's retry now.
			retry = true
		}
	}()

	return nil
}

func (mset *stream) streamSource(iname string) *StreamSource {
	for _, ssi := range mset.cfg.Sources {
		if ssi.iname == iname {
			return ssi
		}
	}
	return nil
}

// Lock should be held.
func (mset *stream) retrySourceConsumerAtSeq(iName string, seq uint64) {
	s := mset.srv

	s.Debugf("Retrying source consumer for '%s > %s'", mset.acc.Name, mset.cfg.Name)

	// setupSourceConsumer will check that the source is still configured.
	mset.setupSourceConsumer(iName, seq, time.Time{})
}

// Lock should be held.
func (mset *stream) cancelSourceConsumer(iname string) {
	if si := mset.sources[iname]; si != nil {
		mset.cancelSourceInfo(si)
		si.sseq, si.dseq = 0, 0
	}
}

// The `si` has been verified to be not nil. The sourceInfo's sub will
// be unsubscribed and set to nil (if not already done) and the
// cname will be reset. The message processing's go routine quit channel
// will be closed if still opened.
//
// Lock should be held
func (mset *stream) cancelSourceInfo(si *sourceInfo) {
	if si.sub != nil {
		mset.unsubscribe(si.sub)
		si.sub = nil
	}
	mset.removeInternalConsumer(si)
	if si.qch != nil {
		close(si.qch)
		si.qch = nil
	}
	if si.msgs != nil {
		si.msgs.drain()
		si.msgs.unregister()
	}
	// If we have a schedule setup go ahead and delete that.
	if t := mset.sourceSetupSchedules[si.iname]; t != nil {
		t.Stop()
		delete(mset.sourceSetupSchedules, si.iname)
	}
}

const sourceConsumerRetryThreshold = 2 * time.Second

// This is the main function to call when needing to setup a new consumer for the source.
// It actually only does the scheduling of the execution of trySetupSourceConsumer in order to implement retry backoff
// and throttle the number of requests.
// Lock should be held.
func (mset *stream) setupSourceConsumer(iname string, seq uint64, startTime time.Time) {
	if mset.sourceSetupSchedules == nil {
		mset.sourceSetupSchedules = map[string]*time.Timer{}
	}

	if _, ok := mset.sourceSetupSchedules[iname]; ok {
		// If there is already a timer scheduled, we don't need to do anything.
		return
	}

	si := mset.sources[iname]
	if si == nil || si.sip { // if sourceInfo was removed or setup is in progress, nothing to do
		return
	}

	// First calculate the delay until the next time we can
	var scheduleDelay time.Duration

	if !si.lreq.IsZero() { // it's not the very first time we are called, compute the delay
		// We want to throttle here in terms of how fast we request new consumers
		if sinceLast := time.Since(si.lreq); sinceLast < sourceConsumerRetryThreshold {
			scheduleDelay = sourceConsumerRetryThreshold - sinceLast
		}
		// Is it a retry? If so, add a backoff
		if si.fails > 0 {
			scheduleDelay += calculateRetryBackoff(si.fails)
		}
	}

	// Always add some jitter
	scheduleDelay += time.Duration(rand.Intn(int(100*time.Millisecond))) + 100*time.Millisecond

	// Schedule the call to trySetupSourceConsumer
	mset.sourceSetupSchedules[iname] = time.AfterFunc(scheduleDelay, func() {
		mset.mu.Lock()
		defer mset.mu.Unlock()

		delete(mset.sourceSetupSchedules, iname)
		mset.trySetupSourceConsumer(iname, seq, startTime)
	})
}

// This is where we will actually try to create a new consumer for the source
// Lock should be held.
func (mset *stream) trySetupSourceConsumer(iname string, seq uint64, startTime time.Time) {
	// Ignore if closed or not leader.
	if mset.closed.Load() || !mset.isLeader() {
		return
	}

	si := mset.sources[iname]
	if si == nil {
		return
	}

	// Cancel previous instance if applicable
	mset.cancelSourceInfo(si)

	ssi := mset.streamSource(iname)
	if ssi == nil {
		return
	}

	si.lreq = time.Now()

	// Determine subjects etc.
	var deliverSubject string
	ext := ssi.External

	if ext != nil && ext.DeliverPrefix != _EMPTY_ {
		deliverSubject = strings.ReplaceAll(ext.DeliverPrefix+syncSubject(".S"), "..", ".")
	} else {
		deliverSubject = syncSubject("$JS.S")
	}

	req := &CreateConsumerRequest{
		Stream: si.name,
		Config: ConsumerConfig{
			DeliverSubject:    deliverSubject,
			AckPolicy:         AckNone,
			AckWait:           22 * time.Hour,
			MaxDeliver:        1,
			Heartbeat:         sourceHealthHB,
			FlowControl:       true,
			Direct:            true,
			InactiveThreshold: sourceHealthCheckInterval,
		},
	}

	// If starting, check any configs.
	if !startTime.IsZero() && seq > 1 {
		req.Config.OptStartTime = &startTime
		req.Config.DeliverPolicy = DeliverByStartTime
	} else if seq <= 1 {
		if ssi.OptStartSeq > 0 {
			req.Config.OptStartSeq = ssi.OptStartSeq
			req.Config.DeliverPolicy = DeliverByStartSequence
		} else {
			// We have not recovered state so check that configured time is less that our first seq time.
			var state StreamState
			mset.store.FastState(&state)
			if ssi.OptStartTime != nil {
				if !state.LastTime.IsZero() && ssi.OptStartTime.Before(state.LastTime) {
					req.Config.OptStartTime = &state.LastTime
				} else {
					req.Config.OptStartTime = ssi.OptStartTime
				}
				req.Config.DeliverPolicy = DeliverByStartTime
			} else if state.FirstSeq > 1 && !state.LastTime.IsZero() {
				req.Config.OptStartTime = &state.LastTime
				req.Config.DeliverPolicy = DeliverByStartTime
			}
		}

	} else {
		req.Config.OptStartSeq = seq
		req.Config.DeliverPolicy = DeliverByStartSequence
	}
	// Filters
	if ssi.FilterSubject != _EMPTY_ {
		req.Config.FilterSubject = ssi.FilterSubject
	}

	var filterSubjects []string
	for _, tr := range ssi.SubjectTransforms {
		filterSubjects = append(filterSubjects, tr.Source)
	}
	req.Config.FilterSubjects = filterSubjects

	respCh := make(chan *JSApiConsumerCreateResponse, 1)
	reply := infoReplySubject()
	crSub, err := mset.subscribeInternal(reply, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
		mset.unsubscribe(sub)
		_, msg := c.msgParts(rmsg)
		var ccr JSApiConsumerCreateResponse
		if err := json.Unmarshal(msg, &ccr); err != nil {
			c.Warnf("JetStream bad source consumer create response: %q", msg)
			return
		}
		select {
		case respCh <- &ccr:
		default:
		}
	})
	if err != nil {
		si.err = NewJSSourceConsumerSetupFailedError(err, Unless(err))
		mset.setupSourceConsumer(iname, seq, startTime)
		return
	}

	var subject string
	if req.Config.FilterSubject != _EMPTY_ {
		req.Config.Name = fmt.Sprintf("src-%s", createConsumerName())
		subject = fmt.Sprintf(JSApiConsumerCreateExT, si.name, req.Config.Name, req.Config.FilterSubject)
	} else if len(req.Config.FilterSubjects) == 1 {
		req.Config.Name = fmt.Sprintf("src-%s", createConsumerName())
		// It is necessary to switch to using FilterSubject here as the extended consumer
		// create API checks for it, so as to not accidentally allow multiple filtered subjects.
		req.Config.FilterSubject = req.Config.FilterSubjects[0]
		req.Config.FilterSubjects = nil
		subject = fmt.Sprintf(JSApiConsumerCreateExT, si.name, req.Config.Name, req.Config.FilterSubject)
	} else {
		subject = fmt.Sprintf(JSApiConsumerCreateT, si.name)
	}
	if ext != nil {
		subject = strings.Replace(subject, JSApiPrefix, ext.ApiPrefix, 1)
		subject = strings.ReplaceAll(subject, "..", ".")
	}

	// Marshal request.
	b, _ := json.Marshal(req)

	// Reset
	si.msgs = nil
	si.err = nil
	si.sip = true

	// Send the consumer create request
	mset.outq.send(newJSPubMsg(subject, _EMPTY_, reply, nil, b, nil, 0))

	go func() {

		var retry bool
		defer func() {
			mset.mu.Lock()
			// Check that this is still valid and if so, clear the "setup in progress" flag.
			if si := mset.sources[iname]; si != nil {
				si.sip = false
				// If we need to retry, schedule now
				// If sub is not nil means we re-established somewhere else so do not re-attempt here.
				if retry && si.sub == nil {
					si.fails++
					// Cancel here since we can not do anything with this consumer at this point.
					mset.cancelSourceInfo(si)
					mset.setupSourceConsumer(iname, seq, startTime)
				} else {
					// Clear on success.
					si.fails = 0
				}
			}
			mset.mu.Unlock()
		}()

		select {
		case ccr := <-respCh:
			mset.mu.Lock()
			// Check that it has not been removed or canceled (si.sub would be nil)
			if si := mset.sources[iname]; si != nil {
				si.err = nil
				if ccr.Error != nil || ccr.ConsumerInfo == nil {
					// Note: this warning can happen a few times when starting up the server when sourcing streams are
					// defined, this is normal as the streams are re-created in no particular order and it is possible
					// that a stream sourcing another could come up before all of its sources have been recreated.
					mset.srv.Warnf("JetStream error response for stream %s create source consumer %s: %+v", mset.cfg.Name, si.name, ccr.Error)
					si.err = ccr.Error
					// Let's retry as soon as possible, but we are gated by sourceConsumerRetryThreshold
					retry = true
					mset.mu.Unlock()
					return
				} else {
					// Check if our shared msg queue and go routine is running or not.
					if mset.smsgs == nil {
						qname := fmt.Sprintf("[ACC:%s] stream sources '%s' msgs", mset.acc.Name, mset.cfg.Name)
						mset.smsgs = newIPQueue[*inMsg](mset.srv, qname)
						mset.srv.startGoRoutine(func() { mset.processAllSourceMsgs() },
							pprofLabels{
								"type":    "source",
								"account": mset.acc.Name,
								"stream":  mset.cfg.Name,
							},
						)
					}

					// Setup actual subscription to process messages from our source.
					if si.sseq != ccr.ConsumerInfo.Delivered.Stream {
						si.sseq = ccr.ConsumerInfo.Delivered.Stream + 1
					}
					// Capture consumer name.
					si.cname = ccr.ConsumerInfo.Name

					// Do not set si.sseq to seq here. si.sseq will be set in processInboundSourceMsg
					si.dseq = 0
					si.qch = make(chan struct{})
					// Set the last seen as now so that we don't fail at the first check.
					si.last.Store(time.Now().UnixNano())

					msgs := mset.smsgs
					sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
						hdr, msg := c.msgParts(copyBytes(rmsg)) // Need to copy.
						mset.queueInbound(msgs, subject, reply, hdr, msg, si, nil)
						si.last.Store(time.Now().UnixNano())
					})
					if err != nil {
						si.err = NewJSSourceConsumerSetupFailedError(err, Unless(err))
						retry = true
						mset.mu.Unlock()
						return
					}
					// Save our sub.
					si.sub = sub
				}
			}
			mset.mu.Unlock()
		case <-time.After(srcConsumerWaitTime):
			mset.unsubscribe(crSub)
			// We already waited 30 seconds, let's retry now.
			retry = true
		}
	}()
}

// This will process all inbound source msgs.
// We mux them into one go routine to avoid lock contention and high cpu and thread thrashing.
// TODO(dlc) make this more then one and pin sources to one of a group.
func (mset *stream) processAllSourceMsgs() {
	s := mset.srv
	defer s.grWG.Done()

	mset.mu.RLock()
	msgs, qch := mset.smsgs, mset.qch
	mset.mu.RUnlock()

	t := time.NewTicker(sourceHealthCheckInterval)
	defer t.Stop()

	// When we detect we are no longer leader, we will cleanup.
	// Should always return right after this is called.
	cleanUp := func() {
		mset.mu.Lock()
		defer mset.mu.Unlock()
		for _, si := range mset.sources {
			mset.cancelSourceConsumer(si.iname)
		}
		mset.smsgs.drain()
		mset.smsgs.unregister()
		mset.smsgs = nil
	}

	for {
		select {
		case <-s.quitCh:
			return
		case <-qch:
			return
		case <-msgs.ch:
			ims := msgs.pop()
			for _, im := range ims {
				if !mset.processInboundSourceMsg(im.si, im) {
					// If we are no longer leader bail.
					if !mset.IsLeader() {
						msgs.recycle(&ims)
						cleanUp()
						return
					}
					break
				}
				im.returnToPool()
			}
			msgs.recycle(&ims)
		case <-t.C:
			// If we are no longer leader bail.
			if !mset.IsLeader() {
				cleanUp()
				return
			}

			// Check health of all sources.
			var stalled []*sourceInfo
			mset.mu.RLock()
			for _, si := range mset.sources {
				if time.Since(time.Unix(0, si.last.Load())) > sourceHealthCheckInterval {
					stalled = append(stalled, si)
				}
			}
			numSources := len(mset.sources)
			mset.mu.RUnlock()

			// This can happen on an update when no longer have sources.
			if numSources == 0 {
				cleanUp()
				return
			}

			// We do not want to block here so do in separate Go routine.
			if len(stalled) > 0 {
				go func() {
					mset.mu.Lock()
					defer mset.mu.Unlock()
					for _, si := range stalled {
						mset.setupSourceConsumer(si.iname, si.sseq+1, time.Time{})
					}
				}()
			}
		}
	}
}

// isControlMsg determines if this is a control message.
func (m *inMsg) isControlMsg() bool {
	return len(m.msg) == 0 && len(m.hdr) > 0 && bytes.HasPrefix(m.hdr, []byte("NATS/1.0 100 "))
}

// Sends a reply to a flow control request.
// Lock should be held.
func (mset *stream) sendFlowControlReply(reply string) {
	if mset.isLeader() && mset.outq != nil {
		mset.outq.sendMsg(reply, nil)
	}
}

// handleFlowControl will properly handle flow control messages for both R==1 and R>1.
// Lock should be held.
func (mset *stream) handleFlowControl(m *inMsg) {
	// If we are clustered we will send the flow control message through the replication stack.
	if mset.isClustered() {
		mset.node.Propose(encodeStreamMsg(_EMPTY_, m.rply, m.hdr, nil, 0, 0, false))
	} else {
		mset.outq.sendMsg(m.rply, nil)
	}
}

// processInboundSourceMsg handles processing other stream messages bound for this stream.
func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
	mset.mu.Lock()
	// If we are no longer the leader cancel this subscriber.
	if !mset.isLeader() {
		mset.cancelSourceConsumer(si.iname)
		mset.mu.Unlock()
		return false
	}

	isControl := m.isControlMsg()

	// Ignore from old subscriptions.
	if !si.isCurrentSub(m.rply) && !isControl {
		mset.mu.Unlock()
		return false
	}

	// Check for heartbeats and flow control messages.
	if isControl {
		var needsRetry bool
		// Flow controls have reply subjects.
		if m.rply != _EMPTY_ {
			mset.handleFlowControl(m)
		} else {
			// For idle heartbeats make sure we did not miss anything.
			if ldseq := parseInt64(getHeader(JSLastConsumerSeq, m.hdr)); ldseq > 0 && uint64(ldseq) != si.dseq {
				needsRetry = true
				mset.retrySourceConsumerAtSeq(si.iname, si.sseq+1)
			} else if fcReply := getHeader(JSConsumerStalled, m.hdr); len(fcReply) > 0 {
				// Other side thinks we are stalled, so send flow control reply.
				mset.outq.sendMsg(string(fcReply), nil)
			}
		}
		mset.mu.Unlock()
		return !needsRetry
	}

	sseq, dseq, dc, _, pending := replyInfo(m.rply)

	if dc > 1 {
		mset.mu.Unlock()
		return false
	}

	// Tracking is done here.
	if dseq == si.dseq+1 {
		si.dseq++
		si.sseq = sseq
	} else if dseq > si.dseq {
		if si.cname == _EMPTY_ {
			si.cname = tokenAt(m.rply, 4)
			si.dseq, si.sseq = dseq, sseq
		} else {
			mset.retrySourceConsumerAtSeq(si.iname, si.sseq+1)
			mset.mu.Unlock()
			return false
		}
	} else {
		mset.mu.Unlock()
		return false
	}

	if pending == 0 {
		si.lag = 0
	} else {
		si.lag = pending - 1
	}
	node := mset.node
	mset.mu.Unlock()

	hdr, msg := m.hdr, m.msg

	// If we are daisy chained here make sure to remove the original one.
	if len(hdr) > 0 {
		hdr = removeHeaderIfPresent(hdr, JSStreamSource)
		// Remove any Nats-Expected- headers as we don't want to validate them.
		hdr = removeHeaderIfPrefixPresent(hdr, "Nats-Expected-")
		// Remove any Nats-Batch- headers, batching is not supported when sourcing.
		hdr = removeHeaderIfPrefixPresent(hdr, "Nats-Batch-")
	}
	// Hold onto the origin reply which has all the metadata.
	hdr = genHeader(hdr, JSStreamSource, si.genSourceHeader(m.subj, m.rply))

	// Do the subject transform for the source if there's one
	if len(si.trs) > 0 {
		for _, tr := range si.trs {
			if tr == nil {
				continue
			} else {
				tsubj, err := tr.Match(m.subj)
				if err == nil {
					m.subj = tsubj
					break
				}
			}
		}
	}

	var err error
	// If we are clustered we need to propose this message to the underlying raft group.
	if node != nil {
		err = mset.processClusteredInboundMsg(m.subj, _EMPTY_, hdr, msg, nil, true)
	} else {
		err = mset.processJetStreamMsg(m.subj, _EMPTY_, hdr, msg, 0, 0, nil, true, true)
	}

	if err != nil {
		s := mset.srv
		if strings.Contains(err.Error(), "no space left") {
			s.Errorf("JetStream out of space, will be DISABLED")
			s.DisableJetStream()
		} else {
			mset.mu.RLock()
			accName, sname, iName := mset.acc.Name, mset.cfg.Name, si.iname
			mset.mu.RUnlock()

			// Can happen temporarily all the time during normal operations when the sourcing stream
			// is working queue/interest with a limit and discard new.
			// TODO - Improve sourcing to WQ with limit and new to use flow control rather than re-creating the consumer.
			if errors.Is(err, ErrMaxMsgs) || errors.Is(err, ErrMaxBytes) {
				// Do not need to do a full retry that includes finding the last sequence in the stream
				// for that source. Just re-create starting with the seq we couldn't store instead.
				mset.mu.Lock()
				mset.retrySourceConsumerAtSeq(iName, si.sseq)
				mset.mu.Unlock()
			} else {
				// Log some warning for errors other than errLastSeqMismatch or errMaxMsgs.
				if !errors.Is(err, errLastSeqMismatch) {
					s.RateLimitWarnf("Error processing inbound source %q for '%s' > '%s': %v",
						iName, accName, sname, err)
				}
				// Retry in all type of errors if we are still leader.
				if mset.isLeader() {
					// This will make sure the source is still in mset.sources map,
					// find the last sequence and then call setupSourceConsumer.
					iNameMap := map[string]struct{}{iName: {}}
					mset.setStartingSequenceForSources(iNameMap)
					mset.mu.Lock()
					mset.retrySourceConsumerAtSeq(iName, si.sseq+1)
					mset.mu.Unlock()
				}
			}
		}
		return false
	}

	return true
}

// Generate a new (2.10) style source header (stream name, sequence number, source filter, source destination transform).
func (si *sourceInfo) genSourceHeader(orig, reply string) string {
	var b strings.Builder
	iNameParts := strings.Split(si.iname, " ")

	b.WriteString(iNameParts[0])
	b.WriteByte(' ')
	// Grab sequence as text here from reply subject.
	var tsa [expectedNumReplyTokens]string
	start, tokens := 0, tsa[:0]
	for i := 0; i < len(reply); i++ {
		if reply[i] == btsep {
			tokens, start = append(tokens, reply[start:i]), i+1
		}
	}
	tokens = append(tokens, reply[start:])
	seq := "1" // Default
	if len(tokens) == expectedNumReplyTokens && tokens[0] == "$JS" && tokens[1] == "ACK" {
		seq = tokens[5]
	}
	b.WriteString(seq)

	b.WriteByte(' ')
	b.WriteString(iNameParts[1])
	b.WriteByte(' ')
	b.WriteString(iNameParts[2])
	b.WriteByte(' ')
	b.WriteString(orig)
	return b.String()
}

// Original version of header that stored ack reply direct.
func streamAndSeqFromAckReply(reply string) (string, string, uint64) {
	tsa := [expectedNumReplyTokens]string{}
	start, tokens := 0, tsa[:0]
	for i := 0; i < len(reply); i++ {
		if reply[i] == btsep {
			tokens, start = append(tokens, reply[start:i]), i+1
		}
	}
	tokens = append(tokens, reply[start:])
	if len(tokens) != expectedNumReplyTokens || tokens[0] != "$JS" || tokens[1] != "ACK" {
		return _EMPTY_, _EMPTY_, 0
	}
	return tokens[2], _EMPTY_, uint64(parseAckReplyNum(tokens[5]))
}

// Extract the stream name, the source index name and the message sequence number from the source header.
// Uses the filter and transform arguments to provide backwards compatibility
func streamAndSeq(shdr string) (string, string, uint64) {
	if strings.HasPrefix(shdr, jsAckPre) {
		return streamAndSeqFromAckReply(shdr)
	}
	// New version which is stream index name <SPC> sequence
	fields := strings.Split(shdr, " ")
	nFields := len(fields)

	if nFields != 2 && nFields <= 3 {
		return _EMPTY_, _EMPTY_, 0
	}

	if nFields >= 4 {
		return fields[0], strings.Join([]string{fields[0], fields[2], fields[3]}, " "), uint64(parseAckReplyNum(fields[1]))
	} else {
		return fields[0], _EMPTY_, uint64(parseAckReplyNum(fields[1]))
	}

}

// Lock should be held.
func (mset *stream) setStartingSequenceForSources(iNames map[string]struct{}) {
	var state StreamState
	mset.store.FastState(&state)

	// Do not reset sseq here so we can remember when purge/expiration happens.
	if state.Msgs == 0 {
		for iName := range iNames {
			si := mset.sources[iName]
			if si == nil {
				continue
			} else {
				si.dseq = 0
			}
		}
		return
	}

	var smv StoreMsg
	for seq := state.LastSeq; seq >= state.FirstSeq; {
		sm, err := mset.store.LoadPrevMsg(seq, &smv)
		if err == ErrStoreEOF || err != nil {
			break
		}
		seq = sm.seq - 1
		if len(sm.hdr) == 0 {
			continue
		}

		ss := getHeader(JSStreamSource, sm.hdr)
		if len(ss) == 0 {
			continue
		}
		streamName, indexName, sseq := streamAndSeq(bytesToString(ss))

		if _, ok := iNames[indexName]; ok {
			si := mset.sources[indexName]
			si.sseq = sseq
			si.dseq = 0
			delete(iNames, indexName)
		} else if indexName == _EMPTY_ && streamName != _EMPTY_ {
			for iName := range iNames {
				// TODO streamSource is a linear walk, to optimize later
				if si := mset.sources[iName]; si != nil && streamName == si.name ||
					(mset.streamSource(iName).External != nil && streamName == si.name+":"+getHash(mset.streamSource(iName).External.ApiPrefix)) {
					si.sseq = sseq
					si.dseq = 0
					delete(iNames, iName)
					break
				}
			}
		}
		if len(iNames) == 0 {
			break
		}
	}
}

// Resets the SourceInfo for all the sources
// lock should be held.
func (mset *stream) resetSourceInfo() {
	// Reset if needed.
	mset.stopSourceConsumers()
	mset.sources = make(map[string]*sourceInfo)

	for _, ssi := range mset.cfg.Sources {
		if ssi.iname == _EMPTY_ {
			ssi.setIndexName()
		}

		var si *sourceInfo

		if len(ssi.SubjectTransforms) == 0 {
			si = &sourceInfo{name: ssi.Name, iname: ssi.iname, sf: ssi.FilterSubject}
		} else {
			sfs := make([]string, len(ssi.SubjectTransforms))
			trs := make([]*subjectTransform, len(ssi.SubjectTransforms))
			for i, str := range ssi.SubjectTransforms {
				tr, err := NewSubjectTransform(str.Source, str.Destination)
				if err != nil {
					mset.srv.Errorf("Unable to get subject transform for source: %v", err)
				}
				sfs[i] = str.Source
				trs[i] = tr
			}
			si = &sourceInfo{name: ssi.Name, iname: ssi.iname, sfs: sfs, trs: trs}
		}
		mset.sources[ssi.iname] = si
	}
}

// This will do a reverse scan on startup or leader election
// searching for the starting sequence number.
// This can be slow in degenerative cases.
// Lock should be held.
func (mset *stream) startingSequenceForSources() {
	if len(mset.cfg.Sources) == 0 {
		return
	}

	// Always reset here.
	mset.resetSourceInfo()

	var state StreamState
	mset.store.FastState(&state)

	// Bail if no messages, meaning no context.
	if state.Msgs == 0 {
		return
	}

	// For short circuiting return.
	expected := len(mset.cfg.Sources)
	seqs := make(map[string]uint64)

	// Stamp our si seq records on the way out.
	defer func() {
		for sname, seq := range seqs {
			// Ignore if not set.
			if seq == 0 {
				continue
			}
			if si := mset.sources[sname]; si != nil {
				si.sseq = seq
				si.dseq = 0
			}
		}
	}()

	update := func(iName string, seq uint64) {
		// Only update active in case we have older ones in here that got configured out.
		if si := mset.sources[iName]; si != nil {
			if _, ok := seqs[iName]; !ok {
				seqs[iName] = seq
			}
		}
	}

	var smv StoreMsg
	for seq := state.LastSeq; ; {
		sm, err := mset.store.LoadPrevMsg(seq, &smv)
		if err == ErrStoreEOF || err != nil {
			break
		}
		seq = sm.seq - 1
		if len(sm.hdr) == 0 {
			continue
		}
		ss := getHeader(JSStreamSource, sm.hdr)
		if len(ss) == 0 {
			continue
		}

		streamName, iName, sseq := streamAndSeq(bytesToString(ss))
		if iName == _EMPTY_ { // Pre-2.10 message header means it's a match for any source using that stream name
			for _, ssi := range mset.cfg.Sources {
				if streamName == ssi.Name || (ssi.External != nil && streamName == ssi.Name+":"+getHash(ssi.External.ApiPrefix)) {
					update(ssi.iname, sseq)
				}
			}
		} else {
			update(iName, sseq)
		}
		if len(seqs) == expected {
			return
		}
	}
}

// Setup our source consumers.
// Lock should be held.
func (mset *stream) setupSourceConsumers() error {
	if mset.outq == nil {
		return errors.New("outq required")
	}
	// Reset if needed.
	for _, si := range mset.sources {
		if si.sub != nil {
			mset.cancelSourceConsumer(si.iname)
		}
	}

	// If we are no longer the leader, give up
	if !mset.isLeader() {
		return nil
	}

	mset.startingSequenceForSources()

	// Setup our consumers at the proper starting position.
	for _, ssi := range mset.cfg.Sources {
		if si := mset.sources[ssi.iname]; si != nil {
			mset.setupSourceConsumer(ssi.iname, si.sseq+1, time.Time{})
		}
	}

	return nil
}

// Will create internal subscriptions for the stream.
// Lock should be held.
func (mset *stream) subscribeToStream() error {
	if mset.active {
		return nil
	}
	for _, subject := range mset.cfg.Subjects {
		if _, err := mset.subscribeInternal(subject, mset.processInboundJetStreamMsg); err != nil {
			return err
		}
	}
	// Check if we need to setup mirroring.
	if mset.cfg.Mirror != nil {
		// setup the initial mirror sourceInfo
		mset.mirror = &sourceInfo{name: mset.cfg.Mirror.Name}
		sfs := make([]string, len(mset.cfg.Mirror.SubjectTransforms))
		trs := make([]*subjectTransform, len(mset.cfg.Mirror.SubjectTransforms))

		for i, tr := range mset.cfg.Mirror.SubjectTransforms {
			// will not fail as already checked before that the transform will work
			subjectTransform, err := NewSubjectTransform(tr.Source, tr.Destination)
			if err != nil {
				mset.srv.Errorf("Unable to get transform for mirror consumer: %v", err)
			}

			sfs[i] = tr.Source
			trs[i] = subjectTransform
		}
		mset.mirror.sfs = sfs
		mset.mirror.trs = trs
		// delay the actual mirror consumer creation for after a delay
		mset.scheduleSetupMirrorConsumerRetry()
	} else if len(mset.cfg.Sources) > 0 && mset.sourcesConsumerSetup == nil {
		// Setup the initial source infos for the sources
		mset.resetSourceInfo()
		// Delay the actual source consumer(s) creation(s) for after a delay if a replicated stream.
		// If it's an R1, this is done at startup and we will do inline.
		if mset.cfg.Replicas == 1 {
			mset.setupSourceConsumers()
		} else {
			mset.sourcesConsumerSetup = time.AfterFunc(time.Duration(rand.Intn(int(500*time.Millisecond)))+100*time.Millisecond, func() {
				mset.mu.Lock()
				mset.setupSourceConsumers()
				mset.mu.Unlock()
			})
		}
	}
	// Check for direct get access.
	// We spin up followers for clustered streams in monitorStream().
	if mset.cfg.AllowDirect {
		if err := mset.subscribeToDirect(); err != nil {
			return err
		}
	}

	mset.active = true
	return nil
}

// Lock should be held.
func (mset *stream) subscribeToDirect() error {
	// We will make this listen on a queue group by default, which can allow mirrors to participate on opt-in basis.
	if mset.directSub == nil {
		dsubj := fmt.Sprintf(JSDirectMsgGetT, mset.cfg.Name)
		if sub, err := mset.queueSubscribeInternal(dsubj, dgetGroup, mset.processDirectGetRequest); err == nil {
			mset.directSub = sub
		} else {
			return err
		}
	}
	// Now the one that will have subject appended past stream name.
	if mset.lastBySub == nil {
		dsubj := fmt.Sprintf(JSDirectGetLastBySubjectT, mset.cfg.Name, fwcs)
		// We will make this listen on a queue group by default, which can allow mirrors to participate on opt-in basis.
		if sub, err := mset.queueSubscribeInternal(dsubj, dgetGroup, mset.processDirectGetLastBySubjectRequest); err == nil {
			mset.lastBySub = sub
		} else {
			return err
		}
	}

	return nil
}

// Lock should be held.
func (mset *stream) unsubscribeToDirect() {
	if mset.directSub != nil {
		mset.unsubscribe(mset.directSub)
		mset.directSub = nil
	}
	if mset.lastBySub != nil {
		mset.unsubscribe(mset.lastBySub)
		mset.lastBySub = nil
	}
}

// Lock should be held.
func (mset *stream) subscribeToMirrorDirect() error {
	if mset.cfg.Mirror == nil {
		return nil
	}
	mirrorName := mset.cfg.Mirror.Name

	// We will make this listen on a queue group by default, which can allow mirrors to participate on opt-in basis.
	if mset.mirrorDirectSub == nil {
		dsubj := fmt.Sprintf(JSDirectMsgGetT, mirrorName)
		// We will make this listen on a queue group by default, which can allow mirrors to participate on opt-in basis.
		if sub, err := mset.queueSubscribeInternal(dsubj, dgetGroup, mset.processDirectGetRequest); err == nil {
			mset.mirrorDirectSub = sub
		} else {
			return err
		}
	}
	// Now the one that will have subject appended past stream name.
	if mset.mirrorLastBySub == nil {
		dsubj := fmt.Sprintf(JSDirectGetLastBySubjectT, mirrorName, fwcs)
		// We will make this listen on a queue group by default, which can allow mirrors to participate on opt-in basis.
		if sub, err := mset.queueSubscribeInternal(dsubj, dgetGroup, mset.processDirectGetLastBySubjectRequest); err == nil {
			mset.mirrorLastBySub = sub
		} else {
			return err
		}
	}

	return nil
}

// Lock should be held.
func (mset *stream) unsubscribeToMirrorDirect() {
	if mset.mirrorDirectSub != nil {
		mset.unsubscribe(mset.mirrorDirectSub)
		mset.mirrorDirectSub = nil
	}
	if mset.mirrorLastBySub != nil {
		mset.unsubscribe(mset.mirrorLastBySub)
		mset.mirrorLastBySub = nil
	}
}

// Stop our source consumers.
// Lock should be held.
func (mset *stream) stopSourceConsumers() {
	for _, si := range mset.sources {
		mset.cancelSourceInfo(si)
	}
}

// Lock should be held.
func (mset *stream) removeInternalConsumer(si *sourceInfo) {
	if si == nil || si.cname == _EMPTY_ {
		return
	}
	si.cname = _EMPTY_
}

// Will unsubscribe from the stream.
// Lock should be held.
func (mset *stream) unsubscribeToStream(stopping, shuttingDown bool) error {
	for _, subject := range mset.cfg.Subjects {
		mset.unsubscribeInternal(subject)
	}
	if mset.mirror != nil {
		mset.cancelSourceInfo(mset.mirror)
		mset.mirror = nil
	}

	if len(mset.sources) > 0 {
		mset.stopSourceConsumers()
	}
	// Clear batching state.
	mset.deleteInflightBatches(shuttingDown)

	if stopping {
		// In case we had a direct get subscriptions.
		mset.unsubscribeToDirect()
		mset.unsubscribeToMirrorDirect()
	}

	if mset.directLeaderSub == nil {
		// Always unsubscribe the leader sub.
		mset.unsubscribe(mset.directLeaderSub)
		mset.directLeaderSub = nil
	}

	mset.active = false
	return nil
}

// Lock should be held.
func (mset *stream) deleteInflightBatches(shuttingDown bool) {
	if mset.batches != nil {
		mset.batches.mu.Lock()
		for batchId, b := range mset.batches.group {
			// If shutting down, do fixup during startup. In-memory batches don't require manual cleanup.
			if shuttingDown {
				b.stopLocked()
			} else {
				b.cleanupLocked(batchId, mset.batches)
			}
		}
		mset.batches.mu.Unlock()
		mset.batches = nil
	}
}

// Lock should be held.
func (mset *stream) deleteBatchApplyState() {
	if batch := mset.batchApply; batch != nil {
		// Need to return entries (if any) to the pool.
		for _, bce := range batch.entries {
			bce.ReturnToPool()
		}
		mset.batchApply = nil
	}
}

// Lock does NOT need to be held, we set the client on setup and never change it at this point.
func (mset *stream) subscribeInternal(subject string, cb msgHandler) (*subscription, error) {
	if mset.closed.Load() {
		return nil, errStreamClosed
	}
	if cb == nil {
		return nil, errInvalidMsgHandler
	}
	c := mset.client
	sid := int(mset.sid.Add(1))
	// Now create the subscription
	return c.processSub([]byte(subject), nil, []byte(strconv.Itoa(sid)), cb, false)
}

// Lock does NOT need to be held, we set the client on setup and never change it at this point.
func (mset *stream) queueSubscribeInternal(subject, group string, cb msgHandler) (*subscription, error) {
	if mset.closed.Load() {
		return nil, errStreamClosed
	}
	if cb == nil {
		return nil, errInvalidMsgHandler
	}
	c := mset.client
	sid := int(mset.sid.Add(1))
	// Now create the subscription
	return c.processSub([]byte(subject), []byte(group), []byte(strconv.Itoa(sid)), cb, false)
}

// This will unsubscribe us from the exact subject given.
// We do not currently track the subs so do not have the sid.
// This should be called only on an update.
// Lock does NOT need to be held, we set the client on setup and never change it at this point.
func (mset *stream) unsubscribeInternal(subject string) error {
	if mset.closed.Load() {
		return errStreamClosed
	}
	c := mset.client
	var sid []byte
	c.mu.Lock()
	for _, sub := range c.subs {
		if subject == string(sub.subject) {
			sid = sub.sid
			break
		}
	}
	c.mu.Unlock()

	if sid != nil {
		return c.processUnsub(sid)
	}
	return nil
}

// Lock should be held.
func (mset *stream) unsubscribe(sub *subscription) {
	if sub == nil || mset.closed.Load() {
		return
	}
	mset.client.processUnsub(sub.sid)
}

func (mset *stream) setupStore(fsCfg *FileStoreConfig) error {
	mset.mu.Lock()
	mset.created = time.Now().UTC()

	switch mset.cfg.Storage {
	case MemoryStorage:
		ms, err := newMemStore(&mset.cfg)
		if err != nil {
			mset.mu.Unlock()
			return err
		}
		mset.store = ms
	case FileStorage:
		s := mset.srv
		prf := s.jsKeyGen(s.getOpts().JetStreamKey, mset.acc.Name)
		if prf != nil {
			// We are encrypted here, fill in correct cipher selection.
			fsCfg.Cipher = s.getOpts().JetStreamCipher
		}
		oldprf := s.jsKeyGen(s.getOpts().JetStreamOldKey, mset.acc.Name)
		cfg := *fsCfg
		cfg.srv = s
		fs, err := newFileStoreWithCreated(cfg, mset.cfg, mset.created, prf, oldprf)
		if err != nil {
			mset.mu.Unlock()
			return err
		}
		mset.store = fs
	}
	// This will fire the callback but we do not require the lock since md will be 0 here.
	mset.store.RegisterStorageUpdates(mset.storeUpdates)
	mset.store.RegisterStorageRemoveMsg(func(seq uint64) {
		if mset.IsClustered() {
			if mset.IsLeader() {
				mset.mu.RLock()
				md := streamMsgDelete{Seq: seq, NoErase: true, Stream: mset.cfg.Name}
				mset.node.Propose(encodeMsgDelete(&md))
				mset.mu.RUnlock()
			}
		} else {
			mset.removeMsg(seq)
		}
	})
	mset.store.RegisterProcessJetStreamMsg(func(im *inMsg) {
		if mset.IsClustered() {
			if mset.IsLeader() {
				mset.processClusteredInboundMsg(im.subj, im.rply, im.hdr, im.msg, im.mt, false)
			}
		} else {
			mset.processJetStreamMsg(im.subj, im.rply, im.hdr, im.msg, 0, 0, im.mt, false, true)
		}
	})
	mset.mu.Unlock()

	return nil
}

// Called for any updates to the underlying stream. We pass through the bytes to the
// jetstream account. We do local processing for stream pending for consumers, but only
// for removals.
// Lock should not be held.
func (mset *stream) storeUpdates(md, bd int64, seq uint64, subj string) {
	// If we have a single negative update then we will process our consumers for stream pending.
	// Purge and Store handled separately inside individual calls.
	if md == -1 && seq > 0 && subj != _EMPTY_ {
		// We use our consumer list mutex here instead of the main stream lock since it may be held already.
		mset.clsMu.RLock()
		if mset.csl != nil {
			mset.csl.Match(subj, func(o *consumer) {
				o.decStreamPending(seq, subj)
			})
		} else {
			for _, o := range mset.cList {
				o.decStreamPending(seq, subj)
			}
		}
		mset.clsMu.RUnlock()
	} else if md < 0 {
		// Batch decrements we need to force consumers to re-calculate num pending.
		mset.clsMu.RLock()
		for _, o := range mset.cList {
			o.streamNumPendingLocked()
		}
		mset.clsMu.RUnlock()
	}

	if mset.jsa != nil {
		mset.jsa.updateUsage(mset.tier, mset.stype, bd)
	}
}

// NumMsgIds returns the number of message ids being tracked for duplicate suppression.
func (mset *stream) numMsgIds() int {
	mset.ddMu.Lock()
	defer mset.ddMu.Unlock()
	return len(mset.ddmap)
}

// checkMsgId will process and check for duplicates.
// mset.ddMu lock should be held.
func (mset *stream) checkMsgId(id string) *ddentry {
	if id == _EMPTY_ || len(mset.ddmap) == 0 {
		return nil
	}
	return mset.ddmap[id]
}

// Will purge the entries that are past the window.
// Should be called from a timer.
func (mset *stream) purgeMsgIds() {
	now := time.Now().UnixNano()
	mset.cfgMu.RLock()
	tmrNext := mset.cfg.Duplicates
	mset.cfgMu.RUnlock()
	window := int64(tmrNext)

	mset.ddMu.Lock()
	defer mset.ddMu.Unlock()

	for i, dde := range mset.ddarr[mset.ddindex:] {
		if now-dde.ts >= window {
			delete(mset.ddmap, dde.id)
		} else {
			mset.ddindex += i
			// Check if we should garbage collect here if we are 1/3 total size.
			if cap(mset.ddarr) > 3*(len(mset.ddarr)-mset.ddindex) {
				mset.ddarr = append([]*ddentry(nil), mset.ddarr[mset.ddindex:]...)
				mset.ddindex = 0
			}
			tmrNext = time.Duration(window - (now - dde.ts))
			break
		}
	}
	if len(mset.ddmap) > 0 {
		// Make sure to not fire too quick
		const minFire = 50 * time.Millisecond
		if tmrNext < minFire {
			tmrNext = minFire
		}
		if mset.ddtmr != nil {
			mset.ddtmr.Reset(tmrNext)
		} else {
			mset.ddtmr = time.AfterFunc(tmrNext, mset.purgeMsgIds)
		}
	} else {
		if mset.ddtmr != nil {
			mset.ddtmr.Stop()
			mset.ddtmr = nil
		}
		mset.ddmap = nil
		mset.ddarr = nil
		mset.ddindex = 0
	}
}

// storeMsgId will store the message id for duplicate detection.
func (mset *stream) storeMsgId(dde *ddentry) {
	mset.ddMu.Lock()
	defer mset.ddMu.Unlock()
	mset.storeMsgIdLocked(dde)
}

// storeMsgIdLocked will store the message id for duplicate detection.
// mset.ddMu lock should be held.
func (mset *stream) storeMsgIdLocked(dde *ddentry) {
	if mset.ddmap == nil {
		mset.ddmap = make(map[string]*ddentry)
	}
	mset.ddmap[dde.id] = dde
	mset.ddarr = append(mset.ddarr, dde)
	if mset.ddtmr == nil {
		mset.ddtmr = time.AfterFunc(mset.cfg.Duplicates, mset.purgeMsgIds)
	}
}

// Fast lookup of msgId.
func getMsgId(hdr []byte) string {
	return string(getHeader(JSMsgId, hdr))
}

// Fast lookup of expected last msgId.
func getExpectedLastMsgId(hdr []byte) string {
	return string(getHeader(JSExpectedLastMsgId, hdr))
}

// Fast lookup of expected stream.
func getExpectedStream(hdr []byte) string {
	return string(getHeader(JSExpectedStream, hdr))
}

// Fast lookup of expected last sequence.
func getExpectedLastSeq(hdr []byte) (uint64, bool) {
	bseq := sliceHeader(JSExpectedLastSeq, hdr)
	if len(bseq) == 0 {
		return 0, false
	}
	return uint64(parseInt64(bseq)), true
}

// Fast lookup of rollups.
func getRollup(hdr []byte) string {
	r := getHeader(JSMsgRollup, hdr)
	if len(r) == 0 {
		return _EMPTY_
	}
	return strings.ToLower(string(r))
}

// Fast lookup of expected stream sequence per subject.
func getExpectedLastSeqPerSubject(hdr []byte) (uint64, bool) {
	bseq := sliceHeader(JSExpectedLastSubjSeq, hdr)
	if len(bseq) == 0 {
		return 0, false
	}
	return uint64(parseInt64(bseq)), true
}

// Fast lookup of expected subject for the expected stream sequence per subject.
func getExpectedLastSeqPerSubjectForSubject(hdr []byte) string {
	return bytesToString(sliceHeader(JSExpectedLastSubjSeqSubj, hdr))
}

// Fast lookup of the message TTL from headers:
// - Positive return value: duration in seconds.
// - Zero return value: no TTL or parse error.
// - Negative return value: never expires.
func getMessageTTL(hdr []byte) (int64, error) {
	ttl := getHeader(JSMessageTTL, hdr)
	if len(ttl) == 0 {
		return 0, nil
	}
	return parseMessageTTL(bytesToString(ttl))
}

// - Positive return value: duration in seconds.
// - Zero return value: no TTL or parse error.
// - Negative return value: never expires.
func parseMessageTTL(ttl string) (int64, error) {
	if strings.ToLower(ttl) == "never" {
		return -1, nil
	}
	dur, err := time.ParseDuration(ttl)
	if err == nil {
		if dur < time.Second {
			return 0, NewJSMessageTTLInvalidError()
		}
		return int64(dur.Seconds()), nil
	}
	t := parseInt64(stringToBytes(ttl))
	if t < 0 {
		// This probably means a parse failure, hence why
		// we have a special case "never" for returning -1.
		// Otherwise we can't know if it's a genuine TTL
		// that says never expire or if it's a parse error.
		return 0, NewJSMessageTTLInvalidError()
	}
	return t, nil
}

// Fast lookup of the message Incr from headers.
// Return includes the value or nil, and success.
func getMessageIncr(hdr []byte) (*big.Int, bool) {
	incr := sliceHeader(JSMessageIncr, hdr)
	if len(incr) == 0 {
		return nil, true
	}
	var v big.Int
	return v.SetString(bytesToString(incr), 10)
}

// Fast lookup of message schedule.
func getMessageSchedule(hdr []byte) (time.Time, bool) {
	if len(hdr) == 0 {
		return time.Time{}, true
	}
	val := bytesToString(sliceHeader(JSSchedulePattern, hdr))
	if val == _EMPTY_ {
		return time.Time{}, true
	}
	if !strings.HasPrefix(val, "@at ") {
		return time.Time{}, false
	}
	t, err := time.Parse(time.RFC3339, val[4:])
	return t, err == nil
}

// Fast lookup of the message schedule TTL from headers.
// The TTL is confirmed to be valid, but the raw TTL string is returned.
func getMessageScheduleTTL(hdr []byte) (string, bool) {
	ttl := getHeader(JSScheduleTTL, hdr)
	if len(ttl) == 0 {
		return _EMPTY_, true
	}
	if _, err := parseMessageTTL(bytesToString(ttl)); err != nil {
		return _EMPTY_, false
	}
	return string(ttl), true
}

// Fast lookup of message schedule target.
func getMessageScheduleTarget(hdr []byte) string {
	if len(hdr) == 0 {
		return _EMPTY_
	}
	return string(getHeader(JSScheduleTarget, hdr))
}

// Fast lookup of message scheduler.
func getMessageScheduler(hdr []byte) string {
	if len(hdr) == 0 {
		return _EMPTY_
	}
	return string(getHeader(JSScheduler, hdr))
}

// Fast lookup of batch ID.
func getBatchId(hdr []byte) string {
	if len(hdr) == 0 {
		return _EMPTY_
	}
	return string(getHeader(JSBatchId, hdr))
}

// Fast lookup of batch sequence.
func getBatchSequence(hdr []byte) (uint64, bool) {
	bseq := sliceHeader(JSBatchSeq, hdr)
	if len(bseq) == 0 {
		return 0, false
	}
	return uint64(parseInt64(bseq)), true
}

// Signal if we are clustered. Will acquire rlock.
func (mset *stream) IsClustered() bool {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.isClustered()
}

// Lock should be held.
func (mset *stream) isClustered() bool {
	return mset.node != nil
}

// Used if we have to queue things internally to avoid the route/gw path.
type inMsg struct {
	subj string
	rply string
	hdr  []byte
	msg  []byte
	si   *sourceInfo
	mt   *msgTrace
	seq  uint64 // seq that can be optionally used for sorting out-of-band.
}

var inMsgPool = sync.Pool{
	New: func() any {
		return &inMsg{}
	},
}

func (im *inMsg) returnToPool() {
	im.subj, im.rply, im.hdr, im.msg, im.si, im.mt = _EMPTY_, _EMPTY_, nil, nil, nil, nil
	inMsgPool.Put(im)
}

func (mset *stream) queueInbound(ib *ipQueue[*inMsg], subj, rply string, hdr, msg []byte, si *sourceInfo, mt *msgTrace) {
	im := inMsgPool.Get().(*inMsg)
	im.subj, im.rply, im.hdr, im.msg, im.si, im.mt = subj, rply, hdr, msg, si, mt
	if _, err := ib.push(im); err != nil {
		im.returnToPool()
		streamName := mset.cfg.Name
		mset.srv.RateLimitWarnf("Dropping messages due to excessive stream ingest rate on '%s' > '%s': %s", mset.acc.Name, streamName, err)
		if rply != _EMPTY_ {
			hdr := []byte("NATS/1.0 429 Too Many Requests\r\n\r\n")
			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: streamName}, Error: NewJSStreamTooManyRequestsError()})
			mset.outq.send(newJSPubMsg(rply, _EMPTY_, _EMPTY_, hdr, b, nil, 0))
		}
	}
}

var dgPool = sync.Pool{
	New: func() any {
		return &directGetReq{}
	},
}

// For when we need to not inline the request.
type directGetReq struct {
	// Copy of this is correct for this.
	req   JSApiMsgGetRequest
	reply string
}

// processDirectGetRequest handles direct get request for stream messages.
func (mset *stream) processDirectGetRequest(_ *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
	if len(reply) == 0 {
		return
	}
	hdr, msg := c.msgParts(rmsg)
	if errorOnRequiredApiLevel(hdr) {
		hdr := []byte("NATS/1.0 412 Required Api Level\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}
	if len(msg) == 0 {
		hdr := []byte("NATS/1.0 408 Empty Request\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}
	var req JSApiMsgGetRequest
	err := json.Unmarshal(msg, &req)
	if err != nil {
		hdr := []byte("NATS/1.0 408 Malformed Request\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}
	// Check if nothing set.
	if req.Seq == 0 && req.LastFor == _EMPTY_ && req.NextFor == _EMPTY_ && len(req.MultiLastFor) == 0 && req.StartTime == nil {
		hdr := []byte("NATS/1.0 408 Empty Request\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}
	// Check we don't have conflicting options set.
	// We do not allow batch mode for lastFor requests.
	if (req.Seq > 0 && req.LastFor != _EMPTY_) ||
		(req.Seq > 0 && req.StartTime != nil) ||
		(req.StartTime != nil && req.LastFor != _EMPTY_) ||
		(req.LastFor != _EMPTY_ && req.NextFor != _EMPTY_) ||
		(req.LastFor != _EMPTY_ && req.Batch > 0) ||
		(req.LastFor != _EMPTY_ && len(req.MultiLastFor) > 0) ||
		(req.NextFor != _EMPTY_ && len(req.MultiLastFor) > 0) ||
		(req.UpToSeq > 0 && req.UpToTime != nil) {
		hdr := []byte("NATS/1.0 408 Bad Request\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}

	inlineOk := c.kind != ROUTER && c.kind != GATEWAY && c.kind != LEAF
	if !inlineOk {
		dg := dgPool.Get().(*directGetReq)
		dg.req, dg.reply = req, reply
		mset.gets.push(dg)
	} else {
		mset.getDirectRequest(&req, reply)
	}
}

// This is for direct get by last subject which is part of the subject itself.
func (mset *stream) processDirectGetLastBySubjectRequest(_ *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
	if len(reply) == 0 {
		return
	}
	hdr, msg := c.msgParts(rmsg)
	if errorOnRequiredApiLevel(hdr) {
		hdr := []byte("NATS/1.0 412 Required Api Level\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}

	var req JSApiMsgGetRequest
	if len(msg) > 0 {
		err := json.Unmarshal(msg, &req)
		if err != nil {
			hdr := []byte("NATS/1.0 408 Malformed Request\r\n\r\n")
			mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
			return
		}
	}

	// Extract the key.
	var key string
	for i, n := 0, 0; i < len(subject); i++ {
		if subject[i] == btsep {
			if n == 4 {
				if start := i + 1; start < len(subject) {
					key = subject[i+1:]
				}
				break
			}
			n++
		}
	}
	if len(key) == 0 {
		hdr := []byte("NATS/1.0 408 Bad Request\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}

	// Error if request fields are provided that don't make sense for this endpoint.
	switch {
	case req.LastFor != _EMPTY_, req.NextFor != _EMPTY_,
		len(req.MultiLastFor) > 0, req.Seq != 0, req.Batch != 0,
		req.StartTime != nil, req.UpToSeq != 0, req.UpToTime != nil:
		hdr := []byte("NATS/1.0 408 Bad Request\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	default:
		req.LastFor = key
	}

	inlineOk := c.kind != ROUTER && c.kind != GATEWAY && c.kind != LEAF
	if !inlineOk {
		dg := dgPool.Get().(*directGetReq)
		dg.req, dg.reply = req, reply
		mset.gets.push(dg)
	} else {
		mset.getDirectRequest(&req, reply)
	}
}

// For direct get batch and multi requests.
const (
	dg   = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\n\r\n"
	dgb  = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\n\r\n"
	eob  = "NATS/1.0 204 EOB\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\n\r\n"
	eobm = "NATS/1.0 204 EOB\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\nNats-UpTo-Sequence: %d\r\n\r\n"
)

// Handle a multi request.
func (mset *stream) getDirectMulti(req *JSApiMsgGetRequest, reply string) {
	// TODO(dlc) - Make configurable?
	const maxAllowedResponses = 1024

	// Ensure this read request is isolated and doesn't interleave with writes.
	mset.mu.RLock()
	defer mset.mu.RUnlock()

	// Grab store and name.
	store, name, s := mset.store, mset.cfg.Name, mset.srv

	// Grab MaxBytes
	mb := req.MaxBytes
	if mb == 0 && s != nil {
		// Fill in with the server's MaxPending.
		mb = int(s.opts.MaxPending)
	}

	upToSeq := req.UpToSeq
	// If we have UpToTime set get the proper sequence.
	if req.UpToTime != nil {
		upToSeq = store.GetSeqFromTime((*req.UpToTime).UTC())
		// Avoid selecting a first sequence that will take us to before the stream first
		// sequence, otherwise we can return messages after the supplied UpToTime.
		if upToSeq <= mset.state().FirstSeq {
			hdr := []byte("NATS/1.0 404 No Results\r\n\r\n")
			mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
			return
		}
		// We need to back off one since this is used to determine start sequence normally,
		// whereas here we want it to be the ceiling.
		upToSeq--
	}
	// If not set, set to the last sequence and remember that for EOB.
	if upToSeq == 0 {
		var state StreamState
		mset.store.FastState(&state)
		upToSeq = state.LastSeq
	}

	seqs, err := store.MultiLastSeqs(req.MultiLastFor, upToSeq, maxAllowedResponses)
	if err != nil {
		var hdr []byte
		if err == ErrTooManyResults {
			hdr = []byte("NATS/1.0 413 Too Many Results\r\n\r\n")
		} else {
			hdr = []byte(fmt.Sprintf("NATS/1.0 500 %v\r\n\r\n", err))
		}
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}
	if len(seqs) == 0 {
		hdr := []byte("NATS/1.0 404 No Results\r\n\r\n")
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
		return
	}

	np, lseq, sentBytes, sent := uint64(len(seqs)), uint64(0), 0, 0
	for _, seq := range seqs {
		if seq < req.Seq {
			if np > 0 {
				np--
			}
			continue
		}
		var svp StoreMsg
		sm, err := store.LoadMsg(seq, &svp)
		if err != nil {
			hdr := []byte("NATS/1.0 404 Message Not Found\r\n\r\n")
			mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
			return
		}

		hdr := sm.hdr
		ts := time.Unix(0, sm.ts).UTC()

		// Decrement num pending. This is an optimization, and we do not continue to look it up for these operations.
		if np > 0 {
			np--
		}
		if len(hdr) == 0 {
			hdr = fmt.Appendf(nil, dgb, name, sm.subj, sm.seq, ts.Format(time.RFC3339Nano), np, lseq)
		} else {
			hdr = copyBytes(hdr)
			hdr = genHeader(hdr, JSStream, name)
			hdr = genHeader(hdr, JSSubject, sm.subj)
			hdr = genHeader(hdr, JSSequence, strconv.FormatUint(sm.seq, 10))
			hdr = genHeader(hdr, JSTimeStamp, ts.Format(time.RFC3339Nano))
			hdr = genHeader(hdr, JSNumPending, strconv.FormatUint(np, 10))
			hdr = genHeader(hdr, JSLastSequence, strconv.FormatUint(lseq, 10))
		}
		// Track our lseq
		lseq = sm.seq
		// Send out our message.
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, sm.msg, nil, 0))
		// Check if we have exceeded max bytes.
		sentBytes += len(sm.subj) + len(sm.hdr) + len(sm.msg)
		if sentBytes >= mb {
			break
		}
		sent++
		if req.Batch > 0 && sent >= req.Batch {
			break
		}
	}

	// Send out EOB
	hdr := fmt.Appendf(nil, eobm, np, lseq, upToSeq)
	mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
}

// Do actual work on a direct msg request.
// This could be called in a Go routine if we are inline for a non-client connection.
func (mset *stream) getDirectRequest(req *JSApiMsgGetRequest, reply string) {
	// Handle multi in separate function.
	if len(req.MultiLastFor) > 0 {
		mset.getDirectMulti(req, reply)
		return
	}

	// Ensure this read request is isolated and doesn't interleave with writes.
	mset.mu.RLock()
	defer mset.mu.RUnlock()

	store, name, s := mset.store, mset.cfg.Name, mset.srv

	var seq uint64
	// Lookup start seq if AsOfTime is set.
	if req.StartTime != nil {
		seq = store.GetSeqFromTime(*req.StartTime)
	} else {
		seq = req.Seq
	}

	wc := subjectHasWildcard(req.NextFor)
	// For tracking num pending if we are batch.
	var np, lseq, validThrough uint64
	var isBatchRequest bool
	batch := req.Batch
	if batch == 0 {
		batch = 1
	} else {
		// This is a batch request, capture initial numPending.
		isBatchRequest = true
		np, validThrough = store.NumPending(seq, req.NextFor, false)
	}

	// Grab MaxBytes
	mb := req.MaxBytes
	if mb == 0 && s != nil {
		// Fill in with the server's MaxPending.
		mb = int(s.opts.MaxPending)
	}
	// Track what we have sent.
	var sentBytes int

	// Loop over batch, which defaults to 1.
	for i := 0; i < batch; i++ {
		var (
			svp StoreMsg
			sm  *StoreMsg
			err error
		)
		if seq > 0 && req.NextFor == _EMPTY_ {
			// Only do direct lookup for a non batch.
			if i == 0 && !isBatchRequest {
				sm, err = store.LoadMsg(seq, &svp)
			} else {
				// We want to use load next with fwcs to step over deleted msgs.
				sm, seq, err = store.LoadNextMsg(fwcs, true, seq, &svp)
			}
			// Bump for next loop if applicable.
			seq++
		} else if req.NextFor != _EMPTY_ {
			sm, seq, err = store.LoadNextMsg(req.NextFor, wc, seq, &svp)
			seq++
		} else {
			// Batch is not applicable here, this is checked before we get here.
			sm, err = store.LoadLastMsg(req.LastFor, &svp)
		}
		if err != nil {
			// For batches, if we stop early we want to do EOB logic below.
			if batch > 1 && i > 0 {
				break
			}
			hdr := []byte("NATS/1.0 404 Message Not Found\r\n\r\n")
			mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
			return
		}

		ts := time.Unix(0, sm.ts).UTC()
		var hdr []byte
		if !req.NoHeaders {
			hdr = sm.hdr
			if isBatchRequest {
				// Decrement num pending. This is an optimization, and we do not continue to look it up for these operations.
				if np > 0 {
					np--
				}
				if len(hdr) == 0 {
					hdr = fmt.Appendf(nil, dgb, name, sm.subj, sm.seq, ts.Format(time.RFC3339Nano), np, lseq)
				} else {
					hdr = copyBytes(hdr)
					hdr = genHeader(hdr, JSStream, name)
					hdr = genHeader(hdr, JSSubject, sm.subj)
					hdr = genHeader(hdr, JSSequence, strconv.FormatUint(sm.seq, 10))
					hdr = genHeader(hdr, JSTimeStamp, ts.Format(time.RFC3339Nano))
					hdr = genHeader(hdr, JSNumPending, strconv.FormatUint(np, 10))
					hdr = genHeader(hdr, JSLastSequence, strconv.FormatUint(lseq, 10))
				}
			} else {
				if len(hdr) == 0 {
					hdr = fmt.Appendf(nil, dg, name, sm.subj, sm.seq, ts.Format(time.RFC3339Nano))
				} else {
					hdr = copyBytes(hdr)
					hdr = genHeader(hdr, JSStream, name)
					hdr = genHeader(hdr, JSSubject, sm.subj)
					hdr = genHeader(hdr, JSSequence, strconv.FormatUint(sm.seq, 10))
					hdr = genHeader(hdr, JSTimeStamp, ts.Format(time.RFC3339Nano))
				}
			}
		}
		// Track our lseq
		lseq = sm.seq
		// Send out our message.
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, sm.msg, nil, 0))
		// Check if we have exceeded max bytes.
		sentBytes += len(sm.subj) + len(sm.hdr) + len(sm.msg)
		if sentBytes >= mb {
			break
		}
	}

	// If batch was requested send EOB.
	if isBatchRequest {
		// Update if the stream's lasts sequence has moved past our validThrough.
		if mset.lastSeq() > validThrough {
			np, _ = store.NumPending(seq, req.NextFor, false)
		}
		hdr := fmt.Appendf(nil, eob, np, lseq)
		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
	}
}

// processInboundJetStreamMsg handles processing messages bound for a stream.
func (mset *stream) processInboundJetStreamMsg(_ *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
	hdr, msg := c.msgParts(copyBytes(rmsg)) // Need to copy.
	if mt, traceOnly := c.isMsgTraceEnabled(); mt != nil {
		// If message is delivered, we need to disable the message trace headers
		// to prevent a trace event to be generated when a stored message
		// is delivered to a consumer and routed.
		if !traceOnly {
			disableTraceHeaders(c, hdr)
		}
		// This will add the jetstream event while in the client read loop.
		// Since the event will be updated in a different go routine, the
		// tracing object will have a separate reference to the JS trace
		// object.
		mt.addJetStreamEvent(mset.name())
	}
	mset.queueInbound(mset.msgs, subject, reply, hdr, msg, nil, c.pa.trace)
}

var (
	errLastSeqMismatch   = errors.New("last sequence mismatch")
	errMsgIdDuplicate    = errors.New("msgid is duplicate")
	errStreamClosed      = errors.New("stream closed")
	errInvalidMsgHandler = errors.New("undefined message handler")
	errStreamMismatch    = errors.New("expected stream does not match")
	errMsgTTLDisabled    = errors.New("message TTL disabled")
)

// processJetStreamMsg is where we try to actually process the stream msg.
func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int64, mt *msgTrace, sourced bool, needLock bool) (retErr error) {
	if mt != nil {
		// Only the leader/standalone will have mt!=nil. On exit, send the
		// message trace event.
		defer func() {
			mt.sendEventFromJetStream(retErr)
		}()
	}

	if mset == nil || mset.closed.Load() {
		return errStreamClosed
	}

	// Hold lock while storing the message, and potentially purging as part of a rollup.
	// If we're writing an atomic batch of multiple messages, the lock is already held.
	if needLock {
		mset.mu.Lock()
		defer mset.mu.Unlock()
	}

	s, store := mset.srv, mset.store

	traceOnly := mt.traceOnly()
	bumpCLFS := func() {
		// Do not bump if tracing and not doing message delivery.
		if traceOnly {
			return
		}
		mset.clMu.Lock()
		mset.clfs++
		mset.clMu.Unlock()
	}

	// Apply the input subject transform if any
	if mset.itr != nil {
		ts, err := mset.itr.Match(subject)
		if err == nil {
			// no filtering: if the subject doesn't map the source of the transform, don't change it
			subject = ts
		}
	}

	var accName string
	if mset.acc != nil {
		accName = mset.acc.Name
	}

	js, jsa, doAck := mset.js, mset.jsa, !mset.cfg.NoAck
	name, stype := mset.cfg.Name, mset.cfg.Storage
	maxMsgSize := int(mset.cfg.MaxMsgSize)
	numConsumers := len(mset.consumers)
	interestRetention := mset.cfg.Retention == InterestPolicy
	allowMsgCounter, allowMsgSchedules := mset.cfg.AllowMsgCounter, mset.cfg.AllowMsgSchedules
	// Snapshot if we are the leader and if we can respond.
	isLeader, isSealed := mset.isLeaderNodeState(), mset.cfg.Sealed
	isClustered := mset.isClustered()
	canConsistencyCheck := !isClustered || traceOnly
	canRespond := doAck && len(reply) > 0 && isLeader
	outq := mset.outq

	var resp = &JSPubAckResponse{}

	var batchId string
	var batchSeq uint64
	if len(hdr) > 0 {
		// Populate batch details.
		if batchId = getBatchId(hdr); batchId != _EMPTY_ {
			batchSeq, _ = getBatchSequence(hdr)
			// Disable consistency checking if this was already done
			// earlier as part of the batch consistency check.
			canConsistencyCheck = traceOnly
		}
	}

	// Bail here if sealed.
	if canConsistencyCheck && isSealed {
		if canRespond && outq != nil {
			resp.PubAck = &PubAck{Stream: name}
			resp.Error = ApiErrors[JSStreamSealedErr]
			b, _ := json.Marshal(resp)
			outq.sendMsg(reply, b)
		}
		return ApiErrors[JSStreamSealedErr]
	}

	var buf [256]byte
	pubAck := append(buf[:0], mset.pubAck...)

	// If this is a non-clustered msg and we are not considered active, meaning no active subscription, do not process.
	if lseq == 0 && ts == 0 && !mset.active {
		return nil
	}

	// For clustering the lower layers will pass our expected lseq. If it is present check for that here.
	var clfs uint64
	if lseq > 0 {
		clfs = mset.getCLFS()
		if lseq != (mset.lseq + clfs) {
			isMisMatch := true
			// We may be able to recover here if we have no state whatsoever, or we are a mirror.
			// See if we have to adjust our starting sequence.
			if mset.lseq == 0 || mset.cfg.Mirror != nil {
				var state StreamState
				mset.store.FastState(&state)
				if state.FirstSeq == 0 {
					mset.store.Compact(lseq + 1)
					mset.lseq = lseq
					isMisMatch = false
				}
			}
			// Really is a mismatch.
			if isMisMatch {
				if canRespond && outq != nil {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = ApiErrors[JSStreamSequenceNotMatchErr]
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return errLastSeqMismatch
			}
		}
	}

	// If we have received this message across an account we may have request information attached.
	// For now remove. TODO(dlc) - Should this be opt-in or opt-out?
	if len(hdr) > 0 {
		hdr = removeHeaderIfPresent(hdr, ClientInfoHdr)
	}

	// Process additional msg headers if still present.
	var msgId string
	var incr *big.Int
	var rollupSub, rollupAll bool

	if len(hdr) > 0 {
		// Certain checks have already been performed if in clustered mode, so only check if not.
		// Note, for cluster mode but with message tracing (without message delivery), we need
		// to do this check here since it was not done in processClusteredInboundMsg().
		if canConsistencyCheck {
			// Counter increments.
			// Only supported on counter streams, and payload must be empty (if not coming from a source).
			var ok bool
			if incr, ok = getMessageIncr(hdr); !ok {
				apiErr := NewJSMessageIncrInvalidError()
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = apiErr
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return apiErr
			} else if incr != nil && !sourced {
				// Only do checks if the message isn't sourced. Otherwise, we need to store verbatim.
				if !allowMsgCounter {
					apiErr := NewJSMessageIncrDisabledError()
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = apiErr
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return apiErr
				} else if len(msg) > 0 {
					apiErr := NewJSMessageIncrPayloadError()
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = apiErr
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return apiErr
				} else {
					// Check for incompatible headers.
					var doErr bool
					if getRollup(hdr) != _EMPTY_ ||
						getExpectedStream(hdr) != _EMPTY_ ||
						getExpectedLastMsgId(hdr) != _EMPTY_ ||
						getExpectedLastSeqPerSubjectForSubject(hdr) != _EMPTY_ {
						doErr = true
					} else if _, ok := getExpectedLastSeq(hdr); ok {
						doErr = true
					} else if _, ok := getExpectedLastSeqPerSubject(hdr); ok {
						doErr = true
					}

					if doErr {
						apiErr := NewJSMessageIncrInvalidError()
						if canRespond {
							resp.PubAck = &PubAck{Stream: name}
							resp.Error = apiErr
							b, _ := json.Marshal(resp)
							outq.sendMsg(reply, b)
						}
						return apiErr
					}
				}
			}

			// Expected stream.
			if sname := getExpectedStream(hdr); sname != _EMPTY_ && sname != name {
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = NewJSStreamNotMatchError()
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return errStreamMismatch
			}

			// TTL'd messages are rejected entirely if TTLs are not enabled on the stream.
			// Shouldn't happen in clustered mode since we should have already caught this
			// in processClusteredInboundMsg, but needed here for non-clustered etc.
			if ttl, _ := getMessageTTL(hdr); !sourced && ttl != 0 && !mset.cfg.AllowMsgTTL {
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = NewJSMessageTTLDisabledError()
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return errMsgTTLDisabled
			}

			// Expected last sequence per subject.
			if seq, exists := getExpectedLastSeqPerSubject(hdr); exists {
				// Allow override of the subject used for the check.
				seqSubj := subject
				if optSubj := getExpectedLastSeqPerSubjectForSubject(hdr); optSubj != _EMPTY_ {
					seqSubj = optSubj
				}

				// TODO(dlc) - We could make a new store func that does this all in one.
				var smv StoreMsg
				var fseq uint64
				sm, err := store.LoadLastMsg(seqSubj, &smv)
				if sm != nil {
					fseq = sm.seq
				}
				if err == ErrStoreMsgNotFound && seq == 0 {
					fseq, err = 0, nil
				}
				if err != nil || fseq != seq {
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = NewJSStreamWrongLastSequenceError(fseq)
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return fmt.Errorf("last sequence by subject mismatch: %d vs %d", seq, fseq)
				}
			} else if getExpectedLastSeqPerSubjectForSubject(hdr) != _EMPTY_ {
				apiErr := NewJSStreamExpectedLastSeqPerSubjectInvalidError()
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = apiErr
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return apiErr
			}

			// Expected last sequence.
			if seq, exists := getExpectedLastSeq(hdr); exists && seq != mset.lseq {
				mlseq := mset.lseq
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = NewJSStreamWrongLastSequenceError(mlseq)
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return fmt.Errorf("last sequence mismatch: %d vs %d", seq, mlseq)
			}

			// Message scheduling.
			if schedule, ok := getMessageSchedule(hdr); !ok {
				apiErr := NewJSMessageSchedulesPatternInvalidError()
				if !allowMsgSchedules {
					apiErr = NewJSMessageSchedulesDisabledError()
				}
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = apiErr
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return apiErr
			} else if !schedule.IsZero() {
				if !allowMsgSchedules {
					apiErr := NewJSMessageSchedulesDisabledError()
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = apiErr
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return apiErr
				} else if scheduleTtl, ok := getMessageScheduleTTL(hdr); !ok {
					apiErr := NewJSMessageSchedulesTTLInvalidError()
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = apiErr
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return apiErr
				} else if scheduleTtl != _EMPTY_ && !mset.cfg.AllowMsgTTL {
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = NewJSMessageTTLDisabledError()
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return errMsgTTLDisabled
				} else if scheduleTarget := getMessageScheduleTarget(hdr); scheduleTarget == _EMPTY_ ||
					!IsValidPublishSubject(scheduleTarget) || SubjectsCollide(scheduleTarget, subject) {
					apiErr := NewJSMessageSchedulesTargetInvalidError()
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = apiErr
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return apiErr
				} else {
					match := slices.ContainsFunc(mset.cfg.Subjects, func(subj string) bool {
						return SubjectsCollide(subj, scheduleTarget)
					})
					if !match {
						apiErr := NewJSMessageSchedulesTargetInvalidError()
						if canRespond {
							resp.PubAck = &PubAck{Stream: name}
							resp.Error = apiErr
							b, _ := json.Marshal(resp)
							outq.sendMsg(reply, b)
						}
						return apiErr
					}

					// Add a rollup sub header if it doesn't already exist.
					// Otherwise, it must exist already as a rollup on the subject.
					if rollup := getRollup(hdr); rollup == _EMPTY_ {
						hdr = genHeader(hdr, JSMsgRollup, JSMsgRollupSubject)
					} else if rollup != JSMsgRollupSubject {
						apiErr := NewJSMessageSchedulesRollupInvalidError()
						if canRespond {
							resp.PubAck = &PubAck{Stream: name}
							resp.Error = apiErr
							b, _ := json.Marshal(resp)
							outq.sendMsg(reply, b)
						}
						return apiErr
					}
				}
			}
		}

		// Dedupe detection. This is done at the cluster level for dedupe detection above the
		// lower layers. But we still need to pull out the msgId.
		if msgId = getMsgId(hdr); msgId != _EMPTY_ {
			// Do real check only if not clustered or traceOnly flag is set.
			if canConsistencyCheck {
				var seq uint64
				mset.ddMu.Lock()
				dde := mset.checkMsgId(msgId)
				if dde != nil {
					seq = dde.seq
				}
				mset.ddMu.Unlock()
				if seq > 0 {
					if canRespond {
						response := append(pubAck, strconv.FormatUint(dde.seq, 10)...)
						response = append(response, ",\"duplicate\": true}"...)
						outq.sendMsg(reply, response)
					}
					return errMsgIdDuplicate
				}
			}
		}

		// Expected last msgId.
		if lmsgId := getExpectedLastMsgId(hdr); lmsgId != _EMPTY_ {
			if lmsgId != mset.lmsgId {
				last := mset.lmsgId
				bumpCLFS()
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = NewJSStreamWrongLastMsgIDError(last)
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return fmt.Errorf("last msgid mismatch: %q vs %q", lmsgId, last)
			}
		}
		// Check for any rollups.
		if rollup := getRollup(hdr); rollup != _EMPTY_ {
			if canConsistencyCheck && (!mset.cfg.AllowRollup || mset.cfg.DenyPurge) {
				err := errors.New("rollup not permitted")
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = NewJSStreamRollupFailedError(err)
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return err
			}
			switch rollup {
			case JSMsgRollupSubject:
				rollupSub = true
			case JSMsgRollupAll:
				rollupAll = true
			default:
				if canConsistencyCheck {
					err := fmt.Errorf("rollup value invalid: %q", rollup)
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = NewJSStreamRollupFailedError(err)
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return err
				}
			}
		}
	}

	if canConsistencyCheck && incr == nil && allowMsgCounter {
		apiErr := NewJSMessageIncrMissingError()
		if canRespond {
			resp.PubAck = &PubAck{Stream: name}
			resp.Error = apiErr
			b, _ := json.Marshal(resp)
			outq.sendMsg(reply, b)
		}
		return apiErr
	}

	// Response Ack.
	var (
		response []byte
		seq      uint64
		err      error
	)

	// Apply increment for counter.
	// But only if it's allowed for this stream. This can happen when we store verbatim for a sourced stream.
	if canConsistencyCheck && incr != nil && allowMsgCounter {
		var initial big.Int
		var sources CounterSources
		var smv StoreMsg
		sm, err := store.LoadLastMsg(subject, &smv)
		if err == nil && sm != nil {
			var val CounterValue
			// Return an error if the counter is broken somehow.
			if json.Unmarshal(sm.msg, &val) != nil {
				apiErr := NewJSMessageCounterBrokenError()
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = apiErr
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return apiErr
			}
			if ncs := sliceHeader(JSMessageCounterSources, sm.hdr); len(ncs) > 0 {
				if err := json.Unmarshal(ncs, &sources); err != nil {
					apiErr := NewJSMessageCounterBrokenError()
					if canRespond {
						resp.PubAck = &PubAck{Stream: name}
						resp.Error = apiErr
						b, _ := json.Marshal(resp)
						outq.sendMsg(reply, b)
					}
					return apiErr
				}
			}
			initial.SetString(val.Value, 10)
		}
		srchdr := sliceHeader(JSStreamSource, hdr)
		if len(srchdr) > 0 {
			// This is a sourced message, so we can't apply Nats-Incr but
			// instead should just update the source count header.
			fields := strings.Split(string(srchdr), " ")
			origStream := fields[0]
			origSubj := subject
			if len(fields) >= 5 {
				origSubj = fields[4]
			}
			var val CounterValue
			if json.Unmarshal(msg, &val) != nil {
				apiErr := NewJSMessageCounterBrokenError()
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = apiErr
					b, _ := json.Marshal(resp)
					outq.sendMsg(reply, b)
				}
				return apiErr
			}
			var sourced big.Int
			sourced.SetString(val.Value, 10)
			if sources == nil {
				sources = map[string]map[string]string{}
			}
			if _, ok := sources[origStream]; !ok {
				sources[origStream] = map[string]string{}
			}
			prevVal := sources[origStream][origSubj]
			sources[origStream][origSubj] = sourced.String()
			// We will also replace the Nats-Incr header with the diff
			// between our last value from this source and this one, so
			// that the arithmetic is always correct.
			var previous big.Int
			previous.SetString(prevVal, 10)
			incr.Sub(&sourced, &previous)
			hdr = setHeader(JSMessageIncr, incr.String(), hdr)
		}
		// Now make the change.
		initial.Add(&initial, incr)
		// Generate the new payload.
		var _msg [128]byte
		msg = fmt.Appendf(_msg[:0], "{%q:%q}", "val", initial.String())
		// Write the updated source count headers.
		if len(sources) > 0 {
			nhdr, err := json.Marshal(sources)
			if err != nil {
				if canRespond {
					resp.PubAck = &PubAck{Stream: name}
					resp.Error = NewJSMessageCounterBrokenError()
					response, _ = json.Marshal(resp)
					outq.sendMsg(reply, response)
				}
				return err
			}
			hdr = setHeader(JSMessageCounterSources, string(nhdr), hdr)
		}

		// Check to see if we are over the max msg size.
		// Subtract to prevent against overflows.
		maxPayload := int64(mset.srv.getOpts().MaxPayload)
		hdrLen, msgLen := int64(len(hdr)), int64(len(msg))
		if hdrLen > maxPayload || msgLen > maxPayload-hdrLen {
			if canRespond {
				resp.PubAck = &PubAck{Stream: name}
				resp.Error = NewJSStreamMessageExceedsMaximumError()
				response, _ = json.Marshal(resp)
				outq.sendMsg(reply, response)
			}
			return ErrMaxPayload
		}
	}

	// Check to see if we are over the max msg size.
	// Subtract to prevent against overflows.
	if canConsistencyCheck && maxMsgSize >= 0 && (len(hdr) > maxMsgSize || len(msg) > maxMsgSize-len(hdr)) {
		if canRespond {
			resp.PubAck = &PubAck{Stream: name}
			resp.Error = NewJSStreamMessageExceedsMaximumError()
			response, _ = json.Marshal(resp)
			outq.sendMsg(reply, response)
		}
		return ErrMaxPayload
	}

	if canConsistencyCheck && len(hdr) > math.MaxUint16 {
		if canRespond {
			resp.PubAck = &PubAck{Stream: name}
			resp.Error = NewJSStreamHeaderExceedsMaximumError()
			response, _ = json.Marshal(resp)
			outq.sendMsg(reply, response)
		}
		return ErrMaxPayload
	}

	// Check to see if we have exceeded our limits.
	// Don't error and log/stepdown if we're tracing when clustered.
	if !isClustered && js.limitsExceeded(stype) {
		s.resourcesExceededError(stype)
		if canRespond {
			resp.PubAck = &PubAck{Stream: name}
			resp.Error = NewJSInsufficientResourcesError()
			response, _ = json.Marshal(resp)
			outq.sendMsg(reply, response)
		}
		// Stepdown regardless.
		if node := mset.node; node != nil {
			node.StepDown()
		}
		return NewJSInsufficientResourcesError()
	}

	var noInterest bool

	// If we are interest based retention and have no consumers then we can skip.
	if interestRetention {
		mset.clsMu.RLock()
		noInterest = numConsumers == 0 || mset.csl == nil || !mset.csl.HasInterest(subject)
		mset.clsMu.RUnlock()
	}

	// Grab timestamp if not already set.
	if ts == 0 && lseq > 0 {
		ts = time.Now().UnixNano()
	}

	mt.updateJetStreamEvent(subject, noInterest)
	if traceOnly {
		return nil
	}

	// Skip msg here.
	if noInterest {
		mset.lseq = store.SkipMsg()
		mset.lmsgId = msgId
		// If we have a msgId make sure to save.
		if msgId != _EMPTY_ {
			mset.storeMsgId(&ddentry{msgId, mset.lseq, ts})
		}
		if canRespond {
			response = append(pubAck, strconv.FormatUint(mset.lseq, 10)...)
			if batchId != _EMPTY_ {
				response = append(response, fmt.Sprintf(",\"batch\":%q,\"count\":%d}", batchId, batchSeq)...)
			} else {
				response = append(response, '}')
			}
			outq.sendMsg(reply, response)
		}
		return nil
	}

	// If here we will attempt to store the message.
	// Assume this will succeed.
	olmsgId := mset.lmsgId
	mset.lmsgId = msgId
	mset.lseq++
	tierName := mset.tier

	// Republish state if needed.
	var tsubj string
	var tlseq uint64
	var thdrsOnly bool
	if mset.tr != nil {
		tsubj, _ = mset.tr.Match(subject)
		if mset.cfg.RePublish != nil {
			thdrsOnly = mset.cfg.RePublish.HeadersOnly
		}
	}
	republish := tsubj != _EMPTY_ && isLeader

	// If we are republishing grab last sequence for this exact subject. Aids in gap detection for lightweight clients.
	if republish {
		var smv StoreMsg
		if sm, _ := store.LoadLastMsg(subject, &smv); sm != nil {
			tlseq = sm.seq
		}
	}

	// If clustered this was already checked and we do not want to check here and possibly introduce skew.
	// Don't error and log if we're tracing when clustered.
	if !isClustered {
		if exceeded, err := jsa.wouldExceedLimits(stype, tierName, mset.cfg.Replicas, subject, hdr, msg); exceeded {
			if err == nil {
				err = NewJSAccountResourcesExceededError()
			}
			s.RateLimitWarnf("JetStream %s resource limits exceeded for account: %q", strings.ToLower(stype.String()), accName)
			if canRespond {
				resp.PubAck = &PubAck{Stream: name}
				resp.Error = err
				response, _ = json.Marshal(resp)
				outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
			}
			return err
		}
	}

	// Find the message TTL if any.
	ttl, err := getMessageTTL(hdr)
	if err != nil {
		if canRespond {
			resp.PubAck = &PubAck{Stream: name}
			resp.Error = NewJSMessageTTLInvalidError()
			response, _ = json.Marshal(resp)
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
		}
		return err
	}

	// If subject delete markers are used, ensure message TTL is that at minimum.
	// Otherwise, subject delete markers could be missed if one already exists for this subject.
	// MaxMsgsPer=1 is an exception, because we'll only ever have one message.
	if ttl > 0 && mset.cfg.SubjectDeleteMarkerTTL > 0 && mset.cfg.MaxMsgsPer != 1 {
		if minTtl := int64(mset.cfg.SubjectDeleteMarkerTTL.Seconds()); ttl < minTtl {
			ttl = minTtl
			hdr = removeHeaderIfPresent(hdr, JSMessageTTL)
			hdr = genHeader(hdr, JSMessageTTL, strconv.FormatInt(ttl, 10))
		}
	}

	// Store actual msg.
	if lseq == 0 && ts == 0 {
		seq, ts, err = store.StoreMsg(subject, hdr, msg, ttl)
	} else {
		// Make sure to take into account any message assignments that we had to skip (clfs).
		seq = lseq + 1 - clfs
		// Check for preAcks and the need to clear it.
		if mset.hasAllPreAcks(seq, subject) {
			mset.clearAllPreAcks(seq)
		}
		err = store.StoreRawMsg(subject, hdr, msg, seq, ts, ttl)
	}

	if err != nil {
		if isPermissionError(err) {
			// messages in block cache could be lost in the worst case.
			// In the clustered mode it is very highly unlikely as a result of replication.
			go mset.srv.DisableJetStream()
			mset.srv.Warnf("Filesystem permission denied while writing msg, disabling JetStream: %v", err)
			return err
		}
		// If we did not succeed put those values back and increment clfs in case we are clustered.
		var state StreamState
		mset.store.FastState(&state)
		mset.lseq = state.LastSeq
		mset.lmsgId = olmsgId
		bumpCLFS()

		switch err {
		case ErrMaxMsgs, ErrMaxBytes, ErrMaxMsgsPerSubject, ErrMsgTooLarge:
			s.RateLimitDebugf("JetStream failed to store a msg on stream '%s > %s': %v", accName, name, err)
		case ErrStoreClosed:
		default:
			s.Errorf("JetStream failed to store a msg on stream '%s > %s': %v", accName, name, err)
		}

		if canRespond {
			resp.PubAck = &PubAck{Stream: name}
			resp.Error = NewJSStreamStoreFailedError(err, Unless(err))
			response, _ = json.Marshal(resp)
			outq.sendMsg(reply, response)
		}
		return err
	}

	// If here we succeeded in storing the message.

	// If we have a msgId make sure to save.
	// This will replace our estimate from the cluster layer if we are clustered.
	if msgId != _EMPTY_ {
		mset.ddMu.Lock()
		if isClustered && isLeader && mset.ddmap != nil {
			if dde := mset.ddmap[msgId]; dde != nil {
				dde.seq, dde.ts = seq, ts
			} else {
				mset.storeMsgIdLocked(&ddentry{msgId, seq, ts})
			}
		} else {
			// R1 or not leader..
			mset.storeMsgIdLocked(&ddentry{msgId, seq, ts})
		}
		mset.ddMu.Unlock()
	}

	// No errors, this is the normal path.
	if rollupSub {
		mset.purgeLocked(&JSApiStreamPurgeRequest{Subject: subject, Keep: 1}, false)
	} else if rollupAll {
		mset.purgeLocked(&JSApiStreamPurgeRequest{Keep: 1}, false)
	} else if scheduleNext := sliceHeader(JSScheduleNext, hdr); len(scheduleNext) > 0 && bytesToString(scheduleNext) == JSScheduleNextPurge {
		// Purge the message schedule.
		scheduler := getMessageScheduler(hdr)
		if scheduler != _EMPTY_ {
			mset.purgeLocked(&JSApiStreamPurgeRequest{Subject: scheduler}, false)
		}
	}

	// Check for republish.
	if republish {
		const ht = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\nNats-Last-Sequence: %d\r\n\r\n"
		const htho = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\nNats-Last-Sequence: %d\r\nNats-Msg-Size: %d\r\n\r\n"
		// When adding to existing headers, will use the fmt.Append version so this skips the headers from above.
		const hoff = 10

		tsStr := time.Unix(0, ts).UTC().Format(time.RFC3339Nano)
		var rpMsg []byte
		if len(hdr) == 0 {
			if !thdrsOnly {
				hdr = fmt.Appendf(nil, ht, name, subject, seq, tsStr, tlseq)
				rpMsg = copyBytes(msg)
			} else {
				hdr = fmt.Appendf(nil, htho, name, subject, seq, tsStr, tlseq, len(msg))
			}
		} else {
			// use hdr[:end:end] to make sure as we add we copy the original hdr.
			end := len(hdr) - LEN_CR_LF
			if !thdrsOnly {
				hdr = fmt.Appendf(hdr[:end:end], ht[hoff:], name, subject, seq, tsStr, tlseq)
				rpMsg = copyBytes(msg)
			} else {
				hdr = fmt.Appendf(hdr[:end:end], htho[hoff:], name, subject, seq, tsStr, tlseq, len(msg))
			}
		}
		outq.send(newJSPubMsg(tsubj, _EMPTY_, _EMPTY_, hdr, rpMsg, nil, seq))
	}

	// Send response here.
	if canRespond {
		response = append(pubAck, strconv.FormatUint(seq, 10)...)
		if batchId != _EMPTY_ {
			response = append(response, fmt.Sprintf(",\"batch\":%q,\"count\":%d}", batchId, batchSeq)...)
		} else if allowMsgCounter {
			var counter CounterValue
			json.Unmarshal(msg, &counter)
			response = append(response, fmt.Sprintf(",\"val\":%q}", counter.Value)...)
		} else {
			response = append(response, '}')
		}
		outq.sendMsg(reply, response)
	}

	// Signal consumers for new messages.
	if numConsumers > 0 {
		mset.sigq.push(newCMsg(subject, seq))
		select {
		case mset.sch <- struct{}{}:
		default:
		}
	}

	return nil
}

// processJetStreamBatchMsg processes a JetStream message that's part of an atomic batch publish.
// Handles constraints around the batch, storing messages, doing consistency checks, and performing the commit.
func (mset *stream) processJetStreamBatchMsg(batchId, subject, reply string, hdr, msg []byte, mt *msgTrace) (retErr error) {
	// For possible error response.
	var response []byte

	mset.mu.RLock()
	canRespond := !mset.cfg.NoAck && len(reply) > 0
	name, stype := mset.cfg.Name, mset.cfg.Storage
	discard, discardNewPer, maxMsgs, maxMsgsPer, maxBytes := mset.cfg.Discard, mset.cfg.DiscardNewPer, mset.cfg.MaxMsgs, mset.cfg.MaxMsgsPer, mset.cfg.MaxBytes
	s, js, jsa, st, r, tierName, outq, node := mset.srv, mset.js, mset.jsa, mset.cfg.Storage, mset.cfg.Replicas, mset.tier, mset.outq, mset.node
	maxMsgSize, lseq := int(mset.cfg.MaxMsgSize), mset.lseq
	isLeader, isClustered, isSealed, allowRollup, denyPurge, allowTTL, allowMsgCounter, allowMsgSchedules, allowAtomicPublish := mset.isLeader(), mset.isClustered(), mset.cfg.Sealed, mset.cfg.AllowRollup, mset.cfg.DenyPurge, mset.cfg.AllowMsgTTL, mset.cfg.AllowMsgCounter, mset.cfg.AllowMsgSchedules, mset.cfg.AllowAtomicPublish
	mset.mu.RUnlock()

	// If message tracing (with message delivery), we will need to send the
	// event on exit in case there was an error (if message was not proposed).
	// Otherwise, the event will be sent from processJetStreamMsg when
	// invoked by the leader (from applyStreamEntries).
	if mt != nil {
		defer func() {
			if retErr != nil {
				mt.sendEventFromJetStream(retErr)
			}
		}()
	}

	// Check that we are the leader. This can be false if we have scaled up from an R1 that had inbound queued messages.
	if !isLeader {
		return NewJSClusterNotLeaderError()
	}

	// Bail here if sealed.
	if isSealed {
		var resp = JSPubAckResponse{PubAck: &PubAck{Stream: mset.name()}, Error: NewJSStreamSealedError()}
		b, _ := json.Marshal(resp)
		mset.outq.sendMsg(reply, b)
		return NewJSStreamSealedError()
	}

	// Check here pre-emptively if we have exceeded this server limits.
	if js.limitsExceeded(stype) {
		s.resourcesExceededError(stype)
		if canRespond {
			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: NewJSInsufficientResourcesError()})
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
		}
		// Stepdown regardless.
		if node := mset.raftNode(); node != nil {
			node.StepDown()
		}
		return NewJSInsufficientResourcesError()
	}

	// Check here pre-emptively if we have exceeded our account limits.
	if exceeded, err := jsa.wouldExceedLimits(st, tierName, r, subject, hdr, msg); exceeded {
		if err == nil {
			err = NewJSAccountResourcesExceededError()
		}
		s.RateLimitWarnf("JetStream account limits exceeded for '%s': %s", jsa.acc().GetName(), err.Error())
		if canRespond {
			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
			resp.Error = err
			response, _ = json.Marshal(resp)
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
		}
		return err
	}

	// Check msgSize if we have a limit set there. Again this works if it goes through but better to be pre-emptive.
	// Subtract to prevent against overflows.
	if maxMsgSize >= 0 && (len(hdr) > maxMsgSize || len(msg) > maxMsgSize-len(hdr)) {
		err := fmt.Errorf("JetStream message size exceeds limits for '%s > %s'", jsa.acc().Name, mset.cfg.Name)
		s.RateLimitWarnf("%s", err.Error())
		if canRespond {
			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
			resp.Error = NewJSStreamMessageExceedsMaximumError()
			response, _ = json.Marshal(resp)
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
		}
		return err
	}

	if !allowAtomicPublish {
		err := NewJSAtomicPublishDisabledError()
		if canRespond {
			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
		}
		return err
	}

	// Batch ID is too long.
	if len(batchId) > 64 {
		err := NewJSAtomicPublishInvalidBatchIDError()
		if canRespond {
			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
		}
		return err
	}

	batchSeq, exists := getBatchSequence(hdr)
	if !exists {
		err := NewJSAtomicPublishMissingSeqError()
		if canRespond {
			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
		}
		return err
	}
	commit := len(sliceHeader(JSBatchCommit, hdr)) != 0

	mset.mu.Lock()
	if mset.batches == nil {
		mset.batches = &batching{
			group: make(map[string]*batchGroup, 1),
		}
	}
	batches := mset.batches
	mset.mu.Unlock()

	respondIncompleteBatch := func() error {
		err := NewJSAtomicPublishIncompleteBatchError()
		if canRespond {
			buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
		}
		return err
	}

	// Get batch.
	batches.mu.Lock()
	b, ok := batches.group[batchId]
	if !ok {
		if batchSeq != 1 {
			batches.mu.Unlock()
			maxBatchSize := streamMaxBatchSize
			opts := s.getOpts()
			if opts.JetStreamLimits.MaxBatchSize > 0 {
				maxBatchSize = opts.JetStreamLimits.MaxBatchSize
			}
			if batchSeq > uint64(maxBatchSize) {
				err := NewJSAtomicPublishTooLargeBatchError(maxBatchSize)
				if canRespond {
					buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
					outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
				}
				return err
			}
			return respondIncompleteBatch()
		}

		// Limits.
		maxInflightPerStream := streamMaxBatchInflightPerStream
		maxInflightTotal := streamMaxBatchInflightTotal
		opts := s.getOpts()
		if opts.JetStreamLimits.MaxBatchInflightPerStream > 0 {
			maxInflightPerStream = opts.JetStreamLimits.MaxBatchInflightPerStream
		}
		if opts.JetStreamLimits.MaxBatchInflightTotal > 0 {
			maxInflightTotal = opts.JetStreamLimits.MaxBatchInflightTotal
		}

		// Confirm we can facilitate an additional batch.
		if len(batches.group)+1 > maxInflightPerStream {
			batches.mu.Unlock()
			return respondIncompleteBatch()
		}

		// Confirm we'll not exceed the server limit.
		if globalInflightBatches.Add(1) > int32(maxInflightTotal) {
			globalInflightBatches.Add(-1)
			batches.mu.Unlock()
			return respondIncompleteBatch()
		}

		var err error
		if b, err = batches.newBatchGroup(mset, batchId); err != nil {
			globalInflightBatches.Add(-1)
			batches.mu.Unlock()
			return respondIncompleteBatch()
		}
		batches.group[batchId] = b
	}

	// Detect gaps.
	b.lseq++
	if b.lseq != batchSeq {
		b.cleanupLocked(batchId, batches)
		batches.mu.Unlock()
		mset.sendStreamBatchAbandonedAdvisory(batchId, BatchIncomplete)
		return respondIncompleteBatch()
	}

	// Confirm the batch doesn't exceed the allowed size.
	maxSize := streamMaxBatchSize
	if maxBatchSize := s.getOpts().JetStreamLimits.MaxBatchSize; maxBatchSize > 0 {
		maxSize = maxBatchSize
	}
	if batchSeq > uint64(maxSize) {
		b.cleanupLocked(batchId, batches)
		batches.mu.Unlock()
		mset.sendStreamBatchAbandonedAdvisory(batchId, BatchLarge)
		err := NewJSAtomicPublishTooLargeBatchError(maxSize)
		if canRespond {
			buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: err})
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
		}
		return err
	}

	// Persist, but optimize if we're committing because we already know last.
	// If the underlying store is file-based we need to persist everything to survive hard kills, because we're R1.
	if !commit || b.store.Type() == FileStorage {
		seq, _, err := b.store.StoreMsg(subject, hdr, msg, 0)
		if err != nil || seq != batchSeq {
			b.cleanupLocked(batchId, batches)
			batches.mu.Unlock()
			mset.sendStreamBatchAbandonedAdvisory(batchId, BatchIncomplete)
			return respondIncompleteBatch()
		}
	}
	if !commit {
		batches.mu.Unlock()
		// Send empty ack to let them know we've persisted the data prior to commit.
		if canRespond {
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, nil, nil, 0))
		}
		return nil
	}

	// Ensure the batch is prepared for the commit and will not be cleaned up while committing.
	if !b.readyForCommit() {
		// Don't do cleanup, this is already done.
		batches.mu.Unlock()
		mset.sendStreamBatchAbandonedAdvisory(batchId, BatchTimeout)
		return respondIncompleteBatch()
	}

	// Proceed with proposing the batch.
	ts := time.Now().UnixNano()

	// Need to hold this lock even if we're not clustered, because we'll
	// be accessing state that requires this lock (even if it's empty).
	mset.clMu.Lock()

	// If not clustered, set up the last sequence.
	if !isClustered {
		mset.clseq, mset.clfs = mset.lseq, 0
	}

	// We only use mset.clseq for clustering and in case we run ahead of actual commits.
	// Check if we need to set initial value here
	if isClustered && (mset.clseq == 0 || mset.clseq < lseq+mset.clfs) {
		// Need to unlock and re-acquire the locks in the proper order.
		mset.clMu.Unlock()
		// Locking order is stream -> batchMu -> clMu
		mset.mu.RLock()
		batch := mset.batchApply
		var batchCount uint64
		if batch != nil {
			batch.mu.Lock()
			batchCount = batch.count
		}
		mset.clMu.Lock()
		// Re-capture
		lseq = mset.lseq
		mset.clseq = lseq + mset.clfs + batchCount
		// Keep hold of the mset.clMu, but unlock the others.
		if batch != nil {
			batch.mu.Unlock()
		}
		mset.mu.RUnlock()
	}

	rollback := func(seq uint64) {
		if isClustered {
			// TODO(mvv): reset in-memory expected header maps
			mset.clseq -= seq - 1
		}
		mset.clMu.Unlock()
	}

	errorOnUnsupported := func(seq uint64, header string) *ApiError {
		apiErr := NewJSAtomicPublishUnsupportedHeaderBatchError(header)
		rollback(seq)
		b.cleanupLocked(batchId, batches)
		batches.mu.Unlock()
		if canRespond {
			buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: apiErr})
			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
		}
		return apiErr
	}

	var (
		entries []*Entry
		bsubj   string
		bhdr    []byte
		bmsg    []byte
		apiErr  *ApiError
		err     error
		smv     StoreMsg
		sm      *StoreMsg
		sz      int
	)

	diff := &batchStagedDiff{}
	for seq := uint64(1); seq <= batchSeq; seq++ {
		if seq == batchSeq && b.store.Type() != FileStorage {
			bsubj, bhdr, bmsg = subject, hdr, msg
		} else if sm, err = b.store.LoadMsg(seq, &smv); sm != nil && err == nil {
			bsubj, bhdr, bmsg = sm.subj, sm.hdr, sm.msg
		} else {
			rollback(seq)
			b.cleanupLocked(batchId, batches)
			batches.mu.Unlock()
			return respondIncompleteBatch()
		}

		// Reject unsupported headers.
		if msgId := getMsgId(bhdr); msgId != _EMPTY_ {
			return errorOnUnsupported(seq, JSMsgId)
		}
		if getExpectedLastMsgId(hdr) != _EMPTY_ {
			return errorOnUnsupported(seq, JSExpectedLastMsgId)
		}

		if bhdr, bmsg, _, apiErr, err = checkMsgHeadersPreClusteredProposal(diff, mset, subject, bhdr, bmsg, false, name, jsa, allowRollup, denyPurge, allowTTL, allowMsgCounter, allowMsgSchedules, discard, discardNewPer, maxMsgSize, maxMsgs, maxMsgsPer, maxBytes); err != nil {
			rollback(seq)
			b.cleanupLocked(batchId, batches)
			batches.mu.Unlock()
			if canRespond {
				buf, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: apiErr})
				outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, buf, nil, 0))
			}
			return err
		}

		if isClustered {
			var _reply string
			isCommit := seq == batchSeq
			if isCommit {
				_reply = reply
			}
			esm := encodeStreamMsgAllowCompressAndBatch(bsubj, _reply, bhdr, bmsg, mset.clseq, ts, false, batchId, seq, isCommit)
			entries = append(entries, newEntry(EntryNormal, esm))
			sz += len(esm)
		}
		mset.clseq++
	}

	// Commit batch.
	if !isClustered {
		// Reset, we only used this to do the batching checks.
		mset.clseq, mset.clfs = 0, 0
		mset.clMu.Unlock()

		// Ensure the whole batch is fully isolated, and reads
		// can only happen after the full batch is committed.
		mset.mu.Lock()
		for seq := uint64(1); seq <= batchSeq; seq++ {
			if seq == batchSeq && b.store.Type() != FileStorage {
				bsubj, bhdr, bmsg = subject, hdr, msg
			} else if sm, err = b.store.LoadMsg(seq, &smv); sm != nil && err == nil {
				bsubj, bhdr, bmsg = sm.subj, sm.hdr, sm.msg
			} else {
				// Should not happen, we've already checked this message existed while doing consistency checks.
				// We'll just exit here, the batch is already inconsistent without the message at this sequence.
				// No use in trying to still store the rest.
				break
			}
			var _reply string
			if seq == batchSeq {
				_reply = reply
			}
			mset.processJetStreamMsg(bsubj, _reply, bhdr, bmsg, 0, 0, mt, false, false)
		}
		mset.mu.Unlock()
	} else {
		// Do a single multi proposal. This ensures we get to push all entries to the proposal queue in-order
		// and not interleaved with other proposals.
		diff.commit(mset)
		if err := node.ProposeMulti(entries); err == nil {
			mset.trackReplicationTraffic(node, sz, r)
		} else {
			// TODO(mvv): reset in-memory expected header maps
			mset.clseq -= batchSeq
		}

		// Check to see if we are being overrun.
		// TODO(dlc) - Make this a limit where we drop messages to protect ourselves, but allow to be configured.
		if mset.clseq-(lseq+mset.clfs) > streamLagWarnThreshold {
			lerr := fmt.Errorf("JetStream stream '%s > %s' has high message lag", jsa.acc().Name, name)
			s.RateLimitWarnf("%s", lerr.Error())
		}
		mset.clMu.Unlock()
	}
	b.cleanupLocked(batchId, batches)
	batches.mu.Unlock()
	return nil
}

// Used to signal inbound message to registered consumers.
type cMsg struct {
	seq  uint64
	subj string
}

// Pool to recycle consumer bound msgs.
var cMsgPool sync.Pool

// Used to queue up consumer bound msgs for signaling.
func newCMsg(subj string, seq uint64) *cMsg {
	var m *cMsg
	cm := cMsgPool.Get()
	if cm != nil {
		m = cm.(*cMsg)
	} else {
		m = new(cMsg)
	}
	m.subj, m.seq = subj, seq

	return m
}

func (m *cMsg) returnToPool() {
	if m == nil {
		return
	}
	m.subj, m.seq = _EMPTY_, 0
	cMsgPool.Put(m)
}

// Go routine to signal consumers.
// Offloaded from stream msg processing.
func (mset *stream) signalConsumersLoop() {
	mset.mu.RLock()
	s, qch, sch, msgs := mset.srv, mset.qch, mset.sch, mset.sigq
	mset.mu.RUnlock()

	for {
		select {
		case <-s.quitCh:
			return
		case <-qch:
			return
		case <-sch:
			cms := msgs.pop()
			for _, m := range cms {
				seq, subj := m.seq, m.subj
				m.returnToPool()
				// Signal all appropriate consumers.
				mset.signalConsumers(subj, seq)
			}
			msgs.recycle(&cms)
		}
	}
}

// This will update and signal all consumers that match.
func (mset *stream) signalConsumers(subj string, seq uint64) {
	mset.clsMu.RLock()
	defer mset.clsMu.RUnlock()
	csl := mset.csl
	if csl == nil {
		return
	}
	csl.Match(subj, func(o *consumer) {
		o.processStreamSignal(seq)
	})
}

// Internal message for use by jetstream subsystem.
type jsPubMsg struct {
	dsubj string // Subject to send to, e.g. _INBOX.xxx
	reply string
	StoreMsg
	o *consumer
}

var jsPubMsgPool sync.Pool

func newJSPubMsg(dsubj, subj, reply string, hdr, msg []byte, o *consumer, seq uint64) *jsPubMsg {
	var m *jsPubMsg
	var buf []byte
	pm := jsPubMsgPool.Get()
	if pm != nil {
		m = pm.(*jsPubMsg)
		buf = m.buf[:0]
		if hdr != nil {
			hdr = append(m.hdr[:0], hdr...)
		}
	} else {
		m = new(jsPubMsg)
	}
	// When getting something from a pool it is critical that all fields are
	// initialized. Doing this way guarantees that if someone adds a field to
	// the structure, the compiler will fail the build if this line is not updated.
	(*m) = jsPubMsg{dsubj, reply, StoreMsg{subj, hdr, msg, buf, seq, 0}, o}

	return m
}

// Gets a jsPubMsg from the pool.
func getJSPubMsgFromPool() *jsPubMsg {
	pm := jsPubMsgPool.Get()
	if pm != nil {
		return pm.(*jsPubMsg)
	}
	return new(jsPubMsg)
}

func (pm *jsPubMsg) returnToPool() {
	if pm == nil {
		return
	}
	pm.subj, pm.dsubj, pm.reply, pm.hdr, pm.msg, pm.o = _EMPTY_, _EMPTY_, _EMPTY_, nil, nil, nil
	if len(pm.buf) > 0 {
		pm.buf = pm.buf[:0]
	}
	if len(pm.hdr) > 0 {
		pm.hdr = pm.hdr[:0]
	}
	jsPubMsgPool.Put(pm)
}

func (pm *jsPubMsg) size() int {
	if pm == nil {
		return 0
	}
	return len(pm.dsubj) + len(pm.reply) + len(pm.hdr) + len(pm.msg)
}

// Queue of *jsPubMsg for sending internal system messages.
type jsOutQ struct {
	*ipQueue[*jsPubMsg]
}

func (q *jsOutQ) sendMsg(subj string, msg []byte) {
	if q != nil {
		q.send(newJSPubMsg(subj, _EMPTY_, _EMPTY_, nil, msg, nil, 0))
	}
}

func (q *jsOutQ) send(msg *jsPubMsg) {
	if q == nil || msg == nil {
		return
	}
	q.push(msg)
}

func (q *jsOutQ) unregister() {
	if q == nil {
		return
	}
	q.ipQueue.unregister()
}

// StoredMsg is for raw access to messages in a stream.
type StoredMsg struct {
	Subject  string    `json:"subject"`
	Sequence uint64    `json:"seq"`
	Header   []byte    `json:"hdrs,omitempty"`
	Data     []byte    `json:"data,omitempty"`
	Time     time.Time `json:"time"`
}

// This is similar to system semantics but did not want to overload the single system sendq,
// or require system account when doing simple setup with jetstream.
func (mset *stream) setupSendCapabilities() {
	mset.mu.Lock()
	defer mset.mu.Unlock()
	if mset.outq != nil {
		return
	}
	qname := fmt.Sprintf("[ACC:%s] stream '%s' sendQ", mset.acc.Name, mset.cfg.Name)
	mset.outq = &jsOutQ{newIPQueue[*jsPubMsg](mset.srv, qname)}
	go mset.internalLoop()
}

// Returns the associated account name.
func (mset *stream) accName() string {
	if mset == nil {
		return _EMPTY_
	}
	mset.mu.RLock()
	acc := mset.acc
	mset.mu.RUnlock()
	return acc.Name
}

// Name returns the stream name.
func (mset *stream) name() string {
	return mset.nameLocked(true)
}

func (mset *stream) nameLocked(needLock bool) string {
	if mset == nil {
		return _EMPTY_
	}
	if needLock {
		mset.mu.RLock()
		defer mset.mu.RUnlock()
	}
	return mset.cfg.Name
}

func (mset *stream) internalLoop() {
	mset.mu.RLock()
	setGoRoutineLabels(pprofLabels{
		"account": mset.acc.Name,
		"stream":  mset.cfg.Name,
	})
	s := mset.srv
	c := s.createInternalJetStreamClient()
	c.registerWithAccount(mset.acc)
	defer c.closeConnection(ClientClosed)
	outq, qch, msgs, gets := mset.outq, mset.qch, mset.msgs, mset.gets

	// For the ack msgs queue for interest retention.
	var (
		amch chan struct{}
		ackq *ipQueue[uint64]
	)
	if mset.ackq != nil {
		ackq, amch = mset.ackq, mset.ackq.ch
	}
	mset.mu.RUnlock()

	// Raw scratch buffer.
	// This should be rarely used now so can be smaller.
	var _r [1024]byte

	// To optimize for not converting a string to a []byte slice.
	var (
		subj  [256]byte
		dsubj [256]byte
		rply  [256]byte
		szb   [10]byte
		hdb   [10]byte
	)

	for {
		select {
		case <-outq.ch:
			pms := outq.pop()
			for _, pm := range pms {
				c.pa.subject = append(dsubj[:0], pm.dsubj...)
				c.pa.deliver = append(subj[:0], pm.subj...)
				c.pa.size = len(pm.msg) + len(pm.hdr)
				c.pa.szb = append(szb[:0], strconv.Itoa(c.pa.size)...)
				if len(pm.reply) > 0 {
					c.pa.reply = append(rply[:0], pm.reply...)
				} else {
					c.pa.reply = nil
				}

				// If we have an underlying buf that is the wire contents for hdr + msg, else construct on the fly.
				var msg []byte
				if len(pm.buf) > 0 {
					msg = pm.buf
				} else {
					if len(pm.hdr) > 0 {
						msg = pm.hdr
						if len(pm.msg) > 0 {
							msg = _r[:0]
							msg = append(msg, pm.hdr...)
							msg = append(msg, pm.msg...)
						}
					} else if len(pm.msg) > 0 {
						// We own this now from a low level buffer perspective so can use directly here.
						msg = pm.msg
					}
				}

				if len(pm.hdr) > 0 {
					c.pa.hdr = len(pm.hdr)
					c.pa.hdb = []byte(strconv.Itoa(c.pa.hdr))
					c.pa.hdb = append(hdb[:0], strconv.Itoa(c.pa.hdr)...)
				} else {
					c.pa.hdr = -1
					c.pa.hdb = nil
				}

				msg = append(msg, _CRLF_...)

				didDeliver, _ := c.processInboundClientMsg(msg)
				c.pa.szb, c.pa.subject, c.pa.deliver = nil, nil, nil

				// Check to see if this is a delivery for a consumer and
				// we failed to deliver the message. If so alert the consumer.
				if pm.o != nil && pm.seq > 0 && !didDeliver {
					pm.o.didNotDeliver(pm.seq, pm.dsubj)
				}
				pm.returnToPool()
			}
			// TODO: Move in the for-loop?
			c.flushClients(0)
			outq.recycle(&pms)
		case <-msgs.ch:
			// This can possibly change now so needs to be checked here.
			isClustered := mset.IsClustered()
			ims := msgs.pop()
			for _, im := range ims {
				// If we are clustered we need to propose this message to the underlying raft group.
				if batchId := getBatchId(im.hdr); batchId != _EMPTY_ {
					mset.processJetStreamBatchMsg(batchId, im.subj, im.rply, im.hdr, im.msg, im.mt)
				} else if isClustered {
					mset.processClusteredInboundMsg(im.subj, im.rply, im.hdr, im.msg, im.mt, false)
				} else {
					mset.processJetStreamMsg(im.subj, im.rply, im.hdr, im.msg, 0, 0, im.mt, false, true)
				}
				im.returnToPool()
			}
			msgs.recycle(&ims)
		case <-gets.ch:
			dgs := gets.pop()
			for _, dg := range dgs {
				mset.getDirectRequest(&dg.req, dg.reply)
				dgPool.Put(dg)
			}
			gets.recycle(&dgs)

		case <-amch:
			seqs := ackq.pop()
			for _, seq := range seqs {
				mset.ackMsg(nil, seq)
			}
			ackq.recycle(&seqs)
		case <-qch:
			return
		case <-s.quitCh:
			return
		}
	}
}

// Used to break consumers out of their monitorConsumer go routines.
func (mset *stream) resetAndWaitOnConsumers() {
	mset.mu.RLock()
	consumers := make([]*consumer, 0, len(mset.consumers))
	for _, o := range mset.consumers {
		consumers = append(consumers, o)
	}
	mset.mu.RUnlock()

	for _, o := range consumers {
		if node := o.raftNode(); node != nil {
			node.StepDown()
			node.Stop()
		}
		if o.isMonitorRunning() {
			o.signalMonitorQuit()
			o.monitorWg.Wait()
		}
	}
}

// Internal function to delete a stream.
func (mset *stream) delete() error {
	if mset == nil {
		return nil
	}
	return mset.stop(true, true)
}

// Internal function to stop or delete the stream.
func (mset *stream) stop(deleteFlag, advisory bool) error {
	mset.mu.RLock()
	js, jsa, name, offlineReason := mset.js, mset.jsa, mset.cfg.Name, mset.offlineReason
	mset.mu.RUnlock()

	if jsa == nil {
		return NewJSNotEnabledForAccountError()
	}

	isShuttingDown := js.isShuttingDown()

	// Remove from our account map first.
	jsa.mu.Lock()
	// Preserve in the account if it's marked offline, to have it remain queryable.
	if deleteFlag || offlineReason == _EMPTY_ {
		delete(jsa.streams, name)
	}
	accName := jsa.account.Name
	jsa.mu.Unlock()

	// Kick monitor and collect consumers first.
	mset.mu.Lock()

	// Mark closed.
	mset.closed.Store(true)

	// Signal to the monitor loop.
	// Can't use qch here.
	if mset.mqch != nil {
		close(mset.mqch)
		mset.mqch = nil
	}

	// Stop responding to sync requests.
	mset.stopClusterSubs()
	// Unsubscribe from direct stream.
	mset.unsubscribeToStream(true, isShuttingDown)

	// Our info sub if we spun it up.
	if mset.infoSub != nil {
		mset.srv.sysUnsubscribe(mset.infoSub)
		mset.infoSub = nil
	}

	// Clean up consumers.
	var obs []*consumer
	for _, o := range mset.consumers {
		obs = append(obs, o)
	}
	// Preserve the consumers if it's marked offline, to have them remain queryable.
	if deleteFlag || offlineReason == _EMPTY_ {
		mset.clsMu.Lock()
		mset.consumers, mset.cList, mset.csl = nil, nil, nil
		mset.clsMu.Unlock()
	}

	// Check if we are a mirror.
	if mset.mirror != nil && mset.mirror.sub != nil {
		mset.unsubscribe(mset.mirror.sub)
		mset.mirror.sub = nil
		mset.removeInternalConsumer(mset.mirror)
	}
	// Now check for sources.
	if len(mset.sources) > 0 {
		for _, si := range mset.sources {
			mset.cancelSourceConsumer(si.iname)
		}
	}
	mset.mu.Unlock()

	for _, o := range obs {
		if !o.isClosed() {
			// Third flag says do not broadcast a signal.
			// TODO(dlc) - If we have an err here we don't want to stop
			// but should we log?
			o.stopWithFlags(deleteFlag, deleteFlag, false, advisory)
			if !isShuttingDown {
				o.signalMonitorQuit()
				o.monitorWg.Wait()
			}
		}
	}

	mset.mu.Lock()
	// Send stream delete advisory after the consumers.
	if deleteFlag && advisory {
		mset.sendDeleteAdvisoryLocked()
	}

	// Quit channel, do this after sending the delete advisory
	if mset.qch != nil {
		close(mset.qch)
		mset.qch = nil
	}

	// Cluster cleanup
	var sa *streamAssignment
	if n := mset.node; n != nil {
		if deleteFlag {
			n.Delete()
			sa = mset.sa
		} else if !isShuttingDown {
			// Stop Raft, unless JetStream is already shutting down, in which case they'll be stopped separately.
			n.Stop()
		}
	}

	// Cleanup duplicate timer if running.
	mset.ddMu.Lock()
	if mset.ddtmr != nil {
		mset.ddtmr.Stop()
		mset.ddtmr = nil
		mset.ddmap = nil
		mset.ddarr = nil
		mset.ddindex = 0
	}
	mset.ddMu.Unlock()

	sysc := mset.sysc
	mset.sysc = nil

	if deleteFlag {
		// Unregistering ipQueues do not prevent them from push/pop
		// just will remove them from the central monitoring map
		mset.msgs.unregister()
		mset.ackq.unregister()
		mset.outq.unregister()
		mset.sigq.unregister()
		mset.smsgs.unregister()
	}

	// Snapshot store.
	store := mset.store
	c := mset.client

	// Clustered cleanup.
	mset.mu.Unlock()

	// Check if the stream assignment has the group node specified.
	// We need this cleared for if the stream gets reassigned here.
	if sa != nil {
		js.mu.Lock()
		if sa.Group != nil {
			sa.Group.node = nil
		}
		js.mu.Unlock()
	}

	if c != nil {
		c.closeConnection(ClientClosed)
	}

	if sysc != nil {
		sysc.closeConnection(ClientClosed)
	}

	if deleteFlag {
		// cleanup directories after the stream
		accDir := filepath.Join(js.config.StoreDir, accName)
		if store != nil {
			// Ignore errors.
			store.Delete(false)
		} else {
			streamDir := filepath.Join(accDir, streamsDir)
			os.RemoveAll(filepath.Join(streamDir, name))
		}
		// Release any resources.
		js.releaseStreamResources(&mset.cfg)
		// Do cleanup in separate go routine similar to how fs will use purge here..
		go func() {
			// no op if not empty
			os.Remove(filepath.Join(accDir, streamsDir))
			os.Remove(accDir)
		}()
	} else if store != nil {
		// Ignore errors.
		store.Stop()
	}

	return nil
}

func (mset *stream) getMsg(seq uint64) (*StoredMsg, error) {
	var smv StoreMsg
	sm, err := mset.store.LoadMsg(seq, &smv)
	if err != nil {
		return nil, err
	}
	// This only used in tests directly so no need to pool etc.
	return &StoredMsg{
		Subject:  sm.subj,
		Sequence: sm.seq,
		Header:   sm.hdr,
		Data:     sm.msg,
		Time:     time.Unix(0, sm.ts).UTC(),
	}, nil
}

// getConsumers will return a copy of all the current consumers for this stream.
func (mset *stream) getConsumers() []*consumer {
	mset.clsMu.RLock()
	defer mset.clsMu.RUnlock()
	return append([]*consumer(nil), mset.cList...)
}

// Lock should be held for this one.
func (mset *stream) numPublicConsumers() int {
	return len(mset.consumers) - mset.directs
}

// This returns all consumers that are not DIRECT.
func (mset *stream) getPublicConsumers() []*consumer {
	mset.clsMu.RLock()
	defer mset.clsMu.RUnlock()

	var obs []*consumer
	for _, o := range mset.cList {
		if !o.cfg.Direct {
			obs = append(obs, o)
		}
	}
	return obs
}

// 2 minutes plus up to 30s jitter.
const (
	defaultCheckInterestStateT = 2 * time.Minute
	defaultCheckInterestStateJ = 30
)

var (
	checkInterestStateT = defaultCheckInterestStateT // Interval
	checkInterestStateJ = defaultCheckInterestStateJ // Jitter (secs)
)

// Will check for interest retention and make sure messages
// that have been acked are processed and removed.
// This will check the ack floors of all consumers, and adjust our first sequence accordingly.
func (mset *stream) checkInterestState() {
	if mset == nil || !mset.isInterestRetention() {
		// If we are limits based nothing to do.
		return
	}

	var ss StreamState
	mset.store.FastState(&ss)

	for _, o := range mset.getConsumers() {
		o.checkStateForInterestStream(&ss)
	}
}

func (mset *stream) isInterestRetention() bool {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.cfg.Retention != LimitsPolicy
}

// NumConsumers reports on number of active consumers for this stream.
func (mset *stream) numConsumers() int {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return len(mset.consumers)
}

// Lock should be held.
func (mset *stream) setConsumer(o *consumer) {
	mset.consumers[o.name] = o
	if len(o.subjf) > 0 {
		mset.numFilter++
	}
	if o.cfg.Direct {
		mset.directs++
	}
	// Now update consumers list as well
	mset.clsMu.Lock()
	mset.cList = append(mset.cList, o)
	if mset.csl == nil {
		mset.csl = gsl.NewSublist[*consumer]()
	}
	for _, sub := range o.signalSubs() {
		mset.csl.Insert(sub, o)
	}
	mset.clsMu.Unlock()
}

// Lock should be held.
func (mset *stream) removeConsumer(o *consumer) {
	if o.cfg.FilterSubject != _EMPTY_ && mset.numFilter > 0 {
		mset.numFilter--
	}
	if o.cfg.Direct && mset.directs > 0 {
		mset.directs--
	}
	if mset.consumers != nil {
		delete(mset.consumers, o.name)
		// Now update consumers list as well
		mset.clsMu.Lock()
		for i, ol := range mset.cList {
			if ol == o {
				mset.cList = append(mset.cList[:i], mset.cList[i+1:]...)
				break
			}
		}
		// Always remove from the leader sublist.
		if mset.csl != nil {
			for _, sub := range o.signalSubs() {
				mset.csl.Remove(sub, o)
			}
		}
		mset.clsMu.Unlock()
	}
}

// swapSigSubs will update signal Subs for a new subject filter.
// consumer lock should not be held.
func (mset *stream) swapSigSubs(o *consumer, newFilters []string) {
	mset.clsMu.Lock()
	o.mu.Lock()

	if o.closed || o.mset == nil {
		o.mu.Unlock()
		mset.clsMu.Unlock()
		return
	}

	if o.sigSubs != nil {
		if mset.csl != nil {
			for _, sub := range o.sigSubs {
				mset.csl.Remove(sub, o)
			}
		}
		o.sigSubs = nil
	}

	if o.isLeader() {
		if mset.csl == nil {
			mset.csl = gsl.NewSublist[*consumer]()
		}
		// If no filters are preset, add fwcs to sublist for that consumer.
		if newFilters == nil {
			mset.csl.Insert(fwcs, o)
			o.sigSubs = append(o.sigSubs, fwcs)
			// If there are filters, add their subjects to sublist.
		} else {
			for _, filter := range newFilters {
				mset.csl.Insert(filter, o)
				o.sigSubs = append(o.sigSubs, filter)
			}
		}
	}
	o.mu.Unlock()
	mset.clsMu.Unlock()

	mset.mu.Lock()
	defer mset.mu.Unlock()

	if mset.numFilter > 0 && len(o.subjf) > 0 {
		mset.numFilter--
	}
	if len(newFilters) > 0 {
		mset.numFilter++
	}
}

// lookupConsumer will retrieve a consumer by name.
func (mset *stream) lookupConsumer(name string) *consumer {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.consumers[name]
}

func (mset *stream) numDirectConsumers() (num int) {
	mset.clsMu.RLock()
	defer mset.clsMu.RUnlock()

	// Consumers that are direct are not recorded at the store level.
	for _, o := range mset.cList {
		o.mu.RLock()
		if o.cfg.Direct {
			num++
		}
		o.mu.RUnlock()
	}
	return num
}

// State will return the current state for this stream.
func (mset *stream) state() StreamState {
	return mset.stateWithDetail(false)
}

func (mset *stream) stateWithDetail(details bool) StreamState {
	// mset.store does not change once set, so ok to reference here directly.
	// We do this elsewhere as well.
	store := mset.store
	if store == nil {
		return StreamState{}
	}

	// Currently rely on store for details.
	if details {
		return store.State()
	}
	// Here we do the fast version.
	var state StreamState
	store.FastState(&state)
	return state
}

func (mset *stream) Store() StreamStore {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.store
}

// Determines if the new proposed partition is unique amongst all consumers.
// Lock should be held.
func (mset *stream) partitionUnique(name string, partitions []string) bool {
	for _, partition := range partitions {
		for n, o := range mset.consumers {
			// Skip the consumer being checked.
			if n == name {
				continue
			}
			if o.subjf == nil {
				return false
			}
			for _, filter := range o.subjf {
				if SubjectsCollide(partition, filter.subject) {
					return false
				}
			}
		}
	}
	return true
}

// Lock should be held.
func (mset *stream) potentialFilteredConsumers() bool {
	numSubjects := len(mset.cfg.Subjects)
	if len(mset.consumers) == 0 || numSubjects == 0 {
		return false
	}
	if numSubjects > 1 || subjectHasWildcard(mset.cfg.Subjects[0]) {
		return true
	}
	return false
}

// Check if there is no interest in this sequence number across our consumers.
// The consumer passed is optional if we are processing the ack for that consumer.
// Write lock should be held.
func (mset *stream) noInterest(seq uint64, obs *consumer) bool {
	return !mset.checkForInterest(seq, obs)
}

// Check if there is no interest in this sequence number and subject across our consumers.
// The consumer passed is optional if we are processing the ack for that consumer.
// Write lock should be held.
func (mset *stream) noInterestWithSubject(seq uint64, subj string, obs *consumer) bool {
	return !mset.checkForInterestWithSubject(seq, subj, obs)
}

// Write lock should be held here for the stream to avoid race conditions on state.
func (mset *stream) checkForInterest(seq uint64, obs *consumer) bool {
	var subj string
	if mset.potentialFilteredConsumers() {
		pmsg := getJSPubMsgFromPool()
		defer pmsg.returnToPool()
		sm, err := mset.store.LoadMsg(seq, &pmsg.StoreMsg)
		if err != nil {
			if err == ErrStoreEOF {
				// Register this as a preAck.
				mset.registerPreAck(obs, seq)
				return true
			}
			mset.clearAllPreAcks(seq)
			return false
		}
		subj = sm.subj
	}
	return mset.checkForInterestWithSubject(seq, subj, obs)
}

// Checks for interest given a sequence and subject.
func (mset *stream) checkForInterestWithSubject(seq uint64, subj string, obs *consumer) bool {
	for _, o := range mset.consumers {
		// If this is us or we have a registered preAck for this consumer continue inspecting.
		if o == obs || mset.hasPreAck(o, seq) {
			continue
		}
		// Check if we need an ack.
		if o.needAck(seq, subj) {
			return true
		}
	}
	mset.clearAllPreAcks(seq)
	return false
}

// Check if we have a pre-registered ack for this sequence.
// Write lock should be held.
func (mset *stream) hasPreAck(o *consumer, seq uint64) bool {
	if o == nil || len(mset.preAcks) == 0 {
		return false
	}
	consumers := mset.preAcks[seq]
	if len(consumers) == 0 {
		return false
	}
	_, found := consumers[o]
	return found
}

// Check if we have all consumers pre-acked for this sequence and subject.
// Write lock should be held.
func (mset *stream) hasAllPreAcks(seq uint64, subj string) bool {
	if len(mset.preAcks) == 0 || len(mset.preAcks[seq]) == 0 {
		return false
	}
	// Since these can be filtered and mutually exclusive,
	// if we have some preAcks we need to check all interest here.
	return mset.noInterestWithSubject(seq, subj, nil)
}

// Check if we have all consumers pre-acked.
// Write lock should be held.
func (mset *stream) clearAllPreAcks(seq uint64) {
	delete(mset.preAcks, seq)
}

// Clear all preAcks below floor.
// Write lock should be held.
func (mset *stream) clearAllPreAcksBelowFloor(floor uint64) {
	for seq := range mset.preAcks {
		if seq < floor {
			delete(mset.preAcks, seq)
		}
	}
}

// This will register an ack for a consumer if it arrives before the actual message.
func (mset *stream) registerPreAckLock(o *consumer, seq uint64) {
	mset.mu.Lock()
	defer mset.mu.Unlock()
	mset.registerPreAck(o, seq)
}

// This will register an ack for a consumer if it arrives before
// the actual message.
// Write lock should be held.
func (mset *stream) registerPreAck(o *consumer, seq uint64) {
	if o == nil {
		return
	}
	if mset.preAcks == nil {
		mset.preAcks = make(map[uint64]map[*consumer]struct{})
	}
	if mset.preAcks[seq] == nil {
		mset.preAcks[seq] = make(map[*consumer]struct{})
	}
	mset.preAcks[seq][o] = struct{}{}
}

// This will clear an ack for a consumer.
// Write lock should be held.
func (mset *stream) clearPreAck(o *consumer, seq uint64) {
	if o == nil || len(mset.preAcks) == 0 {
		return
	}
	if consumers := mset.preAcks[seq]; len(consumers) > 0 {
		delete(consumers, o)
		if len(consumers) == 0 {
			delete(mset.preAcks, seq)
		}
	}
}

// ackMsg is called into from a consumer when we have a WorkQueue or Interest Retention Policy.
// Returns whether the message at seq was removed as a result of the ACK.
// (Or should be removed in the case of clustered streams, since it requires a message delete proposal)
func (mset *stream) ackMsg(o *consumer, seq uint64) bool {
	if seq == 0 {
		return false
	}

	// Don't make this RLock(). We need to have only 1 running at a time to gauge interest across all consumers.
	mset.mu.Lock()
	if mset.closed.Load() || mset.cfg.Retention == LimitsPolicy {
		mset.mu.Unlock()
		return false
	}

	store := mset.store
	var state StreamState
	store.FastState(&state)

	// If this has arrived before we have processed the message itself.
	if seq > state.LastSeq {
		mset.registerPreAck(o, seq)
		mset.mu.Unlock()
		// We have not removed the message, but should still signal so we could retry later
		// since we potentially need to remove it then.
		return true
	}

	// Always clear pre-ack if here.
	mset.clearPreAck(o, seq)

	// Make sure this sequence is not below our first sequence.
	if seq < state.FirstSeq {
		mset.mu.Unlock()
		return false
	}

	var shouldRemove bool
	switch mset.cfg.Retention {
	case WorkQueuePolicy:
		// Normally we just remove a message when its ack'd here but if we have direct consumers
		// from sources and/or mirrors we need to make sure they have delivered the msg.
		shouldRemove = mset.directs <= 0 || mset.noInterest(seq, o)
	case InterestPolicy:
		shouldRemove = mset.noInterest(seq, o)
	}

	// If nothing else to do.
	if !shouldRemove {
		mset.mu.Unlock()
		return false
	}

	if !mset.isClustered() {
		mset.mu.Unlock()
		// If we are here we should attempt to remove.
		if _, err := store.RemoveMsg(seq); err == ErrStoreEOF {
			// This should not happen, but being pedantic.
			mset.registerPreAckLock(o, seq)
		}
		return true
	}

	// Only propose message deletion to the stream if we're consumer leader, otherwise all followers would also propose.
	// We must be the consumer leader, since we know for sure we've stored the message and don't register as pre-ack.
	if o != nil && !o.IsLeader() {
		mset.mu.Unlock()
		// Must still mark as removal if follower. If we become leader later, we must be able to retry the proposal.
		return true
	}

	md := streamMsgDelete{Seq: seq, NoErase: true, Stream: mset.cfg.Name}
	mset.node.ForwardProposal(encodeMsgDelete(&md))
	mset.mu.Unlock()
	return true
}

// Snapshot creates a snapshot for the stream and possibly consumers.
func (mset *stream) snapshot(deadline time.Duration, checkMsgs, includeConsumers bool) (*SnapshotResult, error) {
	if mset.closed.Load() {
		return nil, errStreamClosed
	}
	store := mset.store
	return store.Snapshot(deadline, checkMsgs, includeConsumers)
}

const snapsDir = "__snapshots__"

// RestoreStream will restore a stream from a snapshot.
func (a *Account) RestoreStream(ncfg *StreamConfig, r io.Reader) (*stream, error) {
	if ncfg == nil {
		return nil, errors.New("nil config on stream restore")
	}

	s, jsa, err := a.checkForJetStream()
	if err != nil {
		return nil, err
	}
	js := jsa.js
	if js == nil {
		return nil, NewJSNotEnabledForAccountError()
	}

	cfg, apiErr := s.checkStreamCfg(ncfg, a, false)
	if apiErr != nil {
		return nil, apiErr
	}

	sd := filepath.Join(jsa.storeDir, snapsDir)
	if _, err := os.Stat(sd); os.IsNotExist(err) {
		if err := os.MkdirAll(sd, defaultDirPerms); err != nil {
			return nil, fmt.Errorf("could not create snapshots directory - %v", err)
		}
	}
	sdir, err := os.MkdirTemp(sd, "snap-")
	if err != nil {
		return nil, err
	}
	if _, err := os.Stat(sdir); os.IsNotExist(err) {
		if err := os.MkdirAll(sdir, defaultDirPerms); err != nil {
			return nil, fmt.Errorf("could not create snapshots directory - %v", err)
		}
	}
	defer os.RemoveAll(sdir)

	logAndReturnError := func() error {
		a.mu.RLock()
		err := fmt.Errorf("unexpected content (account=%s)", a.Name)
		if a.srv != nil {
			a.srv.Errorf("Stream restore failed due to %v", err)
		}
		a.mu.RUnlock()
		return err
	}
	sdirCheck := filepath.Clean(sdir) + string(os.PathSeparator)

	_, isClustered := jsa.jetStreamAndClustered()
	jsa.usageMu.RLock()
	selected, tier, hasTier := jsa.selectLimits(cfg.Replicas)
	jsa.usageMu.RUnlock()
	reserved := int64(0)
	if hasTier {
		if isClustered {
			js.mu.RLock()
			_, reserved = tieredStreamAndReservationCount(js.cluster.streams[a.Name], tier, &cfg)
			js.mu.RUnlock()
		} else {
			reserved = jsa.tieredReservation(tier, &cfg)
		}
	}

	var bc int64
	tr := tar.NewReader(s2.NewReader(r))
	for {
		hdr, err := tr.Next()
		if err == io.EOF {
			break // End of snapshot
		}
		if err != nil {
			return nil, err
		}
		if hdr.Typeflag != tar.TypeReg {
			return nil, logAndReturnError()
		}
		bc += hdr.Size
		js.mu.RLock()
		err = js.checkAllLimits(&selected, &cfg, reserved, bc)
		js.mu.RUnlock()
		if err != nil {
			return nil, err
		}
		fpath := filepath.Join(sdir, filepath.Clean(hdr.Name))
		if !strings.HasPrefix(fpath, sdirCheck) {
			return nil, logAndReturnError()
		}
		os.MkdirAll(filepath.Dir(fpath), defaultDirPerms)
		fd, err := os.OpenFile(fpath, os.O_CREATE|os.O_RDWR, 0600)
		if err != nil {
			return nil, err
		}
		_, err = io.Copy(fd, tr)
		fd.Close()
		if err != nil {
			return nil, err
		}
	}

	// Check metadata.
	// The cfg passed in will be the new identity for the stream.
	var fcfg FileStreamInfo
	b, err := os.ReadFile(filepath.Join(sdir, JetStreamMetaFile))
	if err != nil {
		return nil, err
	}
	if err := json.Unmarshal(b, &fcfg); err != nil {
		return nil, err
	}

	// Check to make sure names match.
	if fcfg.Name != cfg.Name {
		return nil, errors.New("stream names do not match")
	}

	// See if this stream already exists.
	if _, err := a.lookupStream(cfg.Name); err == nil {
		return nil, NewJSStreamNameExistRestoreFailedError()
	}
	// Move into the correct place here.
	ndir := filepath.Join(jsa.storeDir, streamsDir, cfg.Name)
	// Remove old one if for some reason it is still here.
	if _, err := os.Stat(ndir); err == nil {
		os.RemoveAll(ndir)
	}
	// Make sure our destination streams directory exists.
	if err := os.MkdirAll(filepath.Join(jsa.storeDir, streamsDir), defaultDirPerms); err != nil {
		return nil, err
	}
	// Move into new location.
	if err := os.Rename(sdir, ndir); err != nil {
		return nil, err
	}

	if cfg.Template != _EMPTY_ {
		if err := jsa.addStreamNameToTemplate(cfg.Template, cfg.Name); err != nil {
			return nil, err
		}
	}
	mset, err := a.addStream(&cfg)
	if err != nil {
		// Make sure to clean up after ourselves here.
		os.RemoveAll(ndir)
		return nil, err
	}
	if !fcfg.Created.IsZero() {
		mset.setCreatedTime(fcfg.Created)
	}

	// Make sure we do an update if the configs have changed.
	if !reflect.DeepEqual(fcfg.StreamConfig, cfg) {
		if err := mset.update(&cfg); err != nil {
			return nil, err
		}
	}

	// Now do consumers.
	odir := filepath.Join(ndir, consumerDir)
	ofis, _ := os.ReadDir(odir)
	for _, ofi := range ofis {
		metafile := filepath.Join(odir, ofi.Name(), JetStreamMetaFile)
		metasum := filepath.Join(odir, ofi.Name(), JetStreamMetaFileSum)
		if _, err := os.Stat(metafile); os.IsNotExist(err) {
			mset.stop(true, false)
			return nil, fmt.Errorf("error restoring consumer [%q]: %v", ofi.Name(), err)
		}
		buf, err := os.ReadFile(metafile)
		if err != nil {
			mset.stop(true, false)
			return nil, fmt.Errorf("error restoring consumer [%q]: %v", ofi.Name(), err)
		}
		if _, err := os.Stat(metasum); os.IsNotExist(err) {
			mset.stop(true, false)
			return nil, fmt.Errorf("error restoring consumer [%q]: %v", ofi.Name(), err)
		}
		var cfg FileConsumerInfo
		if err := json.Unmarshal(buf, &cfg); err != nil {
			mset.stop(true, false)
			return nil, fmt.Errorf("error restoring consumer [%q]: %v", ofi.Name(), err)
		}
		isEphemeral := !isDurableConsumer(&cfg.ConsumerConfig)
		if isEphemeral {
			// This is an ephermal consumer and this could fail on restart until
			// the consumer can reconnect. We will create it as a durable and switch it.
			cfg.ConsumerConfig.Durable = ofi.Name()
		}
		obs, err := mset.addConsumer(&cfg.ConsumerConfig)
		if err != nil {
			mset.stop(true, false)
			return nil, fmt.Errorf("error restoring consumer [%q]: %v", ofi.Name(), err)
		}
		if isEphemeral {
			obs.switchToEphemeral()
		}
		if !cfg.Created.IsZero() {
			obs.setCreatedTime(cfg.Created)
		}
		obs.mu.Lock()
		err = obs.readStoredState()
		obs.mu.Unlock()
		if err != nil {
			mset.stop(true, false)
			return nil, fmt.Errorf("error restoring consumer [%q]: %v", ofi.Name(), err)
		}
	}
	return mset, nil
}

// This is to check for dangling messages on interest retention streams. Only called on account enable.
// Issue https://github.com/nats-io/nats-server/issues/3612
func (mset *stream) checkForOrphanMsgs() {
	mset.mu.RLock()
	consumers := make([]*consumer, 0, len(mset.consumers))
	for _, o := range mset.consumers {
		consumers = append(consumers, o)
	}
	accName, stream := mset.acc.Name, mset.cfg.Name

	var ss StreamState
	mset.store.FastState(&ss)
	mset.mu.RUnlock()

	for _, o := range consumers {
		if err := o.checkStateForInterestStream(&ss); err == errAckFloorHigherThanLastSeq {
			o.mu.RLock()
			s, consumer := o.srv, o.name
			state, _ := o.store.State()
			asflr := state.AckFloor.Stream
			o.mu.RUnlock()
			// Warn about stream state vs our ack floor.
			s.RateLimitWarnf("Detected consumer '%s > %s > %s' ack floor %d is ahead of stream's last sequence %d",
				accName, stream, consumer, asflr, ss.LastSeq)
		}
	}
}

// Check on startup to make sure that consumers replication matches us.
// Interest retention requires replication matches.
func (mset *stream) checkConsumerReplication() {
	mset.mu.RLock()
	defer mset.mu.RUnlock()

	if mset.cfg.Retention != InterestPolicy {
		return
	}

	s, acc := mset.srv, mset.acc
	for _, o := range mset.consumers {
		o.mu.RLock()
		// Consumer replicas 0 can be a legit config for the replicas and we will inherit from the stream
		// when this is the case.
		if mset.cfg.Replicas != o.cfg.Replicas && o.cfg.Replicas != 0 {
			s.Errorf("consumer '%s > %s > %s' MUST match replication (%d vs %d) of stream with interest policy",
				acc, mset.cfg.Name, o.cfg.Name, mset.cfg.Replicas, o.cfg.Replicas)
		}
		o.mu.RUnlock()
	}
}

// Will check if we are running in the monitor already and if not set the appropriate flag.
func (mset *stream) checkInMonitor() bool {
	mset.mu.Lock()
	defer mset.mu.Unlock()

	if mset.inMonitor {
		return true
	}
	mset.inMonitor = true
	return false
}

// Clear us being in the monitor routine.
func (mset *stream) clearMonitorRunning() {
	mset.mu.Lock()
	defer mset.mu.Unlock()
	mset.inMonitor = false

	// We've stopped running the monitor goroutine, now also do additional cleanup.
	mset.deleteBatchApplyState()
}

// Check if our monitor is running.
func (mset *stream) isMonitorRunning() bool {
	mset.mu.RLock()
	defer mset.mu.RUnlock()
	return mset.inMonitor
}

// Adjust accounting for sent messages as part of replication.
func (mset *stream) trackReplicationTraffic(node RaftNode, sz int, r int) {
	// If we are using the system account for NRG, add in the extra sent msgs and bytes to our account
	// so that the end user / account owner has visibility.
	if node.IsSystemAccount() && mset.acc != nil && r > 1 {
		outMsgs := int64(r - 1)
		outBytes := int64(sz * (r - 1))

		mset.acc.stats.Lock()
		mset.acc.stats.outMsgs += outMsgs
		mset.acc.stats.outBytes += outBytes
		mset.acc.stats.rt.outMsgs += outMsgs
		mset.acc.stats.rt.outBytes += outBytes
		mset.acc.stats.Unlock()
	}
}
