Feature/ccm 12934 create letter queue

infrastructure/terraform/components/api/README.md

@@ -66,6 +66,7 @@ No requirements.
 | <a name="module_sqs_supplier_allocator"></a> [sqs\_supplier\_allocator](#module\_sqs\_supplier\_allocator) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.26/terraform-sqs.zip | n/a |
 | <a name="module_supplier_allocator"></a> [supplier\_allocator](#module\_supplier\_allocator) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.29/terraform-lambda.zip | n/a |
 | <a name="module_supplier_ssl"></a> [supplier\_ssl](#module\_supplier\_ssl) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.26/terraform-ssl.zip | n/a |
+| <a name="module_update_letter_queue"></a> [update\_letter\_queue](#module\_update\_letter\_queue) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.29/terraform-lambda.zip | n/a |
 | <a name="module_upsert_letter"></a> [upsert\_letter](#module\_upsert\_letter) | https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.29/terraform-lambda.zip | n/a |
 ## Outputs
 

infrastructure/terraform/components/api/ddb_table_letter_queue.tf

@@ -0,0 +1,44 @@
+resource "aws_dynamodb_table" "letter-queue" {
+  name         = "${local.csi}-letter-queue"
+  billing_mode = "PAY_PER_REQUEST"
+
+  hash_key  = "supplierId"
+  range_key = "queueTimestamp"
+
+  ttl {
+    attribute_name = "ttl"
+    enabled        = true
+  }
+
+  local_secondary_index {
+    name            = "timestamp-index"
+    range_key       = "letterId"
+    projection_type = "ALL"
+  }
+
+  attribute {
+    name = "supplierId"
+    type = "S"
+  }
+
+  attribute {
+    name = "letterId"
+    type = "S"
+  }
+
+  attribute {
+    name = "queueTimestamp"
+    type = "S"
+  }
+
+  point_in_time_recovery {
+    enabled = true
+  }
+
+  tags = merge(
+    local.default_tags,
+    {
+      NHSE-Enable-Dynamo-Backup-Acct = "True"
+    }
+  )
+}

infrastructure/terraform/components/api/event_source_mapping_update_letter_queue.tf

@@ -0,0 +1,11 @@
+resource "aws_lambda_event_source_mapping" "update_letter_queue_kinesis" {
+  event_source_arn                   = aws_kinesis_stream.letter_change_stream.arn
+  function_name                      = module.update_letter_queue.function_arn
+  starting_position                  = "LATEST"
+  batch_size                         = 10
+  maximum_batching_window_in_seconds = 1
+
+  depends_on = [
+    module.update_letter_queue # ensures update letter queue lambda exists
+  ]
+}

infrastructure/terraform/components/api/module_lambda_update_letter_queue.tf

@@ -0,0 +1,75 @@
+module "update_letter_queue" {
+  source = "https://github.com/NHSDigital/nhs-notify-shared-modules/releases/download/v2.0.29/terraform-lambda.zip"
+
+  function_name = "update-letter-queue"
+  description   = "Populates the letter queue table with new pending letters from the letter change stream"
+
+  aws_account_id = var.aws_account_id
+  component      = var.component
+  environment    = var.environment
+  project        = var.project
+  region         = var.region
+  group          = var.group
+
+  log_retention_in_days = var.log_retention_in_days
+  kms_key_arn           = module.kms.key_arn
+
+  iam_policy_document = {
+    body = data.aws_iam_policy_document.update_letter_queue_lambda.json
+  }
+
+  function_s3_bucket      = local.acct.s3_buckets["lambda_function_artefacts"]["id"]
+  function_code_base_path = local.aws_lambda_functions_dir_path
+  function_code_dir       = "update-letter-queue/dist"
+  function_include_common = true
+  handler_function_name   = "handler"
+  runtime                 = "nodejs22.x"
+  memory                  = 512
+  timeout                 = 29
+  log_level               = var.log_level
+
+  force_lambda_code_deploy = var.force_lambda_code_deploy
+  enable_lambda_insights   = false
+
+  log_destination_arn       = local.destination_arn
+  log_subscription_role_arn = local.acct.log_subscription_role_arn
+
+  lambda_env_vars = merge(local.common_lambda_env_vars, {
+    LETTER_QUEUE_TABLE_NAME = aws_dynamodb_table.letter-queue.name,
+    LETTER_QUEUE_TTL_HOURS  = 168 # 7 days
+  })
+}
+
+data "aws_iam_policy_document" "update_letter_queue_lambda" {
+  statement {
+    sid    = "AllowDynamoDBWrite"
+    effect = "Allow"
+
+    actions = [
+      "dynamodb:PutItem",
+    ]
+
+    resources = [
+      aws_dynamodb_table.letter-queue.arn,
+      "${aws_dynamodb_table.letter-queue.arn}/index/*"
+    ]
+  }
+
+  statement {
+    sid    = "AllowKinesisGet"
+    effect = "Allow"
+
+    actions = [
+      "kinesis:GetRecords",
+      "kinesis:GetShardIterator",
+      "kinesis:DescribeStream",
+      "kinesis:DescribeStreamSummary",
+      "kinesis:ListShards",
+      "kinesis:ListStreams",
+    ]
+
+    resources = [
+      aws_kinesis_stream.letter_change_stream.arn
+    ]
+  }
+}

internal/datastore/src/__test__/db.ts

@@ -30,9 +30,11 @@ export async function setupDynamoDBContainer() {
     region: "us-west-2",
     endpoint,
     lettersTableName: "letters",
+    letterQueueTableName: "letter-queue",
     miTableName: "management-info",
     suppliersTableName: "suppliers",
     lettersTtlHours: 1,
+    letterQueueTtlHours: 1,
     miTtlHours: 1,
   };
 
@@ -118,6 +120,32 @@ const createSupplierTableCommand = new CreateTableCommand({
   ],
 });
 
+const createLetterQueueTableCommand = new CreateTableCommand({
+  TableName: "letter-queue",
+  BillingMode: "PAY_PER_REQUEST",
+  KeySchema: [
+    { AttributeName: "supplierId", KeyType: "HASH" }, // Partition key
+    { AttributeName: "letterId", KeyType: "RANGE" }, // Sort key
+  ],
+  LocalSecondaryIndexes: [
+    {
+      IndexName: "timestamp-index",
+      KeySchema: [
+        { AttributeName: "supplierId", KeyType: "HASH" }, // Partition key for LSI
+        { AttributeName: "queueTimestamp", KeyType: "RANGE" }, // Sort key for LSI
+      ],
+      Projection: {
+        ProjectionType: "ALL",
+      },
+    },
+  ],
+  AttributeDefinitions: [
+    { AttributeName: "supplierId", AttributeType: "S" },
+    { AttributeName: "letterId", AttributeType: "S" },
+    { AttributeName: "queueTimestamp", AttributeType: "S" },
+  ],
+});
+
 export async function createTables(context: DBContext) {
   const { ddbClient } = context;
 
@@ -126,26 +154,22 @@ export async function createTables(context: DBContext) {
 
   await ddbClient.send(createMITableCommand);
   await ddbClient.send(createSupplierTableCommand);
+  await ddbClient.send(createLetterQueueTableCommand);
 }
 
 export async function deleteTables(context: DBContext) {
   const { ddbClient } = context;
 
-  await ddbClient.send(
-    new DeleteTableCommand({
-      TableName: "letters",
-    }),
-  );
-
-  await ddbClient.send(
-    new DeleteTableCommand({
-      TableName: "management-info",
-    }),
-  );
-
-  await ddbClient.send(
-    new DeleteTableCommand({
-      TableName: "suppliers",
-    }),
-  );
+  for (const tableName of [
+    "letters",
+    "management-info",
+    "suppliers",
+    "letter-queue",
+  ]) {
+    await ddbClient.send(
+      new DeleteTableCommand({
+        TableName: tableName,
+      }),
+    );
+  }
 }

internal/datastore/src/__test__/letter-queue-repository.test.ts

@@ -0,0 +1,104 @@
+import { Logger } from "pino";
+import {
+  DBContext,
+  createTables,
+  deleteTables,
+  setupDynamoDBContainer,
+} from "./db";
+import LetterQueueRepository from "../letter-queue-repository";
+import { InsertPendingLetter } from "../types";
+import { LetterAlreadyExistsError } from "../errors";
+import { createTestLogger } from "./logs";
+
+function createLetter(letterId = "letter1"): InsertPendingLetter {
+  return {
+    letterId,
+    supplierId: "supplier1",
+    specificationId: "specification1",
+    groupId: "group1",
+  };
+}
+
+// Database tests can take longer, especially with setup and teardown
+jest.setTimeout(30_000);
+
+describe("LetterQueueRepository", () => {
+  let db: DBContext;
+  let letterQueueRepository: LetterQueueRepository;
+  let logger: Logger;
+
+  beforeAll(async () => {
+    db = await setupDynamoDBContainer();
+  });
+
+  beforeEach(async () => {
+    await createTables(db);
+    ({ logger } = createTestLogger());
+
+    letterQueueRepository = new LetterQueueRepository(
+      db.docClient,
+      logger,
+      db.config,
+    );
+  });
+
+  afterEach(async () => {
+    await deleteTables(db);
+    jest.useRealTimers();
+  });
+
+  afterAll(async () => {
+    await db.container.stop();
+  });
+
+  function assertTtl(ttl: number, before: number, after: number) {
+    const expectedLower = Math.floor(
+      before / 1000 + 60 * 60 * db.config.letterQueueTtlHours,
+    );
+    const expectedUpper = Math.floor(
+      after / 1000 + 60 * 60 * db.config.lettersTtlHours,
+    );
+    expect(ttl).toBeGreaterThanOrEqual(expectedLower);
+    expect(ttl).toBeLessThanOrEqual(expectedUpper);
+  }
+
+  describe("putLetter", () => {
+    it("adds a letter to the database", async () => {
+      const before = Date.now();
+
+      const pendingLetter =
+        await letterQueueRepository.putLetter(createLetter());
+
+      const after = Date.now();
+
+      const timestampInMillis = new Date(
+        pendingLetter.queueTimestamp,
+      ).valueOf();
+      expect(timestampInMillis).toBeGreaterThanOrEqual(before);
+      expect(timestampInMillis).toBeLessThanOrEqual(after);
+      assertTtl(pendingLetter.ttl, before, after);
+    });
+
+    it("throws LetterAlreadyExistsError when creating a letter which already exists", async () => {
+      await letterQueueRepository.putLetter(createLetter());
+
+      await expect(
+        letterQueueRepository.putLetter(createLetter()),
+      ).rejects.toThrow(LetterAlreadyExistsError);
+    });
+
+    it("rethrows errors from DynamoDB when creating a letter", async () => {
+      const misconfiguredRepository = new LetterQueueRepository(
+        db.docClient,
+        logger,
+        {
+          ...db.config,
+          letterQueueTableName: "nonexistent-table",
+        },
+      );
+      await expect(
+        misconfiguredRepository.putLetter(createLetter()),
+      ).rejects.toThrow("Cannot do operations on a non-existent table");
+    });
+  });
+});

internal/datastore/src/config.ts

@@ -2,8 +2,10 @@ export type DatastoreConfig = {
   region: string;
   endpoint?: string;
   lettersTableName: string;
+  letterQueueTableName: string;
   miTableName: string;
   suppliersTableName: string;
   lettersTtlHours: number;
+  letterQueueTtlHours: number;
   miTtlHours: number;
 };

internal/datastore/src/errors.ts

@@ -0,0 +1,15 @@
+/**
+ * Error thrown when attempting to create a letter that already exists in the database.
+ */
+// eslint-disable-next-line import-x/prefer-default-export
+export class LetterAlreadyExistsError extends Error {
+  constructor(
+    public readonly supplierId: string,
+    public readonly letterId: string,
+  ) {
+    super(
+      `Letter already exists: supplierId=${supplierId}, letterId=${letterId}`,
+    );
+    this.name = "LetterAlreadyExistsError";
+  }
+}

internal/datastore/src/index.ts

@@ -1,5 +1,7 @@
 export * from "./types";
+export * from "./errors";
 export * from "./mi-repository";
 export * from "./letter-repository";
 export * from "./supplier-repository";
+export { default as LetterQueueRepository } from "./letter-queue-repository";
 export { default as DBHealthcheck } from "./healthcheck";