send it

Author: devksingh4

src/data-archive/main.py

@@ -2,6 +2,8 @@
 import os
 import json
 import logging
+from typing import Any, Callable, Dict
+from datetime import datetime
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
@@ -15,6 +17,12 @@
     logger.error("The 'FIREHOSE_STREAM_NAME' environment variable is not set.")
     raise
 
+TimestampMapper = Dict[str, Callable[[Dict[str, Any]], str]]
+
+ARCHIVE_TIMESTAMP_MAPPER: TimestampMapper = {
+    "infra-core-api-room-requests-status": lambda x: x["createdAt#status"].split("#")[0]
+}
+
 
 def deserialize_dynamodb_item(item):
     """
@@ -53,12 +61,22 @@ def lambda_handler(event, context):
 
             deserialized_data = deserialize_dynamodb_item(old_image)
 
-            # 4. **Construct the Payload**: Create the specified {'table': ..., 'data': ...}
-            #    payload that will be sent to Firehose.
-            payload = {"table": table_name, "data": deserialized_data}
+            # 4. Construct the Payload
+            payload = {
+                "table": table_name,
+                "data": deserialized_data,
+                "timestamp": datetime.now().isoformat(),
+            }
+            if table_name in ARCHIVE_TIMESTAMP_MAPPER:
+                try:
+                    payload["timestamp"] = ARCHIVE_TIMESTAMP_MAPPER[table_name](
+                        deserialized_data
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Failed to extract timestamp for record from {table_name}: {str(e)}. Using now as timestamp."
+                    )
 
-            # 5. **Format for Firehose**: The PutRecordBatch API expects each record
-            #    to have a 'Data' key with a byte-encoded string value.
             firehose_records_to_send.append(
                 {"Data": json.dumps(payload).encode("utf-8")}
             )

terraform/modules/archival/main.tf

@@ -249,7 +249,7 @@ resource "aws_kinesis_firehose_delivery_stream" "dynamic_stream" {
     bucket_arn         = aws_s3_bucket.this.arn
     role_arn           = aws_iam_role.firehose_role.arn
     buffering_interval = 60
-    buffering_size     = 10
+    buffering_size     = 64
     compression_format = "GZIP"
 
     dynamic_partitioning_configuration {
@@ -262,7 +262,7 @@ resource "aws_kinesis_firehose_delivery_stream" "dynamic_stream" {
         type = "MetadataExtraction"
         parameters {
           parameter_name  = "MetadataExtractionQuery"
-          parameter_value = "{table:.table}"
+          parameter_value = "{table:.table, event_ts:.timestamp}"
         }
         parameters {
           parameter_name  = "JsonParsingEngine"
@@ -271,7 +271,7 @@ resource "aws_kinesis_firehose_delivery_stream" "dynamic_stream" {
       }
     }
 
-    prefix              = "table=!{partitionKeyFromQuery:table}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/"
+    prefix              = "table=!{partitionKeyFromQuery:table}/year=!{partitionKeyFromQueryAsTimestamp:event_ts:yyyy}/month=!{partitionKeyFromQueryAsTimestamp:event_ts:MM}/day=!{partitionKeyFromQueryAsTimestamp:event_ts:dd}/"
     error_output_prefix = "firehose-errors/!{firehose:error-output-type}/!{timestamp:yyyy/MM/dd}/"
   }
 }