Engine

Everything that actually moves data lives in one Go module, engine/. It has no HTTP server, no database, no scheduler — it is a pure library. That is deliberate: the cloud worker and the on-prem agent are two thin hosts around the same engine code, so a Postgres read behaves identically whether it happens in our infrastructure or inside your VPC. The engine is built from four parts that compose cleanly:

• Connectors — a Source reads typed records out of an external system; a Sink writes them in.
• The registry — maps a connector kind plus a JSON config to a live Source/Sink, and carries the declarative spec the console renders forms from.
• Operators — compiled from the pipeline document into an ordered chain applied between source and sink.
• The runner — Copy(ctx, src, sink, opts) streams source → operators → sink in one pass, tracking rows read, written, and quarantined.

The dependency arrows all point inward. connector depends on record and nothing else; operator depends on connector, record, and transform; the registry depends on operator — never the reverse (the registry declares a small operator.Registrar interface so operators register into it without an import cycle). The runner sits at the top and knows only the interfaces. No concrete connector name appears anywhere in the core.

Connectors: Source, Sink, Definer

The whole system hangs on two interfaces. A Source can Test its credentials, Discover the streams it exposes (tables, files, collections, API resources), and open a Read iterator over one stream — the cursor argument is an opaque incremental marker ("" means a full snapshot). A Sink can Test, CreateOrMigrateSchema to make the target accept a schema, and Write a batch under a WriteMode. The engine core knows nothing connector-specific.

// Source reads typed records out of an external system.
type Source interface {
	Test(ctx context.Context) error
	Discover(ctx context.Context) ([]Stream, error)
	// cursor is an opaque incremental marker ("" for a full snapshot).
	Read(ctx context.Context, stream Stream, cursor string) (RecordIterator, error)
	Close() error
}

// Sink writes typed records into an external system.
type Sink interface {
	Test(ctx context.Context) error
	CreateOrMigrateSchema(ctx context.Context, stream Stream) error
	Write(ctx context.Context, stream Stream, batch *record.Batch, mode WriteMode) (WriteResult, error)
	Close() error
}

// RecordIterator yields batches until io.EOF, then must Close.
type RecordIterator interface {
	Next(ctx context.Context) (*record.Batch, error)
	Close() error
}

The iterator is a pull model: the runner loops on Next until it returns io.EOF, then Closes. This is what keeps memory bounded — a Postgres source with a hundred-million-row table never materializes more than one batch at a time; it holds an open server cursor and hydrates the next chunk only when asked. Backpressure is implicit: a slow sink means the runner calls Next less often, so the source reads less often.

Two optional interfaces extend a connector without complicating the common case. A Definer validates a user-defined dataset (a SQL query, a file glob) and returns the resulting Stream with an inferred schema. A StreamFinalizer lets a sink commit deferred per-stream work — the runner calls FinishStream once, after the last batch wrote cleanly, which is how an atomic-overwrite sink stages every row and swaps it into the target at the very end.

// Definer: connectors that support user-defined datasets (custom SQL, globs).
type Definer interface {
	Describe(ctx context.Context, def json.RawMessage) (Stream, error)
}

// StreamFinalizer: called once after the last batch of a stream writes
// successfully (never on dry runs) — e.g. an atomic overwrite swap.
type StreamFinalizer interface {
	FinishStream(ctx context.Context, stream Stream, mode WriteMode) error
}

These are the load-bearing shape of Setu's plugin model: a connector implements exactly what it can do. A read-only REST source omits the sink half; a CSV file needs no Definer; only a sink that actually stages implements StreamFinalizer. The runner discovers a capability with a type assertion (dst.(connector.StreamFinalizer)) and simply skips connectors that don't have it.

A Stream is the discoverable unit — a namespace, a name, an advisory Type (table/view/file/collection/api, cosmetic only), a *record.Schema, an optional PrimaryKey (which is what makes upsert and merge possible), and an opaque Definition for user-defined datasets. SelectStreams resolves a requested name list against a discovered set: an empty list means "everything," a name matches by qualified namespace.name or bare name, and a requested name that matches nothing is a hard error — so a stale selection fails loudly instead of silently copying the wrong set.

Incremental reads: pushdown or client filter

The opaque cursor string in Read is the seam for incremental replication, and it degrades gracefully. A source that can push a predicate to the server implements the optional IncrementalSource capability; one that can't gets a client-side filter wrapper for free, so incremental works everywhere but is efficient where the backend allows it.

