Job diário de emails da régua de trial — Plano de implementação

TLDR: constrói o model de histórico e idempotência, o service que decide quais emails da régua estão devidos hoje, o wrapper de envio e a task Celery diária das 10h que faz o fan-out.

Spec: .project/docs/specs/20260825095058_trial_journey_emails_job.md Branch: feat/trial-journey-emails-job

Arquitetura: um service puro (apps/trials/services/trial_journey.py) concentra a régua e toda a decisão — sem rede, sem envio — e é onde mora a maior parte dos testes. Uma task orquestradora varre os UserTrial elegíveis, reserva o envio no banco e faz fan-out para a task de email existente. A camada de email ganha um único método genérico parametrizado por template, em vez de seis métodos quase idênticos.

Stack: Django 5.2, Celery + RabbitMQ, pytest + pytest-django, SMTP via EmailMultiAlternatives.

Restrições globais

Nenhum arquivo em apps/emails/templates/emails/ pode ser criado, alterado ou removido.
O contexto do template é exatamente {"name", "cta_url"} — nunca preço, prazo ou data.
Comando de teste do projeto é make run.test (aceita path=<path>), não make test como diz o CLAUDE.md.
Toda query de UserTrial usa objects (soft-deleted já excluídos), nunca all_objects.
Commits em uma linha, máximo 60 caracteres, sem menção a IA.

Baseline

make run.test verde antes de começar. make check e make test não existem neste projeto.

Ordem e paralelismo

Task 1 e Task 3 são independentes entre si e podem rodar em paralelo.
Task 2 depende da Task 1 (lê email_dispatches).
Task 4 depende de 1, 2 e 3.
Task 5 depende da Task 4.

Task 1: Model genérico de idempotência `EmailDispatch`

Files: - Create: apps/emails/models/__init__.py - Create: apps/emails/models/email_dispatch.py - Create: apps/emails/migrations/0001_emaildispatch.py - Test: tests/emails/test_email_dispatch.py

Interfaces: - Consumes: settings.AUTH_USER_MODEL, BaseModel de apps.common.models - Produces: EmailDispatch com user, source, email_key, sent_at e related_name="email_dispatches" em User

O histórico de envio vive em apps/emails desde o início, não em apps/trials — é o app dono de SendEmails, tasks.py e templates, e qualquer régua futura (não só trial) registra envio na mesma tabela. source é um CharField livre, sem choices fixas: cada app dono do envio define sua própria constante (ex: TRIAL_EMAIL_SOURCE = "trial" na Task 2) em vez de editar este model a cada novo sistema de comunicação.

[ ] Step 1: Write the failing test

```python # tests/emails/test_email_dispatch.py import pytest from django.db import IntegrityError from django.utils import timezone

from apps.emails.models import EmailDispatch

@pytest.mark.django_db def test_dispatch_is_unique_per_user_source_and_email_key(user): # arrange now = timezone.now() EmailDispatch.objects.create(user=user, source=”trial”, email_key=”daily_audio”, sent_at=now)

# act / assert
with pytest.raises(IntegrityError):
    EmailDispatch.objects.create(user=user, source="trial", email_key="daily_audio", sent_at=now)

@pytest.mark.django_db def test_dispatch_allows_different_email_keys_for_same_user_and_source(user): # arrange now = timezone.now()

# act
EmailDispatch.objects.create(user=user, source="trial", email_key="daily_audio", sent_at=now)
EmailDispatch.objects.create(user=user, source="trial", email_key="normose", sent_at=now)

# assert
assert user.email_dispatches.count() == 2

@pytest.mark.django_db def test_dispatch_allows_different_sources_for_same_user_and_email_key(user): # arrange now = timezone.now()

# act
EmailDispatch.objects.create(user=user, source="trial", email_key="welcome", sent_at=now)
EmailDispatch.objects.create(user=user, source="campaign", email_key="welcome", sent_at=now)

# assert
assert user.email_dispatches.count() == 2 ```

user é a fixture genérica de tests/conftest.py.

[ ] Step 2: Run to verify it fails

bash make run.test path=tests/emails/test_email_dispatch.py

Expected: FAIL — ModuleNotFoundError: No module named 'apps.emails.models'

[ ] Step 3: Write minimal implementation

