test notification package

lib/notifications.ts

@@ -1,10 +1,10 @@
-import { sendEmail } from "@lib/integration/notifyConnector";
 import { logMessage } from "@lib/logger";
 import { getRedisInstance } from "@lib/integration/redisConnector";
 import { getOrigin } from "@lib/origin";
 import { NotificationsInterval } from "@gcforms/types";
 import { serverTranslation } from "@i18n";
 import { prisma, prismaErrors } from "@lib/integration/prismaConnector";
+import { notification } from "@gcforms/connectors";
 
 // Hard coded since only one interval is supported currently
 const NOTIFICATIONS_INTERVAL = NotificationsInterval.DAY;
@@ -41,7 +41,7 @@ export const sendNotifications = async (formId: string, titleEn: string, titleFr
     case Status.SINGLE_EMAIL_SENT:
       // Single submissions email sent but not multiple submissions email, send multiple email
       Promise.all([
-        sendEmailNotificationsToAllUsers(users, formId, titleEn, titleFr, true),
+        sendEmailAfterSubmissionProcessed(users, formId, titleEn, titleFr, true),
         setMarker(formId, Status.MULTIPLE_EMAIL_SENT),
       ]);
       break;
@@ -51,7 +51,7 @@ export const sendNotifications = async (formId: string, titleEn: string, titleFr
     default:
       // No email has been sent, send single submission email
       Promise.all([
-        sendEmailNotificationsToAllUsers(users, formId, titleEn, titleFr, false),
+        sendEmailAfterSubmissionProcessed(users, formId, titleEn, titleFr, false),
         setMarker(formId),
       ]);
   }
@@ -80,8 +80,9 @@ export const getNotificationsUsers = async (formId: string) => {
     })
     .catch((e) => prismaErrors(e, null));
 
+  // Can happen with legacy forms that do not have users
   if (!usersAndNotificationsUsers) {
-    logMessage.warn(`_getNotificationsUsers no users found for formId ${formId}`);
+    logMessage.debug(`_getNotificationsUsers no users found for formId ${formId}`);
     return null;
   }
 
@@ -111,8 +112,9 @@ const _getDeliveryOption = async (formId: string) => {
     })
     .catch((e) => prismaErrors(e, null));
 
+  // Can happen with legacy forms that do not have a deliveryOption
   if (!template) {
-    logMessage.warn(`_getDeliveryOption template not found with id ${formId}`);
+    logMessage.debug(`_getDeliveryOption template not found with id ${formId}`);
     return null;
   }
 
@@ -130,18 +132,18 @@ const setMarker = async (formId: string, status: Status = Status.SINGLE_EMAIL_SE
       )
     )
     .catch((err) =>
-      logMessage.error(`setMarker: notification:formId:${formId} failed to set ${err}`)
+      logMessage.warn(`setMarker: notification:formId:${formId} failed to set ${err}`)
     );
 };
 
 const getMarker = async (formId: string) => {
   const redis = await getRedisInstance();
   return redis
     .get(`notification:formId:${formId}`)
-    .catch((err) => logMessage.error(`getMarker: ${err}`));
+    .catch((err) => logMessage.warn(`getMarker: ${err}`));
 };
 
