add enroll later and now

api.saladeaula.digital/app/app.py

@@ -52,6 +52,7 @@ app.include_router(orgs.admins, prefix='/orgs')
 app.include_router(orgs.custom_pricing, prefix='/orgs')
 app.include_router(orgs.scheduled, prefix='/orgs')
 app.include_router(orgs.users, prefix='/orgs')
+app.include_router(orgs.batch_jobs, prefix='/orgs')
 
 
 @app.get('/health')

api.saladeaula.digital/app/config.py

@@ -1,5 +1,7 @@
 import os
 
+TZ = os.getenv('TZ', 'UTC')
+
 USER_TABLE: str = os.getenv('USER_TABLE')  # type: ignore
 ORDER_TABLE: str = os.getenv('ORDER_TABLE')  # type: ignore
 ENROLLMENT_TABLE: str = os.getenv('ENROLLMENT_TABLE')  # type: ignore
@@ -7,6 +9,8 @@ COURSE_TABLE: str = os.getenv('COURSE_TABLE')  # type: ignore
 
 BUCKET_NAME: str = os.getenv('BUCKET_NAME')  # type: ignore
 
+DEDUP_WINDOW_OFFSET_DAYS = 90
+
 PAPERFORGE_API = 'https://paperforge.saladeaula.digital'
 
 INTERNAL_EMAIL_DOMAIN = 'users.noreply.saladeaula.digital'

api.saladeaula.digital/app/middlewares/authentication_middleware.py

@@ -18,7 +18,7 @@ class User(BaseModel):
 
 class AuthenticationMiddleware(BaseMiddlewareHandler):
     """This middleware extracts user authentication details from
-    the jwt_claim authorizer context and makes them available
+    the `jwt_claim` authorizer context and makes them available
     in the application context.
     """
 

api.saladeaula.digital/app/routes/enrollments/enroll.py

@@ -1,6 +1,10 @@
+from datetime import date, datetime, time, timedelta
 from decimal import Decimal
+from http import HTTPStatus
 from typing import Annotated
+from uuid import uuid4
 