```python # apps/emails/models/email_dispatch.py from django.contrib.auth import get_user_model

from apps.common.models import BaseModel, models

User = get_user_model()

class EmailDispatch(BaseModel): user = models.ForeignKey( User, on_delete=models.CASCADE, related_name=”email_dispatches”, verbose_name=”Usuário”, ) source = models.CharField(max_length=32, verbose_name=”Origem”) email_key = models.CharField(max_length=32, verbose_name=”Email”) sent_at = models.DateTimeField(verbose_name=”Enviado em”)

class Meta:
    verbose_name = "Envio de Email"
    verbose_name_plural = "Envios de Email"
    ordering = ["-sent_at"]
    unique_together = ("user", "source", "email_key")

def __str__(self):
    return f"{self.user} — {self.source}:{self.email_key}" ```

python # apps/emails/models/__init__.py from .email_dispatch import EmailDispatch

Gerar a migração (primeira do app emails, que hoje não tem models.py):

bash python manage.py makemigrations emails --name emaildispatch

[ ] Step 4: Run to verify it passes

bash make run.test path=tests/emails/test_email_dispatch.py

Expected: PASS

[ ] Step 5: Commit

bash git add apps/emails/models/ apps/emails/migrations/ tests/emails/test_email_dispatch.py git commit -m "feat: add generic email dispatch model"

Task 2: Service da régua e resolução do email do dia

Files: - Create: apps/trials/services/trial_journey.py - Modify: apps/trials/services/__init__.py - Test: tests/trials/test_trial_journey.py

Interfaces: - Consumes: EmailDispatch (Task 1) e UserTrial de apps.trials.models, DOWNLOAD_PAGE_URL de routes.download - Produces: - JOURNEY_EMAILS: dict[str, JourneyEmail] — JourneyEmail(key, template_name, subject, percent) - PATTERN_SCAN_DELAY_DAYS: int = 5 - TRIAL_EMAIL_SOURCE: str = "trial" — constante de source usada em todo EmailDispatch gravado pela régua de trial - consumed_percent(user_trial, now) -> float | None - resolve_pending_emails(user_trial, now=None) -> list[str] - eligible_user_trials(now=None) -> QuerySet[UserTrial] - cta_url_for(user_trial, email_key) -> str

[ ] Step 1: Write the failing test

