diff --git a/docs/snippets/schemas/v3/index.schema.mdx b/docs/snippets/schemas/v3/index.schema.mdx index 864359251..0a25f6917 100644 --- a/docs/snippets/schemas/v3/index.schema.mdx +++ b/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", @@ -42,7 +43,8 @@ "maxConnectionSyncJobConcurrency": { "type": "number", "description": "The number of connection sync jobs to run concurrently. Defaults to 8.", - "minimum": 1 + "minimum": 1, + "deprecated": true }, "maxRepoIndexingJobConcurrency": { "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", @@ -226,7 +229,8 @@ "maxConnectionSyncJobConcurrency": { "type": "number", "description": "The number of connection sync jobs to run concurrently. Defaults to 8.", - "minimum": 1 + "minimum": 1, + "deprecated": true }, "maxRepoIndexingJobConcurrency": { "type": "number", diff --git a/packages/backend/src/configManager.ts b/packages/backend/src/configManager.ts index 4d4f61ff6..74f050fb4 100644 --- a/packages/backend/src/configManager.ts +++ b/packages/backend/src/configManager.ts @@ -1,12 +1,13 @@ -import { Prisma, PrismaClient } from "@sourcebot/db"; +import { Prisma } from "@sourcebot/db"; import { createLogger, env } from "@sourcebot/shared"; import { ConnectionConfig } from "@sourcebot/schemas/v3/connection.type"; import { loadConfig } from "@sourcebot/shared"; import chokidar, { FSWatcher } from 'chokidar'; -import { ConnectionManager } from "./connectionManager.js"; import { SINGLE_TENANT_ORG_ID } from "./constants.js"; import { syncSearchContexts } from "./ee/syncSearchContexts.js"; import isEqual from 'fast-deep-equal'; +import { JobManager } from "./types.js"; +import { prisma } from "./prisma.js"; const logger = createLogger('config-manager'); @@ -14,8 +15,7 @@ export class ConfigManager { private watcher: FSWatcher; constructor( - private db: PrismaClient, - private connectionManager: ConnectionManager, + private jobManager: JobManager, configPath: string, ) { this.watcher = chokidar.watch(configPath, { @@ -46,14 +46,13 @@ export class ConfigManager { await syncSearchContexts({ contexts: config.contexts, orgId: SINGLE_TENANT_ORG_ID, - db: this.db, }); } private syncConnections = async (connections?: { [key: string]: ConnectionConfig }) => { if (connections) { for (const [key, newConnectionConfig] of Object.entries(connections)) { - const existingConnection = await this.db.connection.findUnique({ + const existingConnection = await prisma.connection.findUnique({ where: { name_orgId: { name: key, @@ -73,7 +72,7 @@ export class ConfigManager { // Either update the existing connection or create a new one. const connection = existingConnection ? - await this.db.connection.update({ + await prisma.connection.update({ where: { id: existingConnection.id, }, @@ -84,7 +83,7 @@ export class ConfigManager { enforcePermissionsForPublicRepos, } }) : - await this.db.connection.create({ + await prisma.connection.create({ data: { name: key, config: newConnectionConfig as unknown as Prisma.InputJsonValue, @@ -102,13 +101,16 @@ export class ConfigManager { if (connectionNeedsSyncing) { logger.debug(`Change detected for connection '${key}' (id: ${connection.id}). Creating sync job.`); - await this.connectionManager.createJobs([connection]); + await this.jobManager.trigger('connection', { + connectionId: connection.id, + orgId: SINGLE_TENANT_ORG_ID, + }) } } } // Delete any connections that are no longer in the config. - const deletedConnections = await this.db.connection.findMany({ + const deletedConnections = await prisma.connection.findMany({ where: { isDeclarative: true, name: { @@ -120,7 +122,7 @@ export class ConfigManager { for (const connection of deletedConnections) { logger.debug(`Deleting connection with name '${connection.name}'. Connection ID: ${connection.id}`); - await this.db.connection.delete({ + await prisma.connection.delete({ where: { id: connection.id, } diff --git a/packages/backend/src/connectionManager.ts b/packages/backend/src/connectionManager.ts deleted file mode 100644 index c9db057e2..000000000 --- a/packages/backend/src/connectionManager.ts +++ /dev/null @@ -1,410 +0,0 @@ -import * as Sentry from "@sentry/node"; -import { Connection, ConnectionSyncJobStatus, PrismaClient } from "@sourcebot/db"; -import { ConnectionConfig } from "@sourcebot/schemas/v3/connection.type"; -import { createLogger, env, loadConfig } from "@sourcebot/shared"; -import { Job, Queue, Worker } from "bullmq"; -import { Redis } from 'ioredis'; -import { WORKER_STOP_GRACEFUL_TIMEOUT_MS } from "./constants.js"; -import { syncSearchContexts } from "./ee/syncSearchContexts.js"; -import { captureEvent } from "./posthog.js"; -import { PromClient } from "./promClient.js"; -import { compileAzureDevOpsConfig, compileBitbucketConfig, compileGenericGitHostConfig, compileGerritConfig, compileGiteaConfig, compileGithubConfig, compileGitlabConfig } from "./repoCompileUtils.js"; -import { Settings } from "./types.js"; -import { setIntervalAsync } from "./utils.js"; - -const LOG_TAG = 'connection-manager'; -const logger = createLogger(LOG_TAG); -const createJobLogger = (jobId: string) => createLogger(`${LOG_TAG}:job:${jobId}`); -const QUEUE_NAME = 'connection-sync-queue'; - -const CONNECTION_SYNC_TIMEOUT_MS = 1000 * 60 * 60 * 2; // 2 hours - -type JobPayload = { - jobId: string, - connectionId: number, - connectionName: string, - orgId: number, -}; - -type JobResult = { - repoCount: number, -} - -export class ConnectionManager { - private worker: Worker; - private queue: Queue; - private abortController: AbortController; - private interval?: NodeJS.Timeout; - - constructor( - private db: PrismaClient, - private settings: Settings, - redis: Redis, - private promClient: PromClient, - ) { - this.abortController = new AbortController(); - - this.queue = new Queue(QUEUE_NAME, { - connection: redis, - defaultJobOptions: { - removeOnComplete: env.REDIS_REMOVE_ON_COMPLETE, - removeOnFail: env.REDIS_REMOVE_ON_FAIL, - attempts: 2, - }, - }); - - this.worker = new Worker( - QUEUE_NAME, - this.runJob.bind(this), - { - connection: redis, - concurrency: this.settings.maxConnectionSyncJobConcurrency, - maxStalledCount: 1, - } - ); - - this.worker.on('completed', this.onJobCompleted.bind(this)); - this.worker.on('failed', this.onJobMaybeFailed.bind(this)); - this.worker.on('stalled', (jobId) => { - // Just log - BullMQ will automatically retry the job (up to maxStalledCount times). - // If all retries fail, onJobMaybeFailed will handle marking it as failed. - logger.warn(`Job ${jobId} stalled - BullMQ will retry`); - }); - this.worker.on('error', (error) => { - logger.error(`Connection syncer worker error:`, error); - }); - } - - public startScheduler() { - logger.debug('Starting scheduler'); - this.interval = setIntervalAsync(async () => { - const thresholdDate = new Date(Date.now() - this.settings.resyncConnectionIntervalMs); - const timeoutDate = new Date(Date.now() - CONNECTION_SYNC_TIMEOUT_MS); - - const connections = await this.db.connection.findMany({ - where: { - AND: [ - { - OR: [ - { syncedAt: null }, - { syncedAt: { lt: thresholdDate } }, - ] - }, - { - NOT: { - syncJobs: { - some: { - OR: [ - // Don't schedule if there are active jobs that were created within the threshold date. - // This handles the case where a job is stuck in a pending state and will never be scheduled. - { - AND: [ - { status: { in: [ConnectionSyncJobStatus.PENDING, ConnectionSyncJobStatus.IN_PROGRESS] } }, - { createdAt: { gt: timeoutDate } }, - ] - }, - // Don't schedule if there are recent failed jobs (within the threshold date). - { - AND: [ - { status: ConnectionSyncJobStatus.FAILED }, - { completedAt: { gt: thresholdDate } }, - ] - } - ] - } - } - } - } - ] - } - }); - - if (connections.length > 0) { - await this.createJobs(connections); - } - }, this.settings.resyncConnectionPollingIntervalMs); - } - - - public async createJobs(connections: Connection[]) { - const jobs = await this.db.connectionSyncJob.createManyAndReturn({ - data: connections.map(connection => ({ - connectionId: connection.id, - })), - include: { - connection: true, - } - }); - - for (const job of jobs) { - logger.debug(`Scheduling job ${job.id} for connection ${job.connection.name} (id: ${job.connectionId})`); - await this.queue.add( - 'connection-sync-job', - { - jobId: job.id, - connectionId: job.connectionId, - connectionName: job.connection.name, - orgId: job.connection.orgId, - }, - { jobId: job.id } - ); - - this.promClient.pendingConnectionSyncJobs.inc({ connection: job.connection.name }); - } - - return jobs.map(job => job.id); - } - - private async runJob(job: Job): Promise { - const { jobId, connectionName } = job.data; - const logger = createJobLogger(jobId); - logger.debug(`Running connection sync job ${jobId} for connection ${connectionName} (id: ${job.data.connectionId})`); - - const currentStatus = await this.db.connectionSyncJob.findUniqueOrThrow({ - where: { - id: jobId, - }, - select: { - status: true, - } - }); - - // Fail safe: if the job is not PENDING (first run) or IN_PROGRESS (retry), it indicates the job - // is in an invalid state and should be skipped. - if (currentStatus.status !== ConnectionSyncJobStatus.PENDING && currentStatus.status !== ConnectionSyncJobStatus.IN_PROGRESS) { - throw new Error(`Job ${jobId} is not in a valid state. Expected: ${ConnectionSyncJobStatus.PENDING} or ${ConnectionSyncJobStatus.IN_PROGRESS}. Actual: ${currentStatus.status}. Skipping.`); - } - - this.promClient.pendingConnectionSyncJobs.dec({ connection: connectionName }); - this.promClient.activeConnectionSyncJobs.inc({ connection: connectionName }); - - const { connection: { config: rawConnectionConfig, orgId } } = await this.db.connectionSyncJob.update({ - where: { - id: jobId, - }, - data: { - status: ConnectionSyncJobStatus.IN_PROGRESS, - }, - select: { - connection: { - select: { - config: true, - orgId: true, - } - } - }, - }); - - const config = rawConnectionConfig as unknown as ConnectionConfig; - - const result = await (async () => { - switch (config.type) { - case 'github': { - return await compileGithubConfig(config, job.data.connectionId, this.abortController.signal); - } - case 'gitlab': { - return await compileGitlabConfig(config, job.data.connectionId); - } - case 'gitea': { - return await compileGiteaConfig(config, job.data.connectionId); - } - case 'gerrit': { - return await compileGerritConfig(config, job.data.connectionId); - } - case 'bitbucket': { - return await compileBitbucketConfig(config, job.data.connectionId); - } - case 'azuredevops': { - return await compileAzureDevOpsConfig(config, job.data.connectionId); - } - case 'git': { - return await compileGenericGitHostConfig(config, job.data.connectionId); - } - } - })(); - - let { repoData, warnings } = result; - - await this.db.connectionSyncJob.update({ - where: { - id: jobId, - }, - data: { - warningMessages: warnings, - }, - }); - - - // Filter out any duplicates by external_id and external_codeHostUrl. - repoData = repoData.filter((repo, index, self) => { - return index === self.findIndex(r => - r.external_id === repo.external_id && - r.external_codeHostUrl === repo.external_codeHostUrl - ); - }) - - // @note: to handle orphaned Repos we delete all RepoToConnection records for this connection, - // and then recreate them when we upsert the repos. For example, if a repo is no-longer - // captured by the connection's config (e.g., it was deleted, marked archived, etc.), it won't - // appear in the repoData array above, and so the RepoToConnection record won't be re-created. - // Repos that have no RepoToConnection records are considered orphaned and can be deleted. - await this.db.$transaction(async (tx) => { - const deleteStart = performance.now(); - await tx.connection.update({ - where: { - id: job.data.connectionId, - }, - data: { - repos: { - deleteMany: {} - } - } - }); - const deleteDuration = performance.now() - deleteStart; - logger.debug(`Deleted all RepoToConnection records for connection ${connectionName} (id: ${job.data.connectionId}) in ${deleteDuration}ms`); - - const totalUpsertStart = performance.now(); - for (const repo of repoData) { - const upsertStart = performance.now(); - await tx.repo.upsert({ - where: { - external_id_external_codeHostUrl_orgId: { - external_id: repo.external_id, - external_codeHostUrl: repo.external_codeHostUrl, - orgId: orgId, - } - }, - update: repo, - create: repo, - }) - const upsertDuration = performance.now() - upsertStart; - logger.debug(`Upserted repo ${repo.displayName} (id: ${repo.external_id}) in ${upsertDuration}ms`); - } - const totalUpsertDuration = performance.now() - totalUpsertStart; - logger.debug(`Upserted ${repoData.length} repos for connection ${connectionName} (id: ${job.data.connectionId}) in ${totalUpsertDuration}ms`); - }, { timeout: env.CONNECTION_MANAGER_UPSERT_TIMEOUT_MS }); - - return { - repoCount: repoData.length, - }; - } - - - private async onJobCompleted(job: Job, result: JobResult) { - try { - const logger = createJobLogger(job.id!); - const { connectionId, connectionName, orgId } = job.data; - - const { connection } = await this.db.connectionSyncJob.update({ - where: { - id: job.id!, - }, - data: { - status: ConnectionSyncJobStatus.COMPLETED, - completedAt: new Date(), - connection: { - update: { - syncedAt: new Date(), - } - } - }, - select: { - connection: true, - } - }); - - // After a connection has synced, we need to re-sync the org's search contexts as - // there may be new repos that match the search context's include/exclude patterns. - if (env.CONFIG_PATH) { - try { - const config = await loadConfig(env.CONFIG_PATH); - - await syncSearchContexts({ - db: this.db, - orgId, - contexts: config.contexts, - }); - } catch (err) { - logger.error(`Failed to sync search contexts for connection ${connectionId}: ${err}`); - Sentry.captureException(err); - } - } - - logger.debug(`Connection sync job ${job.id} for connection ${job.data.connectionName} (id: ${job.data.connectionId}) completed`); - - this.promClient.activeConnectionSyncJobs.dec({ connection: connectionName }); - this.promClient.connectionSyncJobSuccessTotal.inc({ connection: connectionName }); - - const config = connection.config as unknown as ConnectionConfig; - captureEvent('backend_connection_sync_job_completed', { - connectionId: connectionId, - repoCount: result.repoCount, - type: config.type, - }); - } catch (error) { - Sentry.captureException(error); - logger.error(`Exception thrown while executing lifecycle function \`onJobCompleted\`.`, error); - } - } - - private async onJobMaybeFailed(job: Job | undefined, error: Error) { - try { - if (!job) { - logger.error(`Job failed but job object is undefined. Error: ${error.message}`); - return; - } - const jobLogger = createJobLogger(job.id!); - - // @note: we need to check the job state to determine if the job failed, - // or if it is being retried. - const jobState = await job.getState(); - if (jobState !== 'failed') { - jobLogger.warn(`Job ${job.id} for connection ${job.data.connectionName} (id: ${job.data.connectionId}) failed. Retrying...`); - return; - } - - const { connection } = await this.db.connectionSyncJob.update({ - where: { id: job.id }, - data: { - status: ConnectionSyncJobStatus.FAILED, - completedAt: new Date(), - errorMessage: job.failedReason, - }, - select: { - connection: true, - } - }); - - this.promClient.activeConnectionSyncJobs.dec({ connection: connection.name }); - this.promClient.connectionSyncJobFailTotal.inc({ connection: connection.name }); - - jobLogger.error(`Failed job ${job.id} for connection ${connection.name} (id: ${connection.id}). Reason: ${job.failedReason}`); - - const config = connection.config as unknown as ConnectionConfig; - captureEvent('backend_connection_sync_job_failed', { - connectionId: job.data.connectionId, - type: config.type, - }); - } catch (err) { - Sentry.captureException(err); - logger.error(`Exception thrown while executing lifecycle function \`onJobMaybeFailed\`.`, err); - } - } - - public async dispose() { - if (this.interval) { - clearInterval(this.interval); - } - - // Signal all active jobs to abort - this.abortController.abort(); - - // Wait for worker to finish with timeout - await Promise.race([ - this.worker.close(), - new Promise(resolve => setTimeout(resolve, WORKER_STOP_GRACEFUL_TIMEOUT_MS)) - ]); - - await this.queue.close(); - } -} diff --git a/packages/backend/src/connectionWorkload.ts b/packages/backend/src/connectionWorkload.ts new file mode 100644 index 000000000..a3b965fb1 --- /dev/null +++ b/packages/backend/src/connectionWorkload.ts @@ -0,0 +1,141 @@ +import { QueueSpec, Workload } from "./types.js"; +import { prisma } from "./prisma.js"; +import { ConnectionConfig } from "@sourcebot/schemas/v3/index.type"; +import { compileAzureDevOpsConfig, compileBitbucketConfig, compileGenericGitHostConfig, compileGerritConfig, compileGiteaConfig, compileGithubConfig, compileGitlabConfig } from "./repoCompileUtils.js"; +import { createLogger, env, loadConfig } from "@sourcebot/shared"; +import { syncSearchContexts } from "./ee/syncSearchContexts.js"; +import * as Sentry from "@sentry/node"; + + +const connectionQueueSpec: QueueSpec<'connection'> = { + name: 'connection', + jobOptions: { + attempts: 2, + backoff: { type: 'exponential', delayMs: 5000 }, + keep: { completed: 50, failed: 50 } + }, + dedupKey: (data) => `connection:${data.connectionId}` +} + +// @todo +const logger = createLogger('connection-workflow'); + +export const connectionWorkload: Workload<'connection'> = { + spec: connectionQueueSpec, + concurrency: 2, + process: async ({ + data: { + connectionId, + orgId + }, + signal, + }) => { + const connection = await prisma.connection.findUniqueOrThrow({ + where: { + id: connectionId + } + }); + + const config = connection.config as unknown as ConnectionConfig; + + const result = await (async () => { + switch (config.type) { + case 'github': { + return await compileGithubConfig(config, connectionId, signal); + } + case 'gitlab': { + return await compileGitlabConfig(config, connectionId); + } + case 'gitea': { + return await compileGiteaConfig(config, connectionId); + } + case 'gerrit': { + return await compileGerritConfig(config, connectionId); + } + case 'bitbucket': { + return await compileBitbucketConfig(config, connectionId); + } + case 'azuredevops': { + return await compileAzureDevOpsConfig(config, connectionId); + } + case 'git': { + return await compileGenericGitHostConfig(config, connectionId); + } + } + })(); + + let { repoData } = result; + + // Filter out any duplicates by external_id and external_codeHostUrl. + repoData = repoData.filter((repo, index, self) => { + return index === self.findIndex(r => + r.external_id === repo.external_id && + r.external_codeHostUrl === repo.external_codeHostUrl + ); + }) + + // @note: to handle orphaned Repos we delete all RepoToConnection records for this connection, + // and then recreate them when we upsert the repos. For example, if a repo is no-longer + // captured by the connection's config (e.g., it was deleted, marked archived, etc.), it won't + // appear in the repoData array above, and so the RepoToConnection record won't be re-created. + // Repos that have no RepoToConnection records are considered orphaned and can be deleted. + await prisma.$transaction(async (tx) => { + const deleteStart = performance.now(); + await tx.connection.update({ + where: { + id: connectionId, + }, + data: { + repos: { + deleteMany: {} + } + } + }); + const deleteDuration = performance.now() - deleteStart; + logger.debug(`Deleted all RepoToConnection records for connection ${connection.name} (id: ${connectionId}) in ${deleteDuration}ms`); + + const totalUpsertStart = performance.now(); + for (const repo of repoData) { + const upsertStart = performance.now(); + await tx.repo.upsert({ + where: { + external_id_external_codeHostUrl_orgId: { + external_id: repo.external_id, + external_codeHostUrl: repo.external_codeHostUrl, + orgId: orgId, + } + }, + update: repo, + create: repo, + }) + const upsertDuration = performance.now() - upsertStart; + logger.debug(`Upserted repo ${repo.displayName} (id: ${repo.external_id}) in ${upsertDuration}ms`); + } + const totalUpsertDuration = performance.now() - totalUpsertStart; + logger.debug(`Upserted ${repoData.length} repos for connection ${connection.name} (id: ${connectionId}) in ${totalUpsertDuration}ms`); + }, { timeout: env.CONNECTION_MANAGER_UPSERT_TIMEOUT_MS }); + + await prisma.connection.update({ + where: { + id: connectionId, + }, + data: { + syncedAt: new Date(), + } + }); + + // After a connection has synced, we need to re-sync the org's search contexts as + // there may be new repos that match the search context's include/exclude patterns. + try { + const config = await loadConfig(env.CONFIG_PATH); + + await syncSearchContexts({ + orgId, + contexts: config.contexts, + }); + } catch (err) { + logger.error(`Failed to sync search contexts for connection ${connectionId}: ${err}`); + Sentry.captureException(err); + } + } +} diff --git a/packages/backend/src/ee/syncSearchContexts.test.ts b/packages/backend/src/ee/syncSearchContexts.test.ts index 9aa1decfd..89ffec5d6 100644 --- a/packages/backend/src/ee/syncSearchContexts.test.ts +++ b/packages/backend/src/ee/syncSearchContexts.test.ts @@ -21,6 +21,17 @@ vi.mock('../entitlements.js', () => ({ getPlan: vi.fn(() => Promise.resolve('enterprise')), })); +// `syncSearchContexts` imports the prisma singleton, which builds a real client (and needs +// DATABASE_URL) at import time. Stand in for it with an object that `buildDb` re-populates +// with fresh mocks for each test, so tests stay isolated from one another. +const { prismaMock } = vi.hoisted(() => ({ + prismaMock: {} as Record, +})); + +vi.mock('../prisma.js', () => ({ + prisma: prismaMock, +})); + import { syncSearchContexts } from './syncSearchContexts.js'; // Helper to build a repo record with GitLab topics stored in metadata. @@ -64,20 +75,25 @@ const buildDb = (overrides: Partial<{ connectionFindMany: unknown[]; searchContextFindUnique: unknown; searchContextFindMany: unknown[]; -}> = {}): PrismaClient => ({ - repo: { - findMany: vi.fn().mockResolvedValue(overrides.repoFindMany ?? []), - }, - connection: { - findMany: vi.fn().mockResolvedValue(overrides.connectionFindMany ?? []), - }, - searchContext: { - findUnique: vi.fn().mockResolvedValue(overrides.searchContextFindUnique ?? null), - findMany: vi.fn().mockResolvedValue(overrides.searchContextFindMany ?? []), - upsert: vi.fn().mockResolvedValue({}), - delete: vi.fn().mockResolvedValue({}), - }, -} as unknown as PrismaClient); +}> = {}): PrismaClient => { + // Overwrite every delegate so no mock survives from a previous test. + Object.assign(prismaMock, { + repo: { + findMany: vi.fn().mockResolvedValue(overrides.repoFindMany ?? []), + }, + connection: { + findMany: vi.fn().mockResolvedValue(overrides.connectionFindMany ?? []), + }, + searchContext: { + findUnique: vi.fn().mockResolvedValue(overrides.searchContextFindUnique ?? null), + findMany: vi.fn().mockResolvedValue(overrides.searchContextFindMany ?? []), + upsert: vi.fn().mockResolvedValue({}), + delete: vi.fn().mockResolvedValue({}), + }, + }); + + return prismaMock as unknown as PrismaClient; +}; describe('syncSearchContexts - includeTopics', () => { test('includes repos whose topics match an includeTopics entry', async () => { @@ -90,7 +106,6 @@ describe('syncSearchContexts - includeTopics', () => { myContext: { includeTopics: ['backend'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -109,7 +124,6 @@ describe('syncSearchContexts - includeTopics', () => { myContext: { includeTopics: ['backend'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -127,7 +141,6 @@ describe('syncSearchContexts - includeTopics', () => { myContext: { includeTopics: ['backend', 'core'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -146,7 +159,6 @@ describe('syncSearchContexts - includeTopics', () => { myContext: { includeTopics: ['core-*'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -165,7 +177,6 @@ describe('syncSearchContexts - includeTopics', () => { myContext: { includeTopics: ['backend'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -188,7 +199,6 @@ describe('syncSearchContexts - excludeTopics', () => { }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -209,7 +219,6 @@ describe('syncSearchContexts - excludeTopics', () => { }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -230,7 +239,6 @@ describe('syncSearchContexts - excludeTopics', () => { }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -251,7 +259,6 @@ describe('syncSearchContexts - excludeTopics', () => { }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -274,7 +281,6 @@ describe('syncSearchContexts - includeTopics + excludeTopics combined', () => { }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -298,7 +304,6 @@ describe('syncSearchContexts - includeTopics combined with include globs', () => }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -319,7 +324,6 @@ describe('syncSearchContexts - includeTopics combined with include globs', () => }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -340,7 +344,6 @@ describe('syncSearchContexts - GitHub includeTopics', () => { myContext: { includeTopics: ['backend'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -359,7 +362,6 @@ describe('syncSearchContexts - GitHub includeTopics', () => { myContext: { includeTopics: ['core-*'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -377,7 +379,6 @@ describe('syncSearchContexts - GitHub includeTopics', () => { myContext: { includeTopics: ['backend'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -400,7 +401,6 @@ describe('syncSearchContexts - GitHub excludeTopics', () => { }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -422,7 +422,6 @@ describe('syncSearchContexts - mixed GitHub and GitLab repos', () => { myContext: { includeTopics: ['backend'] }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; @@ -446,7 +445,6 @@ describe('syncSearchContexts - mixed GitHub and GitLab repos', () => { }, }, orgId: 1, - db, }); const upsertCall = vi.mocked(db.searchContext.upsert).mock.calls[0][0]; diff --git a/packages/backend/src/ee/syncSearchContexts.ts b/packages/backend/src/ee/syncSearchContexts.ts index 6a02c9062..a3592a30f 100644 --- a/packages/backend/src/ee/syncSearchContexts.ts +++ b/packages/backend/src/ee/syncSearchContexts.ts @@ -1,20 +1,19 @@ import micromatch from "micromatch"; import { createLogger } from "@sourcebot/shared"; -import { PrismaClient } from "@sourcebot/db"; import { repoMetadataSchema, SOURCEBOT_SUPPORT_EMAIL } from "@sourcebot/shared"; import { hasEntitlement } from "../entitlements.js"; import { SearchContext } from "@sourcebot/schemas/v3/index.type"; +import { prisma } from "../prisma.js"; const logger = createLogger('sync-search-contexts'); interface SyncSearchContextsParams { contexts?: { [key: string]: SearchContext } | undefined; orgId: number; - db: PrismaClient; } export const syncSearchContexts = async (params: SyncSearchContextsParams) => { - const { contexts, orgId, db } = params; + const { contexts, orgId } = params; if (!await hasEntitlement("search-contexts")) { if (contexts) { @@ -25,7 +24,7 @@ export const syncSearchContexts = async (params: SyncSearchContextsParams) => { if (contexts) { for (const [key, newContextConfig] of Object.entries(contexts)) { - const allRepos = await db.repo.findMany({ + const allRepos = await prisma.repo.findMany({ where: { orgId, }, @@ -44,7 +43,7 @@ export const syncSearchContexts = async (params: SyncSearchContextsParams) => { } if(newContextConfig.includeConnections) { - const connections = await db.connection.findMany({ + const connections = await prisma.connection.findMany({ where: { orgId, name: { @@ -101,7 +100,7 @@ export const syncSearchContexts = async (params: SyncSearchContextsParams) => { } if (newContextConfig.excludeConnections) { - const connections = await db.connection.findMany({ + const connections = await prisma.connection.findMany({ where: { orgId, name: { @@ -145,7 +144,7 @@ export const syncSearchContexts = async (params: SyncSearchContextsParams) => { }); } - const currentReposInContext = (await db.searchContext.findUnique({ + const currentReposInContext = (await prisma.searchContext.findUnique({ where: { name_orgId: { name: key, @@ -157,7 +156,7 @@ export const syncSearchContexts = async (params: SyncSearchContextsParams) => { } }))?.repos ?? []; - await db.searchContext.upsert({ + await prisma.searchContext.upsert({ where: { name_orgId: { name: key, @@ -195,7 +194,7 @@ export const syncSearchContexts = async (params: SyncSearchContextsParams) => { } } - const deletedContexts = await db.searchContext.findMany({ + const deletedContexts = await prisma.searchContext.findMany({ where: { name: { notIn: Object.keys(contexts ?? {}), @@ -206,7 +205,7 @@ export const syncSearchContexts = async (params: SyncSearchContextsParams) => { for (const context of deletedContexts) { logger.debug(`Deleting search context with name '${context.name}'. ID: ${context.id}`); - await db.searchContext.delete({ + await prisma.searchContext.delete({ where: { id: context.id, } diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index b97fe248f..fee139a87 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -2,24 +2,20 @@ import "./instrument.js"; import * as Sentry from "@sentry/node"; import { createLogger, env, getConfigSettings } from "@sourcebot/shared"; -import { hasEntitlement } from "./entitlements.js"; -import { prisma } from "./prisma.js"; import 'express-async-errors'; import { existsSync } from 'fs'; import { mkdir } from 'fs/promises'; -import { Api } from "./api.js"; -import { AttachmentPruner } from "./attachmentPruner.js"; import { ConfigManager } from "./configManager.js"; -import { ConnectionManager } from './connectionManager.js'; -import { INDEX_CACHE_DIR, REPOS_CACHE_DIR, SHUTDOWN_SIGNALS } from './constants.js'; -import { AccountPermissionSyncer } from "./ee/accountPermissionSyncer.js"; -import { AuditLogPruner } from "./ee/auditLogPruner.js"; +import { INDEX_CACHE_DIR, REPOS_CACHE_DIR, SHUTDOWN_SIGNALS, SINGLE_TENANT_ORG_ID } from './constants.js'; import { GithubAppManager } from "./ee/githubAppManager.js"; -import { RepoPermissionSyncer } from './ee/repoPermissionSyncer.js'; +import { hasEntitlement } from "./entitlements.js"; +import { BullMQJobManager } from "./jobManager.js"; import { shutdownPosthog } from "./posthog.js"; +import { prisma } from "./prisma.js"; import { PromClient } from './promClient.js'; -import { RepoIndexManager } from "./repoIndexManager.js"; import { redis } from "./redis.js"; +import { QueueSpec, Workload } from "./types.js"; +import { connectionWorkload } from "./connectionWorkload.js"; const logger = createLogger('backend-entrypoint'); @@ -50,38 +46,86 @@ if (await hasEntitlement('github-app')) { await GithubAppManager.getInstance().init(prisma); } -const connectionManager = new ConnectionManager(prisma, settings, redis, promClient); -const repoPermissionSyncer = new RepoPermissionSyncer(prisma, settings, redis); -const accountPermissionSyncer = new AccountPermissionSyncer(prisma, settings, redis); -const repoIndexManager = new RepoIndexManager(prisma, settings, redis, promClient); -const configManager = new ConfigManager(prisma, connectionManager, env.CONFIG_PATH); -const auditLogPruner = new AuditLogPruner(prisma); -const attachmentPruner = new AttachmentPruner(prisma); - -connectionManager.startScheduler(); -await repoIndexManager.startScheduler(); -auditLogPruner.startScheduler(); -attachmentPruner.startScheduler(); - -if (env.PERMISSION_SYNC_ENABLED === 'true' && !await hasEntitlement('permission-syncing')) { - logger.warn('Permission syncing is not supported in current plan. Please contact team@sourcebot.dev for assistance.'); +// const connectionManager = new ConnectionManager(prisma, settings, redis, promClient); +// const repoPermissionSyncer = new RepoPermissionSyncer(prisma, settings, redis); +// const accountPermissionSyncer = new AccountPermissionSyncer(prisma, settings, redis); +// const repoIndexManager = new RepoIndexManager(prisma, settings, redis, promClient); +// const auditLogPruner = new AuditLogPruner(prisma); +// const attachmentPruner = new AttachmentPruner(prisma); + +// connectionManager.startScheduler(); +// await repoIndexManager.startScheduler(); +// auditLogPruner.startScheduler(); +// attachmentPruner.startScheduler(); + +// if (env.PERMISSION_SYNC_ENABLED === 'true' && !await hasEntitlement('permission-syncing')) { +// logger.warn('Permission syncing is not supported in current plan. Please contact team@sourcebot.dev for assistance.'); +// } +// else if (env.PERMISSION_SYNC_ENABLED === 'true' && await hasEntitlement('permission-syncing')) { +// if (env.PERMISSION_SYNC_REPO_DRIVEN_ENABLED === 'true') { +// await repoPermissionSyncer.startScheduler(); +// } +// await accountPermissionSyncer.startScheduler(); +// } + +// const api = new Api( +// promClient, +// prisma, +// connectionManager, +// repoIndexManager, +// accountPermissionSyncer, +// ); + +logger.info('Worker started.'); + +// Background jobs run through the JobManager (BullMQ/Redis as the source of truth). Phase 0 +// wires the framework here in place of the old per-manager pollers; the real workloads +// (repo-index, connection-sync, permission syncers) are ported onto it in subsequent phases. +const jobManager = new BullMQJobManager(redis); + +const cronQueueSpec: QueueSpec<'cron'> = { + name: 'cron', + jobOptions: { + attempts: 2, + backoff: { type: 'exponential', delayMs: 5000 }, + keep: { completed: 50, failed: 50 } + } } -else if (env.PERMISSION_SYNC_ENABLED === 'true' && await hasEntitlement('permission-syncing')) { - if (env.PERMISSION_SYNC_REPO_DRIVEN_ENABLED === 'true') { - await repoPermissionSyncer.startScheduler(); + +const cronWorkload: Workload<'cron'> = { + concurrency: 1, + schedule: { every: '5s' }, + spec: cronQueueSpec, + process: async ({ jobId, trigger }) => { + console.log(`cron ${jobId}`); + + const thresholdDate = new Date(Date.now() - settings.resyncConnectionIntervalMs); + const connections = await prisma.connection.findMany({ + where: { + OR: [ + { syncedAt: null }, + { syncedAt: { lt: thresholdDate }} + ] + } + }); + + await Promise.all(connections.map(async (connection) => { + console.log(`Scheduling work for ${connection.id}`); + await trigger('connection', { + connectionId: connection.id, + orgId: SINGLE_TENANT_ORG_ID, + }) + })) } - await accountPermissionSyncer.startScheduler(); } -const api = new Api( - promClient, - prisma, - connectionManager, - repoIndexManager, - accountPermissionSyncer, -); +jobManager.register(cronWorkload); +jobManager.register(connectionWorkload); + +await jobManager.start(); + +const configManager = new ConfigManager(jobManager, env.CONFIG_PATH); -logger.info('Worker started.'); const listenToShutdownSignals = () => { const signals = SHUTDOWN_SIGNALS; @@ -97,17 +141,18 @@ const listenToShutdownSignals = () => { logger.info(`Received ${signal}, cleaning up...`); - await repoIndexManager.dispose() - await connectionManager.dispose() - await repoPermissionSyncer.dispose() - await accountPermissionSyncer.dispose() - await auditLogPruner.dispose() - await attachmentPruner.dispose() + // await repoIndexManager.dispose() + // await connectionManager.dispose() + // await repoPermissionSyncer.dispose() + // await accountPermissionSyncer.dispose() + // await auditLogPruner.dispose() + // await attachmentPruner.dispose() await configManager.dispose() + await jobManager.stop(); await prisma.$disconnect(); await redis.quit(); - await api.dispose(); + // await api.dispose(); await shutdownPosthog(); logger.info('All workers shut down gracefully'); diff --git a/packages/backend/src/jobManager.test.ts b/packages/backend/src/jobManager.test.ts new file mode 100644 index 000000000..5f01538a5 --- /dev/null +++ b/packages/backend/src/jobManager.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, test, vi } from 'vitest'; + +// The module under test creates a logger at import time; stub it so importing pure helpers +// has no side effects (mirrors repoIndexManager.test.ts). +vi.mock('@sourcebot/shared', () => ({ + createLogger: vi.fn(() => ({ + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + })), +})); + +// Mock the constants module directly so its env-derived cache-dir paths don't load. +vi.mock('./constants.js', () => ({ + WORKER_STOP_GRACEFUL_TIMEOUT_MS: 5000, +})); + +import { normalizeJobState, parseDuration } from './jobManager.js'; + +describe('parseDuration', () => { + test.each([ + ['500ms', 500], + ['30s', 30_000], + ['5m', 300_000], + ['6h', 21_600_000], + ['1d', 86_400_000], + ])('parses %s', (input, expected) => { + expect(parseDuration(input)).toBe(expected); + }); + + test('trims surrounding whitespace', () => { + expect(parseDuration(' 10m ')).toBe(600_000); + }); + + test.each(['', '5', 'm', '5x', '1.5h', '-5m', '5 m'])('throws on malformed "%s"', (input) => { + expect(() => parseDuration(input)).toThrow(); + }); +}); + +describe('normalizeJobState', () => { + test.each([ + 'waiting', + 'active', + 'delayed', + 'completed', + 'failed', + 'paused', + ])('passes through "%s"', (state) => { + expect(normalizeJobState(state)).toBe(state); + }); + + test('collapses prioritized and waiting-children to waiting', () => { + expect(normalizeJobState('prioritized')).toBe('waiting'); + expect(normalizeJobState('waiting-children')).toBe('waiting'); + }); + + test('maps anything unrecognized to unknown', () => { + expect(normalizeJobState('something-else')).toBe('unknown'); + expect(normalizeJobState('unknown')).toBe('unknown'); + }); +}); + diff --git a/packages/backend/src/jobManager.ts b/packages/backend/src/jobManager.ts new file mode 100644 index 000000000..35dfa0af0 --- /dev/null +++ b/packages/backend/src/jobManager.ts @@ -0,0 +1,252 @@ +import * as Sentry from "@sentry/node"; +import { createLogger } from "@sourcebot/shared"; +import { Job, Worker } from "bullmq"; +import { Redis } from "ioredis"; +import { WORKER_STOP_GRACEFUL_TIMEOUT_MS } from "./constants.js"; +import { JobProducer } from "./jobProducer.js"; +import { DataOf, JobDetail, JobManager, QueueCounts, QueueName, Schedule, Workload } from "./types.js"; + +const LOG_TAG = 'job-manager'; +const logger = createLogger(LOG_TAG); + +const DURATION_UNITS_MS: Record = { + ms: 1, + s: 1000, + m: 1000 * 60, + h: 1000 * 60 * 60, + d: 1000 * 60 * 60 * 24, +}; + +export const parseDuration = (value: string): number => { + const match = /^(\d+)(ms|s|m|h|d)$/.exec(value.trim()); + if (!match) { + throw new Error(`Invalid duration "${value}". Expected e.g. "500ms", "30s", "5m", "6h", "1d".`); + } + return Number(match[1]) * DURATION_UNITS_MS[match[2]]; +}; + +export const normalizeJobState = (state: string): JobDetail['state'] => { + switch (state) { + case 'waiting': + case 'active': + case 'delayed': + case 'completed': + case 'failed': + case 'paused': + return state; + case 'prioritized': + case 'waiting-children': + return 'waiting'; + default: + return 'unknown'; + } +}; + +const scheduleToRepeat = (schedule: Schedule) => + 'pattern' in schedule ? { pattern: schedule.pattern } : { every: parseDuration(schedule.every) }; + +export class BullMQJobManager implements JobManager { + private readonly workloads = new Map>(); + private readonly workers = new Map(); + private readonly producer: JobProducer; + private readonly abortController = new AbortController(); + + constructor(private readonly connection: Redis) { + this.producer = new JobProducer(connection); + } + + register(workload: Workload): void { + const name = workload.spec.name; + if (this.workloads.has(name)) { + throw new Error(`Workload "${name}" is already registered`); + } + this.workloads.set(name, workload); + } + + async start(): Promise { + if (this.workloads.size === 0) { + logger.debug('start() called with nothing registered; nothing to do'); + return; + } + + for (const workload of this.workloads.values()) { + await this.startWorkload(workload); + } + + logger.info( + `Started ${this.workloads.size} workload(s) [${[...this.workloads.keys()].join(', ')}]`, + ); + } + + async trigger( + workloadName: TName, + data: DataOf + ): Promise { + const workload = this.workloads.get(workloadName); + if (!workload) { + throw new Error(`Cannot trigger unknown workload "${workloadName}"`); + } + await this.producer.enqueue(workload.spec, data); + } + + async status(workloadName: string): Promise { + this.requireRegistered(workloadName); + const counts = await this.producer.queue(workloadName).getJobCounts( + 'waiting', + 'active', + 'delayed', + 'completed', + 'failed', + 'paused', + 'prioritized', + 'waiting-children', + ); + return { + waiting: counts.waiting ?? 0, + active: counts.active ?? 0, + delayed: counts.delayed ?? 0, + completed: counts.completed ?? 0, + failed: counts.failed ?? 0, + paused: counts.paused ?? 0, + prioritized: counts.prioritized ?? 0, + 'waiting-children': counts['waiting-children'] ?? 0, + }; + } + + async jobDetail(workloadName: string, jobId: string): Promise { + this.requireRegistered(workloadName); + const queue = this.producer.queue(workloadName); + const job = await queue.getJob(jobId); + if (!job) { + return null; + } + + const [state, jobLogs] = await Promise.all([ + job.getState(), + queue.getJobLogs(jobId), + ]); + + const enqueuedAt = job.timestamp; + const startedAt = job.processedOn ?? null; + const finishedAt = job.finishedOn ?? null; + + return { + id: job.id ?? jobId, + name: job.name, + state: normalizeJobState(state), + data: job.data, + attemptsMade: job.attemptsMade, + maxAttempts: job.opts.attempts ?? 1, + result: job.returnvalue ?? null, + failedReason: job.failedReason ?? null, + stacktrace: job.stacktrace ?? [], + logs: jobLogs.logs, + enqueuedAt, + startedAt, + finishedAt, + waitMs: startedAt !== null ? startedAt - enqueuedAt : null, + runMs: startedAt !== null && finishedAt !== null ? finishedAt - startedAt : null, + }; + } + + async stop(): Promise { + this.abortController.abort(); + + await Promise.all([...this.workers.values()].map((worker) => + Promise.race([ + worker.close(), + new Promise((resolve) => setTimeout(resolve, WORKER_STOP_GRACEFUL_TIMEOUT_MS)), + ]), + )); + + await this.producer.close(); + + logger.info('Job manager stopped'); + } + + private async startWorkload(workload: Workload): Promise { + const { spec, concurrency, rateLimit, schedule } = workload; + + const queue = this.producer.queue(spec.name); + + const worker = new Worker( + spec.name, + (job) => workload.process({ + data: job.data, + jobId: job.id ?? '', + attemptsMade: job.attemptsMade, + maxAttempts: job.opts.attempts ?? 1, + signal: this.abortController.signal, + log: async (message) => { await job.log(message); }, + updateProgress: (progress) => job.updateProgress(progress), + trigger: (target, data) => this.trigger(target, data), + }), + { + connection: this.connection, + concurrency, + maxStalledCount: 1, + ...(rateLimit + ? { limiter: { max: rateLimit.max, duration: parseDuration(rateLimit.per) } } + : {}), + }, + ); + + worker.on('failed', (job, error) => { + void this.onWorkloadJobFailed(workload, job, error); + }); + worker.on('error', (error) => { + logger.error(`Worker "${spec.name}" error:`, error); + }); + + this.workers.set(spec.name, worker); + + if (schedule) { + // @note: jobs produced by BullMQ's scheduler bypass the deduplication check that + // `Queue.add` goes through, so a dedup key would be silently ignored here. A + // scheduled workload gets its overlap protection from `concurrency` instead: the + // next tick's job is only created once the current one goes active, so at most one + // run is ever queued behind the one in flight. + await queue.upsertJobScheduler( + `schedule:${spec.name}`, + scheduleToRepeat(schedule), + { + name: spec.name, + opts: { + attempts: spec.jobOptions.attempts, + removeOnComplete: { count: spec.jobOptions.keep.completed }, + removeOnFail: { count: spec.jobOptions.keep.failed }, + }, + }, + ); + } + } + + private async onWorkloadJobFailed( + workload: Workload, + job: Job | undefined, + error: Error, + ): Promise { + if (!job) { + return; + } + const maxAttempts = job.opts.attempts ?? 1; + const isTerminal = job.attemptsMade >= maxAttempts; + if (!isTerminal) { + logger.warn(`Workload "${workload.spec.name}" job ${job.id} failed attempt ${job.attemptsMade}/${maxAttempts}; will retry: ${error.message}`); + return; + } + logger.error(`Workload "${workload.spec.name}" job ${job.id} failed terminally after ${job.attemptsMade} attempt(s): ${error.message}`); + try { + await workload.onTerminalFailure?.(job.data, error); + } catch (hookError) { + Sentry.captureException(hookError); + logger.error(`onTerminalFailure for workload "${workload.spec.name}" threw:`, hookError); + } + } + + private requireRegistered(workloadName: string): void { + if (!this.workloads.has(workloadName)) { + throw new Error(`Workload "${workloadName}" is not registered`); + } + } +} diff --git a/packages/backend/src/jobProducer.ts b/packages/backend/src/jobProducer.ts new file mode 100644 index 000000000..8cd28a2e8 --- /dev/null +++ b/packages/backend/src/jobProducer.ts @@ -0,0 +1,36 @@ +import { Queue } from "bullmq"; +import { Redis } from "ioredis"; +import { DataOf, QueueName, QueueSpec } from "./types.js"; + +export class JobProducer { + private readonly queues = new Map(); + + constructor(private readonly connection: Redis) {} + + queue(name: string): Queue { + let queue = this.queues.get(name); + if (!queue) { + queue = new Queue(name, { connection: this.connection }); + this.queues.set(name, queue); + } + return queue; + } + + async enqueue( + spec: QueueSpec, + data: DataOf + ): Promise { + const dedupKey = spec.dedupKey?.(data); + await this.queue(spec.name).add(spec.name, data, { + ...(dedupKey ? { deduplication: { id: dedupKey } } : {}), + attempts: spec.jobOptions.attempts, + backoff: { type: spec.jobOptions.backoff.type, delay: spec.jobOptions.backoff.delayMs }, + removeOnComplete: { count: spec.jobOptions.keep.completed }, + removeOnFail: { count: spec.jobOptions.keep.failed }, + }); + } + + async close(): Promise { + await Promise.all([...this.queues.values()].map((queue) => queue.close())); + } +} diff --git a/packages/backend/src/types.ts b/packages/backend/src/types.ts index 8803b48b9..66295017e 100644 --- a/packages/backend/src/types.ts +++ b/packages/backend/src/types.ts @@ -24,4 +24,111 @@ export type RepoAuthCredentials = { * credentials for this repo. */ connectionConfig?: ConnectionConfig; -} \ No newline at end of file +} + +export interface QueueRegistry { + 'connection': { + connectionId: number, + orgId: number + }, + 'cron': {} +} + +export type QueueName = keyof QueueRegistry; +export type DataOf = QueueRegistry[TName]; + +export interface ProcessContext { + data: DataOf; + jobId: string; + attemptsMade: number; + maxAttempts: number; + signal: AbortSignal; + log(message: string): Promise; + updateProgress(progress: number | object): Promise; + trigger(workload: T, data: DataOf): Promise; +} + +/** + * A QueueSpec defines the specification for a queue, including + * it's name, deduplication key, and settings. + */ +export interface QueueSpec { + name: TName; + dedupKey?(data: DataOf): string; + jobOptions: { + attempts: number; + backoff: { type: 'fixed' | 'exponential'; delayMs: number }; + keep: { completed: number; failed: number }; + }; +} + +export type Schedule = { every: string } | { pattern: string }; + +/** + * A Workload is a single kind of background work, declared + * as the queue it runs on, the code that processes the job, + * and how much of it may run at once. + * + * Jobs reach a workload's queue in one of two ways: someone calls `trigger`, or - if the + * workload declares a `schedule` - the JobManager enqueues one on that cadence. A sweep is + * just a scheduled workload that carries no payload, and whose `process` scans for work and + * triggers it onto other workloads' queues. + */ +export interface Workload { + spec: QueueSpec; + concurrency: number; + /** + * If set, the JobManager enqueues a job on this cadence rather than waiting for someone to + * `trigger` one. Scheduled jobs carry no payload, so `TData` should be `void`. + */ + schedule?: Schedule; + rateLimit?: { max: number; per: string }; + process(ctx: ProcessContext): Promise; + onTerminalFailure?(data: DataOf, err: Error): Promise; +} + +export interface JobManager { + register(w: Workload): void; + + start(): Promise; + stop(): Promise; + + trigger( + workload: TName, + data: DataOf + ): Promise; + + status(workload: string): Promise; + jobDetail(workload: string, jobId: string): Promise; +} + + + +export interface QueueCounts { + waiting: number; + active: number; + delayed: number; + completed: number; + failed: number; + paused: number; + prioritized?: number; + 'waiting-children'?: number; +} + +export interface JobDetail { + id: string; + name: string; + state: 'waiting' | 'active' | 'delayed' | 'completed' | 'failed' | 'paused' | 'unknown'; + data: TData; + attemptsMade: number; + maxAttempts: number; + result?: TResult | null; + failedReason?: string | null; + stacktrace?: string[]; + logs: string[]; + enqueuedAt: number; + startedAt: number | null; + finishedAt: number | null; + waitMs?: number | null; + runMs?: number | null; +} diff --git a/packages/schemas/src/v3/index.schema.ts b/packages/schemas/src/v3/index.schema.ts index 8c1d64b52..c55c72a05 100644 --- a/packages/schemas/src/v3/index.schema.ts +++ b/packages/schemas/src/v3/index.schema.ts @@ -31,7 +31,8 @@ const schema = { "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", @@ -41,7 +42,8 @@ const schema = { "maxConnectionSyncJobConcurrency": { "type": "number", "description": "The number of connection sync jobs to run concurrently. Defaults to 8.", - "minimum": 1 + "minimum": 1, + "deprecated": true }, "maxRepoIndexingJobConcurrency": { "type": "number", @@ -215,7 +217,8 @@ const schema = { "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", @@ -225,7 +228,8 @@ const schema = { "maxConnectionSyncJobConcurrency": { "type": "number", "description": "The number of connection sync jobs to run concurrently. Defaults to 8.", - "minimum": 1 + "minimum": 1, + "deprecated": true }, "maxRepoIndexingJobConcurrency": { "type": "number", diff --git a/packages/schemas/src/v3/index.type.ts b/packages/schemas/src/v3/index.type.ts index 7fa7f5a17..3bede3def 100644 --- a/packages/schemas/src/v3/index.type.ts +++ b/packages/schemas/src/v3/index.type.ts @@ -100,6 +100,7 @@ export interface Settings { */ resyncConnectionIntervalMs?: number; /** + * @deprecated * The polling rate (in milliseconds) at which the db should be checked for connections that need to be re-synced. Defaults to 1 second. */ resyncConnectionPollingIntervalMs?: number; @@ -108,6 +109,7 @@ export interface Settings { */ reindexRepoPollingIntervalMs?: number; /** + * @deprecated * The number of connection sync jobs to run concurrently. Defaults to 8. */ maxConnectionSyncJobConcurrency?: number; diff --git a/schemas/v3/index.json b/schemas/v3/index.json index 874f9f8d5..6dc6255f1 100644 --- a/schemas/v3/index.json +++ b/schemas/v3/index.json @@ -30,7 +30,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", @@ -40,7 +41,8 @@ "maxConnectionSyncJobConcurrency": { "type": "number", "description": "The number of connection sync jobs to run concurrently. Defaults to 8.", - "minimum": 1 + "minimum": 1, + "deprecated": true }, "maxRepoIndexingJobConcurrency": { "type": "number",