// Copyright 2019-2026 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 (
	"bytes"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"math"
	"os"
	"path/filepath"
	"runtime/debug"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/minio/highwayhash"
	"github.com/nats-io/nats-server/v2/server/gsl"
	"github.com/nats-io/nats-server/v2/server/sysmem"
	"github.com/nats-io/nats-server/v2/server/tpm"
	"github.com/nats-io/nkeys"
	"github.com/nats-io/nuid"
)

// JetStreamConfig determines this server's configuration.
// MaxMemory and MaxStore are in bytes.
type JetStreamConfig struct {
	MaxMemory    int64         `json:"max_memory"`              // MaxMemory is the maximum size of memory type streams
	MaxStore     int64         `json:"max_storage"`             // MaxStore is the maximum size of file store type streams
	StoreDir     string        `json:"store_dir,omitempty"`     // StoreDir is where storage files are stored
	SyncInterval time.Duration `json:"sync_interval,omitempty"` // SyncInterval is how frequently we sync to disk in the background by calling fsync
	SyncAlways   bool          `json:"sync_always,omitempty"`   // SyncAlways indicates flushes are done after every write
	Domain       string        `json:"domain,omitempty"`        // Domain is the JetStream domain
	CompressOK   bool          `json:"compress_ok,omitempty"`   // CompressOK indicates if compression is supported
	UniqueTag    string        `json:"unique_tag,omitempty"`    // UniqueTag is the unique tag assigned to this instance
	Strict       bool          `json:"strict,omitempty"`        // Strict indicates if strict JSON parsing is performed
}

// Statistics about JetStream for this server.
type JetStreamStats struct {
	Memory         uint64            `json:"memory"`
	Store          uint64            `json:"storage"`
	ReservedMemory uint64            `json:"reserved_memory"`
	ReservedStore  uint64            `json:"reserved_storage"`
	Accounts       int               `json:"accounts"`
	HAAssets       int               `json:"ha_assets"`
	API            JetStreamAPIStats `json:"api"`
}

type JetStreamAccountLimits struct {
	MaxMemory            int64 `json:"max_memory"`
	MaxStore             int64 `json:"max_storage"`
	MaxStreams           int   `json:"max_streams"`
	MaxConsumers         int   `json:"max_consumers"`
	MaxAckPending        int   `json:"max_ack_pending"`
	MemoryMaxStreamBytes int64 `json:"memory_max_stream_bytes"`
	StoreMaxStreamBytes  int64 `json:"storage_max_stream_bytes"`
	MaxBytesRequired     bool  `json:"max_bytes_required"`
}

type JetStreamTier struct {
	Memory         uint64                 `json:"memory"`
	Store          uint64                 `json:"storage"`
	ReservedMemory uint64                 `json:"reserved_memory"`
	ReservedStore  uint64                 `json:"reserved_storage"`
	Streams        int                    `json:"streams"`
	Consumers      int                    `json:"consumers"`
	Limits         JetStreamAccountLimits `json:"limits"`
}

// JetStreamAccountStats returns current statistics about the account's JetStream usage.
type JetStreamAccountStats struct {
	JetStreamTier                          // in case tiers are used, reflects totals with limits not set
	Domain        string                   `json:"domain,omitempty"`
	API           JetStreamAPIStats        `json:"api"`
	Tiers         map[string]JetStreamTier `json:"tiers,omitempty"` // indexed by tier name
}

// JetStreamAPIStats holds stats about the API usage for this server
type JetStreamAPIStats struct {
	Level    int    `json:"level"`              // Level is the active API level this server implements
	Total    uint64 `json:"total"`              // Total is the total API requests received since start
	Errors   uint64 `json:"errors"`             // Errors is the total API requests that resulted in error responses
	Inflight uint64 `json:"inflight,omitempty"` // Inflight are the number of API requests currently being served
}

// This is for internal accounting for JetStream for this server.
type jetStream struct {
	// These are here first because of atomics on 32bit systems.
	apiInflight    int64
	apiTotal       int64
	apiErrors      int64
	memMax         int64
	memReserved    int64 // Requires JS lock to be held.
	memUsed        int64
	storeMax       int64
	storeReserved  int64 // Requires JS lock to be held.
	storeUsed      int64
	queueLimit     int64
	infoQueueLimit int64
	clustered      int32
	mu             sync.RWMutex
	srv            *Server
	config         JetStreamConfig
	cluster        *jetStreamCluster
	accounts       map[string]*jsAccount
	apiSubs        *Sublist
	infoSubs       *gsl.SimpleSublist // Subjects for info-specific queue.
	started        time.Time

	// System level request to purge a stream move
	accountPurge *subscription

	// Some bools regarding general state.
	metaRecovering bool
	standAlone     bool
	oos            bool
	shuttingDown   bool

	// Atomic versions
	disabled atomic.Bool
}

type remoteUsage struct {
	tiers map[string]*jsaUsage // indexed by tier name
	api   uint64
	err   uint64
}

type jsaStorage struct {
	total jsaUsage
	local jsaUsage
}

// This represents a jetstream enabled account.
// Worth noting that we include the jetstream pointer, this is because
// in general we want to be very efficient when receiving messages on
// an internal sub for a stream, so we will direct link to the stream
// and walk backwards as needed vs multiple hash lookups and locks, etc.
type jsAccount struct {
	mu       sync.RWMutex
	js       *jetStream
	account  *Account
	storeDir string
	inflight sync.Map
	streams  map[string]*stream

	// From server
	sendq *ipQueue[*pubMsg]

	// For limiting only running one checkAndSync at a time.
	sync atomic.Bool

	// Usage/limits related fields that will be protected by usageMu
	usageMu    sync.RWMutex
	limits     map[string]JetStreamAccountLimits // indexed by tierName
	usage      map[string]*jsaStorage            // indexed by tierName
	rusage     map[string]*remoteUsage           // indexed by node id
	apiTotal   uint64
	apiErrors  uint64
	usageApi   uint64
	usageErr   uint64
	updatesPub string
	updatesSub *subscription
	lupdate    time.Time
	utimer     *time.Timer
}

// Track general usage for this account.
type jsaUsage struct {
	mem   int64
	store int64
}

// EnableJetStream will enable JetStream support on this server with the given configuration.
// A nil configuration will dynamically choose the limits and temporary file storage directory.
func (s *Server) EnableJetStream(config *JetStreamConfig) error {
	if s.JetStreamEnabled() {
		return fmt.Errorf("jetstream already enabled")
	}

	s.Noticef("Starting JetStream")
	start := time.Now()
	defer func() {
		s.Noticef("Took %s to start JetStream", time.Since(start))
	}()

	if config == nil || config.MaxMemory <= 0 || config.MaxStore <= 0 {
		var storeDir, domain, uniqueTag string
		var maxStore, maxMem int64
		if config != nil {
			storeDir, domain, uniqueTag = config.StoreDir, config.Domain, config.UniqueTag
			maxStore, maxMem = config.MaxStore, config.MaxMemory
		}
		config = s.dynJetStreamConfig(storeDir, maxStore, maxMem)
		if maxMem > 0 {
			config.MaxMemory = maxMem
		}
		if domain != _EMPTY_ {
			config.Domain = domain
		}
		if uniqueTag != _EMPTY_ {
			config.UniqueTag = uniqueTag
		}
		s.Debugf("JetStream creating dynamic configuration - %s memory, %s disk", friendlyBytes(config.MaxMemory), friendlyBytes(config.MaxStore))
	} else if config.StoreDir != _EMPTY_ {
		config.StoreDir = filepath.Join(config.StoreDir, JetStreamStoreDir)
	}

	cfg := *config
	if cfg.StoreDir == _EMPTY_ {
		cfg.StoreDir = filepath.Join(os.TempDir(), JetStreamStoreDir)
		s.Warnf("Temporary storage directory used, data could be lost on system reboot")
	}

	// We will consistently place the 'jetstream' directory under the storedir that was handed to us. Prior to 2.2.3 though
	// we could have a directory on disk without the 'jetstream' directory. This will check and fix if needed.
	if err := s.checkStoreDir(&cfg); err != nil {
		return err
	}

	return s.enableJetStream(cfg)
}

// Function signature to generate a key encryption key.
type keyGen func(context []byte) ([]byte, error)

// Return a key generation function or nil if encryption not enabled.
func (s *Server) jsKeyGen(jsKey, info string) keyGen {
	if ek := jsKey; ek != _EMPTY_ {
		return func(context []byte) ([]byte, error) {
			h := hmac.New(sha256.New, []byte(ek))
			if _, err := h.Write([]byte(info)); err != nil {
				return nil, err
			}
			if _, err := h.Write(context); err != nil {
				return nil, err
			}
			return h.Sum(nil), nil
		}
	}
	return nil
}

// Decode the encrypted metafile.
func (s *Server) decryptMeta(sc StoreCipher, ekey, buf []byte, acc, context string) ([]byte, bool, error) {
	if len(ekey) < minMetaKeySize {
		return nil, false, errBadKeySize
	}
	var osc StoreCipher
	switch sc {
	case AES:
		osc = ChaCha
	case ChaCha:
		osc = AES
	}
	type prfWithCipher struct {
		keyGen
		StoreCipher
	}
	var prfs []prfWithCipher
	if prf := s.jsKeyGen(s.getOpts().JetStreamKey, acc); prf == nil {
		return nil, false, errNoEncryption
	} else {
		// First of all, try our current encryption keys with both
		// store cipher algorithms.
		prfs = append(prfs, prfWithCipher{prf, sc})
		prfs = append(prfs, prfWithCipher{prf, osc})
	}
	if prf := s.jsKeyGen(s.getOpts().JetStreamOldKey, acc); prf != nil {
		// Then, if we have an old encryption key, try with also with
		// both store cipher algorithms.
		prfs = append(prfs, prfWithCipher{prf, sc})
		prfs = append(prfs, prfWithCipher{prf, osc})
	}

	for i, prf := range prfs {
		rb, err := prf.keyGen([]byte(context))
		if err != nil {
			continue
		}
		kek, err := genEncryptionKey(prf.StoreCipher, rb)
		if err != nil {
			continue
		}
		ns := kek.NonceSize()
		seed, err := kek.Open(nil, ekey[:ns], ekey[ns:], nil)
		if err != nil {
			continue
		}
		aek, err := genEncryptionKey(prf.StoreCipher, seed)
		if err != nil {
			continue
		}
		if aek.NonceSize() != kek.NonceSize() {
			continue
		}
		plain, err := aek.Open(nil, buf[:ns], buf[ns:], nil)
		if err != nil {
			continue
		}
		return plain, i > 0, nil
	}
	return nil, false, fmt.Errorf("unable to recover keys")
}

// Check to make sure directory has the jetstream directory.
// We will have it properly configured here now regardless, so need to look inside.
func (s *Server) checkStoreDir(cfg *JetStreamConfig) error {
	fis, _ := os.ReadDir(cfg.StoreDir)
	// If we have nothing underneath us, could be just starting new, but if we see this we can check.
	if len(fis) != 0 {
		return nil
	}
	// Let's check the directory above. If it has us 'jetstream' but also other stuff that we can
	// identify as accounts then we can fix.
	fis, _ = os.ReadDir(filepath.Dir(cfg.StoreDir))
	// If just one that is us 'jetstream' and all is ok.
	if len(fis) == 1 {
		return nil
	}

	haveJetstreamDir := false
	for _, fi := range fis {
		if fi.Name() == JetStreamStoreDir {
			haveJetstreamDir = true
			break
		}
	}

	for _, fi := range fis {
		// Skip the 'jetstream' directory.
		if fi.Name() == JetStreamStoreDir {
			continue
		}
		// Let's see if this is an account.
		if accName := fi.Name(); accName != _EMPTY_ {
			_, ok := s.accounts.Load(accName)
			if !ok && s.AccountResolver() != nil && nkeys.IsValidPublicAccountKey(accName) {
				// Account is not local but matches the NKEY account public key,
				// this is enough indication to move this directory, no need to
				// fetch the account.
				ok = true
			}
			// If this seems to be an account go ahead and move the directory. This will include all assets
			// like streams and consumers.
			if ok {
				if !haveJetstreamDir {
					err := os.Mkdir(filepath.Join(filepath.Dir(cfg.StoreDir), JetStreamStoreDir), defaultDirPerms)
					if err != nil {
						return err
					}
					haveJetstreamDir = true
				}
				old := filepath.Join(filepath.Dir(cfg.StoreDir), fi.Name())
				new := filepath.Join(cfg.StoreDir, fi.Name())
				s.Noticef("JetStream relocated account %q to %q", old, new)
				if err := os.Rename(old, new); err != nil {
					return err
				}
			}
		}
	}

	return nil
}