```python # tests/trials/test_trial_journey.py from datetime import timedelta

import pytest from django.utils import timezone

from apps.emails.models import EmailDispatch from apps.trials.models import Trial, UserTrial from apps.trials.services.trial_journey import ( TRIAL_EMAIL_SOURCE, consumed_percent, cta_url_for, eligible_user_trials, resolve_pending_emails, )

def make_trial(duration_days=7, communication=True): return Trial.objects.create( name=f”Trial {duration_days} dias”, status=Trial.Status.ACTIVE, duration_days=duration_days, start_date=timezone.now() - timedelta(days=1), end_date=timezone.now() + timedelta(days=60), checkout_url=”https://checkout.example.com/trial”, active_comunication_email=communication, )

def make_user_trial(user, trial, started_at): return UserTrial.objects.create( user=user, trial=trial, started_at=started_at, expires_at=started_at + timedelta(days=trial.duration_days), )

@pytest.mark.django_db def test_consumed_percent_is_none_when_duration_is_zero(trial_user): trial = make_trial(duration_days=0) now = timezone.now() user_trial = make_user_trial(trial_user, trial, now)

assert consumed_percent(user_trial, now) is None

@pytest.mark.django_db @pytest.mark.parametrize( “elapsed_days,expected”, [ (0.5, []), # 7% — nenhum marco (1.0, [“daily_audio”]), # 14% — só o 10 (3.0, [“daily_audio”, “normose”]), # 43% — 10 e 30 (4.0, [“daily_audio”, “normose”, “habits”]), # 57% — 10, 30 e 50 (5.0, [“daily_audio”, “normose”, “habits”, “chat_ai”]), # 71% — 10, 30, 50 e 70 ], ) def test_resolve_returns_every_reached_marker_in_order(trial_user, elapsed_days, expected): trial = make_trial(duration_days=7) started_at = timezone.now() - timedelta(days=elapsed_days) user_trial = make_user_trial(trial_user, trial, started_at)

assert resolve_pending_emails(user_trial) == expected

@pytest.mark.django_db def test_resolve_omits_markers_already_sent(trial_user): trial = make_trial(duration_days=7) now = timezone.now() user_trial = make_user_trial(trial_user, trial, now - timedelta(days=4)) # 57%

EmailDispatch.objects.create(user=user_trial.user, source=TRIAL_EMAIL_SOURCE, email_key="daily_audio", sent_at=now)

assert resolve_pending_emails(user_trial, now) == ["normose", "habits"]

@pytest.mark.django_db def test_resolve_returns_empty_when_everything_was_sent(trial_user): trial = make_trial(duration_days=7) now = timezone.now() user_trial = make_user_trial(trial_user, trial, now - timedelta(days=5)) # 71%

for key in ["daily_audio", "normose", "habits", "chat_ai"]:
    EmailDispatch.objects.create(user=user_trial.user, source=TRIAL_EMAIL_SOURCE, email_key=key, sent_at=now)

assert resolve_pending_emails(user_trial, now) == []

@pytest.mark.django_db def test_resolve_returns_conversion_on_expiration_day(trial_user): trial = make_trial(duration_days=7) now = timezone.now() user_trial = make_user_trial(trial_user, trial, now - timedelta(days=7))

assert "plan_ended" in resolve_pending_emails(user_trial, now)

@pytest.mark.django_db def test_conversion_goes_out_together_with_pending_content_marker(trial_user): “"”No dia da expiração um marco de conteúdo pendente sai junto, não é descartado.””” trial = make_trial(duration_days=7) now = timezone.now() user_trial = make_user_trial(trial_user, trial, now - timedelta(days=7))

for key in ["daily_audio", "normose", "habits"]:
    EmailDispatch.objects.create(user=user_trial.user, source=TRIAL_EMAIL_SOURCE, email_key=key, sent_at=now)

assert resolve_pending_emails(user_trial, now) == ["chat_ai", "plan_ended"]

@pytest.mark.django_db def test_resolve_returns_pattern_scan_five_days_after_expiration(trial_user): trial = make_trial(duration_days=7) now = timezone.now() user_trial = make_user_trial(trial_user, trial, now - timedelta(days=12))

assert "patern_scan" in resolve_pending_emails(user_trial, now)

@pytest.mark.django_db def test_pattern_scan_not_yet_due_before_day_five(trial_user): trial = make_trial(duration_days=7) now = timezone.now() user_trial = make_user_trial(trial_user, trial, now - timedelta(days=11)) # D+4

assert "patern_scan" not in resolve_pending_emails(user_trial, now)

@pytest.mark.django_db def test_seven_day_trial_delivers_every_email_worst_case(trial_user): “"”Cadastro às 9h59 é o pior caso: o job das 10h do dia 1 pega ~0% consumido.””” trial = make_trial(duration_days=7) started_at = timezone.localtime().replace(hour=9, minute=59, second=0, microsecond=0) user_trial = make_user_trial(trial_user, trial, started_at)

delivered = []
for day in range(1, 13):
    run_at = (started_at + timedelta(days=day)).replace(hour=10, minute=0)
    for key in resolve_pending_emails(user_trial, run_at):
        delivered.append(key)
        EmailDispatch.objects.create(user=user_trial.user, source=TRIAL_EMAIL_SOURCE, email_key=key, sent_at=run_at)

assert delivered == ["daily_audio", "normose", "habits", "chat_ai", "plan_ended", "patern_scan"]

@pytest.mark.django_db def test_three_day_trial_loses_no_email(trial_user): “"”Trial curto cruza vários marcos por dia — todos precisam sair.””” trial = make_trial(duration_days=3) started_at = timezone.localtime().replace(hour=14, minute=0, second=0, microsecond=0) user_trial = make_user_trial(trial_user, trial, started_at)

delivered = []
for day in range(1, 10):
    run_at = (started_at + timedelta(days=day)).replace(hour=10, minute=0)
    for key in resolve_pending_emails(user_trial, run_at):
        delivered.append(key)
        EmailDispatch.objects.create(user=user_trial.user, source=TRIAL_EMAIL_SOURCE, email_key=key, sent_at=run_at)

assert sorted(delivered) == sorted(
    ["daily_audio", "normose", "habits", "chat_ai", "plan_ended", "patern_scan"]
)

@pytest.mark.django_db def test_eligible_excludes_trial_without_communication(trial_user): trial = make_trial(communication=False) make_user_trial(trial_user, trial, timezone.now() - timedelta(days=1))

assert eligible_user_trials().count() == 0

@pytest.mark.django_db def test_eligible_excludes_converted_user_trial(trial_user): trial = make_trial() user_trial = make_user_trial(trial_user, trial, timezone.now() - timedelta(days=1)) user_trial.converted_at = timezone.now() user_trial.save(update_fields=[“converted_at”])

assert eligible_user_trials().count() == 0

@pytest.mark.django_db def test_eligible_excludes_user_who_left_trial_status(subscriber_user): “"”converted_at nulo mas subscription_status mudou — o furo do R-012.””” trial = make_trial() make_user_trial(subscriber_user, trial, timezone.now() - timedelta(days=1))

assert eligible_user_trials().count() == 0

@pytest.mark.django_db def test_eligible_excludes_user_trial_past_pattern_scan_window(trial_user): trial = make_trial() make_user_trial(trial_user, trial, timezone.now() - timedelta(days=20))

assert eligible_user_trials().count() == 0

@pytest.mark.django_db def test_eligible_includes_active_trial_user(trial_user): trial = make_trial() make_user_trial(trial_user, trial, timezone.now() - timedelta(days=1))

assert eligible_user_trials().count() == 1

@pytest.mark.django_db def test_cta_url_uses_checkout_for_conversion_and_pattern_scan(trial_user): trial = make_trial() user_trial = make_user_trial(trial_user, trial, timezone.now() - timedelta(days=1))

assert cta_url_for(user_trial, "plan_ended") == trial.checkout_url
assert cta_url_for(user_trial, "patern_scan") == trial.checkout_url
assert cta_url_for(user_trial, "daily_audio") != trial.checkout_url ```

