source

type Capabilities struct {
	Snapshot   bool
	ChunkQuery bool
	Stream     bool
	// MaxConnections is the upper bound on concurrent query connections
	// the source tolerates. 0 means no opinion (unlimited).
	MaxConnections int
	// Modes lists the synchronization modes this source supports.
	Modes []Mode
	// BeforeImage reports whether the source carries the row image that was
	// deleted on a DELETE. MySQL (binlog row) always does; Postgres only
	// per replica identity; Kafka does not (tombstones are null). Append
	// tables with onDelete: record depend on this.
	BeforeImage bool
	// MonotonicSequence reports whether the transport coordinate of a
	// message never reappears (Kafka offset, Kinesis sequence) — the
	// property append-idempotent relies on.
	MonotonicSequence bool
	// RecoverablePosition reports whether the source's position can be
	// used to resume exactly from where it left off. Sources without
	// replay (TCP firehose, webhooks) must declare false so the
	// coordinator does not attempt gapless recovery on them.
	RecoverablePosition bool
}

type Chunk struct {
	Low  []any
	High []any
}

type ChunkSource interface {
	// PK returns the primary key columns, in key order.
	PK() []string
	// Bounds returns the ordered chunk boundary keys.
	Bounds(ctx context.Context) ([][]any, error)
	// Scan reads one chunk, calling fn per row keyed by column name.
	Scan(ctx context.Context, ch Chunk, fn func(row map[string]any) error) error
}

type Discoverer interface {
	// Discover returns every table the connected user may SELECT, limited to
	// schemas when non-empty.
	Discover(ctx context.Context, schemas []string) ([]TableRef, error)
}

type IncrementalSource interface {
	Incremental(ctx context.Context, t TableRef, cursor string, after string) (next string, rows []map[string]any, err error)
}

type Introspector interface {
	Introspect(ctx context.Context, t spec.Table) (core.TableRef, core.Schema, []core.Warning, error)
}

type Mode uint8

type PartitionSource interface {
	// Partitions returns n contiguous ranges covering the table's whole
	// primary-key domain, in order (Partitions(ctx,1)[0] is always the
	// single unbounded range). Returns an error if the primary key shape
	// isn't supported for range partitioning (e.g. more than one column).
	Partitions(ctx context.Context, n int) ([]Chunk, error)
}

type Positioner interface {
	// InitialPosition is the stream start for a first boot (no resume).
	InitialPosition(ctx context.Context) (position.Position, error)
	// ParsePosition decodes a stored cdc.position string.
	ParsePosition(s string) (position.Position, error)
}

type QuerySource interface {
	// NewChunker builds the chunk SELECT source for one table.
	NewChunker(source, pk string, chunkSize int) (ChunkSource, error)
	// CloseQuery releases the source's query connection.
	CloseQuery() error
}

type Reader interface {
	SourceReader
	// Start begins replication from `from`. Called once before the first
	// Next. Sources that need a resume position (MySQL GTID set, Postgres
	// LSN, Kafka offsets) apply it here; sources without replay ignore it.
	Start(ctx context.Context, from position.Position) error
	// Next returns the next columnar batch, or (nil, nil) at a clean end.
	// The caller owns the returned batch and must Release() it.
	Next(ctx context.Context) (*dataplane.Batch, error)
	Close()
	// SetConfirmed installs a callback that returns the minimum position
	// committed to the sink across all tables. The Postgres reader uses
	// this to advance the slot's confirmed_flush_lsn only to the point
	// that has been durably written — never past it.
	SetConfirmed(f func() position.Position)
}

type Runtime struct {
	ServerID  uint32
	Heartbeat time.Duration
	Logger    *slog.Logger
}

type SchemaSetter interface {
	SetSourceSchemas(schemas map[string]core.Schema)
}

type Source interface {
	Streamer
	Positioner
	Introspector
}

type SourceReader interface {
	Synced() position.Position
	Master(ctx context.Context) (position.Position, error)
	// OpenWindow starts the DBLog window for chunkID: the reader captures
	// its own source watermark and, from now on, tags decoded events whose
	// position is strictly past that watermark InWindow for the chunk.
	// Events at or before the watermark are already reflected in the chunk
	// SELECT and must not be tagged. ClearWindow ends the window.
	OpenWindow(ctx context.Context, chunkID uint32)
	ClearWindow()
}

type Streamer interface {
	Open(ctx context.Context, refs []TableRef) (Reader, error)
}

type TableRef = core.TableRef