user syncing + represent sync job status in a seperate table

Author: brendan-kellam

package.json

@@ -14,6 +14,7 @@
         "watch:mcp": "yarn workspace @sourcebot/mcp build:watch",
         "watch:schemas": "yarn workspace @sourcebot/schemas watch",
         "dev:prisma:migrate:dev": "yarn with-env yarn workspace @sourcebot/db prisma:migrate:dev",
+        "dev:prisma:generate": "yarn with-env yarn workspace @sourcebot/db prisma:generate",
         "dev:prisma:studio": "yarn with-env yarn workspace @sourcebot/db prisma:studio",
         "dev:prisma:migrate:reset": "yarn with-env yarn workspace @sourcebot/db prisma:migrate:reset",
         "dev:prisma:db:push": "yarn with-env yarn workspace @sourcebot/db prisma:db:push",

packages/backend/src/constants.ts

@@ -17,3 +17,7 @@ export const DEFAULT_SETTINGS: Settings = {
     repoIndexTimeoutMs: 1000 * 60 * 60 * 2, // 2 hours
     enablePublicAccess: false // deprected, use FORCE_ENABLE_ANONYMOUS_ACCESS instead
 }
+
+export const PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES = [
+    'github',
+];

packages/backend/src/github.ts

@@ -129,6 +129,17 @@ export const getUserIdsWithReadAccessToRepo = async (owner: string, repo: string
     return collaborators.map(collaborator => collaborator.id.toString());
 }
 
