Feat/license deactivates privileges

backend/bin/compile_requirements.sh

@@ -1,7 +1,9 @@
 set -e
 
-pip-compile --no-emit-index-url --upgrade --no-strip-extras multi-account/requirements-dev.in
-pip-compile --no-emit-index-url --upgrade --no-strip-extras multi-account/requirements.in
+pip-compile --no-emit-index-url --upgrade --no-strip-extras multi-account/control-tower/requirements-dev.in
+pip-compile --no-emit-index-url --upgrade --no-strip-extras multi-account/control-tower/requirements.in
+pip-compile --no-emit-index-url --upgrade --no-strip-extras multi-account/log-aggregation/requirements-dev.in
+pip-compile --no-emit-index-url --upgrade --no-strip-extras multi-account/log-aggregation/requirements.in
 pip-compile --no-emit-index-url --upgrade --no-strip-extras compact-connect/requirements-dev.in
 pip-compile --no-emit-index-url --upgrade --no-strip-extras compact-connect/requirements.in
 pip-compile --no-emit-index-url --upgrade --no-strip-extras compact-connect/lambdas/python/attestations/requirements-dev.in

backend/bin/run_python_tests.py

@@ -32,7 +32,8 @@
     'compact-connect/lambdas/python/staff-users',
     'compact-connect/lambdas/python/common',
     'compact-connect',  # CDK tests
-    'multi-account',
+    'multi-account/control-tower',
+    'multi-account/log-aggregation',
 )
 
 

backend/bin/sync_deps.sh

@@ -4,8 +4,10 @@
 )
 
 pip-sync \
-  multi-account/requirements-dev.txt \
-  multi-account/requirements.txt \
+  multi-account/control-tower/requirements-dev.txt \
+  multi-account/control-tower/requirements.txt \
+  multi-account/log-aggregation/requirements-dev.txt \
+  multi-account/log-aggregation/requirements.txt \
   compact-connect/requirements-dev.txt \
   compact-connect/requirements.txt \
   compact-connect/lambdas/python/attestations/requirements-dev.txt \

backend/compact-connect/common_constructs/queued_lambda_processor.py

@@ -28,6 +28,7 @@ def __init__(
         batch_size: int,
         encryption_key: IKey,
         alarm_topic: ITopic,
+        dlq_count_alarm_threshold: int = 10,
     ):
         super().__init__(scope, construct_id)
 
