feat(api): sync progress (#9)

* feat(api): sync progress

* fix(api): set initial sync value

apps/api/src/app/github/services/github-installation.service.ts

@@ -14,6 +14,7 @@ import {
 import { SweetQueue, addJob } from "../../../bull-mq/queues";
 import { logger } from "../../../lib/logger";
 import { octokit } from "../../../lib/octokit";
+import { setInitialSyncProgress } from "../../workspaces/services/workspace.service";
 
 export const syncGitHubInstallation = async (
   gitInstallation: GitHubInstallation,
@@ -30,9 +31,11 @@ export const syncGitHubInstallation = async (
 
   await connectUserToWorkspace(gitProfile, workspace);
 
+  await setInitialSyncProgress(workspace.id);
+
   await addJob(SweetQueue.GITHUB_REPOSITORIES_SYNC, {
     installation: { id: parseInt(installation.gitInstallationId) },
-    shouldSyncRepositoryPullRequests: true,
+    syncPullRequests: true,
   });
 
   if (workspace.organization) {

apps/api/src/app/github/services/github-pull-request.service.ts

@@ -17,6 +17,7 @@ import {
   getTimeToCode,
   getTimeToMerge,
 } from "./github-pull-request-tracking.service";
+import { incrementInitialSyncProgress } from "../../workspaces/services/workspace.service";
 
 interface Author {
   id: string;
@@ -32,7 +33,10 @@ type RepositoryData = Omit<
 export const syncPullRequest = async (
   gitInstallationId: number,
   pullRequestId: string,
-  syncReviews = false
+  { syncReviews, initialSync } = {
+    syncReviews: false,
+    initialSync: false,
+  }
 ) => {
   logger.info("syncPullRequest", {
     installationId: gitInstallationId,
@@ -77,6 +81,10 @@ export const syncPullRequest = async (
     gitPrData
   );
 
+  if (initialSync) {
+    incrementInitialSyncProgress(workspace.id, "done", 1);
+  }
+
   if (syncReviews) {
     logger.debug("syncPullRequest: Adding job to sync reviews", {
       gitPrData,

apps/api/src/app/github/services/github-repository-pull-requests.service.ts

@@ -6,9 +6,13 @@ import {
 import { getBypassRlsPrisma } from "../../../prisma";
 import { ResourceNotFoundException } from "../../errors/exceptions/resource-not-found.exception";
 import { logger } from "../../../lib/logger";
-import { getWorkspaceHandle } from "../../workspaces/services/workspace.service";
+import {
+  getWorkspaceHandle,
+  incrementInitialSyncProgress,
+} from "../../workspaces/services/workspace.service";
 import { BusinessRuleException } from "../../errors/exceptions/business-rule.exception";
 import { JobPriority, SweetQueue, addJobs } from "../../../bull-mq/queues";
+import { sleep } from "radash";
 
 export const syncGitHubRepositoryPullRequests = async (
   repositoryName: string,
@@ -34,19 +38,28 @@ export const syncGitHubRepositoryPullRequests = async (
 
   if (!gitHubPullRequests.length) return;
 
+  await incrementInitialSyncProgress(
+    workspace.id,
+    "waiting",
+    gitHubPullRequests.length
+  );
+
   addJobs(
     SweetQueue.GITHUB_SYNC_PULL_REQUEST,
     gitHubPullRequests.map((pullRequest) => ({
       installation: { id: gitInstallationId },
       pull_request: { node_id: pullRequest.id },
       syncReviews: true,
+      initialSync: true,
     })),
     {
       priority: JobPriority.LOW,
     }
   );
 };
 
+// Note: We could optimize initial sync by fetching all PR data here and saving to our database.
+// For now we dispatch one job per PR instead, to break down the problem and keep the code simpler.
 const fetchGitHubPullRequests = async (
   repositoryName: string,
   owner: string,
@@ -98,6 +111,7 @@ const fetchGitHubPullRequests = async (
 
     hasNextPage = pageInfo.hasNextPage;
     cursor = pageInfo.endCursor;
+    await sleep(500);
   }
 
   return pullRequests;

apps/api/src/app/github/services/github-repository.service.ts

@@ -16,7 +16,7 @@ type RepositoryData = Omit<
 
 export const syncGitHubRepositories = async (
   gitInstallationId: number,
-  shouldSyncRepositoryPullRequests: boolean
+  syncPullRequests: boolean
 ): Promise<void> => {
   logger.info("syncGitHubRepositories", { gitInstallationId });
 
@@ -32,7 +32,7 @@ export const syncGitHubRepositories = async (
   const gitHubRepositories = await fetchGitHubRepositories(gitInstallationId);
   const repositories = await upsertRepositories(workspace, gitHubRepositories);
 
-  if (shouldSyncRepositoryPullRequests) {
+  if (syncPullRequests) {
     const nonArchivedRepositories = repositories.filter(
       (repository) => !repository.archivedAt
     );

apps/api/src/app/github/workers/github-repositories-sync.worker.ts

@@ -8,9 +8,7 @@ import { withDelayedRetryOnRateLimit } from "../services/github-rate-limit.servi
 
 export const githubRepositoriesSyncWorker = createWorker(
   SweetQueue.GITHUB_REPOSITORIES_SYNC,
-  async (
-    job: Job<RepositoryEvent & { shouldSyncRepositoryPullRequests?: boolean }>
-  ) => {
+  async (job: Job<RepositoryEvent & { syncPullRequests?: boolean }>) => {
     const installationId = job.data.installation?.id;
 
     if (!installationId) {
@@ -24,7 +22,7 @@ export const githubRepositoriesSyncWorker = createWorker(
       () =>
         syncGitHubRepositories(
           installationId,
-          job.data.shouldSyncRepositoryPullRequests || false
+          job.data.syncPullRequests || false
         ),
       {
         job,

apps/api/src/app/github/workers/github-sync-pull-request.worker.ts

@@ -15,6 +15,7 @@ export const syncPullRequestWorker = createWorker(
     job: Job<
       (PullRequestSynchronizeEvent | PullRequestOpenedEvent) & {
         syncReviews?: boolean;
+        initialSync?: boolean;
       }
     >
   ) => {
@@ -33,14 +34,14 @@ export const syncPullRequestWorker = createWorker(
     }
 
     const installationId = job.data.installation.id;
+    const options = {
+      syncReviews: job.data.syncReviews || false,
+      initialSync: job.data.initialSync || false,
+    };
 
     await withDelayedRetryOnRateLimit(
       () =>
-        syncPullRequest(
-          installationId,
-          job.data.pull_request.node_id,
-          job.data.syncReviews
-        ),
+        syncPullRequest(installationId, job.data.pull_request.node_id, options),
       {
         job,
         installationId,

apps/api/src/app/workspaces/resolvers/queries/workspace-sync-progress.query.ts

@@ -0,0 +1,16 @@
+import { createFieldResolver } from "../../../../lib/graphql";
+import { logger } from "../../../../lib/logger";
+import { ResourceNotFoundException } from "../../../errors/exceptions/resource-not-found.exception";
+import { getInitialSyncProgress } from "../../services/workspace.service";
+
+export const workspaceSyncProgressQuery = createFieldResolver("Workspace", {
+  initialSyncProgress: async (workspace) => {
+    logger.info("query.initialSyncProgress", { workspaceId: workspace.id });
+
+    if (!workspace.id) {
+      throw new ResourceNotFoundException("Could not find workspace");
+    }
+
+    return getInitialSyncProgress(workspace.id);
+  },
+});

apps/api/src/app/workspaces/resolvers/workspaces.schema.ts

@@ -6,6 +6,8 @@ export default /* GraphQL */ `
     avatar: String
     "The git provider URL to uninstall the sweetr app"
     gitUninstallUrl: String!
+    "A number between 0 and 100 representing the progress of the initial data synchronization with the git provider"
+    initialSyncProgress: Int!
   }
 
   type Query {

apps/api/src/app/workspaces/services/workspace.service.ts

@@ -5,6 +5,9 @@ import {
   Workspace,
 } from "@prisma/client";
 import { getBypassRlsPrisma, getPrisma } from "../../../prisma";
+import { redisConnection } from "../../../bull-mq/redis-connection";
+import { UnknownException } from "../../errors/exceptions/unknown.exception";
+import { captureException } from "../../../lib/sentry";
 
 type WorkspaceWithUserOrOrganization = Workspace & {
   gitProfile: GitProfile | null;
@@ -109,3 +112,51 @@ export const getWorkspaceUninstallGitUrl = (
 
   return `https://github.com/settings/installations/${workspace.installation?.gitInstallationId}`;
 };
+
+export const setInitialSyncProgress = async (workspaceId: number) => {
+  const key = `workspace:${workspaceId}:sync`;
+  const sevenDaysInSeconds = 60 * 60 * 24 * 7;
+
+  await redisConnection
+    .multi()
+    .hset(key, { waiting: 0, done: 0 })
+    .expire(key, sevenDaysInSeconds)
+    .exec();
+};
+
+export const incrementInitialSyncProgress = async (
+  workspaceId: number,
+  field: "waiting" | "done",
+  amount: number
+) => {
+  const key = `workspace:${workspaceId}:sync`;
+
+  await redisConnection.hincrby(key, field, amount);
+};
+
+export const getInitialSyncProgress = async (workspaceId: number) => {
+  try {
+    const progress = await redisConnection.hgetall(
+      `workspace:${workspaceId}:sync`
+    );
+
+    if (!progress || !("waiting" in progress)) return 100;
+
+    const done = Number(progress.done);
+    const waiting = Number(progress.waiting);
+
+    // Avoid division by zero
+    if (waiting === 0) return 0;
+
+    return Math.round((done * 100) / waiting);
+  } catch (error) {
+    captureException(
+      new UnknownException("Redis: Could not get workspace sync progress.", {
+        originalError: error,
+        severity: "warning",
+      })
+    );
+
+    return 100;
+  }
+};

packages/graphql-types/api.ts

@@ -484,6 +484,8 @@ export type Workspace = {
   gitUninstallUrl: Scalars['String']['output'];
   handle: Scalars['String']['output'];
   id: Scalars['SweetID']['output'];
+  /** A number between 0 and 100 representing the progress of the initial data synchronization with the git provider */
+  initialSyncProgress: Scalars['Int']['output'];
   me?: Maybe<Person>;
   name: Scalars['String']['output'];
   people: Array<Person>;
@@ -943,6 +945,7 @@ export type WorkspaceResolvers<ContextType = GraphQLContext, ParentType extends
   gitUninstallUrl?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   handle?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   id?: Resolver<ResolversTypes['SweetID'], ParentType, ContextType>;
+  initialSyncProgress?: Resolver<ResolversTypes['Int'], ParentType, ContextType>;
   me?: Resolver<Maybe<ResolversTypes['Person']>, ParentType, ContextType>;
   name?: Resolver<ResolversTypes['String'], ParentType, ContextType>;
   people?: Resolver<Array<ResolversTypes['Person']>, ParentType, ContextType, Partial<WorkspacePeopleArgs>>;

packages/graphql-types/frontend/graphql.ts

@@ -482,6 +482,8 @@ export type Workspace = {
   gitUninstallUrl: Scalars['String']['output'];
   handle: Scalars['String']['output'];
   id: Scalars['SweetID']['output'];
+  /** A number between 0 and 100 representing the progress of the initial data synchronization with the git provider */
+  initialSyncProgress: Scalars['Int']['output'];
   me?: Maybe<Person>;
   name: Scalars['String']['output'];
   people: Array<Person>;