+import pytz
 from aws_lambda_powertools import Logger
 from aws_lambda_powertools.event_handler.api_gateway import Router
 from aws_lambda_powertools.event_handler.exceptions import (
@@ -8,12 +12,17 @@ from aws_lambda_powertools.event_handler.exceptions import (
 )
 from aws_lambda_powertools.event_handler.openapi.params import Body
 from layercake.batch import BatchProcessor
+from layercake.dateutils import now, ttl
 from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair
-from layercake.extra_types import CnpjStr, CpfStr, NameStr
-from pydantic import UUID4, BaseModel, EmailStr, FutureDate
+from layercake.extra_types import CpfStr, NameStr
+from layercake.strutils import md5_hash
+from pydantic import UUID4, BaseModel, EmailStr, Field, FutureDate
+from typing_extensions import TypedDict
 
+from api_gateway import JSONResponse
 from boto3clients import dynamodb_client
-from config import ENROLLMENT_TABLE
+from config import DEDUP_WINDOW_OFFSET_DAYS, ENROLLMENT_TABLE, TZ
+from exceptions import ConflictError
 from middlewares.authentication_middleware import User as Authenticated
 
 logger = Logger(__name__)
@@ -25,6 +34,9 @@ processor = BatchProcessor()
 class SubscriptionNotFoundError(NotFoundError): ...
 
 
+class DeduplicationConflictError(ConflictError): ...
+
+
 class User(BaseModel):
     id: str | UUID4
     name: NameStr
@@ -36,19 +48,28 @@ class Course(BaseModel):
     id: UUID4
     name: str
     access_period: int
-    unit_price: Decimal
+    unit_price: Decimal = Field(exclude=True)
+
+
+class DeduplicationWindow(BaseModel):
+    offset_days: int
+
+
+class SubscriptionTerms(BaseModel):
+    billing_day: int
 
 
 class Enrollment(BaseModel):
+    id: UUID4 = Field(default_factory=uuid4)
     user: User
     course: Course
     scheduled_for: FutureDate | None = None
+    deduplication_window: DeduplicationWindow | None = None
 
 
 class Org(BaseModel):
     id: str | UUID4
     name: str
-    cnpj: CnpjStr
 
 
 @router.post('/')
@@ -56,33 +77,207 @@ def enroll(
     org_id: Annotated[UUID4 | str, Body(embed=True)],
     enrollments: Annotated[tuple[Enrollment, ...], Body(embed=True)],
 ):
-    created_by: Authenticated = router.context['user']
     org = dyn.collection.get_items(
         KeyPair(
             pk=str(org_id),
             sk='0',
         )
+        + KeyPair(
+            pk=str(org_id),
+            sk='METADATA#SUBSCRIPTION_TERMS',
+            rename_key='terms',
+        )
         + KeyPair(
             pk='SUBSCRIPTION',
             sk=f'ORG#{org_id}',
-            rename_key='subscription',
+            rename_key='subscribed',
         )
     )
 
-    subscribed = 'subscription' in org
-    if not subscribed:
-        return checkout(Org.model_validate(org), enrollments, created_by=created_by)
+    if 'subscribed' not in org:
+        return JSONResponse(
+            status_code=HTTPStatus.NOT_ACCEPTABLE,
+        )
 
-    scheduled, unscheduled = [], []
-    for x in enrollments:
-        (scheduled if x.scheduled_for else unscheduled).append(x)
+    ctx = {
+        'org': Org.model_validate(org),
+        'created_by': router.context['user'],
+        'terms': SubscriptionTerms.model_validate(org['terms']),
+    }
 
-    print(scheduled, created_by)
+    now = [e for e in enrollments if not e.scheduled_for]
+    later = [e for e in enrollments if e.scheduled_for]
+
+    with processor(now, enroll_now, ctx) as batch:
+        now_out = batch.process()
+
+    with processor(later, enroll_later, ctx) as batch:
+        later_out = batch.process()
+
+    return {
+        'enrolled': now_out,
+        'scheduled': later_out,
+    }
 
 
-def checkout(
-    org: Org,
-    enrollments: tuple[Enrollment, ...],
-    created_by: Authenticated,
-):
-    print(org, enrollments, created_by)
+Context = TypedDict(
+    'Context',
+    {
+        'created_by': Authenticated,
+        'org': Org,
+        'terms': SubscriptionTerms,
+    },
+)
+
+
+def enroll_now(enrollment: Enrollment, context: Context):
+    now_ = now()
+    user = enrollment.user
+    course = enrollment.course
+    org: Org = context['org']
+    subscription_terms: SubscriptionTerms = context['terms']
+    created_by: Authenticated = context['created_by']
+    lock_hash = md5_hash(f'{user.id}{course.id}')
+    access_expires_at = now_ + timedelta(days=course.access_period)
+    deduplication_window = enrollment.deduplication_window
+    offset_days = (
+        int(deduplication_window.offset_days)
+        if deduplication_window
+        else DEDUP_WINDOW_OFFSET_DAYS
+    )
+    dedup_lock_ttl = ttl(
+        start_dt=now_,
+        days=course.access_period - offset_days,
+    )
+
+    with dyn.transact_writer() as transact:
+        transact.put(
+            item={
+                'id': enrollment.id,
+                'sk': '0',
+                'user': user.model_dump(),
+                'course': course.model_dump(),
+                'access_expires_at': access_expires_at,
+                'subscription_covered': True,
+                'org_id': org.id,
+                'created_at': now_,
+            }
+        )
+        transact.put(
+            item={
+                'id': enrollment.id,
+                'sk': 'ORG',
+                'name': org.name,
+                'org_id': org.id,
+                'created_at': now_,
+            }
+        )
+        transact.put(
+            item={
+                'id': enrollment.id,
+                'sk': 'CANCEL_POLICY',
+                'created_at': now_,
+            }
+        )
+        transact.put(
+            item={
+                'id': enrollment.id,
+                'sk': 'METADATA#SUBSCRIPTION_COVERED',
+                'org_id': org.id,
+                'billing_day': subscription_terms.billing_day,
+                'created_at': now_,
+            }
+        )
+        transact.put(
+            item={
+                'id': enrollment.id,
+                'sk': 'CREATED_BY',
+                'created_by': {
+                    'id': created_by.id,
+                    'name': created_by.name,
+                },
+                'created_at': now_,
+            }
+        )
+        transact.put(
+            item={
+                'id': enrollment.id,
+                'sk': 'LOCK',
+                'hash': lock_hash,
+                'created_at': now_,
+                'ttl': dedup_lock_ttl,
+            },
+        )
+        transact.put(
+            item={
+                'id': 'LOCK',
+                'sk': lock_hash,
+                'enrollment_id': enrollment.id,
+                'created_at': now_,
+                'ttl': dedup_lock_ttl,
+            },
+            cond_expr='attribute_not_exists(sk)',
+            exc_cls=DeduplicationConflictError,
+        )
+
+        # The deduplication window can be recalculated based on user settings.
+        if deduplication_window:
+            transact.put(
+                item={
+                    'id': enrollment.id,
+                    'sk': 'METADATA#DEDUPLICATION_WINDOW',
+                    'offset_days': offset_days,
+                    'created_at': now_,
+                },
+            )
+    return True
+
+
+def enroll_later(enrollment: Enrollment, context: Context):
+    now_ = now()
+    user = enrollment.user
+    course = enrollment.course
+    scheduled_for = date_to_midnight(enrollment.scheduled_for)  # type: ignore
+    deduplication_window = enrollment.deduplication_window
+    org: Org = context['org']
+    subscription_terms: SubscriptionTerms = context['terms']
+    created_by: Authenticated = context['created_by']
+    lock_hash = md5_hash(f'{user.id}{course.id}')
+
+    with dyn.transact_writer() as transact:
+        transact.put(
+            item={
+                'id': f'SCHEDULED#ORG#{org.id}',
+                'sk': f'{scheduled_for.isoformat()}#{lock_hash}',
+                'user': user.model_dump(),
+                'course': course.model_dump(),
+                'org': org.model_dump(),
+                'created_by': {
+                    'id': created_by.id,
+                    'name': created_by.name,
+                },
+                'subscription_covered': {
+                    'billing_day': subscription_terms.billing_day,
+                },
+                'ttl': ttl(start_dt=scheduled_for),
+                'created_at': now_,
+            }
+            | (
+                {'dedup_window_offset_days': deduplication_window.offset_days}
+                if deduplication_window
+                else {}
+            ),
+        )
+        transact.put(
+            item={
+                'id': 'LOCK#SCHEDULED',
+                'sk': lock_hash,
+                'created_at': now_,
+            },
+            cond_expr='attribute_not_exists(sk)',
+            exc_cls=DeduplicationConflictError,
+        )
+
+
+def date_to_midnight(dt: date) -> datetime:
+    return datetime.combine(dt, time(0, 0)).replace(tzinfo=pytz.timezone(TZ))

api.saladeaula.digital/app/routes/orgs/__init__.py

@@ -2,6 +2,7 @@ from .add import router as add
 from .admins import router as admins
 from .custom_pricing import router as custom_pricing
 from .enrollments.scheduled import router as scheduled
-from .users import router as users
+from .users.add import router as users
+from .users.batch_jobs import router as batch_jobs
 
-__all__ = ['add', 'admins', 'custom_pricing', 'scheduled', 'users']
+__all__ = ['add', 'admins', 'custom_pricing', 'scheduled', 'users', 'batch_jobs']

api.saladeaula.digital/app/routes/orgs/custom_pricing.py

@@ -12,5 +12,5 @@ dyn = DynamoDBPersistenceLayer(COURSE_TABLE, dynamodb_client)
 def get_custom_pricing(org_id: str):
     return dyn.collection.query(
         PartitionKey(f'CUSTOM_PRICING#ORG#{org_id}'),
-        limit=100,
+        limit=150,
     )

api.saladeaula.digital/app/routes/orgs/users/__init__.py

@@ -1,290 +0,0 @@
-from http import HTTPStatus
-from typing import Annotated
-from uuid import uuid4
-
-from aws_lambda_powertools.event_handler.api_gateway import Router
-from aws_lambda_powertools.event_handler.exceptions import NotFoundError
-from aws_lambda_powertools.event_handler.openapi.params import Body
-from layercake.dateutils import now, ttl
-from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair, SortKey
-from layercake.extra_types import CnpjStr, CpfStr, NameStr
-from pydantic import BaseModel, EmailStr, Field
-
-from api_gateway import JSONResponse
-from boto3clients import dynamodb_client
-from config import INTERNAL_EMAIL_DOMAIN, USER_TABLE
-from exceptions import ConflictError
-from middlewares.authentication_middleware import User as Authenticated
-
-router = Router()
-dyn = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client)
-
-
-class Org(BaseModel):
-    id: str | None = Field(default=None, exclude=True)
-    name: str
-    cnpj: CnpjStr
-
-
-class User(BaseModel):
-    name: NameStr
-    cpf: CpfStr
-    email: EmailStr
-
-
-class CPFConflictError(ConflictError): ...
-
-
-class EmailConflictError(ConflictError): ...
-
-
-class UserConflictError(ConflictError): ...
-
-
-class UserNotFoundError(NotFoundError): ...
-
-
-class OrgNotFoundError(NotFoundError): ...
-
-
-@router.post('/<org_id>/users')
-def add(
-    org_id: str,
-    user: Annotated[User, Body(embed=True)],
-    org: Annotated[Org, Body(embed=True)],
-):
-    org.id = org_id
-    created_by: Authenticated = router.context['user']
-
-    if _create_user(user, org, created_by):
-        return JSONResponse(HTTPStatus.CREATED)
-
-    user_id = _get_user_id(user)
-    _add_member(user_id, org)
-
-    return JSONResponse(HTTPStatus.NO_CONTENT)
-
-
-@router.delete('/<org_id>/users/<user_id>')
-def unlink(org_id: str, user_id: str):
-    with dyn.transact_writer() as transact:
-        transact.delete(
-            key=KeyPair(
-                pk=f'orgmembers#{org_id}',
-                # Post-migration: uncomment the following line
-                # pk=f'MEMBER#ORG#{org_id}',
-                sk=user_id,
-            )
-        )
-        transact.delete(
-            key=KeyPair(org_id, f'admins#{user_id}'),
-            # Post-migration: uncomment the following line
-            # key=KeyPair(org_id, f'ADMIN#{user_id}'),
-        )
-        transact.delete(
-            key=KeyPair(user_id, f'SCOPE#{org_id}'),
-        )
-        transact.delete(
-            key=KeyPair(
-                pk=user_id,
-                sk=f'orgs#{org_id}',
-                # Post-migration: uncomment the following line
-                # pk=f'ORG#{org_id}',
-            )
-        )
-        transact.update(
-            key=KeyPair(user_id, '0'),
-            update_expr='DELETE tenant_id :org_id',
-            # Post-migration: uncomment the following line
-            # update_expr='DELETE org_id :org_id',
-            expr_attr_values={':org_id': {org_id}},
-        )
-
-    return JSONResponse(HTTPStatus.NO_CONTENT)
-
-
-def _create_user(
-    user: User,
-    org: Org,
-    created_by: Authenticated,
-) -> bool:
-    now_ = now()
-    user_id = uuid4()
-    email_verified = INTERNAL_EMAIL_DOMAIN in user.email
-
-    try:
-        with dyn.transact_writer() as transact:
-            transact.put(
-                item=user.model_dump()
-                | {
-                    'id': user_id,
-                    'sk': '0',
-                    'email_verified': email_verified,
-                    'tenant_id': {org.id},
-                    # Post-migration (users): uncomment the folloing line
-                    # 'org_id': {org.id},
-                    # 'created_at': now_,
-                    'createDate': now_,
-                    # Makes the email searchable
-                    'emails': {user.email},
-                },
-            )
-            transact.put(
-                item={
-                    'id': user_id,
-                    'sk': 'NEVER_LOGGED',
-                    'created_at': now_,
-                }
-            )
-            transact.put(
-                item={
-                    'id': user_id,
-                    'sk': 'CREATED_BY',
-                    'created_by': {
-                        'id': created_by.id,
-                        'name': created_by.name,
-                    },
-                    'created_at': now_,
-                }
-            )
-            transact.put(
-                item={
-                    'id': user_id,
-                    # Post-migration: rename `emails` to `EMAIL`
-                    'sk': f'emails#{user.email}',
-                    'email_verified': email_verified,
-                    'email_primary': True,
-                    'created_at': now_,
-                    **({'mx_record_exists': True} if email_verified else {}),
-                }
-            )
-
-            if not email_verified:
-                transact.put(
-                    item={
-                        'id': user_id,
-                        'sk': f'EMAIL_VERIFICATION#{uuid4()}',
-                        'welcome': True,
-                        'name': user.name,
-                        'email': user.email,
-                        'org_name': org.name,
-                        'ttl': ttl(start_dt=now_, days=30),
-                        'created_at': now_,
-                    }
-                )
-
-            transact.put(
-                item={
-                    # Post-migration (users): rename `cpf` to `CPF`
-                    'id': 'cpf',
-                    'sk': user.cpf,
-                    'user_id': user_id,
-                    'created_at': now_,
-                },
-                cond_expr='attribute_not_exists(sk)',
-                exc_cls=CPFConflictError,
-            )
-            transact.put(
-                item={
-                    # Post-migration (users): rename `email` to `EMAIL`
-                    'id': 'email',
-                    'sk': user.email,
-                    'user_id': user_id,
-                    'created_at': now_,
-                },
-                cond_expr='attribute_not_exists(sk)',
-                exc_cls=EmailConflictError,
-            )
-            transact.put(
-                item={
-                    'id': user_id,
-                    'sk': f'orgs#{org.id}',
-                    # Post-migration (users): uncomment the following line
-                    # pk=f'ORG#{org.id}',
-                    'name': org.name,
-                    'cnpj': org.cnpj,
-                    'created_at': now_,
-                }
-            )
-            transact.put(
-                item={
-                    'id': f'orgmembers#{org.id}',
-                    # Post-migration (users): uncomment the following line
-                    # pk=f'MEMBER#ORG#{org_id}',
-                    'sk': user_id,
-                    'created_at': now_,
-                }
-            )
-            transact.condition(
-                key=KeyPair(org.id, '0'),  # type: ignore
-                cond_expr='attribute_exists(sk)',
-                exc_cls=OrgNotFoundError,
-            )
-    except (CPFConflictError, EmailConflictError):
-        return False
-    else:
-        return True
-
-
-def _add_member(user_id: str, org: Org) -> None:
-    now_ = now()
-
-    with dyn.transact_writer() as transact:
-        transact.update(
-            key=KeyPair(user_id, '0'),
-            # Post-migration (users): uncomment the following line
-            # update_expr='ADD org_id :org_id',
-            update_expr='ADD tenant_id :org_id',
-            expr_attr_values={
-                ':org_id': {org.id},
-            },
-            cond_expr='attribute_exists(sk)',
-            exc_cls=UserNotFoundError,
-        )
-        transact.put(
-            item={
-                'id': user_id,
-                # Post-migration (users): rename `orgs` to `ORG`
-                'sk': f'orgs#{org.id}',
-                'name': org.name,
-                'cnpj': org.cnpj,
-                'created_at': now_,
-            }
-        )
-        transact.put(
-            item={
-                # Post-migration (users): uncomment the following line
-                # pk=f'MEMBER#ORG#{org_id}',
-                'id': f'orgmembers#{org.id}',
-                'sk': user_id,
-                'created_at': now_,
-            },
-            cond_expr='attribute_not_exists(sk)',
-            exc_cls=UserConflictError,
-        )
-        transact.condition(
-            key=KeyPair(org.id, '0'),  # type: ignore
-            cond_expr='attribute_exists(sk)',
-            exc_cls=OrgNotFoundError,
-        )
-
-
-def _get_user_id(user: User) -> str:
-    user_id = dyn.collection.get_items(
-        KeyPair(
-            pk='email',
-            sk=SortKey(user.email, path_spec='user_id'),
-            rename_key='id',
-        )
-        + KeyPair(
-            pk='cpf',
-            sk=SortKey(user.cpf, path_spec='user_id'),
-            rename_key='id',
-        ),
-        flatten_top=False,
-    ).get('id')
-
-    if not user_id:
-        raise UserNotFoundError('User not found')
-
-    return user_id

