Bug/handle authorizenet delayed reporting (#1201)

### Description List
- Updated batch query logic to start time window just after most recent
collected batch settlement time
- Moved collector timing back by one more hour
- Updated related code to leverage newer CCDataClass and
TestDataGenerator patterns

### Testing List
- Run transaction collection in sandbox
- Code review

Closes #1192 


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

## Release Notes

* **New Features**
* Enhanced transaction history processing with dynamic time window
retrieval for improved accuracy.
* Improved handling of settled and unsettled transaction reconciliation.

* **Bug Fixes**
* Declined transactions from payment processors are now properly
skipped.
  * Better detection and handling of old unsettled transactions.

* **Chores**
* Adjusted transaction processing schedule to account for settlement
delays.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: jusdino

backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/purchase/__init__.py

@@ -1,11 +1,12 @@
 # ruff: noqa: N801, N815 invalid-name
+from marshmallow import ValidationError, validates_schema
+from marshmallow.fields import Dict, List, String
+
 from cc_common.data_model.schema.base_record import ForgivingSchema
 from cc_common.data_model.schema.compact.api import CompactOptionsResponseSchema
 from cc_common.data_model.schema.compact.common import COMPACT_TYPE
 from cc_common.data_model.schema.jurisdiction.api import JurisdictionOptionsResponseSchema
 from cc_common.data_model.schema.jurisdiction.common import JURISDICTION_TYPE
-from marshmallow import ValidationError, validates_schema
-from marshmallow.fields import Dict, List, String
 
 
 class PurchasePrivilegeOptionsResponseSchema(ForgivingSchema):

backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/purchase/api.py

@@ -0,0 +1,72 @@
+# ruff: noqa: N802 we use camelCase to match the marshmallow schema definition
+
+from cc_common.data_model.schema.common import CCDataClass
+from cc_common.data_model.schema.transaction.record import TransactionRecordSchema
+
+
+class TransactionData(CCDataClass):
+    """
+    Class representing a Transaction with read-only properties.
+
+    Unlike several other CCDataClass subclasses, this one does not include setters. This is because
+    transaction records are only created during transaction processing, so we can pass the entire record
+    from the processing into the constructor.
+
+    Note: This class requires valid data when created - it cannot be instantiated empty
+    and populated later.
+    """
+
+    # Define the record schema at the class level
+    _record_schema = TransactionRecordSchema()
+
+    # Require valid data when creating instances
+    _requires_data_at_construction = True
+
+    @property
+    def transactionProcessor(self) -> str:
+        return self._data['transactionProcessor']
+
+    @property
+    def transactionId(self) -> str:
+        return self._data['transactionId']
+
+    @property
+    def batch(self) -> dict:
+        """Batch information containing batchId, settlementState, settlementTimeLocal, and settlementTimeUTC."""
+        return self._data['batch']
+
+    @property
+    def lineItems(self) -> list[dict]:
+        """
+        List of line items, each containing description, itemId, name, quantity, taxable,
+        unitPrice, and optionally privilegeId.
+        """
+        return self._data['lineItems']
+
+    @property
+    def compact(self) -> str:
+        return self._data['compact']
+
+    @property
+    def licenseeId(self) -> str:
+        return self._data['licenseeId']
+
+    @property
+    def responseCode(self) -> str:
+        return self._data['responseCode']
+
+    @property
+    def settleAmount(self) -> str:
+        return self._data['settleAmount']
+
+    @property
+    def submitTimeUTC(self) -> str:
+        return self._data['submitTimeUTC']
+
+    @property
+    def transactionStatus(self) -> str:
+        return self._data['transactionStatus']
+
+    @property
+    def transactionType(self) -> str:
+        return self._data['transactionType']

backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/transaction/__init__.py

@@ -3,7 +3,8 @@
 from boto3.dynamodb.conditions import Key
 
 from cc_common.config import _Config, logger
-from cc_common.data_model.schema.transaction.record import TransactionRecordSchema, UnsettledTransactionRecordSchema
+from cc_common.data_model.schema.transaction import TransactionData
+from cc_common.data_model.schema.transaction.record import UnsettledTransactionRecordSchema
 
 AUTHORIZE_DOT_NET_CLIENT_TYPE = 'authorize.net'
 
@@ -14,21 +15,18 @@ class TransactionClient:
     def __init__(self, config: _Config):
         self.config = config
 
-    def store_transactions(self, transactions: list[dict]) -> None:
+    def store_transactions(self, transactions: list[TransactionData]) -> None:
         """
         Store transaction records in DynamoDB.
 
-        :param compact: The compact name
         :param transactions: List of transaction records to store
         """
         with self.config.transaction_history_table.batch_writer() as batch:
             for transaction in transactions:
                 # Convert UTC timestamp to epoch for sorting
-                transaction_processor = transaction['transactionProcessor']
+                transaction_processor = transaction.transactionProcessor
                 if transaction_processor == AUTHORIZE_DOT_NET_CLIENT_TYPE:
-                    transaction_schema = TransactionRecordSchema()
-
-                    serialized_record = transaction_schema.dump(transaction)
+                    serialized_record = transaction.serialize_to_database_record()
                     batch.put_item(Item=serialized_record)
                 else:
                     raise ValueError(f'Unsupported transaction processor: {transaction_processor}')
@@ -80,6 +78,49 @@ def get_transactions_in_range(self, compact: str, start_epoch: int, end_epoch: i
 
         return all_items
 
+    def get_most_recent_transaction_for_compact(self, compact: str) -> TransactionData:
+        """
+        Get the most recent transaction for a compact.
+
+        Starts by querying the current month's partition key (based on config.current_standard_datetime),
+        then sequentially queries previous months until a record is found.
+
+        :param compact: The compact name
+        :return: The most recent transaction for the compact
+        :raises ValueError: If no transactions are found for the compact
+        """
+        # Start with the current month
+        current_date = self.config.current_standard_datetime.replace(day=1)
+        # During normal operations, the most recent transaction should be no more than two days old, if there were any
+        # transactions in that period. We'll look back up to three months, which should cover most reasonable
+        # situations.
+        max_months_to_check = 3
+
+        for _ in range(max_months_to_check):
+            month_key = current_date.strftime('%Y-%m')
+            pk = f'COMPACT#{compact}#TRANSACTIONS#MONTH#{month_key}'
+
+            # Query for the most recent transaction in this month (descending order, limit 1)
+            response = self.config.transaction_history_table.query(
+                KeyConditionExpression=Key('pk').eq(pk),
+                ScanIndexForward=False,  # Descending order (most recent first)
+                Limit=1,
+            )
+
+            items = response.get('Items', [])
+            if items:
+                # Found a transaction, return it
+                return TransactionData.from_database_record(items[0])
+
+            # Move to previous month
+            if current_date.month == 1:
+                current_date = current_date.replace(year=current_date.year - 1, month=12)
+            else:
+                current_date = current_date.replace(month=current_date.month - 1)
+
+        # No transactions found after checking max_months_to_check months
+        raise ValueError(f'No transactions found for compact: {compact}')
+
     def _query_transactions_for_month(
         self,
         compact: str,
@@ -122,10 +163,13 @@ def _query_transactions_for_month(
 
     def _set_privilege_id_in_line_item(self, line_items: list[dict], item_id_prefix: str, privilege_id: str):
         for line_item in line_items:
-            if line_item.get('itemId').lower().startswith(item_id_prefix.lower()):
+            item_id = line_item.get('itemId')
+            if item_id and item_id.lower().startswith(item_id_prefix.lower()):
                 line_item['privilegeId'] = privilege_id
 
-    def add_privilege_information_to_transactions(self, compact: str, transactions: list[dict]) -> list[dict]:
+    def add_privilege_information_to_transactions(
+        self, compact: str, transactions: list[TransactionData]
+    ) -> list[TransactionData]:
         """
         Add privilege and licensee IDs to transaction line items based on the jurisdiction they were purchased for.
 
@@ -134,7 +178,7 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
         :return: Modified list of transactions with privilege and licensee IDs added to line items
         """
         for transaction in transactions:
-            line_items = transaction['lineItems']
+            line_items = transaction.lineItems
             # Extract jurisdictions from line items with format priv:{compact}-{jurisdiction}-{license type abbr}
             jurisdictions_to_process = set()
             for line_item in line_items:
@@ -145,7 +189,7 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
                     jurisdictions_to_process.add(jurisdiction)
 
             # Query for privilege records using the GSI
-            gsi_pk = f'COMPACT#{compact}#TX#{transaction["transactionId"]}#'
+            gsi_pk = f'COMPACT#{compact}#TX#{transaction.transactionId}#'
             response = self.config.provider_table.query(
                 IndexName=self.config.compact_transaction_id_gsi_name,
                 KeyConditionExpression=Key('compactTransactionIdGSIPK').eq(gsi_pk),
@@ -157,9 +201,9 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
                 logger.error(
                     'No privilege records found for this transaction id.',
                     compact=compact,
-                    transaction_id=transaction['transactionId'],
+                    transaction_id=transaction.transactionId,
                     # attempt to grab the licensee id from the authorize.net data, which may be invalid if it was masked
-                    licensee_id=transaction['licenseeId'],
+                    licensee_id=transaction.licenseeId,
                 )
                 # We mark the data as UNKNOWN so it still shows up in the history,
                 # and move onto the next transaction
@@ -182,16 +226,16 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
                 logger.error(
                     'More than one matching provider id found for a transaction id.',
                     compact=compact,
-                    transaction_id=transaction['transactionId'],
+                    transaction_id=transaction.transactionId,
                     # attempt to grab the licensee id from the authorize.net data, which may be invalid if it was masked
-                    provider_ids=transaction['licenseeId'],
+                    provider_ids=transaction.licenseeId,
                 )
 
             # The licensee id recorded in Authorize.net cannot be trusted, as Authorize.net masks any values that look
             # like a credit card number (consecutive digits separated by dashes). We need to grab the provider id from
             # the privileges associated with this transaction and set the licensee id on the transaction to that value
             # to ensure it is valid.
-            transaction['licenseeId'] = provider_ids.pop()
+            transaction.update({'licenseeId': provider_ids.pop()})
 
             # Process each privilege record
             for jurisdiction in jurisdictions_to_process:
@@ -220,15 +264,16 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
                         'No matching jurisdiction privilege record found for transaction. '
                         'Cannot determine privilege id for this transaction',
                         compact=compact,
-                        transactionId=transaction['transactionId'],
+                        transactionId=transaction.transactionId,
                         jurisdiction=jurisdiction,
-                        provider_id=transaction['licenseeId'],
+                        provider_id=transaction.licenseeId,
                         matching_privilege_records=response.get('Items', []),
                     )
                     # we set the privilege id to UNKNOWN, so that it will be visible in the report
                     self._set_privilege_id_in_line_item(
                         line_items=line_items, item_id_prefix=item_id_prefix, privilege_id='UNKNOWN'
                     )
+            transaction.update({'lineItems': line_items})
 
         return transactions
 
@@ -269,7 +314,7 @@ def store_unsettled_transaction(self, compact: str, transaction_id: str, transac
                 error=str(e),
             )
 
-    def reconcile_unsettled_transactions(self, compact: str, settled_transactions: list[dict]) -> list[str]:
+    def reconcile_unsettled_transactions(self, compact: str, settled_transactions: list[TransactionData]) -> list[str]:
         """
         Reconcile unsettled transactions with settled transactions and detect old unsettled transactions.
 
@@ -297,7 +342,7 @@ def reconcile_unsettled_transactions(self, compact: str, settled_transactions: l
             return []
 
         # Create a set of settled transaction IDs for efficient lookup
-        settled_transaction_ids = {tx['transactionId'] for tx in settled_transactions}
+        settled_transaction_ids = {tx.transactionId for tx in settled_transactions}
 
         # Separate matched and unmatched unsettled transactions
         matched_unsettled = []
@@ -325,6 +370,16 @@ def reconcile_unsettled_transactions(self, compact: str, settled_transactions: l
         cutoff_time = datetime.now(UTC) - timedelta(hours=48)
         old_unsettled_transactions = []
 
+        # We expect that all transactions we process from Authorize.net will match a record we have already
+        # created at the time of purchase, as an unsettled transaction. Any mismatch is an error.
+        matched_unsettled_transaction_ids = {tx['transactionId'] for tx in matched_unsettled}
+        unmatched_settled_transaction_ids = settled_transaction_ids - matched_unsettled_transaction_ids
+        if unmatched_settled_transaction_ids:
+            logger.error(
+                'Unable to reconcile some transactions from Authorize.Net with our unsettled transactions',
+                unreconciled_transactions=unmatched_settled_transaction_ids
+            )
+
         for unsettled_tx in unmatched_unsettled:
             transaction_date = datetime.fromisoformat(unsettled_tx['transactionDate'])
             if transaction_date < cutoff_time:

backend/compact-connect/lambdas/python/common/cc_common/data_model/transaction_client.py

@@ -29,6 +29,37 @@
 DEFAULT_PRIVILEGE_EXPIRATION_DATE = '2025-04-04'
 DEFAULT_PRIVILEGE_UPDATE_DATETIME = '2020-05-05T12:59:59+00:00'
 DEFAULT_COMPACT_TRANSACTION_ID = '1234567890'
+DEFAULT_COMPACT_TRANSACTION_BATCH = {
+    'batchId': '67890',
+    'settlementState': 'settledSuccessfully',
+    'settlementTimeLocal': '2024-01-01T09:00:00',
+    'settlementTimeUTC': '2024-01-01T13:00:00.000Z',
+}
+DEFAULT_COMPACT_TRANSACTION_PRIVILEGE_LINE_ITEM = {
+    'description': 'Compact Privilege for Ohio',
+    'itemId': 'priv:aslp-oh',
+    'name': 'Ohio Compact Privilege',
+    'quantity': '1.0',
+    'taxable': 'False',
+    'unitPrice': '100.00',
+    'privilegeId': 'mock-privilege-id-oh',
+}
+DEFAULT_COMPACT_TRANSACTION_COMPACT_LINE_ITEM = {
+    'description': 'Compact fee applied for each privilege purchased',
+    'itemId': 'aslp-compact-fee',
+    'name': 'ASLP Compact Fee',
+    'quantity': '1',
+    'taxable': 'False',
+    'unitPrice': '10.50',
+}
+DEFAULT_COMPACT_TRANSACTION_FEE_LINE_ITEM = {
+    'description': 'credit card transaction fee',
+    'itemId': 'credit-card-transaction-fee',
+    'name': 'Credit Card Transaction Fee',
+    'quantity': '1',
+    'taxable': 'False',
+    'unitPrice': '3.00',
+}
 DEFAULT_PRIVILEGE_ID = 'SLP-NE-1'
 DEFAULT_MILITARY_AFFILIATION_TYPE = 'militaryMember'
 DEFAULT_MILITARY_STATUS = 'active'
@@ -57,6 +88,7 @@
 PRIVILEGE_RECORD_TYPE = 'privilege'
 PRIVILEGE_UPDATE_RECORD_TYPE = 'privilegeUpdate'
 PROVIDER_RECORD_TYPE = 'provider'
+TRANSACTION_RECORD_TYPE = 'transaction'
 
 # Privilege update default values
 DEFAULT_PRIVILEGE_UPDATE_TYPE = 'renewal'

backend/compact-connect/lambdas/python/common/common_test/test_constants.py

@@ -1,5 +1,6 @@
 # ruff: noqa: F403, F405 star import of test constants file
 import json
+from copy import deepcopy
 from datetime import date, datetime
 from decimal import Decimal
 
@@ -648,6 +649,38 @@ def put_default_jurisdiction_configuration_in_configuration_table(
 
         return jurisdiction_config
 
+    @staticmethod
+    def generate_default_transaction(value_overrides: dict | None = None):
+        """Generate a default transaction"""
+        from cc_common.data_model.schema.transaction import TransactionData
+
+        # We'll fill in any missing batch values with defaults
+        default_batch = deepcopy(DEFAULT_COMPACT_TRANSACTION_BATCH)
+        if value_overrides and 'batch' in value_overrides.keys():
+            default_batch.update(value_overrides.pop('batch'))
+
+        default_transaction = {
+            'transactionProcessor': 'authorize.net',
+            'transactionId': DEFAULT_COMPACT_TRANSACTION_ID,
+            'batch': default_batch,
+            'lineItems': [
+                DEFAULT_COMPACT_TRANSACTION_PRIVILEGE_LINE_ITEM,
+                DEFAULT_COMPACT_TRANSACTION_COMPACT_LINE_ITEM,
+                DEFAULT_COMPACT_TRANSACTION_FEE_LINE_ITEM,
+            ],
+            'compact': DEFAULT_COMPACT,
+            'licenseeId': DEFAULT_PROVIDER_ID,
+            'responseCode': '1',
+            'settleAmount': '113.50',
+            'submitTimeUTC': '2024-01-01T12:00:00.000Z',
+            'transactionStatus': 'settledSuccessfully',
+            'transactionType': 'authCaptureTransaction',
+        }
+        if value_overrides:
+            default_transaction.update(value_overrides)
+
+        return TransactionData.create_new(default_transaction)
+
     @staticmethod
     def put_compact_active_member_jurisdictions(
         compact: str = DEFAULT_COMPACT, postal_abbreviations: list[str] = None

backend/compact-connect/lambdas/python/common/common_test/test_data_generator.py

@@ -104,7 +104,9 @@ def setUpClass(cls):
         # Monkey-patch config object to be sure we have it based
         # on the env vars we set above
         import cc_common.config
+        from common_test.test_data_generator import TestDataGenerator
 
         cls.config = cc_common.config._Config()  # noqa: SLF001 protected-access
         cc_common.config.config = cls.config
         cls.mock_context = MagicMock(name='MockLambdaContext', spec=LambdaContext)
+        cls.test_data_generator = TestDataGenerator