accountPermissionSyncer.ts

import * as Sentry from "@sentry/node";
import { PrismaClient, AccountPermissionSyncJobStatus, Account, PermissionSyncSource} from "@sourcebot/db";
import { env, hasEntitlement, createLogger, loadConfig, decryptOAuthToken, PERMISSION_SYNC_SUPPORTED_IDENTITY_PROVIDERS } from "@sourcebot/shared";
import { Job, Queue, Worker } from "bullmq";
import { Redis } from "ioredis";
import {
    createOctokitFromToken,
    getOAuthScopesForAuthenticatedUser as getGitHubOAuthScopesForAuthenticatedUser,
    getReposForAuthenticatedUser,
} from "../github.js";
import {
    createGitLabFromOAuthToken,
    getOAuthScopesForAuthenticatedUser as getGitLabOAuthScopesForAuthenticatedUser,
    getProjectsForAuthenticatedUser,
} from "../gitlab.js";
import { createBitbucketCloudClient, createBitbucketServerClient, getReposForAuthenticatedBitbucketCloudUser, getReposForAuthenticatedBitbucketServerUser } from "../bitbucket.js";
import { Settings } from "../types.js";
import { setIntervalAsync } from "../utils.js";

const LOG_TAG = 'user-permission-syncer';
const logger = createLogger(LOG_TAG);
const createJobLogger = (jobId: string) => createLogger(`${LOG_TAG}:job:${jobId}`);

const QUEUE_NAME = 'accountPermissionSyncQueue';
const POLLING_INTERVAL_MS = 1000;

