stream

type AggregateFlush struct {
	Attributes   map[string]string
	Body         string
	MessageCount int
}

type BaseOutputConfiguration struct {
	Tracing BaseOutputConfigurationTracing `cfg:"tracing"`
}

type BaseOutputConfigurationAware interface {
	SetTracing(enabled bool)
}

type BaseOutputConfigurationTracing struct {
	Enabled bool `cfg:"enabled" default:"true"`
}

type Chunk [][]byte

type Chunks []Chunk

type CompressionType string

type Consumer struct {
	kernel.EssentialModule
	kernel.ApplicationStage
	// contains filtered or unexported fields
}

type ConsumerCallback[M any] interface {
	Consume(ctx context.Context, model M, attributes map[string]string) (bool, error)
}

type ConsumerCallbackFactory[M any] func(ctx context.Context, config cfg.Config, logger log.Logger) (ConsumerCallback[M], error)

type ConsumerCallbackMap[M any] map[string]ConsumerCallbackFactory[M]

type ConsumerMetadata struct {
	Name         string `json:"name"`
	RetryEnabled bool   `json:"retry_enabled"`
	RetryType    string `json:"retry_type"`
}

type ConsumerRetrySettings struct {
	Enabled   bool          `cfg:"enabled"`
	Type      string        `cfg:"type" default:"sqs"`
	GraceTime time.Duration `cfg:"grace_time" default:"10s"`
}

type ConsumerSettings struct {
	Input       string        `cfg:"input" default:"consumer" validate:"required"`
	Encoding    EncodingType  `cfg:"encoding" default:"application/json"`
	IdleTimeout time.Duration `cfg:"idle_timeout" default:"10s"`
	// GraceTime is the maximum time a record has to be processed once the consumer stops. It is the single
	// authoritative processing deadline and applies to every input, both the primary and the retry one: inputs must
	// stop fetching new messages, while messages they already fetched are processed until this shared deadline. When
	// it expires, in-flight callback contexts are canceled so the inputs can nack or retry any remaining messages.
	//
	// Inputs do not define a processing deadline of their own. Their own grace_time bounds how long they get to
	// acknowledge or commit what was processed, which is a window that only starts once this deadline expired.
	GraceTime             time.Duration                 `cfg:"grace_time" default:"10s"`
	Retry                 ConsumerRetrySettings         `cfg:"retry"`
	Healthcheck           health.HealthCheckSettings    `cfg:"healthcheck"`
	AggregateMessageMode  string                        `cfg:"aggregate_message_mode" default:"atMostOnce" validate:"oneof=atLeastOnce atMostOnce"`
	IgnoreOnGetModelError IgnoreOnGetModelErrorSettings `cfg:"ignore_on_get_model_error"`
}

type EncodeHandler interface {
	Encode(ctx context.Context, data any, attributes map[string]string) (context.Context, map[string]string, error)
	Decode(ctx context.Context, data any, attributes map[string]string) (context.Context, map[string]string, error)
}

type EncodingType string

type FileOutputMode string

type FileOutputSettings struct {
	Filename string         `cfg:"filename"`
	Mode     FileOutputMode `cfg:"mode"     default:"append"`
}

type FileSettings struct {
	Filename    string `cfg:"filename"`
	RunnerCount int    `cfg:"runner_count" default:"1" validate:"min=1"`
}

type IgnorableGetModelError interface {
	error
	// IsIgnorableWithSettings returns true if this error should result in the message being ignored
	// based on the given settings.
	IsIgnorableWithSettings(settings IgnoreOnGetModelErrorSettings) bool
}

type IgnoreOnGetModelErrorSettings struct {
	// UnknownModel indicates whether to ignore messages when the model is unknown.
	// When true, messages with unknown model IDs will be acknowledged and skipped.
	UnknownModel bool `cfg:"unknown_model" default:"false"`
	// UnknownVersion indicates whether to ignore messages when the version is unknown.
	// When true, messages with unknown versions for known models will be acknowledged and skipped.
	UnknownVersion bool `cfg:"unknown_version" default:"false"`
}

type InMemoryInput struct {
	// contains filtered or unexported fields
}

