messaging

type BatchEventHandler func(ctx context.Context, envelopes []*EventEnvelope) error

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

type ConsumerConfig struct {
	BatchSize       int           // Maximum batch size (default: 100)
	BatchTimeout    time.Duration // Maximum time to wait for batch (default: 1s)
	MaxRetries      int           // Maximum retry attempts (default: 3)
	RetryBackoffMin time.Duration // Minimum retry backoff (default: 1s)
	RetryBackoffMax time.Duration // Maximum retry backoff (default: 30s)
	Workers         int           // Number of concurrent workers (default: 4)
}

type ConsumerMetrics struct {
	EventsProcessed   int64
	EventsFailed      int64
	EventsDLQ         int64
	DBWriteErrors     int64 // Track DB write failures
	UnmarshalErrors   int64 // Track unmarshal failures separately
	BatchesProcessed  int64
	BatchesFailed     int64 // Track batch failures
	TotalLatencyMs    int64
	LastProcessedTime time.Time
}

type DLQWriter interface {
	WriteBatch(ctx context.Context, envelopes []*EventEnvelope, err error) error
}

type EventEnvelope struct {
	SchemaVersion string          `json:"schema_version"`
	EventType     string          `json:"event_type"`
	TenantID      string          `json:"tenant_id"`
	EventID       string          `json:"event_id"`
	Timestamp     time.Time       `json:"timestamp"`
	Payload       json.RawMessage `json:"payload"` // Raw JSON for efficient marshaling
}

type EventHandler func(ctx context.Context, envelope *EventEnvelope) error

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

type RecordBatch struct {
	Records   []*kgo.Record
	Envelopes []*EventEnvelope
}