External db sync

apps/backend/package.json

@@ -7,7 +7,7 @@
     "typecheck": "tsc --noEmit",
     "with-env": "dotenv -c development --",
     "with-env:prod": "dotenv -c --",
-    "dev": "concurrently -n \"dev,codegen,prisma-studio\" -k \"next dev --turbopack --port ${NEXT_PUBLIC_STACK_PORT_PREFIX:-81}02\" \"pnpm run codegen:watch\" \"pnpm run prisma-studio\"",
+    "dev": "concurrently -n \"dev,codegen,prisma-studio,cron-jobs\" -k \"next dev --turbopack --port ${NEXT_PUBLIC_STACK_PORT_PREFIX:-81}02\" \"pnpm run codegen:watch\" \"pnpm run prisma-studio\" \"pnpm run run-cron-jobs\"",
     "build": "pnpm run codegen && next build",
     "docker-build": "pnpm run codegen && next build --experimental-build-mode compile",
     "build-self-host-migration-script": "tsup --config scripts/db-migrations.tsup.config.ts",
@@ -35,7 +35,8 @@
     "generate-openapi-fumadocs": "pnpm run with-env tsx scripts/generate-openapi-fumadocs.ts",
     "generate-keys": "pnpm run with-env tsx scripts/generate-keys.ts",
     "db-seed-script": "pnpm run db:seed",
