fix: add update topic (#3364)

Author: rebelchris

__tests__/workers/cdc/primary.ts

@@ -6917,10 +6917,10 @@ describe('organization', () => {
 
     expect(triggerTypedEvent).toHaveBeenCalledTimes(2);
     expect(jest.mocked(triggerTypedEvent).mock.calls[0][1]).toEqual(
-      'api.v1.opportunity-added',
+      'api.v1.opportunity-updated',
     );
     expect(jest.mocked(triggerTypedEvent).mock.calls[1][1]).toEqual(
-      'api.v1.opportunity-added',
+      'api.v1.opportunity-updated',
     );
   });
 

src/common/opportunity/pubsub.ts

@@ -334,10 +334,12 @@ export const notifyJobOpportunity = async ({
   con,
   logger,
   opportunityId,
+  isUpdate = false,
 }: {
   con: DataSource;
   logger: FastifyBaseLogger;
   opportunityId: string;
+  isUpdate?: boolean;
 }) => {
   const [opportunity, organization, keywords, users, locations] =
     await queryReadReplica(con, async ({ queryRunner }) => {
@@ -456,8 +458,11 @@ export const notifyJobOpportunity = async ({
     excludedUserIds,
   });
 
+  const topicName = isUpdate
+    ? 'api.v1.opportunity-updated'
+    : 'api.v1.opportunity-added';
   try {
-    await triggerTypedEvent(logger, 'api.v1.opportunity-added', message);
+    await triggerTypedEvent(logger, topicName, message);
   } catch (_err) {
     const err = _err as Error;
     logger.error({ err, message }, 'failed to send opportunity event');

src/schema/opportunity.ts

@@ -2564,7 +2564,7 @@ export const resolvers: IResolvers<unknown, BaseContext> = traceResolvers<
         .count({
           where: {
             organizationId: organization.id,
-            state: OpportunityState.LIVE,
+            state: In([OpportunityState.LIVE, OpportunityState.IN_REVIEW]),
             flags: JsonContains({ plan: priceId }),
           },
         });

src/workers/cdc/primary.ts

@@ -1382,15 +1382,22 @@ const onOpportunityChange = async (
   if (
     data.payload.op === 'u' &&
     data.payload.after?.type === OpportunityType.JOB &&
-    data.payload.before?.state === OpportunityState.LIVE &&
-    data.payload.after?.state !== OpportunityState.LIVE
+    data.payload.before?.state === OpportunityState.LIVE
   ) {
-    await con
-      .getRepository(Alerts)
-      .update(
-        { opportunityId: data.payload.after!.id },
-        { opportunityId: null },
-      );
+    await notifyJobOpportunity({
+      con,
+      logger,
+      opportunityId: data.payload.after!.id,
+      isUpdate: true,
+    });
+    if (data.payload.after?.state !== OpportunityState.LIVE) {
+      await con
+        .getRepository(Alerts)
+        .update(
+          { opportunityId: data.payload.after!.id },
+          { opportunityId: null },
+        );
+    }
   }
 };
 
@@ -1440,6 +1447,7 @@ const onOrganizationChange = async (
           con,
           logger,
           opportunityId: opportunity.id,
+          isUpdate: true,
         });
       }),
     );