move config syncing into backend

brendan-kellam · brendan-kellam · commit 2f6f2ac6b8b6 · 2025-10-28T10:31:37.000-07:00
diff --git a/LICENSE.md b/LICENSE.md
@@ -2,7 +2,7 @@ Copyright (c) 2025 Taqla Inc.
 
 Portions of this software are licensed as follows:
 
-- All content that resides under the "ee/", "packages/web/src/ee/", "packages/backend/src/ee/", and "packages/shared/src/ee/" directories of this repository, if these directories exist, is licensed under the license defined in "ee/LICENSE".
+- All content that resides under the "ee/", "packages/web/src/ee/" and "packages/backend/src/ee/" directories of this repository, if these directories exist, is licensed under the license defined in "ee/LICENSE".
 - All third party components incorporated into the Sourcebot Software are licensed under the original license provided by the owner of the applicable component.
 - Content outside of the above mentioned directories or restrictions above is available under the "Functional Source License" as defined below.
 
diff --git a/packages/backend/package.json b/packages/backend/package.json
@@ -40,6 +40,7 @@
         "argparse": "^2.0.1",
         "azure-devops-node-api": "^15.1.1",
         "bullmq": "^5.34.10",
+        "chokidar": "^4.0.3",
         "cross-fetch": "^4.0.0",
         "dotenv": "^16.4.5",
         "express": "^4.21.2",
diff --git a/packages/backend/src/configManager.ts b/packages/backend/src/configManager.ts
@@ -0,0 +1,126 @@
+import { Prisma, PrismaClient } from "@sourcebot/db";
+import { createLogger } from "@sourcebot/logger";
+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";
+
+const logger = createLogger('config-manager');
+
+export class ConfigManager {
+    private watcher: FSWatcher;
+
+    constructor(
+        private db: PrismaClient,
+        private connectionManager: ConnectionManager,
+        configPath: string,
+    ) {
+        this.watcher = chokidar.watch(configPath, {
+            ignoreInitial: true,           // Don't fire events for existing files
+            awaitWriteFinish: {
+                stabilityThreshold: 100,   // File size stable for 100ms
+                pollInterval: 100          // Check every 100ms
+            },
+            atomic: true                   // Handle atomic writes (temp file + rename)
+        });
+
+        this.watcher.on('change', async () => {
+            logger.info(`Config file ${configPath} changed. Syncing config.`);
+            try {
+                await this.syncConfig(configPath);
+            } catch (error) {
+                logger.error(`Failed to sync config: ${error}`);
+            }
+        });
+
+        this.syncConfig(configPath);
+    }
+
+    private syncConfig = async (configPath: string) => {
+        const config = await loadConfig(configPath);
+
+        await this.syncConnections(config.connections);
+        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({
+                    where: {
+                        name_orgId: {
+                            name: key,
+                            orgId: SINGLE_TENANT_ORG_ID,
+                        }
+                    }
+                });
+
+
+                const existingConnectionConfig = existingConnection ? existingConnection.config as unknown as ConnectionConfig : undefined;
+                const connectionNeedsSyncing =
+                    !existingConnection ||
+                    (JSON.stringify(existingConnectionConfig) !== JSON.stringify(newConnectionConfig));
+
+                // Either update the existing connection or create a new one.
+                const connection = existingConnection ?
+                    await this.db.connection.update({
+                        where: {
+                            id: existingConnection.id,
+                        },
+                        data: {
+                            config: newConnectionConfig as unknown as Prisma.InputJsonValue,
+                            isDeclarative: true,
+                        }
+                    }) :
+                    await this.db.connection.create({
+                        data: {
+                            name: key,
+                            config: newConnectionConfig as unknown as Prisma.InputJsonValue,
+                            connectionType: newConnectionConfig.type,
+                            isDeclarative: true,
+                            org: {
+                                connect: {
+                                    id: SINGLE_TENANT_ORG_ID,
+                                }
+                            }
+                        }
+                    });
+
+                if (connectionNeedsSyncing) {
+                    const [jobId] = await this.connectionManager.createJobs([connection]);
+                    logger.info(`Change detected for connection '${key}' (id: ${connection.id}). Created sync job ${jobId}.`);
+                }
+            }
+        }
+
+        // Delete any connections that are no longer in the config.
+        const deletedConnections = await this.db.connection.findMany({
+            where: {
+                isDeclarative: true,
+                name: {
+                    notIn: Object.keys(connections ?? {}),
+                },
+                orgId: SINGLE_TENANT_ORG_ID,
+            }
+        });
+
+        for (const connection of deletedConnections) {
+            logger.info(`Deleting connection with name '${connection.name}'. Connection ID: ${connection.id}`);
+            await this.db.connection.delete({
+                where: {
+                    id: connection.id,
+                }
+            })
+        }
+    }
+
+    public dispose = async () => {
+        await this.watcher.close();
+    }
+}
diff --git a/packages/backend/src/connectionManager.ts b/packages/backend/src/connectionManager.ts
@@ -2,13 +2,15 @@ import * as Sentry from "@sentry/node";
 import { Connection, ConnectionSyncJobStatus, PrismaClient } from "@sourcebot/db";
 import { createLogger } from "@sourcebot/logger";
 import { ConnectionConfig } from "@sourcebot/schemas/v3/connection.type";
