feat(backend): add BullMQ JobManager framework

CLAUDE.md

@@ -18,6 +18,26 @@ To build a specific package:
 yarn workspace @sourcebot/<package-name> build
 ```
 
+## Backend Workloads
+
+Use the workload system in `packages/backend` for background work. Define the queue payload and default job behavior in the shared queue registry, implement a `Workload`, and register it with the `JobManager`.
+
+### Execution locks
+
+- Key an execution lock by the logical resource being mutated, not by the job ID. Workloads that mutate the same resource must use the exact same lock key. For example, repo indexing and repo cleanup share the per-repo filesystem and search-index lock, while repo permission syncing uses a separate per-repo permission lock.
+- An execution lock serializes work but does not deduplicate it. Multiple jobs for one resource may still be queued and will execute one at a time.
+- The lock lease is extended automatically while work is running. The workload's `AbortSignal` is aborted if extension fails or the worker shuts down.
+- Abortion is cooperative. Call `signal.throwIfAborted()` before side effects and after long-running or external operations so work stops promptly after losing the lock. The signal cannot cancel an operation that has already been submitted.
+- `onStarted` runs after the execution lock is acquired and immediately before `process`. `onCompleted` and `onTerminalFailure` are BullMQ event hooks and run after the processor has returned and released the lock.
+
+### Lifecycle state
+
+- In `onStarted`, upsert the workload-specific job row as `IN_PROGRESS`. If the parent resource tracks a `latest...JobId`, update that pointer in the same database transaction.
+- Completion and terminal-failure hooks must always update their own historical job row by job ID. Do not condition that update on the job still being latest. Every job row should record its actual outcome.
+- `onTerminalFailure` only runs after the job exhausts all retry attempts. Intermediate failures are retried without marking the lifecycle row as terminally failed.
+- If a completion or failure hook publishes state onto the parent resource, use a conditional `updateMany` keyed by both the resource ID and its `latest...JobId`. This prevents an older hook from overwriting state belonging to a newer job after the execution lock has been released.
+- Parent-resource state written inside `process` is already serialized by the execution lock. It does not need a latest-job conditional merely because the resource tracks the latest job ID.
+
 ## File Naming
 
 Files should use camelCase starting with a lowercase letter:

docs/snippets/schemas/v3/index.schema.mdx

@@ -32,7 +32,8 @@
         "resyncConnectionPollingIntervalMs": {
           "type": "number",
           "description": "The polling rate (in milliseconds) at which the db should be checked for connections that need to be re-synced. Defaults to 1 second.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "reindexRepoPollingIntervalMs": {
           "type": "number",
@@ -52,7 +53,8 @@
         "maxRepoGarbageCollectionJobConcurrency": {
           "type": "number",
           "description": "The number of repo GC jobs to run concurrently. Defaults to 8.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "repoGarbageCollectionGracePeriodMs": {
           "type": "number",
@@ -216,7 +218,8 @@
         "resyncConnectionPollingIntervalMs": {
           "type": "number",
           "description": "The polling rate (in milliseconds) at which the db should be checked for connections that need to be re-synced. Defaults to 1 second.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "reindexRepoPollingIntervalMs": {
           "type": "number",
@@ -236,7 +239,8 @@
         "maxRepoGarbageCollectionJobConcurrency": {
           "type": "number",
           "description": "The number of repo GC jobs to run concurrently. Defaults to 8.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "repoGarbageCollectionGracePeriodMs": {
           "type": "number",

packages/backend/package.json

@@ -22,6 +22,9 @@
         "vitest": "^4.1.4"
     },
     "dependencies": {
+        "@bull-board/api": "6.11.2",
+        "@bull-board/express": "6.11.2",
+        "@bull-board/ui": "6.11.2",
         "@coderabbitai/bitbucket": "^1.1.3",
         "@gitbeaker/rest": "^40.5.1",
         "@octokit/app": "^16.1.1",
@@ -35,7 +38,7 @@
         "@types/express": "^5.0.0",
         "argparse": "^2.0.1",
         "azure-devops-node-api": "^15.1.1",
-        "bullmq": "^5.34.10",
+        "bullmq": "^5.81.3",
         "chokidar": "^4.0.3",
         "cross-fetch": "^4.0.0",
         "dotenv": "^16.4.5",
@@ -46,7 +49,7 @@
         "gitea-js": "^1.22.0",
         "glob": "^11.1.0",
         "http-status-codes": "^2.3.0",
-        "ioredis": "^5.4.2",
+        "ioredis": "^5.11.1",
         "lowdb": "^7.0.1",
         "micromatch": "^4.0.8",
         "p-limit": "^7.2.0",

packages/backend/src/api.ts

@@ -1,19 +1,19 @@
+import { createBullBoard } from '@bull-board/api';
+import { BullMQAdapter } from '@bull-board/api/bullMQAdapter.js';
+import { ExpressAdapter } from '@bull-board/express';
+import { Octokit } from '@octokit/rest';
+import * as Sentry from "@sentry/node";
 import { PrismaClient, RepoIndexingJobType } from '@sourcebot/db';
-import * as Sentry from '@sentry/node';
-import { hasEntitlement } from './entitlements.js';
-import { createLogger, doesIdpSupportPermissionSyncing, env } from '@sourcebot/shared';
+import { createLogger, env, JOB_PRIORITIES } from '@sourcebot/shared';
 import express, { NextFunction, Request, Response } from 'express';
 import 'express-async-errors';
 import * as http from "http";
-import { ConnectionManager } from './connectionManager.js';
-import { AccountPermissionSyncer } from './ee/accountPermissionSyncer.js';
+import z from 'zod';
+import { SINGLE_TENANT_ORG_ID } from './constants.js';
+import { isGitHubRateLimitError, isNotFound } from './errors.js';
 import { PromClient } from './promClient.js';
-import { RepoIndexManager } from './repoIndexManager.js';
 import { createGitHubRepoRecord } from './repoCompileUtils.js';
-import { isGitHubRateLimitError, isNotFound } from './errors.js';
-import { Octokit } from '@octokit/rest';
-import { SINGLE_TENANT_ORG_ID } from './constants.js';
-import z from 'zod';
+import type { JobManager } from './types.js';
 
 const logger = createLogger('api');
 
@@ -26,24 +26,27 @@ export class Api {
     constructor(
         promClient: PromClient,
         private prisma: PrismaClient,
-        private connectionManager: ConnectionManager,
-        private repoIndexManager: RepoIndexManager,
-        private accountPermissionSyncer: AccountPermissionSyncer,
+        private jobManager: JobManager,
     ) {
         const app = express();
         app.use(express.json());
         app.use(express.urlencoded({ extended: true }));
 
+        const bullBoardAdapter = new ExpressAdapter();
+        bullBoardAdapter.setBasePath('/admin/queues');
+        createBullBoard({
+            queues: jobManager.getQueues().map(queue => new BullMQAdapter(queue, { readOnlyMode: true })),
+            serverAdapter: bullBoardAdapter,
+        });
+        app.use('/admin/queues', bullBoardAdapter.getRouter());
+
         // Prometheus metrics endpoint
         app.use('/metrics', async (_req: Request, res: Response) => {
             res.set('Content-Type', promClient.registry.contentType);
             const metrics = await promClient.registry.metrics();
             res.end(metrics);
         });
 
-        app.post('/api/sync-connection', this.syncConnection.bind(this));
-        app.post('/api/index-repo', this.indexRepo.bind(this));
-        app.post('/api/trigger-account-permission-sync', this.triggerAccountPermissionSync.bind(this));
         app.post(`/api/experimental/add-github-repo`, this.experimental_addGithubRepo.bind(this));
 
         app.use((error: unknown, _req: Request, _res: Response, next: NextFunction) => {
@@ -53,97 +56,10 @@ export class Api {
 
         this.server = app.listen(PORT, () => {
             logger.debug(`API server is running on port ${PORT}`);
+            logger.debug(`Bull Board is available at ${workerApiUrl.origin}/admin/queues`);
         });
     }
 
-    private async syncConnection(req: Request, res: Response) {
-        const schema = z.object({
-            connectionId: z.number(),
-        }).strict();
-
-        const parsed = schema.safeParse(req.body);
-        if (!parsed.success) {
-            res.status(400).json({ error: parsed.error.message });
-            return;
-        }
-
-        const { connectionId } = parsed.data;
-        const connection = await this.prisma.connection.findUnique({
-            where: {
-                id: connectionId,
-            }
-        });
-
-        if (!connection) {
-            res.status(404).json({ error: 'Connection not found' });
-            return;
-        }
-
-        const [jobId] = await this.connectionManager.createJobs([connection]);
-
-        res.status(200).json({ jobId });
-    }
-
-    private async indexRepo(req: Request, res: Response) {
-        const schema = z.object({
-            repoId: z.number(),
-        }).strict();
-
-        const parsed = schema.safeParse(req.body);
-        if (!parsed.success) {
-            res.status(400).json({ error: parsed.error.message });
-            return;
-        }
-
-        const { repoId } = parsed.data;
-        const repo = await this.prisma.repo.findUnique({
-            where: { id: repoId },
-        });
-
-        if (!repo) {
-            res.status(404).json({ error: 'Repo not found' });
-            return;
-        }
-
-        const [jobId] = await this.repoIndexManager.createJobs([repo], RepoIndexingJobType.INDEX);
-        res.status(200).json({ jobId });
-    }
-
-    private async triggerAccountPermissionSync(req: Request, res: Response) {
-        if (env.PERMISSION_SYNC_ENABLED !== 'true' || !await hasEntitlement('permission-syncing')) {
-            res.status(403).json({ error: 'Permission syncing is not enabled.' });
-            return;
-        }
-
-        const schema = z.object({
-            accountId: z.string(),
-        }).strict();
-
-        const parsed = schema.safeParse(req.body);
-        if (!parsed.success) {
-            res.status(400).json({ error: parsed.error.message });
-            return;
-        }
-
-        const { accountId } = parsed.data;
-        const account = await this.prisma.account.findUnique({
-            where: { id: accountId },
-        });
-
-        if (!account) {
-            res.status(404).json({ error: 'Account not found' });
-            return;
-        }
-
-        if (!doesIdpSupportPermissionSyncing(account.providerType)) {
-            res.status(400).json({ error: `Provider '${account.providerType}' does not support permission syncing.` });
-            return;
-        }
-
-        const jobId = await this.accountPermissionSyncer.schedulePermissionSyncForAccount(account);
-        res.status(200).json({ jobId });
-    }
-
     private async experimental_addGithubRepo(req: Request, res: Response) {
         const schema = z.object({
             owner: z.string(),
@@ -196,7 +112,14 @@ export class Api {
             create: record,
         });
 
-        const [jobId ] = await this.repoIndexManager.createJobs([repo], RepoIndexingJobType.INDEX);
+        const jobId = await this.jobManager.trigger(
+            'repo-index',
+            {
+                repoId: repo.id,
+                type: RepoIndexingJobType.INDEX,
+            },
+            { priority: JOB_PRIORITIES.INTERACTIVE },
+        );
 
         res.status(200).json({ jobId, repoId: repo.id });
     }