repoPermissionSyncer.ts

import * as Sentry from "@sentry/node";
import { PermissionSyncSource, PrismaClient, Repo, RepoPermissionSyncJobStatus } from "@sourcebot/db";
import { createLogger, PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES } from "@sourcebot/shared";
import { env, hasEntitlement } from "@sourcebot/shared";
import { Job, Queue, Worker } from 'bullmq';
import { Redis } from 'ioredis';
import { createOctokitFromToken, getRepoCollaborators, GITHUB_CLOUD_HOSTNAME } from "../github.js";
import { createGitLabFromPersonalAccessToken, getProjectMembers } from "../gitlab.js";
import { createBitbucketCloudClient, createBitbucketServerClient, getExplicitUserPermissionsForCloudRepo, getUserPermissionsForServerRepo } from "../bitbucket.js";
import { repoMetadataSchema } from "@sourcebot/shared";
import { Settings } from "../types.js";
import { getAuthCredentialsForRepo, setIntervalAsync } from "../utils.js";
import { BitbucketConnectionConfig } from "@sourcebot/schemas/v3/index.type";

type RepoPermissionSyncJob = {
    jobId: string;
}

const QUEUE_NAME = 'repoPermissionSyncQueue';
const POLLING_INTERVAL_MS = 1000;
const LOG_TAG = 'repo-permission-syncer';

const logger = createLogger(LOG_TAG);
const createJobLogger = (jobId: string) => createLogger(`${LOG_TAG}:job:${jobId}`);

export class RepoPermissionSyncer {
    private queue: Queue<RepoPermissionSyncJob>;
    private worker: Worker<RepoPermissionSyncJob>;
    private interval?: NodeJS.Timeout;

    constructor(
        private db: PrismaClient,
        private settings: Settings,
        redis: Redis,
    ) {
        this.queue = new Queue<RepoPermissionSyncJob>(QUEUE_NAME, {
            connection: redis,
        });
        this.worker = new Worker<RepoPermissionSyncJob>(QUEUE_NAME, this.runJob.bind(this), {
            connection: redis,
            concurrency: this.settings.maxRepoPermissionSyncJobConcurrency,
        });
        this.worker.on('completed', this.onJobCompleted.bind(this));
        this.worker.on('failed', this.onJobFailed.bind(this));
    }

