config

type ClickhouseSink struct {
	DSN   string `yaml:"dsn"`
	Table string `yaml:"table"`
}

type Conf struct {
	// Main pipeline configuration.
	Pipeline Pipeline `yaml:"pipeline"`
	// Predefined SQL tables used in the pipeline.
	Tables *Tables `yaml:"tables,omitempty"`
	// List of User-Defined Functions (UDFs) to be used in SQL queries.
	UDFs []UDF `yaml:"udfs,omitempty"`
	// List of SQL commands to execute before processing the pipeline.
	Commands []SQLCommand `yaml:"commands,omitempty"`
}

type ConsoleSink struct{}

type Error struct {
	Policy ErrorPolicy `yaml:"policy"`
}

type ErrorPolicy string

type Handler struct {
	Type               string `yaml:"type"`
	SQL                string `yaml:"sql"`
	SQLResultsCacheDir string `yaml:"sql_results_cache_dir,omitempty"`
	Table              string `yaml:"table,omitempty"`
}

type IcebergSink struct {
	CatalogName string `yaml:"catalog_name"`
	TableName   string `yaml:"table_name"`
}

type KafkaFetch struct {
	// Bytes one fetch may return per broker. Kafka's fetch.max.bytes.
	MaxBytes int `yaml:"max_bytes,omitempty" jsonschema:"minimum=1"`
	// Bytes one fetch may return per partition. Kafka's max.partition.fetch.bytes.
	MaxPartitionBytes int `yaml:"max_partition_bytes,omitempty" jsonschema:"minimum=1"`
	// Fetches held ahead of the pipeline.
	Prefetch int `yaml:"prefetch,omitempty" jsonschema:"minimum=1"`
}

type KafkaSASL struct {
	Mechanism string `yaml:"mechanism" jsonschema:"enum=PLAIN,enum=SCRAM-SHA-256,enum=SCRAM-SHA-512,enum=GSSAPI"`
	Username  string `yaml:"username"`
	Password  string `yaml:"password"`
}

type KafkaSSL struct {
	CALocation                      string `yaml:"ca_location,omitempty"`
	CertificateLocation             string `yaml:"certificate_location,omitempty"`
	KeyLocation                     string `yaml:"key_location,omitempty"`
	KeyPassword                     string `yaml:"key_password,omitempty"`
	EndpointIdentificationAlgorithm string `yaml:"endpoint_identification_algorithm,omitempty"`
}

type KafkaSink struct {
	// List of Kafka brokers.
	Brokers []string `yaml:"brokers"`
	// Target Kafka topic.
	Topic            string     `yaml:"topic"`
	SecurityProtocol string     `yaml:"security_protocol,omitempty"`
	SSL              *KafkaSSL  `yaml:"ssl,omitempty"`
	SASL             *KafkaSASL `yaml:"sasl,omitempty"`
}

type KafkaSource struct {
	Brokers         []string `yaml:"brokers"`
	GroupID         string   `yaml:"group_id"`
	AutoOffsetReset string   `yaml:"auto_offset_reset" jsonschema:"enum=earliest,enum=latest"`
	Topics          []string `yaml:"topics"`

	SecurityProtocol string     `yaml:"security_protocol,omitempty" jsonschema:"enum=SASL_SSL,enum=SSL,enum=SASL_PLAINTEXT,enum=PLAINTEXT"`
	SSL              *KafkaSSL  `yaml:"ssl,omitempty"`
	SASL             *KafkaSASL `yaml:"sasl,omitempty"`

	// Bounds how far the consumer reads ahead of the pipeline. Omit the
	// block to accept the defaults, which bound a backlog replay to a few
	// fetches rather than the backlog.
	Fetch *KafkaFetch `yaml:"fetch,omitempty"`
}

type OnError struct {
	// Defines how errors should be handled.
	Policy string `yaml:"policy" jsonschema:"enum=RAISE,enum=IGNORE,enum=DLQ"`
	// Dead-letter queue configuration. Failed messages will be routed to this
	// sink.
	DLQ *Sink `yaml:"dlq,omitempty"`
}

type Pipeline struct {
	// Name of the pipeline.
	Name string `yaml:"name,omitempty"`
	// Description of the pipeline.
	Description string `yaml:"description,omitempty"`
	// Configuration for the data source.
	Source  Source  `yaml:"source"`
	Handler Handler `yaml:"handler"`
	Sink    Sink    `yaml:"sink"`
	// Messages accumulated before the handler runs. Larger batches trade
	// latency for throughput.
	BatchSize int `yaml:"batch_size,omitempty"`
	// Longest a partial batch waits before it is invoked anyway.
	FlushIntervalSeconds int `yaml:"flush_interval_seconds,omitempty"`
	// Where the pipeline keeps its DuckDB state. Absent means in-memory, and
	// state is lost on a crash.
	State *StateConf `yaml:"state,omitempty"`
	// Global error handling strategy for the pipeline.
	OnError *OnError `yaml:"on_error,omitempty"`
}

