Migrate email sending mechanism from AWS Lambda to Celery task queue (#…

…500)

* Configure Celery workers on local machine

* Move email rendering from AWS Lambda to Celery

* Build emails before building backend and cleanup workers directory

* Pass necessary env variables to email renderer script

* Refactor backend and migraitons stack by extracting common resources to re-usable functions

* Implement a Celery workers CDK stack

* Rename Task class to LambdaTask in backend. Cleanup workers dir

* Add flower service to ECS

* Write documentation for introduced changes

* Add spot capacity provider to flower service

* Remove esbuild from workers and ignore infra+scripts from sonar

* Add a simple test for the send_email celery task

* Add missing env vars to backend tests

* Add more missing envs

docker-compose.ci.yml

@@ -3,6 +3,7 @@ version: "3.4"
 services:
   backend:
     volumes:
+      - ./packages/webapp-libs/webapp-emails/build/email-renderer/:/app/scripts/email/renderer:ro
       - ./packages/backend/cov:/app/cov
       - ./packages/backend/docs:/app/docs
     environment:
@@ -14,7 +15,6 @@ services:
 
   workers:
     volumes:
-      - ./packages/webapp-libs/webapp-emails/build/email-renderer/:/app/packages/workers/emails/renderer
       - ./packages/workers/cov:/app/packages/workers/cov
     environment:
       - CI=true

docker-compose.local.yml

@@ -1,8 +1,8 @@
-version: "3.4"
+version: '3.4'
 
 volumes:
   web_backend_db_data:
-    name: "${PROJECT_NAME}-web-backend-db-data"
+    name: '${PROJECT_NAME}-web-backend-db-data'
     external: true
 
   web_backend_staticfiles: {}
@@ -17,9 +17,12 @@ services:
 
   backend:
     volumes:
+      - ./packages/webapp-libs/webapp-emails/build/email-renderer/:/app/scripts/runtime/email/renderer:ro
       - ./packages/backend/:/app
       - ./packages/backend/docs:/app/docs
       - /app/__pypackages__
+      - /app/node_modules
+      - /app/cdk.out
       - web_backend_staticfiles:/app/static
     env_file:
       - ./packages/backend/.env
@@ -31,10 +34,22 @@ services:
       - localstack
       - mailcatcher
       - workers
+      - flower
+
+  celery_beat:
+    env_file:
+      - ./packages/backend/.env
+
+  celery_default:
+    command: ["./scripts/runtime/run_local_celery_worker_default.sh"]
+    volumes:
+      - ./packages/backend/:/app
+      - /app/__pypackages__
+    env_file:
+      - ./packages/backend/.env
 
   workers:
     volumes:
-      - ./packages/webapp-libs/webapp-emails/build/email-renderer/:/app/packages/workers/emails/renderer
       - ./packages/workers/:/app/packages/workers/
       - /app/packages/workers/node_modules/
       - /app/packages/workers/__pypackages__/
@@ -44,21 +59,31 @@ services:
     environment:
       - AWS_ENDPOINT_URL=http://localstack:4566
       - ENV_STAGE=${ENV_STAGE:-}
-
     depends_on:
       - db
       - mailcatcher
     ports:
-      - "3005:3005"
+      - '3005:3005'
 
   redis:
     volumes:
       - redis_cache:/data
 
+  flower:
+    image: '${PROJECT_NAME}/backend'
+    command: ['./scripts/runtime/run_local_celery_flower.sh']
+    env_file:
+      - ./packages/backend/.env
+    ports:
+      - '5555:5555'
+    depends_on:
+      redis:
+        condition: service_healthy
+
   localstack:
     image: localstack/localstack:2.3.0
     ports:
-      - "4566:4566"
+      - '4566:4566'
     environment:
       - SERVICES=serverless,events,cloudformation,ses,secretsmanager
       - DEFAULT_REGION=eu-west-1
@@ -70,8 +95,8 @@ services:
       - AWS_SECRET_ACCESS_KEY=bar
       - HOST_TMP_FOLDER=/tmp
     volumes:
-      - "/tmp/localstack:/tmp/localstack"
-      - "/var/run/docker.sock:/var/run/docker.sock"
+      - '/tmp/localstack:/tmp/localstack'
+      - '/var/run/docker.sock:/var/run/docker.sock'
     privileged: true
     depends_on:
       - db
