Skip to content

Commit 0a3a63c

Browse files
brendan-kellambrendan-kellam
committed
permission syncer
1 parent b9a91c2 commit 0a3a63c

File tree

8 files changed

+322
-142
lines changed

8 files changed

+322
-142
lines changed

‎packages/backend/src/connectionManager.ts‎

Lines changed: 4 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,6 @@ import { env } from "./env.js";
1111
import * as Sentry from "@sentry/node";
1212
import { loadConfig, syncSearchContexts } from "@sourcebot/shared";
1313

14-
interface IConnectionManager {
15-
scheduleConnectionSync: (connection: Connection) => Promise<void>;
16-
registerPollingCallback: () => void;
17-
dispose: () => void;
18-
}
19-
2014
const QUEUE_NAME = 'connectionSyncQueue';
2115

2216
type JobPayload = {
@@ -30,7 +24,7 @@ type JobResult = {
3024
repoCount: number,
3125
}
3226

33-
export class ConnectionManager implements IConnectionManager {
27+
export class ConnectionManager {
3428
private worker: Worker;
3529
private queue: Queue<JobPayload>;
3630
private logger = createLogger('connection-manager');
@@ -75,8 +69,9 @@ export class ConnectionManager implements IConnectionManager {
7569
});
7670
}
7771

78-
public async registerPollingCallback() {
79-
setInterval(async () => {
72+
public startScheduler() {
73+
this.logger.debug('Starting scheduler');
74+
return setInterval(async () => {
8075
const thresholdDate = new Date(Date.now() - this.settings.resyncConnectionIntervalMs);
8176
const connections = await this.db.connection.findMany({
8277
where: {

‎packages/backend/src/env.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,8 @@ export const env = createEnv({
5252
REPO_SYNC_RETRY_BASE_SLEEP_SECONDS: numberSchema.default(60),
5353

5454
GITLAB_CLIENT_QUERY_TIMEOUT_SECONDS: numberSchema.default(60 * 10),
55+
56+
EXPERIMENT_PERMISSION_SYNC_ENABLED: booleanSchema.default("false"),
5557
},
5658
runtimeEnv: process.env,
5759
emptyStringAsUndefined: true,

‎packages/backend/src/github.ts‎

Lines changed: 49 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -30,35 +30,21 @@ export type OctokitRepository = {
3030
size?: number,
3131
owner: {
3232
avatar_url: string,
33+
login: string,
3334
}
3435
}
3536

3637
const isHttpError = (error: unknown, status: number): boolean => {
37-
return error !== null
38+
return error !== null
3839
&& typeof error === 'object'
39-
&& 'status' in error
40+
&& 'status' in error
4041
&& error.status === status;
4142
}
4243

4344
export const getGitHubReposFromConfig = async (config: GithubConnectionConfig, orgId: number, db: PrismaClient, signal: AbortSignal) => {
44-
const hostname = config.url ?
45-
new URL(config.url).hostname :
46-
GITHUB_CLOUD_HOSTNAME;
47-
48-
const token = config.token ?
49-
await getTokenFromConfig(config.token, orgId, db, logger) :
50-
hostname === GITHUB_CLOUD_HOSTNAME ?
51-
env.FALLBACK_GITHUB_CLOUD_TOKEN :
52-
undefined;
53-
54-
const octokit = new Octokit({
55-
auth: token,
56-
...(config.url ? {
57-
baseUrl: `${config.url}/api/v3`
58-
} : {}),
59-
});
45+
const { octokit, isAuthenticated } = await createOctokitFromConfig(config, orgId, db);
6046

61-
if (token) {
47+
if (isAuthenticated) {
6248
try {
6349
await octokit.rest.users.getAuthenticated();
6450
} catch (error) {
@@ -127,16 +113,51 @@ export const getGitHubReposFromConfig = async (config: GithubConnectionConfig, o
127113
logger.debug(`Found ${repos.length} total repositories.`);
128114

129115
return {
130-
validRepos: repos,
116+
validRepos: repos,
131117
notFound,
132118
};
133119
}
134120

121+
export const getUserIdsWithReadAccessToRepo = async (owner: string, repo: string, octokit: Octokit) => {
122+
const fetchFn = () => octokit.paginate(octokit.repos.listCollaborators, {
123+
owner,
124+
repo,
125+
per_page: 100,
126+
});
127+
128+
const collaborators = await fetchWithRetry(fetchFn, `repo ${owner}/${repo}`, logger);
129+
return collaborators.map(collaborator => collaborator.id.toString());
130+
}
131+
132+
export const createOctokitFromConfig = async (config: GithubConnectionConfig, orgId: number, db: PrismaClient): Promise<{ octokit: Octokit, isAuthenticated: boolean }> => {
133+
const hostname = config.url ?
134+
new URL(config.url).hostname :
135+
GITHUB_CLOUD_HOSTNAME;
136+
137+
const token = config.token ?
138+
await getTokenFromConfig(config.token, orgId, db, logger) :
139+
hostname === GITHUB_CLOUD_HOSTNAME ?
140+
env.FALLBACK_GITHUB_CLOUD_TOKEN :
141+
undefined;
142+
143+
const octokit = new Octokit({
144+
auth: token,
145+
...(config.url ? {
146+
baseUrl: `${config.url}/api/v3`
147+
} : {}),
148+
});
149+
150+
return {
151+
octokit,
152+
isAuthenticated: !!token,
153+
};
154+
}
155+
135156
export const shouldExcludeRepo = ({
136157
repo,
137158
include,
138159
exclude
139-
} : {
160+
}: {
140161
repo: OctokitRepository,
141162
include?: {
142163
topics?: GithubConnectionConfig['topics']
@@ -156,23 +177,23 @@ export const shouldExcludeRepo = ({
156177
reason = `\`exclude.forks\` is true`;
157178
return true;
158179
}
159-
180+
160181
if (!!exclude?.archived && !!repo.archived) {
161182
reason = `\`exclude.archived\` is true`;
162183
return true;
163184
}
164-
185+
165186
if (exclude?.repos) {
166187
if (micromatch.isMatch(repoName, exclude.repos)) {
167188
reason = `\`exclude.repos\` contains ${repoName}`;
168189
return true;
169190
}
170191
}
171-
192+
172193
if (exclude?.topics) {
173194
const configTopics = exclude.topics.map(topic => topic.toLowerCase());
174195
const repoTopics = repo.topics ?? [];
175-
196+
176197
const matchingTopics = repoTopics.filter((topic) => micromatch.isMatch(topic, configTopics));
177198
if (matchingTopics.length > 0) {
178199
reason = `\`exclude.topics\` matches the following topics: ${matchingTopics.join(', ')}`;
@@ -190,17 +211,17 @@ export const shouldExcludeRepo = ({
190211
return true;
191212
}
192213
}
193-
214+
194215
const repoSizeInBytes = repo.size ? repo.size * 1000 : undefined;
195216
if (exclude?.size && repoSizeInBytes) {
196217
const min = exclude.size.min;
197218
const max = exclude.size.max;
198-
219+
199220
if (min && repoSizeInBytes < min) {
200221
reason = `repo is less than \`exclude.size.min\`=${min} bytes.`;
201222
return true;
202223
}
203-
224+
204225
if (max && repoSizeInBytes > max) {
205226
reason = `repo is greater than \`exclude.size.max\`=${max} bytes.`;
206227
return true;

‎packages/backend/src/index.ts‎

Lines changed: 77 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -1,44 +1,36 @@
11
import "./instrument.js";
22

3-
import * as Sentry from "@sentry/node";
3+
import { PrismaClient } from "@sourcebot/db";
4+
import { createLogger } from "@sourcebot/logger";
5+
import { loadConfig } from '@sourcebot/shared';
46
import { existsSync } from 'fs';
57
import { mkdir } from 'fs/promises';
8+
import { Redis } from 'ioredis';
69
import path from 'path';
7-
import { AppContext } from "./types.js";
8-
import { main } from "./main.js"
9-
import { PrismaClient } from "@sourcebot/db";
10+
import { ConnectionManager } from './connectionManager.js';
11+
import { DEFAULT_SETTINGS } from './constants.js';
1012
import { env } from "./env.js";
11-
import { createLogger } from "@sourcebot/logger";
12-
13-
const logger = createLogger('backend-entrypoint');
13+
import { RepoPermissionSyncer } from './permissionSyncer.js';
14+
import { PromClient } from './promClient.js';
15+
import { RepoManager } from './repoManager.js';
16+
import { AppContext } from "./types.js";
1417

1518

16-
// Register handler for normal exit
17-
process.on('exit', (code) => {
18-
logger.info(`Process is exiting with code: ${code}`);
19-
});
19+
const logger = createLogger('backend-entrypoint');
2020

21-
// Register handlers for abnormal terminations
22-
process.on('SIGINT', () => {
23-
logger.info('Process interrupted (SIGINT)');
24-
process.exit(0);
25-
});
21+
const getSettings = async (configPath?: string) => {
22+
if (!configPath) {
23+
return DEFAULT_SETTINGS;
24+
}
2625

27-
process.on('SIGTERM', () => {
28-
logger.info('Process terminated (SIGTERM)');
29-
process.exit(0);
30-
});
26+
const config = await loadConfig(configPath);
3127

32-
// Register handlers for uncaught exceptions and unhandled rejections
33-
process.on('uncaughtException', (err) => {
34-
logger.error(`Uncaught exception: ${err.message}`);
35-
process.exit(1);
36-
});
28+
return {
29+
...DEFAULT_SETTINGS,
30+
...config.settings,
31+
}
32+
}
3733

38-
process.on('unhandledRejection', (reason, promise) => {
39-
logger.error(`Unhandled rejection at: ${promise}, reason: ${reason}`);
40-
process.exit(1);
41-
});
4234

4335
const cacheDir = env.DATA_CACHE_DIR;
4436
const reposPath = path.join(cacheDir, 'repos');
@@ -59,18 +51,60 @@ const context: AppContext = {
5951

6052
const prisma = new PrismaClient();
6153

62-
main(prisma, context)
63-
.then(async () => {
64-
await prisma.$disconnect();
65-
})
66-
.catch(async (e) => {
67-
logger.error(e);
68-
Sentry.captureException(e);
69-
70-
await prisma.$disconnect();
71-
process.exit(1);
72-
})
73-
.finally(() => {
74-
logger.info("Shutting down...");
75-
});
54+
const redis = new Redis(env.REDIS_URL, {
55+
maxRetriesPerRequest: null
56+
});
57+
redis.ping().then(() => {
58+
logger.info('Connected to redis');
59+
}).catch((err: unknown) => {
60+
logger.error('Failed to connect to redis');
61+
logger.error(err);
62+
process.exit(1);
63+
});
64+
65+
const promClient = new PromClient();
66+
67+
const settings = await getSettings(env.CONFIG_PATH);
68+
69+
const connectionManager = new ConnectionManager(prisma, settings, redis);
70+
const repoManager = new RepoManager(prisma, settings, redis, promClient, context);
71+
const permissionSyncer = new RepoPermissionSyncer(prisma, redis);
72+
73+
await repoManager.validateIndexedReposHaveShards();
74+
75+
const connectionManagerInterval = connectionManager.startScheduler();
76+
const repoManagerInterval = repoManager.startScheduler();
77+
const permissionSyncerInterval = env.EXPERIMENT_PERMISSION_SYNC_ENABLED ? permissionSyncer.startScheduler() : null;
78+
79+
80+
const cleanup = async (signal: string) => {
81+
logger.info(`Recieved ${signal}, cleaning up...`);
82+
83+
if (permissionSyncerInterval) {
84+
clearInterval(permissionSyncerInterval);
85+
}
86+
87+
clearInterval(connectionManagerInterval);
88+
clearInterval(repoManagerInterval);
89+
90+
connectionManager.dispose();
91+
repoManager.dispose();
92+
permissionSyncer.dispose();
7693

94+
await prisma.$disconnect();
95+
await redis.quit();
96+
}
97+
98+
process.on('SIGINT', () => cleanup('SIGINT').finally(() => process.exit(0)));
99+
process.on('SIGTERM', () => cleanup('SIGTERM').finally(() => process.exit(0)));
100+
101+
// Register handlers for uncaught exceptions and unhandled rejections
102+
process.on('uncaughtException', (err) => {
103+
logger.error(`Uncaught exception: ${err.message}`);
104+
cleanup('uncaughtException').finally(() => process.exit(1));
105+
});
106+
107+
process.on('unhandledRejection', (reason, promise) => {
108+
logger.error(`Unhandled rejection at: ${promise}, reason: ${reason}`);
109+
cleanup('unhandledRejection').finally(() => process.exit(1));
110+
});

‎packages/backend/src/main.ts‎

Lines changed: 0 additions & 49 deletions
This file was deleted.

0 commit comments

Comments
 (0)