feat: sync user_company to CIO

.infra/application.properties

@@ -8,7 +8,7 @@ debezium.source.database.user=%database_user%
 debezium.source.database.password=%database_pass%
 debezium.source.database.dbname=%database_dbname%
 debezium.source.database.server.name=api
-debezium.source.table.include.list=public.comment,public.user_comment,public.comment_mention,public.source_request,public.post,public.user,public.post_report,public.source_feed,public.settings,public.reputation_event,public.submission,public.user_state,public.notification_v2,public.source_member,public.feature,public.source,public.post_mention,public.content_image,public.comment_report,public.user_post,public.banner,public.post_relation,public.marketing_cta,public.squad_public_request,public.user_streak,public.bookmark
+debezium.source.table.include.list=public.comment,public.user_comment,public.comment_mention,public.source_request,public.post,public.user,public.post_report,public.source_feed,public.settings,public.reputation_event,public.submission,public.user_state,public.notification_v2,public.source_member,public.feature,public.source,public.post_mention,public.content_image,public.comment_report,public.user_post,public.banner,public.post_relation,public.marketing_cta,public.squad_public_request,public.user_streak,public.bookmark,public.user_company
 debezium.source.column.exclude.list=public.post.tsv,public.post.placeholder,public.source.flags
 debezium.source.plugin.name=pgoutput
 debezium.source.heartbeat.interval.ms=60000

.infra/common.ts

@@ -285,6 +285,10 @@ export const workers: Worker[] = [
     topic: 'api.v1.post-visible',
     subscription: 'api.vordr-post-prevented',
   },
+  {
+    topic: 'api.v1.user-company-approved',
+    subscription: 'api.user-company-approved-cio',
+  },
 ];
 
 export const personalizedDigestWorkers: Worker[] = [

__tests__/workers/cdc/primary.ts

@@ -27,6 +27,7 @@ import {
   Bookmark,
   UserStreakAction,
   UserStreakActionType,
+  UserCompany,
 } from '../../../src/entity';
 import {
   notifyCommentCommented,
@@ -3564,6 +3565,84 @@ describe('user streak change', () => {
   });
 });
 