[ ] Step 2: Run to verify it fails

bash make run.test path=tests/trials/test_trial_journey.py

Expected: FAIL — ModuleNotFoundError: No module named 'apps.trials.services.trial_journey'

[ ] Step 3: Write minimal implementation

```python # apps/trials/services/trial_journey.py from dataclasses import dataclass from datetime import timedelta

from django.contrib.auth import get_user_model from django.utils import timezone

from apps.emails.models import EmailDispatch from apps.trials.models import UserTrial from routes.download import DOWNLOAD_PAGE_URL

User = get_user_model()

PATTERN_SCAN_DELAY_DAYS = 5 TRIAL_EMAIL_SOURCE = “trial”

@dataclass(frozen=True) class JourneyEmail: key: str template_name: str subject: str percent: int | None = None

Ordem crescente de percentual — resolve_pending_emails depende disso para

# devolver os marcos devidos na ordem da narrativa da régua. CONTENT_EMAILS = ( JourneyEmail(“daily_audio”, “emails/daily_audio.html”, “Áudio diário | Onion”, 10), JourneyEmail(“normose”, “emails/normose.html”, “Conheça a normose”, 30), JourneyEmail(“habits”, “emails/habits.html”, “Micro-hábitos”, 50), JourneyEmail(“chat_ai”, “emails/chat_ai.html”, “Conheça a Nathalia”, 70), )

CONVERSION_EMAIL = JourneyEmail(“plan_ended”, “emails/plan_ended.html”, “Sua jornada continua”) PATTERN_SCAN_EMAIL = JourneyEmail(“patern_scan”, “emails/patern_scan.html”, “Varredura de Padrões”)

JOURNEY_EMAILS = { email.key: email for email in (*CONTENT_EMAILS, CONVERSION_EMAIL, PATTERN_SCAN_EMAIL) }

CHECKOUT_CTA_KEYS = frozenset({CONVERSION_EMAIL.key, PATTERN_SCAN_EMAIL.key})

def cta_url_for(user_trial, email_key): “"”Conversão e pós-trial vão para o checkout; conteúdo vai para o app.

Os quatro emails de conteúdo apontam para DOWNLOAD_PAGE_URL como placeholder até
o time mobile definir os links de cada conteúdo — trocar aqui não exige tocar em template.
"""
if email_key in CHECKOUT_CTA_KEYS:
    return user_trial.trial.checkout_url
return DOWNLOAD_PAGE_URL

def consumed_percent(user_trial, now): “"”Percentual do período de trial já consumido, ou None quando o período é degenerado.””” total_seconds = (user_trial.expires_at - user_trial.started_at).total_seconds() if total_seconds <= 0: return None return (now - user_trial.started_at).total_seconds() / total_seconds * 100

def resolve_pending_emails(user_trial, now=None): “"”Devolve todos os emails da régua devidos hoje, na ordem, ainda não enviados.

Devolve uma lista e não um único item porque em trial curto o percentual anda muito
por dia e mais de um marco fica devido ao mesmo tempo — todos precisam sair, senão
a régua perde conteúdo. Em trial de 7 dias ou mais isso praticamente não ocorre.

As comparações de data usam <= e não == para que uma execução perdida do job seja
recuperada no dia seguinte em vez de perder a janela para sempre.
"""
now = now or timezone.now()
today = timezone.localdate(now)
expires_on = timezone.localdate(user_trial.expires_at)
sent = set(
    EmailDispatch.objects.filter(
        user=user_trial.user, source=TRIAL_EMAIL_SOURCE
    ).values_list("email_key", flat=True)
)

pending = []

percent = consumed_percent(user_trial, now)
if percent is not None:
    pending.extend(
        email.key
        for email in CONTENT_EMAILS
        if email.key not in sent and percent >= email.percent
    )

if expires_on <= today and CONVERSION_EMAIL.key not in sent:
    pending.append(CONVERSION_EMAIL.key)

if expires_on <= today - timedelta(days=PATTERN_SCAN_DELAY_DAYS) and PATTERN_SCAN_EMAIL.key not in sent:
    pending.append(PATTERN_SCAN_EMAIL.key)

return pending

def eligible_user_trials(now=None): “"”UserTrials que ainda podem receber algum email da régua.

O corte por expires_at implementa "acabou a régua, não envia mais nada": passado o
D+5 nunca há o que enviar, então o registro sai da varredura de vez.

O filtro duplo de conversão é intencional. converted_at sozinho não basta: por R-012
ele só é preenchido no fluxo de webhook para usuário já existente. subscription_status
é a rede de segurança, porque user.activate() troca o status em toda conversão.
"""
now = now or timezone.now()
cutoff = timezone.localdate(now) - timedelta(days=PATTERN_SCAN_DELAY_DAYS)
return (
    UserTrial.objects.select_related("user", "trial")
    .filter(
        trial__active_comunication_email=True,
        converted_at__isnull=True,
        user__subscription_status__iexact=User.SUBSCRIPTION_STATUS_TRIAL,
        started_at__lte=now,
        expires_at__date__gte=cutoff,
    )
    .exclude(user__email="")
    .exclude(user__email__isnull=True)
) ```