type InMemoryOutput struct {
	// contains filtered or unexported fields
}

type InMemoryOutputConfiguration struct {
	BaseOutputConfiguration
	Type string `cfg:"type" default:"inMemory"`
}

type InMemorySettings struct {
	Size        int `cfg:"size" default:"1"`
	RunnerCount int `cfg:"runner_count" default:"1" validate:"min=1"`
}

type InitializeableCallback interface {
	Init(ctx context.Context) error
}

type Input interface {
	// Run provides a steady stream of messages, returned via Data. Run does not return until Stop is called and thus
	// should be called in its own go routine. The only exception to this is if we either fail to produce messages and
	// return an error or if the input is depleted (like an InMemoryInput).
	//
	// Run should only be called once, not all inputs can be resumed.
	Run(ctx context.Context, process InputProcess) error
	// Stop prevents subsequent fetches and causes Run to return as fast as possible. Messages already fetched when Stop
	// is called may still be passed to process, allowing the consumer to finish them during its shutdown grace period.
	// Calling Stop is preferable to canceling the context passed to Run as it allows Run to shut down cleaner (and might
	// take a bit longer, e.g., to finish processing the current batch of messages).
	Stop(ctx context.Context)
	// IsHealthy checks if the input is still able to produce data. An Input is healthy if it produces zero or more
	// messages repeatedly. Producing zero messages would for example happen if the input requested data from an
	// external queue, but the queue was empty. An Input is unhealthy if it is no longer able to produce any messages.
	//
	// If an input exhausts its source (file, finished stream, fixed list, ...), it is still considered as healthy.
	IsHealthy() bool
}

type InputFactory func(ctx context.Context, config cfg.Config, logger log.Logger, name string) (Input, error)

type InputProcess func(ctx context.Context, msg *Message) (ack bool)

type KafkaInputConfiguration struct {
	kafkaConsumer.Settings
	Type string `cfg:"type" default:"kafka"`
}

type KafkaOutputConfiguration struct {
	BaseOutputConfiguration
	cfg.ResourceIdentifier
	Type       string `cfg:"type" default:"kafka"`
	TopicId    string `cfg:"topic_id"`
	Connection string `cfg:"connection" default:"default"`

	// LingerTimeout is the max time the producer will wait for new records before flushing the current batch.
	// When set to 0s, batches will be sent out as fast as possible (or when the size limits are reached with enough back pressure).
	// The kafka library recommends to increase this only when batching with low volume.
	LingerTimeout  time.Duration `cfg:"linger_timeout" default:"0s"`
	RequestTimeout time.Duration `cfg:"request_timeout" default:"10s"`

	MaxBatchSize  int   `cfg:"max_batch_size" default:"10000"`
	MaxBatchBytes int32 `cfg:"max_batch_bytes" default:"1000012"`
}

type KinesisInputConfiguration struct {
	kinesis.Settings
	Type string `cfg:"type" default:"kinesis"`
}

type KinesisOutputConfiguration struct {
	BaseOutputConfiguration
	cfg.ResourceIdentifier
	Type       string `cfg:"type" default:"kinesis"`
	ClientName string `cfg:"client_name" default:"default"`
	StreamName string `cfg:"stream_name"`
}

type KinesisOutputSettings struct {
	cfg.ResourceIdentifier
	ClientName string
	StreamName string
}

type KinsumerAutoscaleModule struct {
	kernel.BackgroundModule
	kernel.ApplicationStage
	// contains filtered or unexported fields
}

type KinsumerAutoscaleModuleDynamoDbNamingSettings struct {
	Pattern string `cfg:"pattern,nodecode" default:"{app.env}-kinsumer-autoscale-leaders"`
}

type KinsumerAutoscaleModuleDynamoDbSettings struct {
	Naming KinsumerAutoscaleModuleDynamoDbNamingSettings `cfg:"naming"`
}

type KinsumerAutoscaleModuleEcsSettings struct {
	Client  string `cfg:"client" default:"default"`
	Cluster string `cfg:"cluster" default:"{app.env}"`
	Service string `cfg:"service" default:"{app.name}"`
}

