spec

type Bootstrap struct {
	Mode    BootstrapMode    `json:"mode,omitempty"`
	StartAt BootstrapStartAt `json:"startAt,omitempty"`
	// Position is the explicit position string when StartAt is "explicit".
	Position string `json:"position,omitempty"`
}

type BootstrapMode string

type BootstrapStartAt string

type CDCConfig struct {
	// Plugin selects the logical decoding plugin: "pgoutput" (default) or
	// "wal2json".
	Plugin string `json:"plugin,omitempty"`
	// InitialWaitTime is the maximum number of seconds the reader waits for
	// the first WAL message before failing with a non-retryable error (min
	// 30, default 300). It detects a misconfigured slot or publication that
	// would otherwise hang forever.
	InitialWaitTime int `json:"initialWaitTime,omitempty"`
}

type ColumnDecl struct {
	// Scalar holds the type string when this entry is a plain JSON string.
	// Empty when Struct/List/Map is set.
	Scalar string
	// Struct holds the field declarations when this entry is
	// {"struct": {name: decl, ...}}. Nil otherwise.
	Struct map[string]ColumnDecl
	// List holds the element declaration when this entry is
	// {"list": decl}. Nil otherwise.
	List *ColumnDecl
	// Map holds the key/value declarations when this entry is
	// {"map": {"key": decl, "value": decl}}. Nil otherwise.
	Map *MapDecl

	// From is the payload path this column is extracted from, for message
	// sources whose payload is a document rather than a row: "totals.grand"
	// reads a nested field. Empty means the column name is the path, which
	// is the common case. Only meaningful for Kafka raw/avro.
	From string
	// Required fails the record when the declared field is absent from the
	// payload, instead of landing NULL. A field can be missing because the
	// contract says it is optional or because the producer broke, and only
	// the operator knows which — so it is declared per column rather than
	// being one global policy.
	Required bool
}

type CommitMode string

type CompactionConfig struct {
	// Interval between compaction runs. Go duration syntax. Default "5m".
	// Measured from the previous run's COMPLETION, not from a fixed
	// schedule: a run longer than the interval does not re-fire immediately
	// (the interval is a throttle between runs, not a start-time cadence).
	Interval string `json:"interval,omitempty"`
	// TargetFileSize is the desired output file size after compaction, e.g.
	// "512Mi". Default "512Mi" (iceberg-go's own default).
	TargetFileSize string `json:"targetFileSize,omitempty"`
	// MinInputFiles is the minimum number of small files in a partition
	// group before compaction rewrites it. Default 5. 0 also means
	// "unset, use the default" — omitempty means the wire form cannot
	// distinguish an explicit 0 from an absent field, so "compact every
	// group down to a single file" is not expressible here. Use 1 for
	// "almost always compact" if that is the intent.
	MinInputFiles uint `json:"minInputFiles,omitempty"`
}

type Defaults struct {
	WriteMode      WriteMode `json:"writeMode,omitempty"`
	TargetFileSize string    `json:"targetFileSize,omitempty"`
}

