introduced firehose and audit bucket

Author: Karthikeyannhs

infrastructure/modules/lambda/lambda.tf

@@ -19,6 +19,7 @@ resource "aws_lambda_function" "eligibility_signposting_lambda" {
     variables = {
       PERSON_TABLE_NAME = var.eligibility_status_table_name,
       RULES_BUCKET_NAME = var.eligibility_rules_bucket_name,
+      AUDIT_BUCKET_NAME = var.eligibility_audit_bucket_name
       ENV               = var.environment
       LOG_LEVEL         = var.log_level
     }

infrastructure/modules/lambda/variables.tf

@@ -33,6 +33,11 @@ variable "eligibility_rules_bucket_name" {
   type        = string
 }
 
+variable "eligibility_audit_bucket_name" {
+  description = "campaign config audit bucket name"
+  type        = string
+}
+
 variable "eligibility_status_table_name" {
   description = "eligibility datastore table name"
   type        = string

infrastructure/stacks/api-layer/lambda.tf

@@ -21,6 +21,7 @@ module "eligibility_signposting_lambda_function" {
   file_name                     = "../../../dist/lambda.zip"
   handler                       = "eligibility_signposting_api.app.lambda_handler"
   eligibility_rules_bucket_name = module.s3_rules_bucket.storage_bucket_name
+  eligibility_audit_bucket_name = module.s3_audit_bucket.storage_bucket_name
   eligibility_status_table_name = module.eligibility_status_table.table_name
   log_level                     = "INFO"
   stack_name                    = local.stack_name

src/eligibility_signposting_api/config/config.py

@@ -34,7 +34,8 @@ def config() -> dict[str, Any]:
             "person_table_name": person_table_name,
             "s3_endpoint": None,
             "rules_bucket_name": rules_bucket_name,
-            "audit_bucket_name":audit_bucket_name,
+            "audit_bucket_name": audit_bucket_name,
+            "firehose_endpoint": None,
             "log_level": log_level,
         }
 
@@ -47,6 +48,7 @@ def config() -> dict[str, Any]:
         "s3_endpoint": URL(os.getenv("S3_ENDPOINT", "http://localhost:4566")),
         "rules_bucket_name": rules_bucket_name,
         "audit_bucket_name": audit_bucket_name,
+        "firehose_endpoint": URL(os.getenv("FIREHOSE_ENDPOINT", "http://localhost:4566")),
         "log_level": log_level,
     }
 

src/eligibility_signposting_api/config/contants.py

@@ -1,3 +1,6 @@
 MAGIC_COHORT_LABEL = "elid_all_people"
 RULE_STOP_DEFAULT = False
-NHS_NUMBER_HEADER_NAME = "nhs-login-nhs-number"
+NHS_NUMBER_HEADER = "nhs-login-nhs-number"
+
+# Infra
+ELIGIBILITY_SIGNPOSTING_AUDIT_STREAM = "eligibility-signposting-audit-stream"

src/eligibility_signposting_api/repos/factory.py

@@ -35,3 +35,11 @@ def dynamodb_resource_factory(
 def s3_service_factory(session: Session, s3_endpoint: Annotated[URL, Inject(param="s3_endpoint")]) -> BaseClient:
     endpoint_url = str(s3_endpoint) if s3_endpoint is not None else None
     return session.client("s3", endpoint_url=endpoint_url)
+
+
+@service(qualifier="firehose")
+def firehose_client_factory(
+    session: Session, firehose_endpoint: Annotated[URL, Inject(param="firehose_endpoint")]
+) -> BaseClient:
+    endpoint_url = str(firehose_endpoint) if firehose_endpoint is not None else None
+    return session.client("firehose", endpoint_url=endpoint_url)

src/eligibility_signposting_api/services/audit_service.py

@@ -0,0 +1,34 @@
+import json
+import logging
+from typing import Annotated
+
+from botocore.client import BaseClient
+from wireup import Inject, service
+
+from eligibility_signposting_api.config.contants import ELIGIBILITY_SIGNPOSTING_AUDIT_STREAM
+
+logger = logging.getLogger(__name__)
+
+
+@service
+class AuditService:
+    def __init__(self, firehose: Annotated[BaseClient, Inject(qualifier="firehose")]) -> None:
+        super().__init__()
+        self.firehose = firehose
+        self.delivery_stream_name = ELIGIBILITY_SIGNPOSTING_AUDIT_STREAM
+
+    def audit(self, audit_record: dict) -> None:
+        """
+        Sends an audit record to the configured Firehose delivery stream.
+
+        Args:
+            audit_record (dict): The audit data to send.
+
+        Returns:
+            str: The Firehose record ID.
+        """
+        response = self.firehose.put_record(
+            DeliveryStreamName=self.delivery_stream_name,
+            Record={"Data": (json.dumps(audit_record) + "\n").encode("utf-8")},
+        )
+        logger.info("Successfully sent to the Firehose", extra={"firehose_record_id": response["RecordId"]})

src/eligibility_signposting_api/wrapper.py

@@ -4,7 +4,7 @@
 
 from mangum.types import LambdaContext, LambdaEvent
 
-from eligibility_signposting_api.config.contants import NHS_NUMBER_HEADER_NAME
+from eligibility_signposting_api.config.contants import NHS_NUMBER_HEADER
 
 logger = logging.getLogger(__name__)
 
@@ -20,7 +20,7 @@ def wrapper(event: LambdaEvent, context: LambdaContext) -> dict[str, int | str]:
             headers = event.get("headers", {})
             path_params = event.get("pathParameters", {})
 
-            header_nhs = headers.get(NHS_NUMBER_HEADER_NAME)
+            header_nhs = headers.get(NHS_NUMBER_HEADER)
             path_nhs = path_params.get("id")
 
             logger.info("nhs numbers from the request", extra={"header_nhs": header_nhs, "path_nhs": path_nhs})

tests/integration/conftest.py

@@ -96,10 +96,11 @@ def iam_client(boto3_session: Session, localstack: URL) -> BaseClient:
 def s3_client(boto3_session: Session, localstack: URL) -> BaseClient:
     return boto3_session.client("s3", endpoint_url=str(localstack))
 
+
 @pytest.fixture(scope="session")
 def firehose_client(boto3_session: Session, localstack: URL) -> BaseClient:
-    session = Session()
-    return session.client("firehose", endpoint_url=str(localstack))
+    return boto3_session.client("firehose", endpoint_url=str(localstack))
+
 
 @pytest.fixture(scope="session")
 def iam_role(iam_client: BaseClient) -> Generator[str]:
@@ -194,6 +195,7 @@ def flask_function(lambda_client: BaseClient, iam_role: str, lambda_zip: Path) -
                 "Variables": {
                     "DYNAMODB_ENDPOINT": os.getenv("LOCALSTACK_INTERNAL_ENDPOINT", "http://localstack:4566/"),
                     "S3_ENDPOINT": os.getenv("LOCALSTACK_INTERNAL_ENDPOINT", "http://localstack:4566/"),
+                    "FIREHOSE_ENDPOINT": os.getenv("LOCALSTACK_INTERNAL_ENDPOINT", "http://localstack:4566/"),
                     "AWS_REGION": AWS_REGION,
                     "LOG_LEVEL": "DEBUG",
                 }
@@ -382,13 +384,15 @@ def rules_bucket(s3_client: BaseClient) -> Generator[BucketName]:
     yield bucket_name
     s3_client.delete_bucket(Bucket=bucket_name)
 
+
 @pytest.fixture(scope="session")
 def audit_bucket(s3_client: BaseClient) -> Generator[BucketName]:
     bucket_name = BucketName(os.getenv("AUDIT_BUCKET_NAME", "test-audit-bucket"))
     s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": AWS_REGION})
     yield bucket_name
     s3_client.delete_bucket(Bucket=bucket_name)
 
+
 @pytest.fixture(scope="class")
 def campaign_config(s3_client: BaseClient, rules_bucket: BucketName) -> Generator[rules.CampaignConfig]:
     campaign: rules.CampaignConfig = rule.CampaignConfigFactory.build(
@@ -419,8 +423,9 @@ def campaign_config(s3_client: BaseClient, rules_bucket: BucketName) -> Generato
 
 
 @pytest.fixture(scope="class")
-def campaign_config_with_magic_cohort(s3_client: BaseClient,
-                                      rules_bucket: BucketName) -> Generator[rules.CampaignConfig]:
+def campaign_config_with_magic_cohort(
+    s3_client: BaseClient, rules_bucket: BucketName
+) -> Generator[rules.CampaignConfig]:
     campaign: rules.CampaignConfig = rule.CampaignConfigFactory.build(
         target="COVID",
         iterations=[

tests/integration/lambda/test_app_running_as_lambda.py

@@ -155,8 +155,8 @@ def get_log_messages(flask_function: str, logs_client: BaseClient) -> list[str]:
 def test_given_nhs_number_in_path_matches_with_nhs_number_in_headers(
     lambda_client: BaseClient,  # noqa:ARG001
     persisted_person: NHSNumber,
-    campaign_config: CampaignConfig, # noqa:ARG001
-    firehose_client: BaseClient, # noqa:ARG001
+    campaign_config: CampaignConfig,  # noqa:ARG001
+    firehose_client: BaseClient,  # noqa:ARG001
     api_gateway_endpoint: URL,
 ):
     # Given