feat(events): stale subscription cleanup — daily cron job in admin worker

Add CleanupStaleSubscriptionsService that finds disabled EventSubscriptions
whose associated task no longer exists in any active worker, and deletes them.
Runs daily at 3 AM UTC via the admin worker cron.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: giovaborgogno

apps/webapp/app/v3/services/adminWorker.server.ts

@@ -8,6 +8,8 @@ import { singleton } from "~/utils/singleton";
 import { tracer } from "../tracer.server";
 import { $replica } from "~/db.server";
 import { RunsBackfillerService } from "../../services/runsBackfiller.server";
+import { CleanupStaleSubscriptionsService } from "./events/cleanupStaleSubscriptions.server";
+import { prisma } from "~/db.server";
 
 function initializeWorker() {
   const redisOptions = {
@@ -26,6 +28,13 @@ function initializeWorker() {
     name: "admin-worker",
     redisOptions,
     catalog: {
+      "admin.cleanupStaleSubscriptions": {
+        schema: z.object({}),
+        visibilityTimeoutMs: 60_000 * 5, // 5 minutes
+        retry: { maxAttempts: 3 },
+        cron: "0 3 * * *", // Daily at 3 AM UTC
+        jitterInMs: 60_000, // 1 minute jitter
+      },
       "admin.backfillRunsToReplication": {
         schema: z.object({
           from: z.coerce.date(),
@@ -50,6 +59,10 @@ function initializeWorker() {
     shutdownTimeoutMs: env.ADMIN_WORKER_SHUTDOWN_TIMEOUT_MS,
     logger: new Logger("AdminWorker", env.ADMIN_WORKER_LOG_LEVEL),
     jobs: {
+      "admin.cleanupStaleSubscriptions": async () => {
+        const service = new CleanupStaleSubscriptionsService(prisma);
+        await service.call();
+      },
       "admin.backfillRunsToReplication": async ({ payload, id }) => {
         if (!runsReplicationInstance) {
           logger.error("Runs replication instance not found");

apps/webapp/app/v3/services/events/cleanupStaleSubscriptions.server.ts

@@ -0,0 +1,58 @@
+import { PrismaClientOrTransaction } from "~/db.server";
+import { logger } from "~/services/logger.server";
+
+/**
+ * Cleans up stale EventSubscriptions — disabled subscriptions whose associated
+ * task no longer exists in any active worker for that environment.
+ */
+export class CleanupStaleSubscriptionsService {
+  constructor(private readonly _prisma: PrismaClientOrTransaction) {}
+
+  async call(): Promise<{ deletedCount: number; scannedCount: number }> {
+    // Find all disabled subscriptions
+    const disabledSubscriptions = await this._prisma.eventSubscription.findMany({
+      where: { enabled: false },
+      select: {
+        id: true,
+        taskSlug: true,
+        projectId: true,
+        environmentId: true,
+      },
+    });
+
+    if (disabledSubscriptions.length === 0) {
+      return { deletedCount: 0, scannedCount: 0 };
+    }
+
+    // For each disabled subscription, check if ANY active worker still has that task
+    const idsToDelete: string[] = [];
+
+    for (const sub of disabledSubscriptions) {
+      const taskExists = await this._prisma.backgroundWorkerTask.findFirst({
+        where: {
+          slug: sub.taskSlug,
+          projectId: sub.projectId,
+          runtimeEnvironmentId: sub.environmentId,
+        },
+        select: { id: true },
+      });
+
+      if (!taskExists) {
+        idsToDelete.push(sub.id);
+      }
+    }
+
+    if (idsToDelete.length > 0) {
+      await this._prisma.eventSubscription.deleteMany({
+        where: { id: { in: idsToDelete } },
+      });
+    }
+
+    logger.info("Cleaned up stale event subscriptions", {
+      deletedCount: idsToDelete.length,
+      scannedCount: disabledSubscriptions.length,
+    });
+
+    return { deletedCount: idsToDelete.length, scannedCount: disabledSubscriptions.length };
+  }
+}