// IncrementalSource is an OPTIONAL Source capability: it pushes a cursor
// predicate to the server. Sources that don't implement it use a client-side
// filter (ClientFilterIterator) instead.
type IncrementalSource interface {
	// Rows WHERE field >= watermark ORDER BY field ASC, batched. watermark nil =
	// full ordered scan (first backfill).
	ReadIncremental(ctx context.Context, stream Stream, field string, watermark *Cursor) (RecordIterator, error)
	// CountRemaining returns the delta size for the progress denominator.
	CountRemaining(ctx context.Context, stream Stream, field string, watermark *Cursor) (n int64, ok bool, err error)
}

Postgres, for example, translates a watermark into WHERE field >= $1 ORDER BY field and streams only the delta; a flat-file source has no server to push to, so the engine wraps its plain iterator in a clientFilterIterator that drops rows below the watermark and tracks the running max. The typed Cursor and its watermark semantics — including how NULL cursor values are handled — are covered in the record model.

The registry: kind → connector, and the spec

The registry is Setu's connector SPI. A connector registers one factory under its kind, and the engine builds a live Source or Sink from a kind plus a JSON config. It is not a global — a host constructs one and registers exactly the connectors and operators it wants available. Adding a connector is, quite literally, registering one factory.

type SourceFactory func(ctx context.Context, config []byte) (connector.Source, error)
type SinkFactory   func(ctx context.Context, config []byte) (connector.Sink, error)

// RegisterConnector registers a spec plus source/sink factories in one call.
// Pass nil for a role the connector does not support.
func (r *Registry) RegisterConnector(spec connector.PluginSpec, src SourceFactory, sink SinkFactory)

func (r *Registry) BuildSource(ctx context.Context, kind string, config []byte) (connector.Source, error)
func (r *Registry) BuildSink(ctx context.Context, kind string, config []byte) (connector.Sink, error)

The same registry also holds operators and their specs, so the catalog the console draws from is one object: connector factories, operator factories, and — via RegisterNodeSpec — spec-only entries for the terminal source/destination nodes that the pipeline compiler realizes into connectors rather than stream processors. connectors.Default() returns a registry pre-loaded with the eleven built-in connectors (Postgres, MySQL, Oracle, SQL Server, ClickHouse, Snowflake, SQLite, MongoDB, CSV, JSON, REST) plus the thirteen default operators; Domain Packs register more on top. Both the apiserver (for spec-driven forms) and the worker (for execution) call the same wiring, so the available set can never drift between "what the UI offers" and "what the engine can run."

Alongside each factory the registry stores a PluginSpec — a declarative descriptor with the connector's display name, category, roles, supported write modes, and its config Fields. The console renders every connection form from this spec, so no connector form is ever hand-written. The same mechanism declares dataset capability: whether Discover lists datasets, and which fields a custom dataset needs.

var Spec = connector.PluginSpec{
	Kind:        "csv",
	DisplayName: "CSV file",
	Category:    "Files",
	Roles:       []connector.Role{connector.RoleSource, connector.RoleSink},
	WriteModes:  []connector.WriteMode{connector.WriteAppend, connector.WriteOverwrite},
	Fields: []connector.ConfigField{
		{Name: "path", Label: "File path", Type: connector.FieldString, Required: true},
		{Name: "delimiter", Label: "Delimiter", Type: connector.FieldString, Default: ","},
		{Name: "header", Label: "Has header row", Type: connector.FieldBool, Default: true},
	},
	Datasets: connector.DatasetCapability{Discover: true, /* Define: … custom globs */},
}

One source of truth for every config form

A ConfigField carries enough to render an input: Type (string, secret, int, bool, select, connection), a label, default, placeholder, help text, and select options. The engine/schema package compiles a field list into a JSON Schema (draft 2020-12): the apiserver serves it, the web builder renders forms from it, and the Monaco code editor uses it for completion and validation. Secrets become format: password with an x-setu-secret marker; a connection field becomes x-setu-connection. Because the schema is generated from the same field list the factory reads, a config UI can never drift from what the connector actually accepts.

Operators: the compiled chain

Setu's execution model is "everything is an operator." A pipeline is a graph of operator nodes whose Kind classifies them: source and destination are the terminal adapters over connectors (no input / no output port), and the interior transform and validate kinds reshape or gate data. Every operator carries a declarative Spec (like a connector's) so the palette and inspector forms are data-driven, and it declares Ports — linear pipelines use a single in/out, but the field exists so a DAG scheduler can wire a join's left/right later without changing the operator contract.

An operator is built from JSON config by a Factory, bound to its input stream via Prepare (which derives the output stream — schema, primary key, name), and invoked per batch via Process. The pipeline document compiles into an ordered Chain; an empty chain is an identity copy.