@@ -81,6 +106,6 @@ services:
   mailcatcher:
     image: sj26/mailcatcher:v0.9.0
     ports:
-      - "1080:1080"
-      - "1025:1025"
+      - '1080:1080'
+      - '1025:1025'
     restart: always

docker-compose.yml

@@ -21,6 +21,7 @@ services:
     build:
       context: ./packages/backend
       target: backend
+    image: "${PROJECT_NAME}/backend"
     command: ["./scripts/runtime/run_local.sh"]
     ports:
       - "5001:5001"
@@ -42,6 +43,22 @@ services:
     stdin_open: true
     tty: true
 
+  celery_beat:
+    image: '${PROJECT_NAME}/backend'
+    command: ["./scripts/runtime/run_celery_beat.sh"]
+    restart: always
+    depends_on:
+      backend:
+        condition: service_started
+
+  celery_default:
+    image: '${PROJECT_NAME}/backend'
+    command: ["./scripts/runtime/run_celery_worker_default.sh"]
+    restart: always
+    depends_on:
+      backend:
+        condition: service_started
+
   workers:
     build:
       context: .

packages/backend/.env.shared

@@ -11,8 +11,8 @@ ENVIRONMENT_NAME=local
 [email protected]
 ADMIN_DEFAULT_PASSWORD=password
 
-TASKS_LOCAL_URL=http://workers:3005
-TASKS_BASE_HANDLER=common.tasks.TaskLocalInvoke
+LAMBDA_TASKS_LOCAL_URL=http://workers:3005
+LAMBDA_TASKS_BASE_HANDLER=common.tasks.LambdaTaskLocalInvoke
 
 DB_CONNECTION={"dbname":"backend","username":"backend","password":"backend","host":"db","port":5432}
 REDIS_CONNECTION=redis://redis:6379
@@ -43,4 +43,13 @@ AWS_XRAY_SDK_ENABLED=False
 
 OTP_AUTH_ISSUER_NAME=example.com
 
-OPENAI_API_KEY=<CHANGE_ME>
+OPENAI_API_KEY=<CHANGE_ME>
+
+EMAIL_BACKEND=django.core.mail.backends.smtp.EmailBackend
+EMAIL_HOST=mailcatcher
+EMAIL_PORT=1025
+[email protected]
+[email protected]
+
+VITE_EMAIL_ASSETS_URL=http://localhost:3000/email-assets
+VITE_WEB_APP_URL=http://localhost:3000

packages/backend/.gitignore

@@ -10,4 +10,6 @@ docs/
 
 # Coverage
 .coverage
