source

type ChangeEvent struct {
	Table     string          `json:"table"`
	Operation string          `json:"operation"` // insert, update, delete
	Data      json.RawMessage `json:"data"`
	OldData   json.RawMessage `json:"old_data,omitempty"`
	Timestamp time.Time       `json:"timestamp"`
}

type PGListener interface {
	// Connect establishes the database connection.
	Connect(ctx context.Context, dsn string) error
	// Listen starts listening on the given channel.
	Listen(ctx context.Context, channel string) error
	// WaitForNotification blocks until a notification arrives or context is cancelled.
	WaitForNotification(ctx context.Context) (payload string, err error)
	// Close closes the connection.
	Close(ctx context.Context) error
}

type PostgresCDCConfig struct {
	DSN          string        `json:"dsn" yaml:"dsn"`
	Tables       []string      `json:"tables" yaml:"tables"`
	Channel      string        `json:"channel" yaml:"channel"`             // LISTEN/NOTIFY channel name
	PollInterval time.Duration `json:"poll_interval" yaml:"poll_interval"` // fallback polling interval
}

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

type RedisStreamClient interface {
	// Connect establishes the Redis connection.
	Connect(ctx context.Context, addr string) error
	// CreateGroup creates a consumer group if it does not already exist.
	// The "startID" parameter is typically "$" (latest) or "0" (all history).
	CreateGroup(ctx context.Context, stream, group, startID string) error
	// ReadGroup reads messages from the stream using the consumer group.
	// Returns up to count messages; blocks until messages are available or ctx is cancelled.
	ReadGroup(ctx context.Context, stream, group, consumer string, count int) ([]RedisStreamMessage, error)
	// Ack acknowledges processing of the given message IDs.
	Ack(ctx context.Context, stream, group string, ids ...string) error
	// Close closes the Redis connection.
	Close() error
}

type RedisStreamConfig struct {
	Addr      string `json:"addr" yaml:"addr"`
	Stream    string `json:"stream" yaml:"stream"`
	Group     string `json:"group" yaml:"group"`       // consumer group name
	Consumer  string `json:"consumer" yaml:"consumer"` // consumer instance name
	BatchSize int    `json:"batch_size" yaml:"batch_size"`
}

type RedisStreamMessage struct {
	ID     string            // stream message ID (e.g., "1645000000000-0")
	Fields map[string]string // key-value pairs from the stream entry
}

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

type SQSBatchEntry struct {
	ID         string
	Body       string
	Attributes map[string]string
}

type SQSBatchResult struct {
	ID        string
	MessageID string
	Error     error
}

type SQSClient interface {
	// ReceiveMessages polls the SQS queue for up to maxMessages.
	ReceiveMessages(ctx context.Context, queueURL string, maxMessages, waitTimeSeconds int) ([]SQSMessage, error)
	// DeleteMessage removes a message from the queue after successful processing.
	DeleteMessage(ctx context.Context, queueURL, receiptHandle string) error
	// SendMessage sends a message to the SQS queue.
	SendMessage(ctx context.Context, queueURL, body string, attributes map[string]string) (messageID string, err error)
	// SendMessageBatch sends multiple messages to the SQS queue.
	SendMessageBatch(ctx context.Context, queueURL string, entries []SQSBatchEntry) ([]SQSBatchResult, error)
}

type SQSConfig struct {
	QueueURL        string `json:"queue_url" yaml:"queue_url"`
	Region          string `json:"region" yaml:"region"`
	MaxMessages     int    `json:"max_messages" yaml:"max_messages"`
	WaitTimeSeconds int    `json:"wait_time_seconds" yaml:"wait_time_seconds"`
}

type SQSMessage struct {
	MessageID     string
	ReceiptHandle string
	Body          string
	Attributes    map[string]string
}

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

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