feat(backend): add BullMQ JobManager framework

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,13 +49,12 @@
         "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",
         "posthog-node": "^5.24.15",
         "prom-client": "^15.1.3",
-        "redlock": "5.0.0-beta.2",
         "simple-git": "^3.36.0",
         "zod": "^3.25.76"
     }

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 });
     }