feat: parse cv to profile

.infra/common.ts

@@ -437,6 +437,10 @@ export const workers: Worker[] = [
     topic: 'api.v1.recruiter-rejected-candidate-match',
     subscription: 'api.recruiter-rejected-candidate-match-email',
   },
+  {
+    topic: 'api.v1.candidate-preference-updated',
+    subscription: 'api.parse-cv-profile',
+  },
 ];
 
 export const personalizedDigestWorkers: Worker[] = [

bin/importProfileFromJSON.ts

@@ -5,14 +5,7 @@ import { z } from 'zod';
 import createOrGetConnection from '../src/db';
 import { type DataSource } from 'typeorm';
 import { readFile } from 'node:fs/promises';
-import { userExperienceInputBaseSchema } from '../src/common/schema/profile';
-import { UserExperienceType } from '../src/entity/user/experiences/types';
-import {
-  importUserExperienceWork,
-  importUserExperienceEducation,
-  importUserExperienceCertification,
-  importUserExperienceProject,
-} from '../src/common/profile/import';
+import { importUserExperienceFromJSON } from '../src/common/profile/import';
 
 /**
  * Import profile from JSON to user by id
@@ -47,55 +40,10 @@ const main = async () => {
 
     const dataJSON = JSON.parse(await readFile(params.path, 'utf-8'));
 
-    const data = z
-      .array(
-        userExperienceInputBaseSchema
-          .pick({
-            type: true,
-          })
-          .loose(),
-      )
-      .parse(dataJSON);
-
-    await con.transaction(async (entityManager) => {
-      for (const item of data) {
-        switch (item.type) {
-          case UserExperienceType.Work:
-            await importUserExperienceWork({
-              data: item,
-              con: entityManager,
-              userId: params.userId,
-            });
-
-            break;
-          case UserExperienceType.Education:
-            await importUserExperienceEducation({
-              data: item,
-              con: entityManager,
-              userId: params.userId,
-            });
-
-            break;
-          case UserExperienceType.Certification:
-            await importUserExperienceCertification({
-              data: item,
-              con: entityManager,
-              userId: params.userId,
-            });
-
-            break;
-          case UserExperienceType.Project:
-            await importUserExperienceProject({
-              data: item,
-              con: entityManager,
-              userId: params.userId,
-            });
-
-            break;
-          default:
-            throw new Error(`Unsupported experience type: ${item.type}`);
-        }
-      }
+    await importUserExperienceFromJSON({
+      con: con.manager,
+      dataJson: dataJSON,
+      userId: params.userId,
     });
   } catch (error) {
     console.error(error instanceof z.ZodError ? z.prettifyError(error) : error);

package.json

@@ -36,7 +36,7 @@
     "@connectrpc/connect-fastify": "^1.6.1",
     "@connectrpc/connect-node": "^1.6.1",
     "@dailydotdev/graphql-redis-subscriptions": "^2.4.3",
-    "@dailydotdev/schema": "0.2.50",
+    "@dailydotdev/schema": "0.2.51",
     "@dailydotdev/ts-ioredis-pool": "^1.0.2",
     "@fastify/cookie": "^11.0.2",
     "@fastify/cors": "^11.1.0",

src/common/profile/import.ts

@@ -2,6 +2,7 @@ import { type EntityManager } from 'typeorm';
 import {
   userExperienceCertificationImportSchema,
   userExperienceEducationImportSchema,
+  userExperienceInputBaseSchema,
   userExperienceProjectImportSchema,
   userExperienceWorkImportSchema,
 } from '../../../src/common/schema/profile';
@@ -15,6 +16,8 @@ import { DatasetLocation } from '../../../src/entity/dataset/DatasetLocation';
 import { UserExperienceEducation } from '../../../src/entity/user/experiences/UserExperienceEducation';
 import { UserExperienceCertification } from '../../../src/entity/user/experiences/UserExperienceCertification';
 import { UserExperienceProject } from '../../../src/entity/user/experiences/UserExperienceProject';
+import z from 'zod';
+import { UserExperienceType } from '../../entity/user/experiences/types';
 
 const resolveUserCompanyPart = async ({
   name,
@@ -306,3 +309,68 @@ export const importUserExperienceProject = async ({
     experienceId,
   };
 };
+
+export const importUserExperienceFromJSON = async ({
+  con,
+  dataJson,
+  userId,
+}: {
+  con: EntityManager;
+  dataJson: unknown;
+  userId: string;
+}) => {
+  if (!userId) {
+    throw new Error('userId is required');
+  }
+
+  const data = z
+    .array(
+      userExperienceInputBaseSchema
+        .pick({
+          type: true,
+        })
+        .loose(),
+    )
+    .parse(dataJson);
+
+  await con.transaction(async (entityManager) => {
+    for (const item of data) {
+      switch (item.type) {
+        case UserExperienceType.Work:
+          await importUserExperienceWork({
+            data: item,
+            con: entityManager,
+            userId,
+          });
+
+          break;
+        case UserExperienceType.Education:
+          await importUserExperienceEducation({
+            data: item,
+            con: entityManager,
+            userId,
+          });
+
+          break;
+        case UserExperienceType.Certification:
+          await importUserExperienceCertification({
+            data: item,
+            con: entityManager,
+            userId,
+          });
+
+          break;
+        case UserExperienceType.Project:
+          await importUserExperienceProject({
+            data: item,
+            con: entityManager,
+            userId,
+          });
+
+          break;
+        default:
+          throw new Error(`Unsupported experience type: ${item.type}`);
+      }
+    }
+  });
+};

src/entity/user/User.ts

@@ -44,6 +44,7 @@ export type UserFlags = Partial<{
     lng: number | null | undefined;
   };
   subdivision: string | null;
+  lastCVParseAt: Date;
 }>;
 
 export type UserFlagsPublic = Pick<UserFlags, 'showPlusGift'>;

src/workers/opportunity/parseCVProfile.ts

@@ -0,0 +1,97 @@
+import {
+  BrokkrParseRequest,
+  CandidatePreferenceUpdated,
+} from '@dailydotdev/schema';
+import type { TypedWorker } from '../worker';
+import { User } from '../../entity/user/User';
+import { getBrokkrClient } from '../../common/brokkr';
+import { updateFlagsStatement } from '../../common/utils';
+import { importUserExperienceFromJSON } from '../../common/profile/import';
+
+export const parseCVProfileWorker: TypedWorker<'api.v1.candidate-preference-updated'> =
+  {
+    subscription: 'api.parse-cv-profile',
+    parseMessage: ({ data }) => CandidatePreferenceUpdated.fromBinary(data),
+    handler: async ({ data }, con) => {
+      const { userId, cv } = data.payload || {};
+
+      if (!cv?.blob || !cv?.bucket) {
+        return;
+      }
+
+      if (!cv?.lastModified) {
+        return;
+      }
+
+      if (!userId) {
+        return;
+      }
+
+      const user: Pick<User, 'flags'> | null = await con
+        .getRepository(User)
+        .findOne({
+          select: ['flags'],
+          where: {
+            id: userId,
+          },
+        });
+
+      if (!user) {
+        return;
+      }
+
+      const lastModifiedCVDate = new Date(cv.lastModified);
+
+      if (Number.isNaN(lastModifiedCVDate.getTime())) {
+        return;
+      }
+
+      const lastProfileParseDate = user.flags.lastCVParseAt || new Date(0);
+
+      if (lastModifiedCVDate <= lastProfileParseDate) {
+        return;
+      }
+
+      const brokkrClient = getBrokkrClient();
+
+      try {
+        await con.getRepository(User).update(
+          { id: userId },
+          {
+            flags: updateFlagsStatement<User>({
+              lastCVParseAt: new Date(),
+            }),
+          },
+        );
+
+        const result = await brokkrClient.garmr.execute(() => {
+          return brokkrClient.instance.parseCV(
+            new BrokkrParseRequest({
+              bucketName: cv.bucket,
+              blobName: cv.blob,
+            }),
+          );
+        });
+
+        const dataJson = JSON.parse(result.parsedCv);
+
+        await importUserExperienceFromJSON({
+          con: con.manager,
+          dataJson,
+          userId,
+        });
+      } catch (error) {
+        // revert to previous date on error
+        await con.getRepository(User).update(
+          { id: userId },
+          {
+            flags: updateFlagsStatement<User>({
+              lastCVParseAt: user.flags.lastCVParseAt,
+            }),
+          },
+        );
+
+        throw error;
+      }
+    },
+  };