feat: user candidate preference location

__tests__/schema/opportunity.ts

@@ -1704,7 +1704,7 @@ describe('query getCandidatePreferences', () => {
           lastModified: new Date('2024-10-10T10:00:00Z'),
         },
         salaryExpectation: { min: '50000', period: SalaryPeriod.ANNUAL },
-        location: [
+        customLocation: [
           { country: 'Norway' },
           { city: 'London', country: 'UK', continent: 'Europe' },
         ],
@@ -1912,7 +1912,6 @@ describe('mutation updateCandidatePreferences', () => {
         roleType: 1.0,
         employmentType: [1, 3],
         salaryExpectation: { min: 70000, period: 1 },
-        location: [{ city: 'Berlin', country: 'Germany' }],
         locationType: [1, 2],
         customKeywords: true,
       },
@@ -1932,7 +1931,6 @@ describe('mutation updateCandidatePreferences', () => {
       roleType: 1.0,
       employmentType: [1, 3], // FULL_TIME, CONTRACT
       salaryExpectation: { min: '70000', period: 1 }, // ANNUAL
-      location: [{ city: 'Berlin', country: 'Germany' }],
       locationType: [1, 2], // REMOTE, ONSITE
       customKeywords: true,
     });

__tests__/workers/cdc/primary.ts

@@ -6809,7 +6809,7 @@ describe('user_candidate_preference', () => {
         min: 100000,
         currency: 'EUR',
       },
-      location: [
+      customLocation: [
         {
           type: 1, // LocationType.REMOTE
           country: 'Germany',

src/common/opportunity/pubsub.ts

@@ -84,27 +84,27 @@ export const notifyOpportunityMatchAccepted = async ({
     return;
   }
 
-  const { match, candidatePreference, keywords } = await queryReadReplica(
-    con,
-    async ({ queryRunner }) => {
+  const { match, candidatePreference, keywords, locationData } =
+    await con.transaction(async (manager) => {
       const [match, candidatePreference] = await Promise.all([
-        queryRunner.manager.getRepository(OpportunityMatch).findOneBy({
+        manager.getRepository(OpportunityMatch).findOneBy({
           opportunityId: data.opportunityId,
           userId: data.userId,
         }),
-        queryRunner.manager
+        manager
           .getRepository(UserCandidatePreference)
-          .findOneBy({ userId: data.userId }),
+          .findOne({ where: { userId: data.userId }, relations: ['location'] }),
       ]);
 
       const keywords = await fetchCandidateKeywords(
-        queryRunner.manager,
+        manager,
         candidatePreference,
       );
 
-      return { match, candidatePreference, keywords };
-    },
-  );
+      const locationData = await candidatePreference?.location;
+
+      return { match, candidatePreference, keywords, locationData };
+    });
 
   if (!match) {
     logger.warn(
@@ -127,6 +127,19 @@ export const notifyOpportunityMatchAccepted = async ({
       ? new Date(candidatePreference.cv.lastModified)
       : candidatePreference.cv.lastModified;
 
+  // Prioritize relational location over customLocation
+  const locationArray = locationData
+    ? [
+        {
+          ...locationData,
+          // Convert null to undefined for protobuf compatibility
+          subdivision: locationData.subdivision ?? undefined,
+          city: locationData.city ?? undefined,
+          externalId: locationData.externalId ?? undefined,
+        },
+      ]
+    : candidatePreference.customLocation || [];
+
   const message = new CandidateAcceptedOpportunityMessage({
     opportunityId: match.opportunityId,
     userId: match.userId,
@@ -135,6 +148,7 @@ export const notifyOpportunityMatchAccepted = async ({
     screening: match.screening,
     candidatePreference: {
       ...candidatePreference,
+      location: locationArray,
       salaryExpectation: new Salary({
         min: candidatePreference.salaryExpectation?.min
           ? BigInt(candidatePreference.salaryExpectation.min)
@@ -430,19 +444,20 @@ export const notifyCandidatePreferenceChange = async ({
   logger: FastifyBaseLogger;
   userId: string;
 }) => {
-  const { candidatePreference, keywords } = await queryReadReplica(
-    con,
-    async ({ queryRunner }) => {
-      const candidatePreference = await queryRunner.manager
+  const { candidatePreference, keywords, locationData } = await con.transaction(
+    async (manager) => {
+      const candidatePreference = await manager
         .getRepository(UserCandidatePreference)
-        .findOneBy({ userId: userId });
+        .findOne({ where: { userId: userId }, relations: ['location'] });
 
       const keywords = await fetchCandidateKeywords(
-        queryRunner.manager,
+        manager,
         candidatePreference,
       );
 
-      return { candidatePreference, keywords };
+      const locationData = await candidatePreference?.location;
+
+      return { candidatePreference, keywords, locationData };
     },
   );
 
@@ -459,9 +474,23 @@ export const notifyCandidatePreferenceChange = async ({
       ? new Date(candidatePreference.cv.lastModified)
       : candidatePreference?.cv?.lastModified;
 
+  // Prioritize relational location over customLocation
+  const locationArray = locationData
+    ? [
+        {
+          ...locationData,
+          // Convert null to undefined for protobuf compatibility
+          subdivision: locationData.subdivision ?? undefined,
+          city: locationData.city ?? undefined,
+          externalId: locationData.externalId ?? undefined,
+        },
+      ]
+    : candidatePreference.customLocation || [];
+
   const message = new CandidatePreferenceUpdated({
     payload: {
       ...candidatePreference,
+      location: locationArray,
       salaryExpectation: new Salary({
         min: candidatePreference.salaryExpectation?.min
           ? BigInt(candidatePreference.salaryExpectation.min)

src/common/schema/userCandidate.ts

@@ -68,6 +68,10 @@ export const candidatePreferenceSchema = z.object({
       }),
     )
     .optional(),
+  externalLocationId: z.preprocess(
+    (val) => (val === '' ? null : val),
+    z.string().nullish().default(null),
+  ),
   locationType: z
     .array(
       z

src/entity/user/UserCandidatePreference.ts

@@ -4,6 +4,7 @@ import {
   Entity,
   Index,
   JoinColumn,
+  ManyToOne,
   OneToOne,
   PrimaryColumn,
   UpdateDateColumn,
@@ -23,6 +24,7 @@ import type {
   UserCandidateCV,
 } from '../../common/schema/userCandidate';
 import { listAllProtoEnumValues } from '../../common';
+import type { DatasetLocation } from '../dataset/DatasetLocation';
 
 export type SalaryExpectation = z.infer<typeof salaryExpectationSchema>;
 
@@ -83,9 +85,21 @@ export class UserCandidatePreference {
   @Column({
     type: 'jsonb',
     default: [],
-    comment: 'Location from protobuf schema',
+    comment: 'Custom location from protobuf schema (legacy)',
   })
-  location: Array<Location> = [];
+  customLocation: Array<Location> = [];
+
+  @Column({ type: 'text', nullable: true, default: null })
+  @Index('IDX_user_candidate_preference_locationId')
+  locationId: string | null;
+
+  @ManyToOne('DatasetLocation', { lazy: true, onDelete: 'SET NULL' })
+  @JoinColumn({
+    name: 'locationId',
+    foreignKeyConstraintName:
+      'FK_user_candidate_preference_dataset_location_locationId',
+  })
+  location: Promise<DatasetLocation> | null;
 
   @Column({
     type: 'integer',

src/graphorm/index.ts

@@ -76,6 +76,7 @@ import type {
   OpportunityFlagsPublic,
 } from '../entity/opportunities/Opportunity';
 import { SubscriptionStatus } from '../common/plus';
+import { isNullOrUndefined } from '../common/object';
 
 const existsByUserAndPost =
   (entity: string, build?: (queryBuilder: QueryBuilder) => QueryBuilder) =>
@@ -1704,6 +1705,32 @@ const obj = new GraphORM({
       },
       location: {
         jsonType: true,
+        select: (_, alias) => `
+          COALESCE(
+            CASE
+              WHEN ${alias}."locationId" IS NOT NULL THEN
+                (
+                  SELECT jsonb_build_array(
+                    jsonb_build_object(
+                      'city', dl.city,
+                      'subdivision', dl.subdivision,
+                      'country', dl.country
+                    )
+                  )
+                  FROM dataset_location dl
+                  WHERE dl.id = ${alias}."locationId"
+                )
+              ELSE NULL
+            END,
+            ${alias}."customLocation"
+          )
+        `,
+        transform: (value: unknown) => {
+          console.log(value);
+          if (isNullOrUndefined(value)) return [];
+          if (Array.isArray(value)) return value;
+          return [value];
+        },
       },
       keywords: {
         relation: {
@@ -1833,18 +1860,22 @@ const obj = new GraphORM({
       },
       location: {
         select: (_, alias) => `
-          COALESCE(
-            (
-              SELECT jsonb_build_object(
-                'city', location->0->>'city',
-                'subdivision', location->0->>'subdivision',
-                'country', location->0->>'country'
-              )
-              FROM user_candidate_preference
-              WHERE "userId" = ${alias}.id
-              LIMIT 1
-            ),
-            ${alias}.flags
+          (
+            SELECT COALESCE(
+              (
+                SELECT jsonb_build_object(
+                  'city', dl.city,
+                  'subdivision', dl.subdivision,
+                  'country', dl.country
+                )
+                FROM dataset_location dl
+                WHERE dl.id = ucp."locationId"
+              ),
+              ucp."customLocation"->0
+            )
+            FROM user_candidate_preference ucp
+            WHERE ucp."userId" = ${alias}.id
+            LIMIT 1
           )
         `,
         transform: (data: Record<string, unknown>): string | null => {

src/migration/1765804151420-1734249600000-MigrateUserCandidatePreferenceLocation.ts

@@ -0,0 +1,62 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class MigrateUserCandidatePreferenceLocation1765804151420
+  implements MigrationInterface
+{
+  name = 'MigrateUserCandidatePreferenceLocation1765804151420';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(/* sql */ `
+      ALTER TABLE "user_candidate_preference"
+        RENAME COLUMN "location" TO "customLocation"
+    `);
+
+    await queryRunner.query(/* sql */ `
+      COMMENT ON COLUMN "user_candidate_preference"."customLocation" IS 'Custom location from protobuf schema (legacy)'
+    `);
+
+    await queryRunner.query(/* sql */ `
+      ALTER TABLE "user_candidate_preference"
+        ADD "locationId" uuid
+    `);
+
+    await queryRunner.query(/* sql */ `
+      CREATE INDEX IF NOT EXISTS "IDX_user_candidate_preference_locationId"
+        ON "user_candidate_preference" ("locationId")
+    `);
+
+    await queryRunner.query(/* sql */ `
+      ALTER TABLE "user_candidate_preference"
+        ADD CONSTRAINT "FK_user_candidate_preference_dataset_location_locationId"
+          FOREIGN KEY ("locationId")
+          REFERENCES "dataset_location"("id")
+          ON DELETE SET NULL
+          ON UPDATE NO ACTION
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(/* sql */ `
+      ALTER TABLE "user_candidate_preference"
+        DROP CONSTRAINT "FK_user_candidate_preference_dataset_location_locationId"
+    `);
+
+    await queryRunner.query(/* sql */ `
+      DROP INDEX IF EXISTS "public"."IDX_user_candidate_preference_locationId"
+    `);
+
+    await queryRunner.query(/* sql */ `
+      ALTER TABLE "user_candidate_preference"
+        DROP COLUMN "locationId"
+    `);
+
+    await queryRunner.query(/* sql */ `
+      ALTER TABLE "user_candidate_preference"
+        RENAME COLUMN "customLocation" TO "location"
+    `);
+
+    await queryRunner.query(/* sql */ `
+      COMMENT ON COLUMN "user_candidate_preference"."location" IS 'Location from protobuf schema'
+    `);
+  }
+}