// This function sets/updates the jetstream encryption key and cipher based
// on options. If the TPM options have been specified, a key is generated
// and sealed by the TPM.
func (s *Server) initJetStreamEncryption() (err error) {
	opts := s.getOpts()

	// The TPM settings and other encryption settings are mutually exclusive.
	if opts.JetStreamKey != _EMPTY_ && opts.JetStreamTpm.KeysFile != _EMPTY_ {
		return fmt.Errorf("JetStream encryption key may not be used with TPM options")
	}
	// if we are using the standard method to set the encryption key just return and carry on.
	if opts.JetStreamKey != _EMPTY_ {
		return nil
	}
	// if the tpm options are not used then no encryption has been configured and return.
	if opts.JetStreamTpm.KeysFile == _EMPTY_ {
		return nil
	}

	if opts.JetStreamTpm.Pcr == 0 {
		// Default PCR to use in the TPM. Values can be 0-23, and most platforms
		// reserve values 0-12 for the OS, boot locker, disc encryption, etc.
		// 16 used for debugging. In sticking to NATS tradition, we'll use 22
		// as the default with the option being configurable.
		opts.JetStreamTpm.Pcr = 22
	}

	// Using the TPM to generate or get the encryption key and update the encryption options.
	opts.JetStreamKey, err = tpm.LoadJetStreamEncryptionKeyFromTPM(opts.JetStreamTpm.SrkPassword,
		opts.JetStreamTpm.KeysFile, opts.JetStreamTpm.KeyPassword, opts.JetStreamTpm.Pcr)

	return err
}

// enableJetStream will start up the JetStream subsystem.
func (s *Server) enableJetStream(cfg JetStreamConfig) error {
	js := &jetStream{srv: s, config: cfg, accounts: make(map[string]*jsAccount), apiSubs: NewSublistNoCache(), infoSubs: gsl.NewSimpleSublist()}
	s.gcbMu.Lock()
	if s.gcbOutMax = s.getOpts().JetStreamMaxCatchup; s.gcbOutMax == 0 {
		s.gcbOutMax = defaultMaxTotalCatchupOutBytes
	}
	s.gcbMu.Unlock()

	atomic.StoreInt64(&js.memMax, cfg.MaxMemory)
	atomic.StoreInt64(&js.storeMax, cfg.MaxStore)

	// TODO: Not currently reloadable.
	atomic.StoreInt64(&js.queueLimit, s.getOpts().JetStreamRequestQueueLimit)
	atomic.StoreInt64(&js.infoQueueLimit, s.getOpts().JetStreamInfoQueueLimit)

	s.js.Store(js)

	// FIXME(dlc) - Allow memory only operation?
	if stat, err := os.Stat(cfg.StoreDir); os.IsNotExist(err) {
		if err := os.MkdirAll(cfg.StoreDir, defaultDirPerms); err != nil {
			return fmt.Errorf("could not create storage directory - %v", err)
		}
	} else {
		// Make sure its a directory and that we can write to it.
		if stat == nil || !stat.IsDir() {
			return fmt.Errorf("storage directory is not a directory")
		}
		tmpfile, err := os.CreateTemp(cfg.StoreDir, "_test_")
		if err != nil {
			return fmt.Errorf("storage directory is not writable")
		}
		tmpfile.Close()
		os.Remove(tmpfile.Name())
	}

	if err := s.initJetStreamEncryption(); err != nil {
		return err
	}

	// JetStream is an internal service so we need to make sure we have a system account.
	// This system account will export the JetStream service endpoints.
	if s.SystemAccount() == nil {
		s.SetDefaultSystemAccount()
	}

	opts := s.getOpts()
	if !opts.DisableJetStreamBanner {
		s.Noticef("    _ ___ _____ ___ _____ ___ ___   _   __  __")
		s.Noticef(" _ | | __|_   _/ __|_   _| _ \\ __| /_\\ |  \\/  |")
		s.Noticef("| || | _|  | | \\__ \\ | | |   / _| / _ \\| |\\/| |")
		s.Noticef(" \\__/|___| |_| |___/ |_| |_|_\\___/_/ \\_\\_|  |_|")
		s.Noticef("")
		s.Noticef("         https://docs.nats.io/jetstream")
		s.Noticef("")
	}
	s.Noticef("---------------- JETSTREAM ----------------")

	if cfg.Strict {
		s.Noticef("  Strict:          %t", cfg.Strict)
	}

	s.Noticef("  Max Memory:      %s", friendlyBytes(cfg.MaxMemory))
	s.Noticef("  Max Storage:     %s", friendlyBytes(cfg.MaxStore))
	s.Noticef("  Store Directory: \"%s\"", cfg.StoreDir)
	if cfg.Domain != _EMPTY_ {
		s.Noticef("  Domain:          %s", cfg.Domain)
	}

	if ek := opts.JetStreamKey; ek != _EMPTY_ {
		s.Noticef("  Encryption:      %s", opts.JetStreamCipher)
	}
	if opts.JetStreamTpm.KeysFile != _EMPTY_ {
		s.Noticef("  TPM File:        %q, Pcr: %d", opts.JetStreamTpm.KeysFile,
			opts.JetStreamTpm.Pcr)
	}
	s.Noticef("  API Level:       %d", JSApiLevel)
	s.Noticef("-------------------------------------------")

	// Setup our internal subscriptions.
	if err := s.setJetStreamExportSubs(); err != nil {
		return fmt.Errorf("setting up internal jetstream subscriptions failed: %v", err)
	}

	// Setup our internal system exports.
	s.Debugf("  Exports:")
	s.Debugf("     %s", jsAllAPI)
	s.setupJetStreamExports()

	standAlone, canExtend := s.standAloneMode(), s.canExtendOtherDomain()
	if standAlone && canExtend && s.getOpts().JetStreamExtHint != jsWillExtend {
		canExtend = false
		s.Noticef("Standalone server started in clustered mode do not support extending domains")
		s.Noticef(`Manually disable standalone mode by setting the JetStream Option "extension_hint: %s"`, jsWillExtend)
	}

	// Indicate if we will be standalone for checking resource reservations, etc.
	js.setJetStreamStandAlone(standAlone && !canExtend)

	// Enable accounts and restore state before starting clustering.
	if err := s.enableJetStreamAccounts(); err != nil {
		return err
	}

	// If we are in clustered mode go ahead and start the meta controller.
	if !standAlone || canExtend {
		if err := s.enableJetStreamClustering(); err != nil {
			return err
		}
		// Set our atomic bool to clustered.
		s.jsClustered.Store(true)
	}

	// Mark when we are up and running.
	js.setStarted()

	return nil
}

const jsNoExtend = "no_extend"
const jsWillExtend = "will_extend"

// This will check if we have a solicited leafnode that shares the system account
// and extension is not manually disabled
func (s *Server) canExtendOtherDomain() bool {
	opts := s.getOpts()
	sysAcc := s.SystemAccount().GetName()
	for _, r := range opts.LeafNode.Remotes {
		if r.LocalAccount == sysAcc {
			for _, denySub := range r.DenyImports {
				if subjectIsSubsetMatch(denySub, raftAllSubj) {
					return false
				}
			}
			return true
		}
	}
	return false
}

func (s *Server) updateJetStreamInfoStatus(enabled bool) {
	s.mu.Lock()
	s.info.JetStream = enabled
	s.mu.Unlock()
}

// restartJetStream will try to re-enable JetStream during a reload if it had been disabled during runtime.
func (s *Server) restartJetStream() error {
	opts := s.getOpts()
	cfg := JetStreamConfig{
		StoreDir:     opts.StoreDir,
		SyncInterval: opts.SyncInterval,
		SyncAlways:   opts.SyncAlways,
		MaxMemory:    opts.JetStreamMaxMemory,
		MaxStore:     opts.JetStreamMaxStore,
		Domain:       opts.JetStreamDomain,
		Strict:       !opts.NoJetStreamStrict,
	}
	s.Noticef("Restarting JetStream")
	err := s.EnableJetStream(&cfg)
	if err != nil {
		s.Warnf("Can't start JetStream: %v", err)
		return s.DisableJetStream()
	}
	s.updateJetStreamInfoStatus(true)
	return nil
}

// checkJetStreamExports will check if we have the JS exports setup
// on the system account, and if not go ahead and set them up.
func (s *Server) checkJetStreamExports() {
	if sacc := s.SystemAccount(); sacc != nil {
		sacc.mu.RLock()
		se := sacc.getServiceExport(jsAllAPI)
		sacc.mu.RUnlock()
		if se == nil {
			s.setupJetStreamExports()
		}
	}
}

func (s *Server) setupJetStreamExports() {
	// Setup our internal system export.
	if err := s.SystemAccount().AddServiceExport(jsAllAPI, nil); err != nil {
		s.Warnf("Error setting up jetstream service exports: %v", err)
	}
}

func (s *Server) jetStreamOOSPending() (wasPending bool) {
	if js := s.getJetStream(); js != nil {
		js.mu.Lock()
		wasPending = js.oos
		js.oos = true
		js.mu.Unlock()
	}
	return wasPending
}

func (s *Server) setJetStreamDisabled() {
	if js := s.getJetStream(); js != nil {
		js.disabled.Store(true)
	}
}

func (s *Server) handleOutOfSpace(mset *stream) {
	if s.JetStreamEnabled() && !s.jetStreamOOSPending() {
		var stream string
		if mset != nil {
			stream = mset.name()
			s.Errorf("JetStream out of %s resources, will be DISABLED", mset.Store().Type())
		} else {
			s.Errorf("JetStream out of resources, will be DISABLED")
		}

		go s.DisableJetStream()

		adv := &JSServerOutOfSpaceAdvisory{
			TypedEvent: TypedEvent{
				Type: JSServerOutOfStorageAdvisoryType,
				ID:   nuid.Next(),
				Time: time.Now().UTC(),
			},
			Server:   s.Name(),
			ServerID: s.ID(),
			Stream:   stream,
			Cluster:  s.cachedClusterName(),
			Domain:   s.getOpts().JetStreamDomain,
		}
		s.publishAdvisory(nil, JSAdvisoryServerOutOfStorage, adv)
	}
}

// DisableJetStream will turn off JetStream and signals in clustered mode
// to have the metacontroller remove us from the peer list.
func (s *Server) DisableJetStream() error {
	if !s.JetStreamEnabled() {
		return nil
	}

	s.setJetStreamDisabled()

	if s.JetStreamIsClustered() {
		isLeader := s.JetStreamIsLeader()
		js, cc := s.getJetStreamCluster()
		if js == nil {
			s.shutdownJetStream()
			return nil
		}
		js.mu.RLock()
		meta := cc.meta
		js.mu.RUnlock()

		if meta != nil {
			if isLeader {
				s.Warnf("JetStream initiating meta leader transfer")
				meta.StepDown()
				select {
				case <-s.quitCh:
					return nil
				case <-time.After(2 * time.Second):
				}
				if !s.JetStreamIsCurrent() {
					s.Warnf("JetStream timeout waiting for meta leader transfer")
				}
			}
			meta.Delete()
		}
	}

	// Update our info status.
	s.updateJetStreamInfoStatus(false)

	// Normal shutdown.
	s.shutdownJetStream()

	// Shut down the RAFT groups.
	s.shutdownRaftNodes()

	return nil
}

