feat(backend): add BullMQ JobManager framework (#1427)

* feat(backend): add BullMQ JobManager framework

Introduces a JobManager over BullMQ: work-queue Workloads with a ProcessContext, CronWorkloads multiplexed onto a shared cron worker with a reconcile() sweep helper, a JobProducer that owns the queues and the deduplicated enqueue path, and a Redis-backed read model (status/jobDetail). Wired into the backend entrypoint with a demo workload.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* wip

* further wip

* further wip

* migrated repo indexing to workload

* further wip

* simplicity: remove much of the web changes

* wip

* add necessary lifecyle hooks

* further wip

* further wip

* migrate attachment & audit log pruning to workload system

* prioritize initial repository indexing

* update retry behaviour

* add locking

* improve logging s.t., we use a log context

* fix: make job scheduler upserts idempotent

* fix: clean up repos from deleted connections

* refactor: rename job scheduler reconciliation

* fix: tolerate missing account permission sync jobs

* fix: address background sync edge cases

* docs: add job manager changelog entry

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: brendan-kellam

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+- Migrated connection syncing, repository indexing, permission syncing, and background pruning from in-process managers and pollers to BullMQ workloads with retries and per-resource execution locking. [#1427](https://github.com/sourcebot-dev/sourcebot/pull/1427)
+
 ### Fixed
 - Upgraded Next.js to 16.3.1 to bound memory retained by high-cardinality dynamic route cache entries. [#1594](https://github.com/sourcebot-dev/sourcebot/pull/1594)
 - Fixed memory leak attributed to CodeMirror allocating objects on heap that were never freed. [#1580](https://github.com/sourcebot-dev/sourcebot/pull/1580)

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,12 +32,14 @@
         "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",
           "description": "The polling rate (in milliseconds) at which the db should be checked for repos that should be re-indexed. Defaults to 1 second.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "maxConnectionSyncJobConcurrency": {
           "type": "number",
@@ -52,7 +54,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,12 +219,14 @@
         "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",
           "description": "The polling rate (in milliseconds) at which the db should be checked for repos that should be re-indexed. Defaults to 1 second.",
-          "minimum": 1
+          "minimum": 1,
+          "deprecated": true
         },
         "maxConnectionSyncJobConcurrency": {
           "type": "number",
@@ -236,7 +241,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 });
     }