python # apps/trials/services/__init__.py — acrescentar ao que já existe from .trial_journey import ( JOURNEY_EMAILS, consumed_percent, cta_url_for, eligible_user_trials, resolve_pending_emails, )

[ ] Step 4: Run to verify it passes

bash make run.test path=tests/trials/test_trial_journey.py

Expected: PASS

[ ] Step 5: Commit

bash git add apps/trials/services/ tests/trials/test_trial_journey.py git commit -m "feat: add trial journey resolution service"

Task 3: Camada de envio de email

Files: - Modify: apps/emails/send_emails.py - Modify: apps/emails/tasks.py - Test: tests/emails/test_trial_journey_email.py

Interfaces: - Consumes: send_email de apps.emails.smtp - Produces: - SendEmails.trial_journey(data) — espera email, subject, template_name, name, cta_url - send_trial_journey_email(data: dict) — task Celery na fila default

Independente das Tasks 1 e 2 — pode rodar em paralelo com a Task 1.

[ ] Step 1: Write the failing test

```python # tests/emails/test_trial_journey_email.py from unittest.mock import patch

from apps.emails.send_emails import SendEmails

PAYLOAD = { “email”: “aluno@example.com”, “name”: “Aluno Teste”, “subject”: “Conheça a normose”, “template_name”: “emails/normose.html”, “cta_url”: “https://download.example.com/”, }

def test_trial_journey_passes_subject_template_and_recipient(): with patch(“apps.emails.send_emails.send_email”) as mock_send: SendEmails.trial_journey(PAYLOAD)

kwargs = mock_send.call_args.kwargs
assert kwargs["to"] == "aluno@example.com"
assert kwargs["subject"] == "Conheça a normose"
assert kwargs["template_name"] == "emails/normose.html"

def test_trial_journey_context_carries_only_name_and_cta(): “"”Preço, prazo e data ficam no checkout — o contexto não pode carregá-los.””” with patch(“apps.emails.send_emails.send_email”) as mock_send: SendEmails.trial_journey(PAYLOAD)

context = mock_send.call_args.kwargs["context"]
assert context == {"name": "Aluno Teste", "cta_url": "https://download.example.com/"}

def test_trial_journey_renders_every_journey_template(): “"”Garante que os seis templates da régua renderizam com o contexto do job.””” from apps.trials.services.trial_journey import JOURNEY_EMAILS

for email in JOURNEY_EMAILS.values():
    with patch("apps.emails.send_emails.send_email") as mock_send:
        SendEmails.trial_journey({**PAYLOAD, "template_name": email.template_name, "subject": email.subject})
    assert mock_send.call_args.kwargs["template_name"] == email.template_name ```