type KinsumerAutoscaleModuleSettings struct {
	Ecs            KinsumerAutoscaleModuleEcsSettings      `cfg:"ecs"`
	Enabled        bool                                    `cfg:"enabled" default:"false"`
	DynamoDb       KinsumerAutoscaleModuleDynamoDbSettings `cfg:"dynamodb"`
	LeaderElection string                                  `cfg:"leader_election" default:"kinsumer-autoscale"`
	Orchestrator   string                                  `cfg:"orchestrator" default:"ecs"`
	Period         time.Duration                           `cfg:"period" default:"1m"`
}

type KinsumerAutoscaleOrchestrator interface {
	GetCurrentTaskCount(ctx context.Context) (int32, error)
	UpdateTaskCount(ctx context.Context, taskCount int32) error
}

type KinsumerAutoscaleOrchestratorFactory func(ctx context.Context, config cfg.Config, logger log.Logger) (KinsumerAutoscaleOrchestrator, error)

type LifecycleManager interface {
	reslife.LifeCycleer
	reslife.Creator
}

type Message struct {
	Attributes map[string]string `json:"attributes"`
	Body       string            `json:"body"`
}

type MessageBodyCompressor interface {
	Compress(body []byte) ([]byte, error)
	Decompress(body []byte) ([]byte, error)
}

type MessageBodyEncoder interface {
	Encode(data any) ([]byte, error)
	Decode(data []byte, out any) error
}

type MessageEncoder interface {
	Encode(ctx context.Context, data any, attributeSets ...map[string]string) (*Message, error)
	Decode(ctx context.Context, msg *Message, out any) (context.Context, map[string]string, error)
}

type MessageEncoderSettings struct {
	Encoding        EncodingType
	Compression     CompressionType
	EncodeHandlers  []EncodeHandler
	ExternalEncoder MessageBodyEncoder
}

type ModelMsg struct {
	CrudType string
	Version  int
	ModelId  string
	Body     string
}

type NoOpOutput struct{}

type Output interface {
	WriteOne(ctx context.Context, msg WritableMessage) error
	Write(ctx context.Context, batch []WritableMessage) error
}

type OutputCapabilities struct {
	// IsPartitionedOutput should be true if the output is writing to more than one shard/partition/bucket, and we need to
	// take care about writing messages to the correct partition.
	IsPartitionedOutput bool
	// ProvidesCompression should be true if the Output natively handles compression.
	ProvidesCompression bool
	// SupportsAggregation should be false if the Output can not handle aggregated messages, e.g. when using a schema registry.
	SupportsAggregation bool
	// MaxBatchSize is the maximum number of messages we can write at once to the output (or nil if there is no limit).
	MaxBatchSize *int
	// MaxMessageSize is the maximum size of a message for this output (or nil if there is no limit on message size).
	MaxMessageSize *int
	// IgnoreProducerDaemonBatchSettings should be true if only the size restrictions specified on the output capabilities should be used.
	// Otherwise, they are only used if lower than the restrictions specified on the producer daemon batch settings.
	IgnoreProducerDaemonBatchSettings bool
}

type OutputChannel interface {
	Read() ([]WritableMessage, bool)
	Write(ctx context.Context, msg []WritableMessage)
	Close(ctx context.Context)
	IsClosed() bool
}

type OutputFactory func(ctx context.Context, config cfg.Config, logger log.Logger, name string) (Output, *OutputCapabilities, error)

type PartitionerRand interface {
	Intn(n int) int
}

type Producer interface {
	WriteOne(ctx context.Context, model any, attributeSets ...map[string]string) error
	Write(ctx context.Context, models any, attributeSets ...map[string]string) error
}

type ProducerDaemonAggregator interface {
	Write(ctx context.Context, msg *Message) ([]AggregateFlush, error)
	Flush() ([]AggregateFlush, error)
}

type ProducerDaemonBatcher interface {
	Append(msg *Message) ([]WritableMessage, error)
	Flush() []WritableMessage
}