-import { loadConfig, syncSearchContexts } from "@sourcebot/shared";
+import { loadConfig } from "@sourcebot/shared";
 import { Job, Queue, ReservedJob, Worker } from "groupmq";
 import { Redis } from 'ioredis';
 import { env } from "./env.js";
 import { compileAzureDevOpsConfig, compileBitbucketConfig, compileGenericGitHostConfig, compileGerritConfig, compileGiteaConfig, compileGithubConfig, compileGitlabConfig } from "./repoCompileUtils.js";
 import { Settings } from "./types.js";
 import { groupmqLifecycleExceptionWrapper } from "./utils.js";
+import { syncSearchContexts } from "./ee/syncSearchContexts.js";
+import { captureEvent } from "./posthog.js";
 
 const LOG_TAG = 'connection-manager';
 const logger = createLogger(LOG_TAG);
@@ -136,6 +138,8 @@ export class ConnectionManager {
                 jobId: job.id,
             });
         }
+
+        return jobs.map(job => job.id);
     }
 
     private async runJob(job: ReservedJob<JobPayload>): Promise<JobResult> {
@@ -297,10 +301,11 @@ export class ConnectionManager {
 
             logger.info(`Connection sync job ${job.id} for connection ${job.data.connectionName} (id: ${job.data.connectionId}) completed`);
 
-            // captureEvent('backend_connection_sync_job_completed', {
-            //     connectionId: connectionId,
-            //     repoCount: result.repoCount,
-            // });
+            const result = job.returnvalue as JobResult;
+            captureEvent('backend_connection_sync_job_completed', {
+                connectionId: connectionId,
+                repoCount: result.repoCount,
+            });
         });
 
     private onJobFailed = async (job: Job<JobPayload>) =>
@@ -332,10 +337,10 @@ export class ConnectionManager {
                 logger.warn(`Failed job ${job.id} for connection ${connection.name} (id: ${connection.id}). Attempt ${attempt} / ${job.opts.attempts}. Retrying.`);
             }
 
-            // captureEvent('backend_connection_sync_job_failed', {
-            //     connectionId: connectionId,
-            //     error: err instanceof BackendException ? err.code : 'UNKNOWN',
-            // });
+            captureEvent('backend_connection_sync_job_failed', {
+                connectionId: job.data.connectionId,
+                error: job.failedReason,
+            });
         });
 
     private onJobStalled = async (jobId: string) =>
@@ -354,6 +359,11 @@ export class ConnectionManager {
             });
 
             logger.error(`Job ${jobId} stalled for connection ${connection.name} (id: ${connection.id})`);
