Push events to queue

terraform/modules/consumer/get-event.tf

@@ -28,6 +28,7 @@ resource "aws_lambda_function" "get_event" {
       client_secret               = var.consumer_client_secret
       lev_api_url                 = var.lev_api_url
       cloudwatch_metric_namespace = "${var.environment}-example-consumer"
+      queue_name                  = module.consumer_queue.queue_name
     }
   }
   tracing_config {

terraform/modules/consumer/lambdas/get_event.py

@@ -26,19 +26,24 @@
 cloudwatch = boto3.client("cloudwatch")
 cloudwatch_namespace = os.environ["cloudwatch_metric_namespace"]
 
+sqs = boto3.client("sqs")
+queue_name = os.environ["queue_name"]
+queue_url = sqs.get_queue_url(QueueName=queue_name)["QueueUrl"]
 
 def lambda_handler(event, _context):
     logger.info(f"## EVENT: {event}")
     logger.info(f"## TIME: {datetime.now()}")
 
     auth_token = get_auth_token(auth_url, client_id, client_secret)
     retrieved_event = get_event(auth_token, event["event_id"])
-    lev_record = get_lev_record(retrieved_event["sourceId"])
+    lev_record = get_lev_record(retrieved_event["data"]["attributes"]["sourceId"])
 
-    for datum in retrieved_event["eventData"].keys():
+    for datum in retrieved_event["data"]["attributes"]["eventData"].keys():
         assert_matches_lev(retrieved_event, lev_record, datum)
 
-    return {"event_id": retrieved_event["eventId"]}
+    push_event_to_queue(retrieved_event)
+
+    return {"event_id": retrieved_event["data"]["id"]}
 
 
 def get_event(auth_token: str, event_id: str):
@@ -51,6 +56,15 @@ def get_event(auth_token: str, event_id: str):
     return json.loads(response.read())
 
 
+def push_event_to_queue(event):
+    queue_event = event["data"]["attributes"]
+    queue_event["id"] = event["data"]["id"]
+    sqs.send_message(
+        QueueUrl=queue_url,
+        MessageBody=json.dumps(queue_event)
+    )
+
+
 def get_lev_record(record_id: str):
     record_request = request.Request(
         f"{lev_api_url}/v1/registration/death/{record_id}",
@@ -64,9 +78,12 @@ def assert_matches_lev(retrieved_event, lev_record, datum):
     """Fetch the matching record from the LEV API and assert matches.
 
     This is not needed in a normal consumer. Only for validation."""
-    provided_data = retrieved_event["eventData"][datum]
-    lev_key = map_to_lev_key(datum)
-    lev_data = lev_record["deceased"][lev_key]
+    provided_data = retrieved_event["data"]["attributes"]["eventData"][datum]
+    if datum == "registrationDate":
+        lev_data = lev_record["date"]
+    else:
+        lev_key = map_to_lev_key(datum)
+        lev_data = lev_record["deceased"][lev_key]
     if provided_data != lev_data:
         logger.error(f"Data mismatch. Provided data: {provided_data}, lev key: {lev_key}, lev data: {lev_data}")
         record_metric(cloudwatch, cloudwatch_namespace, "GET_EVENT.DataMatchFailure", 1)
@@ -75,7 +92,7 @@ def assert_matches_lev(retrieved_event, lev_record, datum):
 
 
 def map_to_lev_key(key: str) -> str:
-    if key == "firstName":
+    if key == "firstNames":
         return "forenames"
     elif key == "lastName":
         return "surname"
@@ -85,3 +102,5 @@ def map_to_lev_key(key: str) -> str:
         return "dateOfDeath"
     elif key == "address":
         return "address"
+    else:
+        return key

terraform/modules/consumer/sqs.tf

@@ -1,7 +1,7 @@
 module "consumer_queue" {
-  source = "../sqs"
+  source      = "../sqs"
   environment = var.environment
-  queue_name = "${var.environment}-example-consumer-queue"
+  queue_name  = "${var.environment}-example-consumer-queue"
 }
 
 data "aws_iam_policy_document" "lambda_sqs_access" {