type ProducerDaemonSettings struct {
	Enabled bool `cfg:"enabled" default:"false"`

	// Amount of time spend waiting for messages before sending out a batch.
	Interval time.Duration `cfg:"interval" default:"1m"`

	// Size of the buffer channel, i.e., how many messages can be in-flight at once? Generally it is a good idea to match
	// this with the number of runners.
	BufferSize int `cfg:"buffer_size" default:"10" validate:"min=1"`

	// Number of daemons running in the background, writing complete batches to the output.
	RunnerCount int `cfg:"runner_count" default:"10" validate:"min=1"`

	// How many SQS messages do we submit in a single batch? SQS can accept up to 10 messages per batch.
	// SNS doesn't support batching, so the value doesn't matter for SNS.
	BatchSize int `cfg:"batch_size" default:"10" validate:"min=1"`

	// How large may the sum of all messages in the aggregation be? For SQS you can't send more than 256 KB in one batch,
	// for SNS a single message can't be larger than 256 KB. We use 252 KB as default to leave some room for request
	// encoding and overhead.
	BatchMaxSize int `cfg:"batch_max_size" default:"258048" validate:"min=0"`

	// How many stream.Messages do we pack together in a single batch (one message in SQS) at once?
	AggregationSize int `cfg:"aggregation_size" default:"1" validate:"min=1"`

	// Maximum size in bytes of a batch. Defaults to 64 KB to leave some room for encoding overhead.
	// Set to 0 to disable limiting the maximum size for a batch (it will still not put more than BatchSize messages
	// in a batch).
	//
	// Note: Gosoline can't ensure your messages stay below this size if your messages are quite large (especially when
	// using compression). Imagine you already aggregated 40kb of compressed messages (around 53kb when base64 encoded)
	// and are now writing a message that compresses to 20 kb. Now your buffer reaches 60 kb and 80 kb base64 encoded.
	// Gosoline will not already output a 53 kb message if you requested 64 kb messages (it would accept a 56 kb message),
	// but after writing the next message
	AggregationMaxSize int `cfg:"aggregation_max_size" default:"65536" validate:"min=0"`

	// If you are writing to an output using a partition key, we ensure messages are still distributed to a partition
	// according to their partition key (although not necessary the same partition as without the producer daemon).
	// For this, we split the messages into buckets while collecting them, thus potentially aggregating more messages in
	// memory (depending on the number of buckets you configure).
	//
	// Note: This still does not guarantee that your messages are perfectly ordered - this is impossible as soon as you
	// have more than once producer. However, messages with the same partition key will end up in the same shard, so if
	// you are reading two different shards and one is much further behind than the other, you will not see messages
	// *massively* out of order - it should be roughly bounded by the time you buffer messages (the Interval setting) and
	// thus be not much more than a minute (using the default setting) instead of hours (if one shard is half a day behind
	// while the other is up-to-date).
	//
	// Second note: If you change the amount of partitions, messages might move between buckets and thus end up in different
	// shards than before. Thus, only do this if you can handle it (e.g., because no shard is currently lagging behind).
	PartitionBucketCount int `cfg:"partition_bucket_count" default:"128" validate:"min=1"`

	// Additional attributes we append to each message
	MessageAttributes map[string]string `cfg:"message_attributes"`
}

type ProducerMetadata struct {
	Name          string `json:"name"`
	DaemonEnabled bool   `json:"daemon_enabled"`
}

type ProducerOption func(p *producerOptions)

type ProducerSettings struct {
	Output      string                 `cfg:"output"`
	Encoding    EncodingType           `cfg:"encoding"`
	Compression CompressionType        `cfg:"compression" default:"none"`
	Daemon      ProducerDaemonSettings `cfg:"daemon"`
}

type ProtobufEncodable interface {
	ToMessage() (proto.Message, error)
	EmptyMessage() proto.Message
	FromMessage(message proto.Message) error
}

type RawMessage struct {
	Body    any
	Encoder MessageBodyEncoder
}

type RedisListInputSettings struct {
	ServerName         string
	Key                string
	WaitTime           time.Duration
	HealthcheckTimeout time.Duration
	RunnerCount        int
}

type RedisListOutputSettings struct {
	ServerName string
	Key        string
	BatchSize  int
}

type RetryHandler interface {
	Put(ctx context.Context, msg *Message) error
}

type RetryHandlerFactory func(ctx context.Context, config cfg.Config, logger log.Logger, name string) (Input, RetryHandler, error)

