sink

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

type ClickHouseStorageConfig struct {
	ClickHouse clickhouse.Conn
	Database   string
}

type InsertEventsQuery struct {
	Database string
	Messages []SinkMessage
}

type InsertInvalidQuery struct {
	Database string
	Messages []SinkMessage
}

type MeterStore struct {
	Meters []models.Meter
}

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

type Offset struct {
	Offset int64
}

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

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

type ProcessingControl int32

type ProcessingError struct {
	Message           string
	ProcessingControl ProcessingControl
}

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

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

type SinkConfig struct {
	Logger          *slog.Logger
	Tracer          trace.Tracer
	MetricMeter     metric.Meter
	MeterRepository meter.Repository
	Storage         Storage
	Deduplicator    dedupe.Deduplicator
	Consumer        *kafka.Consumer
	// MinCommitCount is the minimum number of messages to wait before flushing the buffer.
	// Whichever happens earlier MinCommitCount or MaxCommitWait will trigger a flush.
	MinCommitCount int
	// MaxCommitWait is the maximum time to wait before flushing the buffer
	MaxCommitWait time.Duration
	// NamespaceRefetch is the interval to refetch exsisting namespaces and meters
	// this information is used to configure which topics the consumer subscribes and
	// the meter configs used in event validation.
	NamespaceRefetch time.Duration
	// OnFlushSuccess is an optional lifecycle hook
	OnFlushSuccess func(string, int64)
}

type SinkMessage struct {
	Namespace    string
	KafkaMessage *kafka.Message
	Serialized   *serializer.CloudEventsKafkaPayload
	Error        *ProcessingError
}

type Storage interface {
	BatchInsert(ctx context.Context, messages []SinkMessage) error
	BatchInsertInvalid(ctx context.Context, messages []SinkMessage) error
}