wip subscription org

enrollments-events/app/enrollment.py

@@ -16,7 +16,7 @@ DeduplicationWindow = TypedDict('DeduplicationWindow', {'offset_days': int})
 
 class LinkedEntity(str):
     def __new__(cls, id: str, type: str) -> Self:
-        return super().__new__(cls, '#'.join([type.upper(), id]))
+        return super().__new__(cls, '#'.join([type.lower(), id]))
 
     def __init__(self, id: str, type: str) -> None:
         # __init__ is used to store the parameters for later reference.
@@ -33,7 +33,7 @@ class Slot:
     @property
     def order_id(self) -> LinkedEntity:
         idx, _ = self.sk.split('#')
-        return LinkedEntity(idx, 'order')
+        return LinkedEntity(idx, 'ORDER')
 
 
 class LifecycleEvents(str, Enum):
@@ -142,13 +142,15 @@ def enroll(
         )
 
         for entity in linked_entities:
-            type = entity.type.lower()
+            keyprefix = entity.type.lower()
             transact.put(
                 item={
                     'id': enrollment.id,
-                    'sk': f'linked_entities#{type}',
+                    # Post-migration: uncomment the following line
+                    # 'sk': f'LINKED_ENTITIES#{entity.type}',
+                    'sk': f'linked_entities#{entity.type}',
                     'created_at': now_,
-                    f'{type}_id': entity.id,
+                    f'{keyprefix}_id': entity.id,
                 }
             )
 
