feat(routines): add workspace-aware routine runs

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/db/package.json

@@ -35,11 +35,12 @@
     "dist"
   ],
   "scripts": {
-    "build": "tsc && cp -r src/migrations dist/migrations",
+    "check:migrations": "tsx src/check-migration-numbering.ts",
+    "build": "pnpm run check:migrations && tsc && cp -r src/migrations dist/migrations",
     "clean": "rm -rf dist",
-    "typecheck": "tsc --noEmit",
-    "generate": "tsc -p tsconfig.json && drizzle-kit generate",
-    "migrate": "tsx src/migrate.ts",
+    "typecheck": "pnpm run check:migrations && tsc --noEmit",
+    "generate": "pnpm run check:migrations && tsc -p tsconfig.json && drizzle-kit generate",
+    "migrate": "pnpm run check:migrations && tsx src/migrate.ts",
     "seed": "tsx src/seed.ts"
   },
   "dependencies": {

packages/db/src/check-migration-numbering.ts

@@ -0,0 +1,89 @@
+import { readdir, readFile } from "node:fs/promises";
+import { fileURLToPath } from "node:url";
+
+const migrationsDir = fileURLToPath(new URL("./migrations", import.meta.url));
+const journalPath = fileURLToPath(new URL("./migrations/meta/_journal.json", import.meta.url));
+
+type JournalFile = {
+  entries?: Array<{
+    idx?: number;
+    tag?: string;
+  }>;
+};
+
+function migrationNumber(value: string): string | null {
+  const match = value.match(/^(\d{4})_/);
+  return match ? match[1] : null;
+}
+
+function ensureNoDuplicates(values: string[], label: string) {
+  const seen = new Map<string, string>();
+
+  for (const value of values) {
+    const number = migrationNumber(value);
+    if (!number) {
+      throw new Error(`${label} entry does not start with a 4-digit migration number: ${value}`);
+    }
+    const existing = seen.get(number);
+    if (existing) {
+      throw new Error(`Duplicate migration number ${number} in ${label}: ${existing}, ${value}`);
+    }
+    seen.set(number, value);
+  }
+}
+
+function ensureStrictlyOrdered(values: string[], label: string) {
+  const sorted = [...values].sort();
+  for (let index = 0; index < values.length; index += 1) {
+    if (values[index] !== sorted[index]) {
+      throw new Error(
+        `${label} are out of order at position ${index}: expected ${sorted[index]}, found ${values[index]}`,
+      );
+    }
+  }
+}
+
+function ensureJournalMatchesFiles(migrationFiles: string[], journalTags: string[]) {
+  const journalFiles = journalTags.map((tag) => `${tag}.sql`);
+
+  if (journalFiles.length !== migrationFiles.length) {
+    throw new Error(
+      `Migration journal/file count mismatch: journal has ${journalFiles.length}, files have ${migrationFiles.length}`,
+    );
+  }
+
+  for (let index = 0; index < migrationFiles.length; index += 1) {
+    const migrationFile = migrationFiles[index];
+    const journalFile = journalFiles[index];
+    if (migrationFile !== journalFile) {
+      throw new Error(
+        `Migration journal/file order mismatch at position ${index}: journal has ${journalFile}, files have ${migrationFile}`,
+      );
+    }
+  }
+}
+
+async function main() {
+  const migrationFiles = (await readdir(migrationsDir))
+    .filter((entry) => entry.endsWith(".sql"))
+    .sort();
+
+  ensureNoDuplicates(migrationFiles, "migration files");
+  ensureStrictlyOrdered(migrationFiles, "migration files");
+
+  const rawJournal = await readFile(journalPath, "utf8");
+  const journal = JSON.parse(rawJournal) as JournalFile;
+  const journalTags = (journal.entries ?? [])
+    .map((entry, index) => {
+      if (typeof entry.tag !== "string" || entry.tag.length === 0) {
+        throw new Error(`Migration journal entry ${index} is missing a tag`);
+      }
+      return entry.tag;
+    });
+
+  ensureNoDuplicates(journalTags, "migration journal");
+  ensureStrictlyOrdered(journalTags, "migration journal");
+  ensureJournalMatchesFiles(migrationFiles, journalTags);
+}
+
+await main();

packages/db/src/client.test.ts

@@ -305,6 +305,99 @@ describeEmbeddedPostgres("applyPendingMigrations", () => {
 
       const finalState = await inspectMigrations(connectionString);
       expect(finalState.status).toBe("upToDate");
+
+      const verifySql = postgres(connectionString, { max: 1, onnotice: () => {} });
+      try {
+        const constraints = await verifySql.unsafe<{ conname: string }[]>(
+          `
+            SELECT conname
+            FROM pg_constraint
+            WHERE conname IN (
+              'feedback_exports_company_id_companies_id_fk',
+              'feedback_exports_feedback_vote_id_feedback_votes_id_fk',
+              'feedback_exports_issue_id_issues_id_fk',
+              'feedback_votes_company_id_companies_id_fk',
+              'feedback_votes_issue_id_issues_id_fk'
+            )
+            ORDER BY conname
+          `,
+        );
+        expect(constraints.map((row) => row.conname)).toEqual([
+          "feedback_exports_company_id_companies_id_fk",
+          "feedback_exports_feedback_vote_id_feedback_votes_id_fk",
+          "feedback_exports_issue_id_issues_id_fk",
+          "feedback_votes_company_id_companies_id_fk",
+          "feedback_votes_issue_id_issues_id_fk",
+        ]);
+      } finally {
+        await verifySql.end();
+      }
+    },
+    20_000,
+  );
+
+  it(
+    "replays migration 0048 safely when routines.variables already exists",
+    async () => {
+      const connectionString = await createTempDatabase();
+
+      await applyPendingMigrations(connectionString);
+
+      const sql = postgres(connectionString, { max: 1, onnotice: () => {} });
+      try {
+        const flashyMarrowHash = await migrationHash("0048_flashy_marrow.sql");
+
+        await sql.unsafe(
+          `DELETE FROM "drizzle"."__drizzle_migrations" WHERE hash = '${flashyMarrowHash}'`,
+        );
+
+        const columns = await sql.unsafe<{ column_name: string }[]>(
+          `
+            SELECT column_name
+            FROM information_schema.columns
+            WHERE table_schema = 'public'
+              AND table_name = 'routines'
+              AND column_name = 'variables'
+          `,
+        );
+        expect(columns).toHaveLength(1);
+      } finally {
+        await sql.end();
+      }
+
+      const pendingState = await inspectMigrations(connectionString);
+      expect(pendingState).toMatchObject({
+        status: "needsMigrations",
+        pendingMigrations: ["0048_flashy_marrow.sql"],
+        reason: "pending-migrations",
+      });
+
+      await applyPendingMigrations(connectionString);
+
+      const finalState = await inspectMigrations(connectionString);
+      expect(finalState.status).toBe("upToDate");
+
+      const verifySql = postgres(connectionString, { max: 1, onnotice: () => {} });
+      try {
+        const columns = await verifySql.unsafe<{ column_name: string; is_nullable: string; data_type: string }[]>(
+          `
+            SELECT column_name, is_nullable, data_type
+            FROM information_schema.columns
+            WHERE table_schema = 'public'
+              AND table_name = 'routines'
+              AND column_name = 'variables'
+          `,
+        );
+        expect(columns).toEqual([
+          expect.objectContaining({
+            column_name: "variables",
+            is_nullable: "NO",
+            data_type: "jsonb",
+          }),
+        ]);
+      } finally {
+        await verifySql.end();
+      }
     },
     20_000,
   );

packages/db/src/migrations/0048_flashy_marrow.sql

@@ -0,0 +1 @@
+ALTER TABLE "routines" ADD COLUMN IF NOT EXISTS "variables" jsonb DEFAULT '[]'::jsonb NOT NULL;

packages/db/src/migrations/meta/_journal.json

@@ -337,6 +337,13 @@
       "when": 1775137972687,
       "tag": "0047_overjoyed_groot",
       "breakpoints": true
+    },
+    {
+      "idx": 48,
+      "version": "7",
+      "when": 1775145655557,
+      "tag": "0048_flashy_marrow",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/routines.ts

@@ -15,6 +15,7 @@ import { companySecrets } from "./company_secrets.js";
 import { issues } from "./issues.js";
 import { projects } from "./projects.js";
 import { goals } from "./goals.js";
+import type { RoutineVariable } from "@paperclipai/shared";
 
 export const routines = pgTable(
   "routines",
@@ -31,6 +32,7 @@ export const routines = pgTable(
     status: text("status").notNull().default("active"),
     concurrencyPolicy: text("concurrency_policy").notNull().default("coalesce_if_active"),
     catchUpPolicy: text("catch_up_policy").notNull().default("skip_missed"),
+    variables: jsonb("variables").$type<RoutineVariable[]>().notNull().default([]),
     createdByAgentId: uuid("created_by_agent_id").references(() => agents.id, { onDelete: "set null" }),
     createdByUserId: text("created_by_user_id"),
     updatedByAgentId: uuid("updated_by_agent_id").references(() => agents.id, { onDelete: "set null" }),