feat(db): implement database migration system with version tracking a…

backend/src/api/app.ts

@@ -19,7 +19,7 @@ import {
   type StellarReleasePaymentFn,
 } from "../payment";
 import { agentsRouter } from "./routes/agents";
-import { healthRouter } from "./routes/health";
+import { healthRouter, getMigrationsHandler } from "./routes/health";
 import { createStatsRouter } from "./routes/stats";
 import { createTasksRouter } from "./routes/tasks";
 import { rateLimitMiddleware, registerRateLimitMiddleware } from "./middleware/rateLimit";
@@ -96,8 +96,10 @@ export function createApp(opts: AppOptions = {}): {
     heartbeatService.start();
   }
 
-  // ── Health routes ───────────────────────────────────────────────────────────
+  // ── Health & Migration routes ───────────────────────────────────────────────
   app.use("/health", healthRouter);
+  app.get("/migrations", getMigrationsHandler);
+
 
   // ── Stats routes ───────────────────────────────────────────────────────────
   app.use("/api/stats", createStatsRouter(getTaskDb()));

backend/src/api/routes/health.ts

@@ -1,5 +1,8 @@
 import { Router, Request, Response } from "express";
 import { getConfig } from "../../config";
+import { getTaskDb } from "../../db/tasks";
+import { getMigrationStatus } from "../../db/migrations";
+
 
 const router = Router();
 
@@ -122,6 +125,42 @@ router.get("/ready", async (_req: Request, res: Response) => {
   res.status(allOk ? 200 : 500).json({ status: allOk ? "ok" : "error", checks });
 });
 
+/**
+ * @openapi
+ * /migrations:
+ *   get:
+ *     summary: Database migration status
+ *     operationId: getMigrationsStatus
+ *     description: Returns the list of available database migrations and their application status.
+ *     tags: [Migrations]
+ *     security: []
+ *     responses:
+ *       200:
+ *         description: List of migration statuses
+ *         content:
+ *           application/json:
+ *             schema:
+ *               type: object
+ *               properties:
+ *                 status:
+ *                   type: string
+ *                 migrations:
+ *                   type: array
+ */
+export function getMigrationsHandler(_req: Request, res: Response): void {
+  try {
+    const db = getTaskDb();
+    const migrations = getMigrationStatus(db);
+    res.json({ status: "ok", migrations });
+  } catch (error) {
+    res.status(500).json({ status: "error", error: String(error) });
+  }
+}
+
+
+router.get("/migrations", getMigrationsHandler);
+
+
 async function checkVenice(apiKey: string): Promise<"ok" | "unreachable"> {
   try {
     const ctrl = new AbortController();

backend/src/config/index.ts

@@ -15,6 +15,13 @@ const envSchema = z.object({
   NPM_PACKAGE_VERSION: z.string().default(pkg.version ?? "0.1.0"),
   GRACEFUL_SHUTDOWN_TIMEOUT: z.coerce.number().int().positive().default(30),
 
+  // ── Database Migrations ───────────────────────────────────────────────────
+  /** Whether to automatically run database migrations on server startup. Default: true. */
+  AUTO_MIGRATE: z
+    .string()
+    .optional()
+    .transform((val) => val === undefined || val === "" || val === "true" || val === "1"),
+
   // ── Input validation ────────────────────────────────────────────────────────
   /** Maximum allowed length (characters) for a task prompt. Default: 10 000. */
   MAX_PROMPT_LENGTH: z.coerce.number().int().positive().default(10_000),

backend/src/db/migrations/001_add_stats_indexes.sql

@@ -1,3 +1,29 @@
 -- Add indexes to speed up dashboard and analytics queries
+CREATE TABLE IF NOT EXISTS tasks (
+  id              TEXT PRIMARY KEY,
+  prompt          TEXT NOT NULL,
+  walletPublicKey TEXT NOT NULL DEFAULT '',
+  status          TEXT NOT NULL DEFAULT 'queued',
+  dagJson         TEXT NOT NULL DEFAULT '[]',
+  createdAt       TEXT NOT NULL,
+  updatedAt       TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS payments (
+  taskId        TEXT NOT NULL,
+  nodeId        TEXT NOT NULL,
+  balanceId     TEXT NOT NULL,
+  status        TEXT NOT NULL DEFAULT 'locked',
+  amountStroops TEXT NOT NULL,
+  txHash        TEXT,
+  PRIMARY KEY (taskId, nodeId)
+);
+
 CREATE INDEX IF NOT EXISTS idx_tasks_created_at ON tasks ("createdAt");
 CREATE INDEX IF NOT EXISTS idx_payments_status ON payments (status);
+
+
+-- DOWN
+DROP INDEX IF EXISTS idx_tasks_created_at;
+DROP INDEX IF EXISTS idx_payments_status;
+

backend/src/db/migrations/002_create_tasks_table.sql

@@ -19,3 +19,9 @@ CREATE TABLE IF NOT EXISTS task_events (
 );
 
 CREATE INDEX IF NOT EXISTS idx_task_events_taskId ON task_events (taskId);
+
+-- DOWN
+DROP INDEX IF EXISTS idx_task_events_taskId;
+DROP TABLE IF EXISTS task_events;
+DROP TABLE IF EXISTS tasks;
+

backend/src/db/migrations/003_create_schema_migrations.sql

@@ -0,0 +1,10 @@
+-- Migration 003: create schema_migrations table
+CREATE TABLE IF NOT EXISTS schema_migrations (
+  id          INTEGER PRIMARY KEY AUTOINCREMENT,
+  version     TEXT NOT NULL UNIQUE,
+  name        TEXT NOT NULL,
+  applied_at  TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+-- DOWN
+DROP TABLE IF EXISTS schema_migrations;

backend/src/db/migrations/index.ts

@@ -0,0 +1,73 @@
+import type Database from "better-sqlite3";
+import {
+  runUp,
+  runDown,
+  getStatus,
+  loadMigrations,
+  parseMigrationSql,
+  type Migration,
+  type MigrationStatus,
+  type MigrationResult,
+  type MigrationOptions,
+  type RollbackOptions,
+} from "./runner";
+import {
+  getAppliedMigrations,
+  recordMigration,
+  removeMigration,
+  isApplied,
+  ensureMigrationsTable,
+  type AppliedMigration,
+} from "./tracker";
+
+/**
+ * Runs all pending migrations (UP) on the database.
+ */
+export function runMigrations(
+  db: Database.Database,
+  options: MigrationOptions = {},
+): MigrationResult {
+  return runUp(db, options);
+}
+
+/**
+ * Rolls back applied migrations (DOWN) on the database.
+ */
+export function rollbackMigrations(
+  db: Database.Database,
+  options: RollbackOptions = {},
+): MigrationResult {
+  return runDown(db, options);
+}
+
+/**
+ * Gets the current status of all migrations on the database.
+ */
+export function getMigrationStatus(
+  db: Database.Database,
+  options: { migrationsDir?: string } = {},
+): MigrationStatus[] {
+  return getStatus(db, options.migrationsDir);
+}
+
+export {
+  runUp,
+  runDown,
+  getStatus,
+  loadMigrations,
+  parseMigrationSql,
+  getAppliedMigrations,
+  recordMigration,
+  removeMigration,
+  isApplied,
+  ensureMigrationsTable,
+};
+
+export type {
+  Migration,
+  MigrationStatus,
+  MigrationResult,
+  MigrationOptions,
+  RollbackOptions,
+  AppliedMigration,
+};