[ ] Step 2: Run to verify it fails

bash make run.test path=tests/emails/test_trial_journey_email.py

Expected: FAIL — AttributeError: type object 'SendEmails' has no attribute 'trial_journey'

[ ] Step 3: Write minimal implementation

```python # apps/emails/send_emails.py — acrescentar à classe SendEmails @staticmethod def trial_journey(data) -> None: “"”Envia um email da régua de trial.

    Um único método parametrizado por template, em vez de um por email da régua.
    O contexto é deliberadamente mínimo: nenhum template da régua pede preço,
    prazo ou data, e o job não os passa.
    """
    email_config = {
        'subject': data.get('subject'),
        'from_email': f"Onion <naoresponder@{DEFAULT_DOMAIN_EMAIL}>",
        'to': data.get('email'),
        'template_name': data.get('template_name'),
        'context': {
            'name': data.get('name'),
            'cta_url': data.get('cta_url'),
        },
    }
    return send_email(**email_config) ```

python # apps/emails/tasks.py — acrescentar ao final @app.task(base=BaseTaskWithRetry, queue=resolve_queue("default")) def send_trial_journey_email(data: dict): return SendEmails.trial_journey(data)

Criar tests/emails/__init__.py vazio se o diretório ainda não existir.

[ ] Step 4: Run to verify it passes

bash make run.test path=tests/emails/test_trial_journey_email.py

Expected: PASS

[ ] Step 5: Commit

bash git add apps/emails/ tests/emails/ git commit -m "feat: add trial journey email sender"

Task 4: Task orquestradora com reserva e fan-out

Files: - Create: apps/trials/tasks.py - Test: tests/trials/test_tasks.py

Interfaces: - Consumes: eligible_user_trials, resolve_pending_emails, cta_url_for, JOURNEY_EMAILS, TRIAL_EMAIL_SOURCE (Task 2); EmailDispatch (Task 1); send_trial_journey_email (Task 3) - Produces: send_trial_journey_emails() — task Celery, retorna {"status": ..., "sent": {email_key: count}}

[ ] Step 1: Write the failing test

