Architecture

Setu answers one hard question: how do you move data between systems when the cloud often can’t reach one side of the transfer, and the credentials for that side must never leave the customer’s network? The answer is a strict control-plane / data-plane split plus a hybrid edge agent. The cloud decides what should happen and records what did; the actual extract → transform → load runs wherever the data lives.

Read this page as the map. Every claim here is grounded in the router (apiserver/internal/api/api.go), the placement function (placement.go), and the schema migrations under apiserver/internal/db/migration/. The concepts page then walks the same machinery from an operator’s point of view.

The three planes

Everything in Setu belongs to exactly one of three planes. The boundary between them is the whole design. State the invariant plainly: the control plane never opens a connector and never moves a row; the data plane never makes a product decision. That separation is what lets one half live in Setu’s cloud and the other half live inside a customer’s VPC.

The Go monorepo

Setu is a single Go workspace. go.work stitches five modules together so the apiserver, worker, and agent can all import the shared engine without a published package. The dependency arrows all point inward toward engine:

go 1.26.4

use (
	./agent
	./apiserver
	./engine
	./packs
	./worker
)

Containers & the blob seam

Here is the whole system as running processes. Note the two data-plane hosts flanking a shared object store: the blob is the seam that lets extract and load run on different sides of a trust boundary.

The apiserver never connects out to the agent — it can’t, the agent has no inbound ports. Instead the agent pulls work by long-polling /agent/v1/jobs/next. Both the worker and the agent run the engine and both read/write the same blob store under an identical key layout, so an extract staged by an agent can be loaded by the cloud worker (or vice-versa):

// worker/internal/migrate/staged.go and apiserver/internal/api/agent_jobs.go
// both format the run's staging prefix identically, so whichever host stages the
// extract, the other can read it back.
func runPrefix(workspaceID, pipelineID, runID uuid.UUID) string {
	return fmt.Sprintf("ws/%s/pl/%s/run/%s/", workspaceID, pipelineID, runID)
}

The router: every subsystem at a glance

The apiserver’s Router() is the single source of truth for what the control plane does. A short middleware chain wraps everything, then four trust zones mount side by side:

r := chi.NewRouter()
r.Use(middleware.RequestID)                 // correlation id per request
r.Use(middleware.Recoverer)                 // a panic becomes a 500, not a crash
r.Use(corsMiddleware(corsAllowlist()))      // console origin allowlist

r.Get("/healthz", ...)                      // open: liveness
r.Get("/readyz", s.readyz)                  // open: readiness (pings the pool)
r.Handle("/metrics", metrics.Handler())     // open: Prometheus scrape
r.Post("/internal/scheduler/tick", ...)     // machine cron: its own bearer token

r.Route("/api/v1", ...)                     // USER trust boundary (session cookie)
r.Route("/agent/v1", ...)                   // AGENT trust boundary (agent key)

Inside /api/v1 (behind authMiddleware) live the whole console surface: auth & profile, workspaces / members / invites, connectors and connections, operators, pipelines, schedules, datasets & folders, mapping suggestions, search, overview, notifications, runs (with SSE event streams and artifact download), secrets, and agents. Inside /agent/v1 lives the edge protocol only. The subsystems, grouped:

The data model

Every row is scoped to a workspace (migration 0013 stamped a workspace_id onto every top-level table and closed a systemic IDOR). Users reach a workspace through a membership. A connection carries a placement (cloud or agent) and, when agent-placed, an agent_id — this single pair decides where each stage of a run executes. A pipeline wires a source connection to a target and stores a canonical spec Document; a run is one execution with a status, a phase, and a dry_run flag; a bound agent_job carries edge work by kind.

Fields that carry the design

• connection.placement + connection.agent_id — the routing decision (migration 0024). Cloud connections run their stage in the worker; agent connections pin their stage to that agent. The FK is ON DELETE SET NULL, so revoking an agent never cascade-drops the connection.
• connection.config — JSONB persisted verbatim. Secret references (${secret.NAME}) are kept as-is; plaintext credentials are never stored. The provider column was renamed from kind in migration 0006 for the dataset-centric model.
• run.status — a real Postgres ENUM (pending → running → succeeded | failed | cancelled). run.phase — a text column with a CHECK, widened over time: 0019 added extract|load, 0022 added direct.
• run_event — an append-only, identity-keyed live log streamed to the console over SSE. run_artifact — one row per staged run pointing at the blob prefix, with an expires_at the GC sweeps.
• extract_checkpoint / load_checkpoint — per-(run, stream) resume cursors (0020 / 0019). A retried run reads these to skip already-staged parts and already-loaded rows.
• pipeline_state — the durable per-(pipeline, stream) incremental watermark (0002), advanced only after a load commits — distinct from the per-run extract_checkpoint.
• agent.key_hash — only the SHA-256 of the agent key is stored; the plaintext key is shown once at enrollment. agent_job.payload carries IDs only, never secrets.