type Enrich struct {
	// Table names the reference (diagnostics and duplicate detection).
	Table string `json:"table"`
	// Source is the reference read: a plain SQL connection and a query
	// returning the full reference image. Types matter at the join: cast
	// in SQL (CAST(id AS CHAR)) when the event column's type differs.
	Source EnrichSource `json:"source"`
	// On maps event column → reference column (the join key pair). One
	// pair today; a composite key is a future need, not a current one.
	On map[string]string `json:"on"`
	// Select limits the reference columns taken; ["*"] takes all (the
	// wildcard sugar). Empty is rejected — declare the columns explicitly
	// or use ["*"].
	Select []string `json:"select,omitempty"`
	// As renames reference columns on the way into the event (ref column
	// → destination name). Keys must appear in Select when Select is set.
	As map[string]string `json:"as,omitempty"`
	// JoinType is required — there is no universal miss policy: left
	// passes the event with NULL reference columns, inner drops the event.
	JoinType string `json:"joinType"`
	// Refresh is the full re-read interval (e.g. 5m). Empty means the
	// default (5m). A refresh swaps the map atomically: in-flight events
	// finish on the old image.
	Refresh string `json:"refresh,omitempty"`
	// OnColdStart governs events that arrive before the first reference
	// load completes: buffer (default) holds them up to BufferLimits and
	// drains in order once the reference is hot; pass processes them
	// immediately against the cold map (a miss follows JoinType); drop
	// discards them.
	OnColdStart string `json:"onColdStart,omitempty"`
	// BufferLimits bounds the cold-start queue: MaxEvents caps memory
	// (the oldest event is evacuated, and follows JoinType), MaxWait caps
	// latency (an event queued longer follows JoinType at drain time).
	BufferLimits EnrichBufferLimits `json:"bufferLimits,omitempty"`
	// MaxRows caps the reference image's row count: the broadcast join
	// holds the whole reference in RAM (refTable + a keyIndex entry per
	// row), so an unbounded reference is an unbounded, unannounced OOM.
	// Zero means the package default (enrich.DefaultMaxRows). A load that
	// returns more rows than this is rejected loudly, not truncated
	// silently — the reference is small by contract; a reference that
	// outgrows this needs a raised limit or a different tool, not a
	// silent truncation.
	MaxRows int `json:"maxRows,omitempty"`
}

type EnrichBufferLimits struct {
	MaxEvents int    `json:"maxEvents,omitempty"`
	MaxWait   string `json:"maxWait,omitempty"`
}

type EnrichSource struct {
	URI   string `json:"uri"`
	Query string `json:"query"`
}

type Filter struct {
	All       []Filter   `json:"all,omitempty"`
	Any       []Filter   `json:"any,omitempty"`
	Not       *Filter    `json:"not,omitempty"`
	Predicate *Predicate `json:"where,omitempty"`
}

type Maintenance struct {
	Enabled        bool                  `json:"enabled,omitempty"`
	Compaction     *CompactionConfig     `json:"compaction,omitempty"`
	SnapshotExpiry *SnapshotExpiryConfig `json:"snapshotExpiry,omitempty"`
	OrphanCleanup  *OrphanCleanupConfig  `json:"orphanCleanup,omitempty"`
}

type MapDecl struct {
	Key   ColumnDecl `json:"key"`
	Value ColumnDecl `json:"value"`
}

type OnDelete string

type Operator string

type OrphanCleanupConfig struct {
	// Interval between cleanup runs. Go duration syntax. Default "1h".
	// Measured from the previous run's COMPLETION, not from a fixed
	// schedule (a throttle between runs, not a start-time cadence).
	Interval string `json:"interval,omitempty"`
	// OlderThan: only files older than this are eligible for deletion. Go
	// duration syntax. Default "72h" (3 days) — iceberg-go's own default.
	// At least MinOrphanCleanupOlderThan (1h).
	OlderThan string `json:"olderThan,omitempty"`
}

type PostgresSource struct {
	Host     string            `json:"host,omitempty"`
	Port     int               `json:"port,omitempty"`
	Database string            `json:"database,omitempty"`
	Username string            `json:"username,omitempty"`
	Password string            `json:"password,omitempty"`
	Params   map[string]string `json:"params,omitempty"`
	SSL      *SSLConfig        `json:"ssl,omitempty"`
	SSH      *SSHConfig        `json:"ssh,omitempty"`
	// MaxThreads limits the number of concurrent connections for snapshot
	// chunk SELECTs. 1..32, default runtime.NumCPU() (clamped to 32).
	MaxThreads int `json:"maxThreads,omitempty"`
	// RetryCount is the number of transient-connection retries with
	// exponential backoff before failing. Default 3. The field is
	// omitempty, so an explicit 0 is indistinguishable from omission and
	// resolves to the default rather than disabling retries.
	RetryCount int `json:"retryCount,omitempty"`
	// Schemas limits table discovery to these schemas. Empty discovers every
	// accessible schema. Ignored unless Discover is set.
	Schemas []string `json:"schemas,omitempty"`
	// Discover replaces the explicit tables list with every table the
	// connected user may SELECT (base tables and partitioned parents), so a
	// pipeline can replicate a whole schema without enumerating it.
	// Mutually exclusive with an explicit tables list.
	Discover bool `json:"discover,omitempty"`
	// CDC holds the logical-decoding knobs for the replication reader.
	CDC *CDCConfig `json:"cdc,omitempty"`
}