func (s *Server) enableJetStreamAccounts() error {
	// Reuse the same task workers across all accounts, so that we don't explode
	// with a large number of goroutines on multi-account systems.
	tq := parallelTaskQueue(len(dios))
	defer close(tq)

	// If we have no configured accounts setup then setup imports on global account.
	if s.globalAccountOnly() {
		gacc := s.GlobalAccount()
		gacc.mu.Lock()
		if len(gacc.jsLimits) == 0 {
			gacc.jsLimits = defaultJSAccountTiers
		}
		gacc.mu.Unlock()
		if err := s.configJetStream(gacc, tq); err != nil {
			return err
		}
	} else if err := s.configAllJetStreamAccounts(tq); err != nil {
		return fmt.Errorf("Error enabling jetstream on configured accounts: %v", err)
	}
	return nil
}

// enableAllJetStreamServiceImportsAndMappings turns on all service imports and mappings for jetstream for this account.
func (a *Account) enableAllJetStreamServiceImportsAndMappings() error {
	a.mu.RLock()
	s := a.srv
	a.mu.RUnlock()

	if s == nil {
		return fmt.Errorf("jetstream account not registered")
	}

	var dstAccName string
	if sacc := s.SystemAccount(); sacc != nil {
		dstAccName = sacc.Name
	}

	if !a.serviceImportExists(dstAccName, jsAllAPI) {
		// Capture si so we can turn on implicit sharing with JetStream layer.
		// Make sure to set "to" otherwise will incur performance slow down.
		si, err := a.addServiceImport(s.SystemAccount(), jsAllAPI, jsAllAPI, nil)
		if err != nil {
			return fmt.Errorf("Error setting up jetstream service imports for account: %v", err)
		}
		a.mu.Lock()
		si.share = true
		a.mu.Unlock()
	}

	// Check if we have a Domain specified.
	// If so add in a subject mapping that will allow local connected clients to reach us here as well.
	if opts := s.getOpts(); opts.JetStreamDomain != _EMPTY_ {
		mappings := generateJSMappingTable(opts.JetStreamDomain)
		a.mu.RLock()
		for _, m := range a.mappings {
			delete(mappings, m.src)
		}
		a.mu.RUnlock()
		for src, dest := range mappings {
			if err := a.AddMapping(src, dest); err != nil {
				s.Errorf("Error adding JetStream domain mapping: %v", err)
			}
		}
	}

	return nil
}

// enableJetStreamInfoServiceImportOnly will enable the single service import responder.
// Should we do them all regardless?
func (a *Account) enableJetStreamInfoServiceImportOnly() error {
	// Check if this import would be overshadowed. This can happen when accounts
	// are importing from another account for JS access.
	if a.serviceImportShadowed(JSApiAccountInfo) {
		return nil
	}

	return a.enableAllJetStreamServiceImportsAndMappings()
}

func (s *Server) configJetStream(acc *Account, tq chan<- func()) error {
	if acc == nil {
		return nil
	}
	acc.mu.RLock()
	jsLimits := acc.jsLimits
	acc.mu.RUnlock()
	if jsLimits != nil {
		// Check if already enabled. This can be during a reload.
		if acc.JetStreamEnabled() {
			if err := acc.enableAllJetStreamServiceImportsAndMappings(); err != nil {
				return err
			}
			if err := acc.UpdateJetStreamLimits(jsLimits); err != nil {
				return err
			}
		} else {
			if err := acc.EnableJetStream(jsLimits, tq); err != nil {
				return err
			}
			if s.gateway.enabled {
				s.switchAccountToInterestMode(acc.GetName())
			}
		}
	} else if acc != s.SystemAccount() {
		if acc.JetStreamEnabled() {
			acc.DisableJetStream()
		}
		// We will setup basic service imports to respond to
		// requests if JS is enabled for this account.
		if err := acc.enableJetStreamInfoServiceImportOnly(); err != nil {
			return err
		}
	}
	return nil
}

// configAllJetStreamAccounts walk all configured accounts and turn on jetstream if requested.
func (s *Server) configAllJetStreamAccounts(tq chan<- func()) error {
	// Check to see if system account has been enabled. We could arrive here via reload and
	// a non-default system account.
	s.checkJetStreamExports()

	// Bail if server not enabled. If it was enabled and a reload turns it off
	// that will be handled elsewhere.
	js := s.getJetStream()
	if js == nil {
		return nil
	}

	// Snapshot into our own list. Might not be needed.
	s.mu.RLock()
	if s.sys != nil {
		// clustered stream removal will perform this cleanup as well
		// this is mainly for initial cleanup
		saccName := s.sys.account.Name
		accStoreDirs, _ := os.ReadDir(js.config.StoreDir)
		for _, acc := range accStoreDirs {
			if accName := acc.Name(); accName != saccName {
				// no op if not empty
				accDir := filepath.Join(js.config.StoreDir, accName)
				os.Remove(filepath.Join(accDir, streamsDir))
				os.Remove(accDir)
			}
		}
	}

	var jsAccounts []*Account
	s.accounts.Range(func(k, v any) bool {
		jsAccounts = append(jsAccounts, v.(*Account))
		return true
	})
	accounts := &s.accounts
	s.mu.RUnlock()

	// Process any jetstream enabled accounts here. These will be accounts we are
	// already aware of at startup etc.
	for _, acc := range jsAccounts {
		if err := s.configJetStream(acc, tq); err != nil {
			return err
		}
	}

	// Now walk all the storage we have and resolve any accounts that we did not process already.
	// This is important in resolver/operator models.
	fis, _ := os.ReadDir(js.config.StoreDir)
	for _, fi := range fis {
		if accName := fi.Name(); accName != _EMPTY_ {
			// Only load up ones not already loaded since they are processed above.
			if _, ok := accounts.Load(accName); !ok {
				if acc, err := s.lookupAccount(accName); err != nil && acc != nil {
					if err := s.configJetStream(acc, tq); err != nil {
						return err
					}
				}
			}
		}
	}

	return nil
}

// Mark our started time.
func (js *jetStream) setStarted() {
	js.mu.Lock()
	defer js.mu.Unlock()
	js.started = time.Now()
}

func (js *jetStream) isEnabled() bool {
	if js == nil {
		return false
	}
	return !js.disabled.Load()
}

// Mark that we will be in standlone mode.
func (js *jetStream) setJetStreamStandAlone(isStandAlone bool) {
	if js == nil {
		return
	}
	js.mu.Lock()
	defer js.mu.Unlock()
	if js.standAlone = isStandAlone; js.standAlone {
		// Update our server atomic.
		js.srv.isMetaLeader.Store(true)
		js.accountPurge, _ = js.srv.systemSubscribe(JSApiAccountPurge, _EMPTY_, false, nil, js.srv.jsLeaderAccountPurgeRequest)
	} else if js.accountPurge != nil {
		js.srv.sysUnsubscribe(js.accountPurge)
	}
}

// JetStreamEnabled reports if jetstream is enabled for this server.
func (s *Server) JetStreamEnabled() bool {
	return s.getJetStream().isEnabled()
}

// JetStreamEnabledForDomain will report if any servers have JetStream enabled within this domain.
func (s *Server) JetStreamEnabledForDomain() bool {
	if s.JetStreamEnabled() {
		return true
	}

	var jsFound bool
	// If we are here we do not have JetStream enabled for ourselves, but we need to check all connected servers.
	// TODO(dlc) - Could optimize and memoize this.
	s.nodeToInfo.Range(func(k, v any) bool {
		// This should not be dependent on online status, so only check js.
		if v.(nodeInfo).js {
			jsFound = true
			return false
		}
		return true
	})

	return jsFound
}

// Will signal that all pull requests for consumers on this server are now invalid.
func (s *Server) signalPullConsumers() {
	js := s.getJetStream()
	if js == nil {
		return
	}

	js.mu.RLock()
	defer js.mu.RUnlock()

	// In case we have stale pending requests.
	const hdr = "NATS/1.0 409 Server Shutdown\r\n" + JSPullRequestPendingMsgs + ": %d\r\n" + JSPullRequestPendingBytes + ": %d\r\n\r\n"
	var didSend bool

	for _, jsa := range js.accounts {
		jsa.mu.RLock()
		for _, stream := range jsa.streams {
			stream.mu.RLock()
			for _, o := range stream.consumers {
				o.mu.RLock()
				// Only signal on R1.
				if o.cfg.Replicas <= 1 {
					for reply, wr := range o.pendingRequests() {
						shdr := fmt.Sprintf(hdr, wr.n, wr.b)
						o.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, []byte(shdr), nil, nil, 0))
						didSend = true
					}
				}
				o.mu.RUnlock()
			}
			stream.mu.RUnlock()
		}
		jsa.mu.RUnlock()
	}
	// Give time for migration information to make it out of our server.
	if didSend {
		time.Sleep(50 * time.Millisecond)
	}
}

// Helper for determining if we are shutting down.
func (js *jetStream) isShuttingDown() bool {
	js.mu.RLock()
	defer js.mu.RUnlock()
	return js.shuttingDown
}

// Shutdown jetstream for this server.
func (s *Server) shutdownJetStream() {
	js := s.getJetStream()
	if js == nil {
		return
	}

	s.Noticef("Initiating JetStream Shutdown...")
	defer s.Noticef("JetStream Shutdown")

	// If we have folks blocked on sync requests, unblock.
	// Send 1 is enough, but use select in case they were all present.
	select {
	case s.syncOutSem <- struct{}{}:
	default:
	}

	var _a [512]*Account
	accounts := _a[:0]

	js.mu.Lock()
	// Collect accounts.
	for _, jsa := range js.accounts {
		if a := jsa.acc(); a != nil {
			accounts = append(accounts, a)
		}
	}
	accPurgeSub := js.accountPurge
	js.accountPurge = nil
	// Signal we are shutting down.
	js.shuttingDown = true
	js.mu.Unlock()

	if accPurgeSub != nil {
		s.sysUnsubscribe(accPurgeSub)
	}

	for _, a := range accounts {
		a.removeJetStream()
	}

	s.js.Store(nil)

	js.mu.Lock()
	js.accounts = nil

	var qch chan struct{}
	var stopped chan struct{}
	if cc := js.cluster; cc != nil {
		if cc.qch != nil {
			qch, stopped = cc.qch, cc.stopped
			cc.qch, cc.stopped = nil, nil
		}
		js.stopUpdatesSub()
		if cc.c != nil {
			cc.c.closeConnection(ClientClosed)
			cc.c = nil
		}
		cc.meta = nil
		// Set our atomic bool to false.
		s.jsClustered.Store(false)
	}
	js.mu.Unlock()

	// If we were clustered signal the monitor cluster go routine.
	// We will wait for a bit for it to close.
	// Do this without the lock.
	if qch != nil {
		close(qch) // Must be close() to signal *all* listeners
		select {
		case <-stopped:
		case <-time.After(10 * time.Second):
			s.Warnf("Did not receive signal for successful shutdown of cluster routine")
		}
	}
}

// JetStreamConfig will return the current config. Useful if the system
// created a dynamic configuration. A copy is returned.
func (s *Server) JetStreamConfig() *JetStreamConfig {
	var c *JetStreamConfig
	if js := s.getJetStream(); js != nil {
		js.mu.RLock()
		copy := js.config
		c = &(copy)
		js.mu.RUnlock()
	}
	return c
}

// StoreDir returns the current JetStream directory.
func (s *Server) StoreDir() string {
	js := s.getJetStream()
	if js == nil {
		return _EMPTY_
	}
	return js.config.StoreDir
}

// JetStreamNumAccounts returns the number of enabled accounts this server is tracking.
func (s *Server) JetStreamNumAccounts() int {
	js := s.getJetStream()
	if js == nil {
		return 0
	}
	js.mu.Lock()
	defer js.mu.Unlock()
	return len(js.accounts)
}

// JetStreamReservedResources returns the reserved resources if JetStream is enabled.
func (s *Server) JetStreamReservedResources() (int64, int64, error) {
	js := s.getJetStream()
	if js == nil {
		return -1, -1, NewJSNotEnabledForAccountError()
	}
	js.mu.RLock()
	defer js.mu.RUnlock()
	return js.memReserved, js.storeReserved, nil
}

func (s *Server) getJetStream() *jetStream {
	return s.js.Load()
}

func (a *Account) assignJetStreamLimits(limits map[string]JetStreamAccountLimits) {
	a.mu.Lock()
	a.jsLimits = limits
	a.mu.Unlock()
}

