Merge pull request #773 from techmatters/SJH-taxonomy_tweak

Tweak taxonomy mappings for KHP imports - trivial PR, no review required

hrm-domain/hrm-core/contact/contactsReindexService.ts

@@ -56,7 +56,7 @@ export const reindexContactsStream = async (
           const { MessageId } = await publishContactToSearchIndex({
             accountSid,
             contact,
-            operation: 'index',
+            operation: 'reindex',
           });
 
           this.push(

hrm-domain/hrm-core/jobs/search/publishToSearchIndex.ts

@@ -19,11 +19,41 @@ import { getSsmParameter } from '../../config/ssmCache';
 import { IndexMessage } from '@tech-matters/hrm-search-config';
 import { CaseService, Contact } from '@tech-matters/hrm-types';
 import { AccountSID, HrmAccountId } from '@tech-matters/types';
+import { publishSns, PublishSnsParams } from '@tech-matters/sns-client';
+import { SsmParameterNotFound } from '@tech-matters/ssm-cache';
+
+type DeleteNotificationPayload = {
+  accountSid: HrmAccountId;
+  operation: 'delete';
+  id: string;
+};
+
+type UpsertCaseNotificationPayload = {
+  accountSid: HrmAccountId;
+  operation: 'update' | 'create' | 'reindex';
+  case: CaseService;
+};
+
+type UpsertContactNotificationPayload = {
+  accountSid: HrmAccountId;
+  operation: 'update' | 'create' | 'reindex';
+  contact: Contact;
+};
+
+type NotificationPayload =
+  | DeleteNotificationPayload
+  | UpsertCaseNotificationPayload
+  | UpsertContactNotificationPayload;
 
 const PENDING_INDEX_QUEUE_SSM_PATH = `/${process.env.NODE_ENV}/${
   process.env.AWS_REGION ?? process.env.AWS_DEFAULT_REGION
 }/sqs/jobs/hrm-search-index/queue-url-consumer`;
 
+const getSnsSsmPath = dataType =>
+  `/${process.env.NODE_ENV}/${
+    process.env.AWS_REGION ?? process.env.AWS_DEFAULT_REGION
+  }/sns/hrm/${dataType}/update-notifications-sns-topic`;
+
 const publishToSearchIndex = async ({
   message,
   messageGroupId,
@@ -46,14 +76,89 @@ const publishToSearchIndex = async ({
   }
 };
 
+const publishToSns = async ({
+  entityType,
+  payload,
+  messageGroupId,
+}: {
+  entityType: 'contact' | 'case';
+  payload: NotificationPayload;
+  messageGroupId: string;
+}) => {
+  try {
+    const topicArn = await getSsmParameter(getSnsSsmPath(entityType));
+    const publishParameters: PublishSnsParams = {
+      topicArn,
+      message: JSON.stringify({ ...payload, entityType }),
+      messageGroupId,
+      messageAttributes: { operation: payload.operation },
+    };
+    console.debug('Publishing HRM entity update:', publishParameters);
+    return await publishSns(publishParameters);
+  } catch (err) {
+    if (err instanceof SsmParameterNotFound) {
+      console.debug(
+        `No SNS topic stored in SSM parameter ${getSnsSsmPath(
+          entityType,
+        )}. Skipping publish.`,
+      );
+      return;
+    }
+    console.error(
+      `Error trying to publish message to SNS topic stored in SSM parameter ${getSnsSsmPath(
+        entityType,
+      )}`,
+      err,
+    );
+  }
+};
+
+const publishEntityToSearchIndex = async (
+  accountSid: HrmAccountId,
+  entityType: 'contact' | 'case',
+  entity: Contact | CaseService,
+  operation: IndexMessage['operation'] | 'reindex',
+) => {
+  const messageGroupId = `${accountSid}-${entityType}-${entity.id}`;
+  if (operation === 'remove') {
+    await publishToSns({
+      entityType,
+      payload: { accountSid, id: entity.id.toString(), operation: 'delete' },
+      messageGroupId,
+    });
+  } else {
+    const publishOperation = operation === 'reindex' ? 'reindex' : 'update';
+    await publishToSns({
+      entityType,
+      // Update / create are identical for now, will differentiate between the 2 ops in a follow pu refactor PR
+      payload: {
+        accountSid,
+        [entityType]: entity,
+        operation: publishOperation,
+      } as NotificationPayload,
+      messageGroupId,
+    });
+  }
+  const indexOperation = operation === 'reindex' ? 'index' : operation;
+  return publishToSearchIndex({
+    message: {
+      accountSid,
+      type: entityType,
+      [entityType]: entity,
+      operation: indexOperation,
+    } as IndexMessage,
+    messageGroupId: `${accountSid}-${entityType}-${entity.id}`,
+  });
+};
+
 export const publishContactToSearchIndex = async ({
   accountSid,
   contact,
   operation,
 }: {
   accountSid: HrmAccountId;
   contact: Contact;
-  operation: IndexMessage['operation'];
+  operation: IndexMessage['operation'] | 'reindex';
 }) => {
   console.info(
     `[generalised-search-contacts]: Indexing Started. Account SID: ${accountSid}, Contact ID: ${
@@ -64,10 +169,7 @@ export const publishContactToSearchIndex = async ({
       contact.updatedAt ?? contact.createdAt
     }/${operation})`,
   );
-  return publishToSearchIndex({
-    message: { accountSid, type: 'contact', contact, operation },
-    messageGroupId: `${accountSid}-contact-${contact.id}`,
-  });
+  return publishEntityToSearchIndex(accountSid, 'contact', contact, operation);
 };
 
 export const publishCaseToSearchIndex = async ({
@@ -77,7 +179,7 @@ export const publishCaseToSearchIndex = async ({
 }: {
   accountSid: AccountSID;
   case: CaseService;
-  operation: IndexMessage['operation'];
+  operation: IndexMessage['operation'] | 'reindex';
 }) => {
   console.info(
     `[generalised-search-cases]: Indexing Request Started. Account SID: ${accountSid}, Case ID: ${
@@ -88,8 +190,5 @@ export const publishCaseToSearchIndex = async ({
       caseObj.updatedAt ?? caseObj.createdAt
     }/${operation})`,
   );
-  return publishToSearchIndex({
-    message: { accountSid, type: 'case', case: caseObj, operation },
-    messageGroupId: `${accountSid}-case-${caseObj.id}`,
-  });
+  return publishEntityToSearchIndex(accountSid, 'case', caseObj, operation);
 };

hrm-domain/hrm-core/package.json

@@ -30,6 +30,7 @@
     "@tech-matters/resources-service": "^1.0.0",
     "@tech-matters/s3-client": "^1.0.0",
     "@tech-matters/sqs-client": "^1.0.0",
+    "@tech-matters/sns-client": "^1.0.0",
     "@tech-matters/ssm-cache": "^1.0.0",
     "@tech-matters/twilio-client": "^1.0.0",
     "@tech-matters/twilio-worker-auth": "^1.0.0",

packages/sns-client/index.ts

@@ -14,7 +14,12 @@
  * along with this program.  If not, see https://www.gnu.org/licenses/.
  */
 
-import { PublishCommand, SNSClient, SNSClientConfig } from '@aws-sdk/client-sns';
+import {
+  MessageAttributeValue,
+  PublishCommand,
+  SNSClient,
+  SNSClientConfig,
+} from '@aws-sdk/client-sns';
 
 const convertToEndpoint = (endpointUrl: string) => {
   const url: URL = new URL(endpointUrl);
@@ -46,14 +51,23 @@ export const sns = new SNSClient(getSnsConfig());
 export type PublishSnsParams = {
   topicArn: string;
   message: string;
+  messageGroupId?: string;
+  messageAttributes?: Record<string, string>;
 };
 
 export const publishSns = async (params: PublishSnsParams) => {
-  const { topicArn, message } = params;
+  const { topicArn, message, messageGroupId, messageAttributes } = params;
+  const messageAttributesPayload: Record<string, MessageAttributeValue> = {};
+  Object.entries(messageAttributes ?? {}).forEach(
+    ([key, value]) =>
+      (messageAttributesPayload[key] = { DataType: 'String', StringValue: value }),
+  );
 
   const command = new PublishCommand({
     TopicArn: topicArn,
     Message: message,
+    MessageAttributes: messageAttributesPayload,
+    ...(messageGroupId ? { MessageGroupId: messageGroupId } : {}),
   });
 
   return sns.send(command);

resources-domain/lambdas/import-producer/src/khpMappings.ts

@@ -251,8 +251,10 @@ const KHP_MAPPING_NODE_SITES: { children: MappingNode } = {
   },
 };
 
+const MAX_TAXONOMY_DEPTH = 5;
+
 // TODO: this is an array of arrays, is this shape correct?
-const KHP_MAPPING_NODE_TAXONOMIES: { children: MappingNode } = {
+const KHP_MAPPING_NODE_TAXONOMIES = (depth: number = 0): { children: MappingNode } => ({
   children: {
     '{taxonomyIndex}': {
       children: {
@@ -272,6 +274,10 @@ const KHP_MAPPING_NODE_TAXONOMIES: { children: MappingNode } = {
             info: ctx => ctx.parentValue,
           },
         ),
+        ...(depth + 1 <= MAX_TAXONOMY_DEPTH
+          ? { ancestors: KHP_MAPPING_NODE_TAXONOMIES(depth + 1) }
+          : {}),
+        // Deprecated
         ancestorTaxonomies: {
           children: {
             '{ancestorIndex}': {
@@ -299,15 +305,20 @@ const KHP_MAPPING_NODE_TAXONOMIES: { children: MappingNode } = {
       },
     },
   },
-};
+});
 
 export const KHP_MAPPING_NODE: MappingNode = {
   _id: resourceFieldMapping('id'),
   // TODO: Remove this and all other 'objectId' mappings once the updated Arctic API is deployed to production
   objectId: resourceFieldMapping('id'),
   timeSequence: resourceFieldMapping('importSequenceId'),
   sites: KHP_MAPPING_NODE_SITES,
-  taxonomies: KHP_MAPPING_NODE_TAXONOMIES,
+  taxonomies: {
+    ...KHP_MAPPING_NODE_TAXONOMIES(),
+    children: {
+      '{topIndex}': KHP_MAPPING_NODE_TAXONOMIES(),
+    },
+  },
   name: resourceFieldMapping('name', ctx => ctx.currentValue.en || ctx.currentValue.fr),
   updatedAt: resourceFieldMapping('lastUpdated'),
   createdAt: attributeMapping('dateTimeAttributes', 'sourceCreatedAt'),
@@ -677,13 +688,14 @@ export const KHP_MAPPING_NODE: MappingNode = {
         children: {
           objectId: { children: {} },
           _id: { children: {} },
+          __v: {},
+          name: {},
           '{language}': referenceAttributeMapping(
             'feeStructure/{feeStructureSourceIndex}',
             'khp-fee-structure-source',
             {
-              value: ctx => ctx.parentValue.en,
+              value: ctx => ctx.parentValue[ctx.captures.language],
               language: ctx => ctx.captures.language,
-              // value: ctx => ctx.currentValue.en || ctx.currentValue.fr,
             },
           ),
         },
@@ -695,6 +707,10 @@ export const KHP_MAPPING_NODE: MappingNode = {
     children: {
       '{howIsServiceOfferedIndex}': {
         children: {
+          objectId: { children: {} },
+          _id: { children: {} },
+          __v: {},
+          name: {},
           '{language}': referenceAttributeMapping(
             'howIsServiceOffered/{howIsServiceOfferedIndex}',
             'khp-how-is-service-offered',

resources-domain/lambdas/import-producer/tests/fixtures/sampleResources.ts

@@ -7994,7 +7994,13 @@ export const withSites_20240823 = {
   },
   eligibilityMaxAge: 30,
   eligibilityMinAge: 18,
-  feeStructureSource: [],
+  feeStructureSource: [
+    {
+      en: 'Free',
+      fr: 'Gratuit',
+      objectId: '66635363033cc0b95a022076',
+    },
+  ],
   feeStructureSourceFreeTextEn: null,
   feeStructureSourceFreeTextFr: null,
   howIsServiceOffered: [