```python # tests/trials/test_tasks.py from datetime import timedelta from unittest.mock import patch

import pytest from django.test import override_settings from django.utils import timezone

from apps.emails.models import EmailDispatch from apps.trials.models import Trial, UserTrial from apps.trials.tasks import send_trial_journey_emails

def make_trial(communication=True): return Trial.objects.create( name=”Trial 7 dias”, status=Trial.Status.ACTIVE, duration_days=7, start_date=timezone.now() - timedelta(days=1), end_date=timezone.now() + timedelta(days=60), checkout_url=”https://checkout.example.com/trial”, active_comunication_email=communication, )

def make_user_trial(user, trial, days_ago): started_at = timezone.now() - timedelta(days=days_ago) return UserTrial.objects.create( user=user, trial=trial, started_at=started_at, expires_at=started_at + timedelta(days=trial.duration_days), )

@override_settings(SEND_EMAIL=True) @pytest.mark.django_db def test_task_dispatches_one_email_per_eligible_user(trial_user): make_user_trial(trial_user, make_trial(), days_ago=1)

with patch("apps.trials.tasks.send_trial_journey_email") as mock_task:
    result = send_trial_journey_emails()

assert mock_task.delay.call_count == 1
assert result["sent"] == {"daily_audio": 1}

@override_settings(SEND_EMAIL=True) @pytest.mark.django_db def test_task_reserves_dispatch_before_sending(trial_user): make_user_trial(trial_user, make_trial(), days_ago=1)

with patch("apps.trials.tasks.send_trial_journey_email"):
    send_trial_journey_emails()

assert trial_user.email_dispatches.filter(email_key="daily_audio").exists()

@override_settings(SEND_EMAIL=True) @pytest.mark.django_db def test_task_does_not_resend_on_second_run_same_day(trial_user): make_user_trial(trial_user, make_trial(), days_ago=1)

with patch("apps.trials.tasks.send_trial_journey_email") as mock_task:
    send_trial_journey_emails()
    send_trial_journey_emails()

assert mock_task.delay.call_count == 1

@override_settings(SEND_EMAIL=True) @pytest.mark.django_db def test_task_payload_carries_subject_template_and_cta(trial_user): trial = make_trial() make_user_trial(trial_user, trial, days_ago=1)

with patch("apps.trials.tasks.send_trial_journey_email") as mock_task:
    send_trial_journey_emails()

payload = mock_task.delay.call_args.args[0]
assert payload["email"] == trial_user.email
assert payload["subject"] == "Áudio diário | Onion"
assert payload["template_name"] == "emails/daily_audio.html"
assert payload["cta_url"]

@override_settings(SEND_EMAIL=False) @pytest.mark.django_db def test_task_reserves_nothing_when_sending_is_disabled(trial_user): make_user_trial(trial_user, make_trial(), days_ago=1)

with patch("apps.trials.tasks.send_trial_journey_email") as mock_task:
    result = send_trial_journey_emails()

assert mock_task.delay.call_count == 0
assert EmailDispatch.objects.count() == 0
assert result["status"] == "skipped"

@override_settings(SEND_EMAIL=True) @pytest.mark.django_db def test_task_skips_user_without_pending_marker(trial_user): make_user_trial(trial_user, make_trial(), days_ago=0) # ~0% consumido

with patch("apps.trials.tasks.send_trial_journey_email") as mock_task:
    send_trial_journey_emails()

assert mock_task.delay.call_count == 0

@override_settings(SEND_EMAIL=True) @pytest.mark.django_db def test_task_sends_every_due_email_in_the_same_run(trial_user): “"”Trial de 7 dias com 71% consumido tem quatro marcos devidos de uma vez.””” make_user_trial(trial_user, make_trial(), days_ago=5)

with patch("apps.trials.tasks.send_trial_journey_email") as mock_task:
    result = send_trial_journey_emails()

assert mock_task.delay.call_count == 4
assert result["sent"] == {"daily_audio": 1, "normose": 1, "habits": 1, "chat_ai": 1}
assert EmailDispatch.objects.count() == 4 ```

[ ] Step 2: Run to verify it fails

bash make run.test path=tests/trials/test_tasks.py

Expected: FAIL — ModuleNotFoundError: No module named 'apps.trials.tasks'

[ ] Step 3: Write minimal implementation

```python # apps/trials/tasks.py from celery.utils.log import get_task_logger from django.conf import settings from django.utils import timezone

from apps.common.tasks import BaseTaskWithRetry from apps.emails.models import EmailDispatch from apps.emails.tasks import send_trial_journey_email from apps.trials.services.trial_journey import ( JOURNEY_EMAILS, TRIAL_EMAIL_SOURCE, cta_url_for, eligible_user_trials, resolve_pending_emails, ) from config.celery import app from config.queues import queue as resolve_queue

logger = get_task_logger(name)

@app.task(base=BaseTaskWithRetry, queue=resolve_queue(“default”)) def send_trial_journey_emails(): “"”Envia todos os emails da régua de trial devidos hoje.

A reserva do EmailDispatch acontece antes do fan-out, não depois do envio:
duas execuções próximas do job enfileirariam o mesmo email duas vezes antes de
qualquer linha existir. Reservando antes, a unique_together serializa antes do
envio e o pior caso vira um email não enviado em vez de um email duplicado.
"""
if not settings.SEND_EMAIL:
    return {"status": "skipped", "reason": "send_email_disabled", "sent": {}}

now = timezone.now()
sent = {}

for user_trial in eligible_user_trials(now):
    for email_key in resolve_pending_emails(user_trial, now):
        _, created = EmailDispatch.objects.get_or_create(
            user=user_trial.user,
            source=TRIAL_EMAIL_SOURCE,
            email_key=email_key,
            defaults={"sent_at": now},
        )
        if not created:
            continue

        journey_email = JOURNEY_EMAILS[email_key]
        send_trial_journey_email.delay({
            "email": user_trial.user.email,
            "name": user_trial.user.name,
            "subject": journey_email.subject,
            "template_name": journey_email.template_name,
            "cta_url": cta_url_for(user_trial, email_key),
        })
        sent[email_key] = sent.get(email_key, 0) + 1

logger.info("send_trial_journey_emails | sent=%s", sent)
return {"status": "done", "sent": sent} ```