@@ -157,7 +159,7 @@ def enroll(
                 item={
                     'id': enrollment.id,
                     # Post-migration: uncomment the following line
-                    # 'sk': 'metadata#parent_slot',
+                    # 'sk': 'METADATA#SOURCE_SLOT',
                     'sk': 'parent_vacancy',
                     'vacancy': asdict(slot),
                     'created_at': now_,

enrollments-events/app/events/enroll.py

@@ -6,6 +6,7 @@ from aws_lambda_powertools.utilities.data_classes import (
     event_source,
 )
 from aws_lambda_powertools.utilities.typing import LambdaContext
+from layercake.batch import BatchProcessor
 from layercake.dateutils import now
 from layercake.dynamodb import (
     DynamoDBPersistenceLayer,
@@ -24,16 +25,17 @@ logger = Logger(__name__)
 order_layer = DynamoDBPersistenceLayer(ORDER_TABLE, dynamodb_client)
 course_layer = DynamoDBPersistenceLayer(COURSE_TABLE, dynamodb_client)
 enrollment_layer = DynamoDBPersistenceLayer(ENROLLMENT_TABLE, dynamodb_client)
+processor = BatchProcessor()
 
 
 @event_source(data_class=EventBridgeEvent)
 @logger.inject_lambda_context
-def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> list[str]:
+def lambda_handler(event: EventBridgeEvent, context: LambdaContext):
     new_image = event.detail['new_image']
-    now_ = now()
-    order_id = new_image['id']
     order = order_layer.collection.get_items(
-        TransactKey(order_id) + SortKey('0') + SortKey('items', path_spec='items'),
+        TransactKey(new_image['id'])
+        + SortKey('0')
+        + SortKey('items', path_spec='items'),
     )
     items = {
         item['id']: int(item['quantity'])
@@ -51,23 +53,18 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> list[str]
         cpf=order['cpf'],
     )
 
-    ids = []
+    with processor(
+        records=courses,
+        handler=_handler,
+        context={
+            'user': user,
+            'order_id': new_image['id'],
+        },
+    ) as fp:
+        result = fp.process()
+        logger.info('Processed courses', result=result)
 
-    for course in courses:
+    return order_layer.update_item(
-        enrollment = Enrollment(
-            id=uuid4(),
-            user=user,
-            course=course,
-        )
-        enroll(
-            enrollment,
-            persistence_layer=enrollment_layer,
-            deduplication_window=DeduplicationWindow(offset_days=90),
-            linked_entities=frozenset({LinkedEntity(order_id, 'ORDER')}),
-        )
-        ids.append(str(enrollment.id))
-
-    order_layer.update_item(
         key=KeyPair(new_image['id'], new_image['sk']),
         update_expr='SET #status = :status, updated_at = :updated_at',
         expr_attr_names={
@@ -75,12 +72,27 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> list[str]
         },
         expr_attr_values={
             ':status': 'SUCCESS',
-            ':updated_at': now_,
+            ':updated_at': now(),
         },
         cond_expr='attribute_exists(sk)',
     )
 
-    return ids
+
+def _handler(record: Course, context: dict) -> Enrollment:
+    enrollment = Enrollment(
+        id=uuid4(),
+        user=context['user'],
+        course=record,
+    )
+
+    enroll(
+        enrollment,
+        persistence_layer=enrollment_layer,
+        deduplication_window=DeduplicationWindow(offset_days=90),
+        linked_entities=frozenset({LinkedEntity(context['order_id'], 'ORDER')}),
+    )
+
+    return enrollment
 
 
 def _get_courses(ids: set) -> tuple[Course, ...]:
@@ -89,5 +101,4 @@ def _get_courses(ids: set) -> tuple[Course, ...]:
         KeyChain(pairs),
         flatten_top=False,
     )
-    courses = tuple(Course(id=idx, **obj) for idx, obj in result.items())  # type: ignore
+    return tuple(Course(id=idx, **obj) for idx, obj in result.items())  # type: ignore
-    return courses

enrollments-events/app/events/stopgap/patch_konviva.py

@@ -71,7 +71,7 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
 
 class CourseNotFoundError(Exception):
     def __init__(self, *args):
-        super().__init__('Course not found')
+        super().__init__('Course not found in SQLite')
 
 
 # Post-migration: remove the following function

enrollments-events/app/events/stopgap/set_terms_if_subscribed.py

@@ -0,0 +1,48 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import (
+    EventBridgeEvent,
+    event_source,
+)
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from layercake.dateutils import now
+from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair
+
+from boto3clients import dynamodb_client
+from config import (
+    ENROLLMENT_TABLE,
+    USER_TABLE,
+)
+
+logger = Logger(__name__)
+user_layer = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client)
+enrollment_layer = DynamoDBPersistenceLayer(ENROLLMENT_TABLE, dynamodb_client)
+
+
+@event_source(data_class=EventBridgeEvent)
+@logger.inject_lambda_context
+def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
+    new_image = event.detail['new_image']
+    data = user_layer.get_item(
+        # Post-migration: uncomment the following line
+        # KeyPair(new_image['org_id'], 'METADATA#BILLING_TERMS'),
+        KeyPair(new_image['tenant_id'], 'metadata#billing_policy'),
+    )
+
+    if not data:
+        return False
+
+    try:
+        enrollment_layer.put_item(
+            item={
+                'id': new_image['id'],
+                'sk': 'METADATA#BILLING_TERMS',
+                'org_id': new_image['tenant_id'],
+                'billing_day': data['billing_day'],
+                'created_at': now(),
+            },
+            cond_expr='attribute_not_exists(sk)',
+        )
+    except Exception:
+        return False
+    else:
+        return True

enrollments-events/template.yaml

@@ -42,10 +42,35 @@ Resources:
     Properties:
       RetentionInDays: 90
 
-  EventPatchEnrollFunction:
+  EventSetTermsIfSubscribedFunction:
     Type: AWS::Serverless::Function
     Properties:
-      Handler: events.stopgap.patch_enroll.lambda_handler
+      Handler: events.stopgap.set_terms_if_subscribed.lambda_handler
+      LoggingConfig:
+        LogGroup: !Ref EventLog
+      Policies:
+        - DynamoDBCrudPolicy:
+            TableName: !Ref EnrollmentTable
+        - DynamoDBReadPolicy:
+            TableName: !Ref UserTable
+      Events:
+        DynamoDBEvent:
+          Type: EventBridgeRule
+          Properties:
+            Pattern:
+              resources: [!Ref EnrollmentTable]
+              detail-type: [INSERT]
+              detail:
+                new_image:
+                  sk: ["0"]
+                  # Post-migration: rename `tenant_id` to `org_id`
+                  tenant_id:
+                    - exists: true
+
+  EventPatchCourseMetadataFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Handler: events.stopgap.patch_course_metadata.lambda_handler
       LoggingConfig:
         LogGroup: !Ref EventLog
       Policies:

enrollments-events/tests/events/stopgap/test_patch_course_metadata.py

@@ -1,4 +1,4 @@
-import app.events.stopgap.patch_enroll as app
+import app.events.stopgap.patch_course_metadata as app
 from aws_lambda_powertools.utilities.typing import LambdaContext
 from layercake.dynamodb import (
     DynamoDBPersistenceLayer,

enrollments-events/tests/events/test_enroll.py

@@ -1,6 +1,6 @@
 import app.events.enroll as app
 from aws_lambda_powertools.utilities.typing import LambdaContext
-from layercake.dynamodb import DynamoDBPersistenceLayer, PartitionKey
+from layercake.dynamodb import DynamoDBPersistenceLayer
 
 
 def test_enroll(
@@ -17,10 +17,4 @@ def test_enroll(
             }
         }
     }
-    ids = app.lambda_handler(event, lambda_context)  # type: ignore
+    assert app.lambda_handler(event, lambda_context)  # type: ignore
-    print(ids)
-
-    result = dynamodb_persistence_layer.collection.query(PartitionKey(str(ids[0])))
-    print(result)
-
-    # assert len(result['items']) == 4

order-events/app/events/assign_org_id.py

@@ -1,69 +0,0 @@
-from aws_lambda_powertools import Logger
-from aws_lambda_powertools.utilities.data_classes import (
-    EventBridgeEvent,
-    event_source,
-)
-from aws_lambda_powertools.utilities.typing import LambdaContext
-from layercake.dateutils import now
-from layercake.dynamodb import (
-    DynamoDBPersistenceLayer,
-    KeyPair,
-    SortKey,
-)
-
-from boto3clients import dynamodb_client
-from config import ORDER_TABLE, USER_TABLE
-
-logger = Logger(__name__)
-user_layer = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client)
-order_layer = DynamoDBPersistenceLayer(ORDER_TABLE, dynamodb_client)
-
-
-@event_source(data_class=EventBridgeEvent)
-@logger.inject_lambda_context
-def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
-    new_image = event.detail['new_image']
-    now_ = now()
-    data = user_layer.collection.get_items(
-        KeyPair(
-            pk='cnpj',
-            sk=SortKey(new_image['cnpj'], path_spec='user_id'),
-            rename_key='org_id',
-        )
-        + KeyPair(
-            pk='email',
-            sk=SortKey(new_image['email'], path_spec='user_id'),
-            rename_key='user_id',
-        ),
-        flatten_top=False,
-    )
-
-    # Sometimes the function executes before the user insertion completes,
-    # so an exception is raised to trigger a retry.
-    if len(data) < 2:
-        raise ValueError('IDs not found')
-
-    logger.info('IDs found', data=data)
-
-    with order_layer.transact_writer() as transact:
-        transact.update(
-            key=KeyPair(new_image['id'], '0'),
-            update_expr='SET tenant_id = :org_id, updated_at = :updated_at',
-            expr_attr_values={
-                ':org_id': data['org_id'],
-                ':updated_at': now_,
-            },
-        )
-
-        transact.update(
-            key=KeyPair(new_image['id'], 'author'),
-            update_expr='SET user_id = :user_id, updated_at = :updated_at',
-            expr_attr_values={
-                ':user_id': data['user_id'],
-                ':updated_at': now_,
-            },
-        )
-
-        logger.info('IDs updated')
-
-    return True

order-events/app/events/assign_user_id.py

@@ -1,55 +0,0 @@
-from aws_lambda_powertools import Logger
-from aws_lambda_powertools.utilities.data_classes import (
-    EventBridgeEvent,
-    event_source,
-)
-from aws_lambda_powertools.utilities.typing import LambdaContext
-from layercake.dateutils import now
-from layercake.dynamodb import (
-    DynamoDBPersistenceLayer,
-    KeyPair,
-    SortKey,
-)
-
-from boto3clients import dynamodb_client
-from config import ORDER_TABLE, USER_TABLE
-
-logger = Logger(__name__)
-user_layer = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client)
-order_layer = DynamoDBPersistenceLayer(ORDER_TABLE, dynamodb_client)
-
-
-@event_source(data_class=EventBridgeEvent)
-@logger.inject_lambda_context
-def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
-    new_image = event.detail['new_image']
-    now_ = now()
-    data = user_layer.collection.get_items(
-        KeyPair(
-            pk='cpf',
-            sk=SortKey(new_image['cpf'], path_spec='user_id'),
-            rename_key='user_id',
-        )
-        + KeyPair(
-            pk='email',
-            sk=SortKey(new_image['email'], path_spec='user_id'),
-            rename_key='user_id',
-        ),
-        flatten_top=False,
-    )
-
-    # Sometimes the function executes before the user insertion completes,
-    # so an exception is raised to trigger a retry.
-    if not data:
-        raise ValueError('User ID not found')
-
-    order_layer.update_item(
-        key=KeyPair(new_image['id'], '0'),
-        update_expr='SET user_id = :user_id, updated_at = :updated_at',
-        expr_attr_values={
-            ':user_id': data['user_id'],
-            ':updated_at': now_,
-        },
-    )
-
-    return True

