stream

type AcknowledgeableInput interface {
	Input
	// Ack acknowledges a single message. If possible, prefer calling AckBatch as it is more efficient.
	Ack(ctx context.Context, msg *Message, ack bool) error
	// AckBatch does the same as calling Ack for every single message would, but it might use fewer calls to an external
	// service.
	AckBatch(ctx context.Context, msgs []*Message, acks []bool) error
}

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

type BaseConsumerCallback interface {
	GetModel(attributes map[string]string) any
}

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

type BaseOutputConfigurationAware interface {
	SetTracing(enabled bool)
}

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

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

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

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

type BatchConsumerCallbackMap[M any] map[string]BatchConsumerCallbackFactory[M]

type BatchConsumerSettings struct {
	IdleTimeout      time.Duration `cfg:"idle_timeout" default:"10s"`
	BatchSize        int           `cfg:"batch_size" default:"1"`
	ConsumeGraceTime time.Duration `cfg:"consume_grace_time" default:"10s"`
}

type Chunk [][]byte

type Chunks []Chunk

type CompressionType string

type Consumer struct {
	// 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"`
	RunnerCount  int    `json:"runner_count"`
}

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"`
	RunnerCount          int                        `cfg:"runner_count" default:"1" validate:"min=1"`
	Encoding             EncodingType               `cfg:"encoding" default:"application/json"`
	IdleTimeout          time.Duration              `cfg:"idle_timeout" default:"10s"`
	AcknowledgeGraceTime time.Duration              `cfg:"acknowledge_grace_time" default:"10s"`
	ConsumeGraceTime     time.Duration              `cfg:"consume_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"`
}

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"`
	Blocking bool   `cfg:"blocking"`
}

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

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) error
	// Stop causes Run to return as fast as possible. 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)
	// Data returns a channel containing the messages produced by this input.
	Data() <-chan *Message
	// 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 KafkaInputConfiguration struct {
	kafkaConsumer.Settings
	Type string `cfg:"type" default:"kafka"`
}

type KafkaOutputConfiguration struct {
	BaseOutputConfiguration
	Type        string `cfg:"type" default:"kafka"`
	Project     string `cfg:"project"`
	Family      string `cfg:"family"`
	Group       string `cfg:"group"`
	Application string `cfg:"application"`
	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 KafkaSourceMessage struct {
	kgo.Record
}

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

type KinesisOutputConfiguration struct {
	BaseOutputConfiguration
	Type        string `cfg:"type" default:"kinesis"`
	Project     string `cfg:"project"`
	Family      string `cfg:"family"`
	Group       string `cfg:"group"`
	Application string `cfg:"application"`
	ClientName  string `cfg:"client_name" default:"default"`
	StreamName  string `cfg:"stream_name"`
}

type KinesisOutputSettings struct {
	cfg.AppId
	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:"{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:"{env}"`
	Service string `cfg:"service" default:"{app_group}-{app_name}"`
}

type KinsumerAutoscaleModuleSettings struct {
	Ecs            KinsumerAutoscaleModuleEcsSettings      `cfg:"ecs"`
	Enabled        bool                                    `cfg:"enabled" default:"true"`
	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"`
	// contains filtered or unexported fields
}

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 {
	cfg.AppId
	ServerName         string
	Key                string
	WaitTime           time.Duration
	HealthcheckTimeout time.Duration
}

type RedisListOutputSettings struct {
	cfg.AppId
	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.AppId
	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"`
	RunnerCount         int                        `cfg:"runner_count" default:"1"`
	QueueId             string                     `cfg:"queue_id"`
	Healthcheck         health.HealthCheckSettings `cfg:"healthcheck"`
}

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

type RunnableBatchConsumerCallback[M any] interface {
	BatchConsumerCallback[M]
	RunnableCallback
}

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

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

type RunnableUntypedBatchConsumerCallback interface {
	UntypedBatchConsumerCallback
	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
	ProtobufMessageIndex []int
	Model                any
}

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

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

type SnsInputConfiguration struct {
	Type                string                        `cfg:"type" default:"sns"`
	ConsumerId          string                        `cfg:"id" validate:"required"`
	Family              string                        `cfg:"family" default:""`
	Group               string                        `cfg:"group" default:""`
	Application         string                        `cfg:"application" default:""`
	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"`
	RunnerCount         int                           `cfg:"runner_count" default:"1" validate:"min=1"`
	RedrivePolicy       sqs.RedrivePolicy             `cfg:"redrive_policy"`
	ClientName          string                        `cfg:"client_name" default:"default"`
	Healthcheck         health.HealthCheckSettings    `cfg:"healthcheck"`
}

type SnsInputSettings struct {
	cfg.AppId
	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"`
	RunnerCount         int                        `cfg:"runner_count"`
	ClientName          string                     `cfg:"client_name"`
	Healthcheck         health.HealthCheckSettings `cfg:"healthcheck"`
}

type SnsInputTarget struct {
	cfg.AppId
	TopicId    string
	Attributes map[string]string
	ClientName string
}

type SnsInputTargetConfiguration struct {
	Family      string            `cfg:"family"`
	Group       string            `cfg:"group" validate:"required"`
	Application string            `cfg:"application" validate:"required"`
	TopicId     string            `cfg:"topic_id" validate:"required"`
	Attributes  map[string]string `cfg:"attributes"`
	ClientName  string            `cfg:"client_name" default:"default"`
}

type SnsOutputConfiguration struct {
	BaseOutputConfiguration
	Type        string `cfg:"type" default:"sns"`
	Project     string `cfg:"project"`
	Family      string `cfg:"family"`
	Group       string `cfg:"group"`
	Application string `cfg:"application"`
	TopicId     string `cfg:"topic_id" validate:"required"`
	ClientName  string `cfg:"client_name" default:"default"`
}

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

type SqsInputSettings struct {
	cfg.AppId
	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"`
	RunnerCount         int                        `cfg:"runner_count"`
	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
	Type              string            `cfg:"type" default:"sqs"`
	Project           string            `cfg:"project"`
	Family            string            `cfg:"family"`
	Group             string            `cfg:"group"`
	Application       string            `cfg:"application"`
	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 {
	cfg.AppId
	ClientName        string
	Fifo              sqs.FifoSettings
	QueueId           string
	RedrivePolicy     sqs.RedrivePolicy
	VisibilityTimeout int
}

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

type UntypedBatchConsumerCallback interface {
	BaseConsumerCallback
	Consume(ctx context.Context, models []any, attributes []map[string]string) ([]bool, error)
}

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

type UntypedBatchConsumerCallbackMap map[string]UntypedBatchConsumerCallbackFactory

type UntypedConsumerCallback interface {
	BaseConsumerCallback
	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)
}