type Predicate struct {
	Column string   `json:"col"`
	Op     Operator `json:"op"`
	Value  any      `json:"value,omitempty"`
}

type SSHConfig struct {
	Host       string `json:"host,omitempty"`
	Port       int    `json:"port,omitempty"`
	Username   string `json:"username,omitempty"`
	Password   string `json:"password,omitempty"`
	PrivateKey string `json:"privateKey,omitempty"`
	// Passphrase decrypts the private key when encrypted. Empty means
	// unencrypted.
	Passphrase string `json:"passphrase,omitempty"`
	// KnownHosts is the OpenSSH known_hosts file the bastion's host key is
	// verified against. Empty falls back to ~/.ssh/known_hosts. Host key
	// verification is never disabled: a missing file is an error, not a
	// silent trust-everything.
	KnownHosts string `json:"knownHosts,omitempty"`
}

type SSLConfig struct {
	// Mode selects the TLS behavior: "disable" (default), "require",
	// "verify-ca", "verify-full".
	Mode string `json:"mode,omitempty"`
	// CA is the path to the server CA certificate PEM file (used by
	// verify-ca and verify-full).
	CA string `json:"ca,omitempty"`
	// Cert is the path to the client certificate PEM file (mutual TLS).
	Cert string `json:"cert,omitempty"`
	// Key is the path to the client private key PEM file (mutual TLS).
	Key string `json:"key,omitempty"`
}

type Sink struct {
	// Type selects the sink implementation ("iceberg+rest"). Empty defaults
	// to "iceberg+rest"; a future sink declares its own type and the spec
	// names it explicitly.
	Type         string   `json:"type,omitempty"`
	URI          string   `json:"uri"`
	Namespace    string   `json:"namespace"`
	Warehouse    string   `json:"warehouse,omitempty"`
	ClientID     string   `json:"clientId,omitempty"`
	ClientSecret string   `json:"clientSecret,omitempty"`
	Scope        string   `json:"scope,omitempty"`
	Defaults     Defaults `json:"defaults"`
	// CommitMode selects how a sink that cannot commit data and position in
	// one atomic write sequences the two. Couchbase-only: any other sink
	// type rejects a non-empty commitMode in Validate rather than silently
	// ignoring it. Empty means the sink's own default ("fast": data first,
	// control document last — recovery replays the batch idempotently).
	// "atomic" wraps data and control document in a distributed transaction,
	// closing the recovery window at the cost of transaction overhead per
	// batch.
	CommitMode CommitMode `json:"commitMode,omitempty"`
	// EvolveSchema opts into additive schema evolution on the target table.
	// Off by default: the sink is fail-closed on schema divergence and
	// hard-errors until an operator intervenes. When on, EnsureTable evolves
	// an existing table to the resolved schema by adding missing columns
	// (nullable) and applying only Iceberg's allowed type promotions
	// (int→long, float→double, decimal widening, timestamp precision
	// widening) — never a narrowing, rename, or drop. Iceberg-only: other
	// sinks ignore it.
	EvolveSchema bool `json:"evolveSchema,omitempty"`
	// Maintenance configures background Iceberg table maintenance
	// (compaction, snapshot expiry, orphan cleanup). Iceberg-only: any
	// other sink type rejects a maintenance block in Validate rather than
	// silently ignoring it. Nil disables it entirely.
	Maintenance *Maintenance `json:"maintenance,omitempty"`
}