    public startScheduler() {
        if (!hasEntitlement('permission-syncing')) {
            throw new Error('Permission syncing is not supported in current plan.');
        }

        logger.debug('Starting scheduler');

        this.interval = setIntervalAsync(async () => {
            // @todo: make this configurable
            const thresholdDate = new Date(Date.now() - this.settings.repoDrivenPermissionSyncIntervalMs);

            const repos = await this.db.repo.findMany({
                // Repos need their permissions to be synced against the code host when...
                where: {
                    AND: [
                        // They are not public. Public repositories are always visible to all users, therefore we don't
                        // need to explicitly perform permission syncing for them.
                        // @see: packages/web/src/prisma.ts
                        {
                            isPublic: false
                        },
                        // They belong to a code host that supports permissions syncing
                        {
                            external_codeHostType: {
                                in: PERMISSION_SYNC_SUPPORTED_CODE_HOST_TYPES,
                            }
                        },
                        // They have at least one connection with permission enforcement enabled
                        {
                            connections: {
                                some: {
                                    connection: {
                                        enforcePermissions: true,
                                    }
                                }
                            }
                        },
                        // They have not been synced within the threshold date.
                        {
                            OR: [
                                { permissionSyncedAt: null },
                                { permissionSyncedAt: { lt: thresholdDate } },
                            ],
                        },
                        // There aren't any active or recently failed jobs.
                        {
                            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);
        }, POLLING_INTERVAL_MS);
    }

    public async dispose() {
        if (this.interval) {
            clearInterval(this.interval);
        }
        await this.worker.close(/* force = */ true);
        await this.queue.close();
    }

    private async schedulePermissionSync(repos: Repo[]) {
        // @note: we don't perform this in a transaction because
        // we want to avoid the situation where a job is created and run
        // prior to the transaction being committed.
        const jobs = await this.db.repoPermissionSyncJob.createManyAndReturn({
            data: repos.map(repo => ({
                repoId: repo.id,
            })),
            include: {
                repo: true,
            }
        });

        await this.queue.addBulk(jobs.map((job) => ({
            name: 'repoPermissionSyncJob',
            data: {
                jobId: job.id,
            },
            opts: {
                removeOnComplete: env.REDIS_REMOVE_ON_COMPLETE,
                removeOnFail: env.REDIS_REMOVE_ON_FAIL,
                // Priority 1 (high) for never-synced, Priority 2 (normal) for re-sync
                priority: job.repo.permissionSyncedAt === null ? 1 : 2,
            }
        })))
    }

    private async runJob(job: Job<RepoPermissionSyncJob>) {
        const id = job.data.jobId;
        const logger = createJobLogger(id);

        const { repo } = await this.db.repoPermissionSyncJob.update({
            where: {
                id,
            },
            data: {
                status: RepoPermissionSyncJobStatus.IN_PROGRESS,
            },
            select: {
                repo: {
                    include: {
                        connections: {
                            include: {
                                connection: true,
                            }
                        }
                    }
                }
            }
        });

        if (!repo) {
            throw new Error(`Repo ${id} not found`);
        }

        logger.info(`Syncing permissions for repo ${repo.displayName}...`);

        const credentials = await getAuthCredentialsForRepo(repo, logger);
        if (!credentials) {
            throw new Error(`No credentials found for repo ${id}`);
        }

        const {
            accountIds,
            isPartialSync = false,
        } = await (async (): Promise<{
            accountIds: string[],
            isPartialSync?: boolean
        }> => {
            if (repo.external_codeHostType === 'github') {
                const isGitHubCloud = credentials.hostUrl ? new URL(credentials.hostUrl).hostname === GITHUB_CLOUD_HOSTNAME : true;
                const { octokit } = await createOctokitFromToken({
                    token: credentials.token,
                    url: isGitHubCloud ? undefined : credentials.hostUrl,
                });

                // @note: this is a bit of a hack since the displayName _might_ not be set..
                // however, this property was introduced many versions ago and _should_ be set
                // on each connection sync. Let's throw an error just in case.
                if (!repo.displayName) {
                    throw new Error(`Repo ${id} does not have a displayName`);
                }

                const [owner, repoName] = repo.displayName.split('/');

                const collaborators = await getRepoCollaborators(owner, repoName, octokit);
                const githubUserIds = collaborators.map(collaborator => collaborator.id.toString());

                const accounts = await this.db.account.findMany({
                    where: {
                        provider: 'github',
                        providerAccountId: {
                            in: githubUserIds,
                        }
                    },
                });

                return {
                    accountIds: accounts.map(account => account.id),
                }
            } else if (repo.external_codeHostType === 'gitlab') {
                const api = await createGitLabFromPersonalAccessToken({
                    token: credentials.token,
                    url: credentials.hostUrl,
                });

                const projectId = repo.external_id;
                if (!projectId) {
                    throw new Error(`Repo ${id} does not have an external_id`);
                }

                const members = await getProjectMembers(projectId, api);
                const gitlabUserIds = members.map(member => member.id.toString());

                const accounts = await this.db.account.findMany({
                    where: {
                        provider: 'gitlab',
                        providerAccountId: {
                            in: gitlabUserIds,
                        }
                    },
                });

                return {
                    accountIds: accounts.map(account => account.id),
                }
            } else if (repo.external_codeHostType === 'bitbucketCloud') {
                const config = credentials.connectionConfig as BitbucketConnectionConfig | undefined;
                if (!config) {
                    throw new Error(`No connection config found for repo ${id}`);
                }

                const client = createBitbucketCloudClient(config.user, credentials.token);

                const parsedMetadata = repoMetadataSchema.safeParse(repo.metadata);
                if (!parsedMetadata.success) {
                    throw new Error(`Repo ${id} has invalid metadata: ${JSON.stringify(parsedMetadata.error.errors)}`);
                }
                const bitbucketCloudMetadata = parsedMetadata.data.codeHostMetadata?.bitbucketCloud;
                if (!bitbucketCloudMetadata) {
                    throw new Error(`Repo ${id} is missing required Bitbucket Cloud metadata (workspace/repoSlug)`);
                }

                const { workspace, repoSlug } = bitbucketCloudMetadata;

                // @note: The Bitbucket Cloud permissions API only returns users who have been *directly*
                // granted access to this repository. Users who have access via a group added to the repo,
                // via project-level membership, or via a group in a project are NOT captured here.
                // These users will still gain access through user-driven syncing (accountPermissionSyncer),
                // but there may be a delay of up to `userDrivenPermissionSyncIntervalMs` before
                // they see the repository in Sourcebot.
                // @see: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-repositories/#api-repositories-workspace-repo-slug-permissions-config-users-get
                const users = await getExplicitUserPermissionsForCloudRepo(client, workspace, repoSlug);
                const userAccountIds = users.map(u => u.accountId);

                const accounts = await this.db.account.findMany({
                    where: {
                        provider: 'bitbucket-cloud',
                        providerAccountId: {
                            in: userAccountIds,
                        }
                    },
                });

                return {
                    accountIds: accounts.map(account => account.id),
                    // Since we only fetch users who have been explicitly granted access to the repo,
                    // this is a partial sync.
                    isPartialSync: true,
                }
            } else if (repo.external_codeHostType === 'bitbucketServer') {
                const parsedMetadata = repoMetadataSchema.safeParse(repo.metadata);
                if (!parsedMetadata.success) {
                    throw new Error(`Repo ${id} has invalid metadata: ${JSON.stringify(parsedMetadata.error.errors)}`);
                }
                const bitbucketServerMetadata = parsedMetadata.data.codeHostMetadata?.bitbucketServer;
                if (!bitbucketServerMetadata) {
                    throw new Error(`Repo ${id} is missing required Bitbucket Server metadata (projectKey/repoSlug)`);
                }

                const { projectKey, repoSlug } = bitbucketServerMetadata;
                const hostUrl = credentials.hostUrl;

                if (!hostUrl) {
                    throw new Error(`No host URL found for Bitbucket Server repo ${id}`);
                }

                // @note: This covers users with direct repo-level and project-level permissions.
                // Users with access only via groups are NOT captured here. Those users will
                // still gain access through account-driven syncing (accountPermissionSyncer).
                const client = createBitbucketServerClient(hostUrl, /* user = */ undefined, credentials.token);
                const users = await getUserPermissionsForServerRepo(client, projectKey, repoSlug);
                const userIds = users.map(u => u.userId);

                const accounts = await this.db.account.findMany({
                    where: {
                        provider: 'bitbucket-server',
                        providerAccountId: { in: userIds },
                    }
                });

                return {
                    accountIds: accounts.map(account => account.id),
                    isPartialSync: true,
                }
            }

            throw new Error(`Unsupported code host type: ${repo.external_codeHostType}`);
        })();

        await this.db.$transaction([
            this.db.repo.update({
                where: {
                    id: repo.id,
                },
                data: {
                    permittedAccounts: {
                        // @note: if this is a partial sync, we only want to delete the repo-driven permissions
                        // since we don't want to overwrite the account-driven permissions.
                        deleteMany: isPartialSync ? {
                            source: PermissionSyncSource.REPO_DRIVEN,
                        } : {},
                    }
                }
            }),
            this.db.accountToRepoPermission.createMany({
                data: accountIds.map(accountId => ({
                    accountId,
                    repoId: repo.id,
                    source: PermissionSyncSource.REPO_DRIVEN,
                })),
                skipDuplicates: true,
            })
        ]);
    }

    private async onJobCompleted(job: Job<RepoPermissionSyncJob>) {
        const logger = createJobLogger(job.data.jobId);

        const { repo } = await this.db.repoPermissionSyncJob.update({
            where: {
                id: job.data.jobId,
            },
            data: {
                status: RepoPermissionSyncJobStatus.COMPLETED,
                repo: {
                    update: {
                        permissionSyncedAt: new Date(),
                    }
                },
                completedAt: new Date(),
            },
            select: {
                repo: true
            }
        });

        logger.info(`Permissions synced for repo ${repo.displayName ?? repo.name}`);
    }

    private async onJobFailed(job: Job<RepoPermissionSyncJob> | undefined, err: Error) {
        const logger = createJobLogger(job?.data.jobId ?? 'unknown');

        Sentry.captureException(err, {
            tags: {
                jobId: job?.data.jobId,
                queue: QUEUE_NAME,
            }
        });

        const errorMessage = (repoName: string) => `Repo permission sync job failed for repo ${repoName}: ${err.message}`;

        if (job) {
            const { repo } = await this.db.repoPermissionSyncJob.update({
                where: {
                    id: job.data.jobId,
                },
                data: {
                    status: RepoPermissionSyncJobStatus.FAILED,
                    completedAt: new Date(),
                    errorMessage: err.message,
                },
                select: {
                    repo: true
                },
            });
            logger.error(errorMessage(repo.displayName ?? repo.name));
        } else {
            logger.error(errorMessage('unknown repo (id not found)'));
        }
    }
}