type RetryHandlerNoop struct{}

type RetryHandlerSettings struct {
	After       time.Duration `cfg:"after" default:"1m"`
	MaxAttempts int           `cfg:"max_attempts" default:"3"`
}

type RetryHandlerSqs struct {
	// contains filtered or unexported fields
}

type RetryHandlerSqsSettings struct {
	cfg.ResourceIdentifier
	RetryHandlerSettings
	ClientName          string                     `cfg:"client_name" default:"default"`
	MaxNumberOfMessages int32                      `cfg:"max_number_of_messages" default:"10" validate:"min=1,max=10"`
	WaitTime            int32                      `cfg:"wait_time" default:"10"`
	AckGraceTime        time.Duration              `cfg:"ack_grace_time" default:"10s"`
	RunnerCount         int                        `cfg:"runner_count" default:"1"`
	AcknowledgementMode SqsAcknowledgementMode     `cfg:"acknowledgement_mode" default:"individual" validate:"oneof=individual batch"`
	QueueId             string                     `cfg:"queue_id"`
	Healthcheck         health.HealthCheckSettings `cfg:"healthcheck"`
}

type RetryingInput interface {
	GetRetryHandler() (Input, RetryHandler)
}

type RunnableCallback interface {
	Run(ctx context.Context) error
}

type RunnableConsumerCallback[M any] interface {
	ConsumerCallback[M]
	RunnableCallback
}

type RunnableUntypedConsumerCallback interface {
	UntypedConsumerCallback
	RunnableCallback
}

type SchemaRegistryAwareInput interface {
	Input
	// InitSchemaRegistry initializes the schema registry and returns the encoder/decoder corresponding to the schema
	InitSchemaRegistry(ctx context.Context, settings SchemaSettingsWithEncoding) (MessageBodyEncoder, error)
}

type SchemaRegistryAwareOutput interface {
	Output
	// InitSchemaRegistry initializes the schema registry and returns the encoder/decoder corresponding to the schema
	InitSchemaRegistry(ctx context.Context, settings SchemaSettingsWithEncoding) (MessageBodyEncoder, error)
}

type SchemaSettings struct {
	Subject              string
	Schema               string
	AutoRegister         bool
	ProtobufMessageIndex []int
	Model                any
}

type SchemaSettingsAwareCallback interface {
	GetSchemaSettings() (*SchemaSettings, error)
}

type SchemaSettingsWithEncoding struct {
	Subject              string
	Schema               string
	Encoding             EncodingType
	AutoRegister         bool
	ProtobufMessageIndex []int
	Model                any
}

type SnsInputConfiguration struct {
	cfg.ResourceIdentifier
	Type                string                        `cfg:"type" default:"sns"`
	ConsumerId          string                        `cfg:"id" validate:"required"`
	Targets             []SnsInputTargetConfiguration `cfg:"targets" validate:"min=1"`
	MaxNumberOfMessages int32                         `cfg:"max_number_of_messages" default:"10" validate:"min=1,max=10"`
	WaitTime            int32                         `cfg:"wait_time" default:"3" validate:"min=1"`
	VisibilityTimeout   int                           `cfg:"visibility_timeout" default:"30" validate:"min=1"`
	GraceTime           time.Duration                 `cfg:"grace_time" default:"10s"`
	RunnerCount         int                           `cfg:"runner_count" default:"1" validate:"min=1"`
	AcknowledgementMode SqsAcknowledgementMode        `cfg:"acknowledgement_mode" default:"individual" validate:"oneof=individual batch"`
	RedrivePolicy       sqs.RedrivePolicy             `cfg:"redrive_policy"`
	ClientName          string                        `cfg:"client_name" default:"default"`
	Healthcheck         health.HealthCheckSettings    `cfg:"healthcheck"`
}

