feat: add notification and AWS utils to connector package

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;
+};