// EnableJetStream will enable JetStream on this account with the defined limits.
// This is a helper for JetStreamEnableAccount.
func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits, tq chan<- func()) error {
	a.mu.RLock()
	s := a.srv
	a.mu.RUnlock()

	if s == nil {
		return fmt.Errorf("jetstream account not registered")
	}

	if s.SystemAccount() == a {
		return fmt.Errorf("jetstream can not be enabled on the system account")
	}

	s.mu.RLock()
	if s.sys == nil {
		s.mu.RUnlock()
		return ErrServerNotRunning
	}
	sendq := s.sys.sendq
	s.mu.RUnlock()

	// No limits means we dynamically set up limits.
	// We also place limits here so we know that the account is configured for JetStream.
	if len(limits) == 0 {
		limits = defaultJSAccountTiers
	}

	a.assignJetStreamLimits(limits)

	js := s.getJetStream()
	if js == nil {
		return NewJSNotEnabledError()
	}

	js.mu.Lock()
	// Accounts get reset to nil on shutdown, since we re-acquire the locks here, we need to check again.
	if js.accounts == nil {
		js.mu.Unlock()
		return NewJSNotEnabledError()
	}

	if jsa, ok := js.accounts[a.Name]; ok {
		a.mu.Lock()
		a.js = jsa
		a.mu.Unlock()
		js.mu.Unlock()
		return a.enableAllJetStreamServiceImportsAndMappings()
	}

	// Check the limits against existing reservations.
	if err := js.sufficientResources(limits); err != nil {
		js.mu.Unlock()
		return err
	}

	sysNode := s.Node()

	jsa := &jsAccount{js: js, account: a, limits: limits, streams: make(map[string]*stream), sendq: sendq, usage: make(map[string]*jsaStorage)}
	jsa.storeDir = filepath.Join(js.config.StoreDir, a.Name)

	// A single server does not need to do the account updates at this point.
	if js.cluster != nil || !s.standAloneMode() {
		jsa.usageMu.Lock()
		jsa.utimer = time.AfterFunc(usageTick, jsa.sendClusterUsageUpdateTimer)
		// Cluster mode updates to resource usage. System internal prevents echos.
		jsa.updatesPub = fmt.Sprintf(jsaUpdatesPubT, a.Name, sysNode)
		jsa.updatesSub, _ = s.sysSubscribe(fmt.Sprintf(jsaUpdatesSubT, a.Name), jsa.remoteUpdateUsage)
		jsa.usageMu.Unlock()
	}

	js.accounts[a.Name] = jsa
	// Stamp inside account as well. Needs to be done under js's lock.
	a.mu.Lock()
	a.js = jsa
	a.mu.Unlock()
	js.mu.Unlock()

	// Create the proper imports here.
	if err := a.enableAllJetStreamServiceImportsAndMappings(); err != nil {
		return err
	}

	s.Debugf("Enabled JetStream for account %q", a.Name)
	if l, ok := limits[_EMPTY_]; ok {
		s.Debugf("  Max Memory:      %s", friendlyBytes(l.MaxMemory))
		s.Debugf("  Max Storage:     %s", friendlyBytes(l.MaxStore))
	} else {
		for t, l := range limits {
			s.Debugf("  Tier: %s", t)
			s.Debugf("    Max Memory:      %s", friendlyBytes(l.MaxMemory))
			s.Debugf("    Max Storage:     %s", friendlyBytes(l.MaxStore))
		}
	}

	// Clean up any old snapshots that were orphaned while staging.
	os.RemoveAll(filepath.Join(js.config.StoreDir, snapStagingDir))

	sdir := filepath.Join(jsa.storeDir, streamsDir)
	if _, err := os.Stat(sdir); os.IsNotExist(err) {
		if err := os.MkdirAll(sdir, defaultDirPerms); err != nil {
			return fmt.Errorf("could not create storage streams directory - %v", err)
		}
		// Just need to make sure we can write to the directory.
		// Remove the directory will create later if needed.
		os.RemoveAll(sdir)
		// when empty remove parent directory, which may have been created as well
		os.Remove(jsa.storeDir)
	} else {
		// Restore any state here.
		s.Debugf("Recovering JetStream state for account %q", a.Name)
	}

	// Remember if we should be encrypted and what cipher we think we should use.
	encrypted := s.getOpts().JetStreamKey != _EMPTY_
	sc := s.getOpts().JetStreamCipher

	doConsumers := func(mset *stream, odir string) {
		ofis, _ := os.ReadDir(odir)
		if len(ofis) > 0 {
			s.Noticef("  Recovering %d consumers for stream - '%s > %s'", len(ofis), mset.accName(), mset.name())
		}
		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) {
				s.Warnf("    Missing consumer metafile %q", metafile)
				continue
			}
			buf, err := os.ReadFile(metafile)
			if err != nil {
				s.Warnf("    Error reading consumer metafile %q: %v", metafile, err)
				continue
			}
			if _, err := os.Stat(metasum); os.IsNotExist(err) {
				s.Warnf("    Missing consumer checksum for %q", metasum)
				continue
			}

			// Check if we are encrypted.
			if key, err := os.ReadFile(filepath.Join(odir, ofi.Name(), JetStreamMetaFileKey)); err == nil {
				s.Debugf("  Consumer metafile is encrypted, reading encrypted keyfile")
				// Decode the buffer before proceeding.
				ctxName := mset.name() + tsep + ofi.Name()
				nbuf, _, err := s.decryptMeta(sc, key, buf, a.Name, ctxName)
				if err != nil {
					s.Warnf("  Error decrypting our consumer metafile: %v", err)
					continue
				}
				buf = nbuf
			}

			var cfg FileConsumerInfo
			decoder := json.NewDecoder(bytes.NewReader(buf))
			decoder.DisallowUnknownFields()
			strictErr := decoder.Decode(&cfg)
			if strictErr != nil {
				cfg = FileConsumerInfo{}
				if err := json.Unmarshal(buf, &cfg); err != nil {
					s.Warnf("    Error unmarshalling consumer metafile %q: %v", metafile, err)
					continue
				}
			}
			if supported := supportsRequiredApiLevel(cfg.Metadata); !supported || strictErr != nil {
				var offlineReason string
				if !supported {
					apiLevel := getRequiredApiLevel(cfg.Metadata)
					if strictErr != nil {
						offlineReason = fmt.Sprintf("unsupported - config error: %s", strings.TrimPrefix(strictErr.Error(), "json: "))
					} else {
						offlineReason = fmt.Sprintf("unsupported - required API level: %s, current API level: %d", apiLevel, JSApiLevel)
					}
					s.Warnf("  Detected unsupported consumer '%s > %s > %s': %s", a.Name, mset.name(), cfg.Name, offlineReason)
				} else {
					offlineReason = fmt.Sprintf("decoding error: %v", strictErr)
					s.Warnf("  Error unmarshalling consumer metafile %q: %v", metafile, strictErr)
				}
				singleServerMode := !s.JetStreamIsClustered() && s.standAloneMode()
				if singleServerMode {
					if !mset.closed.Load() {
						s.Warnf("  Stopping unsupported stream '%s > %s'", a.Name, mset.name())
						mset.mu.Lock()
						mset.offlineReason = fmt.Sprintf("stopped - unsupported consumer %q", cfg.Name)
						mset.mu.Unlock()
						mset.stop(false, false)
					}

					// Fake a consumer, so we can respond to API requests as single-server.
					o := &consumer{
						mset:          mset,
						js:            s.getJetStream(),
						acc:           a,
						srv:           s,
						cfg:           cfg.ConsumerConfig,
						active:        false,
						stream:        mset.name(),
						name:          cfg.Name,
						dseq:          1,
						sseq:          1,
						created:       time.Now().UTC(),
						closed:        true,
						offlineReason: offlineReason,
					}
					if !cfg.Created.IsZero() {
						o.created = cfg.Created
					}

					mset.mu.Lock()
					mset.setConsumer(o)
					mset.mu.Unlock()
				}
				continue
			}

			isEphemeral := !isDurableConsumer(&cfg.ConsumerConfig)
			if isEphemeral {
				// This is an ephemeral 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.addConsumerWithAssignment(&cfg.ConsumerConfig, _EMPTY_, nil, true, ActionCreateOrUpdate, false)
			if err != nil {
				s.Warnf("    Error adding consumer '%s > %s > %s': %v", a.Name, mset.name(), cfg.Name, err)
				continue
			}
			if isEphemeral {
				obs.switchToEphemeral()
			}
			if !cfg.Created.IsZero() {
				obs.setCreatedTime(cfg.Created)
			}
		}
	}

	// Now recover the streams.
	fis, _ := os.ReadDir(sdir)
	doStream := func(fi os.DirEntry) error {
		plaintext := true
		mdir := filepath.Join(sdir, fi.Name())
		// Check for partially deleted streams. They are marked with "." prefix.
		if strings.HasPrefix(fi.Name(), tsep) {
			go os.RemoveAll(mdir)
			return nil
		}
		key := sha256.Sum256([]byte(fi.Name()))
		hh, err := highwayhash.NewDigest64(key[:])
		if err != nil {
			return err
		}
		metafile := filepath.Join(mdir, JetStreamMetaFile)
		metasum := filepath.Join(mdir, JetStreamMetaFileSum)
		if _, err := os.Stat(metafile); os.IsNotExist(err) {
			s.Warnf("  Missing stream metafile for %q", metafile)
			return nil
		}
		buf, err := os.ReadFile(metafile)
		if err != nil {
			s.Warnf("  Error reading metafile %q: %v", metafile, err)
			return nil
		}
		if _, err := os.Stat(metasum); os.IsNotExist(err) {
			s.Warnf("  Missing stream checksum file %q", metasum)
			return nil
		}
		sum, err := os.ReadFile(metasum)
		if err != nil {
			s.Warnf("  Error reading Stream metafile checksum %q: %v", metasum, err)
			return nil
		}
		hh.Write(buf)
		var hb [highwayhash.Size64]byte
		checksum := hex.EncodeToString(hh.Sum(hb[:0]))
		if checksum != string(sum) {
			s.Warnf("  Stream metafile %q: checksums do not match %q vs %q", metafile, sum, checksum)
			return nil
		}

		// Track if we are converting ciphers.
		var convertingCiphers bool

		// Check if we are encrypted.
		keyFile := filepath.Join(mdir, JetStreamMetaFileKey)
		keyBuf, err := os.ReadFile(keyFile)
		if err == nil {
			s.Debugf("  Stream metafile is encrypted, reading encrypted keyfile")
			if len(keyBuf) < minMetaKeySize {
				s.Warnf("  Bad stream encryption key length of %d", len(keyBuf))
				return nil
			}
			// Decode the buffer before proceeding.
			var nbuf []byte
			nbuf, convertingCiphers, err = s.decryptMeta(sc, keyBuf, buf, a.Name, fi.Name())
			if err != nil {
				s.Warnf("  Error decrypting our stream metafile: %v", err)
				return nil
			}
			buf = nbuf
			plaintext = false
		}

		var cfg FileStreamInfo
		decoder := json.NewDecoder(bytes.NewReader(buf))
		decoder.DisallowUnknownFields()
		strictErr := decoder.Decode(&cfg)
		if strictErr != nil {
			cfg = FileStreamInfo{}
			if err := json.Unmarshal(buf, &cfg); err != nil {
				s.Warnf("  Error unmarshalling stream metafile %q: %v", metafile, err)
				return nil
			}
		}
		if supported := supportsRequiredApiLevel(cfg.Metadata); !supported || strictErr != nil {
			var offlineReason string
			if !supported {
				apiLevel := getRequiredApiLevel(cfg.Metadata)
				if strictErr != nil {
					offlineReason = fmt.Sprintf("unsupported - config error: %s", strings.TrimPrefix(strictErr.Error(), "json: "))
				} else {
					offlineReason = fmt.Sprintf("unsupported - required API level: %s, current API level: %d", apiLevel, JSApiLevel)
				}
				s.Warnf("  Detected unsupported stream '%s > %s': %s", a.Name, cfg.StreamConfig.Name, offlineReason)
			} else {
				offlineReason = fmt.Sprintf("decoding error: %v", strictErr)
				s.Warnf("  Error unmarshalling stream metafile %q: %v", metafile, strictErr)
			}
			singleServerMode := !s.JetStreamIsClustered() && s.standAloneMode()
			if singleServerMode {
				// Fake a stream, so we can respond to API requests as single-server.
				mset := &stream{
					acc:           a,
					jsa:           jsa,
					cfg:           cfg.StreamConfig,
					js:            js,
					srv:           s,
					stype:         cfg.Storage,
					consumers:     make(map[string]*consumer),
					active:        false,
					created:       time.Now().UTC(),
					offlineReason: offlineReason,
				}
				if !cfg.Created.IsZero() {
					mset.created = cfg.Created
				}
				mset.closed.Store(true)

				jsa.mu.Lock()
				jsa.streams[cfg.Name] = mset
				jsa.mu.Unlock()

				// Now do the consumers.
				odir := filepath.Join(sdir, fi.Name(), consumerDir)
				doConsumers(mset, odir)
			}
			return nil
		}

		// We had a bug that set a default de dupe window on mirror, despite that being not a valid config
		fixCfgMirrorWithDedupWindow(&cfg.StreamConfig)

		// We had a bug that could allow subjects in that had prefix or suffix spaces. We check for that here
		// and will patch them on the fly for now. We will warn about them.
		var hadSubjErr bool
		for i, subj := range cfg.StreamConfig.Subjects {
			if !IsValidSubject(subj) {
				s.Warnf("  Detected bad subject %q while adding stream %q, will attempt to repair", subj, cfg.Name)
				if nsubj := strings.TrimSpace(subj); IsValidSubject(nsubj) {
					s.Warnf("  Bad subject %q repaired to %q", subj, nsubj)
					cfg.StreamConfig.Subjects[i] = nsubj
				} else {
					s.Warnf("  Error recreating stream %q: %v", cfg.Name, "invalid subject")
					hadSubjErr = true
					break
				}
			}
		}
		if hadSubjErr {
			return nil
		}

		// The other possible bug is assigning subjects to mirrors, so check for that and patch as well.
		if cfg.StreamConfig.Mirror != nil && len(cfg.StreamConfig.Subjects) > 0 {
			s.Warnf("  Detected subjects on a mirrored stream %q, will remove", cfg.Name)
			cfg.StreamConfig.Subjects = nil
		}

		s.Noticef("  Starting restore for stream '%s > %s'", a.Name, cfg.StreamConfig.Name)
		rt := time.Now()

		// Log if we are converting from plaintext to encrypted.
		if encrypted {
			if plaintext {
				s.Noticef("  Encrypting stream '%s > %s'", a.Name, cfg.StreamConfig.Name)
			} else if convertingCiphers {
				s.Noticef("  Converting to %s for stream '%s > %s'", sc, a.Name, cfg.StreamConfig.Name)
				// Remove the key file to have system regenerate with the new cipher.
				os.Remove(keyFile)
			}
		}

		// Add in the stream.
		mset, err := a.recoverStream(&cfg.StreamConfig)
		if err != nil {
			s.Warnf("  Error recreating stream %q: %v", cfg.Name, err)
			// If we removed a keyfile from above make sure to put it back.
			if convertingCiphers {
				err := os.WriteFile(keyFile, keyBuf, defaultFilePerms)
				if err != nil {
					s.Warnf("  Error replacing meta keyfile for stream %q: %v", cfg.Name, err)
				}
			}
			return nil
		}
		if !cfg.Created.IsZero() {
			mset.setCreatedTime(cfg.Created)
		}

		// Might need to recover from a partial batch write, but only if a single replica stream.
		if cfg.AllowAtomicPublish && cfg.Replicas == 1 {
			var (
				ok            bool
				smv           StoreMsg
				batchId       string
				batchSeq      uint64
				commit        bool
				commitEob     bool
				batchStoreDir string
				store         StreamStore
				state         StreamState
			)
			// Check if the last message was part of a batch.
			sm, err := mset.store.LoadLastMsg(fwcs, &smv)
			if err != nil || sm == nil {
				goto SKIP
			}
			batchId = getBatchId(sm.hdr)
			batchSeq, ok = getBatchSequence(sm.hdr)
			commit = len(sliceHeader(JSBatchCommit, sm.hdr)) != 0
			if batchId == _EMPTY_ || !ok || commit {
				goto SKIP
			}
			// We've observed a partial batch write. Write the remainder of the batch.
			batchSeq++
			_, batchStoreDir = getBatchStoreDir(jsa.storeDir, cfg.Name, batchId)
			if _, err = os.Stat(batchStoreDir); err != nil {
				s.Errorf("  Failed restoring partial batch write for stream '%s > %s' at sequence %d: %v",
					mset.accName(), mset.name(), batchSeq, err)
				goto SKIP
			}
			store, err = newBatchStore(mset, batchId, cfg.Replicas, cfg.Storage, jsa.storeDir, cfg.Name)
			if err != nil {
				s.Errorf("  Failed restoring partial batch write for stream '%s > %s' at sequence %d: %v",
					mset.accName(), mset.name(), batchSeq, err)
				goto SKIP
			}
			store.FastState(&state)
			sm, err = store.LoadMsg(state.LastSeq, &smv)
			if err != nil || sm == nil {
				s.Errorf("  Failed restoring partial batch write for stream '%s > %s' at sequence %d: last msg not found %d",
					mset.accName(), mset.name(), batchSeq, state.LastSeq)
				goto SKIP
			}
			commitEob = bytes.Equal(sliceHeader(JSBatchCommit, sm.hdr), []byte("eob"))
			// If the commit ends with an "End Of Batch" message, we don't store this.
			if commitEob {
				state.LastSeq--
			}
			s.Noticef("  Restoring partial batch write for stream '%s > %s' (seq %d to %d)",
				mset.accName(), mset.name(), batchSeq, state.LastSeq)
			// Loop through items that weren't persisted yet.
			for seq := batchSeq; seq <= state.LastSeq; seq++ {
				sm, err = store.LoadMsg(seq, &smv)
				if err != nil || sm == nil {
					s.Errorf("  Failed restoring partial batch write for stream '%s > %s' at sequence %d: %v",
						mset.accName(), mset.name(), seq, err)
					break
				}
				hdr := sm.hdr
				// If committed by EOB, the last message must get the normal commit header.
				if commitEob && seq == state.LastSeq {
					hdr = genHeader(hdr, JSBatchCommit, "1")
				}
				mset.processJetStreamMsg(sm.subj, _EMPTY_, hdr, sm.msg, 0, 0, nil, false, true)
			}
			store.Delete(true)
		SKIP:
			os.RemoveAll(filepath.Join(sdir, fi.Name(), batchesDir))
		}

		state := mset.state()
		s.Noticef("  Restored %s messages for stream '%s > %s' in %v",
			comma(int64(state.Msgs)), mset.accName(), mset.name(), time.Since(rt).Round(time.Millisecond))

		// Now do the consumers.
		odir := filepath.Join(sdir, fi.Name(), consumerDir)
		doConsumers(mset, odir)

		// Collect to check for dangling messages.
		// TODO(dlc) - Can be removed eventually.
		if cfg.StreamConfig.Retention == InterestPolicy {
			mset.checkForOrphanMsgs()
			mset.checkConsumerReplication()
		}

		return nil
	}

	if tq != nil {
		// If a parallelTaskQueue was provided then use that for concurrency.
		var wg sync.WaitGroup
		wg.Add(len(fis))
		for _, fi := range fis {
			tq <- func() {
				doStream(fi)
				wg.Done()
			}
		}
		wg.Wait()
	} else {
		// No parallelTaskQueue provided, do inline as before.
		for _, fi := range fis {
			doStream(fi)
		}
	}

	// Make sure to cleanup any old remaining snapshots.
	os.RemoveAll(filepath.Join(jsa.storeDir, snapsDir))

	s.Debugf("JetStream state for account %q recovered", a.Name)

	return nil
}