+export const getReposThatAuthenticatedUserHasReadAccessTo = async (octokit: Octokit) => {
+    const fetchFn = () => octokit.paginate(octokit.repos.listForAuthenticatedUser, {
+        per_page: 100,
+        // @todo: do we need to set a visibility to private only?
+        // visibility: 'private'
+    });
+
+    const repos = await fetchWithRetry(fetchFn, `authenticated user`, logger);
+    return repos.map(repo => repo.id.toString());
+}
+
 export const createOctokitFromConfig = async (config: GithubConnectionConfig, orgId: number, db: PrismaClient): Promise<{ octokit: Octokit, isAuthenticated: boolean }> => {
     const hostname = config.url ?
         new URL(config.url).hostname :

packages/backend/src/index.ts

@@ -10,10 +10,11 @@ import path from 'path';
 import { ConnectionManager } from './connectionManager.js';
 import { DEFAULT_SETTINGS } from './constants.js';
 import { env } from "./env.js";
-import { RepoPermissionSyncer } from './permissionSyncer.js';
+import { RepoPermissionSyncer } from './repoPermissionSyncer.js';
 import { PromClient } from './promClient.js';
 import { RepoManager } from './repoManager.js';
 import { AppContext } from "./types.js";
+import { UserPermissionSyncer } from "./userPermissionSyncer.js";
 
 
 const logger = createLogger('backend-entrypoint');
@@ -68,28 +69,34 @@ const settings = await getSettings(env.CONFIG_PATH);
 
 const connectionManager = new ConnectionManager(prisma, settings, redis);
 const repoManager = new RepoManager(prisma, settings, redis, promClient, context);
-const permissionSyncer = new RepoPermissionSyncer(prisma, redis);
+const repoPermissionSyncer = new RepoPermissionSyncer(prisma, redis);
+const userPermissionSyncer = new UserPermissionSyncer(prisma, redis);
 
 await repoManager.validateIndexedReposHaveShards();
 
 const connectionManagerInterval = connectionManager.startScheduler();
 const repoManagerInterval = repoManager.startScheduler();
-const permissionSyncerInterval = env.EXPERIMENT_PERMISSION_SYNC_ENABLED === 'true' ? permissionSyncer.startScheduler() : null;
+const repoPermissionSyncerInterval = env.EXPERIMENT_PERMISSION_SYNC_ENABLED === 'true' ? repoPermissionSyncer.startScheduler() : null;
+const userPermissionSyncerInterval = env.EXPERIMENT_PERMISSION_SYNC_ENABLED === 'true' ? userPermissionSyncer.startScheduler() : null;
 
 
 const cleanup = async (signal: string) => {
     logger.info(`Recieved ${signal}, cleaning up...`);
 
-    if (permissionSyncerInterval) {
-        clearInterval(permissionSyncerInterval);
+    if (userPermissionSyncerInterval) {
+        clearInterval(userPermissionSyncerInterval);
+    }
+    if (repoPermissionSyncerInterval) {
+        clearInterval(repoPermissionSyncerInterval);
     }
 
     clearInterval(connectionManagerInterval);
     clearInterval(repoManagerInterval);
 
     connectionManager.dispose();
     repoManager.dispose();
-    permissionSyncer.dispose();
+    repoPermissionSyncer.dispose();
+    userPermissionSyncer.dispose();
 
     await prisma.$disconnect();
     await redis.quit();

packages/backend/src/permissionSyncer.ts renamed to packages/backend/src/repoPermissionSyncer.ts

@@ -1,5 +1,5 @@
 import * as Sentry from "@sentry/node";
-import { PrismaClient, Repo, RepoPermissionSyncStatus } from "@sourcebot/db";
+import { PrismaClient, Repo, RepoPermissionSyncJobStatus } from "@sourcebot/db";
 import { createLogger } from "@sourcebot/logger";
 import { BitbucketConnectionConfig } from "@sourcebot/schemas/v3/bitbucket.type";
 import { GiteaConnectionConfig } from "@sourcebot/schemas/v3/gitea.type";
@@ -10,16 +10,16 @@ import { Redis } from 'ioredis';
 import { env } from "./env.js";
 import { createOctokitFromConfig, getUserIdsWithReadAccessToRepo } from "./github.js";
 import { RepoWithConnections } from "./types.js";
+import { PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES } from "./constants.js";
 
 type RepoPermissionSyncJob = {
-    repoId: number;
+    jobId: string;
 }
 
 const QUEUE_NAME = 'repoPermissionSyncQueue';
 
-const logger = createLogger('permission-syncer');
+const logger = createLogger('repo-permission-syncer');
 
-const SUPPORTED_CODE_HOST_TYPES = ['github'];
 
 export class RepoPermissionSyncer {
     private queue: Queue<RepoPermissionSyncJob>;
@@ -46,47 +46,55 @@ export class RepoPermissionSyncer {
         return setInterval(async () => {
             // @todo: make this configurable
             const thresholdDate = new Date(Date.now() - 1000 * 60 * 60 * 24);
+
             const repos = await this.db.repo.findMany({
                 // Repos need their permissions to be synced against the code host when...
                 where: {
                     // They belong to a code host that supports permissions syncing
                     AND: [
                         {
                             external_codeHostType: {
-                                in: SUPPORTED_CODE_HOST_TYPES,
+                                in: PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES,
                             }
                         },
-                        // and, they either require a sync (SYNC_NEEDED) or have been in a completed state (SYNCED or FAILED)
-                        // for > some duration (default 24 hours)
                         {
                             OR: [
-                                {
-                                    permissionSyncStatus: RepoPermissionSyncStatus.SYNC_NEEDED
-                                },
-                                {
-                                    AND: [
-                                        {
-                                            OR: [
-                                                { permissionSyncStatus: RepoPermissionSyncStatus.SYNCED },
-                                                { permissionSyncStatus: RepoPermissionSyncStatus.FAILED },
-                                            ]
-                                        },
-                                        {
-                                            OR: [
-                                                { permissionSyncJobLastCompletedAt: null },
-                                                { permissionSyncJobLastCompletedAt: { lt: thresholdDate } }
-                                            ]
-                                        }
-                                    ]
+                                { permissionSyncedAt: null },
+                                { permissionSyncedAt: { lt: thresholdDate } },
+                            ],
+                        },
+                        {
+                            NOT: {
+                                permissionSyncJobs: {
+                                    some: {
+                                        OR: [
+                                            // Don't schedule if there are active jobs
+                                            {
+                                                status: {
+                                                    in: [
+                                                        RepoPermissionSyncJobStatus.PENDING,
+                                                        RepoPermissionSyncJobStatus.IN_PROGRESS,
+                                                    ],
+                                                }
+                                            },
+                                            // Don't schedule if there are recent failed jobs (within the threshold date). Note `gt` is used here since this is a inverse condition.
+                                            {
+                                                AND: [
+                                                    { status: RepoPermissionSyncJobStatus.FAILED },
+                                                    { completedAt: { gt: thresholdDate } },
+                                                ]
+                                            }
+                                        ]
+                                    }
                                 }
-                            ]
+                            }
                         },
                     ]
                 }
             });
 
             await this.schedulePermissionSync(repos);
-        }, 1000 * 30);
+        }, 1000 * 5);
     }
 
     public dispose() {
@@ -96,15 +104,16 @@ export class RepoPermissionSyncer {
 
     private async schedulePermissionSync(repos: Repo[]) {
         await this.db.$transaction(async (tx) => {
-            await tx.repo.updateMany({
-                where: { id: { in: repos.map(repo => repo.id) } },
-                data: { permissionSyncStatus: RepoPermissionSyncStatus.IN_SYNC_QUEUE },
+            const jobs = await tx.repoPermissionSyncJob.createManyAndReturn({
+                data: repos.map(repo => ({
+                    repoId: repo.id,
+                })),
             });
 
-            await this.queue.addBulk(repos.map(repo => ({
+            await this.queue.addBulk(jobs.map((job) => ({
                 name: 'repoPermissionSyncJob',
                 data: {
-                    repoId: repo.id,
+                    jobId: job.id,
                 },
                 opts: {
                     removeOnComplete: env.REDIS_REMOVE_ON_COMPLETE,
@@ -115,21 +124,25 @@ export class RepoPermissionSyncer {
     }
 
     private async runJob(job: Job<RepoPermissionSyncJob>) {
-        const id = job.data.repoId;
-        const repo = await this.db.repo.update({
+        const id = job.data.jobId;
+        const { repo } = await this.db.repoPermissionSyncJob.update({
             where: {
-                id
+                id,
             },
             data: {
-                permissionSyncStatus: RepoPermissionSyncStatus.SYNCING,
+                status: RepoPermissionSyncJobStatus.IN_PROGRESS,
             },
-            include: {
-                connections: {
+            select: {
+                repo: {
                     include: {
-                        connection: true,
-                    },
-                },
-            },
+                        connections: {
+                            include: {
+                                connection: true,
+                            }
+                        }
+                    }
+                }
+            }
         });
 
         if (!repo) {
@@ -171,34 +184,43 @@ export class RepoPermissionSyncer {
             return [];
         })();
 
-        await this.db.repo.update({
-            where: {
-                id: repo.id,
-            },
-            data: {
-                permittedUsers: {
-                    deleteMany: {},
+        await this.db.$transaction([
+            this.db.repo.update({
+                where: {
+                    id: repo.id,
+                },
+                data: {
+                    permittedUsers: {
+                        deleteMany: {},
+                    }
                 }
-            }
-        });
-
-        await this.db.userToRepoPermission.createMany({
-            data: userIds.map(userId => ({
-                userId,
-                repoId: repo.id,
-            })),
-        });
+            }),
+            this.db.userToRepoPermission.createMany({
+                data: userIds.map(userId => ({
+                    userId,
+                    repoId: repo.id,
+                })),
+            })
+        ]);
     }
 
     private async onJobCompleted(job: Job<RepoPermissionSyncJob>) {
-        const repo = await this.db.repo.update({
+        const { repo } = await this.db.repoPermissionSyncJob.update({
             where: {
-                id: job.data.repoId,
+                id: job.data.jobId,
             },
             data: {
-                permissionSyncStatus: RepoPermissionSyncStatus.SYNCED,
-                permissionSyncJobLastCompletedAt: new Date(),
+                status: RepoPermissionSyncJobStatus.COMPLETED,
+                repo: {
+                    update: {
+                        permissionSyncedAt: new Date(),
+                    }
+                },
+                completedAt: new Date(),
             },
+            select: {
+                repo: true
+            }
         });
 
         logger.info(`Permissions synced for repo ${repo.displayName ?? repo.name}`);
@@ -207,21 +229,25 @@ export class RepoPermissionSyncer {
     private async onJobFailed(job: Job<RepoPermissionSyncJob> | undefined, err: Error) {
         Sentry.captureException(err, {
             tags: {
-                repoId: job?.data.repoId,
+                jobId: job?.data.jobId,
                 queue: QUEUE_NAME,
             }
         });
 
-        const errorMessage = (repoName: string) => `Repo permission sync job failed for repo ${repoName}: ${err}`;
+        const errorMessage = (repoName: string) => `Repo permission sync job failed for repo ${repoName}: ${err.message}`;
 
         if (job) {
-            const repo = await this.db.repo.update({
+            const { repo } = await this.db.repoPermissionSyncJob.update({
                 where: {
-                    id: job?.data.repoId,
+                    id: job.data.jobId,
                 },
                 data: {
-                    permissionSyncStatus: RepoPermissionSyncStatus.FAILED,
-                    permissionSyncJobLastCompletedAt: new Date(),
+                    status: RepoPermissionSyncJobStatus.FAILED,
+                    completedAt: new Date(),
+                    errorMessage: err.message,
+                },
+                select: {
+                    repo: true
                 },
             });
             logger.error(errorMessage(repo.displayName ?? repo.name));