kafka

type Config[T any] struct {
	// Brokers is the seed broker list, e.g. {"localhost:9092"} or MSK bootstrap servers.
	Brokers []string

	// Topic is the Kafka topic to consume from.
	Topic string

	// ConsumerGroup is the Kafka consumer-group ID. All workers sharing this ID
	// cooperatively partition the topic between them.
	ConsumerGroup string

	// Decode converts a raw message value to T. Use JSONDecoder[T]() for JSON-encoded
	// records, or supply your own for Avro / Protobuf / etc.
	Decode Decoder[T]

	// OnDecodeError, if non-nil, is called for every message whose Decode returned
	// an error. The default behavior is to drop the record silently and advance —
	// fine for development but dangerous in production. Wire this to a DLQ producer
	// or a metrics.Recorder.RecordError to surface poison pills.
	OnDecodeError func(raw []byte, partition int32, offset int64, err error)

	// EventID, if non-nil, is called on every decoded record to compute the
	// at-least-once dedup key. The default is "<topic>:<partition>:<offset>"
	// — globally unique within the cluster's history but only deduplicates
	// Kafka-side redeliveries (rebalance after a crash before commit). For
	// CDC pipelines where the same logical change can be produced twice
	// (Debezium retransmits, dual-write fixers), set this to extract the
	// upstream identifier (e.g. Mongo `_id`, Postgres LSN) from the record.
	EventID func(T) string

	// OnFetchError, if non-nil, is called for partition-level fetch errors that the
	// client is going to retry internally (broker bounce, leader change, etc).
	// Default: drop. Wire to logging / metrics to surface persistent failures.
	OnFetchError func(topic string, partition int32, err error)

	// Extra lets callers append additional franz-go options (TLS, SASL, etc).
	Extra []kgo.Opt
}

type Decoder[T any] func([]byte) (T, error)

type Source[T any] struct {
	// contains filtered or unexported fields
}