Request trust boundaries

The router mounts four distinct trust zones, each with its own authentication. A request never crosses from one into another:

r.Get("/healthz", ...)                 // open: liveness
r.Get("/readyz", s.readyz)             // open: readiness (pings the pool)
r.Handle("/metrics", metrics.Handler())// open: Prometheus scrape

// Machine cron → force a scheduling pass. Its own bearer token, not a session.
r.Post("/internal/scheduler/tick", s.schedulerTick)

r.Route("/api/v1", func(r chi.Router) {   // USER trust boundary
    r.Post("/auth/login", s.login)        // unauthenticated: mints a session
    r.Group(func(r chi.Router) {
        r.Use(s.authMiddleware)           // everything else needs a live session
        // ...connections, pipelines, runs, datasets, agents...
    })
})

r.Route("/agent/v1", func(r chi.Router) { // AGENT trust boundary
    r.Post("/enroll", s.enroll)           // public: exchange a token for a key
    r.Group(func(r chi.Router) {
        r.Use(s.agentAuthMiddleware)      // else: authenticate by agent key
        r.Get("/jobs/next", s.agentJobNext)
        r.Post("/jobs/{id}/result", s.agentJobResult)
    })
})

func (s *Server) agentAuthMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		key := bearerToken(r)                       // Authorization: Bearer <key>
		if key == "" { s.unauthenticated(w); return }
		hash := auth.HashToken(key)                 // only the hash is ever stored
		agent, err := s.q.GetAgentByKeyHash(r.Context(), &hash)
		if err != nil { s.unauthenticated(w); return }
		// Liveness is advisory: never fail the request if the touch write errors.
		_ = s.q.TouchAgentSeen(r.Context(), sqlc.TouchAgentSeenParams{ID: agent.ID, Version: agent.Version})
		ctx := context.WithValue(r.Context(), ctxKeyAgentID, agent.ID)
		ctx = context.WithValue(ctx, ctxKeyWorkspaceID, agent.WorkspaceID)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}

Request lifecycle: a run, end to end

Tie it together with the path a single run takes. The operator clicks Run; from there the control plane decides placement and either enqueues a cloud job or dispatches an agent job — but it never moves a byte itself.

1. 1
   Create the run. runPipeline loads the pipeline in the caller’s workspace and calls CreateRun — a pending row with phase = extract and dry_run from the query string.
2. 2
   Derive placement. orchestrateRun loads the source and destination connections and calls the pure derivePlacement(src.placement, src.agent, dst.placement, dst.agent), yielding where extract and load each run.
3. 3
   Reachability gate. The stage that runs first on an agent (extract if the source is agent-placed, else load) is checked against agentReachable (active + seen within 90s). If it’s offline, the run is marked failed with an actionable message instead of hanging on an un-leased job.
4. 4
   Dispatch. If extract is agent-placed, CreateAgentJob(kind=extract) enqueues an outbound job (IDs only) and emits a run event. Otherwise enq.EnqueueMigrate(runID) puts a River job on the queue for the cloud worker.
5. 5
   Execute. The chosen host runs the engine: extract → (stage to blob) → load, or a direct single pass. Progress lands as run_event rows and live counters on the run.
6. 6
   Bridge the seam (if crossing). When extract ran on an agent, onAgentExtractResult records the artifact, flips phase to load, and dispatches the load to the cloud worker or the destination’s agent — per the destination placement.
7. 7
   Observe. The console tails GET /runs/{id}/events as Server-Sent Events until the run reaches a terminal status.

The edge job protocol

Because the agent has no inbound ports, all edge work flows through an outbound queue the agent long-polls. The lease protocol is a small durable state machine in Postgres — no message broker required.

-- Atomically claim the oldest pending job for this agent. FOR UPDATE SKIP LOCKED
-- lets concurrent leasers each grab a distinct row without blocking, so the same
-- job is never handed out twice.
UPDATE agent_job SET
  status = 'leased',
  attempt = attempt + 1,
  lease_expires_at = now() + make_interval(secs => sqlc.arg('lease_seconds')::int),
  updated_at = now()
WHERE id = (
  SELECT j.id FROM agent_job j
  WHERE j.agent_id = sqlc.arg('agent_id') AND j.status = 'pending'
  ORDER BY j.created_at
  FOR UPDATE SKIP LOCKED
  LIMIT 1
)
RETURNING *;

With the planes, the data model, and the lifecycle in place, the core concepts page walks through what those tables mean to an operator: connections and placement, datasets, pipelines, runs, and the staged-vs-direct execution decision.