order-events/app/events/remove_slots_if_canceled.py

@@ -34,15 +34,17 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
         limit=100,
     )
 
-    logger.info('Slots found', total_items=len(result['items']), slots=result['items'])
+    logger.info(
+        'Slots found',
+        total_items=len(result['items']),
+        slots=result['items'],
+    )
 
     with enrollment_layer.batch_writer() as batch:
         for pair in result['items']:
             batch.delete_item(
                 Key={
-                    # Post-migration: Uncomment the following line
+                    'id': {'S': pair['id']},
-                    # 'id': {'S': f'SLOT#ORG#{org_id}'},
-                    'id': {'S': f'vacancies#{org_id}'},
                     'sk': {'S': pair['sk']},
                 }
             )

order-events/app/events/stopgap/remove_slots.py

@@ -35,7 +35,7 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
 
     # Skip if billing policy is missing or order is less than or equal to zero
     if not policy or data['total'] <= 0:
-        logger.info('Missing billing policy.')
+        logger.info('Missing billing policy')
         return False
 
     logger.info(f'Billing policy from Org ID "{org_id}" found', policy=policy)
@@ -56,10 +56,9 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
 
     with enrollment_layer.batch_writer() as batch:
         for pair in result['items']:
-            org_id = pair['id']
             batch.delete_item(
                 Key={
-                    'id': {'S': f'vacancies#{org_id}'},
+                    'id': {'S': pair['id']},
                     'sk': {'S': pair['sk']},
                 }
             )

