diff --git a/README.md b/README.md index c40d32c..358b1b7 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ You need Docker and Go 1.26+. ```bash make up # starts Postgres, Redis, and a demo DB seeded with 1M rows + # (the demo data is deploy/sandbox/seed-target.sql — sandbox only) make run # starts Marathon on :8000 ``` diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 62bb056..a87982c 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -51,7 +51,7 @@ services: ports: - "5434:5432" volumes: - - ./seed-target.sql:/docker-entrypoint-initdb.d/01-seed.sql:ro + - ./sandbox/seed-target.sql:/docker-entrypoint-initdb.d/01-seed.sql:ro healthcheck: test: ["CMD-SHELL", "pg_isready -U demo"] interval: 2s diff --git a/deploy/seed-target.sql b/deploy/sandbox/seed-target.sql similarity index 50% rename from deploy/seed-target.sql rename to deploy/sandbox/seed-target.sql index 3e37f4f..2b86275 100644 --- a/deploy/seed-target.sql +++ b/deploy/sandbox/seed-target.sql @@ -1,4 +1,7 @@ --- Demo "customer" table: 1M users whose full_name needs backfilling. +-- Sandbox/demo fixture only — NOT part of Marathon. +-- docker-compose mounts this into the fake "customer" target database so a +-- local `make up` comes pre-loaded with a realistic backfill scenario: +-- 1,000,000 users whose full_name is empty and needs backfilling. CREATE TABLE users ( id BIGSERIAL PRIMARY KEY, first_name TEXT NOT NULL, diff --git a/migrations/001_create_tasks.go b/migrations/001_create_tasks.go deleted file mode 100644 index 5dfe8ba..0000000 --- a/migrations/001_create_tasks.go +++ /dev/null @@ -1,25 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -func createTasks() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.Exec(` - CREATE TABLE IF NOT EXISTS tasks ( - id BIGSERIAL PRIMARY KEY, - name TEXT NOT NULL UNIQUE, - target_dsn TEXT NOT NULL, - source_table TEXT NOT NULL, - cursor_column TEXT NOT NULL, - row_filter TEXT NOT NULL DEFAULT '', - batch_size INT NOT NULL DEFAULT 1000, - operation_sql TEXT NOT NULL, - rate_per_sec INT NOT NULL DEFAULT 0, - created_at TIMESTAMPTZ NOT NULL DEFAULT now() - )`) - - return err - }, - } -} diff --git a/migrations/001_schema.go b/migrations/001_schema.go new file mode 100644 index 0000000..d800b41 --- /dev/null +++ b/migrations/001_schema.go @@ -0,0 +1,99 @@ +package migrations + +import "gofr.dev/pkg/gofr/migration" + +// schema creates the entire control-store schema in one migration. The repo +// ships this as its baseline, so there is no need for the incremental steps the +// schema was built up from during development. Tables are created in +// foreign-key dependency order: connections → tasks → runs → children. +func schema() migration.Migrate { + return migration.Migrate{ + UP: func(d migration.Datasource) error { + _, err := d.SQL.Exec(` + -- Saved, named database connections (DSN encrypted at rest). + 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() + ); + + -- Reusable data-operation definitions. + CREATE TABLE IF NOT EXISTS tasks ( + id BIGSERIAL PRIMARY KEY, + name TEXT NOT NULL UNIQUE, + connection_id BIGINT REFERENCES connections(id), + target_driver TEXT NOT NULL DEFAULT 'postgres', + target_dsn TEXT NOT NULL DEFAULT '', + source_table TEXT NOT NULL, + cursor_column TEXT NOT NULL, + row_filter TEXT NOT NULL DEFAULT '', + batch_size INT NOT NULL DEFAULT 1000, + operation_type TEXT NOT NULL DEFAULT 'sql', + operation_sql TEXT NOT NULL DEFAULT '', + operation_url TEXT NOT NULL DEFAULT '', + rate_per_sec INT NOT NULL DEFAULT 0, + adaptive BOOLEAN NOT NULL DEFAULT false, + schedule_seconds INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + + -- One execution of a task. + CREATE TABLE IF NOT EXISTS runs ( + id BIGSERIAL PRIMARY KEY, + task_id BIGINT NOT NULL REFERENCES tasks(id), + state TEXT NOT NULL, + rows_total BIGINT NOT NULL DEFAULT -1, + rows_processed BIGINT NOT NULL DEFAULT 0, + rows_affected BIGINT NOT NULL DEFAULT 0, + last_cursor TEXT NOT NULL DEFAULT '', + error TEXT NOT NULL DEFAULT '', + heartbeat_at TIMESTAMPTZ, + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + finished_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + CREATE INDEX IF NOT EXISTS runs_state_idx ON runs(state); + CREATE INDEX IF NOT EXISTS runs_task_idx ON runs(task_id); + -- At most one active run per task (atomic guard against duplicate runs). + CREATE UNIQUE INDEX IF NOT EXISTS runs_one_active_per_task + ON runs (task_id) WHERE state IN ('queued','running','paused'); + + -- Append-only checkpoint log: "everything up to cursor is committed". + CREATE TABLE IF NOT EXISTS checkpoints ( + id BIGSERIAL PRIMARY KEY, + run_id BIGINT NOT NULL REFERENCES runs(id), + cursor TEXT NOT NULL, + batch_rows INT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + CREATE INDEX IF NOT EXISTS checkpoints_run_idx ON checkpoints(run_id, id DESC); + + -- Rows whose operation failed; inspected and retried by the operator. + CREATE TABLE IF NOT EXISTS quarantined_rows ( + id BIGSERIAL PRIMARY KEY, + run_id BIGINT NOT NULL REFERENCES runs(id), + row_key TEXT NOT NULL, + payload JSONB, + error TEXT NOT NULL, + retry_state TEXT NOT NULL DEFAULT 'pending', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + CREATE INDEX IF NOT EXISTS quarantine_run_idx ON quarantined_rows(run_id); + + -- Immutable audit log of run lifecycle events. + CREATE TABLE IF NOT EXISTS run_events ( + id BIGSERIAL PRIMARY KEY, + run_id BIGINT NOT NULL REFERENCES runs(id), + actor TEXT NOT NULL DEFAULT 'system', + event TEXT NOT NULL, + detail TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + CREATE INDEX IF NOT EXISTS run_events_run_idx ON run_events(run_id);`) + + return err + }, + } +} diff --git a/migrations/002_create_runs.go b/migrations/002_create_runs.go deleted file mode 100644 index 4e0e21d..0000000 --- a/migrations/002_create_runs.go +++ /dev/null @@ -1,29 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -func createRuns() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.Exec(` - CREATE TABLE IF NOT EXISTS runs ( - id BIGSERIAL PRIMARY KEY, - task_id BIGINT NOT NULL REFERENCES tasks(id), - state TEXT NOT NULL, - rows_total BIGINT NOT NULL DEFAULT -1, - rows_processed BIGINT NOT NULL DEFAULT 0, - rows_affected BIGINT NOT NULL DEFAULT 0, - last_cursor TEXT NOT NULL DEFAULT '', - error TEXT NOT NULL DEFAULT '', - heartbeat_at TIMESTAMPTZ, - started_at TIMESTAMPTZ NOT NULL DEFAULT now(), - finished_at TIMESTAMPTZ, - created_at TIMESTAMPTZ NOT NULL DEFAULT now() - ); - CREATE INDEX IF NOT EXISTS runs_state_idx ON runs(state); - CREATE INDEX IF NOT EXISTS runs_task_idx ON runs(task_id)`) - - return err - }, - } -} diff --git a/migrations/003_create_checkpoints.go b/migrations/003_create_checkpoints.go deleted file mode 100644 index 8faafeb..0000000 --- a/migrations/003_create_checkpoints.go +++ /dev/null @@ -1,21 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -func createCheckpoints() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.Exec(` - CREATE TABLE IF NOT EXISTS checkpoints ( - id BIGSERIAL PRIMARY KEY, - run_id BIGINT NOT NULL REFERENCES runs(id), - cursor TEXT NOT NULL, - batch_rows INT NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT now() - ); - CREATE INDEX IF NOT EXISTS checkpoints_run_idx ON checkpoints(run_id, id DESC)`) - - return err - }, - } -} diff --git a/migrations/004_create_quarantine.go b/migrations/004_create_quarantine.go deleted file mode 100644 index 82c938d..0000000 --- a/migrations/004_create_quarantine.go +++ /dev/null @@ -1,25 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -// Quarantine lands in Phase 2; the table ships now so early runs already -// have somewhere to record row-level failures once the executor emits them. -func createQuarantine() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.Exec(` - CREATE TABLE IF NOT EXISTS quarantined_rows ( - id BIGSERIAL PRIMARY KEY, - run_id BIGINT NOT NULL REFERENCES runs(id), - row_key TEXT NOT NULL, - payload JSONB, - error TEXT NOT NULL, - retry_state TEXT NOT NULL DEFAULT 'pending', - created_at TIMESTAMPTZ NOT NULL DEFAULT now() - ); - CREATE INDEX IF NOT EXISTS quarantine_run_idx ON quarantined_rows(run_id)`) - - return err - }, - } -} diff --git a/migrations/005_create_connections.go b/migrations/005_create_connections.go deleted file mode 100644 index 16ecb92..0000000 --- a/migrations/005_create_connections.go +++ /dev/null @@ -1,22 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -// Connections (named, encrypted DSNs) land in Phase 4; MVP tasks carry their -// DSN directly. Table ships now to keep migration numbering stable. -func createConnections() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.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() - )`) - - return err - }, - } -} diff --git a/migrations/006_create_run_events.go b/migrations/006_create_run_events.go deleted file mode 100644 index ba4f4d4..0000000 --- a/migrations/006_create_run_events.go +++ /dev/null @@ -1,22 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -func createRunEvents() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.Exec(` - CREATE TABLE IF NOT EXISTS run_events ( - id BIGSERIAL PRIMARY KEY, - run_id BIGINT NOT NULL REFERENCES runs(id), - actor TEXT NOT NULL DEFAULT 'system', - event TEXT NOT NULL, - detail TEXT NOT NULL DEFAULT '', - created_at TIMESTAMPTZ NOT NULL DEFAULT now() - ); - CREATE INDEX IF NOT EXISTS run_events_run_idx ON run_events(run_id)`) - - return err - }, - } -} diff --git a/migrations/007_task_operation_type.go b/migrations/007_task_operation_type.go deleted file mode 100644 index d58638b..0000000 --- a/migrations/007_task_operation_type.go +++ /dev/null @@ -1,19 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -// Adds HTTP-callback operation support: tasks can post batches to the -// customer's own service instead of running SQL on the target. -func addOperationType() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.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 ''; - ALTER TABLE tasks ALTER COLUMN operation_sql DROP NOT NULL`) - - return err - }, - } -} diff --git a/migrations/008_task_target_driver.go b/migrations/008_task_target_driver.go deleted file mode 100644 index 2ec3fcf..0000000 --- a/migrations/008_task_target_driver.go +++ /dev/null @@ -1,17 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -// Adds MySQL target support: tasks record which driver/dialect their target -// database uses. -func addTargetDriver() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.Exec(` - ALTER TABLE tasks - ADD COLUMN IF NOT EXISTS target_driver TEXT NOT NULL DEFAULT 'postgres'`) - - return err - }, - } -} diff --git a/migrations/009_task_schedule.go b/migrations/009_task_schedule.go deleted file mode 100644 index 4a57aca..0000000 --- a/migrations/009_task_schedule.go +++ /dev/null @@ -1,16 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -// Adds recurring-task support: a task can start a run every N seconds. -func addSchedule() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.Exec(` - ALTER TABLE tasks - ADD COLUMN IF NOT EXISTS schedule_seconds INT NOT NULL DEFAULT 0`) - - return err - }, - } -} diff --git a/migrations/010_task_connection.go b/migrations/010_task_connection.go deleted file mode 100644 index 8cfcfed..0000000 --- a/migrations/010_task_connection.go +++ /dev/null @@ -1,16 +0,0 @@ -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/011_task_adaptive.go b/migrations/011_task_adaptive.go deleted file mode 100644 index 3f77d6e..0000000 --- a/migrations/011_task_adaptive.go +++ /dev/null @@ -1,17 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -// Adds the adaptive auto-throttle opt-in: when true, the worker steers the rate -// by observed target-DB latency, using rate_per_sec as the ceiling. -func addAdaptive() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - _, err := d.SQL.Exec(` - ALTER TABLE tasks - ADD COLUMN IF NOT EXISTS adaptive BOOLEAN NOT NULL DEFAULT false`) - - return err - }, - } -} diff --git a/migrations/012_one_active_run.go b/migrations/012_one_active_run.go deleted file mode 100644 index c3efce8..0000000 --- a/migrations/012_one_active_run.go +++ /dev/null @@ -1,31 +0,0 @@ -package migrations - -import "gofr.dev/pkg/gofr/migration" - -// Enforces at most one active (queued/running/paused) run per task at the DB -// level. This closes the check-then-insert race in StartRun: two concurrent -// starts (manual + scheduler, or double-click) can no longer both create a run. -func oneActiveRunPerTask() migration.Migrate { - return migration.Migrate{ - UP: func(d migration.Datasource) error { - // Defensive: collapse any pre-existing duplicates before adding the - // unique index, keeping only the newest active run per task. - if _, err := d.SQL.Exec(` - UPDATE runs SET state = 'killed', finished_at = now() - WHERE state IN ('queued','running','paused') - AND id NOT IN ( - SELECT max(id) FROM runs - WHERE state IN ('queued','running','paused') - GROUP BY task_id)`); err != nil { - return err - } - - _, err := d.SQL.Exec(` - CREATE UNIQUE INDEX IF NOT EXISTS runs_one_active_per_task - ON runs (task_id) - WHERE state IN ('queued','running','paused')`) - - return err - }, - } -} diff --git a/migrations/all.go b/migrations/all.go index eb6402f..7f0d29c 100644 --- a/migrations/all.go +++ b/migrations/all.go @@ -1,22 +1,12 @@ -// Package migrations defines the control-store schema. Each migration is -// numbered and immutable once shipped. +// Package migrations defines the control-store schema. The full schema ships as +// a single baseline migration; add new numbered migrations after it for future +// changes. package migrations import "gofr.dev/pkg/gofr/migration" func All() map[int64]migration.Migrate { return map[int64]migration.Migrate{ - 1: createTasks(), - 2: createRuns(), - 3: createCheckpoints(), - 4: createQuarantine(), - 5: createConnections(), - 6: createRunEvents(), - 7: addOperationType(), - 8: addTargetDriver(), - 9: addSchedule(), - 10: addTaskConnection(), - 11: addAdaptive(), - 12: oneActiveRunPerTask(), + 1: schema(), } }