// Return whether we require MaxBytes to be set and if > 0 an upper limit for stream size exists
// Both limits are independent of each other.
func (a *Account) maxBytesLimits(cfg *StreamConfig) (bool, int64) {
	a.mu.RLock()
	jsa := a.js
	a.mu.RUnlock()
	if jsa == nil {
		return false, 0
	}
	jsa.usageMu.RLock()
	var replicas int
	if cfg != nil {
		replicas = cfg.Replicas
	}
	selectedLimits, _, ok := jsa.selectLimits(replicas)
	jsa.usageMu.RUnlock()
	if !ok {
		return false, 0
	}
	maxStreamBytes := int64(0)
	if cfg.Storage == MemoryStorage {
		maxStreamBytes = selectedLimits.MemoryMaxStreamBytes
	} else {
		maxStreamBytes = selectedLimits.StoreMaxStreamBytes
	}
	return selectedLimits.MaxBytesRequired, maxStreamBytes
}

// NumStreams will return how many streams we have.
func (a *Account) numStreams() int {
	a.mu.RLock()
	jsa := a.js
	a.mu.RUnlock()
	if jsa == nil {
		return 0
	}
	jsa.mu.Lock()
	n := len(jsa.streams)
	jsa.mu.Unlock()
	return n
}

// Streams will return all known streams.
func (a *Account) streams() []*stream {
	return a.filteredStreams(_EMPTY_)
}

func (a *Account) filteredStreams(filter string) []*stream {
	a.mu.RLock()
	jsa := a.js
	a.mu.RUnlock()

	if jsa == nil {
		return nil
	}

	jsa.mu.RLock()
	defer jsa.mu.RUnlock()

	var msets []*stream
	for _, mset := range jsa.streams {
		if filter != _EMPTY_ {
			mset.cfgMu.RLock()
			for _, subj := range mset.cfg.Subjects {
				if SubjectsCollide(filter, subj) {
					msets = append(msets, mset)
					break
				}
			}
			mset.cfgMu.RUnlock()
		} else {
			msets = append(msets, mset)
		}
	}

	return msets
}

// lookupStream will lookup a stream by name.
func (a *Account) lookupStream(name string) (*stream, error) {
	a.mu.RLock()
	jsa := a.js
	a.mu.RUnlock()

	if jsa == nil {
		return nil, NewJSNotEnabledForAccountError()
	}
	jsa.mu.RLock()
	defer jsa.mu.RUnlock()

	mset, ok := jsa.streams[name]
	if !ok {
		return nil, NewJSStreamNotFoundError()
	}
	return mset, nil
}

// UpdateJetStreamLimits will update the account limits for a JetStream enabled account.
func (a *Account) UpdateJetStreamLimits(limits map[string]JetStreamAccountLimits) error {
	a.mu.RLock()
	s, jsa := a.srv, a.js
	a.mu.RUnlock()

	if s == nil {
		return fmt.Errorf("jetstream account not registered")
	}
	js := s.getJetStream()
	if js == nil {
		return NewJSNotEnabledError()
	}
	if jsa == nil {
		return NewJSNotEnabledForAccountError()
	}

	if len(limits) == 0 {
		limits = defaultJSAccountTiers
	}

	// Calculate the delta between what we have and what we want.
	jsa.usageMu.RLock()
	dl := diffCheckedLimits(jsa.limits, limits)
	jsa.usageMu.RUnlock()

	js.mu.Lock()
	// Check the limits against existing reservations.
	if err := js.sufficientResources(dl); err != nil {
		js.mu.Unlock()
		return err
	}
	js.mu.Unlock()

	// Update
	jsa.usageMu.Lock()
	jsa.limits = limits
	jsa.usageMu.Unlock()

	return nil
}

func diffCheckedLimits(a, b map[string]JetStreamAccountLimits) map[string]JetStreamAccountLimits {
	diff := map[string]JetStreamAccountLimits{}
	for t, la := range a {
		// in a, not in b will return 0
		lb := b[t]
		diff[t] = JetStreamAccountLimits{
			MaxMemory: lb.MaxMemory - la.MaxMemory,
			MaxStore:  lb.MaxStore - la.MaxStore,
		}
	}
	for t, lb := range b {
		if la, ok := a[t]; !ok {
			// only in b not in a. (in a and b already covered)
			diff[t] = JetStreamAccountLimits{
				MaxMemory: lb.MaxMemory - la.MaxMemory,
				MaxStore:  lb.MaxStore - la.MaxStore,
			}
		}
	}
	return diff
}