type SQLCommand struct {
	// Name of the command for reference.
	Name string `yaml:"name"`
	// SQL statements to execute.
	SQL string `yaml:"sql"`
}

type SQLCommandSink struct {
	SQL           string                   `yaml:"sql"`
	Substitutions []SQLCommandSubstitution `yaml:"substitutions,omitempty"`
}

type SQLCommandSubstitution struct {
	Var  string `yaml:"var"`
	Type string `yaml:"type" jsonschema:"enum=uuid4"`
}

type Sink struct {
	// Sink identifier.
	Type string `yaml:"type"`
	// Format settings (e.g., for Parquet).
	Format *SinkFormat `yaml:"format,omitempty"`
	// Kafka-specific sink configuration.
	Kafka *KafkaSink `yaml:"kafka,omitempty"`
	// Console output sink configuration.
	Console *ConsoleSink `yaml:"console,omitempty"`
	// SQL-command sink configuration.
	SQLCommand *SQLCommandSink `yaml:"sqlcommand,omitempty"`
	// Iceberg-specific sink configuration.
	Iceberg *IcebergSink `yaml:"iceberg,omitempty"`
	// ClickHouse-specific sink configuration.
	Clickhouse *ClickhouseSink `yaml:"clickhouse,omitempty"`
	// Bounds how long this sink keeps trying a destination that is not
	// answering. Omit to accept the defaults; set max_attempts to 1 to turn
	// retrying off. The kafka sink ignores this: franz-go already retries a
	// produce with its own backoff.
	Retry *SinkRetry `yaml:"retry,omitempty"`
}

type SinkFormat struct {
	Type string `yaml:"type" jsonschema:"enum=parquet"`
}

type SinkRetry struct {
	// Total attempts, including the first. 1 disables retrying.
	MaxAttempts int `yaml:"max_attempts,omitempty"`
	// Wait before the first retry. Doubles each attempt.
	InitialBackoffMS int `yaml:"initial_backoff_ms,omitempty"`
	// Ceiling on the backoff.
	MaxBackoffMS int `yaml:"max_backoff_ms,omitempty"`
	// Bounds the whole ladder, not one attempt. Keep it below
	// pipeline.flush_interval_seconds: the retry runs inside the open state
	// transaction, whose clock the window depends on.
	DeadlineSeconds int `yaml:"deadline_seconds,omitempty"`
}

type Source struct {
	Type      string           `yaml:"type"`
	Kafka     *KafkaSource     `yaml:"kafka,omitempty"`
	Websocket *WebsocketSource `yaml:"websocket,omitempty"`
	Webhook   *WebhookSource   `yaml:"webhook,omitempty"`
	Error     *Error           `yaml:"error,omitempty"`
}

type StateConf struct {
	// File backing the pipeline's DuckDB database. Window state and Kafka
	// offsets are stored here and committed together.
	Path string `yaml:"path"`
}

type TableManager struct {
	TumblingWindow *TumblingWindow `yaml:"tumbling_window"`
	Sink           Sink            `yaml:"sink"`
}

type TableSQL struct {
	Name    string        `yaml:"name"`
	SQL     string        `yaml:"sql"`
	Manager *TableManager `yaml:"manager,omitempty"`
}

type Tables struct {
	// List of tables with their SQL definitions and management configurations.
	SQL []TableSQL `yaml:"sql"`
}

type TumblingWindow struct {
	CollectSQL string `yaml:"collect_closed_windows_sql"`
	DeleteSQL  string `yaml:"delete_closed_windows_sql"`
	// How often to collect closed windows. Optional; the manager applies its
	// own default when this is absent or not positive.
	PollIntervalSecs int `yaml:"poll_interval_seconds,omitempty"`
}

type UDF struct {
	FunctionName string `yaml:"function_name"`
	ImportPath   string `yaml:"import_path"`
}

type WebhookHMAC struct {
	Header string `yaml:"header"`
	SigKey string `yaml:"sig_key"`
	Secret string `yaml:"secret"`
}

type WebhookSource struct {
	SignatureType string       `yaml:"signature_type,omitempty" jsonschema:"enum=hmac"`
	HMAC          *WebhookHMAC `yaml:"hmac,omitempty"`
}

type WebsocketSource struct {
	URI string `yaml:"uri"`
}