chore(worker): Migrate off of groupmq

packages/backend/package.json

@@ -45,14 +45,14 @@
         "git-url-parse": "^16.1.0",
         "gitea-js": "^1.22.0",
         "glob": "^11.1.0",
-        "groupmq": "^1.0.0",
         "http-status-codes": "^2.3.0",
         "ioredis": "^5.4.2",
         "lowdb": "^7.0.1",
         "micromatch": "^4.0.8",
         "p-limit": "^7.2.0",
         "posthog-node": "^5.17.4",
         "prom-client": "^15.1.3",
+        "redlock": "5.0.0-beta.2",
         "simple-git": "^3.27.0",
         "zod": "^3.25.74"
     }

packages/backend/src/constants.ts

@@ -13,7 +13,7 @@ export const REPOS_CACHE_DIR = path.join(env.DATA_CACHE_DIR, 'repos');
 export const INDEX_CACHE_DIR = path.join(env.DATA_CACHE_DIR, 'index');
 
 // Maximum time to wait for current job to finish
-export const GROUPMQ_WORKER_STOP_GRACEFUL_TIMEOUT_MS = 5 * 1000; // 5 seconds
+export const WORKER_STOP_GRACEFUL_TIMEOUT_MS = 5 * 1000; // 5 seconds
 
 // List of shutdown signals
 export const SHUTDOWN_SIGNALS: string[] = [

packages/backend/src/github.ts

@@ -306,15 +306,27 @@ const getReposOwnedByUsers = async (users: string[], octokit: Octokit, signal: A
                     // the username as a parameter.
                     // @see: https://github.com/orgs/community/discussions/24382#discussioncomment-3243958
                     // @see: https://api.github.com/search/repositories?q=user:USERNAME
-                    const searchResults = await octokitToUse.paginate(octokitToUse.rest.search.repos, {
+
+                    // @note: We use paginate.iterator() instead of paginate() to check
+                    // signal.aborted between pages. paginate() only passes the signal to
+                    // individual fetch requests but doesn't check abort state between pages.
+                    const allRepos: OctokitRepository[] = [];
+                    const iterator = octokitToUse.paginate.iterator(octokitToUse.rest.search.repos, {
                         q: query,
                         per_page: 100,
                         request: {
                             signal,
                         },
                     });
 
-                    return searchResults as OctokitRepository[];
+                    for await (const { data: repos } of iterator) {
+                        if (signal.aborted) {
+                            throw new DOMException('Operation aborted', 'AbortError');
+                        }
+                        allRepos.push(...(repos as OctokitRepository[]));
+                    }
+
+                    return allRepos;
                 };
 
                 return fetchWithRetry(fetchFn, `user ${user}`, logger);
@@ -357,13 +369,28 @@ const getReposForOrgs = async (orgs: string[], octokit: Octokit, signal: AbortSi
 
             const octokitToUse = await getOctokitWithGithubApp(octokit, org, url, `org ${org}`);
             const { durationMs, data } = await measure(async () => {
-                const fetchFn = () => octokitToUse.paginate(octokitToUse.repos.listForOrg, {
-                    org: org,
-                    per_page: 100,
-                    request: {
-                        signal
+                // @note: We use paginate.iterator() instead of paginate() to check
+                // signal.aborted between pages. paginate() only passes the signal to
+                // individual fetch requests but doesn't check abort state between pages.
+                const fetchFn = async () => {
+                    const allRepos: OctokitRepository[] = [];
+                    const iterator = octokitToUse.paginate.iterator(octokitToUse.repos.listForOrg, {
+                        org: org,
+                        per_page: 100,
+                        request: {
+                            signal
+                        }
+                    });
+
+                    for await (const { data: repos } of iterator) {
+                        if (signal.aborted) {
+                            throw new DOMException('Operation aborted', 'AbortError');
+                        }
+                        allRepos.push(...repos);
                     }
-                });
+
+                    return allRepos;
+                };
 
                 return fetchWithRetry(fetchFn, `org ${org}`, logger);
             });