Add Supabase-backed sync queue backend with cron processor while keeping BullMQ

Author: marcus

.env.example

@@ -8,4 +8,10 @@ SUPABASE_SERVICE_ROLE_KEY="your-service-role-key"
 MASTER_KEY="base64-32-byte-key"
 REDIS_URL="redis://localhost:6379"
 
+# Queue backend: bull (default) or supabase
+SYNC_QUEUE_BACKEND="bull"
+
+# Required only for Supabase queue cron processor endpoint
+CRON_SECRET="change-me"
+
 APP_URL="http://localhost:3000"

README.md

@@ -288,6 +288,8 @@ Copy `.env.example` to `.env.local` and set:
 - `SUPABASE_SERVICE_ROLE_KEY`
 - `MASTER_KEY` (base64-encoded 32-byte key)
 - `REDIS_URL`
+- `SYNC_QUEUE_BACKEND` (`bull` or `supabase`)
+- `CRON_SECRET` (required when `SYNC_QUEUE_BACKEND=supabase`)
 - `APP_URL`
 - `NEXT_PUBLIC_APP_NAME` (optional, default `ContributionPulse`)
 - `NEXT_PUBLIC_APP_SLUG` (optional, default derived from app name)
@@ -313,6 +315,50 @@ npm run worker:nightly
 
 `worker:nightly` registers the repeatable nightly sync schedule. Run it once per environment.
 
+### Supabase queue mode (no Redis worker required)
+If you want to run without BullMQ/Redis for low-cost environments:
+
+1. Set:
+```bash
+SYNC_QUEUE_BACKEND=supabase
+CRON_SECRET=<strong-random-secret>
+```
+
+2. Do not run `npm run worker` for queue processing.
+
+3. Trigger queue processing by calling:
+```bash
+POST /api/internal/sync/process
+Authorization: Bearer <CRON_SECRET>
+```
+
+You can process manually:
+```bash
+npm run queue:process
+```
+
+4. Schedule this endpoint with Supabase Cron (example every minute):
+```sql
+select
+  cron.schedule(
+    'process-contribution-sync-queue',
+    '* * * * *',
+    $$
+    select
+      net.http_post(
+        url := 'https://YOUR_APP_DOMAIN/api/internal/sync/process?limit=10',
+        headers := jsonb_build_object(
+          'Content-Type', 'application/json',
+          'Authorization', 'Bearer YOUR_CRON_SECRET'
+        ),
+        body := '{}'::jsonb
+      );
+    $$
+  );
+```
+
+This keeps BullMQ code in the project for future scale, while allowing Redis-free operation today.
+
 ---
 
 ## Testing

package.json

@@ -12,7 +12,8 @@
     "prisma:generate": "prisma generate",
     "prisma:migrate": "prisma migrate dev",
     "worker": "tsx scripts/worker.ts",