// Return reserved bytes for memory and file store streams for this account on this server.
// Lock should be held.
func (jsa *jsAccount) reservedStorage(tier string) (mem, store uint64) {
	for _, mset := range jsa.streams {
		cfg := &mset.cfg
		if (tier == _EMPTY_ || tier == tierName(cfg.Replicas)) && cfg.MaxBytes > 0 {
			switch cfg.Storage {
			case FileStorage:
				store += uint64(cfg.MaxBytes)
			case MemoryStorage:
				mem += uint64(cfg.MaxBytes)
			}
		}
	}
	return mem, store
}

// Return reserved bytes for memory and file store streams for this account in clustered mode.
// js lock should be held.
func reservedStorage(sas map[string]*streamAssignment, tier string) (mem, store uint64) {
	for _, sa := range sas {
		cfg := sa.Config
		if (tier == _EMPTY_ || tier == tierName(cfg.Replicas)) && cfg.MaxBytes > 0 {
			switch cfg.Storage {
			case FileStorage:
				store += uint64(cfg.MaxBytes)
			case MemoryStorage:
				mem += uint64(cfg.MaxBytes)
			}
		}
	}
	return mem, store
}

// JetStreamUsage reports on JetStream usage and limits for an account.
func (a *Account) JetStreamUsage() JetStreamAccountStats {
	a.mu.RLock()
	jsa, aname := a.js, a.Name
	accJsLimits := a.jsLimits
	a.mu.RUnlock()

	var stats JetStreamAccountStats
	if jsa != nil {
		js := jsa.js
		js.mu.RLock()
		cc := js.cluster
		singleServer := cc == nil
		jsa.mu.RLock()
		jsa.usageMu.RLock()
		stats.Memory, stats.Store = jsa.storageTotals()
		stats.Domain = js.config.Domain
		stats.API = JetStreamAPIStats{
			Level:  JSApiLevel,
			Total:  jsa.apiTotal,
			Errors: jsa.apiErrors,
		}
		if singleServer {
			stats.ReservedMemory, stats.ReservedStore = jsa.reservedStorage(_EMPTY_)
		} else {
			stats.ReservedMemory, stats.ReservedStore = reservedStorage(cc.streams[aname], _EMPTY_)
		}
		l, defaultTier := jsa.limits[_EMPTY_]
		if defaultTier {
			stats.Limits = l
		} else {
			skipped := 0
			stats.Tiers = make(map[string]JetStreamTier)
			for t, total := range jsa.usage {
				if _, ok := jsa.limits[t]; !ok && (*total) == (jsaStorage{}) {
					// skip tiers not present that don't contain a count
					// In case this shows an empty stream, that tier will be added when iterating over streams
					skipped++
				} else {
					tier := JetStreamTier{
						Memory: uint64(total.total.mem),
						Store:  uint64(total.total.store),
						Limits: jsa.limits[t],
					}
					if singleServer {
						tier.ReservedMemory, tier.ReservedStore = jsa.reservedStorage(t)
					} else {
						tier.ReservedMemory, tier.ReservedStore = reservedStorage(cc.streams[aname], t)
					}
					stats.Tiers[t] = tier
				}
			}
			if len(accJsLimits) != len(jsa.usage)-skipped {
				// insert unused limits
				for t, lim := range accJsLimits {
					if _, ok := stats.Tiers[t]; !ok {
						tier := JetStreamTier{Limits: lim}
						if singleServer {
							tier.ReservedMemory, tier.ReservedStore = jsa.reservedStorage(t)
						} else {
							tier.ReservedMemory, tier.ReservedStore = reservedStorage(cc.streams[aname], t)
						}
						stats.Tiers[t] = tier
					}
				}
			}
		}
		jsa.usageMu.RUnlock()

		// Clustered
		if cc := js.cluster; cc != nil {
			sas := cc.streams[aname]
			if defaultTier {
				stats.Streams = len(sas)
				stats.ReservedMemory, stats.ReservedStore = reservedStorage(sas, _EMPTY_)
			}
			for _, sa := range sas {
				if defaultTier {
					stats.Consumers += len(sa.consumers)
				} else {
					stats.Streams++
					streamTier := tierName(sa.Config.Replicas)
					su, ok := stats.Tiers[streamTier]
					if !ok {
						su = JetStreamTier{}
					}
					su.Streams++
					stats.Tiers[streamTier] = su

					// Now consumers, check each since could be different tiers.
					for _, ca := range sa.consumers {
						stats.Consumers++
						consumerTier := tierName(ca.Config.replicas(sa.Config))
						cu, ok := stats.Tiers[consumerTier]
						if !ok {
							cu = JetStreamTier{}
						}
						cu.Consumers++
						stats.Tiers[consumerTier] = cu
					}
				}
			}
		} else {
			if defaultTier {
				stats.Streams = len(jsa.streams)
			}
			for _, mset := range jsa.streams {
				consCount := mset.numConsumers()
				stats.Consumers += consCount
				if !defaultTier {
					u, ok := stats.Tiers[mset.tier]
					if !ok {
						u = JetStreamTier{}
					}
					u.Streams++
					stats.Streams++
					u.Consumers += consCount
					stats.Tiers[mset.tier] = u
				}
			}
		}
		jsa.mu.RUnlock()
		js.mu.RUnlock()
	}
	return stats
}

// DisableJetStream will disable JetStream for this account.
func (a *Account) DisableJetStream() error {
	return a.removeJetStream()
}

// removeJetStream is called when JetStream has been disabled for this account.
func (a *Account) removeJetStream() error {
	a.mu.Lock()
	s := a.srv
	a.js = nil
	a.mu.Unlock()

	if s == nil {
		return fmt.Errorf("jetstream account not registered")
	}

	js := s.getJetStream()
	if js == nil {
		return NewJSNotEnabledForAccountError()
	}

	return js.disableJetStream(js.lookupAccount(a))
}

// Disable JetStream for the account.
func (js *jetStream) disableJetStream(jsa *jsAccount) error {
	if jsa == nil || jsa.account == nil {
		return NewJSNotEnabledForAccountError()
	}

	js.mu.Lock()
	delete(js.accounts, jsa.account.Name)
	js.mu.Unlock()

	jsa.delete()

	return nil
}

// jetStreamConfigured reports whether the account has JetStream configured, regardless of this
// servers JetStream status.
func (a *Account) jetStreamConfigured() bool {
	if a == nil {
		return false
	}
	a.mu.RLock()
	defer a.mu.RUnlock()
	return len(a.jsLimits) > 0
}

// JetStreamEnabled is a helper to determine if jetstream is enabled for an account.
func (a *Account) JetStreamEnabled() bool {
	if a == nil {
		return false
	}
	a.mu.RLock()
	enabled := a.js != nil
	a.mu.RUnlock()
	return enabled
}

func (jsa *jsAccount) remoteUpdateUsage(sub *subscription, c *client, _ *Account, subject, _ string, msg []byte) {
	// jsa.js.srv is immutable and guaranteed to no be nil, so no lock needed.
	s := jsa.js.srv

	jsa.usageMu.Lock()
	defer jsa.usageMu.Unlock()

	if len(msg) < minUsageUpdateLen {
		s.Warnf("Ignoring remote usage update with size too short")
		return
	}
	var rnode string
	if li := strings.LastIndexByte(subject, btsep); li > 0 && li < len(subject) {
		rnode = subject[li+1:]
	}
	if rnode == _EMPTY_ {
		s.Warnf("Received remote usage update with no remote node")
		return
	}
	rUsage, ok := jsa.rusage[rnode]
	if !ok {
		if jsa.rusage == nil {
			jsa.rusage = make(map[string]*remoteUsage)
		}
		rUsage = &remoteUsage{tiers: make(map[string]*jsaUsage)}
		jsa.rusage[rnode] = rUsage
	}
	updateTotal := func(tierName string, memUsed, storeUsed int64) {
		total, ok := jsa.usage[tierName]
		if !ok {
			total = &jsaStorage{}
			jsa.usage[tierName] = total
		}
		// Update the usage for this remote.
		if usage := rUsage.tiers[tierName]; usage != nil {
			// Decrement our old values.
			total.total.mem -= usage.mem
			total.total.store -= usage.store
			usage.mem, usage.store = memUsed, storeUsed
		} else {
			rUsage.tiers[tierName] = &jsaUsage{memUsed, storeUsed}
		}
		total.total.mem += memUsed
		total.total.store += storeUsed
	}

	var le = binary.LittleEndian
	apiTotal, apiErrors := le.Uint64(msg[16:]), le.Uint64(msg[24:])
	memUsed, storeUsed := int64(le.Uint64(msg[0:])), int64(le.Uint64(msg[8:]))

	// We later extended the data structure to support multiple tiers
	var excessRecordCnt uint32
	var tierName string

	if len(msg) >= usageMultiTiersLen {
		excessRecordCnt = le.Uint32(msg[minUsageUpdateLen:])
		length := le.Uint64(msg[minUsageUpdateLen+4:])
		// Need to protect past this point in case this is wrong.
		if uint64(len(msg)) < usageMultiTiersLen+length {
			s.Warnf("Received corrupt remote usage update")
			return
		}
		tierName = string(msg[usageMultiTiersLen : usageMultiTiersLen+length])
		msg = msg[usageMultiTiersLen+length:]
	}
	updateTotal(tierName, memUsed, storeUsed)
	for ; excessRecordCnt > 0 && len(msg) >= usageRecordLen; excessRecordCnt-- {
		memUsed, storeUsed := int64(le.Uint64(msg[0:])), int64(le.Uint64(msg[8:]))
		length := le.Uint64(msg[16:])
		if uint64(len(msg)) < usageRecordLen+length {
			s.Warnf("Received corrupt remote usage update on excess record")
			return
		}
		tierName = string(msg[usageRecordLen : usageRecordLen+length])
		msg = msg[usageRecordLen+length:]
		updateTotal(tierName, memUsed, storeUsed)
	}
	jsa.apiTotal -= rUsage.api
	jsa.apiErrors -= rUsage.err
	rUsage.api = apiTotal
	rUsage.err = apiErrors
	jsa.apiTotal += apiTotal
	jsa.apiErrors += apiErrors
}

// When we detect a skew of some sort this will verify the usage reporting is correct.
// No locks should be held.
func (jsa *jsAccount) checkAndSyncUsage(tierName string, storeType StorageType) {
	// This will run in a separate go routine, so check that we are only running once.
	if !jsa.sync.CompareAndSwap(false, true) {
		return
	}
	defer jsa.sync.Store(false)

	// Hold the account read lock and the usage lock while we calculate.
	// We scope by tier and storage type, but if R3 File has 200 streams etc. could
	// show a pause. I did test with > 100 non-active streams and was 80-200ns or so.
	// Should be rare this gets called as well.
	jsa.mu.RLock()
	defer jsa.mu.RUnlock()
	js := jsa.js
	if js == nil {
		return
	}
	s := js.srv

	// We need to collect the stream stores before we acquire the usage lock since in storeUpdates the
	// stream lock could be held if deletion are inline with storing a new message, e.g. via limits.
	var stores []StreamStore
	for _, mset := range jsa.streams {
		mset.mu.RLock()
		if mset.tier == tierName && mset.stype == storeType && mset.store != nil {
			stores = append(stores, mset.store)
		}
		mset.mu.RUnlock()
	}

	// Now range and qualify, hold usage lock to prevent updates.
	jsa.usageMu.Lock()
	defer jsa.usageMu.Unlock()

	usage, ok := jsa.usage[tierName]
	if !ok {
		return
	}

	// Collect current total for all stream stores that matched.
	var total int64
	var state StreamState
	for _, store := range stores {
		store.FastState(&state)
		total += int64(state.Bytes)
	}

	var needClusterUpdate bool
	// If we do not match on our calculations compute delta and adjust.
	if storeType == MemoryStorage {
		if total != usage.local.mem {
			s.Warnf("MemStore usage drift of %v vs %v detected for account %q",
				friendlyBytes(total), friendlyBytes(usage.local.mem), jsa.account.GetName())
			delta := total - usage.local.mem
			usage.local.mem += delta
			usage.total.mem += delta
			atomic.AddInt64(&js.memUsed, delta)
			needClusterUpdate = true
		}
	} else {
		if total != usage.local.store {
			s.Warnf("FileStore usage drift of %v vs %v detected for account %q",
				friendlyBytes(total), friendlyBytes(usage.local.store), jsa.account.GetName())
			delta := total - usage.local.store
			usage.local.store += delta
			usage.total.store += delta
			atomic.AddInt64(&js.storeUsed, delta)
			needClusterUpdate = true
		}
	}

	// Publish our local updates if in clustered mode.
	if needClusterUpdate && js.isClusteredNoLock() {
		jsa.sendClusterUsageUpdate()
	}
}

