common/message/stream.go

// 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 message

import (
	"fmt"
	"time"
)

// 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"`
	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"`
}

// Used to guide placement of streams and meta controllers in clustered JetStream.
type Placement struct {
	Cluster   string   `json:"cluster,omitempty"`
	Tags      []string `json:"tags,omitempty"`
	Preferred string   `json:"preferred,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.
}

// 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"`
}

// 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"`
}

// 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"`
}

type StreamConsumerLimits struct {
	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
	MaxAckPending     int           `json:"max_ack_pending,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
)

// MarshalJSON 将 PersistModeType 序列化为字符串
func (pm PersistModeType) MarshalJSON() ([]byte, error) {
	switch pm {
	case DefaultPersistMode:
		return []byte(`"default"`), nil
	case AsyncPersistMode:
		return []byte(`"async"`), nil
	default:
		return nil, fmt.Errorf("can not marshal %v", pm)
	}
}

// UnmarshalJSON 将字符串反序列化为 PersistModeType
func (pm *PersistModeType) UnmarshalJSON(data []byte) error {
	switch string(data) {
	case `"default"`:
		*pm = DefaultPersistMode
	case `"async"`:
		*pm = AsyncPersistMode
	default:
		return fmt.Errorf("unknown persist mode: %s", string(data))
	}
	return nil
}

type StoreCompression uint8

const (
	NoCompression StoreCompression = iota
	S2Compression
)

// MarshalJSON 将 StoreCompression 序列化为字符串
func (sc StoreCompression) MarshalJSON() ([]byte, error) {
	switch sc {
	case NoCompression:
		return []byte(`"none"`), nil
	case S2Compression:
		return []byte(`"s2"`), nil
	default:
		return nil, fmt.Errorf("can not marshal %v", sc)
	}
}

// UnmarshalJSON 将字符串反序列化为 StoreCompression
func (sc *StoreCompression) UnmarshalJSON(data []byte) error {
	switch string(data) {
	case `"none"`:
		*sc = NoCompression
	case `"s2"`:
		*sc = S2Compression
	default:
		return fmt.Errorf("unknown store compression: %s", string(data))
	}
	return nil
}