type AccountPermissionSyncJob = {
    jobId: string;
}

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

    constructor(
        private db: PrismaClient,
        private settings: Settings,
        redis: Redis,
    ) {
        this.queue = new Queue<AccountPermissionSyncJob>(QUEUE_NAME, {
            connection: redis,
        });
        this.worker = new Worker<AccountPermissionSyncJob>(QUEUE_NAME, this.runJob.bind(this), {
            connection: redis,
            concurrency: this.settings.maxAccountPermissionSyncJobConcurrency,
        });
        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 () => {
            const thresholdDate = new Date(Date.now() - this.settings.userDrivenPermissionSyncIntervalMs);

            const accounts = await this.db.account.findMany({
                where: {
                    AND: [
                        {
                            provider: {
                                in: PERMISSION_SYNC_SUPPORTED_IDENTITY_PROVIDERS
                            }
                        },
                        {
                            OR: [
                                { permissionSyncedAt: null },
                                { permissionSyncedAt: { lt: thresholdDate } },
                            ]
                        },
                        {
                            NOT: {
                                permissionSyncJobs: {
                                    some: {
                                        OR: [
                                            // Don't schedule if there are active jobs
                                            {
                                                status: {
                                                    in: [
                                                        AccountPermissionSyncJobStatus.PENDING,
                                                        AccountPermissionSyncJobStatus.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: AccountPermissionSyncJobStatus.FAILED },
                                                    { completedAt: { gt: thresholdDate } },
                                                ]
                                            }
                                        ]
                                    }
                                }
                            }
                        },
                    ]
                }
            });

            await this.schedulePermissionSync(accounts);
        }, POLLING_INTERVAL_MS);
    }

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

    public async schedulePermissionSyncForAccount(account: Account) {
        const [job] = await this.db.accountPermissionSyncJob.createManyAndReturn({
            data: [{ accountId: account.id }],
        });

        await this.queue.add('accountPermissionSyncJob', {
            jobId: job.id,
        }, {
            removeOnComplete: env.REDIS_REMOVE_ON_COMPLETE,
            removeOnFail: env.REDIS_REMOVE_ON_FAIL,
            priority: 1,
        });

        return job.id;
    }

    private async schedulePermissionSync(accounts: Account[]) {
        // @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.accountPermissionSyncJob.createManyAndReturn({
            data: accounts.map(account => ({
                accountId: account.id,
            })),
            include: {
                account: true,
            }
        });

        await this.queue.addBulk(jobs.map((job) => ({
            name: 'accountPermissionSyncJob',
            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.account.permissionSyncedAt === null ? 1 : 2,
            }
        })))
    }

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

        const { account } = await this.db.accountPermissionSyncJob.update({
            where: {
                id,
            },
            data: {
                status: AccountPermissionSyncJobStatus.IN_PROGRESS,
            },
            select: {
                account: {
                    include: {
                        user: true,
                    }
                }
            }
        });

        const config = await loadConfig(env.CONFIG_PATH);

        logger.info(`Syncing permissions for ${account.provider} account (id: ${account.id}) for user ${account.user.email}...`);

        // Decrypt tokens (stored encrypted in the database)
        const accessToken = decryptOAuthToken(account.access_token);

        // Get a list of all repos that the user has access to from all connected accounts.
        const repoIds = await (async () => {
            const aggregatedRepoIds: Set<number> = new Set();

            if (account.provider === 'github') {
                if (!accessToken) {
                    throw new Error(`User '${account.user.email}' does not have an GitHub OAuth access token associated with their GitHub account. Please re-authenticate with GitHub to refresh the token.`);
                }

                // @hack: we don't have a way of identifying specific identity providers in the config file.
                // Instead, we'll use the first connection of type 'github' and hope for the best.
                const baseUrl = Array.from(Object.values(config.connections ?? {}))
                    .find(connection => connection.type === 'github')?.url;

                const { octokit } = await createOctokitFromToken({
                    token: accessToken,
                    url: baseUrl,
                });

                const scopes = await getGitHubOAuthScopesForAuthenticatedUser(octokit, accessToken);

                // Token supports scope introspection (classic PAT or OAuth app token)
                if (scopes !== null) {
                    if (!scopes.includes('repo')) {
                        throw new Error(`OAuth token with scopes [${scopes.join(', ')}] is missing the 'repo' scope required for permission syncing. Please re-authorize with GitHub to grant the required scope.`);
                    }
                }

                // @note: we only care about the private repos since we don't need to build a mapping
                // for public repos.
                // @see: packages/web/src/prisma.ts
                let githubRepos;
                try {
                    githubRepos = await getReposForAuthenticatedUser(/* visibility = */ 'private', octokit);
                } catch (error) {
                    if (error && typeof error === 'object' && 'status' in error) {
                        const status = (error as { status: number }).status;
                        if (status === 401 || status === 403) {
                            throw new Error(
                                `GitHub API returned ${status} error. Your token may have expired or lacks the required permissions. ` +
                                `Please re-authorize with GitHub to grant the necessary access.`
                            );
                        }
                    }
                    throw error;
                }
                const gitHubRepoIds = githubRepos.map(repo => repo.id.toString());

                const repos = await this.db.repo.findMany({
                    where: {
                        external_codeHostType: 'github',
                        external_id: {
                            in: gitHubRepoIds,
                        }
                    }
                });

                repos.forEach(repo => aggregatedRepoIds.add(repo.id));
            } else if (account.provider === 'gitlab') {
                if (!accessToken) {
                    throw new Error(`User '${account.user.email}' does not have a GitLab OAuth access token associated with their GitLab account. Please re-authenticate with GitLab to refresh the token.`);
                }

                // @hack: we don't have a way of identifying specific identity providers in the config file.
                // Instead, we'll use the first connection of type 'gitlab' and hope for the best.
                const baseUrl = Array.from(Object.values(config.connections ?? {}))
                    .find(connection => connection.type === 'gitlab')?.url

                const api = await createGitLabFromOAuthToken({
                    oauthToken: accessToken,
                    url: baseUrl,
                });

                const scopes = await getGitLabOAuthScopesForAuthenticatedUser(api);
                if (!scopes.includes('read_api')) {
                    throw new Error(`OAuth token with scopes [${scopes.join(', ')}] is missing the 'read_api' scope required for permission syncing.`);
                }

                // @note: we only care about the private repos since we don't need to build a
                // mapping for public or internal repos. Note that internal repos are _not_
                // enforced by permission syncing and therefore we don't need to fetch them
                // here.
                // 
                // @see: packages/web/src/prisma.ts
                const gitLabProjectIds = (
                    await getProjectsForAuthenticatedUser('private', api)
                ).map(project => project.id.toString());

                const repos = await this.db.repo.findMany({
                    where: {
                        external_codeHostType: 'gitlab',
                        external_id: {
                            in: gitLabProjectIds,
                        }
                    }
                });

                repos.forEach(repo => aggregatedRepoIds.add(repo.id));
            } else if (account.provider === 'bitbucket-cloud') {
                if (!accessToken) {
                    throw new Error(`User '${account.user.email}' does not have a Bitbucket Cloud OAuth access token associated with their account. Please re-authenticate with Bitbucket Cloud to refresh the token.`);
                }

                // @note: we don't pass a user here since we want to use a bearer token
                // for authentication.
                const client = createBitbucketCloudClient(/* user = */ undefined, accessToken)
                const bitbucketRepos = await getReposForAuthenticatedBitbucketCloudUser(client);
                const bitbucketRepoUuids = bitbucketRepos.map(repo => repo.uuid);

                const repos = await this.db.repo.findMany({
                    where: {
                        external_codeHostType: 'bitbucketCloud',
                        external_id: {
                            in: bitbucketRepoUuids,
                        }
                    }
                });

                repos.forEach(repo => aggregatedRepoIds.add(repo.id));
            } else if (account.provider === 'bitbucket-server') {
                if (!accessToken) {
                    throw new Error(`User '${account.user.email}' does not have a Bitbucket Server OAuth access token associated with their account. Please re-authenticate with Bitbucket Server to refresh the token.`);
                }

                // @hack: we don't have a way of identifying specific identity providers in the config file.
                // Instead, we'll use the first Bitbucket Server connection's URL as the base URL.
                const baseUrl = Array.from(Object.values(config.connections ?? {}))
                    .find(connection => connection.type === 'bitbucket' && connection.deploymentType === 'server')?.url;

                if (!baseUrl) {
                    throw new Error(`No Bitbucket Server connection URL found in config for account ${account.id}`);
                }

                const client = createBitbucketServerClient(baseUrl, /* user = */ undefined, accessToken);
                const serverRepos = await getReposForAuthenticatedBitbucketServerUser(client);
                const serverRepoIds = serverRepos.map(r => r.id);

                const repos = await this.db.repo.findMany({
                    where: {
                        external_codeHostType: 'bitbucketServer',
                        external_id: { in: serverRepoIds },
                    }
                });

                repos.forEach(repo => aggregatedRepoIds.add(repo.id));
            } else {
                throw new Error(`Unsupported code host type: ${account.provider}`);
            }

            return Array.from(aggregatedRepoIds);
        })();

        await this.db.$transaction([
            this.db.account.update({
                where: {
                    id: account.id,
                },
                data: {
                    accessibleRepos: {
                        deleteMany: {},
                    }
                }
            }),
            this.db.accountToRepoPermission.createMany({
                data: repoIds.map(repoId => ({
                    accountId: account.id,
                    repoId,
                    source: PermissionSyncSource.ACCOUNT_DRIVEN,
                })),
                skipDuplicates: true,
            })
        ]);
    }

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

        const { account } = await this.db.accountPermissionSyncJob.update({
            where: {
                id: job.data.jobId,
            },
            data: {
                status: AccountPermissionSyncJobStatus.COMPLETED,
                account: {
                    update: {
                        permissionSyncedAt: new Date(),
                    },
                },
                completedAt: new Date(),
            },
            select: {
                account: {
                    include: {
                        user: true,
                    }
                }
            }
        });

        logger.info(`Permissions synced for ${account.provider} account (id: ${account.id}) for user ${account.user.email}`);
    }

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

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

        const errorMessage = (accountId: string, email: string) => `Account permission sync job failed for account (id: ${accountId}) for user ${email}: ${err.message}`;

        if (job) {
            const { account } = await this.db.accountPermissionSyncJob.update({
                where: {
                    id: job.data.jobId,
                },
                data: {
                    status: AccountPermissionSyncJobStatus.FAILED,
                    completedAt: new Date(),
                    errorMessage: err.message,
                },
                select: {
                    account: {
                        include: {
                            user: true,
                        }
                    }
                }
            });

            logger.error(errorMessage(account.id, account.user.email ?? 'unknown user (email not found)'));
        } else {
            logger.error(errorMessage('unknown account (id not found)', 'unknown user (id not found)'));
        }
    }
}