// Updates accounting on in use memory and storage. This is called from locally
// by the lower storage layers.
func (jsa *jsAccount) updateUsage(tierName string, storeType StorageType, delta int64) {
	// jsa.js is immutable and cannot be nil, so ok w/o lock.
	js := jsa.js
	// updateUsage() may be invoked under the mset's lock, so we can't get
	// the js' lock to check if clustered. So use this function that make
	// use of an atomic to do the check without having data race reports.
	isClustered := js.isClusteredNoLock()

	var needsCheck bool
	jsa.usageMu.Lock()
	s, ok := jsa.usage[tierName]
	if !ok {
		s = &jsaStorage{}
		jsa.usage[tierName] = s
	}
	if storeType == MemoryStorage {
		s.local.mem += delta
		s.total.mem += delta
		atomic.AddInt64(&js.memUsed, delta)
		needsCheck = s.local.mem < 0
	} else {
		s.local.store += delta
		s.total.store += delta
		atomic.AddInt64(&js.storeUsed, delta)
		needsCheck = s.local.store < 0
	}
	// Publish our local updates if in clustered mode.
	if isClustered {
		jsa.sendClusterUsageUpdate()
	}
	jsa.usageMu.Unlock()

	if needsCheck {
		// We could be holding the stream lock from up in the stack, and this
		// will want the jsa lock, which would violate locking order.
		// So do this in a Go routine. The function will check if it is already running.
		go jsa.checkAndSyncUsage(tierName, storeType)
	}
}

var usageTick = 1500 * time.Millisecond

func (jsa *jsAccount) sendClusterUsageUpdateTimer() {
	jsa.usageMu.Lock()
	defer jsa.usageMu.Unlock()
	jsa.sendClusterUsageUpdate()
	if jsa.utimer != nil {
		jsa.utimer.Reset(usageTick)
	}
}

// For usage fields.
const (
	minUsageUpdateLen    = 32
	stackUsageUpdate     = 72
	usageRecordLen       = 24
	usageMultiTiersLen   = 44
	apiStatsAndNumTiers  = 20
	minUsageUpdateWindow = 250 * time.Millisecond
)

// Send updates to our account usage for this server.
// jsa.usageMu lock should be held.
func (jsa *jsAccount) sendClusterUsageUpdate() {
	// These values are absolute so we can limit send rates.
	now := time.Now()
	if now.Sub(jsa.lupdate) < minUsageUpdateWindow {
		return
	}
	jsa.lupdate = now

	lenUsage := len(jsa.usage)
	if lenUsage == 0 {
		return
	}
	// every base record contains mem/store/len(tier) as well as the tier name
	l := usageRecordLen * lenUsage
	for tier := range jsa.usage {
		l += len(tier)
	}
	// first record contains api/usage errors as well as count for extra base records
	l += apiStatsAndNumTiers

	var raw [stackUsageUpdate]byte
	var b []byte
	if l > stackUsageUpdate {
		b = make([]byte, l)
	} else {
		b = raw[:l]
	}

	var i int
	var le = binary.LittleEndian
	for tier, usage := range jsa.usage {
		le.PutUint64(b[i+0:], uint64(usage.local.mem))
		le.PutUint64(b[i+8:], uint64(usage.local.store))
		if i == 0 {
			le.PutUint64(b[16:], jsa.usageApi)
			le.PutUint64(b[24:], jsa.usageErr)
			le.PutUint32(b[32:], uint32(len(jsa.usage)-1))
			le.PutUint64(b[36:], uint64(len(tier)))
			copy(b[usageMultiTiersLen:], tier)
			i = usageMultiTiersLen + len(tier)
		} else {
			le.PutUint64(b[i+16:], uint64(len(tier)))
			copy(b[i+usageRecordLen:], tier)
			i += usageRecordLen + len(tier)
		}
	}
	jsa.sendq.push(newPubMsg(nil, jsa.updatesPub, _EMPTY_, nil, nil, b, noCompression, false, false))
}

func (js *jetStream) wouldExceedLimits(storeType StorageType, sz int) bool {
	var (
		total *int64
		max   *int64
	)
	if storeType == MemoryStorage {
		total, max = &js.memUsed, &js.memMax
	} else {
		total, max = &js.storeUsed, &js.storeMax
	}
	return (atomic.LoadInt64(total) + int64(sz)) > atomic.LoadInt64(max)
}

func (js *jetStream) limitsExceeded(storeType StorageType) bool {
	return js.wouldExceedLimits(storeType, 0)
}

func tierName(replicas int) string {
	// TODO (mh) this is where we could select based off a placement tag as well "qos:tier"
	if replicas == 0 {
		replicas = 1
	}
	return fmt.Sprintf("R%d", replicas)
}

func isSameTier(cfgA, cfgB *StreamConfig) bool {
	a := max(1, cfgA.Replicas)
	b := max(1, cfgB.Replicas)
	// TODO (mh) this is where we could select based off a placement tag as well "qos:tier"
	return a == b
}

func (jsa *jsAccount) jetStreamAndClustered() (*jetStream, bool) {
	jsa.mu.RLock()
	js := jsa.js
	jsa.mu.RUnlock()
	return js, js.isClustered()
}

// jsa.usageMu read lock should be held.
func (jsa *jsAccount) selectLimits(replicas int) (JetStreamAccountLimits, string, bool) {
	if selectedLimits, ok := jsa.limits[_EMPTY_]; ok {
		return selectedLimits, _EMPTY_, true
	}
	tier := tierName(replicas)
	if selectedLimits, ok := jsa.limits[tier]; ok {
		return selectedLimits, tier, true
	}
	return JetStreamAccountLimits{}, _EMPTY_, false
}

// Lock should be held.
func (jsa *jsAccount) countStreams(tier string, cfg *StreamConfig) (streams int) {
	for _, sa := range jsa.streams {
		// Don't count the stream toward the limit if it already exists.
		if (tier == _EMPTY_ || isSameTier(&sa.cfg, cfg)) && sa.cfg.Name != cfg.Name {
			streams++
		}
	}
	return streams
}

// jsa.usageMu read lock (at least) should be held.
func (jsa *jsAccount) storageTotals() (uint64, uint64) {
	mem := uint64(0)
	store := uint64(0)
	for _, sa := range jsa.usage {
		mem += uint64(sa.total.mem)
		store += uint64(sa.total.store)
	}
	return mem, store
}

func (jsa *jsAccount) limitsExceeded(storeType StorageType, tierName string, replicas int) (bool, *ApiError) {
	return jsa.wouldExceedLimits(storeType, tierName, replicas, _EMPTY_, nil, nil)
}

func (jsa *jsAccount) wouldExceedLimits(storeType StorageType, tierName string, replicas int, subj string, hdr, msg []byte) (bool, *ApiError) {
	jsa.usageMu.RLock()
	defer jsa.usageMu.RUnlock()

	selectedLimits, ok := jsa.limits[tierName]
	if !ok {
		return true, NewJSNoLimitsError()
	}
	inUse := jsa.usage[tierName]
	if inUse == nil {
		// Imply totals of 0
		return false, nil
	}
	r := int64(replicas)
	// Make sure replicas is correct.
	if r < 1 {
		r = 1
	}
	// This is for limits. If we have no tier, consider all to be flat, vs tiers like R3 where we want to scale limit by replication.
	lr := r
	if tierName == _EMPTY_ {
		lr = 1
	}

	// Since tiers are flat we need to scale limit up by replicas when checking.
	if storeType == MemoryStorage {
		totalMem := inUse.total.mem + (int64(memStoreMsgSize(subj, hdr, msg)) * r)
		if selectedLimits.MaxMemory >= 0 && totalMem > selectedLimits.MaxMemory*lr {
			return true, nil
		}
	} else {
		totalStore := inUse.total.store + (int64(fileStoreMsgSize(subj, hdr, msg)) * r)
		if selectedLimits.MaxStore >= 0 && totalStore > selectedLimits.MaxStore*lr {
			return true, nil
		}
	}

	return false, nil
}

// Check account limits.
// Read Lock should be held
func (js *jetStream) checkAccountLimits(selected *JetStreamAccountLimits, config *StreamConfig, currentRes int64) error {
	return js.checkLimits(selected, config, false, currentRes, 0)
}

// Check account and server limits.
// Read Lock should be held
func (js *jetStream) checkAllLimits(selected *JetStreamAccountLimits, config *StreamConfig, currentRes, maxBytesOffset int64) error {
	return js.checkLimits(selected, config, true, currentRes, maxBytesOffset)
}

// Check if a new proposed msg set while exceed our account limits.
// Lock should be held.
func (js *jetStream) checkLimits(selected *JetStreamAccountLimits, config *StreamConfig, checkServer bool, currentRes, maxBytesOffset int64) error {
	// Check MaxConsumers
	if config.MaxConsumers > 0 && selected.MaxConsumers > 0 && config.MaxConsumers > selected.MaxConsumers {
		return NewJSMaximumConsumersLimitError()
	}
	// stream limit is checked separately on stream create only!
	// Check storage, memory or disk.
	return js.checkBytesLimits(selected, config.MaxBytes, config.Storage, checkServer, currentRes, maxBytesOffset)
}

// Check if additional bytes will exceed our account limits and optionally the server itself.
// Read Lock should be held.
func (js *jetStream) checkBytesLimits(selectedLimits *JetStreamAccountLimits, addBytes int64, storage StorageType, checkServer bool, currentRes, maxBytesOffset int64) error {
	if addBytes < 0 {
		addBytes = 1
	}
	totalBytes := addSaturate(addBytes, maxBytesOffset)

	switch storage {
	case MemoryStorage:
		// Account limits defined.
		if selectedLimits.MaxMemory >= 0 && (currentRes > selectedLimits.MaxMemory || totalBytes > selectedLimits.MaxMemory-currentRes) {
			return NewJSMemoryResourcesExceededError()
		}
		// Check if this server can handle request.
		if checkServer && (js.memReserved > js.config.MaxMemory || totalBytes > js.config.MaxMemory-js.memReserved) {
			return NewJSMemoryResourcesExceededError()
		}
	case FileStorage:
		// Account limits defined.
		if selectedLimits.MaxStore >= 0 && (currentRes > selectedLimits.MaxStore || totalBytes > selectedLimits.MaxStore-currentRes) {
			return NewJSStorageResourcesExceededError()
		}
		// Check if this server can handle request.
		if checkServer && (js.storeReserved > js.config.MaxStore || totalBytes > js.config.MaxStore-js.storeReserved) {
			return NewJSStorageResourcesExceededError()
		}
	}

	return nil
}

func (jsa *jsAccount) acc() *Account {
	return jsa.account
}

// Delete the JetStream resources.
func (jsa *jsAccount) delete() {
	var streams []*stream

	jsa.mu.Lock()
	// The update timer and subs need to be protected by usageMu lock
	jsa.usageMu.Lock()
	if jsa.utimer != nil {
		jsa.utimer.Stop()
		jsa.utimer = nil
	}
	if jsa.updatesSub != nil && jsa.js.srv != nil {
		s := jsa.js.srv
		s.sysUnsubscribe(jsa.updatesSub)
		jsa.updatesSub = nil
	}
	jsa.usageMu.Unlock()

	for _, ms := range jsa.streams {
		streams = append(streams, ms)
	}
	jsa.mu.Unlock()

	for _, mset := range streams {
		mset.stop(false, false)
	}
}