type SnsInputSettings struct {
	Identity            cfg.Identity               `cfg:"identity"`
	QueueId             string                     `cfg:"queue_id"`
	MaxNumberOfMessages int32                      `cfg:"max_number_of_messages" default:"10" validate:"min=1,max=10"`
	WaitTime            int32                      `cfg:"wait_time"`
	RedrivePolicy       sqs.RedrivePolicy          `cfg:"redrive_policy"`
	VisibilityTimeout   int                        `cfg:"visibility_timeout"`
	GraceTime           time.Duration              `cfg:"grace_time" default:"10s"`
	RunnerCount         int                        `cfg:"runner_count"`
	AcknowledgementMode SqsAcknowledgementMode     `cfg:"acknowledgement_mode" default:"individual" validate:"oneof=individual batch"`
	ClientName          string                     `cfg:"client_name"`
	Healthcheck         health.HealthCheckSettings `cfg:"healthcheck"`
}

type SnsInputTarget struct {
	Identity     cfg.Identity
	TopicId      string
	FilterPolicy map[string][]string
	ClientName   string
}

type SnsInputTargetConfiguration struct {
	cfg.ResourceIdentifier
	TopicId      string              `cfg:"topic_id" validate:"required"`
	FilterPolicy map[string][]string `cfg:"filter_policy"`
	ClientName   string              `cfg:"client_name" default:"default"`
}

type SnsOutputConfiguration struct {
	BaseOutputConfiguration
	cfg.ResourceIdentifier
	Type       string `cfg:"type" default:"sns"`
	TopicId    string `cfg:"topic_id" validate:"required"`
	ClientName string `cfg:"client_name" default:"default"`
}

type SnsOutputSettings struct {
	Identity   cfg.Identity
	TopicId    string
	ClientName string
}

type SqsAcknowledgementMode string

type SqsInputSettings struct {
	Identity            cfg.Identity `cfg:"identity"`
	QueueId             string       `cfg:"queue_id"`
	MaxNumberOfMessages int32        `cfg:"max_number_of_messages" default:"10" validate:"min=1,max=10"`
	WaitTime            int32        `cfg:"wait_time"`
	VisibilityTimeout   int          `cfg:"visibility_timeout"`
	// GraceTime bounds acknowledging already processed messages after the input context was canceled. It does not
	// bound processing itself, which is owned by stream.consumer.<name>.grace_time.
	GraceTime           time.Duration              `cfg:"grace_time" default:"10s"`
	RunnerCount         int                        `cfg:"runner_count"`
	AcknowledgementMode SqsAcknowledgementMode     `cfg:"acknowledgement_mode" default:"individual" validate:"oneof=individual batch"`
	Fifo                sqs.FifoSettings           `cfg:"fifo"`
	RedrivePolicy       sqs.RedrivePolicy          `cfg:"redrive_policy"`
	ClientName          string                     `cfg:"client_name"`
	Unmarshaller        string                     `cfg:"unmarshaller" default:"msg"`
	Healthcheck         health.HealthCheckSettings `cfg:"healthcheck"`
}

type SqsOutputConfiguration struct {
	BaseOutputConfiguration
	cfg.ResourceIdentifier
	Type              string            `cfg:"type" default:"sqs"`
	QueueId           string            `cfg:"queue_id" validate:"required"`
	VisibilityTimeout int               `cfg:"visibility_timeout" default:"30" validate:"gt=0"`
	RedrivePolicy     sqs.RedrivePolicy `cfg:"redrive_policy"`
	Fifo              sqs.FifoSettings  `cfg:"fifo"`
	ClientName        string            `cfg:"client_name" default:"default"`
}

type SqsOutputSettings struct {
	Identity          cfg.Identity
	ClientName        string
	Fifo              sqs.FifoSettings
	QueueId           string
	RedrivePolicy     sqs.RedrivePolicy
	VisibilityTimeout int
}

type UnmarshallerFunc func(data *string) (*Message, error)

type UntypedConsumerCallback interface {
	GetModel(attributes map[string]string) (any, error)
	Consume(ctx context.Context, model any, attributes map[string]string) (bool, error)
}

type UntypedConsumerCallbackFactory func(ctx context.Context, config cfg.Config, logger log.Logger) (UntypedConsumerCallback, error)

type UntypedConsumerCallbackMap map[string]UntypedConsumerCallbackFactory

type WritableMessage interface {
	MarshalToBytes() ([]byte, error)
	MarshalToString() (string, error)
}