Saved connections vault (encrypted DSNs)

cmd/marathon/main.go

@@ -6,6 +6,7 @@ import (
 	"gofr.dev/pkg/gofr"
 
 	"marathon/internal/auth"
+	connhandler "marathon/internal/handler/connection"
 	runhandler "marathon/internal/handler/run"
 	taskhandler "marathon/internal/handler/task"
 	"marathon/internal/service/orchestrator"
@@ -34,6 +35,7 @@ func main() {
 	tasks := taskhandler.New()
 	runs := runhandler.New(orch)
 	live := runhandler.NewLive(hub, runs)
+	conns := connhandler.New()
 
 	// Tasks
 	app.POST("/tasks", tasks.Create)
@@ -53,6 +55,13 @@ func main() {
 	app.GET("/runs/{id}/quarantine", runs.ListQuarantine)
 	app.POST("/runs/{id}/quarantine/retry", runs.RetryQuarantine)
 
+	// Saved connections
+	app.POST("/connections", conns.Create)
+	app.GET("/connections", conns.List)
+	app.DELETE("/connections/{id}", conns.Delete)
+	app.POST("/connections/{id}/test", conns.Test)
+	app.POST("/connections/test", conns.TestDraft)
+
 	// Orphan sweeper: every 15s, resume 'running' runs that have no live
 	// runner in this process (i.e. we crashed and restarted mid-run).
 	app.AddCronJob("*/15 * * * * *", "orphan-sweeper", orch.Sweep)

cmd/server/main.go

@@ -7,6 +7,7 @@ import (
 
 	"marathon/internal/auth"
 	"marathon/internal/coord"
+	connhandler "marathon/internal/handler/connection"
 	runhandler "marathon/internal/handler/run"
 	taskhandler "marathon/internal/handler/task"
 	"marathon/internal/service/orchestrator"
@@ -33,6 +34,7 @@ func main() {
 	tasks := taskhandler.New()
 	runs := runhandler.New(orch)
 	live := runhandler.NewLive(hub, runs)
+	conns := connhandler.New()
 
 	app.POST("/tasks", tasks.Create)
 	app.GET("/tasks", tasks.List)
@@ -49,6 +51,12 @@ func main() {
 	app.GET("/runs/{id}/quarantine", runs.ListQuarantine)
 	app.POST("/runs/{id}/quarantine/retry", runs.RetryQuarantine)
 
+	app.POST("/connections", conns.Create)
+	app.GET("/connections", conns.List)
+	app.DELETE("/connections/{id}", conns.Delete)
+	app.POST("/connections/{id}/test", conns.Test)
+	app.POST("/connections/test", conns.TestDraft)
+
 	// Fleet monitor: reap dead-worker leases and finalize drained runs.
 	app.AddCronJob("*/5 * * * * *", "fleet-monitor", orch.MonitorFleet)
 

internal/engine/connection_test.go

@@ -0,0 +1,79 @@
+package engine
+
+import (
+	"context"
+	"testing"
+
+	"marathon/internal/models"
+	"marathon/internal/store"
+)
+
+// Proves the saved-connection path: a task carries no DSN, only a connection_id;
+// the DSN is stored encrypted, resolved at run time, and the backfill runs.
+func TestIntegration_SavedConnection(t *testing.T) {
+	control, target := openOrSkip(t)
+	controlSchema(t, control)
+
+	const n = 5000
+	seedTarget(t, target, "items", n, "")
+
+	targetDSN := dsn("MARATHON_TEST_TARGET_DSN", "postgres://demo:demo@localhost:5434/demo?sslmode=disable")
+
+	conns := store.ConnectionStore{}
+	conn, err := conns.Create(context.Background(), control, "conn-test", "postgres", targetDSN)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// The task references the connection and carries no DSN of its own.
+	task := &models.Task{
+		Name: "via-connection", ConnectionID: &conn.ID,
+		SourceTable: "items", CursorColumn: "id", BatchSize: 1000,
+		OperationSQL: "UPDATE items SET applied = applied + 1 WHERE id >= $1 AND id <= $2",
+	}
+	newTask(t, control, task)
+
+	if task.TargetDSN != "" {
+		t.Fatalf("task should not store a DSN when using a connection, got %q", task.TargetDSN)
+	}
+
+	// Resolve fills the DSN/driver from the (decrypted) connection.
+	if err := conns.ResolveInto(context.Background(), control, task); err != nil {
+		t.Fatal(err)
+	}
+
+	if task.TargetDSN != targetDSN || task.TargetDriver != "postgres" {
+		t.Fatalf("resolve produced dsn=%q driver=%q", task.TargetDSN, task.TargetDriver)
+	}
+
+	run, err := (store.RunStore{}).Create(context.Background(), control, task.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	NewRunner(control, nopLogger{}, *task, run.ID).Run(context.Background())
+
+	assertState(t, control, run.ID, models.RunSucceeded)
+	assertCount(t, target, "SELECT count(*) FROM items WHERE applied <> 1", 0, "rows not applied exactly once")
+
+	// A connection in use by a task can't be deleted.
+	if err := conns.Delete(context.Background(), control, conn.ID); err == nil {
+		t.Error("deleting an in-use connection should be refused")
+	}
+
+	// An unused connection deletes fine.
+	spare, err := conns.Create(context.Background(), control, "spare", "postgres", targetDSN)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := conns.Delete(context.Background(), control, spare.ID); err != nil {
+		t.Fatalf("deleting an unused connection should work: %v", err)
+	}
+
+	// Resolving a non-existent connection errors.
+	missing := int64(999999)
+	if err := conns.ResolveInto(context.Background(), control, &models.Task{ConnectionID: &missing}); err == nil {
+		t.Error("resolving a missing connection should error")
+	}
+}

internal/engine/integration_test.go

@@ -88,16 +88,23 @@ func controlSchema(t *testing.T, db *sql.DB) {
 
 	// Idempotently reconcile columns added by later migrations, so the test is
 	// robust whether the control DB is fresh or predates them.
+	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS connections (
+		id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL UNIQUE, driver TEXT NOT NULL,
+		encrypted_dsn TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())`); err != nil {
+		t.Fatalf("control connections ddl: %v", err)
+	}
+
 	if _, err := db.Exec(`ALTER TABLE tasks
 		ADD COLUMN IF NOT EXISTS operation_type   TEXT NOT NULL DEFAULT 'sql',
 		ADD COLUMN IF NOT EXISTS operation_url    TEXT NOT NULL DEFAULT '',
 		ADD COLUMN IF NOT EXISTS target_driver    TEXT NOT NULL DEFAULT 'postgres',
-		ADD COLUMN IF NOT EXISTS schedule_seconds INT  NOT NULL DEFAULT 0`); err != nil {
+		ADD COLUMN IF NOT EXISTS schedule_seconds INT  NOT NULL DEFAULT 0,
+		ADD COLUMN IF NOT EXISTS connection_id    BIGINT REFERENCES connections(id)`); err != nil {
 		t.Fatalf("control alter: %v", err)
 	}
 
 	// Clean slate. TRUNCATE ... CASCADE resets identities too.
-	if _, err := db.Exec(`TRUNCATE run_events, quarantined_rows, checkpoints, runs, tasks RESTART IDENTITY CASCADE`); err != nil {
+	if _, err := db.Exec(`TRUNCATE run_events, quarantined_rows, checkpoints, runs, tasks, connections RESTART IDENTITY CASCADE`); err != nil {
 		t.Fatalf("truncate control: %v", err)
 	}
 }

internal/fleet/worker.go

@@ -87,6 +87,11 @@ func (m *Manager) work(ctx context.Context, run models.Run) {
 		return
 	}
 
+	if err := (store.ConnectionStore{}).ResolveInto(ctx, m.db, &task); err != nil {
+		m.log.Errorf("fleet worker %s: resolve connection for run %d: %v", m.id, run.ID, err)
+		return
+	}
+
 	queue := coord.NewQueue(m.rdb, run.ID)
 
 	w := engine.NewLeaseWorker(m.id, m.db, m.log, queue, task, run.ID)

internal/handler/connection/handler.go

@@ -0,0 +1,112 @@
+// Package connection exposes CRUD for saved database connections plus a
+// "test" endpoint that verifies a DSN actually connects. DSNs are encrypted at
+// rest and never returned.
+package connection
+
+import (
+	"context"
+	"database/sql"
+	"strconv"
+	"time"
+
+	"gofr.dev/pkg/gofr"
+	gofrhttp "gofr.dev/pkg/gofr/http"
+
+	"marathon/internal/engine/batch"
+	"marathon/internal/store"
+)
+
+type Handler struct {
+	conns store.ConnectionStore
+}
+
+func New() *Handler {
+	return &Handler{}
+}
+
+type createRequest struct {
+	Name   string `json:"name"`
+	Driver string `json:"driver"`
+	DSN    string `json:"dsn"`
+}
+
+func (r createRequest) valid() bool {
+	return r.Name != "" && (r.Driver == "postgres" || r.Driver == "mysql") && r.DSN != ""
+}
+
+// Create handles POST /connections.
+func (h *Handler) Create(ctx *gofr.Context) (any, error) {
+	var req createRequest
+	if err := ctx.Bind(&req); err != nil || !req.valid() {
+		return nil, gofrhttp.ErrorInvalidParam{Params: []string{"name, driver (postgres|mysql), dsn"}}
+	}
+
+	return h.conns.Create(ctx, ctx.SQL, req.Name, req.Driver, req.DSN)
+}
+
+// List handles GET /connections.
+func (h *Handler) List(ctx *gofr.Context) (any, error) {
+	return h.conns.List(ctx, ctx.SQL)
+}
+
+// Delete handles DELETE /connections/{id}.
+func (h *Handler) Delete(ctx *gofr.Context) (any, error) {
+	id, err := strconv.ParseInt(ctx.PathParam("id"), 10, 64)
+	if err != nil {
+		return nil, gofrhttp.ErrorInvalidParam{Params: []string{"id"}}
+	}
+
+	if err := h.conns.Delete(ctx, ctx.SQL, id); err != nil {
+		return nil, err
+	}
+
+	return map[string]bool{"deleted": true}, nil
+}
+
+// Test handles POST /connections/{id}/test — opens the connection and pings it.
+func (h *Handler) Test(ctx *gofr.Context) (any, error) {
+	id, err := strconv.ParseInt(ctx.PathParam("id"), 10, 64)
+	if err != nil {
+		return nil, gofrhttp.ErrorInvalidParam{Params: []string{"id"}}
+	}
+
+	driver, dsn, err := h.conns.Resolve(ctx, ctx.SQL, id)
+	if err != nil {
+		return nil, err
+	}
+
+	return probe(batch.DialectFor(driver).DriverName(), dsn), nil
+}
+
+// TestDraft handles POST /connections/test — checks a DSN before saving it.
+func (h *Handler) TestDraft(ctx *gofr.Context) (any, error) {
+	var req createRequest
+	if err := ctx.Bind(&req); err != nil || req.DSN == "" {
+		return nil, gofrhttp.ErrorInvalidParam{Params: []string{"driver, dsn"}}
+	}
+
+	return probe(batch.DialectFor(req.Driver).DriverName(), req.DSN), nil
+}
+
+type probeResult struct {
+	OK    bool   `json:"ok"`
+	Error string `json:"error,omitempty"`
+}
+
+// probe opens the DSN and pings it with a short timeout. Never leaks the DSN.
+func probe(driverName, dsn string) probeResult {
+	db, err := sql.Open(driverName, dsn)
+	if err != nil {
+		return probeResult{Error: err.Error()}
+	}
+	defer db.Close()
+
+	pctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+	defer cancel()
+
+	if err := db.PingContext(pctx); err != nil {
+		return probeResult{Error: err.Error()}
+	}
+
+	return probeResult{OK: true}
+}

internal/handler/task/handler.go

@@ -64,5 +64,9 @@ func (h *Handler) DryRun(ctx *gofr.Context) (any, error) {
 		return nil, err
 	}
 
+	if err := (store.ConnectionStore{}).ResolveInto(ctx, ctx.SQL, &task); err != nil {
+		return nil, err
+	}
+
 	return dryrun.Analyze(ctx, task)
 }

internal/models/connection.go

@@ -0,0 +1,12 @@
+package models
+
+import "time"
+
+// Connection is a saved, named database connection. The DSN is encrypted at
+// rest and never returned by the API — only the metadata is exposed.
+type Connection struct {
+	ID        int64     `json:"id"`
+	Name      string    `json:"name"`
+	Driver    string    `json:"driver"` // "postgres" | "mysql"
+	CreatedAt time.Time `json:"created_at"`
+}

internal/models/task.go

@@ -24,6 +24,7 @@ const (
 type Task struct {
 	ID            int64     `json:"id"`
 	Name          string    `json:"name"`
+	ConnectionID  *int64    `json:"connection_id"` // when set, driver+DSN come from a saved connection
 	TargetDriver  string    `json:"target_driver"` // "postgres" (default) or "mysql"
 	TargetDSN     string    `json:"target_dsn"`
 	SourceTable   string    `json:"source_table"`
@@ -64,7 +65,12 @@ func (t *Task) Validate() error {
 		return errNameRequired
 	}
 
-	if strings.TrimSpace(t.TargetDSN) == "" {
+	usesConnection := t.ConnectionID != nil && *t.ConnectionID > 0
+
+	// With a saved connection the DSN is supplied at run time; without one it's
+	// required up front. The driver is always known (set from the connection by
+	// the handler, or provided inline) so operation validation can proceed.
+	if !usesConnection && strings.TrimSpace(t.TargetDSN) == "" {
 		return errDSNRequired
 	}