type SnapshotExpiryConfig struct {
	// Interval between expiry runs. Go duration syntax. Default "10m".
	// Measured from the previous run's COMPLETION, not from a fixed
	// schedule (a throttle between runs, not a start-time cadence).
	Interval string `json:"interval,omitempty"`
	// RetainLast is the minimum number of snapshots kept regardless of age.
	// Default 1.
	RetainLast int `json:"retainLast,omitempty"`
	// MaxAge is the safety window described above: no snapshot younger than
	// this is ever expired. Go duration syntax. Default "168h" (7 days).
	MaxAge string `json:"maxAge,omitempty"`
}

type Source struct {
	Kind string `json:"kind"`
	URI  string `json:"uri"`
	// SnapshotURI is a READ-ONLY connection string the workers use for the
	// snapshot chunk SELECT. The full replication URI stays coordinator-only;
	// the worker never needs replication credentials, so a deployment can
	// grant the worker a SELECT-only user. When empty, URI is used (the
	// pre-scoping behavior).
	SnapshotURI string `json:"snapshotUri,omitempty"`
	// ServerID is the MySQL replication server_id. It MUST be unique across
	// every pipeline reading from the same source server: two pipelines
	// sharing one server_id fight over the same replication stream and
	// corrupt it. The operator's admission webhook rejects a duplicate
	// serverId within a namespace.
	ServerID      string `json:"serverId,omitempty"`
	SlotName      string `json:"slotName,omitempty"`
	SnapshotMode  string `json:"snapshotMode,omitempty"`
	BootstrapAdds string `json:"bootstrapServers,omitempty"`
	GroupID       string `json:"groupId,omitempty"`
	// MaxReconnectAttempts bounds the MySQL binlog reader's reconnect budget
	// (go-mysql canal's max_reconnect_attempts). 0 (omitted) resolves to the
	// default 3. Without a bound a permanently broken stream — purged binlog,
	// revoked grant, server_id collision — retries forever instead of failing.
	MaxReconnectAttempts int `json:"maxReconnectAttempts,omitempty"`
	// PartitionedByPrimaryKey declares the Kafka topics are partitioned by
	// the key, so ordering (and thus upsert correctness) holds within a
	// key. Kafka only orders inside a partition: if the same key landed in
	// different partitions, an upsert could apply a stale version silently.
	// The engine cannot verify this — it must be a conscious operator
	// assertion. Required for writeMode: upsert on a Kafka source.
	PartitionedByPrimaryKey bool `json:"partitionedByPrimaryKey,omitempty"`
	// Format selects the Kafka message decoder: "debezium" (default) parses
	// the envelope into typed rows; "raw" lands the payload verbatim
	// without interpreting it (bronze landing); "avro" decodes
	// Confluent-Avro records resolved by schema id from the registry. Raw
	// and avro require append-only tables.
	Format string `json:"format,omitempty"`
	// SchemaRegistry is the Confluent-compatible schema registry base URL
	// (e.g. http://registry:8081), required when format is avro.
	SchemaRegistry string `json:"schemaRegistry,omitempty"`
	// Postgres configures a PostgreSQL source via structured fields instead
	// of a URI. When present, uri is ignored for connection building (but
	// slotName and snapshotUri remain flat). Nil means the source uses uri.
	Postgres *PostgresSource `json:"postgres,omitempty"`
}

type Spec struct {
	Pipeline string  `json:"pipeline"`
	Source   Source  `json:"source"`
	Sink     Sink    `json:"sink"`
	Tables   []Table `json:"tables"`
}