+describe('user company approved', () => {
+  type ObjectType = UserCompany;
+  const base: ChangeObject<ObjectType> = {
+    userId: '1',
+    code: '123456',
+    email: '[email protected]',
+    verified: true,
+    createdAt: new Date().getTime(),
+    updatedAt: new Date().getTime(),
+    companyId: null,
+    flags: {},
+  };
+
+  it('should not notify on creation when company id not set', async () => {
+    const after: ChangeObject<ObjectType> = base;
+    await expectSuccessfulBackground(
+      worker,
+      mockChangeMessage<ObjectType>({
+        after,
+        before: null,
+        op: 'c',
+        table: 'user_company',
+      }),
+    );
+    expect(triggerTypedEvent).not.toHaveBeenCalled();
+  });
+
+  it('should notify on creation when company id is set', async () => {
+    const after: ChangeObject<ObjectType> = { ...base, companyId: '1' };
+    await expectSuccessfulBackground(
+      worker,
+      mockChangeMessage<ObjectType>({
+        after,
+        before: null,
+        op: 'c',
+        table: 'user_company',
+      }),
+    );
+    expect(triggerTypedEvent).toHaveBeenCalled();
+    expect(jest.mocked(triggerTypedEvent).mock.calls[0].slice(1)).toEqual([
+      'api.v1.user-company-approved',
+      { userCompany: after },
+    ]);
+  });
+
+  it('should not notify on update when company not changed', async () => {
+    const after: ChangeObject<ObjectType> = base;
+    await expectSuccessfulBackground(
+      worker,
+      mockChangeMessage<ObjectType>({
+        after,
+        before: { ...after },
+        op: 'u',
+        table: 'user_company',
+      }),
+    );
+    expect(triggerTypedEvent).not.toHaveBeenCalled();
+  });
+
+  it('should notify on update when company id is changed', async () => {
+    const after: ChangeObject<ObjectType> = { ...base, companyId: '1' };
+    await expectSuccessfulBackground(
+      worker,
+      mockChangeMessage<ObjectType>({
+        after,
+        before: base,
+        op: 'u',
+        table: 'user_company',
+      }),
+    );
+    expect(triggerTypedEvent).toHaveBeenCalled();
+    expect(jest.mocked(triggerTypedEvent).mock.calls[0].slice(1)).toEqual([
+      'api.v1.user-company-approved',
+      { userCompany: after },
+    ]);
+  });
+});
+
 describe('bookmark change', () => {
   type ObjectType = Bookmark;
   const base: ChangeObject<ObjectType> = {

__tests__/workers/userCompanyApprovedCio.ts

@@ -0,0 +1,101 @@
+import { expectSuccessfulTypedBackground, saveFixtures } from '../helpers';
+import worker from '../../src/workers/userCompanyApprovedCio';
+import { User } from '../../src/entity';
+import { DataSource } from 'typeorm';
+import createOrGetConnection from '../../src/db';
+import { typedWorkers } from '../../src/workers';
+import { usersFixture } from '../fixture';
+import nock from 'nock';
+import { ChangeObject } from '../../src/types';
+import { PubSubSchema } from '../../src/common';
+import { cioV2 as cio } from '../../src/cio';
+import { UserCompany } from '../../src/entity/UserCompany';
+import { Company } from '../../src/entity/Company';
+
+jest.mock('../../src/cio', () => ({
+  ...(jest.requireActual('../../src/cio') as Record<string, unknown>),
+  cioV2: { request: { post: jest.fn() } },
+}));
+
+let con: DataSource;
+
+beforeAll(async () => {
+  con = await createOrGetConnection();
+});
+
+beforeEach(async () => {
+  jest.clearAllMocks();
+  nock.cleanAll();
+  process.env.CIO_SITE_ID = 'wolololo';
+});
+
+describe('userCompanyApprovedCio worker', () => {
+  type ObjectType = Partial<UserCompany>;
+  const base: ChangeObject<ObjectType> = {
+    userId: '1',
+    code: '123456',
+    email: '[email protected]',
+    verified: true,
+    createdAt: new Date().getTime(),
+    updatedAt: new Date().getTime(),
+    companyId: 'dailydev',
+    flags: {},
+  };
+
+  beforeEach(async () => {
+    jest.resetAllMocks();
+    await saveFixtures(con, User, [usersFixture[0]]);
+    await con.getRepository(Company).save({
+      id: 'dailydev',
+      name: 'daily.dev',
+      image: 'cloudinary.com/dailydev/121232121/image',
+    });
+  });
+
+  it('should be registered', () => {
+    const registeredWorker = typedWorkers.find(
+      (item) => item.subscription === worker.subscription,
+    );
+
+    expect(registeredWorker).toBeDefined();
+  });
+
+  it('should not update if no company id provided', async () => {
+    await expectSuccessfulTypedBackground(worker, {
+      userCompany: { ...base, companyId: null },
+    } as unknown as PubSubSchema['api.v1.user-company-approved']);
+    expect(cio.request.post).not.toHaveBeenCalled();
+  });
+
+  it('should not update if no company is not found', async () => {
+    await expectSuccessfulTypedBackground(worker, {
+      userCompany: { ...base, companyId: '1' },
+    } as unknown as PubSubSchema['api.v1.user-company-approved']);
+    expect(cio.request.post).not.toHaveBeenCalled();
+  });
+
+  it('should update customer.io', async () => {
+    await expectSuccessfulTypedBackground(worker, {
+      userCompany: base,
+    } as unknown as PubSubSchema['api.v1.user-company-approved']);
+    expect(cio.request.post).toHaveBeenCalledWith('undefined/entity', {
+      identifiers: {
+        object_type_id: '4',
+        object_id: 'dailydev',
+      },
+      type: 'object',
+      action: 'identify',
+      attributes: {
+        name: 'daily.dev',
+        image: 'cloudinary.com/dailydev/121232121/image',
+      },
+      cio_relationships: [
+        {
+          identifiers: {
+            id: '1',
+          },
+        },
+      ],
+    });
+  });
+});

src/cio.ts

@@ -9,12 +9,26 @@ import {
   getShortGenericInviteLink,
 } from './common';
 import { FastifyBaseLogger } from 'fastify';
+import type { UserCompany } from './entity/UserCompany';
+import type { Company } from './entity/Company';
 
 export const cio = new TrackClient(
   process.env.CIO_SITE_ID,
   process.env.CIO_API_KEY,
 );
 
+/**
+ * Specific client for using v2 of the Customer.io API.
+ */
+export const cioV2 = new TrackClient(
+  process.env.CIO_SITE_ID,
+  process.env.CIO_API_KEY,
+  { url: 'https://track.customer.io/api/v2' },
+);
+
+// Magic number from customer.io
+const CIO_COMPANY_OBJECT_ID = '4';
+
 export function dateToCioTimestamp(date: Date): number {
   return Math.floor(date.getTime() / 1000);
 }
@@ -102,3 +116,38 @@ export async function identifyUser(
     throw err;
   }
 }
+
+export async function identifyUserCompany(
+  log: FastifyBaseLogger,
+  cio: TrackClient,
+  userCompany: ChangeObject<UserCompany>,
+  company: Company,
+): Promise<void> {
+  try {
+    await cio.request.post(`${cio.trackRoot}/entity`, {
+      identifiers: {
+        object_type_id: CIO_COMPANY_OBJECT_ID,
+        object_id: company.id,
+      },
+      type: 'object',
+      action: 'identify',
+      attributes: {
+        name: company.name,
+        image: company.image,
+      },
+      cio_relationships: [
+        {
+          identifiers: {
+            id: userCompany.userId,
+          },
+        },
+      ],
+    });
+  } catch (err) {
+    if (err instanceof CustomerIORequestError && err.statusCode === 400) {
+      log.warn({ err }, 'failed to update user company in cio');
+      return;
+    }
+    throw err;
+  }
+}

src/common/typedPubsub.ts

@@ -1,10 +1,11 @@
 import { ChangeObject } from '../types';
-import {
+import type {
   Post,
   SourceRequest,
   SquadPublicRequest,
   User,
   UserStreak,
+  UserCompany,
 } from '../entity';
 import {
   EventLogger,
@@ -88,6 +89,9 @@ export type PubSubSchema = {
   'api.v1.post-visible': {
     post: ChangeObject<Post>;
   };
+  'api.v1.user-company-approved': {
+    userCompany: ChangeObject<UserCompany>;
+  };
 };
 
 export async function triggerTypedEvent<T extends keyof PubSubSchema>(

src/entity/index.ts

@@ -32,3 +32,4 @@ export * from './Invite';
 export * from './ContentImage';
 export * from './MarketingCta';
 export * from './SquadPublicRequest';
+export * from './UserCompany';

src/workers/cdc/primary.ts

@@ -36,6 +36,7 @@ import {
   PostRelationType,
   normalizeCollectionPostSources,
   CollectionPost,
+  UserCompany,
 } from '../../entity';
 import {
   notifyCommentCommented,
@@ -878,6 +879,24 @@ const onUserStreakChange = async (
   }
 };
 
+const onUserCompanyCompanyChange = async (
+  _: DataSource,
+  logger: FastifyBaseLogger,
+  data: ChangeMessage<UserCompany>,
+) => {
+  const creationWithCompany =
+    data.payload.op === 'c' && !!data.payload.after?.companyId;
+  const updateWithDifferentCompany =
+    data.payload.op === 'u' &&
+    !!data.payload.after?.companyId &&
+    data.payload.before?.companyId !== data.payload.after?.companyId;
+  if (creationWithCompany || updateWithDifferentCompany) {
+    await triggerTypedEvent(logger, 'api.v1.user-company-approved', {
+      userCompany: data.payload.after!,
+    });
+  }
+};
+
 const onBookmarkChange = async (
   con: DataSource,
   logger: FastifyBaseLogger,
@@ -993,6 +1012,9 @@ const worker: Worker = {
         case getTableName(con, Bookmark):
           await onBookmarkChange(con, logger, data);
           break;
+        case getTableName(con, UserCompany):
+          await onUserCompanyCompanyChange(con, logger, data);
+          break;
       }
     } catch (err) {
       logger.error(

src/workers/index.ts

@@ -52,6 +52,7 @@ import userStreakUpdatedCio from './userStreakUpdatedCio';
 import { vordrPostCommentPrevented } from './vordrPostCommentPrevented';
 import { vordrPostPrevented } from './vordrPostPrevented';
 import { postAddedSlackChannelSendWorker } from './postAddedSlackChannelSend';
+import userCompanyApprovedCio from './userCompanyApprovedCio';
 
 export { Worker } from './worker';
 
@@ -107,6 +108,7 @@ export const typedWorkers: BaseTypedWorker<any>[] = [
   vordrPostCommentPrevented,
   vordrPostPrevented,
   postAddedSlackChannelSendWorker,
+  userCompanyApprovedCio,
 ];
 
 export const personalizedDigestWorkers: Worker[] = [