// Lookup the jetstream account for a given account.
func (js *jetStream) lookupAccount(a *Account) *jsAccount {
	if a == nil {
		return nil
	}
	js.mu.RLock()
	jsa := js.accounts[a.Name]
	js.mu.RUnlock()
	return jsa
}

// Report on JetStream stats and usage for this server.
func (js *jetStream) usageStats() *JetStreamStats {
	var stats JetStreamStats
	js.mu.RLock()
	stats.Accounts = len(js.accounts)
	stats.ReservedMemory = uint64(js.memReserved)
	stats.ReservedStore = uint64(js.storeReserved)
	s := js.srv
	js.mu.RUnlock()
	stats.API.Level = JSApiLevel
	stats.API.Total = uint64(atomic.LoadInt64(&js.apiTotal))
	stats.API.Errors = uint64(atomic.LoadInt64(&js.apiErrors))
	stats.API.Inflight = uint64(atomic.LoadInt64(&js.apiInflight))
	// Make sure we do not report negative.
	used := atomic.LoadInt64(&js.memUsed)
	if used < 0 {
		used = 0
	}
	stats.Memory = uint64(used)
	used = atomic.LoadInt64(&js.storeUsed)
	if used < 0 {
		used = 0
	}
	stats.Store = uint64(used)
	stats.HAAssets = s.numRaftNodes()
	return &stats
}

// Check to see if we have enough system resources for this account.
// Lock should be held.
func (js *jetStream) sufficientResources(limits map[string]JetStreamAccountLimits) error {
	// If we are clustered we do not really know how many resources will be ultimately available.
	// This needs to be handled out of band.
	// If we are a single server, we can make decisions here.
	if limits == nil || !js.standAlone {
		return nil
	}

	totalMaxBytes := func(limits map[string]JetStreamAccountLimits) (int64, int64) {
		totalMaxMemory := int64(0)
		totalMaxStore := int64(0)
		for _, l := range limits {
			if l.MaxMemory > 0 {
				totalMaxMemory += l.MaxMemory
			}
			if l.MaxStore > 0 {
				totalMaxStore += l.MaxStore
			}
		}
		return totalMaxMemory, totalMaxStore
	}

	totalMaxMemory, totalMaxStore := totalMaxBytes(limits)

	// Reserved is now specific to the MaxBytes for streams.
	if js.memReserved+totalMaxMemory > js.config.MaxMemory {
		return NewJSMemoryResourcesExceededError()
	}
	if js.storeReserved+totalMaxStore > js.config.MaxStore {
		return NewJSStorageResourcesExceededError()
	}

	// Since we know if we are here we are single server mode, check the account reservations.
	var storeReserved, memReserved int64
	for _, jsa := range js.accounts {
		if jsa.account.IsExpired() {
			continue
		}
		jsa.usageMu.RLock()
		maxMemory, maxStore := totalMaxBytes(jsa.limits)
		jsa.usageMu.RUnlock()
		memReserved += maxMemory
		storeReserved += maxStore
	}

	if memReserved+totalMaxMemory > js.config.MaxMemory {
		return NewJSMemoryResourcesExceededError()
	}
	if storeReserved+totalMaxStore > js.config.MaxStore {
		return NewJSStorageResourcesExceededError()
	}

	return nil
}

// This will reserve the stream resources requested.
// This will spin off off of MaxBytes.
func (js *jetStream) reserveStreamResources(cfg *StreamConfig) {
	if cfg == nil || cfg.MaxBytes <= 0 {
		return
	}

	js.mu.Lock()
	switch cfg.Storage {
	case MemoryStorage:
		js.memReserved += cfg.MaxBytes
	case FileStorage:
		js.storeReserved += cfg.MaxBytes
	}
	s, clustered := js.srv, !js.standAlone
	js.mu.Unlock()
	// If clustered send an update to the system immediately.
	if clustered {
		s.sendStatszUpdate()
	}
}

// Release reserved resources held by a stream.
func (js *jetStream) releaseStreamResources(cfg *StreamConfig) {
	if cfg == nil || cfg.MaxBytes <= 0 {
		return
	}

	js.mu.Lock()
	switch cfg.Storage {
	case MemoryStorage:
		js.memReserved -= cfg.MaxBytes
	case FileStorage:
		js.storeReserved -= cfg.MaxBytes
	}
	s, clustered := js.srv, !js.standAlone
	js.mu.Unlock()
	// If clustered send an update to the system immediately.
	if clustered {
		s.sendStatszUpdate()
	}
}

const (
	// JetStreamStoreDir is the prefix we use.
	JetStreamStoreDir = "jetstream"
	// JetStreamMaxStoreDefault is the default disk storage limit. 1TB
	JetStreamMaxStoreDefault = 1024 * 1024 * 1024 * 1024
	// JetStreamMaxMemDefault is only used when we can't determine system memory. 256MB
	JetStreamMaxMemDefault = 1024 * 1024 * 256
	// snapshot staging for restores.
	snapStagingDir = ".snap-staging"
)

// Dynamically create a config with a tmp based directory (repeatable) and 75% of system memory.
func (s *Server) dynJetStreamConfig(storeDir string, maxStore, maxMem int64) *JetStreamConfig {
	jsc := &JetStreamConfig{}
	if storeDir != _EMPTY_ {
		jsc.StoreDir = filepath.Join(storeDir, JetStreamStoreDir)
	} else {
		// Create one in tmp directory, but make it consistent for restarts.
		jsc.StoreDir = filepath.Join(os.TempDir(), "nats", JetStreamStoreDir)
		s.Warnf("Temporary storage directory used, data could be lost on system reboot")
	}

	opts := s.getOpts()

	// Strict mode.
	jsc.Strict = !opts.NoJetStreamStrict

	// Sync options.
	jsc.SyncInterval = opts.SyncInterval
	jsc.SyncAlways = opts.SyncAlways

	if opts.maxStoreSet && maxStore >= 0 {
		jsc.MaxStore = maxStore
	} else {
		jsc.MaxStore = diskAvailable(jsc.StoreDir)
	}

	if opts.maxMemSet && maxMem >= 0 {
		jsc.MaxMemory = maxMem
	} else {
		// Estimate to 75% of total memory if we can determine system memory.
		if sysMem := sysmem.Memory(); sysMem > 0 {
			// Check if we have been limited with GOMEMLIMIT and if lower use that value.
			if gml := debug.SetMemoryLimit(-1); gml != math.MaxInt64 && gml < sysMem {
				s.Debugf("JetStream detected GOMEMLIMIT of %v", friendlyBytes(gml))
				sysMem = gml
			}
			jsc.MaxMemory = sysMem / 4 * 3
		} else {
			jsc.MaxMemory = JetStreamMaxMemDefault
		}
	}

	return jsc
}

// Helper function.
func (a *Account) checkForJetStream() (*Server, *jsAccount, error) {
	a.mu.RLock()
	s := a.srv
	jsa := a.js
	a.mu.RUnlock()

	if s == nil || jsa == nil {
		return nil, nil, NewJSNotEnabledForAccountError()
	}

	return s, jsa, nil
}

type Number interface {
	int | int8 | int16 | int32 | int64 | uint | uint8 | uint16 | uint32 | uint64 | float32 | float64
}

// friendlyBytes returns a string with the given bytes int64
// represented as a size, such as 1KB, 10MB, etc...
func friendlyBytes[T Number](bytes T) string {
	fbytes := float64(bytes)
	base := 1024
	pre := []string{"K", "M", "G", "T", "P", "E"}
	if fbytes < float64(base) {
		return fmt.Sprintf("%v B", fbytes)
	}
	exp := int(math.Log(fbytes) / math.Log(float64(base)))
	index := exp - 1
	return fmt.Sprintf("%.2f %sB", fbytes/math.Pow(float64(base), float64(exp)), pre[index])
}

func isValidName(name string) bool {
	if name == _EMPTY_ {
		return false
	}
	return !strings.ContainsAny(name, " \t\r\n\f.*>")
}

func isValidAssetName(name string) bool {
	if name == _EMPTY_ {
		return false
	}
	return !strings.ContainsAny(name, " \t\r\n\f.*>\\/")
}

// To throttle the out of resources errors.
func (s *Server) resourcesExceededError(storeType StorageType) {
	var didAlert bool

	s.rerrMu.Lock()
	if now := time.Now(); now.Sub(s.rerrLast) > 10*time.Second {
		s.Errorf("JetStream %s resource limits exceeded for server", strings.ToLower(storeType.String()))
		s.rerrLast = now
		didAlert = true
	}
	s.rerrMu.Unlock()

	// If we are meta leader we should relinquish that here.
	if didAlert {
		if js := s.getJetStream(); js != nil {
			js.mu.RLock()
			if cc := js.cluster; cc != nil && cc.meta != nil {
				cc.meta.StepDown()
			}
			js.mu.RUnlock()
		}
	}
}

// For validating options.
func validateJetStreamOptions(o *Options) error {
	// in non operator mode, the account names need to be configured
	if len(o.JsAccDefaultDomain) > 0 {
		if len(o.TrustedOperators) == 0 {
			for a, domain := range o.JsAccDefaultDomain {
				found := false
				if isReservedAccount(a) {
					found = true
				} else {
					for _, acc := range o.Accounts {
						if a == acc.GetName() {
							if len(acc.jsLimits) > 0 && domain != _EMPTY_ {
								return fmt.Errorf("default_js_domain contains account name %q with enabled JetStream", a)
							}
							found = true
							break
						}
					}
				}
				if !found {
					return fmt.Errorf("in non operator mode, `default_js_domain` references non existing account %q", a)
				}
			}
		} else {
			for a := range o.JsAccDefaultDomain {
				if !nkeys.IsValidPublicAccountKey(a) {
					return fmt.Errorf("default_js_domain contains account name %q, which is not a valid public account nkey", a)
				}
			}
		}
		for a, d := range o.JsAccDefaultDomain {
			sacc := DEFAULT_SYSTEM_ACCOUNT
			if o.SystemAccount != _EMPTY_ {
				sacc = o.SystemAccount
			}
			if a == sacc {
				return fmt.Errorf("system account %q can not be in default_js_domain", a)
			}
			if d == _EMPTY_ {
				continue
			}
			if sub := fmt.Sprintf(jsDomainAPI, d); !IsValidSubject(sub) {
				return fmt.Errorf("default_js_domain contains account %q with invalid domain name %q", a, d)
			}
		}
	}
	if o.JetStreamDomain != _EMPTY_ {
		if subj := fmt.Sprintf(jsDomainAPI, o.JetStreamDomain); !IsValidSubject(subj) {
			return fmt.Errorf("invalid domain name: derived %q is not a valid subject", subj)
		}

		if !isValidName(o.JetStreamDomain) {
			return fmt.Errorf("invalid domain name: may not contain ., * or >")
		}
	}
	// If not clustered no checks needed past here.
	if !o.JetStream || o.Cluster.Port == 0 {
		return nil
	}
	if o.ServerName == _EMPTY_ {
		return fmt.Errorf("jetstream cluster requires `server_name` to be set")
	}
	if o.Cluster.Name == _EMPTY_ {
		return fmt.Errorf("jetstream cluster requires `cluster.name` to be set")
	}

	h := strings.ToLower(o.JetStreamExtHint)
	switch h {
	case jsWillExtend, jsNoExtend, _EMPTY_:
		o.JetStreamExtHint = h
	default:
		return fmt.Errorf("expected 'no_extend' for string value, got '%s'", h)
	}

	if o.JetStreamMaxCatchup < 0 {
		return fmt.Errorf("jetstream max catchup cannot be negative")
	}
	return nil
}

// We had a bug that set a default de dupe window on mirror, despite that being not a valid config
func fixCfgMirrorWithDedupWindow(cfg *StreamConfig) {
	if cfg == nil || cfg.Mirror == nil {
		return
	}
	if cfg.Duplicates != 0 {
		cfg.Duplicates = 0
	}
}

func (s *Server) handleWritePermissionError() {
	//TODO Check if we should add s.jetStreamOOSPending in condition
	if s.JetStreamEnabled() {
		s.Errorf("File system permission denied while writing, disabling JetStream")

		go s.DisableJetStream()

		//TODO Send respective advisory if needed, same as in handleOutOfSpace
	}
}
