fix

Author: marcus

src/app/api/sync/events/route.ts

@@ -1,5 +1,6 @@
 import { NextResponse } from "next/server";
 import { requireAppUser } from "@/server/auth/user";
+import { getQueueBackend } from "@/server/queue/queue";
 import { createSyncEventSubscriber, getSyncEventsChannelForUser } from "@/server/queue/sync-events";
 
 export const runtime = "nodejs";
@@ -14,14 +15,27 @@ function encodeSse(data: unknown, event?: string): Uint8Array {
 export async function GET(request: Request) {
   const { appUser } = await requireAppUser();
   const channel = getSyncEventsChannelForUser(appUser.id);
+  const backend = getQueueBackend();
 
   let cleanupRef: (() => Promise<void>) | null = null;
 
   const stream = new ReadableStream<Uint8Array>({
     async start(controller) {
-      const subscriber = createSyncEventSubscriber();
       let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
 
+      if (backend === "supabase") {
+        controller.enqueue(encodeSse({ ok: true, connected: true, mode: "supabase", timestamp: new Date().toISOString() }, "connected"));
+        heartbeatTimer = setInterval(() => {
+          controller.enqueue(encodeSse({ t: Date.now() }, "ping"));
+        }, 20000);
+        cleanupRef = async () => {
+          if (heartbeatTimer) clearInterval(heartbeatTimer);
+        };
+        return;
+      }
+
+      const subscriber = createSyncEventSubscriber();
+
       const onMessage = (incomingChannel: string, message: string) => {
         if (incomingChannel !== channel) return;
         controller.enqueue(encodeSse(message));

src/server/queue/queue.ts

@@ -7,13 +7,16 @@ export type SyncProvider = "GITLAB" | "AZURE_DEVOPS" | "GITHUB";
 export type QueueBackend = "bull" | "supabase";
 
 const backend: QueueBackend = process.env.SYNC_QUEUE_BACKEND === "supabase" ? "supabase" : "bull";
-
-const connection = new IORedis(process.env.REDIS_URL ?? "redis://localhost:6379", {
-  maxRetriesPerRequest: null,
-  enableReadyCheck: false,
-});
-
-export const syncQueue = new Queue("contribution-sync", { connection });
+let queueInstance: Queue | null = null;
+function getSyncQueue(): Queue {
+  if (queueInstance) return queueInstance;
+  const connection = new IORedis(process.env.REDIS_URL ?? "redis://localhost:6379", {
+    maxRetriesPerRequest: null,
+    enableReadyCheck: false,
+  });
+  queueInstance = new Queue("contribution-sync", { connection });
+  return queueInstance;
+}
 
 export type SyncJobOptions = {
   from?: string;
@@ -60,7 +63,7 @@ export async function enqueueUserSync(userId: string, options?: SyncJobOptions)
     return;
   }
 
-  await syncQueue.add("sync-user", { userId, options }, { attempts: 3, backoff: { type: "exponential", delay: 1000 } });
+  await getSyncQueue().add("sync-user", { userId, options }, { attempts: 3, backoff: { type: "exponential", delay: 1000 } });
 }
 
 export async function scheduleNightlySync() {
@@ -69,7 +72,7 @@ export async function scheduleNightlySync() {
     return;
   }
 
-  await syncQueue.add(
+  await getSyncQueue().add(
     "nightly-sync",
     {},
     {
@@ -99,7 +102,7 @@ export async function listBackfillJobsForUser(userId: string, limit = 8): Promis
   }
 
   try {
-    const jobs = await syncQueue.getJobs(["waiting", "active", "completed", "failed", "delayed"], 0, 200);
+    const jobs = await getSyncQueue().getJobs(["waiting", "active", "completed", "failed", "delayed"], 0, 200);
     const collected: BackfillJobView[] = [];
 
     for (const job of jobs) {
@@ -145,7 +148,7 @@ export async function retryBackfillJobForUser(userId: string, jobId: string): Pr
   }
 
   try {
-    const job = await syncQueue.getJob(jobId);
+    const job = await getSyncQueue().getJob(jobId);
     if (!job || job.name !== "sync-user" || !isBackfillJob(job.data) || job.data.userId !== userId) {
       return { ok: false };
     }
@@ -177,7 +180,7 @@ export async function removeBackfillJobForUser(userId: string, jobId: string): P
   }
 
   try {
-    const job = await syncQueue.getJob(jobId);
+    const job = await getSyncQueue().getJob(jobId);
     if (!job || job.name !== "sync-user" || !isBackfillJob(job.data) || job.data.userId !== userId) {
       return { ok: false };
     }
@@ -198,7 +201,7 @@ export async function removeCompletedBackfillJobsForUser(userId: string): Promis
   }
 
   try {
-    const jobs = await syncQueue.getJobs(["completed"], 0, 300);
+    const jobs = await getSyncQueue().getJobs(["completed"], 0, 300);
     let removed = 0;
 
     for (const job of jobs) {

src/server/queue/sync-events.ts

@@ -1,5 +1,6 @@
 import IORedis from "ioredis";
 import { safeLog } from "@/server/crypto/logging";
+import { getQueueBackend } from "@/server/queue/queue";
 
 const REDIS_URL = process.env.REDIS_URL ?? "redis://localhost:6379";
 const CHANNEL_PREFIX = "pow:sync-events:user:";
@@ -20,6 +21,8 @@ function getChannel(userId: string): string {
 }
 
 export async function publishSyncEvent(payload: Omit<SyncEventPayload, "timestamp">): Promise<void> {
+  if (getQueueBackend() === "supabase") return;
+
   const publisher = new IORedis(REDIS_URL, {
     maxRetriesPerRequest: null,
     enableReadyCheck: false,