Consolidate 12 migrations into a single baseline schema

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
 ```
 

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

deploy/seed-target.sql → 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,

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
+		},
+	}
+}