Add replay-ability for transaction collection process

Previously, if a transaction history collection job failed to process for
whatever reason (such as a service outage), it was incredibly difficult to run
the processing for that specific time period again. This change leverages the
fact that event bridge sends in the timestamp of the time in which the scheduled
task was triggered.

By going off of this time, rather than generating the timestamp in the lambda,
developers can simply rerun the failed step function execution, and it will
process all the transactions for that same time period. Due to the idempotent
nature of the transaction pk/sk design, we will never see duplicate transaction
records for the same transaction ids.

Author: landonshumway-ia

backend/compact-connect/lambdas/python/purchases/handlers/transaction_history.py

@@ -1,5 +1,5 @@
 import json
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from aws_lambda_powertools.utilities.typing import LambdaContext
 from cc_common.config import config, logger
@@ -25,6 +25,7 @@ def process_settled_transactions(event: dict, context: LambdaContext) -> dict:
 
     :param event: Lambda event containing:
         - compact: The compact name
+        - scheduledTime: The scheduled execution time from EventBridge for replay-ability
         - lastProcessedTransactionId: Optional last processed transaction ID
         - currentBatchId: Optional current batch ID being processed
         - processedBatchIds: Optional list of batch IDs that have been processed, this ensures we don't process the same
@@ -33,14 +34,16 @@ def process_settled_transactions(event: dict, context: LambdaContext) -> dict:
     :return: Dictionary indicating processing status and optional pagination info
     """
     compact = event['compact']
+    scheduled_time = event['scheduledTime']
     last_processed_transaction_id = event.get('lastProcessedTransactionId')
     current_batch_id = event.get('currentBatchId')
     processed_batch_ids = event.get('processedBatchIds', [])
 
+    # Use the scheduled time from EventBridge for replay-ability
     # By default, the authorize.net accounts batch settlements at 4:00pm Pacific Time.
     # This daily collector runs an hour later (5pm PST, which is 1am UTC) to collect
     # all settled transaction for the last 24 hours.
-    end_time = config.current_standard_datetime.replace(hour=1, minute=0, second=0, microsecond=0)
+    end_time = datetime.fromisoformat(scheduled_time).replace(hour=1, minute=0, second=0, microsecond=0)
     start_time = end_time - timedelta(days=1)
 
     # Format timestamps for API call
@@ -81,6 +84,7 @@ def process_settled_transactions(event: dict, context: LambdaContext) -> dict:
     # Return appropriate response based on whether there are more transactions to process
     response = {
         'compact': compact,  # Always include the compact name
+        'scheduledTime': scheduled_time,  # Preserve scheduled time for subsequent iterations
         'status': 'IN_PROGRESS' if not _all_transactions_processed(transaction_response) else 'COMPLETE',
         'processedBatchIds': transaction_response['processedBatchIds'],
     }

backend/compact-connect/lambdas/python/purchases/tests/function/test_handlers/test_transaction_history.py

@@ -8,9 +8,6 @@
 
 TEST_COMPACT = 'aslp'
 TEST_AUD_LICENSE_TYPE_ABBR = 'aud'
-MOCK_START_TIME = '2024-01-01T00:00:00Z'
-MOCK_END_TIME = '2024-01-02T00:00:00Z'
-MOCK_TRANSACTION_LIMIT = 500
 
 # Test transaction data
 MOCK_TRANSACTION_ID = '12345'
@@ -28,6 +25,8 @@
 MOCK_CURRENT_BATCH_ID = 'mock_current_batch_id'
 MOCK_PROCESSED_BATCH_IDS = ['mock_processed_batch_id']
 
+MOCK_SCHEDULED_TIME = '2024-01-01T01:00:00Z'
+
 
 def _generate_mock_transaction(
     transaction_id=MOCK_TRANSACTION_ID,
@@ -139,6 +138,7 @@ def _add_mock_privilege_update_to_database(
     def _when_testing_non_paginated_event(self, test_compact=TEST_COMPACT):
         return {
             'compact': test_compact,
+            'scheduledTime': MOCK_SCHEDULED_TIME,  # Mock EventBridge scheduled time
             'lastProcessedTransactionId': None,
             'currentBatchId': None,
             'processedBatchIds': None,
@@ -147,6 +147,7 @@ def _when_testing_non_paginated_event(self, test_compact=TEST_COMPACT):
     def _when_testing_paginated_event(self, test_compact=TEST_COMPACT):
         return {
             'compact': test_compact,
+            'scheduledTime': MOCK_SCHEDULED_TIME,  # Mock EventBridge scheduled time
             'lastProcessedTransactionId': MOCK_LAST_PROCESSED_TRANSACTION_ID,
             'currentBatchId': MOCK_CURRENT_BATCH_ID,
             'processedBatchIds': MOCK_PROCESSED_BATCH_IDS,
@@ -196,7 +197,15 @@ def test_process_settled_transactions_returns_complete_status(self, mock_purchas
         event = self._when_testing_non_paginated_event()
         resp = process_settled_transactions(event, self.mock_context)
 
-        self.assertEqual({'compact': 'aslp', 'processedBatchIds': [MOCK_BATCH_ID], 'status': 'COMPLETE'}, resp)
+        self.assertEqual(
+            {
+                'compact': 'aslp',
+                'scheduledTime': MOCK_SCHEDULED_TIME,
+                'processedBatchIds': [MOCK_BATCH_ID],
+                'status': 'COMPLETE',
+            },
+            resp,
+        )
 
     @patch('handlers.transaction_history.PurchaseClient')
     def test_process_settled_transactions_passes_pagination_values_into_purchase_client(
@@ -287,6 +296,42 @@ def test_process_settled_transactions_stores_transactions_in_dynamodb(self, mock
             stored_transactions['Items'],
         )
 
+    @patch('handlers.transaction_history.PurchaseClient')
+    def test_process_settled_transactions_does_not_duplicate_identical_transaction_records(
+        self, mock_purchase_client_constructor
+    ):
+        """Test that transactions are stored in DynamoDB."""
+        from handlers.transaction_history import process_settled_transactions
+
+        # in this test, there is one transaction, and one privilege. These should map together using the default
+        # transaction id and privilege id
+        self._when_purchase_client_returns_transactions(mock_purchase_client_constructor)
+        self._add_mock_privilege_to_database()
+
+        event = self._when_testing_non_paginated_event()
+
+        process_settled_transactions(event, self.mock_context)
+
+        # Verify transactions were stored in DynamoDB
+        stored_transactions = self.config.transaction_history_table.query(
+            KeyConditionExpression='pk = :pk',
+            ExpressionAttributeValues={':pk': f'COMPACT#{TEST_COMPACT}#TRANSACTIONS#MONTH#2024-01'},
+        )
+
+        self.assertEqual(1, len(stored_transactions['Items']))
+
+        # now run the lambda again with the same payload, the duplicate transaction record should overwrite the
+        # existing one
+        process_settled_transactions(event, self.mock_context)
+
+        # Verify transactions were stored in DynamoDB
+        stored_transactions = self.config.transaction_history_table.query(
+            KeyConditionExpression='pk = :pk',
+            ExpressionAttributeValues={':pk': f'COMPACT#{TEST_COMPACT}#TRANSACTIONS#MONTH#2024-01'},
+        )
+
+        self.assertEqual(1, len(stored_transactions['Items']))
+
     @patch('handlers.transaction_history.PurchaseClient')
     def test_process_settled_transactions_returns_in_progress_status_with_pagination_values(
         self, mock_purchase_client_constructor
@@ -302,6 +347,7 @@ def test_process_settled_transactions_returns_in_progress_status_with_pagination
         self.assertEqual(
             {
                 'compact': TEST_COMPACT,
+                'scheduledTime': MOCK_SCHEDULED_TIME,
                 'status': 'IN_PROGRESS',
                 'lastProcessedTransactionId': MOCK_LAST_PROCESSED_TRANSACTION_ID,
                 'currentBatchId': MOCK_CURRENT_BATCH_ID,
@@ -369,6 +415,7 @@ def test_process_settled_transactions_returns_batch_failure_status_after_process
             {
                 'status': 'IN_PROGRESS',
                 'compact': TEST_COMPACT,
+                'scheduledTime': MOCK_SCHEDULED_TIME,
                 'currentBatchId': MOCK_BATCH_ID,
                 'lastProcessedTransactionId': mock_first_iteration_failed_transaction_id,
                 'processedBatchIds': [],
@@ -391,6 +438,7 @@ def test_process_settled_transactions_returns_batch_failure_status_after_process
             {
                 'status': 'BATCH_FAILURE',
                 'compact': TEST_COMPACT,
+                'scheduledTime': MOCK_SCHEDULED_TIME,
                 'processedBatchIds': [MOCK_BATCH_ID],
                 'batchFailureErrorMessage': json.dumps(
                     {

backend/compact-connect/stacks/transaction_monitoring_stack/transaction_history_processing_workflow.py

@@ -18,7 +18,6 @@
     LogLevel,
     LogOptions,
     Pass,
-    Result,
     StateMachine,
     Succeed,
     TaskInput,
@@ -98,11 +97,16 @@ def __init__(
         )
 
         # Create Step Function definition
-        # set initial values for the process task to use
+        # Transform EventBridge input to extract time and create payload for replay-ability
         self.initialize_state = Pass(
             self,
             f'{compact}-InitializeState',
-            result=Result.from_object({'Payload': {'compact': compact, 'processedBatchIds': []}}),
+            parameters={
+                'compact': compact,
+                'scheduledTime.$': '$.time',  # Extract time from EventBridge event
+                'processedBatchIds': [],
+            },
+            result_path='$.Payload',
         )
 
         self.processor_task = LambdaInvoke(

backend/compact-connect/tests/app/test_transaction_monitoring.py

@@ -150,6 +150,23 @@ def test_workflow_generates_expected_process_transaction_history_lambda_invoke_s
             ),
         )
 
+    def test_workflow_generates_expected_initialize_step(self):
+        aslp_transaction_history_proccessing_workflow = (
+            self.app.prod_backend_pipeline_stack.prod_stage.transaction_monitoring_stack.compact_state_machines['aslp']
+        )
+
+        self.assertEqual(
+            {
+                'Next': 'aslp-ProcessTransactionHistory',
+                'Parameters': {'compact': 'aslp', 'processedBatchIds': [], 'scheduledTime.$': '$.time'},
+                'ResultPath': '$.Payload',
+                'Type': 'Pass',
+            },
+            self.remove_dynamic_tokens_numbers(
+                aslp_transaction_history_proccessing_workflow.initialize_state.to_state_json()
+            ),
+        )
+
     def test_workflow_generates_expected_choice_step(self):
         aslp_transaction_history_proccessing_workflow = (
             self.app.prod_backend_pipeline_stack.prod_stage.transaction_monitoring_stack.compact_state_machines['aslp']