@@ -92,7 +93,11 @@ def __init__(
         )
 
         self._add_queue_alarms(
-            retention_period=retention_period, queue=self.queue, dlq=self.dlq, alarm_topic=alarm_topic
+            retention_period=retention_period,
+            queue=self.queue,
+            dlq=self.dlq,
+            alarm_topic=alarm_topic,
+            dlq_count_alarm_threshold=dlq_count_alarm_threshold,
         )
 
         QueryDefinition(
@@ -113,6 +118,7 @@ def _add_queue_alarms(
         queue: IQueue,
         dlq: IQueue,
         alarm_topic: ITopic,
+        dlq_count_alarm_threshold: int = 10,
     ):
         # Alarm if messages are older than half the queue retention period
         message_age_alarm = Alarm(
@@ -135,7 +141,7 @@ def _add_queue_alarms(
             'DLQMessagesAlarm',
             metric=dlq.metric_approximate_number_of_messages_visible(),
             evaluation_periods=1,
-            threshold=10,
+            threshold=dlq_count_alarm_threshold,
             actions_enabled=True,
             alarm_description=f'{dlq.node.path} high message volume',
             comparison_operator=ComparisonOperator.GREATER_THAN_THRESHOLD,

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

@@ -1,3 +1,4 @@
+import json
 import time
 from datetime import date
 from urllib.parse import quote
@@ -20,6 +21,7 @@
 )
 from cc_common.data_model.schema.military_affiliation.record import MilitaryAffiliationRecordSchema
 from cc_common.data_model.schema.privilege.record import PrivilegeUpdateRecordSchema
+from cc_common.event_batch_writer import EventBatchWriter
 from cc_common.exceptions import (
     CCAwsServiceException,
     CCInternalException,
@@ -846,7 +848,13 @@ def process_registration_values(
 
     @logger_inject_kwargs(logger, 'compact', 'provider_id', 'jurisdiction', 'license_type')
     def deactivate_privilege(
-        self, *, compact: str, provider_id: str, jurisdiction: str, license_type_abbr: str
+        self,
+        *,
+        compact: str,
+        provider_id: str,
+        jurisdiction: str,
+        license_type_abbr: str,
+        event_batch_writer: EventBatchWriter | None = None,
     ) -> None:
         """
         Deactivate a privilege for a provider in a jurisdiction.
@@ -928,5 +936,26 @@ def deactivate_privilege(
                 },
             ],
         )
+        event_entry = {
+            'Source': 'org.compactconnect.provider-data',
+            'DetailType': 'privilege.deactivation',
+            'Detail': json.dumps(
+                {
+                    'eventTime': config.current_standard_datetime.isoformat(),
+                    'compact': compact,
+                    'jurisdiction': jurisdiction,
+                    'providerId': provider_id,
+                    'licenseTypeAbbr': license_type_abbr,
+                    'privilegeId': privilege_record['privilegeId'],
+                }
+            ),
+            'EventBusName': config.event_bus_name,
+        }
+        # We'll support using a provided event batch writer to send the event to the event bus
+        if event_batch_writer:
+            event_batch_writer.put_event(Entry=event_entry)
+        else:
+            # If no event batch writer is provided, we'll use the default event bus client
+            config.events_client.put_events(Entries=[event_entry])
 
         return privilege_record

backend/compact-connect/lambdas/python/provider-data-v1/tests/unit/test_event_batch_writer.py → backend/compact-connect/lambdas/python/common/tests/unit/test_event_batch_writer.py

@@ -10,7 +10,7 @@
 
 class TestEventBatchWriter(TstLambdas):
     def test_write_big_batch(self):
-        from event_batch_writer import EventBatchWriter
+        from cc_common.event_batch_writer import EventBatchWriter
 
         put_count = []
 
@@ -38,7 +38,7 @@ def mock_put_items(Entries: list[dict]):  # noqa: N801 invalid-name
         self.assertEqual(13, mock_client.put_events.call_count)
 
     def test_write_small_batch(self):
-        from event_batch_writer import EventBatchWriter
+        from cc_common.event_batch_writer import EventBatchWriter
 
         put_count = []
 
@@ -66,7 +66,7 @@ def test_write_batch(self):
         """Making sure that, in the event that we exit with exactly 0 messages remaining, we don't try
         to put an empty batch
         """
-        from event_batch_writer import EventBatchWriter
+        from cc_common.event_batch_writer import EventBatchWriter
 
         put_count = []
 
@@ -93,7 +93,7 @@ def mock_put_items(Entries: list[dict]):  # noqa: N801 invalid-name
         self.assertEqual(1, mock_client.put_events.call_count)
 
     def test_exception_recovery(self):
-        from event_batch_writer import EventBatchWriter
+        from cc_common.event_batch_writer import EventBatchWriter
 
         put_count = []
 
@@ -127,15 +127,15 @@ def test_bad_use(self):
         """EventBatchWriter requires that it be used as a context manager (in a `with EventBatchWriter(...):` block)
         Trying to use it otherwise should raise an exception.
         """
-        from event_batch_writer import EventBatchWriter
+        from cc_common.event_batch_writer import EventBatchWriter
 
         # If a developer uses this wrong
         writer = EventBatchWriter(MagicMock())
         with self.assertRaises(RuntimeError):
             writer.put_event(Entry={})
 
     def test_entry_failures(self):
-        from event_batch_writer import EventBatchWriter
+        from cc_common.event_batch_writer import EventBatchWriter
 
         put_count = []
 
@@ -170,7 +170,7 @@ def mock_put_items(Entries: list[dict]):  # noqa: N801 invalid-name
 
     def test_write_custom_batch_size(self):
         """Override the default batch size of 10"""
-        from event_batch_writer import EventBatchWriter
+        from cc_common.event_batch_writer import EventBatchWriter
 
         put_count = []
 

backend/compact-connect/lambdas/python/provider-data-v1/handlers/bulk_upload.py

@@ -8,14 +8,14 @@
 from botocore.response import StreamingBody
 from cc_common.config import config, logger
 from cc_common.data_model.schema.license.api import LicenseGeneralResponseSchema, LicensePostRequestSchema
+from cc_common.event_batch_writer import EventBatchWriter
 from cc_common.exceptions import CCInternalException
 from cc_common.utils import (
     ResponseEncoder,
     api_handler,
     authorize_compact_jurisdiction,
     send_licenses_to_preprocessing_queue,
 )
-from event_batch_writer import EventBatchWriter
 from license_csv_reader import LicenseCSVReader
 from marshmallow import ValidationError
 

backend/compact-connect/lambdas/python/provider-data-v1/handlers/ingest.py

@@ -7,9 +7,9 @@
 from cc_common.data_model.schema.common import ProviderEligibilityStatus, UpdateCategory
 from cc_common.data_model.schema.license.ingest import LicenseIngestSchema
 from cc_common.data_model.schema.license.record import LicenseUpdateRecordSchema
+from cc_common.event_batch_writer import EventBatchWriter
 from cc_common.exceptions import CCNotFoundException
 from cc_common.utils import sqs_handler
-from event_batch_writer import EventBatchWriter
 
 license_schema = LicenseIngestSchema()
 license_update_schema = LicenseUpdateRecordSchema()

backend/compact-connect/lambdas/python/provider-data-v1/handlers/privileges.py

@@ -4,9 +4,9 @@
 from aws_lambda_powertools.utilities.typing import LambdaContext
 from cc_common.config import config, logger, metrics
 from cc_common.data_model.schema.common import CCPermissionsAction
+from cc_common.event_batch_writer import EventBatchWriter
 from cc_common.exceptions import CCAccessDeniedException, CCInternalException, CCInvalidRequestException
-from cc_common.utils import api_handler, authorize_compact, get_event_scopes
-from event_batch_writer import EventBatchWriter
+from cc_common.utils import api_handler, authorize_compact, get_event_scopes, sqs_handler
 
 
 @api_handler
@@ -42,40 +42,47 @@ def deactivate_privilege(event: dict, context: LambdaContext):  # noqa: ARG001 u
             raise CCInvalidRequestException(f'Invalid license type abbreviation: {license_type_abbr}')
 
         logger.info('Deactivating privilege')
-        deactivated_privilege_record = config.data_client.deactivate_privilege(
+        config.data_client.deactivate_privilege(
             compact=compact,
             provider_id=provider_id,
             jurisdiction=jurisdiction,
             license_type_abbr=license_type_abbr,
         )
-        with EventBatchWriter(config.events_client) as event_writer:
-            event_writer.put_event(
-                Entry={
-                    'Source': 'org.compactconnect.provider-data',
-                    'DetailType': 'privilege.deactivation',
-                    'Detail': json.dumps(
-                        {
-                            'eventTime': config.current_standard_datetime.isoformat(),
-                            'compact': compact,
-                            'jurisdiction': jurisdiction,
-                            'providerId': provider_id,
-                        }
-                    ),
-                    'EventBusName': config.event_bus_name,
-                }
-            )
+    return {'message': 'OK'}
+
 
-        # Send email notifications for privilege deactivation
-        failed_to_send_notification = False
-        deactivated_privilege_id = deactivated_privilege_record['privilegeId']
+@sqs_handler
+def privilege_deactivation_message_handler(message: dict):
+    """
+    Handle privilege deactivation messages from the event bus.
+    This handler is responsible for sending email notifications to both the provider
+    and the jurisdiction when a privilege is deactivated.
+
+    If notification sending fails, the function will raise a CCInternalException,
+    which will cause the SQS handler decorator to report the message as a failure
+    and the message will be retried according to the queue's retry policy.
+    """
+    compact = message.get('compact')
+    provider_id = message.get('providerId')
+    jurisdiction = message.get('jurisdiction')
+    privilege_id = message.get('privilegeId')
+
+    with logger.append_context_keys(
+        compact=compact, provider_id=provider_id, jurisdiction=jurisdiction, privilege_id=privilege_id
+    ):
+        logger.info('Processing privilege deactivation notification')
+
         # Get provider information to retrieve email and name
         provider = config.data_client.get_provider(compact=compact, provider_id=provider_id, detail=False)['items'][0]
+
+        error_messages = []
+
+        # Send notification to the provider
         try:
             provider_email = provider.get('compactConnectRegisteredEmailAddress')
             if not provider_email:
                 logger.error('Provider email not found, cannot send provider notification', provider_id=provider_id)
             else:
-                # Send notification to the provider
                 logger.info(
                     'Sending privilege deactivation notification to provider',
                     provider_id=provider_id,
@@ -84,32 +91,30 @@ def deactivate_privilege(event: dict, context: LambdaContext):  # noqa: ARG001 u
                 config.email_service_client.send_provider_privilege_deactivation_email(
                     compact=compact,
                     provider_email=provider_email,
-                    privilege_id=deactivated_privilege_id,
+                    privilege_id=privilege_id,
                 )
         except CCInternalException as e:
-            # Log the error but don't fail the deactivation process
-            logger.error('Failed to send provider privilege deactivation notifications', exception=str(e))
-            failed_to_send_notification = True
+            error_message = f'Failed to send provider privilege deactivation notification: {str(e)}'
+            logger.error(error_message, exc_info=e)
+            error_messages.append(error_message)
 
+        # Send notification to the jurisdiction
         try:
-            # Send notification to the jurisdiction
             logger.info('Sending privilege deactivation notification to jurisdiction', jurisdiction=jurisdiction)
             provider_first_name = provider['givenName']
             provider_last_name = provider['familyName']
             config.email_service_client.send_jurisdiction_privilege_deactivation_email(
                 compact=compact,
                 jurisdiction=jurisdiction,
-                privilege_id=deactivated_privilege_id,
+                privilege_id=privilege_id,
                 provider_first_name=provider_first_name,
                 provider_last_name=provider_last_name,
             )
         except CCInternalException as e:
-            # Log the error but don't fail the deactivation process
-            logger.error('Failed to send jurisdiction privilege deactivation notifications', exception=str(e))
-            failed_to_send_notification = True
+            error_message = f'Failed to send jurisdiction privilege deactivation notification: {str(e)}'
+            logger.error(error_message, exc_info=e)
+            error_messages.append(error_message)
 
-        if failed_to_send_notification:
-            logger.error('One or more deactivation notifications failed to send. Pushing metric to fire alert.')
-            metrics.add_metric(name='privilege-deactivation-notification-failed', unit=MetricUnit.Count, value=1)
-
-    return {'message': 'OK'}
+        # If there were any errors, raise an exception to trigger a retry
+        if error_messages:
+            raise CCInternalException('; '.join(error_messages))