diff --git a/cmd/marathon/main.go b/cmd/marathon/main.go index cd8a33b..d290fb3 100644 --- a/cmd/marathon/main.go +++ b/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) diff --git a/cmd/server/main.go b/cmd/server/main.go index ed61035..01b37cc 100644 --- a/cmd/server/main.go +++ b/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) diff --git a/docs/screenshots/connections.png b/docs/screenshots/connections.png new file mode 100644 index 0000000..68ceb35 Binary files /dev/null and b/docs/screenshots/connections.png differ diff --git a/docs/screenshots/wizard-connection.png b/docs/screenshots/wizard-connection.png new file mode 100644 index 0000000..1c26201 Binary files /dev/null and b/docs/screenshots/wizard-connection.png differ diff --git a/internal/engine/connection_test.go b/internal/engine/connection_test.go new file mode 100644 index 0000000..64956db --- /dev/null +++ b/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") + } +} diff --git a/internal/engine/integration_test.go b/internal/engine/integration_test.go index 12cfc25..bd5e772 100644 --- a/internal/engine/integration_test.go +++ b/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) } } diff --git a/internal/fleet/worker.go b/internal/fleet/worker.go index 63c0341..ed37dee 100644 --- a/internal/fleet/worker.go +++ b/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) diff --git a/internal/handler/connection/handler.go b/internal/handler/connection/handler.go new file mode 100644 index 0000000..07e9efa --- /dev/null +++ b/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} +} diff --git a/internal/handler/task/handler.go b/internal/handler/task/handler.go index 783c052..29a0560 100644 --- a/internal/handler/task/handler.go +++ b/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) } diff --git a/internal/models/connection.go b/internal/models/connection.go new file mode 100644 index 0000000..ec823a4 --- /dev/null +++ b/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"` +} diff --git a/internal/models/task.go b/internal/models/task.go index c363df1..043a854 100644 --- a/internal/models/task.go +++ b/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 } diff --git a/internal/secret/secret.go b/internal/secret/secret.go new file mode 100644 index 0000000..2f4dea2 --- /dev/null +++ b/internal/secret/secret.go @@ -0,0 +1,100 @@ +// Package secret encrypts sensitive values (database connection strings) before +// they are written to the control store. It uses AES-256-GCM with a key derived +// from a passphrase, so a leaked database dump does not leak DSNs. +package secret + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "errors" + "fmt" + "io" + "os" + "sync" +) + +// devPassphrase is used when MARATHON_SECRET is unset — fine for local/dev, but +// production deployments must set MARATHON_SECRET to protect stored DSNs. +const devPassphrase = "marathon-insecure-dev-secret" + +var errCiphertext = errors.New("malformed ciphertext") + +// Box encrypts and decrypts strings with a fixed key. +type Box struct { + aead cipher.AEAD +} + +// NewBox derives a 256-bit key from the passphrase (SHA-256) and returns a Box. +func NewBox(passphrase string) (*Box, error) { + key := sha256.Sum256([]byte(passphrase)) + + block, err := aes.NewCipher(key[:]) + if err != nil { + return nil, fmt.Errorf("new cipher: %w", err) + } + + aead, err := cipher.NewGCM(block) + if err != nil { + return nil, fmt.Errorf("new gcm: %w", err) + } + + return &Box{aead: aead}, nil +} + +// Encrypt returns a base64 token of nonce||ciphertext. +func (b *Box) Encrypt(plaintext string) (string, error) { + nonce := make([]byte, b.aead.NonceSize()) + if _, err := io.ReadFull(rand.Reader, nonce); err != nil { + return "", fmt.Errorf("read nonce: %w", err) + } + + sealed := b.aead.Seal(nonce, nonce, []byte(plaintext), nil) + + return base64.StdEncoding.EncodeToString(sealed), nil +} + +// Decrypt reverses Encrypt. It fails if the token was tampered with. +func (b *Box) Decrypt(token string) (string, error) { + raw, err := base64.StdEncoding.DecodeString(token) + if err != nil { + return "", fmt.Errorf("%w: %v", errCiphertext, err) + } + + ns := b.aead.NonceSize() + if len(raw) < ns { + return "", errCiphertext + } + + nonce, body := raw[:ns], raw[ns:] + + plain, err := b.aead.Open(nil, nonce, body, nil) + if err != nil { + return "", fmt.Errorf("%w: %v", errCiphertext, err) + } + + return string(plain), nil +} + +var ( + defaultBox *Box + defaultOnce sync.Once +) + +// Default returns the process Box, built once from MARATHON_SECRET (or the dev +// passphrase if unset). +func Default() *Box { + defaultOnce.Do(func() { + pass := os.Getenv("MARATHON_SECRET") + if pass == "" { + pass = devPassphrase + } + + // A SHA-256 key never makes NewBox fail, so the error is unreachable. + defaultBox, _ = NewBox(pass) + }) + + return defaultBox +} diff --git a/internal/secret/secret_test.go b/internal/secret/secret_test.go new file mode 100644 index 0000000..d38922b --- /dev/null +++ b/internal/secret/secret_test.go @@ -0,0 +1,63 @@ +package secret + +import "testing" + +func TestRoundTrip(t *testing.T) { + box, err := NewBox("a-passphrase") + if err != nil { + t.Fatal(err) + } + + for _, plain := range []string{"", "postgres://u:p@host:5432/db?sslmode=require", "unicode ✓ café"} { + token, err := box.Encrypt(plain) + if err != nil { + t.Fatalf("encrypt %q: %v", plain, err) + } + + if token == plain { + t.Errorf("ciphertext equals plaintext for %q", plain) + } + + got, err := box.Decrypt(token) + if err != nil { + t.Fatalf("decrypt: %v", err) + } + + if got != plain { + t.Errorf("round trip: got %q, want %q", got, plain) + } + } +} + +func TestEncryptIsNondeterministic(t *testing.T) { + box, _ := NewBox("k") + a, _ := box.Encrypt("same") + b, _ := box.Encrypt("same") + + if a == b { + t.Error("two encryptions of the same value should differ (random nonce)") + } +} + +func TestWrongKeyFails(t *testing.T) { + a, _ := NewBox("key-a") + b, _ := NewBox("key-b") + + token, _ := a.Encrypt("secret") + if _, err := b.Decrypt(token); err == nil { + t.Error("decrypt with the wrong key should fail") + } +} + +func TestTamperFails(t *testing.T) { + box, _ := NewBox("k") + token, _ := box.Encrypt("secret") + + // Flip a character in the middle of the token. + bad := []byte(token) + bad[len(bad)/2] ^= 0x01 + + if _, err := box.Decrypt(string(bad)); err == nil { + t.Error("decrypt of tampered ciphertext should fail") + } +} diff --git a/internal/service/orchestrator/service.go b/internal/service/orchestrator/service.go index 2501e83..7f6c593 100644 --- a/internal/service/orchestrator/service.go +++ b/internal/service/orchestrator/service.go @@ -86,6 +86,10 @@ func (o *Orchestrator) StartRun(ctx *gofr.Context, taskID int64) (models.Run, er return models.Run{}, err } + if err := (store.ConnectionStore{}).ResolveInto(ctx, ctx.SQL, &task); err != nil { + return models.Run{}, err + } + busy, err := o.runs.ActiveForTask(ctx, ctx.SQL, taskID) if err != nil { return models.Run{}, err @@ -312,6 +316,10 @@ func (o *Orchestrator) RetryQuarantine(ctx *gofr.Context, runID int64) (RetryRes return RetryResult{}, err } + if err := (store.ConnectionStore{}).ResolveInto(ctx, ctx.SQL, &task); err != nil { + return RetryResult{}, err + } + keys, err := o.quarantine.PendingKeys(ctx, ctx.SQL, runID) if err != nil { return RetryResult{}, err diff --git a/internal/store/connection.go b/internal/store/connection.go new file mode 100644 index 0000000..5353b6f --- /dev/null +++ b/internal/store/connection.go @@ -0,0 +1,129 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "marathon/internal/models" + "marathon/internal/secret" +) + +// ErrConnectionNotFound is returned when a connection id doesn't exist. +var ErrConnectionNotFound = errors.New("connection not found") + +// ErrConnectionInUse is returned when deleting a connection that a task uses. +var ErrConnectionInUse = errors.New("connection is in use by a task") + +type ConnectionStore struct{} + +// Create encrypts the DSN and stores the connection, returning its metadata. +func (ConnectionStore) Create(ctx context.Context, db DB, name, driver, dsn string) (models.Connection, error) { + enc, err := secret.Default().Encrypt(dsn) + if err != nil { + return models.Connection{}, fmt.Errorf("encrypt dsn: %w", err) + } + + var c models.Connection + + err = db.QueryRowContext(ctx, ` + INSERT INTO connections (name, driver, encrypted_dsn) + VALUES ($1, $2, $3) + RETURNING id, name, driver, created_at`, + name, driver, enc, + ).Scan(&c.ID, &c.Name, &c.Driver, &c.CreatedAt) + if err != nil { + return models.Connection{}, fmt.Errorf("create connection: %w", err) + } + + return c, nil +} + +// List returns all connections' metadata (never the DSN). +func (ConnectionStore) List(ctx context.Context, db DB) ([]models.Connection, error) { + rows, err := db.QueryContext(ctx, + `SELECT id, name, driver, created_at FROM connections ORDER BY name`) + if err != nil { + return nil, fmt.Errorf("list connections: %w", err) + } + defer rows.Close() + + out := []models.Connection{} + + for rows.Next() { + var c models.Connection + if err := rows.Scan(&c.ID, &c.Name, &c.Driver, &c.CreatedAt); err != nil { + return nil, fmt.Errorf("scan connection: %w", err) + } + + out = append(out, c) + } + + return out, rows.Err() +} + +// Resolve returns a connection's driver and decrypted DSN — for internal use +// (opening the target), never exposed over the API. +func (ConnectionStore) Resolve(ctx context.Context, db DB, id int64) (driver, dsn string, err error) { + var enc string + + err = db.QueryRowContext(ctx, + `SELECT driver, encrypted_dsn FROM connections WHERE id = $1`, id).Scan(&driver, &enc) + if errors.Is(err, sql.ErrNoRows) { + return "", "", fmt.Errorf("%w: %d", ErrConnectionNotFound, id) + } + + if err != nil { + return "", "", fmt.Errorf("resolve connection %d: %w", id, err) + } + + dsn, err = secret.Default().Decrypt(enc) + if err != nil { + return "", "", fmt.Errorf("decrypt connection %d dsn: %w", id, err) + } + + return driver, dsn, nil +} + +// Delete removes a connection, refusing if a task still references it. +func (ConnectionStore) Delete(ctx context.Context, db DB, id int64) error { + var used int + if err := db.QueryRowContext(ctx, + `SELECT count(*) FROM tasks WHERE connection_id = $1`, id).Scan(&used); err != nil { + return fmt.Errorf("check connection %d usage: %w", id, err) + } + + if used > 0 { + return fmt.Errorf("%w: %d task(s) reference it", ErrConnectionInUse, used) + } + + res, err := db.ExecContext(ctx, `DELETE FROM connections WHERE id = $1`, id) + if err != nil { + return fmt.Errorf("delete connection %d: %w", id, err) + } + + if n, _ := res.RowsAffected(); n == 0 { + return fmt.Errorf("%w: %d", ErrConnectionNotFound, id) + } + + return nil +} + +// ResolveInto fills a task's TargetDriver/TargetDSN from its saved connection, +// if it references one. No-op for tasks that carry an inline DSN. +func (s ConnectionStore) ResolveInto(ctx context.Context, db DB, task *models.Task) error { + if task.ConnectionID == nil || *task.ConnectionID == 0 { + return nil + } + + driver, dsn, err := s.Resolve(ctx, db, *task.ConnectionID) + if err != nil { + return err + } + + task.TargetDriver = driver + task.TargetDSN = dsn + + return nil +} diff --git a/internal/store/task.go b/internal/store/task.go index c5aad6e..5c8671d 100644 --- a/internal/store/task.go +++ b/internal/store/task.go @@ -14,11 +14,11 @@ type TaskStore struct{} func (TaskStore) Create(ctx context.Context, db DB, t *models.Task) error { err := db.QueryRowContext(ctx, ` - INSERT INTO tasks (name, target_driver, target_dsn, source_table, cursor_column, row_filter, + INSERT INTO tasks (name, connection_id, target_driver, target_dsn, source_table, cursor_column, row_filter, batch_size, operation_type, operation_sql, operation_url, rate_per_sec, schedule_seconds) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) RETURNING id, created_at`, - t.Name, t.TargetDriver, t.TargetDSN, t.SourceTable, t.CursorColumn, t.RowFilter, + t.Name, t.ConnectionID, t.TargetDriver, t.TargetDSN, t.SourceTable, t.CursorColumn, t.RowFilter, t.BatchSize, t.OperationType, t.OperationSQL, t.OperationURL, t.RatePerSec, t.ScheduleSeconds, ).Scan(&t.ID, &t.CreatedAt) if err != nil { @@ -42,10 +42,10 @@ func (TaskStore) Get(ctx context.Context, db DB, id int64) (models.Task, error) var t models.Task err := db.QueryRowContext(ctx, ` - SELECT id, name, target_driver, target_dsn, source_table, cursor_column, row_filter, + SELECT id, name, connection_id, target_driver, target_dsn, source_table, cursor_column, row_filter, batch_size, operation_type, operation_sql, operation_url, rate_per_sec, schedule_seconds, created_at FROM tasks WHERE id = $1`, id, - ).Scan(&t.ID, &t.Name, &t.TargetDriver, &t.TargetDSN, &t.SourceTable, &t.CursorColumn, &t.RowFilter, + ).Scan(&t.ID, &t.Name, &t.ConnectionID, &t.TargetDriver, &t.TargetDSN, &t.SourceTable, &t.CursorColumn, &t.RowFilter, &t.BatchSize, &t.OperationType, &t.OperationSQL, &t.OperationURL, &t.RatePerSec, &t.ScheduleSeconds, &t.CreatedAt) if err != nil { return models.Task{}, fmt.Errorf("get task %d: %w", id, err) @@ -56,7 +56,7 @@ func (TaskStore) Get(ctx context.Context, db DB, id int64) (models.Task, error) func (TaskStore) List(ctx context.Context, db DB) ([]models.Task, error) { rows, err := db.QueryContext(ctx, ` - SELECT id, name, target_driver, target_dsn, source_table, cursor_column, row_filter, + SELECT id, name, connection_id, target_driver, target_dsn, source_table, cursor_column, row_filter, batch_size, operation_type, operation_sql, operation_url, rate_per_sec, schedule_seconds, created_at FROM tasks ORDER BY id DESC`) if err != nil { @@ -69,7 +69,7 @@ func (TaskStore) List(ctx context.Context, db DB) ([]models.Task, error) { for rows.Next() { var t models.Task - if err := rows.Scan(&t.ID, &t.Name, &t.TargetDriver, &t.TargetDSN, &t.SourceTable, &t.CursorColumn, + if err := rows.Scan(&t.ID, &t.Name, &t.ConnectionID, &t.TargetDriver, &t.TargetDSN, &t.SourceTable, &t.CursorColumn, &t.RowFilter, &t.BatchSize, &t.OperationType, &t.OperationSQL, &t.OperationURL, &t.RatePerSec, &t.ScheduleSeconds, &t.CreatedAt); err != nil { return nil, fmt.Errorf("scan task: %w", err) diff --git a/migrations/010_task_connection.go b/migrations/010_task_connection.go new file mode 100644 index 0000000..8cfcfed --- /dev/null +++ b/migrations/010_task_connection.go @@ -0,0 +1,16 @@ +package migrations + +import "gofr.dev/pkg/gofr/migration" + +// Lets a task reference a saved connection instead of carrying an inline DSN. +func addTaskConnection() migration.Migrate { + return migration.Migrate{ + UP: func(d migration.Datasource) error { + _, err := d.SQL.Exec(` + ALTER TABLE tasks + ADD COLUMN IF NOT EXISTS connection_id BIGINT REFERENCES connections(id)`) + + return err + }, + } +} diff --git a/migrations/all.go b/migrations/all.go index fc54f80..ca02ad8 100644 --- a/migrations/all.go +++ b/migrations/all.go @@ -15,5 +15,6 @@ func All() map[int64]migration.Migrate { 7: addOperationType(), 8: addTargetDriver(), 9: addSchedule(), + 10: addTaskConnection(), } } diff --git a/web/src/App.tsx b/web/src/App.tsx index 014a663..96bac40 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -13,7 +13,11 @@ type View = type Theme = "light" | "dark"; function initialView(): View { - const r = new URLSearchParams(location.search).get("run"); + const p = new URLSearchParams(location.search); + const v = p.get("view"); + if (v === "setup") return { name: "setup" }; + if (v === "new") return { name: "new" }; + const r = p.get("run"); return r ? { name: "run", runId: Number(r) } : { name: "tasks" }; } diff --git a/web/src/api.ts b/web/src/api.ts index f6e3243..8bb0693 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -1,4 +1,5 @@ import type { + Connection, DryRunReport, QuarantinedRow, Run, @@ -51,6 +52,19 @@ export const api = { `/runs/${id}/quarantine/retry`, { method: "POST" } ), + + listConnections: () => call("/connections"), + createConnection: (c: { name: string; driver: string; dsn: string }) => + call("/connections", { method: "POST", body: JSON.stringify(c) }), + deleteConnection: (id: number) => + call<{ deleted: boolean }>(`/connections/${id}`, { method: "DELETE" }), + testConnection: (id: number) => + call<{ ok: boolean; error?: string }>(`/connections/${id}/test`, { method: "POST" }), + testDraftConnection: (c: { driver: string; dsn: string }) => + call<{ ok: boolean; error?: string }>("/connections/test", { + method: "POST", + body: JSON.stringify(c), + }), }; // liveSocket opens the progress websocket and performs the {run_id} handshake. diff --git a/web/src/pages/Setup.tsx b/web/src/pages/Setup.tsx index a78d9d5..f51beca 100644 --- a/web/src/pages/Setup.tsx +++ b/web/src/pages/Setup.tsx @@ -1,46 +1,140 @@ +import { useEffect, useState } from "react"; +import { api } from "../api"; +import type { Connection } from "../types"; +import { timeAgo } from "../components/widgets"; + +function Connections() { + const [conns, setConns] = useState([]); + const [name, setName] = useState(""); + const [driver, setDriver] = useState<"postgres" | "mysql">("postgres"); + const [dsn, setDsn] = useState(""); + const [msg, setMsg] = useState<{ kind: "ok" | "err"; text: string } | null>(null); + const [busy, setBusy] = useState(""); + + const refresh = () => api.listConnections().then(setConns).catch(() => {}); + useEffect(() => { refresh(); }, []); + + const test = async () => { + setBusy("test"); setMsg(null); + try { + const r = await api.testDraftConnection({ driver, dsn }); + setMsg(r.ok ? { kind: "ok", text: "Connected successfully." } : { kind: "err", text: r.error || "Connection failed." }); + } catch (e) { setMsg({ kind: "err", text: (e as Error).message }); } + setBusy(""); + }; + + const save = async () => { + setBusy("save"); setMsg(null); + try { + await api.createConnection({ name, driver, dsn }); + setName(""); setDsn(""); setMsg({ kind: "ok", text: "Connection saved." }); + refresh(); + } catch (e) { setMsg({ kind: "err", text: (e as Error).message }); } + setBusy(""); + }; + + const testSaved = async (id: number) => { + setMsg(null); + try { + const r = await api.testConnection(id); + setMsg(r.ok ? { kind: "ok", text: "Connected successfully." } : { kind: "err", text: r.error || "Connection failed." }); + } catch (e) { setMsg({ kind: "err", text: (e as Error).message }); } + }; + + const remove = async (id: number) => { + try { await api.deleteConnection(id); refresh(); } + catch (e) { setMsg({ kind: "err", text: (e as Error).message }); } + }; + + const dsnHint = driver === "mysql" ? "user:pass@tcp(host:3306)/dbname" : "postgres://user:pass@host:5432/dbname?sslmode=require"; + + return ( +
+

Saved connections

+

+ Store a database connection once and reference it by name when you create an operation. + The connection string is encrypted at rest and never shown again. +

+ + {conns.length > 0 && ( +
+ {conns.map((c) => ( +
+ {c.name} + {c.driver} + added {timeAgo(c.created_at)} +
+ + +
+ ))} +
+ )} + +
+
+ + setName(e.target.value)} placeholder="prod-users-db" /> +
+
+ +
+ + +
+
+
+
+ + setDsn(e.target.value)} spellCheck={false} placeholder={dsnHint} /> +
Format: {dsnHint}
+
+ +
+ +
+ +
+ {msg &&
{msg.text}
} +
+ ); +} + export function Setup({ onNew }: { onNew: () => void }) { return ( <>

Connect your system

-
How MARATHON reaches your database or service to do the work.
+
Save a connection, and learn how Marathon reaches your database or service.
+ +

The model

- MARATHON is self-hosted — you run it inside your own network (Docker, Kubernetes, a VM). - It needs network access to your database host and nothing else. Your data never leaves your infrastructure; - there is no MARATHON cloud in the path. + Marathon is self-hosted — you run it inside your own network (Docker, Kubernetes, a VM). + It needs network access to your database host and nothing else. Your data never leaves your infrastructure.

There are two ways to apply a change to each batch. You choose per operation:

Option A — SQL directly on the database

-

MARATHON connects to your database and runs a range UPDATE/DELETE per batch, throttled and checkpointed.

-
    -
  • Give it a connection string (DSN). Postgres and MySQL are supported:
    • -
+

Marathon connects to your database and runs a range UPDATE/DELETE per batch, throttled and checkpointed.

{`Postgres : postgres://user:pass@host:5432/dbname?sslmode=require MySQL : user:pass@tcp(host:3306)/dbname`}
-
    -
  • Use a least-privilege user. It only needs to read the cursor column and write the target table:
    • -