type Operator interface {
	Spec() Spec
	Prepare(in connector.Stream) (connector.Stream, error)
	// rejected carries rows removed for cause (with reasons) for quarantine;
	// rows intentionally filtered out are simply absent, not rejected.
	Process(ctx context.Context, b *record.Batch) (out *record.Batch, rejected []transform.Rejected, err error)
}

// Compile binds an ordered operator list to a concrete input stream, threading
// each operator's output as the next operator's input.
func Compile(in connector.Stream, ops []Operator, names []string) (*Chain, error)
func (c *Chain) Output() connector.Stream // the stream the sink will see
func (c *Chain) Apply(ctx, in *record.Batch) (final *record.Batch, trace []StageOutput, err error)

The built-in operators are select, compute, expression, cast, filter, validate, limit, dedupe, regex, jsonextract, split, flatten, and aggregate. Two distinctions in Process matter. First, rows an operator rejects for cause are returned separately as transform.Rejected (the row plus human-readable reasons) so the runner can quarantine them; rows a filter simply drops are just absent from out, never rejected. Second, Apply threads a batch through every stage and returns a trace — one StageOutput per stage, with its emitted rows and its rejects — which is exactly what the preview API renders "data after each stage" from, and what the runner attributes quarantine counts to.

An optional Flusher capability lets a buffering operator (like aggregate) emit its accumulated result once, after the last batch, via Chain.Finish. Finish walks operators in order; when one flushes, its output is threaded through every downstream operator's Process so a later aggregator can buffer it before its own flush. Non-buffering operators simply don't implement Flusher, so the SPI is unchanged for them.

The runner: one streaming pass

runner.Copy is the single code path both hosts use. It resolves the write mode (defaulting to append), discovers streams if none were pinned, and copies each stream in turn, accumulating Stats. Per stream it compiles the operator chain, prepares the target schema (unless it is a dry run), then loops: pull a batch, apply the chain, quarantine any rejects, write the survivors, emit progress. Afterward it drains buffering operators and calls a StreamFinalizer if the sink is one.

// Copy moves all selected streams from src to dst and returns aggregate
// stats. It is the single code path used by both the worker and the agent.
func Copy(ctx context.Context, src connector.Source, dst connector.Sink, opts Options) (Stats, error)

// … inside copyStream, the streaming core:
for {
	if err := ctx.Err(); err != nil { return st, err } // honor cancellation
	batch, err := it.Next(ctx)
	if errors.Is(err, io.EOF) { break }
	st.RowsRead += int64(batch.Len())

	final, trace, err := chain.Apply(ctx, batch)      // run operators
	stageStats = accumulateStages(stageStats, target.Qualified(), batch.Len(), trace)
	for _, stage := range trace {                      // quarantine rejects
		st.RowsQuarantined += int64(len(stage.Rejected))
		if opts.OnReject != nil { opts.OnReject(target.Qualified(), stage.Name, stage.Rejected) }
	}
	if opts.DryRun {
		st.RowsWritten += int64(final.Len())           // count, never write
	} else if final.Len() > 0 {
		res, _ := dst.Write(ctx, target, final, mode)  // stream to sink
		st.RowsWritten += res.RowsWritten
	}
	opts.OnProgress(Progress{RowsRead: st.RowsRead, RowsWritten: st.RowsWritten})
}

Options is the whole control surface. It carries the write Mode; an optional Streams restriction (empty means "discover everything"); a TargetPrefix that namespaces target names (e.g. staging_) while keeping schema and namespace; the Operators builder (returning the ordered chain and per-stage display names for a stream); a DryRun flag that reads and transforms but never issues DDL or writes; a Total row denominator for a determinate progress ring; and three callbacks.

The loop checks ctx.Err() before every batch, so cancelling the context stops the copy at a batch boundary and returns the partial stats — no half-written batch, no orphaned iterator (the defer it.Close() always runs). Every error is wrapped with the qualified stream name, so a failure names the stream, the phase (read / transform / write / finalize), and the underlying cause.

One engine, two hosts

The worker and the agent differ only in where they run and how they get config — never in what Copy does. The worker shares the control plane's environment and reads jobs from a queue; the agent runs inside your network and receives a serialized job (including a blob Spec) over an authenticated channel. Both build the same connectors from the same registry and call the same Copy.

The one execution choice that lives above the engine, in the worker, is direct vs. staged. A small incremental delta (at or below SETU_DIRECT_MAX_ROWS, default 50,000) streams directly source → operators → destination. Anything larger — or any hybrid run where extract and load happen in different processes — stages through blob. Both are the same runner.Copy; only the sink (real destination vs. a blob sink) and the number of passes change. The blob seam is covered in its own page.

Where to go next

The record model is the currency every connector and operator speaks; the blob staging layer is the seam that lets a run split across two processes. Both are covered next.