-const sendEmailNotificationsToAllUsers = async (
+const sendEmailAfterSubmissionProcessed = async (
   users: {
     email: string;
     enabled: boolean;
@@ -151,47 +153,31 @@ const sendEmailNotificationsToAllUsers = async (
   formTitleFr: string,
   multipleSubmissions: boolean = false
 ) => {
-  if (!Array.isArray(users) || users.length === 0) {
-    logMessage.error("sendEmailNotificationsToAllUsers missing users");
-    return;
-  }
-  users.forEach(
-    ({ email, enabled }) =>
-      enabled && sendEmailNotification(email, formId, formTitleEn, formTitleFr, multipleSubmissions)
-  );
-};
-
-const sendEmailNotification = async (
-  email: string,
-  formId: string,
-  formTitleEn: string,
-  formTitleFr: string,
-  multipleSubmissions: boolean = false
-) => {
-  const { t } = await serverTranslation("form-builder");
-  const HOST = await getOrigin();
-  await sendEmail(
-    email,
-    {
+  try {
+    const { t } = await serverTranslation("form-builder");
+    const HOST = await getOrigin();
+
+    if (!Array.isArray(users) || users.length === 0) {
+      logMessage.debug("sendEmailNotificationsToAllUsers missing users");
+      return;
+    }
+    const emails = users.filter(({ enabled }) => enabled).map(({ email }) => email);
+
+    await notification.sendDeferred({
+      notificationId: formId,
+      emails,
       subject: multipleSubmissions
         ? t("settings.notifications.email.multipleSubmissions.subject")
         : t("settings.notifications.email.singleSubmission.subject"),
-      formResponse: multipleSubmissions
+      body: multipleSubmissions
         ? await multipleSubmissionsEmailTemplate(HOST, formTitleEn, formTitleFr)
         : await singleSubmissionEmailTemplate(HOST, formTitleEn, formTitleFr),
-    },
-    "notification"
-  )
-    .then(() =>
-      logMessage.debug(
-        `sendEmailNotification sent email to ${email} with formId ${formId} for type ${
-          multipleSubmissions ? "multiple email" : "single email"
-        }`
-      )
-    )
-    .catch(() =>
-      logMessage.error(`sendEmailNotification failed to send email ${email} with formId ${formId}`)
+    });
+  } catch (error) {
+    logMessage.warn(
+      `sendEmailNotification failed for formId ${formId} with error: ${(error as Error).message}`
     );
+  }
 };
 
 const singleSubmissionEmailTemplate = async (

packages/connectors/CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.2.14] - 2025-12-15
+
+- Add notification helpers
+- Add utils for AWS
+
 ## [2.2.13] - 2025-10-27
 
 - Update AWS SDK package from version `3.917.0`

packages/connectors/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@gcforms/connectors",
-  "version": "2.2.13",
+  "version": "2.2.14",
   "author": "Canadian Digital Service",
   "license": "MIT",
   "publishConfig": {

packages/connectors/src/gc-notify-connector.ts

@@ -1,5 +1,5 @@
 import { Agent } from "https";
-import { getAwsSecret } from "./getAwsSecret";
+import { getAwsSecret } from "./utils";
 import axios, { AxiosError } from "axios";
 
 const API_URL: string = "https://api.notification.canada.ca";

packages/connectors/src/index.ts

@@ -1,2 +1,3 @@
 export { GCNotifyConnector, type Personalisation } from "./gc-notify-connector";
 export { PostgresConnector } from "./postgres-connector";
+export { notification } from "./notification";

packages/connectors/src/notification.ts

@@ -0,0 +1,135 @@
+import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
+import { SendMessageCommand, SQSClient } from "@aws-sdk/client-sqs";
+import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";
+import { randomUUID } from "crypto";
+import { ErrorWithCause } from "./types/errors";
+import { getAwsSQSQueueURL } from "./utils";
+
+const DYNAMODB_NOTIFICATION_TABLE_NAME = "Notification";
+
+const globalConfig = {
+  region: process.env.AWS_REGION ?? "ca-central-1",
+};
+
+const dynamoDBDocumentClient = DynamoDBDocumentClient.from(
+  new DynamoDBClient({
+    ...globalConfig,
+    // SDK retries use exponential backoff with jitter by default
+    maxAttempts: 15,
+  })
+);
+
+const sqsClient = new SQSClient({
+  ...globalConfig,
+});
+
+/**
+ * Creates a notification record in DynamoDB and enqueues it for immediate sending.
+ *
+ * @param emails - Array of email addresses to send the notification to
+ * @param subject - Email subject line
+ * @param body - Email body content
+ */
+const sendImmediate = async ({
+  emails,
+  subject,
+  body,
+}: {
+  notificationId?: string;
+  emails: string[];
+  subject: string;
+  body: string;
+}): Promise<void> => {
+  const notificationId = randomUUID();
+  try {
+    await _createRecord({ notificationId, emails, subject, body });
+    await enqueueDeferred(notificationId);
+  } catch (error) {
+    throw new ErrorWithCause(`Error creating immediate notification id ${notificationId}`, {
+      cause: error,
+    });
+  }
+};
+
+/**
+ * Creates a notification record in DynamoDB for deferred sending. Once the related
+ * process is completed it can enqueue the notification for sending by calling
+ * enqueueDeferredNotification with the related notificationId.
+ *
+ * @param notificationId - Unique identifier for the notification to enqueue and
+ *   used by the notification lambda to look up the record in DynamoDB.
+ */
+const sendDeferred = async ({
+  notificationId,
+  emails,
+  subject,
+  body,
+}: {
+  notificationId: string;
+  emails: string[];
+  subject: string;
+  body: string;
+}): Promise<void> => {
+  try {
+    await _createRecord({ notificationId, emails, subject, body });
+  } catch (error) {
+    throw new ErrorWithCause(`Error creating deferred notification id ${notificationId}`, {
+      cause: error,
+    });
+  }
+};
+
+const _createRecord = async ({
+  notificationId,
+  emails,
+  subject,
+  body,
+}: {
+  notificationId: string;
+  emails: string[];
+  subject: string;
+  body: string;
+}): Promise<void> => {
+  try {
+    const ttl = Math.floor(Date.now() / 1000) + 86400; // 24 hours from now
+    const command = new PutCommand({
+      TableName: DYNAMODB_NOTIFICATION_TABLE_NAME,
+      Item: {
+        NotificationID: notificationId,
+        Emails: emails,
+        Subject: subject,
+        Body: body,
+        TTL: ttl,
+      },
+    });
+    await dynamoDBDocumentClient.send(command);
+  } catch (error) {
+    throw new ErrorWithCause(`Could not create record`, { cause: error });
+  }
+};
+
+const enqueueDeferred = async (notificationId: string): Promise<void> => {
+  try {
+    const queueUrl = await getAwsSQSQueueURL("NOTIFICATION_QUEUE_URL", "notification_queue");
+    if (!queueUrl) {
+      throw new Error("Notification Queue not connected");
+    }
+
+    const command = new SendMessageCommand({
+      MessageBody: JSON.stringify({ notificationId }),
+      QueueUrl: queueUrl,
+    });
+    const sendMessageCommandOutput = await sqsClient.send(command);
+    if (!sendMessageCommandOutput.MessageId) {
+      throw new Error("Received null SQS message identifier");
+    }
+  } catch (error) {
+    throw new ErrorWithCause(`Could not enqueue`, { cause: error });
+  }
+};
+
+export const notification = {
+  sendImmediate,
+  sendDeferred,
+  enqueueDeferred,
+};

packages/connectors/src/postgres-connector.ts

@@ -1,4 +1,4 @@
-import { getAwsSecret } from "./getAwsSecret";
+import { getAwsSecret } from "./utils";
 import postgres, { Sql } from "postgres";
 
 export class PostgresConnector {

packages/connectors/src/types/errors.ts

@@ -0,0 +1,17 @@
+/**
+ * Custom Error class that supports adding a "cause" for re-throwing errors.
+ */
+export class ErrorWithCause extends Error {
+  cause?: unknown;
+
+  constructor(message: string, options?: { cause?: unknown }) {
+    super(message);
+    this.name = "ErrorWithCause";
+    this.cause = options?.cause;
+
+    // Maintains proper stack trace
+    if (Error.captureStackTrace) {
+      Error.captureStackTrace(this, ErrorWithCause);
+    }
+  }
+}

packages/connectors/src/utils.ts

@@ -0,0 +1,32 @@
+import { SQSClient, GetQueueUrlCommand } from "@aws-sdk/client-sqs";
+import { SecretsManagerClient, GetSecretValueCommand } from "@aws-sdk/client-secrets-manager";
+
+const globalConfig = {
+  region: process.env.AWS_REGION ?? "ca-central-1",
+};
+
+const sqsClient = new SQSClient({
+  ...globalConfig,
+});
+
+export function getAwsSecret(secretIdentifier: string): Promise<string | undefined> {
+  return new SecretsManagerClient()
+    .send(new GetSecretValueCommand({ SecretId: secretIdentifier }))
+    .then((commandOutput) => commandOutput.SecretString);
+}
+
+export const getAwsSQSQueueURL = async (
+  urlEnvName: string,
+  urlQueueName: string
+): Promise<string | null> => {
+  if (process.env[urlEnvName]) {
+    return process.env[urlEnvName];
+  }
+
+  const data = await sqsClient.send(
+    new GetQueueUrlCommand({
+      QueueName: urlQueueName,
+    })
+  );
+  return data.QueueUrl ?? null;
+};