Implement parallel processing with retry logic for repository mirroring and syncing operations

src/pages/api/job/mirror-org.ts

@@ -6,6 +6,8 @@ import { createGitHubClient } from "@/lib/github";
 import { mirrorGitHubOrgToGitea } from "@/lib/gitea";
 import { repoStatusEnum } from "@/types/Repository";
 import { type MembershipRole } from "@/types/organizations";
+import { processWithRetry } from "@/lib/utils/concurrency";
+import { createMirrorJob } from "@/lib/helpers";
 
 export const POST: APIRoute = async ({ request }) => {
   try {
@@ -61,31 +63,73 @@ export const POST: APIRoute = async ({ request }) => {
       );
     }
 
-    // Fire async mirroring without blocking response
+    // Fire async mirroring without blocking response, using parallel processing
     setTimeout(async () => {
-      for (const org of orgs) {
-        if (!config.githubConfig.token) {
-          throw new Error("GitHub token is missing in config.");
-        }
+      if (!config.githubConfig.token) {
+        throw new Error("GitHub token is missing in config.");
+      }
 
-        const octokit = createGitHubClient(config.githubConfig.token);
+      // Create a single Octokit instance to be reused
+      const octokit = createGitHubClient(config.githubConfig.token);
 
-        try {
+      // Define the concurrency limit - adjust based on API rate limits
+      // Using a lower concurrency for organizations since each org might contain many repos
+      const CONCURRENCY_LIMIT = 2;
+
+      // Process organizations in parallel with retry capability
+      await processWithRetry(
+        orgs,
+        async (org) => {
+          // Prepare organization data
+          const orgData = {
+            ...org,
+            status: repoStatusEnum.parse("imported"),
+            membershipRole: org.membershipRole as MembershipRole,
+            lastMirrored: org.lastMirrored ?? undefined,
+            errorMessage: org.errorMessage ?? undefined,
+          };
+
+          // Log the start of mirroring
+          console.log(`Starting mirror for organization: ${org.name}`);
+
+          // Create a mirror job entry to track progress
+          await createMirrorJob({
+            userId: config.userId || "",
+            organizationId: org.id,
+            organizationName: org.name,
+            message: `Started mirroring organization: ${org.name}`,
+            details: `Organization ${org.name} is now in the mirroring queue.`,
+            status: "mirroring",
+          });
+
+          // Mirror the organization
           await mirrorGitHubOrgToGitea({
             config,
             octokit,
-            organization: {
-              ...org,
-              status: repoStatusEnum.parse("imported"),
-              membershipRole: org.membershipRole as MembershipRole,
-              lastMirrored: org.lastMirrored ?? undefined,
-              errorMessage: org.errorMessage ?? undefined,
-            },
+            organization: orgData,
           });
-        } catch (error) {
-          console.error(`Mirror failed for organization ${org.name}:`, error);
+
+          return org;
+        },
+        {
+          concurrencyLimit: CONCURRENCY_LIMIT,
+          maxRetries: 2,
+          retryDelay: 3000,
+          onProgress: (completed, total, result) => {
+            const percentComplete = Math.round((completed / total) * 100);
+            console.log(`Organization mirroring progress: ${percentComplete}% (${completed}/${total})`);
+
+            if (result) {
+              console.log(`Successfully mirrored organization: ${result.name}`);
+            }
+          },
+          onRetry: (org, error, attempt) => {
+            console.log(`Retrying organization ${org.name} (attempt ${attempt}): ${error.message}`);
+          }
         }
-      }
+      );
+
+      console.log("All organization mirroring tasks completed");
     }, 0);
 
     const responsePayload: MirrorOrgResponse = {

src/pages/api/job/mirror-repo.ts

@@ -8,6 +8,8 @@ import {
   mirrorGitHubOrgRepoToGiteaOrg,
 } from "@/lib/gitea";
 import { createGitHubClient } from "@/lib/github";
+import { processWithRetry } from "@/lib/utils/concurrency";
+import { createMirrorJob } from "@/lib/helpers";
 
 export const POST: APIRoute = async ({ request }) => {
   try {
@@ -63,52 +65,84 @@ export const POST: APIRoute = async ({ request }) => {
       );
     }
 
-    // Start async mirroring in background
+    // Start async mirroring in background with parallel processing
     setTimeout(async () => {
-      for (const repo of repos) {
-        if (!config.githubConfig.token) {
-          throw new Error("GitHub token is missing.");
-        }
+      if (!config.githubConfig.token) {
+        throw new Error("GitHub token is missing.");
+      }
 
-        const octokit = createGitHubClient(config.githubConfig.token);
+      // Create a single Octokit instance to be reused
+      const octokit = createGitHubClient(config.githubConfig.token);
 
-        try {
+      // Define the concurrency limit - adjust based on API rate limits
+      const CONCURRENCY_LIMIT = 3;
+
+      // Process repositories in parallel with retry capability
+      await processWithRetry(
+        repos,
+        async (repo) => {
+          // Prepare repository data
+          const repoData = {
+            ...repo,
+            status: repoStatusEnum.parse("imported"),
+            organization: repo.organization ?? undefined,
+            lastMirrored: repo.lastMirrored ?? undefined,
+            errorMessage: repo.errorMessage ?? undefined,
+            forkedFrom: repo.forkedFrom ?? undefined,
+            visibility: repositoryVisibilityEnum.parse(repo.visibility),
+            mirroredLocation: repo.mirroredLocation || "",
+          };
+
+          // Log the start of mirroring
+          console.log(`Starting mirror for repository: ${repo.name}`);
+
+          // Create a mirror job entry to track progress
+          await createMirrorJob({
+            userId: config.userId || "",
+            repositoryId: repo.id,
+            repositoryName: repo.name,
+            message: `Started mirroring repository: ${repo.name}`,
+            details: `Repository ${repo.name} is now in the mirroring queue.`,
+            status: "mirroring",
+          });
+
+          // Mirror the repository based on whether it's in an organization
           if (repo.organization && config.githubConfig.preserveOrgStructure) {
             await mirrorGitHubOrgRepoToGiteaOrg({
               config,
               octokit,
               orgName: repo.organization,
-              repository: {
-                ...repo,
-                status: repoStatusEnum.parse("imported"),
-                organization: repo.organization ?? undefined,
-                lastMirrored: repo.lastMirrored ?? undefined,
-                errorMessage: repo.errorMessage ?? undefined,
-                forkedFrom: repo.forkedFrom ?? undefined,
-                visibility: repositoryVisibilityEnum.parse(repo.visibility),
-                mirroredLocation: repo.mirroredLocation || "",
-              },
+              repository: repoData,
             });
           } else {
             await mirrorGithubRepoToGitea({
               octokit,
-              repository: {
-                ...repo,
-                status: repoStatusEnum.parse("imported"),
-                organization: repo.organization ?? undefined,
-                lastMirrored: repo.lastMirrored ?? undefined,
-                errorMessage: repo.errorMessage ?? undefined,
-                forkedFrom: repo.forkedFrom ?? undefined,
-                visibility: repositoryVisibilityEnum.parse(repo.visibility),
-                mirroredLocation: repo.mirroredLocation || "",
-              },
+              repository: repoData,
               config,
             });
           }
-        } catch (error) {
-          console.error(`Mirror failed for repo ${repo.name}:`, error);
+
+          return repo;
+        },
+        {
+          concurrencyLimit: CONCURRENCY_LIMIT,
+          maxRetries: 2,
+          retryDelay: 2000,
+          onProgress: (completed, total, result) => {
+            const percentComplete = Math.round((completed / total) * 100);
+            console.log(`Mirroring progress: ${percentComplete}% (${completed}/${total})`);
+
+            if (result) {
+              console.log(`Successfully mirrored repository: ${result.name}`);
+            }
+          },
+          onRetry: (repo, error, attempt) => {
+            console.log(`Retrying repository ${repo.name} (attempt ${attempt}): ${error.message}`);
+          }
         }
-      }
+      );
+
+      console.log("All repository mirroring tasks completed");
     }, 0);
 
     const responsePayload: MirrorRepoResponse = {

src/pages/api/job/retry-repo.ts

@@ -10,6 +10,8 @@ import {
 import { createGitHubClient } from "@/lib/github";
 import { repoStatusEnum, repositoryVisibilityEnum } from "@/types/Repository";
 import type { RetryRepoRequest, RetryRepoResponse } from "@/types/retry";
+import { processWithRetry } from "@/lib/utils/concurrency";
+import { createMirrorJob } from "@/lib/helpers";
 
 export const POST: APIRoute = async ({ request }) => {
   try {
@@ -65,10 +67,21 @@ export const POST: APIRoute = async ({ request }) => {
       );
     }
 
-    // Start background retry
+    // Start background retry with parallel processing
     setTimeout(async () => {
-      for (const repo of repos) {
-        try {
+      // Create a single Octokit instance to be reused if needed
+      const octokit = config.githubConfig.token
+        ? createGitHubClient(config.githubConfig.token)
+        : null;
+
+      // Define the concurrency limit - adjust based on API rate limits
+      const CONCURRENCY_LIMIT = 3;
+
+      // Process repositories in parallel with retry capability
+      await processWithRetry(
+        repos,
+        async (repo) => {
+          // Prepare repository data
           const visibility = repositoryVisibilityEnum.parse(repo.visibility);
           const status = repoStatusEnum.parse(repo.status);
           const repoData = {
@@ -81,6 +94,20 @@ export const POST: APIRoute = async ({ request }) => {
             forkedFrom: repo.forkedFrom ?? undefined,
           };
 
+          // Log the start of retry operation
+          console.log(`Starting retry for repository: ${repo.name}`);
+
+          // Create a mirror job entry to track progress
+          await createMirrorJob({
+            userId: config.userId || "",
+            repositoryId: repo.id,
+            repositoryName: repo.name,
+            message: `Started retry operation for repository: ${repo.name}`,
+            details: `Repository ${repo.name} is now in the retry queue.`,
+            status: "imported",
+          });
+
+          // Determine if the repository exists in Gitea
           let owner = getGiteaRepoOwner({
             config,
             repository: repoData,
@@ -93,16 +120,21 @@ export const POST: APIRoute = async ({ request }) => {
           });
 
           if (present) {
+            // If the repository exists, sync it
             await syncGiteaRepo({ config, repository: repoData });
             console.log(`Synced existing repo: ${repo.name}`);
           } else {
+            // If the repository doesn't exist, mirror it
             if (!config.githubConfig.token) {
               throw new Error("GitHub token is missing.");
             }
 
+            if (!octokit) {
+              throw new Error("Octokit client is not initialized.");
+            }
+
             console.log(`Importing repo: ${repo.name} ${owner}`);
 
-            const octokit = createGitHubClient(config.githubConfig.token);
             if (repo.organization && config.githubConfig.preserveOrgStructure) {
               await mirrorGitHubOrgRepoToGiteaOrg({
                 config,
@@ -124,10 +156,28 @@ export const POST: APIRoute = async ({ request }) => {
               });
             }
           }
-        } catch (err) {
-          console.error(`Failed to retry repo ${repo.name}:`, err);
+
+          return repo;
+        },
+        {
+          concurrencyLimit: CONCURRENCY_LIMIT,
+          maxRetries: 2,
+          retryDelay: 2000,
+          onProgress: (completed, total, result) => {
+            const percentComplete = Math.round((completed / total) * 100);
+            console.log(`Retry progress: ${percentComplete}% (${completed}/${total})`);
+
+            if (result) {
+              console.log(`Successfully processed repository: ${result.name}`);
+            }
+          },
+          onRetry: (repo, error, attempt) => {
+            console.log(`Retrying repository ${repo.name} (attempt ${attempt}): ${error.message}`);
+          }
         }
-      }
+      );
+
+      console.log("All repository retry tasks completed");
     }, 0);
 
     const responsePayload: RetryRepoResponse = {

src/pages/api/job/sync-repo.ts

@@ -5,6 +5,8 @@ import { eq, inArray } from "drizzle-orm";
 import { repositoryVisibilityEnum, repoStatusEnum } from "@/types/Repository";
 import { syncGiteaRepo } from "@/lib/gitea";
 import type { SyncRepoResponse } from "@/types/sync";
+import { processWithRetry } from "@/lib/utils/concurrency";
+import { createMirrorJob } from "@/lib/helpers";
 
 export const POST: APIRoute = async ({ request }) => {
   try {
@@ -60,26 +62,66 @@ export const POST: APIRoute = async ({ request }) => {
       );
     }
 
-    // Start async mirroring in background
+    // Start async mirroring in background with parallel processing
     setTimeout(async () => {
-      for (const repo of repos) {
-        try {
+      // Define the concurrency limit - adjust based on API rate limits
+      const CONCURRENCY_LIMIT = 5;
+
+      // Process repositories in parallel with retry capability
+      await processWithRetry(
+        repos,
+        async (repo) => {
+          // Prepare repository data
+          const repoData = {
+            ...repo,
+            status: repoStatusEnum.parse(repo.status),
+            organization: repo.organization ?? undefined,
+            lastMirrored: repo.lastMirrored ?? undefined,
+            errorMessage: repo.errorMessage ?? undefined,
+            forkedFrom: repo.forkedFrom ?? undefined,
+            visibility: repositoryVisibilityEnum.parse(repo.visibility),
+          };
+
+          // Log the start of syncing
+          console.log(`Starting sync for repository: ${repo.name}`);
+
+          // Create a mirror job entry to track progress
+          await createMirrorJob({
+            userId: config.userId || "",
+            repositoryId: repo.id,
+            repositoryName: repo.name,
+            message: `Started syncing repository: ${repo.name}`,
+            details: `Repository ${repo.name} is now in the syncing queue.`,
+            status: "syncing",
+          });
+
+          // Sync the repository
           await syncGiteaRepo({
             config,
-            repository: {
-              ...repo,
-              status: repoStatusEnum.parse(repo.status),
-              organization: repo.organization ?? undefined,
-              lastMirrored: repo.lastMirrored ?? undefined,
-              errorMessage: repo.errorMessage ?? undefined,
-              forkedFrom: repo.forkedFrom ?? undefined,
-              visibility: repositoryVisibilityEnum.parse(repo.visibility),
-            },
+            repository: repoData,
           });
-        } catch (error) {
-          console.error(`Sync failed for repo ${repo.name}:`, error);
+
+          return repo;
+        },
+        {
+          concurrencyLimit: CONCURRENCY_LIMIT,
+          maxRetries: 2,
+          retryDelay: 2000,
+          onProgress: (completed, total, result) => {
+            const percentComplete = Math.round((completed / total) * 100);
+            console.log(`Syncing progress: ${percentComplete}% (${completed}/${total})`);
+
+            if (result) {
+              console.log(`Successfully synced repository: ${result.name}`);
+            }
+          },
+          onRetry: (repo, error, attempt) => {
+            console.log(`Retrying sync for repository ${repo.name} (attempt ${attempt}): ${error.message}`);
+          }
         }
-      }
+      );
+
+      console.log("All repository syncing tasks completed");
     }, 0);
 
     const responsePayload: SyncRepoResponse = {