Core concepts

// stagePlacement says where one stage of a run executes: the cloud worker
// (Cloud=true) or a specific bound agent (Cloud=false, AgentID set).
type stagePlacement struct {
	Cloud   bool
	AgentID uuid.UUID
}

// derivePlacement decides where extract and load run from the two connection
// placements. It is a pure function — no I/O, trivially unit-tested.
func derivePlacement(srcPlacement string, srcAgent uuid.UUID, dstPlacement string, dstAgent uuid.UUID) runPlacement {
	return runPlacement{
		Extract: stageFor(srcPlacement, srcAgent), // reads the source
		Load:    stageFor(dstPlacement, dstAgent), // writes the destination
	}
}

func stageFor(placement string, agent uuid.UUID) stagePlacement {
	if placement == "agent" {
		return stagePlacement{Cloud: false, AgentID: agent}
	}
	return stagePlacement{Cloud: true}
}

{
  "host": "10.0.4.12",
  "port": 5432,
  "database": "orders",
  "user": "setu_ro",
  "password": "${secret.ORDERS_DB_PASSWORD}",
  "sslmode": "require"
}

for _, d := range discovered {
	row, err := q.UpsertDataset(ctx, sqlc.UpsertDatasetParams{
		ConnectionID: conn.ID, Provider: conn.Provider,
		PhysicalName: d.PhysicalName, Type: d.Type,
		Schema: d.Schema, RowEstimate: d.RowEstimate,
		PrimaryKey: pk, WorkspaceID: workspaceID, // dataset inherits its connection's workspace
	})
	// ...
}
// Empty-discovery safety: never mass-flag the whole catalog missing on a
// transient empty discovery. Skip the sweep and report Missing=0.
if len(discovered) == 0 {
	return Summary{Total: len(rows)}, rows, nil
}
missing, err := q.MarkDatasetsMissing(ctx, sqlc.MarkDatasetsMissingParams{
	ConnectionID: conn.ID, Present: PresentNames(discovered),
})

// Document is the JSON-serialised pipeline body stored in pipeline.spec.
type Document struct {
	Version int            `json:"version"`
	Params  map[string]any `json:"params,omitempty"`
	Vars    map[string]any `json:"vars,omitempty"`
	Nodes   []Node         `json:"nodes"`
	Edges   []Edge         `json:"edges,omitempty"`
}

// SyncConfig describes how a source reads across runs: a full re-read (default)
// or an incremental read bounded by a cursor column.
type SyncConfig struct {
	Mode           string            `json:"mode,omitempty"` // "full" | "incremental"
	CursorField    string            `json:"cursorField,omitempty"`
	CursorByStream map[string]string `json:"cursorByStream,omitempty"` // per-stream override
}

type DestinationConfig struct {
	ConnectionID string `json:"connectionId"`
	Mode         string `json:"mode,omitempty"`         // append|overwrite|upsert|merge
	SchemaPolicy string `json:"schemaPolicy,omitempty"` // create|strict
	TargetPrefix string `json:"targetPrefix,omitempty"` // e.g. "staging_"
}

{
  "version": 1,
  "nodes": [
    { "id": "source", "type": "source",
      "config": { "connectionId": "…", "streams": ["public.orders"],
                  "sync": { "mode": "incremental", "cursorField": "updated_at" } } },
    { "id": "destination", "type": "destination",
      "config": { "connectionId": "…", "mode": "upsert", "schemaPolicy": "create" } }
  ],
  "edges": [ { "from": "source", "to": "destination" } ]
}

CREATE TYPE run_status AS ENUM ('pending','running','succeeded','failed','cancelled');

CREATE TABLE run (
    id           uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    pipeline_id  uuid NOT NULL REFERENCES pipeline(id) ON DELETE CASCADE,
    status       run_status NOT NULL DEFAULT 'pending',
    phase        text NOT NULL DEFAULT 'extract'         -- 0019, widened by 0022
                 CHECK (phase IN ('extract','load','direct')),
    rows_read        bigint NOT NULL DEFAULT 0,
    rows_written     bigint NOT NULL DEFAULT 0,
    rows_quarantined bigint NOT NULL DEFAULT 0,           -- 0002
    rows_total       bigint NOT NULL DEFAULT 0,           -- 0020: progress denominator
    dry_run          boolean NOT NULL DEFAULT false,      -- 0002
    cancel_requested boolean NOT NULL DEFAULT false,      -- 0011: cooperative cancel
    error        text,
    started_at   timestamptz,
    finished_at  timestamptz
);

// directMaxRows: at/below this incremental delta, stream DIRECTLY (no blob).
// Default 50000 (~one blob part). Env: SETU_DIRECT_MAX_ROWS (0 disables direct).
func directMaxRows() int64 { /* env override, else 50000 */ }

// Only a fresh (still-extracting) run whose load is NOT pinned to an agent can go
// direct; a run resumed at the load phase must read from the staged blob.
if run.Phase == "extract" && !loadOnAgent {
	forceDirect := !stagedEnabled()          // SETU_STAGED_EXECUTION=off kill switch
	autoDirect := false
	if res.Incremental {
		plans, todo, total, sized, err := w.directDelta(ctx, src, res, streams, run.PipelineID, ecs)
		// ...
		autoDirect = sized && total <= directMaxRows() // small delta → direct
	}
	if forceDirect || autoDirect {
		return w.runDirect(runCtx, ctx, q, runID, src, dst, todo, res, run.DryRun, ...)
	}
}
// Otherwise: extract the source into blob, set phase=load, then load from blob.

if run.Phase == "extract" {
	// ...runExtract stages to blob...
	q.SetRunPhase(ctx, sqlc.SetRunPhaseParams{ID: runID, Phase: "load"})
	if loadOnAgent {
		q.CreateAgentJob(ctx, sqlc.CreateAgentJobParams{
			AgentID: uuid.UUID(dstConn.AgentID.Bytes), WorkspaceID: workspaceID,
			Kind: "load", RunID: pgtype.UUID{Bytes: runID, Valid: true},
			ConnectionID: pgtype.UUID{Bytes: dstID, Valid: true}, Payload: payload,
		})
		return runner.Stats{}, errHandedOffToAgent // run continues on the edge
	}
}

// Secrets is the edge secret store: name→value. Connection configs pulled
// from the control plane carry ${secret.NAME} references (never plaintext);
// the agent resolves them from this map at extract time. Populated via
// `setu secret set <NAME> <VALUE>` and written 0600 under the user's config dir.
Secrets map[string]string `json:"secrets,omitempty"`

// Config is returned VERBATIM (secret refs intact). Never resolve here.
cfg := json.RawMessage(conn.Config)
writeJSON(w, http.StatusOK, extractSpecDTO{
	RunID:      runID,
	BlobPrefix: runBlobPrefix(ws, run.PipelineID, runID),
	Provider:   conn.Provider,
	Config:     cfg,               // {secret.X{'}'} still unresolved
	Streams:    streams,
	Blob:       engineblob.SpecFromEnv(), // same backend the cloud reads back from
})

CREATE TABLE pipeline_schedule (
    pipeline_id  uuid PRIMARY KEY REFERENCES pipeline(id) ON DELETE CASCADE,
    cron         text NOT NULL,                 -- 5-field cron expression
    enabled      boolean NOT NULL DEFAULT true,
    dry_run      boolean NOT NULL DEFAULT false,
    next_run_at  timestamptz,                   -- NULL when disabled
    last_run_at  timestamptz
);
CREATE INDEX pipeline_schedule_due_idx
    ON pipeline_schedule (next_run_at)
    WHERE enabled AND next_run_at IS NOT NULL;