-    "worker:nightly": "tsx scripts/nightly.ts"
+    "worker:nightly": "tsx scripts/nightly.ts",
+    "queue:process": "tsx scripts/process-supabase-queue.ts"
   },
   "dependencies": {
     "@prisma/client": "6.5.0",

prisma/migrations/20260303143000_add_supabase_sync_jobs/migration.sql

@@ -0,0 +1,30 @@
+CREATE TYPE "SyncJobStatus" AS ENUM ('QUEUED', 'RUNNING', 'COMPLETED', 'FAILED');
+
+CREATE TABLE "SyncJob" (
+  "id" TEXT NOT NULL,
+  "userId" TEXT NOT NULL,
+  "provider" "Provider",
+  "from" TIMESTAMP(3),
+  "to" TIMESTAMP(3),
+  "backfillYear" INTEGER,
+  "status" "SyncJobStatus" NOT NULL DEFAULT 'QUEUED',
+  "attemptCount" INTEGER NOT NULL DEFAULT 0,
+  "maxAttempts" INTEGER NOT NULL DEFAULT 3,
+  "availableAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "lockedAt" TIMESTAMP(3),
+  "startedAt" TIMESTAMP(3),
+  "finishedAt" TIMESTAMP(3),
+  "errorMessage" TEXT,
+  "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "updatedAt" TIMESTAMP(3) NOT NULL,
+
+  CONSTRAINT "SyncJob_pkey" PRIMARY KEY ("id")
+);
+
+ALTER TABLE "SyncJob"
+  ADD CONSTRAINT "SyncJob_userId_fkey"
+  FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+CREATE INDEX "SyncJob_status_availableAt_idx" ON "SyncJob"("status", "availableAt");
+CREATE INDEX "SyncJob_userId_createdAt_idx" ON "SyncJob"("userId", "createdAt");
+CREATE INDEX "SyncJob_userId_backfillYear_idx" ON "SyncJob"("userId", "backfillYear");

prisma/schema.prisma

@@ -20,6 +20,13 @@ enum SyncState {
   FAILED
 }
 
+enum SyncJobStatus {
+  QUEUED
+  RUNNING
+  COMPLETED
+  FAILED
+}
+
 model User {
   id               String         @id @default(cuid())
   supabaseUserId   String         @unique
@@ -30,6 +37,7 @@ model User {
   dailyActivities  DailyActivity[]
   manualHighlights ManualHighlight[]
   publicShares     PublicShare[]
+  syncJobs         SyncJob[]
 
   @@index([email])
 }
@@ -94,3 +102,27 @@ model PublicShare {
 
   @@index([userId])
 }
+
+model SyncJob {
+  id          String       @id @default(cuid())
+  userId       String
+  provider     Provider?
+  from         DateTime?
+  to           DateTime?
+  backfillYear Int?
+  status       SyncJobStatus @default(QUEUED)
+  attemptCount Int          @default(0)
+  maxAttempts  Int          @default(3)
+  availableAt  DateTime     @default(now())
+  lockedAt     DateTime?
+  startedAt    DateTime?
+  finishedAt   DateTime?
+  errorMessage String?
+  createdAt    DateTime     @default(now())
+  updatedAt    DateTime     @updatedAt
+  user         User         @relation(fields: [userId], references: [id], onDelete: Cascade)
+
+  @@index([status, availableAt])
+  @@index([userId, createdAt])
+  @@index([userId, backfillYear])
+}

scripts/process-supabase-queue.ts

@@ -0,0 +1,23 @@
+import "dotenv/config";
+import { getQueueBackend } from "../src/server/queue/queue";
+import { processSupabaseSyncQueue } from "../src/server/queue/supabase-processor";
+
+async function main() {
+  if (getQueueBackend() !== "supabase") {
+    // eslint-disable-next-line no-console
+    console.log("SYNC_QUEUE_BACKEND is not supabase; nothing to process.");
+    return;
+  }
+
+  const result = await processSupabaseSyncQueue(10);
+  // eslint-disable-next-line no-console
+  console.log("Supabase queue processed", result);
+}
+
+main()
+  .then(() => process.exit(0))
+  .catch((error) => {
+    // eslint-disable-next-line no-console
+    console.error(error);
+    process.exit(1);
+  });

src/app/api/integrations/azure-devops/route.ts

@@ -1,5 +1,4 @@
 import { NextResponse } from "next/server";
-import { Provider } from "@prisma/client";
 import { encryptSecret } from "@/server/crypto/encryption";
 import { prisma } from "@/server/db/prisma";
 import { requireAppUser } from "@/server/auth/user";
@@ -19,11 +18,11 @@ export async function POST(request: Request) {
 
   await prisma.integration.upsert({
     where: {
-      userId_provider: { userId: appUser.id, provider: Provider.AZURE_DEVOPS },
+      userId_provider: { userId: appUser.id, provider: "AZURE_DEVOPS" },
     },
     create: {
       userId: appUser.id,
-      provider: Provider.AZURE_DEVOPS,
+      provider: "AZURE_DEVOPS",
       encryptedToken: encrypted.ciphertext,
       tokenIv: encrypted.iv,
       tokenTag: encrypted.tag,

src/app/api/integrations/disconnect/route.ts

@@ -1,5 +1,4 @@
 import { NextResponse } from "next/server";
-import { Provider } from "@prisma/client";
 import { requireAppUser } from "@/server/auth/user";
 import { prisma } from "@/server/db/prisma";
 
@@ -15,7 +14,7 @@ export async function POST(request: Request) {
     provider = String(form.get("provider") ?? "");
   }
 
-  if (provider !== Provider.GITLAB && provider !== Provider.AZURE_DEVOPS && provider !== Provider.GITHUB) {
+  if (provider !== "GITLAB" && provider !== "AZURE_DEVOPS" && provider !== "GITHUB") {
     return NextResponse.json({ error: "Invalid provider" }, { status: 400 });
   }
 

src/app/api/integrations/gitlab/route.ts

@@ -1,5 +1,4 @@
 import { NextResponse } from "next/server";
-import { Provider } from "@prisma/client";
 import { encryptSecret } from "@/server/crypto/encryption";
 import { prisma } from "@/server/db/prisma";
 import { requireAppUser } from "@/server/auth/user";
@@ -18,11 +17,11 @@ export async function POST(request: Request) {
 
   await prisma.integration.upsert({
     where: {
-      userId_provider: { userId: appUser.id, provider: Provider.GITLAB },
+      userId_provider: { userId: appUser.id, provider: "GITLAB" },
     },
     create: {
       userId: appUser.id,
-      provider: Provider.GITLAB,
+      provider: "GITLAB",
       encryptedToken: encrypted.ciphertext,
       tokenIv: encrypted.iv,
       tokenTag: encrypted.tag,

src/app/api/internal/sync/process/route.ts

@@ -0,0 +1,31 @@
+import { NextResponse } from "next/server";
+import { getQueueBackend } from "@/server/queue/queue";
+import { processSupabaseSyncQueue } from "@/server/queue/supabase-processor";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+function isAuthorized(request: Request): boolean {
+  const secret = process.env.CRON_SECRET;
+  if (!secret) return false;
+  const bearer = request.headers.get("authorization");
+  if (bearer === `Bearer ${secret}`) return true;
+  const xSecret = request.headers.get("x-cron-secret");
+  return xSecret === secret;
+}
+
+export async function POST(request: Request) {
+  if (!isAuthorized(request)) {
+    return NextResponse.json({ ok: false, error: "unauthorized" }, { status: 401 });
+  }
+
+  if (getQueueBackend() !== "supabase") {
+    return NextResponse.json({ ok: false, error: "sync_queue_backend_not_supabase" }, { status: 400 });
+  }
+
+  const url = new URL(request.url);
+  const limit = Number(url.searchParams.get("limit") ?? "5");
+  const result = await processSupabaseSyncQueue(Number.isFinite(limit) ? limit : 5);
+
+  return NextResponse.json({ ok: true, ...result });
+}