order-events/template.yaml

@@ -42,10 +42,10 @@ Resources:
     Properties:
       RetentionInDays: 90
 
-  EventAssignOrgIdFunction:
+  EventAppendOrgIdFunction:
     Type: AWS::Serverless::Function
     Properties:
-      Handler: events.assign_org_id.lambda_handler
+      Handler: events.append_org_id.lambda_handler
       LoggingConfig:
         LogGroup: !Ref EventLog
       Policies:
@@ -69,10 +69,10 @@ Resources:
                   tenant_id:
                     - exists: false
 
-  EventAssignUserIdFunction:
+  EventAppendUserIdFunction:
     Type: AWS::Serverless::Function
     Properties:
-      Handler: events.assign_user_id.lambda_handler
+      Handler: events.append_user_id.lambda_handler
       LoggingConfig:
         LogGroup: !Ref EventLog
       Policies:

order-events/tests/events/test_assign_org_id.py

@@ -1,27 +0,0 @@
-from aws_lambda_powertools.utilities.typing import LambdaContext
-from layercake.dynamodb import DynamoDBPersistenceLayer, PartitionKey
-
-import events.assign_org_id as app
-
-
-def test_assign_org_id(
-    dynamodb_seeds,
-    dynamodb_persistence_layer: DynamoDBPersistenceLayer,
-    lambda_context: LambdaContext,
-):
-    event = {
-        'detail': {
-            'new_image': {
-                'id': '9omWNKymwU5U4aeun6mWzZ',
-                'cnpj': '15608435000190',
-                'email': 'sergio@somosbeta.com.br',
-            }
-        }
-    }
-
-    assert app.lambda_handler(event, lambda_context)  # type: ignore
-
-    r = dynamodb_persistence_layer.collection.query(
-        PartitionKey('9omWNKymwU5U4aeun6mWzZ')
-    )
-    assert 2 == len(r['items'])

order-events/tests/events/test_assign_user_id.py

@@ -1,27 +0,0 @@
-from aws_lambda_powertools.utilities.typing import LambdaContext
-from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair
-
-import events.assign_user_id as app
-
-
-def test_assign_user_id(
-    dynamodb_seeds,
-    dynamodb_persistence_layer: DynamoDBPersistenceLayer,
-    lambda_context: LambdaContext,
-):
-    event = {
-        'detail': {
-            'new_image': {
-                'id': '9omWNKymwU5U4aeun6mWzZ',
-                'cpf': '07879819908',
-                'email': 'sergio@somosbeta.com.br',
-            }
-        }
-    }
-
-    assert app.lambda_handler(event, lambda_context)  # type: ignore
-
-    r = dynamodb_persistence_layer.collection.get_item(
-        KeyPair('9omWNKymwU5U4aeun6mWzZ', '0')
-    )
-    assert 'user_id' in r