-coverage.xml
+coverage.xml
+
+scripts/runtime/email/renderer/*

packages/backend/.test.env

@@ -15,7 +15,7 @@ REDIS_CONNECTION=redis://redis:6379
 
 WORKERS_EVENT_BUS_NAME=local-workers
 
-TASKS_BASE_HANDLER=common.tasks.Task
+LAMBDA_TASKS_BASE_HANDLER=common.tasks.LambdaTask
 
 AWS_DEFAULT_REGION=eu-west-1
 PARENT_HOST=example.org
@@ -31,4 +31,10 @@ AWS_ENDPOINT_URL=
 
 OTP_AUTH_ISSUER_NAME=example.com
 
-OPENAI_API_KEY=sk-example
+OPENAI_API_KEY=sk-example
+
+EMAIL_BACKEND=django.core.mail.backends.locmem.EmailBackend
+EMAIL_FROM_ADDRESS=[email protected]
+EMAIL_REPLY_ADDRESS=[email protected]
+VITE_EMAIL_ASSETS_URL=http://localhost:3000/email-assets
+VITE_WEB_APP_URL=http://localhost:3000

packages/backend/Dockerfile

@@ -15,12 +15,15 @@ ENV PYTHONUNBUFFERED 1
 ENV PIP_NO_CACHE_DIR off
 
 
-RUN apt-get update && apt-get install -y gcc postgresql-client ca-certificates jq \
+RUN apt-get update && apt-get install -y gcc postgresql-client ca-certificates jq curl \
   && update-ca-certificates \
   && pip install --upgrade pip \
   && pip install --no-cache-dir setuptools pdm~=2.5.2 awscli==1.32.24
 
 
+RUN curl -fsSL https://deb.nodesource.com/setup_20.x | bash - \
+  && apt-get --no-install-recommends install -y nodejs
+
 COPY --from=chamber /chamber /bin/chamber
 
 WORKDIR /pkgs

packages/backend/apps/content/tasks.py

@@ -5,11 +5,11 @@
 
 logger = logging.getLogger(__name__)
 
-module_name, package = settings.TASKS_BASE_HANDLER.rsplit(".", maxsplit=1)
-Task = getattr(importlib.import_module(module_name), package)
+module_name, package = settings.LAMBDA_TASKS_BASE_HANDLER.rsplit(".", maxsplit=1)
+LambdaTask = getattr(importlib.import_module(module_name), package)
 
 
-class ContentfulSync(Task):
+class ContentfulSync(LambdaTask):
     def __init__(self, name: str):
         super().__init__(name=name, source='backend.contentfulSync')
 

packages/backend/apps/finances/tests/test_webhooks.py

@@ -2,6 +2,7 @@
 
 import calleee
 import pytest
+from unittest.mock import patch
 from djstripe import models as djstripe_models
 from djstripe.enums import RefundStatus, RefundFailureReason
 
@@ -77,7 +78,8 @@ def test_subscription_schedule_from_subscription(
 
 
 class TestSendSubscriptionErrorEmail:
-    def test_send_email_on_invoice_payment_failed(self, webhook_event_factory, subscription, task_apply):
+    @patch('common.emails.send_email')
+    def test_send_email_on_invoice_payment_failed(self, send_email, webhook_event_factory, subscription):
         webhook_event = webhook_event_factory(
             type='invoice.payment_failed',
             data={
@@ -91,9 +93,12 @@ def test_send_email_on_invoice_payment_failed(self, webhook_event_factory, subsc
 
         webhook_event.invoke_webhook_handlers()
 
-        task_apply.assert_email_sent(notifications.SubscriptionErrorEmail, subscription.customer.subscriber.email)
+        send_email.apply_async.assert_called_with(
+            (subscription.customer.subscriber.email, notifications.SubscriptionErrorEmail.name, None)
+        )
 
-    def test_send_email_on_invoice_payment_required(self, webhook_event_factory, subscription, task_apply):
+    @patch('common.emails.send_email')
+    def test_send_email_on_invoice_payment_required(self, send_email, webhook_event_factory, subscription):
         webhook_event = webhook_event_factory(
             type='invoice.payment_action_required',
             data={
@@ -107,20 +112,27 @@ def test_send_email_on_invoice_payment_required(self, webhook_event_factory, sub
 
         webhook_event.invoke_webhook_handlers()
 
-        task_apply.assert_email_sent(notifications.SubscriptionErrorEmail, subscription.customer.subscriber.email)
+        send_email.apply_async.assert_called_with(
+            (subscription.customer.subscriber.email, notifications.SubscriptionErrorEmail.name, None)
+        )
 
 
 class TestSendTrialExpiresSoonEmail:
-    def test_previously_trialing_subscription_is_canceled(self, webhook_event_factory, customer, task_apply):
+    @patch('common.emails.send_email')
+    def test_previously_trialing_subscription_is_canceled(self, send_email, webhook_event_factory, customer):
         webhook_event = webhook_event_factory(
             type='customer.subscription.trial_will_end',
             data={'object': {'object': 'subscription', 'customer': customer.id, 'trial_end': 1617103425}},
         )
 
         webhook_event.invoke_webhook_handlers()
 
-        task_apply.assert_email_sent(
-            notifications.TrialExpiresSoonEmail, customer.subscriber.email, {'expiry_date': '2021-03-30T11:23:45Z'}
+        send_email.apply_async.assert_called_with(
+            (
+                customer.subscriber.email,
+                notifications.TrialExpiresSoonEmail.name,
+                {'expiry_date': '2021-03-30T11:23:45Z'},
+            )
         )
 
 

packages/backend/apps/users/tasks.py

@@ -2,10 +2,10 @@
 
 from django.conf import settings
 
-module_name, package = settings.TASKS_BASE_HANDLER.rsplit(".", maxsplit=1)
-Task = getattr(importlib.import_module(module_name), package)
+module_name, package = settings.LAMBDA_TASKS_BASE_HANDLER.rsplit(".", maxsplit=1)
+LambdaTask = getattr(importlib.import_module(module_name), package)
 
 
-class ExportUserData(Task):
+class ExportUserData(LambdaTask):
     def __init__(self):
         super().__init__(name="EXPORT_USER_DATA", source='backend.export_user')