-    "verify-data-integrity": "pnpm run with-env tsx scripts/verify-data-integrity.ts"
+    "verify-data-integrity": "pnpm run with-env tsx scripts/verify-data-integrity.ts",
+    "run-cron-jobs": "pnpm run with-env tsx scripts/run-cron-jobs.ts"
   },
   "prisma": {
     "seed": "pnpm run db-seed-script"

apps/backend/prisma/migrations/20251125030551_add_sequence_id/migration.sql

@@ -0,0 +1,18 @@
+CREATE SEQUENCE  global_seq_id
+    AS BIGINT
+    START 1
+    INCREMENT BY 11
+    NO MINVALUE
+    NO MAXVALUE;
+
+-- SPLIT_STATEMENT_SENTINEL
+ALTER TABLE "ContactChannel" ADD COLUMN  "sequenceId" BIGINT;
+
+-- SPLIT_STATEMENT_SENTINEL
+ALTER TABLE "ProjectUser" ADD COLUMN  "sequenceId" BIGINT;
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE UNIQUE INDEX  "ContactChannel_sequenceId_key" ON "ContactChannel"("sequenceId");
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE UNIQUE INDEX  "ProjectUser_sequenceId_key" ON "ProjectUser"("sequenceId");

apps/backend/prisma/migrations/20251125172224_add_outgoing_request/migration.sql

@@ -0,0 +1,13 @@
+
+CREATE TABLE  "OutgoingRequest" (
+    "id" UUID NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "qstashOptions" JSONB NOT NULL,
+    "fulfilledAt" TIMESTAMP(3),
+
+    CONSTRAINT "OutgoingRequest_pkey" PRIMARY KEY ("id")
+);
+
+
+CREATE INDEX  "OutgoingRequest_fulfilledAt_idx" ON "OutgoingRequest"("fulfilledAt");
+

apps/backend/prisma/migrations/20251125180000_add_deleted_row_table/migration.sql

@@ -0,0 +1,21 @@
+CREATE TABLE "DeletedRow" (
+    "id" UUID NOT NULL,
+    "tenancyId" UUID NOT NULL,
+    "tableName" TEXT NOT NULL,
+    "sequenceId" BIGINT,
+    "primaryKey" JSONB NOT NULL,
+    "data" JSONB,
+    "deletedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "fulfilledAt" TIMESTAMP(3),
+
+    CONSTRAINT "DeletedRow_pkey" PRIMARY KEY ("id")
+);
+
+
+CREATE UNIQUE INDEX "DeletedRow_sequenceId_key" ON "DeletedRow"("sequenceId");
+
+CREATE INDEX "DeletedRow_tableName_idx" ON "DeletedRow"("tableName");
+
+CREATE INDEX "DeletedRow_tenancyId_idx" ON "DeletedRow"("tenancyId");
+
+

apps/backend/prisma/migrations/20251128210000_add_should_update_sequence_id_columns/migration.sql

@@ -0,0 +1,8 @@
+ALTER TABLE "ProjectUser" ADD COLUMN "shouldUpdateSequenceId" BOOLEAN NOT NULL DEFAULT TRUE;
+
+-- SPLIT_STATEMENT_SENTINEL
+ALTER TABLE "ContactChannel" ADD COLUMN "shouldUpdateSequenceId" BOOLEAN NOT NULL DEFAULT TRUE;
+
+-- SPLIT_STATEMENT_SENTINEL
+ALTER TABLE "DeletedRow" ADD COLUMN "shouldUpdateSequenceId" BOOLEAN NOT NULL DEFAULT TRUE;
+

apps/backend/prisma/migrations/20251128210001_add_should_update_sequence_id_indexes/migration.sql

@@ -0,0 +1,8 @@
+CREATE INDEX "ProjectUser_shouldUpdateSequenceId_idx" ON "ProjectUser"("shouldUpdateSequenceId") WHERE "shouldUpdateSequenceId" = TRUE;
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE INDEX "ContactChannel_shouldUpdateSequenceId_idx" ON "ContactChannel"("shouldUpdateSequenceId") WHERE "shouldUpdateSequenceId" = TRUE;
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE INDEX "DeletedRow_shouldUpdateSequenceId_idx" ON "DeletedRow"("shouldUpdateSequenceId") WHERE "shouldUpdateSequenceId" = TRUE;
+

apps/backend/prisma/migrations/20251128210002_add_reset_sequence_id_function_and_triggers/migration.sql

@@ -0,0 +1,30 @@
+-- SINGLE_STATEMENT_SENTINEL
+CREATE FUNCTION reset_sequence_id_on_update()
+RETURNS TRIGGER AS $$
+BEGIN
+  NEW."shouldUpdateSequenceId" := TRUE;
+  RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE TRIGGER mark_should_update_sequence_id_project_user
+BEFORE UPDATE ON "ProjectUser"
+FOR EACH ROW
+WHEN (OLD."shouldUpdateSequenceId" = FALSE)
+EXECUTE FUNCTION reset_sequence_id_on_update();
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE TRIGGER mark_should_update_sequence_id_contact_channel
+BEFORE UPDATE ON "ContactChannel"
+FOR EACH ROW
+WHEN (OLD."shouldUpdateSequenceId" = FALSE)
+EXECUTE FUNCTION reset_sequence_id_on_update();
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE TRIGGER mark_should_update_sequence_id_deleted_row
+BEFORE UPDATE ON "DeletedRow"
+FOR EACH ROW
+WHEN (OLD."shouldUpdateSequenceId" = FALSE)
+EXECUTE FUNCTION reset_sequence_id_on_update();
+

apps/backend/prisma/migrations/20251128210003_log_deleted_row_function_and_triggers/migration.sql

@@ -0,0 +1,55 @@
+-- SINGLE_STATEMENT_SENTINEL
+CREATE FUNCTION log_deleted_row()
+RETURNS TRIGGER AS $function$
+DECLARE
+  row_data jsonb;
+  pk jsonb := '{}'::jsonb;
+  col record;
+BEGIN
+  row_data := to_jsonb(OLD);
+
+  FOR col IN
+    SELECT a.attname
+    FROM pg_index i
+    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
+    WHERE i.indrelid = TG_RELID
+      AND i.indisprimary
+  LOOP
+    pk := pk || jsonb_build_object(col.attname, row_data -> col.attname);
+  END LOOP;
+
+  INSERT INTO "DeletedRow" (
+    "id",
+    "tenancyId",
+    "tableName",
+    "primaryKey",
+    "data",
+    "deletedAt",
+    "shouldUpdateSequenceId"
+  )
+  VALUES (
+    gen_random_uuid(), 
+    OLD."tenancyId", 
+    TG_TABLE_NAME, 
+    pk,
+    row_data, 
+    NOW(),
+    TRUE
+  );
+
+  RETURN OLD;
+END;
+$function$ LANGUAGE plpgsql;
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE TRIGGER log_deleted_row_project_user
+BEFORE DELETE ON "ProjectUser"
+FOR EACH ROW
+EXECUTE FUNCTION log_deleted_row();
+
+-- SPLIT_STATEMENT_SENTINEL
+CREATE TRIGGER log_deleted_row_contact_channel
+BEFORE DELETE ON "ContactChannel"
+FOR EACH ROW
+EXECUTE FUNCTION log_deleted_row();
+

apps/backend/prisma/migrations/20251128210004_add_sync_functions/migration.sql

@@ -0,0 +1,100 @@
+-- SINGLE_STATEMENT_SENTINEL
+CREATE FUNCTION enqueue_tenant_sync(p_tenant_id uuid)
+RETURNS void AS $$
+BEGIN
+  INSERT INTO "OutgoingRequest" ("id", "createdAt", "qstashOptions", "fulfilledAt")
+  SELECT
+    gen_random_uuid(),
+    NOW(),
+    json_build_object(
+      'url',  '/api/latest/internal/external-db-sync/sync-engine',
+      'body', json_build_object('tenantId', p_tenant_id)
+    ),
+    NULL
+  WHERE NOT EXISTS (
+    SELECT 1
+    FROM "OutgoingRequest"
+    WHERE "fulfilledAt" IS NULL
+      AND ("qstashOptions"->'body'->>'tenantId')::uuid = p_tenant_id
+  );
+END;
+$$ LANGUAGE plpgsql;
+-- SPLIT_STATEMENT_SENTINEL
+-- SINGLE_STATEMENT_SENTINEL
+CREATE FUNCTION backfill_null_sequence_ids()
+RETURNS void AS $$
+DECLARE
+  v_tenancy_id uuid;
+BEGIN
+  FOR v_tenancy_id IN
+    WITH rows_to_update AS (
+      SELECT "tenancyId", "projectUserId"
+      FROM "ProjectUser"
+      WHERE "shouldUpdateSequenceId" = TRUE
+      OR "sequenceId" IS NULL
+      LIMIT 1000
+      FOR UPDATE SKIP LOCKED
+    ),
+    updated_rows AS (
+      UPDATE "ProjectUser" pu
+      SET "sequenceId" = nextval('global_seq_id'),
+          "shouldUpdateSequenceId" = FALSE
+      FROM rows_to_update r
+      WHERE pu."tenancyId"     = r."tenancyId"
+        AND pu."projectUserId" = r."projectUserId"
+      RETURNING pu."tenancyId"
+    )
+    SELECT DISTINCT "tenancyId" FROM updated_rows
+  LOOP
+    PERFORM enqueue_tenant_sync(v_tenancy_id);
+  END LOOP;
+
+  FOR v_tenancy_id IN
+    WITH rows_to_update AS (
+      SELECT "tenancyId", "projectUserId", "id"
+      FROM "ContactChannel"
+      WHERE "shouldUpdateSequenceId" = TRUE
+      OR "sequenceId" IS NULL
+      LIMIT 1000
+      FOR UPDATE SKIP LOCKED
+    ),
+    updated_rows AS (
+      UPDATE "ContactChannel" cc
+      SET "sequenceId" = nextval('global_seq_id'),
+          "shouldUpdateSequenceId" = FALSE
+      FROM rows_to_update r
+      WHERE cc."tenancyId"     = r."tenancyId"
+        AND cc."projectUserId" = r."projectUserId"
+        AND cc."id"            = r."id"
+      RETURNING cc."tenancyId"
+    )
+    SELECT DISTINCT "tenancyId" FROM updated_rows
+  LOOP
+    PERFORM enqueue_tenant_sync(v_tenancy_id);
+  END LOOP;
+
+  FOR v_tenancy_id IN
+    WITH rows_to_update AS (
+      SELECT "id", "tenancyId"
+      FROM "DeletedRow"
+      WHERE "shouldUpdateSequenceId" = TRUE
+      OR "sequenceId" IS NULL
+      LIMIT 1000
+      FOR UPDATE SKIP LOCKED
+    ),
+    updated_rows AS (
+      UPDATE "DeletedRow" dr
+      SET "sequenceId" = nextval('global_seq_id'),
+          "shouldUpdateSequenceId" = FALSE
+      FROM rows_to_update r
+      WHERE dr."id" = r."id"
+      RETURNING dr."tenancyId"
+    )
+    SELECT DISTINCT "tenancyId" FROM updated_rows
+  LOOP
+    PERFORM enqueue_tenant_sync(v_tenancy_id);
+  END LOOP;
+
+END;
+$$ LANGUAGE plpgsql;
+

apps/backend/prisma/schema.prisma

@@ -173,6 +173,9 @@ model ProjectUser {
   createdAt DateTime @default(now())
   updatedAt DateTime @updatedAt
 
+  sequenceId BigInt? @unique
+  shouldUpdateSequenceId Boolean @default(true)
+
   displayName            String?
   serverMetadata         Json?
   clientReadOnlyMetadata Json?
@@ -252,6 +255,9 @@ model ContactChannel {
   createdAt DateTime @default(now())
   updatedAt DateTime @updatedAt
 
+  sequenceId BigInt? @unique
+  shouldUpdateSequenceId Boolean @default(true)
+
   type        ContactChannelType
   isPrimary   BooleanTrue?
   usedForAuth BooleanTrue?
@@ -861,7 +867,7 @@ model CacheEntry {
 
 model SubscriptionInvoice {
   id String @default(uuid()) @db.Uuid
-  tenancyId String @db.Uuid
+  tenancyId String  @db.Uuid
   stripeSubscriptionId String
   stripeInvoiceId String
   isSubscriptionCreationInvoice Boolean
@@ -874,3 +880,32 @@ model SubscriptionInvoice {
   @@id([tenancyId, id])
   @@unique([tenancyId, stripeInvoiceId])
 }
+
+model OutgoingRequest {
+  id String @id @default(uuid()) @db.Uuid
+
+  createdAt DateTime @default(now())
+
+  qstashOptions Json
+  fulfilledAt   DateTime?
+
+  @@index([fulfilledAt])
+}
+
+model DeletedRow {
+  id        String @id @default(uuid()) @db.Uuid
+  tenancyId String @db.Uuid
+  tableName String
+
+  sequenceId             BigInt?  @unique
+  shouldUpdateSequenceId Boolean  @default(true)
+
+  primaryKey Json
+  data       Json?
+
+  deletedAt   DateTime  @default(now())
+  fulfilledAt DateTime?
+
+  @@index([tableName])
+  @@index([tenancyId])
+}

apps/backend/scripts/run-cron-jobs.ts

@@ -0,0 +1,40 @@
+import { getEnvVariable } from "@stackframe/stack-shared/dist/utils/env";
+import { StackAssertionError } from "@stackframe/stack-shared/dist/utils/errors";
+import { runAsynchronously } from "@stackframe/stack-shared/dist/utils/promises";
+
+const endpoints = [
+    "/api/latest/internal/external-db-sync/sequencer",
+    "/api/latest/internal/external-db-sync/poller",
+];
+
+async function main() {
+  console.log("Starting cron jobs...");
+  const cronSecret = getEnvVariable('CRON_SECRET');
+
+  const baseUrl = `http://localhost:${getEnvVariable('NEXT_PUBLIC_STACK_PORT_PREFIX', '81')}02`;
+
+  const run = (endpoint: string) => runAsynchronously(async () => {
+    console.log(`Running ${endpoint}...`);
+    const res = await fetch(`${baseUrl}${endpoint}`, {
+      headers: { 'Authorization': `Bearer ${cronSecret}` },
+    });
+    if (!res.ok) throw new StackAssertionError(`Failed to call ${endpoint}: ${res.status} ${res.statusText}\n${await res.text()}`, { res });
+    console.log(`${endpoint} completed.`);
+  });
+
+  for (const endpoint of endpoints) {
+    runAsynchronously(async () => {
+      while (true) {
+        run(endpoint);
+        // Vercel only guarantees minute-granularity for cron jobs, so we randomize the interval
+        await wait(Math.random() * 120_000);
+      }
+    }); 
+  }
+}
+
+// eslint-disable-next-line no-restricted-syntax
+main().catch((err) => {
+  console.error(err);
+  process.exit(1);
+});