+

Use a least-privilege user — it only needs to read the cursor column and write the target table:

{`CREATE USER marathon WITH PASSWORD '••••'; GRANT SELECT ON users TO marathon; -- to walk rows GRANT UPDATE ON users TO marathon; -- to apply the change`}
-
    -
  • Network: allow MARATHON's host/pod to reach the DB (VPC peering, security group, or same cluster).
    • -
  • Placeholders: the batch bounds are $1/$2 on Postgres, two ? on MySQL.
    • -

Option B — HTTP callback to your service

- When the logic belongs in your codebase (any language), MARATHON still reads each batch from the DB, + When the logic belongs in your codebase (any language), Marathon still reads each batch from the DB, then POSTs the rows to an endpoint you provide. Your service does the work and reports per-row results.

{`→ POST https://your-service/marathon/apply @@ -49,19 +143,7 @@ GRANT UPDATE ON users TO marathon; -- to apply the change`}
← 200 OK { "results": [ { "row_key": "i:1001", "ok": true }, { "row_key": "i:1002", "ok": false, "error": "…" } ] }`}
-
    -
  • Rows returned ok:false are quarantined (not retried inline) — the run keeps going.
    • -
  • A non-2xx response or timeout fails the batch, so the run can back off and resume safely.
    • -
  • You still provide a read DSN so MARATHON can page through the rows.
    • -
- -

What MARATHON guarantees while connected

-
    -
  • Polite to production: a token-bucket rate limit (and adaptive throttle in fleet mode) keeps load bounded.
    • -
  • Crash-safe: every batch checkpoints — a killed run resumes from where it stopped, never from zero.
    • -
  • Auditable: every run and control action is recorded in an immutable log.
    • -
  • Access-controlled: set an API key (MARATHON_KEYS) to require viewer/operator/admin roles.
    • -
+

Rows returned ok:false are quarantined; the run keeps going.

diff --git a/web/src/pages/TaskWizard.tsx b/web/src/pages/TaskWizard.tsx index 8142768..2203037 100644 --- a/web/src/pages/TaskWizard.tsx +++ b/web/src/pages/TaskWizard.tsx @@ -1,6 +1,6 @@ -import { useState } from "react"; +import { useEffect, useState } from "react"; import { api } from "../api"; -import type { DryRunReport } from "../types"; +import type { Connection, DryRunReport } from "../types"; import { fmt } from "../components/widgets"; const PG_OP = "UPDATE users SET full_name = first_name || ' ' || last_name WHERE id >= $1 AND id <= $2 AND full_name IS NULL"; @@ -11,6 +11,7 @@ const MY_DSN = "user:pass@tcp(host:3306)/dbname"; type Form = { name: string; + connection_id: number | null; target_driver: "postgres" | "mysql"; target_dsn: string; source_table: string; @@ -26,6 +27,7 @@ type Form = { const initial: Form = { name: "", + connection_id: null, target_driver: "postgres", target_dsn: "postgres://demo:demo@localhost:5434/demo?sslmode=disable", source_table: "users", @@ -49,11 +51,49 @@ export function TaskWizard({ onSetup: () => void; }) { const [f, setF] = useState
(initial); + const [mode, setMode] = useState<"dsn" | "saved">("dsn"); + const [conns, setConns] = useState([]); const [createdId, setCreatedId] = useState(null); const [dry, setDry] = useState(null); const [err, setErr] = useState(""); const [busy, setBusy] = useState(""); + useEffect(() => { + api.listConnections().then((list) => { + setConns(list); + // Prefer a saved connection when the user has one. + if (list.length > 0) { + setMode("saved"); + const c = list[0]; + const drv = (c.driver === "mysql" ? "mysql" : "postgres") as "postgres" | "mysql"; + setF((prev) => ({ ...prev, connection_id: c.id, target_driver: drv, target_dsn: "" })); + } + }).catch(() => {}); + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []); + + // Pick a saved connection: set connection_id, adopt its driver (for the SQL + // placeholder style), and clear the inline DSN. + const pickConnection = (id: number) => { + const c = conns.find((x) => x.id === id); + const drv = (c?.driver === "mysql" ? "mysql" : "postgres") as "postgres" | "mysql"; + setCreatedId(null); setDry(null); + setF((prev) => ({ + ...prev, + connection_id: id, + target_driver: drv, + target_dsn: "", + operation_sql: prev.operation_sql === PG_OP || prev.operation_sql === MY_OP ? (drv === "mysql" ? MY_OP : PG_OP) : prev.operation_sql, + })); + }; + + const switchMode = (m: "dsn" | "saved") => { + setMode(m); + setCreatedId(null); setDry(null); + if (m === "dsn") setF((prev) => ({ ...prev, connection_id: null, target_dsn: initial.target_dsn })); + else if (conns.length) pickConnection(conns[0].id); + }; + const set = (k: K, v: Form[K]) => { setCreatedId(null); // any edit invalidates the saved draft setDry(null); @@ -142,25 +182,50 @@ export function TaskWizard({
2

Connect your database

-

MARATHON connects directly to your database with a connection string. It's self-hosted, so it runs inside your network — nothing leaves.

+

Use a saved connection, or enter a connection string directly. Marathon is self-hosted, so nothing leaves your network.

-
- - + +
-
- - set("target_dsn", e.target.value)} spellCheck={false} /> -
Format: {dsnHint}
-
- -
- 🔒 Least privilege: create a dedicated DB user that can only SELECT the cursor column and UPDATE/DELETE the target table. MARATHON only ever touches the rows your operation matches. -
+ {mode === "saved" ? ( + conns.length === 0 ? ( +
+ No saved connections yet. Add one on the page, then come back. +
+ ) : ( +
+ + +
The connection string stays encrypted — it's resolved only when a run starts.
+
+ ) + ) : ( + <> +
+ +
+ + +
+
+
+ + set("target_dsn", e.target.value)} spellCheck={false} /> +
Format: {dsnHint}
+
+
+ 🔒 Least privilege: a dedicated DB user that can only SELECT the cursor column and UPDATE/DELETE the target table. +
+ + )}
diff --git a/web/src/types.ts b/web/src/types.ts index 35332ca..659d71a 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -1,6 +1,14 @@ +export interface Connection { + id: number; + name: string; + driver: string; // "postgres" | "mysql" + created_at: string; +} + export interface Task { id: number; name: string; + connection_id: number | null; target_driver: string; // "postgres" | "mysql" target_dsn: string; source_table: string;