api.saladeaula.digital/app/routes/orgs/users/add.py

@@ -0,0 +1,290 @@
+from http import HTTPStatus
+from typing import Annotated
+from uuid import uuid4
+
+from aws_lambda_powertools.event_handler.api_gateway import Router
+from aws_lambda_powertools.event_handler.exceptions import NotFoundError
+from aws_lambda_powertools.event_handler.openapi.params import Body
+from layercake.dateutils import now, ttl
+from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair, SortKey
+from layercake.extra_types import CnpjStr, CpfStr, NameStr
+from pydantic import BaseModel, EmailStr, Field
+
+from api_gateway import JSONResponse
+from boto3clients import dynamodb_client
+from config import INTERNAL_EMAIL_DOMAIN, USER_TABLE
+from exceptions import ConflictError
+from middlewares.authentication_middleware import User as Authenticated
+
+router = Router()
+dyn = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client)
+
+
+class Org(BaseModel):
+    id: str | None = Field(default=None, exclude=True)
+    name: str
+    cnpj: CnpjStr
+
+
+class User(BaseModel):
+    name: NameStr
+    cpf: CpfStr
+    email: EmailStr
+
+
+class CPFConflictError(ConflictError): ...
+
+
+class EmailConflictError(ConflictError): ...
+
+
+class UserConflictError(ConflictError): ...
+
+
+class UserNotFoundError(NotFoundError): ...
+
+
+class OrgNotFoundError(NotFoundError): ...
+
+
+@router.post('/<org_id>/users')
+def add(
+    org_id: str,
+    user: Annotated[User, Body(embed=True)],
+    org: Annotated[Org, Body(embed=True)],
+):
+    org.id = org_id
+    created_by: Authenticated = router.context['user']
+
+    if _create_user(user, org, created_by):
+        return JSONResponse(HTTPStatus.CREATED)
+
+    user_id = _get_user_id(user)
+    _add_member(user_id, org)
+
+    return JSONResponse(HTTPStatus.NO_CONTENT)
+
+
+@router.delete('/<org_id>/users/<user_id>')
+def unlink(org_id: str, user_id: str):
+    with dyn.transact_writer() as transact:
+        transact.delete(
+            key=KeyPair(
+                pk=f'orgmembers#{org_id}',
+                # Post-migration: uncomment the following line
+                # pk=f'MEMBER#ORG#{org_id}',
+                sk=user_id,
+            )
+        )
+        transact.delete(
+            key=KeyPair(org_id, f'admins#{user_id}'),
+            # Post-migration: uncomment the following line
+            # key=KeyPair(org_id, f'ADMIN#{user_id}'),
+        )
+        transact.delete(
+            key=KeyPair(user_id, f'SCOPE#{org_id}'),
+        )
+        transact.delete(
+            key=KeyPair(
+                pk=user_id,
+                sk=f'orgs#{org_id}',
+                # Post-migration: uncomment the following line
+                # pk=f'ORG#{org_id}',
+            )
+        )
+        transact.update(
+            key=KeyPair(user_id, '0'),
+            update_expr='DELETE tenant_id :org_id',
+            # Post-migration: uncomment the following line
+            # update_expr='DELETE org_id :org_id',
+            expr_attr_values={':org_id': {org_id}},
+        )
+
+    return JSONResponse(HTTPStatus.NO_CONTENT)
+
+
+def _create_user(
+    user: User,
+    org: Org,
+    created_by: Authenticated,
+) -> bool:
+    now_ = now()
+    user_id = uuid4()
+    email_verified = INTERNAL_EMAIL_DOMAIN in user.email
+
+    try:
+        with dyn.transact_writer() as transact:
+            transact.put(
+                item=user.model_dump()
+                | {
+                    'id': user_id,
+                    'sk': '0',
+                    'email_verified': email_verified,
+                    'tenant_id': {org.id},
+                    # Post-migration (users): uncomment the folloing line
+                    # 'org_id': {org.id},
+                    # 'created_at': now_,
+                    'createDate': now_,
+                    # Makes the email searchable
+                    'emails': {user.email},
+                },
+            )
+            transact.put(
+                item={
+                    'id': user_id,
+                    'sk': 'NEVER_LOGGED',
+                    'created_at': now_,
+                }
+            )
+            transact.put(
+                item={
+                    'id': user_id,
+                    'sk': 'CREATED_BY',
+                    'created_by': {
+                        'id': created_by.id,
+                        'name': created_by.name,
+                    },
+                    'created_at': now_,
+                }
+            )
+            transact.put(
+                item={
+                    'id': user_id,
+                    # Post-migration: rename `emails` to `EMAIL`
+                    'sk': f'emails#{user.email}',
+                    'email_verified': email_verified,
+                    'email_primary': True,
+                    'created_at': now_,
+                    **({'mx_record_exists': True} if email_verified else {}),
+                }
+            )
+
+            if not email_verified:
+                transact.put(
+                    item={
+                        'id': user_id,
+                        'sk': f'EMAIL_VERIFICATION#{uuid4()}',
+                        'welcome': True,
+                        'name': user.name,
+                        'email': user.email,
+                        'org_name': org.name,
+                        'ttl': ttl(start_dt=now_, days=30),
+                        'created_at': now_,
+                    }
+                )
+
+            transact.put(
+                item={
+                    # Post-migration (users): rename `cpf` to `CPF`
+                    'id': 'cpf',
+                    'sk': user.cpf,
+                    'user_id': user_id,
+                    'created_at': now_,
+                },
+                cond_expr='attribute_not_exists(sk)',
+                exc_cls=CPFConflictError,
+            )
+            transact.put(
+                item={
+                    # Post-migration (users): rename `email` to `EMAIL`
+                    'id': 'email',
+                    'sk': user.email,
+                    'user_id': user_id,
+                    'created_at': now_,
+                },
+                cond_expr='attribute_not_exists(sk)',
+                exc_cls=EmailConflictError,
+            )
+            transact.put(
+                item={
+                    'id': user_id,
+                    'sk': f'orgs#{org.id}',
+                    # Post-migration (users): uncomment the following line
+                    # pk=f'ORG#{org.id}',
+                    'name': org.name,
+                    'cnpj': org.cnpj,
+                    'created_at': now_,
+                }
+            )
+            transact.put(
+                item={
+                    'id': f'orgmembers#{org.id}',
+                    # Post-migration (users): uncomment the following line
+                    # pk=f'MEMBER#ORG#{org_id}',
+                    'sk': user_id,
+                    'created_at': now_,
+                }
+            )
+            transact.condition(
+                key=KeyPair(org.id, '0'),  # type: ignore
+                cond_expr='attribute_exists(sk)',
+                exc_cls=OrgNotFoundError,
+            )
+    except (CPFConflictError, EmailConflictError):
+        return False
+    else:
+        return True
+
+
+def _add_member(user_id: str, org: Org) -> None:
+    now_ = now()
+
+    with dyn.transact_writer() as transact:
+        transact.update(
+            key=KeyPair(user_id, '0'),
+            # Post-migration (users): uncomment the following line
+            # update_expr='ADD org_id :org_id',
+            update_expr='ADD tenant_id :org_id',
+            expr_attr_values={
+                ':org_id': {org.id},
+            },
+            cond_expr='attribute_exists(sk)',
+            exc_cls=UserNotFoundError,
+        )
+        transact.put(
+            item={
+                'id': user_id,
+                # Post-migration (users): rename `orgs` to `ORG`
+                'sk': f'orgs#{org.id}',
+                'name': org.name,
+                'cnpj': org.cnpj,
+                'created_at': now_,
+            }
+        )
+        transact.put(
+            item={
+                # Post-migration (users): uncomment the following line
+                # pk=f'MEMBER#ORG#{org_id}',
+                'id': f'orgmembers#{org.id}',
+                'sk': user_id,
+                'created_at': now_,
+            },
+            cond_expr='attribute_not_exists(sk)',
+            exc_cls=UserConflictError,
+        )
+        transact.condition(
+            key=KeyPair(org.id, '0'),  # type: ignore
+            cond_expr='attribute_exists(sk)',
+            exc_cls=OrgNotFoundError,
+        )
+
+
+def _get_user_id(user: User) -> str:
+    user_id = dyn.collection.get_items(
+        KeyPair(
+            pk='email',
+            sk=SortKey(user.email, path_spec='user_id'),
+            rename_key='id',
+        )
+        + KeyPair(
+            pk='cpf',
+            sk=SortKey(user.cpf, path_spec='user_id'),
+            rename_key='id',
+        ),
+        flatten_top=False,
+    ).get('id')
+
+    if not user_id:
+        raise UserNotFoundError('User not found')
+
+    return user_id