[ ] Step 4: Run to verify it passes

bash make run.test path=tests/trials/test_tasks.py

Expected: PASS

[ ] Step 5: Commit

bash git add apps/trials/tasks.py tests/trials/test_tasks.py git commit -m "feat: add trial journey emails task"

Task 5: Agendamento no beat e documentação da regra

Files: - Modify: config/celery_defaults.py - Create: .project/docs/rules/trials/trial_journey_emails.md - Modify: .project/docs/README.md - Test: tests/trials/test_beat_schedule.py

Interfaces: - Consumes: apps.trials.tasks.send_trial_journey_emails (Task 4) - Produces: entrada send-trial-journey-emails em CELERY_BEAT_SCHEDULE

[ ] Step 1: Write the failing test

```python # tests/trials/test_beat_schedule.py from celery.schedules import crontab

from config.celery_defaults import CELERY_BEAT_SCHEDULE

def test_trial_journey_job_runs_daily_at_ten(): entry = CELERY_BEAT_SCHEDULE[“send-trial-journey-emails”]

assert entry["task"] == "apps.trials.tasks.send_trial_journey_emails"
assert entry["schedule"] == crontab(hour=10, minute=0) ```

[ ] Step 2: Run to verify it fails

bash make run.test path=tests/trials/test_beat_schedule.py

Expected: FAIL — KeyError: 'send-trial-journey-emails'

[ ] Step 3: Write minimal implementation

python # config/celery_defaults.py — acrescentar ao dict CELERY_BEAT_SCHEDULE 'send-trial-journey-emails': { 'task': 'apps.trials.tasks.send_trial_journey_emails', 'schedule': crontab(hour=10, minute=0), },

CELERY_TIMEZONE já é America/Sao_Paulo, então hour=10 é 10h de Brasília sem conversão.

Criar .project/docs/rules/trials/trial_journey_emails.md com frontmatter id, title, scope: trials, certainty: high, e o conteúdo em português seguindo o padrão dos demais arquivos de rules/trials/: a tabela da régua com os seis emails e seus marcos; a resolução por lista de pendentes (todos os marcos percentuais atingidos, mais plan_ended a partir do dia da expiração e patern_scan a partir do D+5); as três condições de parada (conversão confirmada pelo filtro duplo, régua completa, janela D+5 vencida); a garantia de que nenhum email da régua se perde, com a nota de que em trial curto mais de um sai no mesmo dia e isso é intencional; e a justificativa dos marcos 10/30/50/70 por distribuição. Referenciar a spec e os testes vinculados.

Acrescentar a linha correspondente ao índice em .project/docs/README.md.

[ ] Step 4: Run to verify it passes

bash make run.test path=tests/trials/test_beat_schedule.py make run.test

Expected: PASS — suíte completa verde

[ ] Step 5: Commit

bash git add config/celery_defaults.py tests/trials/test_beat_schedule.py .project/docs/ git commit -m "feat: schedule trial journey emails job"

Verificação final

Depois da Task 5, antes de abrir o PR:

make run.test — suíte completa verde.
python manage.py makemigrations --check --dry-run — sem migração pendente.
Em staging com SEND_EMAIL=True, rodar send_trial_journey_emails() no shell e confirmar no inbox que o email chega renderizado, com header, assunto e botão.
Rodar a task duas vezes seguidas e confirmar que o segundo envio não acontece.
Confirmar que a task aparece na fila onion-{env} no RabbitMQ e que o beat a agenda às 10h.

Antes de ligar em produção: nenhum dos seis templates jamais passou pelo SMTP — cta_url nunca foi passado por código algum. Enviar os seis manualmente em staging e revisar a renderização antes de considerar o job pronto.

Fora do escopo deste plano

Registrado na spec, não implementado aqui: correção do href="#" de chat_ai.html, remoção do órfão audio_diario.html, rotas reais dos CTAs de conteúdo e opt-out de email.

Job diário de emails da régua de trial — Plano de implementação

Restrições globais

Baseline

Ordem e paralelismo

Task 1: Model genérico de idempotência EmailDispatch

Task 2: Service da régua e resolução do email do dia

Ordem crescente de percentual — resolve_pending_emails depende disso para

Task 3: Camada de envio de email

Task 4: Task orquestradora com reserva e fan-out

Task 5: Agendamento no beat e documentação da regra

Verificação final

Fora do escopo deste plano

Task 1: Model genérico de idempotência `EmailDispatch`