type Table struct {
	Source      string    `json:"source"`
	Target      string    `json:"target"`
	PrimaryKey  []string  `json:"primaryKey,omitempty"`
	PartitionBy []string  `json:"partitionBy,omitempty"`
	Filter      *Filter   `json:"filter,omitempty"`
	WriteMode   WriteMode `json:"writeMode,omitempty"`
	OnDelete    OnDelete  `json:"onDelete,omitempty"`
	// Identity declares the transport-metadata columns that make an
	// append-idempotent table logically idempotent. Each entry is the
	// destination column (as) of a transport metadata column declared in
	// Metadata. Empty outside append-idempotent.
	Identity []string `json:"identity,omitempty"`
	// Workers partitions this table's work across N worker groups by a
	// contiguous primary-key range — like Spark/Flink partition an
	// executor pool over a hot table, instead of capping it at one
	// worker's throughput. Nil or Number<=1 means today's single-worker
	// behavior: no partitioning, one group. A worker group's name is
	// always derived (see Table.WorkerGroupNames) — there is no
	// operator-chosen worker name. CPU/Memory are Kubernetes resource
	// quantities (e.g. "2", "4Gi") applied to every worker Deployment
	// this table provisions; ignored outside Kubernetes provisioning.
	Workers           *WorkerSpec `json:"workers,omitempty"`
	CreateIfNotExists bool        `json:"createIfNotExists,omitempty"`
	FilterImmutable   bool        `json:"filterImmutable,omitempty"`
	// Metadata lands pipeline metadata columns (op, commit_ts, position, ...)
	// in the target table. The destination name is explicit via As.
	Metadata []core.MetadataColumn `json:"metadata,omitempty"`
	// Cast overrides one source column's canonical type. Key is the source
	// column name; value is the textual canonical target (e.g. "string",
	// "decimal(20,4)", "timestamptz(assume_utc)").
	Cast map[string]string `json:"cast,omitempty"`
	// Columns defines the source schema explicitly for sources that cannot
	// be introspected (e.g. Kafka). Each entry is either a scalar type
	// string ("int64", "decimal(20,4)") or a composite declaration
	// (struct/list/map, nested to any depth) — see ColumnDecl. Ignored for
	// SQL sources which introspect automatically.
	Columns map[string]ColumnDecl `json:"columns,omitempty"`
	// ColumnFilter limits the source columns read and emitted (snapshot
	// SELECT list, CDC projection, and the resolved schema) to this subset.
	// Empty means all columns. The filter is applied at the source boundary,
	// before the Arrow hot-path. In upsert mode every primary-key column
	// must be listed — the sink resolves the equality key by column name,
	// so an excluded key column would break the target table.
	ColumnFilter []string `json:"columnFilter,omitempty"`
	// ChunkColumn selects the snapshot chunking strategy for SQL sources.
	// Empty uses the source's default (Postgres: physical CTID ranges).
	// When set, it must name a primary-key column: an integer/float column
	// splits by value range (batch-size), any other type by cursor stepping
	// (next-query). A key-ordered column keeps the chunk range routable to
	// the same worker as the live stream when workers > 1.
	ChunkColumn string `json:"chunkColumn,omitempty"`
	// Mode selects the sync model: "" or "cdc" (default, log-based) or
	// "incremental" (cursor column, no replication slot). Only sources that
	// declare ModeIncremental support it (Postgres today).
	Mode string `json:"mode,omitempty"`
	// Cursor names the column incremental mode orders by (e.g. updated_at).
	// Required when Mode is "incremental".
	Cursor string `json:"cursor,omitempty"`
	// Bootstrap configures how the initial snapshot is handled.
	Bootstrap *Bootstrap `json:"bootstrap,omitempty"`
	// Enrich joins each event against reference tables loaded in memory
	// (broadcast hash join): the reference is read whole at boot and fully
	// re-read on a refresh interval — never CDC'd, never queried per event.
	// Applied in declaration order; an inner-join miss at any reference
	// drops the event.
	Enrich []Enrich `json:"enrich,omitempty"`
}

type ValidateOption func(*validateOptions)

type WorkerSpec struct {
	Number int `json:"number,omitempty"`
	// Max caps how many partitions this table may be scaled to at
	// runtime (issue #312). Zero means the pipeline-wide default. A table
	// that sets it ignores the global cap.
	Max    int    `json:"max,omitempty"`
	CPU    string `json:"cpu,omitempty"`
	Memory string `json:"memory,omitempty"`
}

type WriteMode string