+
+            captureEvent('backend_connection_sync_job_failed', {
+                connectionId: connection.id,
+                error: 'Job stalled',
+            });
         });
 
     private async onWorkerError(error: Error) {
diff --git a/packages/backend/src/constants.ts b/packages/backend/src/constants.ts
@@ -1,6 +1,8 @@
 import { env } from "./env.js";
 import path from "path";
 
+export const SINGLE_TENANT_ORG_ID = 1;
+
 export const PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES = [
     'github',
 ];
diff --git a/packages/backend/src/ee/syncSearchContexts.ts b/packages/backend/src/ee/syncSearchContexts.ts
@@ -1,8 +1,7 @@
 import micromatch from "micromatch";
 import { createLogger } from "@sourcebot/logger";
 import { PrismaClient } from "@sourcebot/db";
-import { getPlan, hasEntitlement } from "../entitlements.js";
-import { SOURCEBOT_SUPPORT_EMAIL } from "../constants.js";
+import { getPlan, hasEntitlement, SOURCEBOT_SUPPORT_EMAIL } from "@sourcebot/shared";
 import { SearchContext } from "@sourcebot/schemas/v3/index.type";
 
 const logger = createLogger('sync-search-contexts');
diff --git a/packages/backend/src/env.ts b/packages/backend/src/env.ts
@@ -47,7 +47,7 @@ export const env = createEnv({
         DEBUG_ENABLE_GROUPMQ_LOGGING: booleanSchema.default('false'),
 
         DATABASE_URL: z.string().url().default("postgresql://postgres:postgres@localhost:5432/postgres"),
-        CONFIG_PATH: z.string().optional(),
+        CONFIG_PATH: z.string(),
 
         CONNECTION_MANAGER_UPSERT_TIMEOUT_MS: numberSchema.default(300000),
         REPO_SYNC_RETRY_BASE_SLEEP_SECONDS: numberSchema.default(60),
@@ -56,6 +56,8 @@ export const env = createEnv({
 
         EXPERIMENT_EE_PERMISSION_SYNC_ENABLED: booleanSchema.default('false'),
         AUTH_EE_GITHUB_BASE_URL: z.string().optional(),
+
+        FORCE_ENABLE_ANONYMOUS_ACCESS: booleanSchema.default('false'),
     },
     runtimeEnv: process.env,
     emptyStringAsUndefined: true,
diff --git a/packages/backend/src/github.ts b/packages/backend/src/github.ts
@@ -256,7 +256,7 @@ const getReposOwnedByUsers = async (users: string[], octokit: Octokit, signal: A
 const getReposForOrgs = async (orgs: string[], octokit: Octokit, signal: AbortSignal, url?: string) => {
     const results = await Promise.allSettled(orgs.map(async (org) => {
         try {
-            logger.info(`Fetching repository info for org ${org}...`);
+            logger.debug(`Fetching repository info for org ${org}...`);
 
             const octokitToUse = await getOctokitWithGithubApp(octokit, org, url, `org ${org}`);
             const { durationMs, data } = await measure(async () => {
@@ -271,7 +271,7 @@ const getReposForOrgs = async (orgs: string[], octokit: Octokit, signal: AbortSi
                 return fetchWithRetry(fetchFn, `org ${org}`, logger);
             });
 
-            logger.info(`Found ${data.length} in org ${org} in ${durationMs}ms.`);
+            logger.debug(`Found ${data.length} in org ${org} in ${durationMs}ms.`);
             return {
                 type: 'valid' as const,
                 data
@@ -305,7 +305,7 @@ const getRepos = async (repoList: string[], octokit: Octokit, signal: AbortSigna
     const results = await Promise.allSettled(repoList.map(async (repo) => {
         try {
             const [owner, repoName] = repo.split('/');
-            logger.info(`Fetching repository info for ${repo}...`);
+            logger.debug(`Fetching repository info for ${repo}...`);
 
             const octokitToUse = await getOctokitWithGithubApp(octokit, owner, url, `repo ${repo}`);
             const { durationMs, data: result } = await measure(async () => {
@@ -320,7 +320,7 @@ const getRepos = async (repoList: string[], octokit: Octokit, signal: AbortSigna
                 return fetchWithRetry(fetchFn, repo, logger);
             });
 
-            logger.info(`Found info for repository ${repo} in ${durationMs}ms`);
+            logger.debug(`Found info for repository ${repo} in ${durationMs}ms`);
             return {
                 type: 'valid' as const,
                 data: [result.data]
diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts
@@ -6,14 +6,15 @@ import { getConfigSettings, hasEntitlement } from '@sourcebot/shared';
 import { existsSync } from 'fs';
 import { mkdir } from 'fs/promises';
 import { Redis } from 'ioredis';
+import { ConfigManager } from "./configManager.js";
 import { ConnectionManager } from './connectionManager.js';
 import { INDEX_CACHE_DIR, REPOS_CACHE_DIR } from './constants.js';
+import { GithubAppManager } from "./ee/githubAppManager.js";
 import { RepoPermissionSyncer } from './ee/repoPermissionSyncer.js';
 import { UserPermissionSyncer } from "./ee/userPermissionSyncer.js";
-import { GithubAppManager } from "./ee/githubAppManager.js";
 import { env } from "./env.js";
-import { RepoIndexManager } from "./repoIndexManager.js";
 import { PromClient } from './promClient.js';
+import { RepoIndexManager } from "./repoIndexManager.js";
 
 
 const logger = createLogger('backend-entrypoint');
@@ -53,6 +54,7 @@ const connectionManager = new ConnectionManager(prisma, settings, redis);
 const repoPermissionSyncer = new RepoPermissionSyncer(prisma, settings, redis);
 const userPermissionSyncer = new UserPermissionSyncer(prisma, settings, redis);
 const repoIndexManager = new RepoIndexManager(prisma, settings, redis, promClient);
+const configManager = new ConfigManager(prisma, connectionManager, env.CONFIG_PATH);
 
 connectionManager.startScheduler();
 repoIndexManager.startScheduler();
@@ -66,11 +68,13 @@ else if (env.EXPERIMENT_EE_PERMISSION_SYNC_ENABLED === 'true' && hasEntitlement(
     userPermissionSyncer.startScheduler();
 }
 
+logger.info('Worker started.');
+
 const cleanup = async (signal: string) => {
     logger.info(`Received ${signal}, cleaning up...`);
 
     const shutdownTimeout = 30000; // 30 seconds
-    
+
     try {
         await Promise.race([
             Promise.all([
@@ -79,8 +83,9 @@ const cleanup = async (signal: string) => {
                 repoPermissionSyncer.dispose(),
                 userPermissionSyncer.dispose(),
                 promClient.dispose(),
+                configManager.dispose(),
             ]),
-            new Promise((_, reject) => 
+            new Promise((_, reject) =>
                 setTimeout(() => reject(new Error('Shutdown timeout')), shutdownTimeout)
             )
         ]);
diff --git a/packages/db/prisma/migrations/20251026194617_add_connection_job_table/migration.sql b/packages/db/prisma/migrations/20251026194617_add_connection_job_table/migration.sql
@@ -0,0 +1,30 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `syncStatus` on the `Connection` table. All the data in the column will be lost.
+  - You are about to drop the column `syncStatusMetadata` on the `Connection` table. All the data in the column will be lost.
+
+*/
+-- CreateEnum
+CREATE TYPE "ConnectionSyncJobStatus" AS ENUM ('PENDING', 'IN_PROGRESS', 'COMPLETED', 'FAILED');
+
+-- AlterTable
+ALTER TABLE "Connection" DROP COLUMN "syncStatus",
+DROP COLUMN "syncStatusMetadata";
+
+-- CreateTable
+CREATE TABLE "ConnectionSyncJob" (
+    "id" TEXT NOT NULL,
+    "status" "ConnectionSyncJobStatus" NOT NULL DEFAULT 'PENDING',
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+    "completedAt" TIMESTAMP(3),
+    "warningMessages" TEXT[],
+    "errorMessage" TEXT,
+    "connectionId" INTEGER NOT NULL,
+
+    CONSTRAINT "ConnectionSyncJob_pkey" PRIMARY KEY ("id")
+);
+
+-- AddForeignKey
+ALTER TABLE "ConnectionSyncJob" ADD CONSTRAINT "ConnectionSyncJob_connectionId_fkey" FOREIGN KEY ("connectionId") REFERENCES "Connection"("id") ON DELETE CASCADE ON UPDATE CASCADE;
diff --git a/packages/db/prisma/migrations/20251026194628_ensure_single_tenant_org/migration.sql b/packages/db/prisma/migrations/20251026194628_ensure_single_tenant_org/migration.sql
@@ -0,0 +1,9 @@
+-- Ensure single tenant organization exists
+INSERT INTO "Org" (id, name, domain, "inviteLinkId", "createdAt", "updatedAt")
+VALUES (1, 'default', '~', gen_random_uuid(), NOW(), NOW())
+ON CONFLICT (id) DO NOTHING;
+
+-- Backfill inviteLinkId for any existing orgs that don't have one
+UPDATE "Org"
+SET "inviteLinkId" = gen_random_uuid()
+WHERE "inviteLinkId" IS NULL;
diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma
@@ -134,11 +134,6 @@ model Connection {
   updatedAt          DateTime             @updatedAt
   repos              RepoToConnection[]
 
-  /// @deprecated
-  syncStatus         ConnectionSyncStatus @default(SYNC_NEEDED)
-  /// @deprecated
-  syncStatusMetadata Json?
-
   // The type of connection (e.g., github, gitlab, etc.)
   connectionType String
 
diff --git a/packages/shared/src/index.server.ts b/packages/shared/src/index.server.ts
diff --git a/packages/web/package.json b/packages/web/package.json
diff --git a/packages/web/src/app/api/(server)/stripe/route.ts b/packages/web/src/app/api/(server)/stripe/route.ts
diff --git a/packages/web/src/initialize.ts b/packages/web/src/initialize.ts
diff --git a/yarn.lock b/yarn.lock
