diff --git a/docs/api.md b/docs/api.md index 58c3b62..f331db7 100644 --- a/docs/api.md +++ b/docs/api.md @@ -885,6 +885,235 @@ Sandbox deploys install this dispatcher as `datamailer-cmp-callbacks-worker`. Operators can inspect recent callback status, attempt counts, next retry time, delivery time, and the last error from the client detail page and Django admin. +## Client Callbacks + +When a callback endpoint is configured for a client, Relay announces delivery +and engagement transitions for that client with signed, versioned, redacted +JSON events. This channel is additive to the CMP callback contract above. It +carries transactional deliveries and client-level subscription changes for +unification-plan clients; campaign-recipient transitions stay on the CMP +channel. Payloads carry identifiers and safe reason codes only, so a consumer +can log and store them as-is. + +### Configuration + +One endpoint per client, managed by operators on the client record: + +| Field | Meaning | +|---|---| +| URL | The one HTTPS endpoint Relay posts to. Per-message URLs and redirects to other origins are not permitted. | +| Signing secret | Dedicated HMAC signing secret, separate from every client API key. Write-only: it is shown once at provisioning and never returned afterwards. | +| Previous signing secret | Retained during a rotation for a bounded verification overlap; the receiver accepts either secret until the overlap ends. | +| Contract version | Integer version of the event payload, currently `1`. | +| Enabled | Disablement stops all deliveries immediately; in-flight pending callbacks fail terminally with `endpoint_disabled`. | + +Real secret provisioning is infrastructure work and uses approved +server-side secret handling; secrets never appear in URLs, logs, errors, or +docs examples. + +### Event types and reason codes + +Relay posts these event types after the corresponding transport state commits: + +| Event type | Fired on | Reason codes | +|---|---|---| +| `delivery.accepted` | message queued, then handed to the provider | none | +| `delivery.delivered` | provider confirmed delivery | none | +| `delivery.bounced` | provider reported a bounce | `hard_bounce`, `soft_bounce` | +| `delivery.complained` | recipient filed a complaint | `complaint` | +| `delivery.suppressed` | send was suppressed before delivery | `unverified`, `invalid_email`, `global_unsubscribe`, `client_unsubscribe`, `audience_unsubscribe`, `hard_bounce`, `complaint`, `duplicate`, `category_unsubscribe`, `missing_category_scope`, `suppressed` | +| `engagement.opened` | tracked open | none | +| `engagement.clicked` | tracked click | none | +| `subscription.changed` | subscribe or unsubscribe state changed | `subscribed`, `unsubscribed` | + +`reason_code` is omitted when the table shows "none". A permanently failed +send is not announced on this channel; the synchronous send response and the +reconciliation endpoint below carry that outcome. + +### Payload + +The body is the canonical JSON serialization of the event (sorted keys, no +whitespace), signed exactly as sent on every attempt: + +```json +{ + "bounce_type": "hard", + "client_reference": "registration-user-123", + "contract_version": 1, + "event_id": "dd8b8f17-6d55-5094-ae45-83cd8ac37b96", + "event_type": "delivery.bounced", + "message_id": "1042", + "reason_code": "hard_bounce", + "sequence": 3, + "template_key": "registration-welcome", + "timestamp": "2026-09-08T12:00:00+00:00" +} +``` + +| Field | Meaning | +|---|---| +| `contract_version` | Payload contract version. Additive changes do not bump it; removing or repurposing a field does. | +| `event_id` | Stable UUID for one transition. Duplicate deliveries of the same event reuse the id, so receivers can deduplicate on it. | +| `event_type` | One of the types above. | +| `timestamp` | When the transition committed (ISO 8601). | +| `sequence` | 1-based counter per message. Ordering of deliveries is best-effort (retries reorder); use `sequence` to reorder if you need strict ordering. | +| `message_id` | Relay transactional message id as a string; `null` for contact-level transitions such as subscription changes. | +| `client_reference` | Your idempotency key from the original send; `null` when the transition has no transactional message. | +| `template_key` | Template key for transactional messages, empty otherwise. | +| `bounce_type` | `hard` or `soft`; present only on `delivery.bounced`. | +| `reason_code` | Safe reason code from the table above, when applicable. | + +The payload contains nothing else. No recipient address, subject, body, +template context, custom headers, credentials, raw provider payload, or +arbitrary metadata is ever included. + +### Signature headers + +Every request carries: + +| Header | Meaning | +|---|---| +| `X-Relay-Timestamp` | Unix timestamp of the attempt | +| `X-Relay-Signature` | `sha256=` + HMAC-SHA256 over `.` with the callback signing secret | +| `X-Relay-Event-Id` | the `event_id`, for quick routing and dedup | +| `X-Relay-Event-Type` | the `event_type` | +| `X-Relay-Contract-Version` | the `contract_version` | +| `X-Relay-Attempt` | 1-based attempt counter | + +There is no `Authorization` header and no Bearer credential. The signature +scheme is identical to webhook task deliveries. + +A deterministic verification fixture lives at +`tests/fixtures/client_callback_contract_v1.json`: it contains a canonical +body, a signing secret, a timestamp, and the expected signature, plus a +rotated-secret signature. Consumer contract tests should verify the fixture +instead of inventing their own vectors. + +### Reference receiver + +The receiver verifies origin and freshness with the callback signing secret. +The replay window is part of the receiver's contract; 5 minutes is a workable +default. During a rotation, verify against the current secret and then the +previous one. + +```python +import hashlib +import hmac +import time + +REPLAY_WINDOW_SECONDS = 300 + + +class Reject(Exception): + pass + + +def verify_relay_callback(body: bytes, headers: dict, secrets: list[str], *, now=None) -> None: + """Verify X-Relay-* headers on a client callback delivery. Raises Reject.""" + now = now or time.time() + timestamp = headers.get("X-Relay-Timestamp", "") + signature = headers.get("X-Relay-Signature", "") + try: + stamp = int(timestamp) + except ValueError: + raise Reject("missing or malformed timestamp") from None + if abs(now - stamp) > REPLAY_WINDOW_SECONDS: + raise Reject("timestamp outside replay window") + expected = [ + "sha256=" + + hmac.new(secret.encode(), timestamp.encode() + b"." + body, hashlib.sha256).hexdigest() + for secret in secrets + ] + if not any(hmac.compare_digest(candidate, signature) for candidate in expected): + raise Reject("bad signature") +``` + +### Delivery, retries, and deduplication + +Callback rows are created in the same transaction as the transport state they +announce and dispatched only after that transaction commits, so a committed +transition is never silently missing its callback. A duplicate transition +never creates a second row: work is deduplicated on client plus `event_id`. + +Delivery responds to a receiver that answers `2xx`. Duplicate successful +acknowledgements are treated as success. Relay retries: + +| Outcome | Class | +|---|---| +| Connection error or timeout | retry with bounded exponential backoff | +| HTTP `429` | retry with bounded exponential backoff | +| HTTP `5xx` | retry with bounded exponential backoff | +| Any other `4xx` | fail terminally | +| Redirect (`3xx`) | fail terminally; redirects are never followed | +| Endpoint disabled | fail terminally as `endpoint_disabled` | +| Retry exhaustion | fail terminally | + +Backoff starts at 60 seconds and doubles per attempt, capped at 6 hours, with +a small deterministic jitter. Relay records the attempt count, the response +status class (`2xx`/`3xx`/`4xx`/`5xx`), a safe error code, and the next +attempt time. Response bodies and headers are never stored. + +Callback failure never regresses Relay transport state: a delivered message +stays delivered whether or not the callback succeeded. If callbacks are +delayed or lost, reconcile against `GET /api/transactional/messages?since=` +(described next). + +Run the dispatcher with: + +```bash +python manage.py process_client_callbacks --batch-size 25 +``` + +Sandbox deploys install this dispatcher as +`datamailer-client-callbacks-worker`. Operators can inspect recent callback +status, attempt counts, next retry time, delivery time, and the last safe +error from the client detail page and Django admin. + +## Transactional Message Reconciliation + +`GET /api/transactional/messages?since=` returns every transactional +message of the authenticated client whose `updated_at` is at or after +`since`, oldest first, capped at 1000 rows. It is the pull-based complement +to client callbacks and the authoritative view for recovery after missed or +failed callback deliveries. + +Authentication and errors are the same as the rest of the client API: Bearer +client API key; a missing or unparseable `since` returns `validation_error`. + +```text +GET /api/transactional/messages?since=2026-09-08T00:00:00%2B00:00 +``` + +Response: + +```json +{ + "messages": [ + { + "id": "1042", + "client_reference": "registration-user-123", + "status": "bounced", + "template_key": "registration-welcome", + "template_version": 1, + "reason_code": "hard_bounce", + "updated_at": "2026-09-08T12:00:00+00:00" + } + ] +} +``` + +| Field | Meaning | +|---|---| +| `id` | Transactional message id as a string. | +| `client_reference` | The idempotency key from the original send. | +| `status` | `queued`, `retrying` (in flight to the provider), `sent` (accepted by the provider), `delivered`, `suppressed`, `failed`, `bounced` (hard bounce), or `complained`. | +| `template_key` | Template key the message was sent with. | +| `template_version` | Published template version the message was sent with; `1` for messages sent before versions were recorded. | +| `reason_code` | Safe reason code, empty when there is nothing to report; provider diagnostics are never included. | +| `updated_at` | ISO 8601; use it as the next poll's `since`. | + +Messages without an idempotency key are never reported. + ## Mailchimp Sync Datamailer can push contacts into a client's Mailchimp audience with a tag when diff --git a/docs/infra-deploy.md b/docs/infra-deploy.md index c9eb11c..c2ca628 100644 --- a/docs/infra-deploy.md +++ b/docs/infra-deploy.md @@ -22,6 +22,7 @@ For this intermediate step, the sandbox deploy installs these systemd units on t /opt/datamailer/.venv/bin/python manage.py process_sqs_worker campaign --batch-size 10 --wait-time 20 /opt/datamailer/.venv/bin/python manage.py process_sqs_worker ses-webhooks --batch-size 10 --wait-time 20 /opt/datamailer/.venv/bin/python manage.py process_cmp_callbacks --batch-size 25 --idle-sleep 5 +/opt/datamailer/.venv/bin/python manage.py process_client_callbacks --batch-size 25 --idle-sleep 5 ``` The commands long-poll their SQS queues, call the same handlers used by the future Lambda workers, delete only successfully processed records, and leave failed records for SQS retry/DLQ behavior. This is intentionally a sandbox bridge, not the final production architecture. diff --git a/docs/operations.md b/docs/operations.md index 88f2b25..48054f5 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -49,18 +49,21 @@ python manage.py db_worker python manage.py drain_sqs_ingress ses-webhooks --batch-size 10 --wait-time 20 python manage.py drain_sqs_ingress inbound-email --batch-size 10 --wait-time 20 python manage.py process_cmp_callbacks --batch-size 25 --idle-sleep 5 +python manage.py process_client_callbacks --batch-size 25 --idle-sleep 5 ``` The SQS workers use the same message contracts and handlers as Lambda. The CMP callback dispatcher reads the local `cmp_callbacks` outbox and retries failed -HTTP callbacks with backoff. Once the sandbox uses shared Postgres/RDS, replace +HTTP callbacks with backoff. The client callback dispatcher does the same for +the `client_callbacks` outbox used by generic `CallbackEndpoint` consumers. +Once the sandbox uses shared Postgres/RDS, replace the EC2 SQS worker services with SQS event-source Lambda workers; keep one scheduled or long-running callback dispatcher for the outbox. Staff operators can inspect the same worker state in the dashboard or as JSON at `/api/workers/status`. The endpoint reports systemd state where available plus local backlog counts for transactional messages, campaign recipients, and due -CMP callbacks. +CMP and client callbacks. Sandbox deploys must also provision the CMP client scope before configuring senders: diff --git a/mailing/admin.py b/mailing/admin.py index 9da43b4..19fd1eb 100644 --- a/mailing/admin.py +++ b/mailing/admin.py @@ -1,11 +1,14 @@ +from django import forms from django.contrib import admin from mailing.models import ( Audience, + CallbackEndpoint, Campaign, CampaignRecipient, Client, ClientApiKey, + ClientCallback, CmpCallback, Contact, ContactTag, @@ -269,6 +272,31 @@ class EmailEventAdmin(admin.ModelAdmin): autocomplete_fields = ("campaign", "campaign_recipient", "transactional_message", "contact", "client", "audience") +class CallbackEndpointForm(forms.ModelForm): + """Keeps the signing secret write-only: blank keeps the current secret.""" + + class Meta: + model = CallbackEndpoint + fields = "__all__" + widgets = {"signing_secret": forms.PasswordInput(render_value=False)} + + def clean_signing_secret(self): + new_secret = self.cleaned_data.get("signing_secret", "") + if not new_secret and self.instance and self.instance.pk: + return self.instance.signing_secret + return new_secret + + +@admin.register(CallbackEndpoint) +class CallbackEndpointAdmin(admin.ModelAdmin): + form = CallbackEndpointForm + readonly_fields = ("created_at", "updated_at", "secret_rotated_at", "disabled_at") + list_display = ("client", "enabled", "url", "contract_version", "secret_rotated_at", "updated_at") + list_filter = ("enabled",) + search_fields = ("client__name", "client__slug", "url") + autocomplete_fields = ("client",) + + @admin.register(CmpCallback) class CmpCallbackAdmin(admin.ModelAdmin): readonly_fields = ( @@ -300,6 +328,35 @@ class CmpCallbackAdmin(admin.ModelAdmin): autocomplete_fields = ("email_event", "contact", "client", "audience") +@admin.register(ClientCallback) +class ClientCallbackAdmin(CreatedAtReadOnlyMixin, admin.ModelAdmin): + readonly_fields = ( + "updated_at", + "last_attempt_at", + "delivered_at", + "body_hash", + ) + list_display = ( + "event_type", + "status", + "client", + "attempt_count", + "next_attempt_at", + "delivered_at", + "created_at", + ) + list_filter = ("status", "event_type", "client") + search_fields = ( + "event_id", + "event_type", + "client_reference", + "client__name", + "client__slug", + "last_error", + ) + autocomplete_fields = ("email_event", "transactional_message", "campaign_recipient", "client", "endpoint") + + @admin.register(MailchimpTagMapping) class MailchimpTagMappingAdmin(admin.ModelAdmin): readonly_fields = ("created_at", "updated_at") diff --git a/mailing/management/commands/process_client_callbacks.py b/mailing/management/commands/process_client_callbacks.py new file mode 100644 index 0000000..d5eafcb --- /dev/null +++ b/mailing/management/commands/process_client_callbacks.py @@ -0,0 +1,43 @@ +from time import sleep + +from django.core.management.base import BaseCommand + +from mailing.services.client_callbacks import process_due_client_callbacks + + +class Command(BaseCommand): + help = "Dispatch due client callbacks from the tenant-scoped HMAC outbox." + + def add_arguments(self, parser): + parser.add_argument( + "--once", + action="store_true", + help="Process one batch and exit.", + ) + parser.add_argument( + "--batch-size", + type=int, + default=25, + help="Maximum callback rows to process per batch.", + ) + parser.add_argument( + "--idle-sleep", + type=float, + default=5.0, + help="Seconds to sleep between empty batches in continuous mode.", + ) + + def handle(self, *args, **options): + batch_size = options["batch_size"] + if batch_size < 1: + self.stderr.write("batch-size must be at least 1.") + return + + self.stdout.write("Starting client callback dispatcher") + while True: + result = process_due_client_callbacks(limit=batch_size) + self.stdout.write("processed={processed} delivered={delivered} failed={failed}".format(**result)) + if options["once"]: + return + if result["processed"] == 0: + sleep(options["idle_sleep"]) diff --git a/mailing/migrations/0027_callbackendpoint_clientcallback_and_more.py b/mailing/migrations/0027_callbackendpoint_clientcallback_and_more.py new file mode 100644 index 0000000..0bd8409 --- /dev/null +++ b/mailing/migrations/0027_callbackendpoint_clientcallback_and_more.py @@ -0,0 +1,113 @@ +# Generated by Django 6.0.5 on 2026-09-07 23:09 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('mailing', '0025_client_relay_webhook_fields'), + ('mailing', '0026_emailtemplate_category_emailtemplate_markdown_body_and_more'), + ] + + operations = [ + migrations.CreateModel( + name='CallbackEndpoint', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('url', models.URLField(max_length=2048)), + ('signing_secret', models.CharField(max_length=255)), + ('previous_signing_secret', models.CharField(blank=True, max_length=255)), + ('secret_rotated_at', models.DateTimeField(blank=True, null=True)), + ('contract_version', models.PositiveIntegerField(default=1)), + ('enabled', models.BooleanField(default=True)), + ('disabled_at', models.DateTimeField(blank=True, null=True)), + ('disabled_reason', models.CharField(blank=True, max_length=255)), + ], + options={ + 'db_table': 'callback_endpoints', + 'ordering': ['client__organization__slug', 'client__slug'], + }, + ), + migrations.CreateModel( + name='ClientCallback', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('event_id', models.UUIDField(editable=False, unique=True)), + ('event_type', models.CharField(max_length=80)), + ('contract_version', models.PositiveIntegerField(default=1)), + ('client_reference', models.CharField(blank=True, max_length=255)), + ('message_kind', models.CharField(blank=True, max_length=20)), + ('message_ref', models.CharField(blank=True, max_length=64)), + ('template_key', models.CharField(blank=True, max_length=120)), + ('sequence', models.PositiveIntegerField(default=0)), + ('payload', models.JSONField(default=dict)), + ('body', models.TextField()), + ('body_hash', models.CharField(max_length=64)), + ('status', models.CharField(choices=[('pending', 'Pending'), ('delivered', 'Delivered'), ('failed', 'Failed')], default='pending', max_length=20)), + ('attempt_count', models.PositiveIntegerField(default=0)), + ('max_attempts', models.PositiveIntegerField(default=8)), + ('next_attempt_at', models.DateTimeField(db_index=True)), + ('last_attempt_at', models.DateTimeField(blank=True, null=True)), + ('delivered_at', models.DateTimeField(blank=True, null=True)), + ('response_status_class', models.CharField(blank=True, max_length=8)), + ('last_error_code', models.CharField(blank=True, max_length=64)), + ('last_error', models.CharField(blank=True, max_length=500)), + ], + options={ + 'db_table': 'client_callbacks', + 'ordering': ['id'], + }, + ), + migrations.AddField( + model_name='callbackendpoint', + name='client', + field=models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='callback_endpoint', to='mailing.client'), + ), + migrations.AddField( + model_name='clientcallback', + name='campaign_recipient', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='client_callbacks', to='mailing.campaignrecipient'), + ), + migrations.AddField( + model_name='clientcallback', + name='client', + field=models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='client_callbacks', to='mailing.client'), + ), + migrations.AddField( + model_name='clientcallback', + name='email_event', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='client_callbacks', to='mailing.emailevent'), + ), + migrations.AddField( + model_name='clientcallback', + name='endpoint', + field=models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='callbacks', to='mailing.callbackendpoint'), + ), + migrations.AddField( + model_name='clientcallback', + name='transactional_message', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='client_callbacks', to='mailing.transactionalmessage'), + ), + migrations.AddIndex( + model_name='clientcallback', + index=models.Index(fields=['status', 'next_attempt_at'], name='client_cb_status_next_idx'), + ), + migrations.AddIndex( + model_name='clientcallback', + index=models.Index(fields=['client', 'status', 'created_at'], name='client_cb_cli_status_idx'), + ), + migrations.AddIndex( + model_name='clientcallback', + index=models.Index(fields=['client', 'message_kind', 'message_ref', 'sequence'], name='client_cb_msg_seq_idx'), + ), + migrations.AddConstraint( + model_name='clientcallback', + constraint=models.UniqueConstraint(fields=('client', 'event_id'), name='unique_client_callback_event'), + ), + ] diff --git a/mailing/models.py b/mailing/models.py index bca0eca..b0e7342 100644 --- a/mailing/models.py +++ b/mailing/models.py @@ -737,6 +737,136 @@ def __str__(self): return f"{self.event_type} at {self.created_at}" +class CallbackEndpoint(TimeStampedModel): + """Tenant-scoped callback destination and signing secret. + + One endpoint per client. The signing secret is a dedicated callback + credential, separate from every client API key, and is write-only once + provisioned: it is never serialized by any API or operator view. Rotation + keeps the previous secret for a bounded verification overlap so a receiver + can verify either secret while the switch propagates. + """ + + client = models.OneToOneField(Client, on_delete=models.CASCADE, related_name="callback_endpoint") + url = models.URLField(max_length=2048) + signing_secret = models.CharField(max_length=255) + previous_signing_secret = models.CharField(max_length=255, blank=True) + secret_rotated_at = models.DateTimeField(null=True, blank=True) + contract_version = models.PositiveIntegerField(default=1) + enabled = models.BooleanField(default=True) + disabled_at = models.DateTimeField(null=True, blank=True) + disabled_reason = models.CharField(max_length=255, blank=True) + + class Meta: + db_table = "callback_endpoints" + ordering = ["client__organization__slug", "client__slug"] + + def rotate_secret(self, new_secret, *, rotated_at): + self.previous_signing_secret = self.signing_secret + self.signing_secret = new_secret + self.secret_rotated_at = rotated_at + self.save(update_fields=["previous_signing_secret", "signing_secret", "secret_rotated_at", "updated_at"]) + + def disable(self, reason, *, disabled_at): + self.enabled = False + self.disabled_at = disabled_at + self.disabled_reason = reason[:255] + self.save(update_fields=["enabled", "disabled_at", "disabled_reason", "updated_at"]) + + def enable(self): + self.enabled = True + self.disabled_at = None + self.disabled_reason = "" + self.save(update_fields=["enabled", "disabled_at", "disabled_reason", "updated_at"]) + + def __str__(self): + state = "enabled" if self.enabled else "disabled" + return f"{self.client.slug} callback endpoint ({state})" + + +class ClientCallbackStatus(models.TextChoices): + PENDING = "pending", "Pending" + DELIVERED = "delivered", "Delivered" + FAILED = "failed", "Failed" + + +class ClientCallback(TimeStampedModel): + """Outbox row for one tenant-scoped, versioned, redacted callback event. + + Created in the same transaction as the transport transition it announces + and dispatched only after that transaction commits. The exact canonical + body and its hash are persisted, so every attempt signs and sends the same + logical event. Rows carry no recipient, body, context, credential, or raw + provider data: redaction is enforced at creation, not at delivery. + """ + + client = models.ForeignKey(Client, on_delete=models.PROTECT, related_name="client_callbacks") + endpoint = models.ForeignKey(CallbackEndpoint, on_delete=models.PROTECT, related_name="callbacks") + email_event = models.ForeignKey( + EmailEvent, + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="client_callbacks", + ) + transactional_message = models.ForeignKey( + TransactionalMessage, + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="client_callbacks", + ) + campaign_recipient = models.ForeignKey( + CampaignRecipient, + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="client_callbacks", + ) + event_id = models.UUIDField(unique=True, editable=False) + event_type = models.CharField(max_length=80) + contract_version = models.PositiveIntegerField(default=1) + client_reference = models.CharField(max_length=255, blank=True) + message_kind = models.CharField(max_length=20, blank=True) + message_ref = models.CharField(max_length=64, blank=True) + template_key = models.CharField(max_length=120, blank=True) + sequence = models.PositiveIntegerField(default=0) + payload = models.JSONField(default=dict) + body = models.TextField() + body_hash = models.CharField(max_length=64) + status = models.CharField( + max_length=20, + choices=ClientCallbackStatus.choices, + default=ClientCallbackStatus.PENDING, + ) + attempt_count = models.PositiveIntegerField(default=0) + max_attempts = models.PositiveIntegerField(default=8) + next_attempt_at = models.DateTimeField(db_index=True) + last_attempt_at = models.DateTimeField(null=True, blank=True) + delivered_at = models.DateTimeField(null=True, blank=True) + response_status_class = models.CharField(max_length=8, blank=True) + last_error_code = models.CharField(max_length=64, blank=True) + last_error = models.CharField(max_length=500, blank=True) + + class Meta: + db_table = "client_callbacks" + ordering = ["id"] + constraints = [ + models.UniqueConstraint(fields=["client", "event_id"], name="unique_client_callback_event"), + ] + indexes = [ + models.Index(fields=["status", "next_attempt_at"], name="client_cb_status_next_idx"), + models.Index(fields=["client", "status", "created_at"], name="client_cb_cli_status_idx"), + models.Index( + fields=["client", "message_kind", "message_ref", "sequence"], + name="client_cb_msg_seq_idx", + ), + ] + + def __str__(self): + return f"{self.event_type} {self.event_id} ({self.status})" + + class CmpCallbackStatus(models.TextChoices): PENDING = "pending", "Pending" DELIVERED = "delivered", "Delivered" diff --git a/mailing/services/api.py b/mailing/services/api.py index 33f7635..7ad1589 100644 --- a/mailing/services/api.py +++ b/mailing/services/api.py @@ -27,6 +27,7 @@ Subscription, SubscriptionStatus, TransactionalMessage, + TransactionalMessageStatus, normalize_tag_filter, ) from mailing.models import ( @@ -36,6 +37,7 @@ from mailing.services.campaign_sender import render_campaign_message, send_campaign_test_message from mailing.services.campaigns import queue_campaign from mailing.services.categories import TRANSACTIONAL_CATEGORY, category_label, validate_canonical_category +from mailing.services.client_callbacks import SUPPRESSION_REASON_CODES, emit_client_callback from mailing.services.cmp_callbacks import emit_cmp_contact_event from mailing.services.contacts import ( assign_tag, @@ -1106,6 +1108,7 @@ def upsert_contact_for_client(data, authenticated_client): metadata={"source": "api"}, ) emit_cmp_contact_event(event) + emit_client_callback(event) for tag_name in tags: assign_tag(contact, scope.audience, tag_name) @@ -1427,6 +1430,7 @@ def update_contact_suppression_for_client(contact_id, data, authenticated_client metadata={"source": "api", "reason": reason}, ) emit_cmp_contact_event(event) + emit_client_callback(event) elif field == "global_unsubscribed_at": event = EmailEvent.objects.create( contact=contact, @@ -1436,6 +1440,7 @@ def update_contact_suppression_for_client(contact_id, data, authenticated_client metadata={"source": "api", "reason": reason}, ) emit_cmp_contact_event(event) + emit_client_callback(event) return contact_payload(contact, scope.audience, scope.client, requested_email=contact.email) @@ -1563,6 +1568,81 @@ def get_transactional_message_status_for_client(message_id, authenticated_client } +RECONCILIATION_MESSAGE_LIMIT = 1000 + +# TransactionalMessage.status mapped onto the reconciliation vocabulary the +# client packages understand. "sending" is in flight, so it is reported as +# "retrying"; a sent message that already has a delivery timestamp is +# reported as "delivered". +RECONCILIATION_STATUS_BY_MESSAGE_STATUS = { + TransactionalMessageStatus.QUEUED: "queued", + TransactionalMessageStatus.SENDING: "retrying", + TransactionalMessageStatus.SENT: "sent", + TransactionalMessageStatus.FAILED: "failed", + TransactionalMessageStatus.SKIPPED: "suppressed", + TransactionalMessageStatus.BOUNCED: "bounced", + TransactionalMessageStatus.COMPLAINED: "complained", +} + + +def validate_reconciliation_since(value): + if not isinstance(value, str) or not value.strip(): + raise ApiValidationError({"since": "required"}) + parsed = parse_datetime(value.strip()) + if parsed is None: + raise ApiValidationError({"since": "must_be_iso_datetime"}) + if timezone.is_naive(parsed): + parsed = timezone.make_aware(parsed, timezone=timezone.get_current_timezone()) + return parsed + + +def reconciliation_reason_code(message): + """A safe, pattern-checked reason code; never raw provider diagnostics.""" + metadata = message.metadata or {} + if message.status == TransactionalMessageStatus.SKIPPED: + reason = str(metadata.get("reason", "")) + return reason if reason in SUPPRESSION_REASON_CODES else "suppressed" + if message.status == TransactionalMessageStatus.BOUNCED: + return "hard_bounce" + if message.status == TransactionalMessageStatus.COMPLAINED: + return "complaint" + if message.status == TransactionalMessageStatus.FAILED: + return "send_failed" + if message.status == TransactionalMessageStatus.SENT and (message.last_error or "") == "soft_bounce": + return "soft_bounce" + return "" + + +def reconciliation_message_item(message): + status = RECONCILIATION_STATUS_BY_MESSAGE_STATUS[message.status] + if message.status == TransactionalMessageStatus.SENT and message.delivered_at: + status = "delivered" + # Real template versions exist since R1.3; messages sent before a version + # was recorded report 1, matching the send contract's fallback. + return { + "id": str(message.id), + "client_reference": message.idempotency_key, + "status": status, + "template_key": message.template_key, + "template_version": message.template_version if message.template_version is not None else 1, + "reason_code": reconciliation_reason_code(message), + "updated_at": isoformat(message.updated_at), + } + + +def get_transactional_messages_since_for_client(raw_since, authenticated_client): + since = validate_reconciliation_since(raw_since) + messages = ( + TransactionalMessage.objects.filter( + client=authenticated_client, + idempotency_key__gt="", + updated_at__gte=since, + ) + .order_by("updated_at", "id")[:RECONCILIATION_MESSAGE_LIMIT] + ) + return {"messages": [reconciliation_message_item(message) for message in messages]} + + def get_contact_history_for_client(contact_id, data, authenticated_client): contact, scope = validate_existing_contact_scope(contact_id, data, authenticated_client) limit = validate_history_limit(data.get("limit")) diff --git a/mailing/services/client_callbacks.py b/mailing/services/client_callbacks.py new file mode 100644 index 0000000..da66ccf --- /dev/null +++ b/mailing/services/client_callbacks.py @@ -0,0 +1,449 @@ +"""Generic tenant-scoped client callbacks for delivery and engagement events. + +Every client-visible transport transition appends an ``EmailEvent``; this +module turns exactly those transitions into durable, versioned, redacted +callback work. One ``ClientCallback`` outbox row is created inside the same +transaction that commits the transition, and a polling dispatcher signs and +posts it after commit. The row is deduplicated on ``(client, event_id)``, so +repeated transition processing or duplicate provider events never create +duplicate callback work. + +The wire contract is documented in ``docs/api.md``: + +- the body is the canonical JSON event with an explicit ``contract_version``; +- requests are signed with HMAC-SHA-256 over ``.`` using + the client's dedicated callback signing secret (never a Bearer credential); +- retries use bounded exponential backoff for transport errors, ``429`` and + ``5xx``; other ``4xx`` and redirects are terminal; and +- payloads carry stable identifiers, timestamps, and safe reason codes only -- + never recipient addresses, message content, credentials, or raw provider + data. +""" + +import hashlib +import hmac +import json +import logging +import socket +import urllib.error +import urllib.request +import uuid +from datetime import timedelta + +from django.conf import settings +from django.utils import timezone + +from mailing.models import ( + CallbackEndpoint, + ClientCallback, + ClientCallbackStatus, + EmailEventType, +) + +logger = logging.getLogger(__name__) + +SIGNATURE_PREFIX = "sha256=" +HTTP_REDIRECT_CLASSES = {301, 302, 303, 307, 308} +HTTP_RATE_LIMIT_STATUS = 429 + +CALLBACK_EVENT_TYPES = { + EmailEventType.QUEUED: "delivery.accepted", + EmailEventType.SENT: "delivery.accepted", + EmailEventType.SKIPPED: "delivery.suppressed", + EmailEventType.DELIVERED: "delivery.delivered", + EmailEventType.BOUNCE: "delivery.bounced", + EmailEventType.COMPLAINT: "delivery.complained", + EmailEventType.OPEN: "engagement.opened", + EmailEventType.CLICK: "engagement.clicked", + EmailEventType.SUBSCRIBE: "subscription.changed", + EmailEventType.UNSUBSCRIBE: "subscription.changed", +} + +# The only metadata values safe enough to become a reason code. Anything else +# collapses to a generic code so provider diagnostics never leak to clients. +SUPPRESSION_REASON_CODES = { + "unverified", + "invalid_email", + "global_unsubscribe", + "client_unsubscribe", + "audience_unsubscribe", + "hard_bounce", + "complaint", + "duplicate", + "suppressed", + "category_unsubscribe", + "missing_category_scope", +} + +MESSAGE_KIND_TRANSACTIONAL = "transactional" +MESSAGE_KIND_CAMPAIGN = "campaign" + + +def callback_event_id(event): + """Stable event id for one transition: the same on every reprocessing.""" + return uuid.uuid5(uuid.NAMESPACE_URL, f"urn:relay:email-event:{event.pk}") + + +def canonical_callback_body(payload): + return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def callback_signature(secret, timestamp, body): + """Sign ``.`` exactly as webhook tasks are signed.""" + digest = hmac.new(secret.encode(), timestamp.encode() + b"." + body, hashlib.sha256).hexdigest() + return SIGNATURE_PREFIX + digest + + +def signed_callback_headers(endpoint, callback, *, attempt): + timestamp = str(int(timezone.now().timestamp())) + signature = callback_signature(endpoint.signing_secret, timestamp, callback.body.encode("utf-8")) + return { + "Content-Type": "application/json", + "User-Agent": "relay/1", + "X-Relay-Timestamp": timestamp, + "X-Relay-Signature": signature, + "X-Relay-Event-Id": str(callback.event_id), + "X-Relay-Event-Type": callback.event_type, + "X-Relay-Contract-Version": str(callback.contract_version), + "X-Relay-Attempt": str(attempt), + } + + +def build_callback_payload(event, endpoint, *, sequence): + """Build the redacted, versioned event payload for one transition. + + Returns ``None`` when the transition has no client-visible callback (no + owning client, unmapped event type). The payload carries identifiers, + timestamps, and safe reason codes only; redaction canaries (recipient + address, subject, body, context, credentials, raw provider data) never + enter it. + """ + event_type = CALLBACK_EVENT_TYPES.get(event.event_type) + if event_type is None: + return None + + message_kind, message_ref, template_key, client_reference = message_identity(event) + payload = { + "contract_version": endpoint.contract_version, + "event_id": str(callback_event_id(event)), + "event_type": event_type, + "timestamp": event.created_at.isoformat(), + "sequence": sequence, + "message_id": message_ref or None, + "client_reference": client_reference or None, + "template_key": template_key, + } + if event_type == "delivery.bounced": + payload["bounce_type"] = "hard" if is_hard_bounce(event.metadata or {}) else "soft" + reason = safe_reason_code(event) + if reason: + payload["reason_code"] = reason + return payload + + +def message_identity(event): + if event.transactional_message_id and event.transactional_message is not None: + return ( + MESSAGE_KIND_TRANSACTIONAL, + str(event.transactional_message_id), + event.transactional_message.template_key, + event.transactional_message.idempotency_key, + ) + if event.transactional_message_id: + return MESSAGE_KIND_TRANSACTIONAL, str(event.transactional_message_id), "", "" + if event.campaign_recipient_id: + return MESSAGE_KIND_CAMPAIGN, str(event.campaign_recipient_id), "", "" + return "", "", "", "" + + +def safe_reason_code(event): + metadata = event.metadata or {} + if event.event_type == EmailEventType.BOUNCE: + return "hard_bounce" if is_hard_bounce(metadata) else "soft_bounce" + if event.event_type == EmailEventType.COMPLAINT: + return "complaint" + if event.event_type == EmailEventType.SKIPPED: + reason = str(metadata.get("reason", "")) + return reason if reason in SUPPRESSION_REASON_CODES else "suppressed" + if event.event_type == EmailEventType.SUBSCRIBE: + return "subscribed" + if event.event_type == EmailEventType.UNSUBSCRIBE: + return "unsubscribed" + return "" + + +def is_hard_bounce(metadata): + bounce_type = (metadata.get("bounce_type") or "").casefold() + bounce_sub_type = (metadata.get("bounce_sub_type") or "").casefold() + return bounce_type == "permanent" or bounce_sub_type in {"general", "suppressed", "onaccountsuppressionlist"} + + +def next_callback_sequence(client, message_kind, message_ref): + if not message_kind: + return ClientCallback.objects.filter(client=client, message_kind="").count() + 1 + return ( + ClientCallback.objects.filter( + client=client, + message_kind=message_kind, + message_ref=message_ref, + ).count() + + 1 + ) + + +def emit_client_callback(event): + """Create the durable callback work for one committed-pending transition. + + Call inside the transaction that appends the ``EmailEvent`` so the callback + row commits atomically with the state it announces. Deduplicated on + ``(client, event_id)``: repeated transition processing returns the existing + row instead of creating a second one. + """ + client = event.client + if client is None or event.event_type not in CALLBACK_EVENT_TYPES: + return None + if event.campaign_recipient_id is not None: + # Campaign-scoped transitions stay on the CMP channel. Client + # callbacks carry transactional deliveries and client-level + # subscription changes, the transitions whose receivers own a + # client_reference. + return None + endpoint = CallbackEndpoint.objects.filter(client=client, enabled=True).first() + if endpoint is None: + return None + + message_kind, message_ref, _, _ = message_identity(event) + payload = build_callback_payload( + event, + endpoint, + sequence=next_callback_sequence(client, message_kind, message_ref), + ) + if payload is None: + return None + body = canonical_callback_body(payload) + callback, _ = ClientCallback.objects.get_or_create( + client=client, + event_id=uuid.UUID(payload["event_id"]), + defaults={ + "endpoint": endpoint, + "email_event": event, + "transactional_message_id": event.transactional_message_id, + "campaign_recipient_id": event.campaign_recipient_id, + "event_type": payload["event_type"], + "contract_version": payload["contract_version"], + "client_reference": payload["client_reference"] or "", + "message_kind": message_kind, + "message_ref": message_ref, + "template_key": payload["template_key"], + "sequence": payload["sequence"], + "payload": payload, + "body": body.decode("utf-8"), + "body_hash": hashlib.sha256(body).hexdigest(), + "max_attempts": settings.CLIENT_CALLBACK_MAX_ATTEMPTS, + "next_attempt_at": timezone.now(), + }, + ) + return callback + + +def due_client_callbacks(*, limit=25, now=None): + now = now or timezone.now() + return ( + ClientCallback.objects.select_related("client", "endpoint") + .filter(status=ClientCallbackStatus.PENDING, next_attempt_at__lte=now) + .order_by("next_attempt_at", "id")[:limit] + ) + + +def process_due_client_callbacks(*, limit=25, now=None): + processed = 0 + delivered = 0 + failed = 0 + for callback in due_client_callbacks(limit=limit, now=now): + processed += 1 + if dispatch_client_callback(callback): + delivered += 1 + else: + failed += 1 + return {"processed": processed, "delivered": delivered, "failed": failed} + + +def dispatch_client_callback(callback, *, now=None): + """Deliver one callback attempt. Never touches transport state.""" + # Re-read the endpoint: enable/disable may have changed since the callback + # row was created or last attempted. + endpoint = CallbackEndpoint.objects.filter(pk=callback.endpoint_id, enabled=True).first() + if endpoint is None: + mark_callback_failed(callback, "endpoint_disabled", "Callback endpoint is disabled.") + return False + + try: + post_callback(endpoint, callback) + except RetryableCallbackError as exc: + logger.warning("client callback retry event_id=%s code=%s", callback.event_id, exc.code) + mark_callback_retry(callback, exc.code, str(exc), response_status_class=exc.status_class, now=now) + return False + except PermanentCallbackError as exc: + mark_callback_failed(callback, exc.code, str(exc), response_status_class=exc.status_class) + return False + + mark_callback_delivered(callback, now=now) + return True + + +def post_callback(endpoint, callback): + """POST the canonical body. Returns the 2xx status or raises.""" + body = callback.body.encode("utf-8") + headers = signed_callback_headers(endpoint, callback, attempt=callback.attempt_count + 1) + request = urllib.request.Request(endpoint.url, data=body, headers=headers, method="POST") + timeout = settings.CLIENT_CALLBACK_TIMEOUT_SECONDS + try: + with NO_REDIRECT_OPENER.open(request, timeout=timeout) as response: + status = response.status + except urllib.error.HTTPError as exc: + exc.close() + if exc.code in HTTP_REDIRECT_CLASSES: + raise PermanentCallbackError( + "callback endpoint redirected; redirects are not followed", + code="redirect_not_allowed", + status_class="3xx", + ) from exc + if exc.code >= 500 or exc.code == HTTP_RATE_LIMIT_STATUS: + code = "rate_limited" if exc.code == HTTP_RATE_LIMIT_STATUS else "http_5xx" + raise RetryableCallbackError( + f"callback endpoint returned HTTP {exc.code}", code=code, status_class="5xx" if exc.code >= 500 else "4xx" + ) from exc + raise PermanentCallbackError( + f"callback endpoint returned HTTP {exc.code}", code="http_4xx", status_class="4xx" + ) from exc + except (urllib.error.URLError, TimeoutError, socket.timeout, OSError) as exc: + code = "timeout" if isinstance(exc, (TimeoutError, socket.timeout)) else "connection_error" + raise RetryableCallbackError(f"callback delivery failed: {exc}", code=code, status_class="") from exc + + if not 200 <= status < 300: + raise RetryableCallbackError( + f"callback endpoint returned HTTP {status}", code="http_5xx", status_class="5xx" + ) + return status + + +class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +NO_REDIRECT_OPENER = urllib.request.build_opener(_NoRedirectHandler) + + +class RetryableCallbackError(Exception): + def __init__(self, message, *, code, status_class=""): + self.code = code + self.status_class = status_class + super().__init__(message) + + +class PermanentCallbackError(Exception): + def __init__(self, message, *, code, status_class=""): + self.code = code + self.status_class = status_class + super().__init__(message) + + +def mark_callback_delivered(callback, *, now=None): + now = now or timezone.now() + callback.status = ClientCallbackStatus.DELIVERED + callback.attempt_count += 1 + callback.last_attempt_at = now + callback.delivered_at = now + callback.response_status_class = "2xx" + callback.last_error_code = "" + callback.last_error = "" + callback.save( + update_fields=[ + "status", + "attempt_count", + "last_attempt_at", + "delivered_at", + "response_status_class", + "last_error_code", + "last_error", + "updated_at", + ] + ) + + +def mark_callback_retry(callback, error_code, error, *, response_status_class="", now=None): + now = now or timezone.now() + callback.attempt_count += 1 + callback.last_attempt_at = now + callback.response_status_class = response_status_class + callback.last_error_code = error_code + callback.last_error = error[:500] + if callback.attempt_count >= callback.max_attempts: + callback.status = ClientCallbackStatus.FAILED + else: + callback.next_attempt_at = now + retry_delay(callback) + callback.save( + update_fields=[ + "status", + "attempt_count", + "next_attempt_at", + "last_attempt_at", + "response_status_class", + "last_error_code", + "last_error", + "updated_at", + ] + ) + + +def mark_callback_failed(callback, error_code, error, *, response_status_class=""): + now = timezone.now() + callback.status = ClientCallbackStatus.FAILED + callback.attempt_count += 1 + callback.last_attempt_at = now + callback.response_status_class = response_status_class + callback.last_error_code = error_code + callback.last_error = error[:500] + callback.save( + update_fields=[ + "status", + "attempt_count", + "last_attempt_at", + "response_status_class", + "last_error_code", + "last_error", + "updated_at", + ] + ) + + +def retry_delay(callback): + """Bounded exponential backoff with jitter deterministic in the callback. + + ``base * 2^(attempt-1)`` grows by a quarter of itself at most, with the + fraction derived from the stable event id and attempt number, so the + schedule is fully determined by stored data. + """ + attempt = max(callback.attempt_count, 1) + base = settings.CLIENT_CALLBACK_RETRY_BASE_SECONDS + cap = settings.CLIENT_CALLBACK_RETRY_MAX_DELAY_SECONDS + stall = int(hashlib.sha256(f"{callback.event_id}:{attempt}".encode()).hexdigest()[:8], 16) / 0xFFFFFFFF + delay = base * (2 ** (attempt - 1)) + delay = min(delay * (1 + 0.25 * stall), cap) + return timedelta(seconds=delay) + + +def callback_backlog_status(): + """Safe operator metrics: backlog, oldest age, recent failures. No PII.""" + now = timezone.now() + pending = ClientCallback.objects.filter(status=ClientCallbackStatus.PENDING) + oldest = pending.order_by("created_at").values_list("created_at", flat=True).first() + return { + "pending": pending.count(), + "oldest_pending_age_s": int((now - oldest).total_seconds()) if oldest else 0, + "failed_24h": ClientCallback.objects.filter( + status=ClientCallbackStatus.FAILED, + last_attempt_at__gte=now - timedelta(hours=24), + ).count(), + } diff --git a/mailing/services/ses_webhooks.py b/mailing/services/ses_webhooks.py index 083a1c0..65b92b7 100644 --- a/mailing/services/ses_webhooks.py +++ b/mailing/services/ses_webhooks.py @@ -24,6 +24,7 @@ TransactionalMessageStatus, ) from mailing.queue_contracts import CONTRACT_VERSION, SES_WEBHOOKS_CONTRACT, validate_ses_webhook_message +from mailing.services.client_callbacks import emit_client_callback, is_hard_bounce from mailing.services.cmp_callbacks import emit_cmp_contact_event SNS_NOTIFICATION = "Notification" @@ -305,6 +306,10 @@ def apply_correlated_updates(payload, source, event): notification_type == "bounce" and is_hard_bounce(metadata) ): emit_cmp_contact_event(event) + # Every correlated transactional transition is client-visible on the + # client callback channel, including soft bounces: reason_code carries + # the hard/soft distinction. + emit_client_callback(event) def apply_campaign_recipient_update(recipient, notification_type, occurred_at, metadata): @@ -419,12 +424,6 @@ def mark_complained(contact, occurred_at): ) -def is_hard_bounce(metadata): - bounce_type = (metadata.get("bounce_type") or "").casefold() - bounce_sub_type = (metadata.get("bounce_sub_type") or "").casefold() - return bounce_type == "permanent" or bounce_sub_type in {"general", "suppressed", "onaccountsuppressionlist"} - - def event_timestamp(payload): metadata = payload.get("metadata") or {} for key in ("event_timestamp", "mail_timestamp"): diff --git a/mailing/services/tracking.py b/mailing/services/tracking.py index c3a0556..3857fbc 100644 --- a/mailing/services/tracking.py +++ b/mailing/services/tracking.py @@ -5,6 +5,7 @@ from django.utils import timezone from mailing.models import Campaign, CampaignRecipient, CampaignRecipientStatus, EmailEvent, EmailEventType +from mailing.services.client_callbacks import emit_client_callback from mailing.services.cmp_callbacks import emit_cmp_contact_event from mailing.services.contacts import unsubscribe_contact from mailing.services.tokens import get_recipient_by_tracking_token, get_recipient_by_unsubscribe_token @@ -50,6 +51,7 @@ def record_open(raw_token): recipient.save(update_fields=update_fields) event = _create_campaign_event(recipient, EmailEventType.OPEN) emit_cmp_contact_event(event) + emit_client_callback(event) refresh_campaign_engagement_counts(recipient.campaign) return recipient @@ -81,6 +83,7 @@ def record_click(raw_token, destination_url): recipient.save(update_fields=update_fields) event = _create_campaign_event(recipient, EmailEventType.CLICK, url=destination_url) emit_cmp_contact_event(event) + emit_client_callback(event) refresh_campaign_engagement_counts(recipient.campaign) return recipient @@ -128,6 +131,7 @@ def apply_unsubscribe(raw_token, scope): metadata={"scope": scope}, ) emit_cmp_contact_event(event) + emit_client_callback(event) refresh_campaign_engagement_counts(campaign) return recipient diff --git a/mailing/services/transactional.py b/mailing/services/transactional.py index b445d96..309d3fb 100644 --- a/mailing/services/transactional.py +++ b/mailing/services/transactional.py @@ -34,6 +34,7 @@ subscription_confirm_url, validate_canonical_category, ) +from mailing.services.client_callbacks import emit_client_callback from mailing.services.cmp_callbacks import emit_cmp_contact_event from mailing.services.contacts import is_transactional_email_allowed, normalize_email, upsert_contact from mailing.services.recipient_lists import ( @@ -1153,6 +1154,7 @@ def append_transactional_event(message, event_type, metadata): metadata=metadata, ) emit_cmp_contact_event(event) + emit_client_callback(event) return event diff --git a/mailing/services/transactional_sender.py b/mailing/services/transactional_sender.py index 19f9b2c..acf9d65 100644 --- a/mailing/services/transactional_sender.py +++ b/mailing/services/transactional_sender.py @@ -5,6 +5,7 @@ from mailing.aws import ses_client from mailing.models import EmailEvent, EmailEventType, TransactionalMessage, TransactionalMessageStatus +from mailing.services.client_callbacks import emit_client_callback from mailing.services.cmp_callbacks import emit_cmp_contact_event from mailing.ses import send_email @@ -191,6 +192,7 @@ def _append_event(message, event_type, metadata): metadata=metadata, ) emit_cmp_contact_event(event) + emit_client_callback(event) return event diff --git a/mailing/services/worker_status.py b/mailing/services/worker_status.py index 68ec34d..ae6b780 100644 --- a/mailing/services/worker_status.py +++ b/mailing/services/worker_status.py @@ -12,6 +12,8 @@ CampaignRecipient, CampaignRecipientStatus, CampaignStatus, + ClientCallback, + ClientCallbackStatus, CmpCallback, CmpCallbackStatus, RecipientListImportJob, @@ -89,6 +91,13 @@ class WorkerDefinition: "process_cmp_callbacks", "Due callbacks", ), + WorkerDefinition( + "client-callbacks", + "Client callbacks", + "relay-client-callbacks-worker.service", + "process_client_callbacks", + "Due callbacks", + ), WorkerDefinition( "recipient-list-imports", "Recipient list imports", @@ -220,6 +229,11 @@ def _backlog_count(worker_key: str) -> int | None: Q(status=CmpCallbackStatus.PENDING, next_attempt_at__lte=timezone.now()) | Q(status=CmpCallbackStatus.FAILED) ).count() + if worker_key == "client-callbacks": + return ClientCallback.objects.filter( + Q(status=ClientCallbackStatus.PENDING, next_attempt_at__lte=timezone.now()) + | Q(status=ClientCallbackStatus.FAILED) + ).count() if worker_key == "recipient-list-imports": return RecipientListImportJob.objects.filter( status__in=[ diff --git a/mailing/tests/__init__.py b/mailing/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mailing/tests/callback_helpers.py b/mailing/tests/callback_helpers.py new file mode 100644 index 0000000..3e86f95 --- /dev/null +++ b/mailing/tests/callback_helpers.py @@ -0,0 +1,64 @@ +"""Shared helpers for client callback tests. + +Nothing here is collected by pytest: the module only exists so the callback +fixtures, the fake no-redirect opener, and the HTTP error builders stay +identical across every test file that exercises the dispatcher. +""" + +from urllib.error import HTTPError + +from mailing.models import CallbackEndpoint + +DEFAULT_CALLBACK_URL = "https://callback.example.com/hooks" +DEFAULT_CALLBACK_SECRET = "callback-signing-secret" + + +class CallbackResponse: + def __init__(self, status=200): + self.status = status + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + return False + + +def create_callback_endpoint(client, *, url=DEFAULT_CALLBACK_URL, secret=DEFAULT_CALLBACK_SECRET): + return CallbackEndpoint.objects.create(client=client, url=url, signing_secret=secret) + + +class FakeCallbackOpener: + """Stands in for the dispatcher's no-redirect opener, recording requests. + + Outcomes are consumed one per request: an ``int`` answers with that HTTP + status, an ``Exception`` instance is raised instead. + """ + + def __init__(self, outcomes=None): + self.requests = [] + self.outcomes = list(outcomes or []) + + def open(self, request, *, timeout): + self.requests.append( + { + "url": request.full_url, + "body": bytes(request.data), + "headers": {key.lower(): value for key, value in request.header_items()}, + "timeout": timeout, + } + ) + outcome = self.outcomes.pop(0) if self.outcomes else 200 + if isinstance(outcome, Exception): + raise outcome + return CallbackResponse(outcome) + + +def install_fake_opener(monkeypatch, outcomes=None): + opener = FakeCallbackOpener(outcomes) + monkeypatch.setattr("mailing.services.client_callbacks.NO_REDIRECT_OPENER", opener) + return opener + + +def http_error(code): + return HTTPError(DEFAULT_CALLBACK_URL, code, "error", hdrs=None, fp=None) diff --git a/mailing/tests/test_client_api.py b/mailing/tests/test_client_api.py index 7ad3a9a..69f8434 100644 --- a/mailing/tests/test_client_api.py +++ b/mailing/tests/test_client_api.py @@ -7,12 +7,14 @@ from mailing.models import ( Audience, + CallbackEndpoint, Campaign, CampaignRecipient, CampaignRecipientStatus, CategoryPreference, Client, ClientApiKey, + ClientCallback, CmpCallback, Contact, ContactSourceMetadata, @@ -1057,7 +1059,24 @@ def test_contact_erase_deletes_live_state_and_anonymizes_history(client, audienc event_type=EmailEventType.SENT, metadata={"email": "person@example.com"}, ) - callback = CmpCallback.objects.create( + endpoint = CallbackEndpoint.objects.create( + client=api_client_record, + url="https://courses.example.com/webhook", + signing_secret="callback-signing-secret", + ) + callback = ClientCallback.objects.create( + email_event=event, + transactional_message=message, + client=api_client_record, + endpoint=endpoint, + event_id="00000000-0000-5000-8000-000000000001", + event_type="delivery.accepted", + payload={}, + body="{}", + body_hash="0" * 64, + next_attempt_at=timezone.now(), + ) + cmp_callback = CmpCallback.objects.create( email_event=event, contact=contact, audience=audience, @@ -1100,12 +1119,15 @@ def test_contact_erase_deletes_live_state_and_anonymizes_history(client, audienc recipient.refresh_from_db() event.refresh_from_db() callback.refresh_from_db() + cmp_callback.refresh_from_db() assert message.email == contact.email assert message.context == {} assert message.metadata == {"erased": True} assert recipient.email == contact.email assert event.metadata == {"erased": True} - assert callback.payload == {"erased": True} + # Callback rows are redacted at creation, so erasure has nothing to scrub. + assert callback.payload == {} + assert cmp_callback.payload == {"erased": True} assert Contact.objects.filter(normalized_email="person@example.com").exists() is False second_response = post_json( diff --git a/mailing/tests/test_client_callbacks.py b/mailing/tests/test_client_callbacks.py new file mode 100644 index 0000000..e0e2ff7 --- /dev/null +++ b/mailing/tests/test_client_callbacks.py @@ -0,0 +1,632 @@ +"""Generic tenant-scoped client callbacks (plan issue R1.4, relay issue #17). + +Covers: the redacted versioned event payload, event-type mapping, row +deduplication under repeated transition processing, the timestamped HMAC +header contract (no Bearer credential), the deterministic contract fixture, +retry classification with bounded backoff, terminal failures (permanent 4xx, +redirects, retry exhaustion, endpoint disablement), per-message sequencing, +and proof that callback failure never regresses transport state. +""" + +import hashlib +import hmac +import json +from datetime import timedelta +from pathlib import Path + +import pytest +from django.conf import settings +from django.utils import timezone +from django.utils.dateparse import parse_datetime + +from mailing.models import ( + Audience, + CallbackEndpoint, + Campaign, + CampaignRecipient, + Client, + ClientCallback, + ClientCallbackStatus, + Contact, + EmailEvent, + EmailEventType, + EmailTemplate, + Organization, + TransactionalMessage, + TransactionalMessageStatus, +) +from mailing.services.client_callbacks import ( + callback_event_id, + callback_signature, + canonical_callback_body, + dispatch_client_callback, + emit_client_callback, + process_due_client_callbacks, +) +from mailing.tests.callback_helpers import ( + DEFAULT_CALLBACK_SECRET, + DEFAULT_CALLBACK_URL, + create_callback_endpoint, + http_error, + install_fake_opener, +) + +pytestmark = pytest.mark.django_db + +FIXTURE_PATH = Path(__file__).resolve().parents[2] / "tests" / "fixtures" / "client_callback_contract_v1.json" + + +@pytest.fixture +def organization(): + return Organization.objects.create(name="DataTalksClub", slug="datatalksclub") + + +@pytest.fixture +def app_client(organization): + return Client.objects.create(organization=organization, name="DTC Courses", slug="dtc-courses") + + +@pytest.fixture +def other_client(organization): + return Client.objects.create(organization=organization, name="Newsletter", slug="dtc-newsletter") + + +@pytest.fixture +def contact(): + return Contact.objects.create(email="learner@example.com") + + +@pytest.fixture +def template(app_client): + return EmailTemplate.objects.create( + client=app_client, + key="registration-welcome", + name="Registration welcome", + subject="Welcome", + is_transactional=True, + ) + + +@pytest.fixture +def transactional_message(app_client, contact, template): + return TransactionalMessage.objects.create( + client=app_client, + contact=contact, + email=contact.email, + template=template, + template_key=template.key, + status=TransactionalMessageStatus.QUEUED, + idempotency_key="registration-user-123", + subject="Welcome to the course", + html_body="

Welcome to the course

", + text_body="Welcome to the course", + ) + + +def callback_endpoint(app_client): + return create_callback_endpoint(app_client) + + +def transition_event(transactional_message, event_type=EmailEventType.BOUNCE, metadata=None): + return EmailEvent.objects.create( + transactional_message=transactional_message, + contact=transactional_message.contact, + client=transactional_message.client, + event_type=event_type, + metadata=metadata or {}, + ) + + +# --- Event payload: mapping, redaction, dedup ------------------------------- + + +@pytest.mark.parametrize( + ("email_event_type", "callback_event_type", "reason_code"), + [ + (EmailEventType.QUEUED, "delivery.accepted", None), + (EmailEventType.SENT, "delivery.accepted", None), + (EmailEventType.SKIPPED, "delivery.suppressed", "suppressed"), + (EmailEventType.DELIVERED, "delivery.delivered", None), + (EmailEventType.BOUNCE, "delivery.bounced", "soft_bounce"), + (EmailEventType.COMPLAINT, "delivery.complained", "complaint"), + (EmailEventType.OPEN, "engagement.opened", None), + (EmailEventType.CLICK, "engagement.clicked", None), + (EmailEventType.SUBSCRIBE, "subscription.changed", "subscribed"), + (EmailEventType.UNSUBSCRIBE, "subscription.changed", "unsubscribed"), + ], +) +def test_every_transition_maps_to_one_versioned_callback_event( + transactional_message, + app_client, + email_event_type, + callback_event_type, + reason_code, +): + callback_endpoint(app_client) + event = transition_event(transactional_message, email_event_type) + + callback = emit_client_callback(event) + + assert callback is not None + assert callback.event_type == callback_event_type + assert callback.contract_version == 1 + assert callback.payload["event_type"] == callback_event_type + if reason_code is None: + assert "reason_code" not in callback.payload + else: + assert callback.payload["reason_code"] == reason_code + + +def test_payload_carries_identifiers_and_never_redaction_canaries( + transactional_message, + app_client, +): + callback_endpoint(app_client) + event = transition_event( + transactional_message, + EmailEventType.BOUNCE, + metadata={"bounce_type": "Permanent", "bounce_sub_type": "General", "diagnostic": "smtp: quit whining"}, + ) + + callback = emit_client_callback(event) + body = callback.body + + assert set(callback.payload) == { + "bounce_type", + "client_reference", + "contract_version", + "event_id", + "event_type", + "reason_code", + "sequence", + "message_id", + "template_key", + "timestamp", + } + assert callback.payload["message_id"] == str(transactional_message.pk) + assert callback.payload["client_reference"] == "registration-user-123" + assert callback.payload["template_key"] == "registration-welcome" + assert callback.payload["bounce_type"] == "hard" + assert callback.payload["reason_code"] == "hard_bounce" + assert parse_datetime(callback.payload["timestamp"]) is not None + for canary in ( + "learner@example.com", + "Welcome to the course", + "smtp: quit whining", + "diagnostic", + ): + assert canary not in body + assert callback.body_hash == hashlib.sha256(body.encode("utf-8")).hexdigest() + assert callback.body == canonical_callback_body(callback.payload).decode("utf-8") + + +def test_soft_bounce_and_omitted_bounce_type_metadata_report_soft(transactional_message, app_client): + callback_endpoint(app_client) + soft = emit_client_callback(transition_event(transactional_message, metadata={"bounce_type": "Transient"})) + bare = emit_client_callback(transition_event(transactional_message)) + + assert soft.payload["bounce_type"] == "soft" + assert soft.payload["reason_code"] == "soft_bounce" + assert bare.payload["bounce_type"] == "soft" + + +def test_contact_level_subscription_event_has_no_message_identity(app_client, contact): + callback_endpoint(app_client) + event = EmailEvent.objects.create( + contact=contact, + client=app_client, + event_type=EmailEventType.UNSUBSCRIBE, + metadata={"source": "api"}, + ) + + callback = emit_client_callback(event) + + assert callback is not None + assert callback.event_type == "subscription.changed" + assert callback.payload["message_id"] is None + assert callback.payload["client_reference"] is None + assert callback.payload["template_key"] == "" + + +def test_campaign_and_send_failure_transitions_create_no_callback( + transactional_message, + app_client, + organization, + contact, +): + callback_endpoint(app_client) + audience = Audience.objects.create(organization=organization, name="Newsletter", slug="newsletter") + campaign = Campaign.objects.create(audience=audience, client=app_client, subject="Weekly update") + recipient = CampaignRecipient.objects.create( + campaign=campaign, + contact=contact, + email=contact.email, + ses_message_id="ses-campaign-1", + ) + campaign_event = EmailEvent.objects.create( + campaign_recipient=recipient, + client=app_client, + event_type=EmailEventType.DELIVERED, + metadata={}, + ) + + assert emit_client_callback(campaign_event) is None + assert emit_client_callback(transition_event(transactional_message, EmailEventType.FAILED)) is None + assert ClientCallback.objects.count() == 0 + + +def test_emit_is_deduplicated_per_client_and_event(transactional_message, app_client): + callback_endpoint(app_client) + event = transition_event(transactional_message) + + first = emit_client_callback(event) + second = emit_client_callback(event) + + assert first.pk == second.pk + assert ClientCallback.objects.count() == 1 + + +def test_transitions_without_a_client_or_endpoint_create_no_callback( + transactional_message, + app_client, + other_client, + contact, +): + callback_endpoint(app_client) + event = transition_event(transactional_message) + + # A transition owned by another client must not deliver to this endpoint. + other_event = EmailEvent.objects.create( + contact=contact, + client=other_client, + event_type=EmailEventType.DELIVERED, + metadata={}, + ) + + assert emit_client_callback(event) is not None + + endpoint_disabled = CallbackEndpoint.objects.get(client=app_client) + endpoint_disabled.disable("rotation test", disabled_at=timezone.now()) + orphan = transition_event(transactional_message) + assert emit_client_callback(orphan) is None + assert emit_client_callback(other_event) is None + assert ClientCallback.objects.count() == 1 + + +def test_sequence_is_monotonic_per_message(transactional_message, app_client): + callback_endpoint(app_client) + + first = emit_client_callback(transition_event(transactional_message, EmailEventType.QUEUED)) + second = emit_client_callback(transition_event(transactional_message, EmailEventType.SENT)) + third = emit_client_callback(transition_event(transactional_message, EmailEventType.DELIVERED)) + + assert [first.sequence, second.sequence, third.sequence] == [1, 2, 3] + + +def test_event_id_is_stable_for_one_transition(transactional_message, app_client): + callback_endpoint(app_client) + event = transition_event(transactional_message) + + assert str(callback_event_id(event)) == emit_client_callback(event).payload["event_id"] + + +# --- Signing and dispatch --------------------------------------------------- + + +def signed_request(opener): + request = opener.requests[0] + return ( + request, + json.loads(request["body"].decode("utf-8")), + request["body"].decode("utf-8"), + ) + + +def test_dispatch_posts_the_canonical_body_with_timestamped_hmac( + transactional_message, + app_client, + monkeypatch, +): + callback_endpoint(app_client) + callback = emit_client_callback(transition_event(transactional_message)) + opener = install_fake_opener(monkeypatch) + + assert dispatch_client_callback(callback) is True + + request, body, raw_body = signed_request(opener) + assert request["url"] == DEFAULT_CALLBACK_URL + assert "authorization" not in request["headers"] + assert request["headers"]["content-type"] == "application/json" + assert body == callback.payload + assert raw_body == callback.body + expected = callback_signature(DEFAULT_CALLBACK_SECRET, request["headers"]["x-relay-timestamp"], request["body"]) + assert request["headers"]["x-relay-signature"] == expected + assert request["headers"]["x-relay-event-id"] == str(callback.event_id) + assert request["headers"]["x-relay-contract-version"] == "1" + assert request["headers"]["x-relay-attempt"] == "1" + + callback.refresh_from_db() + assert callback.status == ClientCallbackStatus.DELIVERED + assert callback.attempt_count == 1 + assert callback.response_status_class == "2xx" + assert callback.delivered_at is not None + + +def test_retry_after_transport_failure_then_success_keeps_one_identical_body( + transactional_message, + app_client, + monkeypatch, +): + callback_endpoint(app_client) + callback = emit_client_callback(transition_event(transactional_message)) + opener = install_fake_opener(monkeypatch, outcomes=[TimeoutError("timed out"), 200]) + + assert dispatch_client_callback(callback) is False + callback.refresh_from_db() + first_body = callback.body + assert callback.status == ClientCallbackStatus.PENDING + assert callback.attempt_count == 1 + assert callback.last_error_code == "timeout" + assert callback.next_attempt_at > callback.created_at + + callback.next_attempt_at = timezone.now() + callback.save(update_fields=["next_attempt_at", "updated_at"]) + assert dispatch_client_callback(callback) is True + + callback.refresh_from_db() + assert callback.status == ClientCallbackStatus.DELIVERED + assert callback.attempt_count == 2 + assert callback.body == first_body + request = opener.requests[1] + expected = callback_signature(DEFAULT_CALLBACK_SECRET, request["headers"]["x-relay-timestamp"], request["body"]) + assert request["headers"]["x-relay-signature"] == expected + assert request["headers"]["x-relay-attempt"] == "2" + + +@pytest.mark.parametrize( + ("outcome", "expected_code", "expected_class", "retryable"), + [ + (http_error(429), "rate_limited", "4xx", True), + (http_error(500), "http_5xx", "5xx", True), + (http_error(503), "http_5xx", "5xx", True), + (http_error(404), "http_4xx", "4xx", False), + (http_error(410), "http_4xx", "4xx", False), + (http_error(302), "redirect_not_allowed", "3xx", False), + ], +) +def test_http_outcome_classification( + transactional_message, + app_client, + monkeypatch, + outcome, + expected_code, + expected_class, + retryable, +): + callback_endpoint(app_client) + callback = emit_client_callback(transition_event(transactional_message)) + install_fake_opener(monkeypatch, outcomes=[outcome]) + + assert dispatch_client_callback(callback) is False + + callback.refresh_from_db() + assert callback.last_error_code == expected_code + assert callback.response_status_class == expected_class + if retryable: + assert callback.status == ClientCallbackStatus.PENDING + assert callback.next_attempt_at > callback.last_attempt_at + else: + assert callback.status == ClientCallbackStatus.FAILED + + +def test_retry_exhaustion_becomes_a_terminal_failure(transactional_message, app_client, monkeypatch): + callback_endpoint(app_client) + callback = emit_client_callback(transition_event(transactional_message)) + callback.max_attempts = 2 + callback.save(update_fields=["max_attempts", "updated_at"]) + install_fake_opener(monkeypatch, outcomes=[http_error(500), http_error(500)]) + + assert dispatch_client_callback(callback) is False + callback.refresh_from_db() + assert callback.status == ClientCallbackStatus.PENDING + assert callback.attempt_count == 1 + + assert dispatch_client_callback(callback) is False + callback.refresh_from_db() + assert callback.status == ClientCallbackStatus.FAILED + assert callback.attempt_count == 2 + assert callback.last_error_code == "http_5xx" + + +def test_disabled_endpoint_fails_pending_callbacks_without_posting( + transactional_message, + app_client, + monkeypatch, +): + endpoint = callback_endpoint(app_client) + callback = emit_client_callback(transition_event(transactional_message)) + opener = install_fake_opener(monkeypatch) + endpoint.disable("revoked", disabled_at=timezone.now()) + + assert dispatch_client_callback(callback) is False + + assert opener.requests == [] + callback.refresh_from_db() + assert callback.status == ClientCallbackStatus.FAILED + assert callback.last_error_code == "endpoint_disabled" + + +def test_rotation_signs_with_the_new_secret_and_keeps_the_previous_one( + transactional_message, + app_client, + monkeypatch, +): + endpoint = callback_endpoint(app_client) + callback = emit_client_callback(transition_event(transactional_message)) + rotated_at = timezone.now() + endpoint.rotate_secret("rotated-callback-secret", rotated_at=rotated_at) + opener = install_fake_opener(monkeypatch) + + assert dispatch_client_callback(callback) is True + + request = opener.requests[0] + expected = callback_signature("rotated-callback-secret", request["headers"]["x-relay-timestamp"], request["body"]) + assert request["headers"]["x-relay-signature"] == expected + endpoint.refresh_from_db() + assert endpoint.previous_signing_secret == DEFAULT_CALLBACK_SECRET + assert endpoint.secret_rotated_at is not None + + +def test_callback_failure_never_regresses_transport_state(transactional_message, app_client, monkeypatch): + transactional_message.status = TransactionalMessageStatus.SENT + transactional_message.delivered_at = timezone.now() + transactional_message.save(update_fields=["status", "delivered_at", "updated_at"]) + callback_endpoint(app_client) + callback = emit_client_callback(transition_event(transactional_message, EmailEventType.DELIVERED)) + install_fake_opener(monkeypatch, outcomes=[http_error(500)]) + + assert dispatch_client_callback(callback) is False + + transactional_message.refresh_from_db() + assert transactional_message.status == TransactionalMessageStatus.SENT + assert transactional_message.delivered_at is not None + + +def test_backoff_grows_bounded_and_deterministic(transactional_message, app_client): + callback_endpoint(app_client) + callback = emit_client_callback(transition_event(transactional_message)) + callback.attempt_count = 1 + + delays = [] + for _ in range(6): + delays.append(dispatch_retry_delay(callback)) + callback.attempt_count += 1 + + assert delays[0] < delays[-1] + assert max(delays) <= settings.CLIENT_CALLBACK_RETRY_MAX_DELAY_SECONDS + # Deterministic: recomputing the schedule for the same attempt is stable. + callback.attempt_count = 3 + assert dispatch_retry_delay(callback) == dispatch_retry_delay(callback) + + +def dispatch_retry_delay(callback): + from mailing.services.client_callbacks import retry_delay # noqa: PLC0415 - test-local import + + return retry_delay(callback).total_seconds() + + +# --- Batch dispatcher -------------------------------------------------------- + + +def test_process_due_dispatches_only_due_rows(transactional_message, app_client, monkeypatch): + callback_endpoint(app_client) + due = emit_client_callback(transition_event(transactional_message, EmailEventType.QUEUED)) + future = emit_client_callback(transition_event(transactional_message, EmailEventType.SENT)) + future.next_attempt_at = timezone.now() + timedelta(hours=1) + future.save(update_fields=["next_attempt_at", "updated_at"]) + opener = install_fake_opener(monkeypatch) + + result = process_due_client_callbacks() + + assert result == {"processed": 1, "delivered": 1, "failed": 0} + assert len(opener.requests) == 1 + due.refresh_from_db() + future.refresh_from_db() + assert due.status == ClientCallbackStatus.DELIVERED + assert future.status == ClientCallbackStatus.PENDING + + +# --- Deterministic contract fixture ----------------------------------------- + + +def load_contract_fixture(): + return json.loads(FIXTURE_PATH.read_text()) + + +def test_contract_fixture_signature_matches_the_relay_signing_function(): + fixture = load_contract_fixture() + + signature = callback_signature( + fixture["signing_secret"], + fixture["timestamp"], + fixture["canonical_body"].encode("utf-8"), + ) + + assert signature == fixture["expected_signature"] + assert json.loads(fixture["canonical_body"]) == fixture["event"] + + +def reference_receiver_verify(body, headers, secrets, *, now, replay_window=300): + """Verbatim mirror of the reference receiver documented in docs/api.md. + + ``secrets`` is the list of accepted signing secrets: the current one and, + during a rotation, the previous one for the bounded overlap. + """ + timestamp = headers.get("X-Relay-Timestamp", "") + signature = headers.get("X-Relay-Signature", "") + try: + stamp = int(timestamp) + except ValueError: + raise ReceiverRejection("missing or malformed timestamp") from None + if abs(now - stamp) > replay_window: + raise ReceiverRejection("timestamp outside replay window") + expected = [ + "sha256=" + hmac.new(secret.encode(), timestamp.encode() + b"." + body, hashlib.sha256).hexdigest() + for secret in secrets + ] + if not any(hmac.compare_digest(candidate, signature) for candidate in expected): + raise ReceiverRejection("bad signature") + + +class ReceiverRejection(Exception): + pass + + +def test_reference_receiver_accepts_the_contract_fixture_and_rotation_overlap(): + fixture = load_contract_fixture() + body = fixture["canonical_body"].encode("utf-8") + now = int(fixture["timestamp"]) + current_headers = { + "X-Relay-Timestamp": fixture["timestamp"], + "X-Relay-Signature": fixture["rotated_expected_signature"], + } + previous_headers = { + "X-Relay-Timestamp": fixture["timestamp"], + "X-Relay-Signature": fixture["expected_signature"], + } + overlap_secrets = [fixture["rotated_signing_secret"], fixture["signing_secret"]] + + # Before rotation: only the current secret verifies. + reference_receiver_verify(body, previous_headers, [fixture["signing_secret"]], now=now) + with pytest.raises(ReceiverRejection, match="bad signature"): + reference_receiver_verify(body, current_headers, [fixture["signing_secret"]], now=now) + # During the bounded rotation overlap: either secret verifies. + reference_receiver_verify(body, current_headers, overlap_secrets, now=now) + reference_receiver_verify(body, previous_headers, overlap_secrets, now=now) + + +def test_reference_receiver_rejects_tampered_expired_and_forged_callbacks(): + fixture = load_contract_fixture() + body = fixture["canonical_body"].encode("utf-8") + now = int(fixture["timestamp"]) + headers = { + "X-Relay-Timestamp": fixture["timestamp"], + "X-Relay-Signature": fixture["expected_signature"], + } + + tampered = json.loads(fixture["canonical_body"]) + tampered["event_id"] = "00000000-0000-5000-8000-000000000009" + tampered_body = canonical_callback_body(tampered) + with pytest.raises(ReceiverRejection, match="bad signature"): + reference_receiver_verify(tampered_body, headers, [fixture["signing_secret"]], now=now) + + with pytest.raises(ReceiverRejection, match="replay window"): + reference_receiver_verify(body, headers, [fixture["signing_secret"]], now=now + 301) + + forged = dict(headers, **{"X-Relay-Signature": "sha256=" + "0" * 64}) + with pytest.raises(ReceiverRejection, match="bad signature"): + reference_receiver_verify(body, forged, [fixture["signing_secret"]], now=now) + + missing = {"X-Relay-Signature": headers["X-Relay-Signature"]} + with pytest.raises(ReceiverRejection, match="malformed timestamp"): + reference_receiver_verify(body, missing, [fixture["signing_secret"]], now=now) diff --git a/mailing/tests/test_ses_webhooks.py b/mailing/tests/test_ses_webhooks.py index 8912acd..bdbfc5a 100644 --- a/mailing/tests/test_ses_webhooks.py +++ b/mailing/tests/test_ses_webhooks.py @@ -18,8 +18,8 @@ CampaignRecipient, CampaignRecipientStatus, Client, - CmpCallback, - CmpCallbackStatus, + ClientCallback, + ClientCallbackStatus, Contact, EmailEvent, EmailEventType, @@ -28,10 +28,11 @@ TransactionalMessage, TransactionalMessageStatus, ) -from mailing.services.cmp_callbacks import process_due_cmp_callbacks +from mailing.services.client_callbacks import process_due_client_callbacks from mailing.services.contacts import is_marketing_email_allowed, is_transactional_email_allowed from mailing.services.ses_webhooks import SNS_MOCK_SIGNATURE, canonical_sns_message from mailing.sqs import records_from_messages +from mailing.tests.callback_helpers import create_callback_endpoint, install_fake_opener from mailing.workers import ses_webhooks_handler pytestmark = pytest.mark.django_db @@ -262,12 +263,8 @@ def test_worker_delivery_updates_transactional_message(transactional_message): ) -@override_settings(CMP_WEBHOOK_URL="https://cmp.example.com/api/datamailer/events", CMP_WEBHOOK_TOKEN="secret") -def test_worker_delivery_emits_cmp_callback(recipient, monkeypatch): - monkeypatch.setattr( - "mailing.services.cmp_callbacks.transaction.on_commit", - lambda callback: callback(), - ) +def test_worker_campaign_delivery_emits_no_client_callback(recipient, app_client): + create_callback_endpoint(app_client) response = ses_webhooks_handler( records_from_payloads([("message-1", webhook_payload("delivery", "sns-cmp-delivery", "ses-campaign-1"))]), @@ -275,11 +272,27 @@ def test_worker_delivery_emits_cmp_callback(recipient, monkeypatch): ) assert response == {"batchItemFailures": []} - callback = CmpCallback.objects.get() - assert callback.status == CmpCallbackStatus.PENDING - assert callback.event_type == "message.delivered" - assert callback.payload["event_type"] == "message.delivered" - assert callback.payload["email"] == recipient.contact.normalized_email + # Campaign-recipient transitions stay on the CMP channel; the client + # callback contract carries transactional deliveries only. + assert ClientCallback.objects.count() == 0 + + +def test_worker_transactional_delivery_emits_client_callback(transactional_message, app_client): + create_callback_endpoint(app_client) + + response = ses_webhooks_handler( + records_from_payloads([("message-1", webhook_payload("delivery", "sns-tx-cb-delivery", "ses-tx-1"))]), + None, + ) + + assert response == {"batchItemFailures": []} + callback = ClientCallback.objects.get() + assert callback.status == ClientCallbackStatus.PENDING + assert callback.event_type == "delivery.delivered" + assert callback.payload["event_type"] == "delivery.delivered" + assert callback.payload["message_id"] == str(transactional_message.pk) + assert callback.payload["client_reference"] == "tx-1" + assert transactional_message.contact.normalized_email not in callback.body def test_worker_hard_bounce_suppresses_campaign_contact(recipient, audience, app_client): @@ -303,36 +316,36 @@ def test_worker_hard_bounce_suppresses_campaign_contact(recipient, audience, app assert EmailEvent.objects.filter(event_type=EmailEventType.BOUNCE, campaign_recipient=recipient).count() == 1 -@override_settings(CMP_WEBHOOK_URL="https://cmp.example.com/api/datamailer/events", CMP_WEBHOOK_TOKEN="secret") -def test_worker_hard_bounce_emits_cmp_callback(recipient, monkeypatch): - posts = [] - - class Response: - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, traceback): - return False - - def fake_urlopen(request, *, timeout): - posts.append( - { - "url": request.full_url, - "json": json.loads(request.data.decode("utf-8")), - "headers": dict(request.header_items()), - "timeout": timeout, - } - ) - return Response() - - monkeypatch.setattr( - "mailing.services.cmp_callbacks.urlopen", - fake_urlopen, - ) - monkeypatch.setattr( - "mailing.services.cmp_callbacks.transaction.on_commit", - lambda callback: callback(), +def test_worker_hard_bounce_emits_signed_client_callback(transactional_message, app_client, monkeypatch): + endpoint = create_callback_endpoint(app_client) + opener = install_fake_opener(monkeypatch) + payload = webhook_payload( + "bounce", + "sns-bounce-cmp", + "ses-tx-1", + metadata={"bounce_type": "Permanent", "bounce_sub_type": "General"}, ) + + ses_webhooks_handler(records_from_payloads([("message-1", payload)]), None) + assert ClientCallback.objects.filter(status=ClientCallbackStatus.PENDING).count() == 1 + process_due_client_callbacks() + + assert len(opener.requests) == 1 + request = opener.requests[0] + assert request["url"] == endpoint.url + assert "authorization" not in request["headers"] + assert request["headers"]["x-relay-signature"].startswith("sha256=") + assert request["headers"]["x-relay-event-type"] == "delivery.bounced" + body = json.loads(request["body"].decode("utf-8")) + assert body["event_type"] == "delivery.bounced" + assert body["bounce_type"] == "hard" + assert body["reason_code"] == "hard_bounce" + assert body["client_reference"] == "tx-1" + assert transactional_message.contact.normalized_email not in request["body"].decode("utf-8") + + +def test_worker_campaign_hard_bounce_emits_no_client_callback(recipient, app_client): + create_callback_endpoint(app_client) payload = webhook_payload( "bounce", "sns-bounce-cmp", @@ -341,18 +354,8 @@ def fake_urlopen(request, *, timeout): ) ses_webhooks_handler(records_from_payloads([("message-1", payload)]), None) - assert CmpCallback.objects.filter(status=CmpCallbackStatus.PENDING).count() == 1 - process_due_cmp_callbacks() - - assert len(posts) == 1 - body = posts[0]["json"] - assert posts[0]["url"] == "https://cmp.example.com/api/datamailer/events" - assert posts[0]["headers"]["Authorization"] == "Bearer secret" - assert body["event_type"] == "contact.hard_bounced" - assert body["email"] == recipient.contact.normalized_email - assert body["audience"] == recipient.campaign.audience.slug - assert body["client"] == recipient.campaign.client.slug - assert body["metadata"]["bounce_type"] == "Permanent" + + assert ClientCallback.objects.count() == 0 def test_worker_raw_sns_hard_bounce_suppresses_campaign_contact(recipient, audience, app_client): diff --git a/mailing/tests/test_tracking_unsubscribe.py b/mailing/tests/test_tracking_unsubscribe.py index 5ed4e6d..339f842 100644 --- a/mailing/tests/test_tracking_unsubscribe.py +++ b/mailing/tests/test_tracking_unsubscribe.py @@ -1,4 +1,3 @@ -import json from urllib.parse import parse_qs, urlparse import pytest @@ -12,8 +11,9 @@ CampaignRecipient, CampaignRecipientStatus, Client, + ClientCallback, + ClientCallbackStatus, CmpCallback, - CmpCallbackStatus, Contact, EmailEvent, EmailEventType, @@ -21,7 +21,6 @@ Subscription, SubscriptionStatus, ) -from mailing.services.cmp_callbacks import process_due_cmp_callbacks from mailing.services.public_urls import ( campaign_recipient_public_urls, click_redirect_url, @@ -30,6 +29,7 @@ ) from mailing.services.tokens import ensure_campaign_recipient_tokens, get_recipient_by_tracking_token, token_hash from mailing.services.tracking import TRANSPARENT_GIF +from mailing.tests.callback_helpers import create_callback_endpoint pytestmark = pytest.mark.django_db @@ -216,7 +216,8 @@ def test_click_redirect_records_repeated_clicks_and_redirects(client, recipient) @override_settings(CMP_WEBHOOK_URL="https://cmp.example.com/api/datamailer/events", CMP_WEBHOOK_TOKEN="secret") -def test_tracking_open_and_click_emit_cmp_callbacks(client, recipient, monkeypatch): +def test_tracking_open_and_click_emit_cmp_callbacks_only(client, recipient, app_client, monkeypatch): + create_callback_endpoint(app_client) monkeypatch.setattr( "mailing.services.cmp_callbacks.transaction.on_commit", lambda callback: callback(), @@ -231,10 +232,13 @@ def test_tracking_open_and_click_emit_cmp_callbacks(client, recipient, monkeypat assert open_response.status_code == 200 assert click_response.status_code == 302 + # Campaign-recipient engagement stays on the CMP channel; the client + # callback contract carries transactional deliveries only. assert list(CmpCallback.objects.order_by("id").values_list("event_type", flat=True)) == [ "message.opened", "message.clicked", ] + assert ClientCallback.objects.count() == 0 @pytest.mark.parametrize( @@ -319,31 +323,8 @@ def test_unsubscribe_post_applies_scope_idempotently_and_records_events(client, @override_settings(CMP_WEBHOOK_URL="https://cmp.example.com/api/datamailer/events", CMP_WEBHOOK_TOKEN="secret") -def test_unsubscribe_post_emits_cmp_callback(client, recipient, monkeypatch): - posts = [] - - class Response: - def __enter__(self): - return self - - def __exit__(self, exc_type, exc, traceback): - return False - - def fake_urlopen(request, *, timeout): - posts.append( - { - "url": request.full_url, - "json": json.loads(request.data.decode("utf-8")), - "headers": dict(request.header_items()), - "timeout": timeout, - } - ) - return Response() - - monkeypatch.setattr( - "mailing.services.cmp_callbacks.urlopen", - fake_urlopen, - ) +def test_unsubscribe_post_emits_cmp_callback_only(client, recipient, app_client, monkeypatch): + create_callback_endpoint(app_client) monkeypatch.setattr( "mailing.services.cmp_callbacks.transaction.on_commit", lambda callback: callback(), @@ -356,17 +337,13 @@ def fake_urlopen(request, *, timeout): ) assert response.status_code == 200 - assert CmpCallback.objects.filter(status=CmpCallbackStatus.PENDING).count() == 1 - process_due_cmp_callbacks() - assert len(posts) == 1 - body = posts[0]["json"] - assert posts[0]["url"] == "https://cmp.example.com/api/datamailer/events" - assert posts[0]["headers"]["Authorization"] == "Bearer secret" - assert body["event_type"] == "subscription.unsubscribed" - assert body["email"] == recipient.contact.normalized_email - assert body["audience"] == recipient.campaign.audience.slug - assert body["client"] == recipient.campaign.client.slug - assert body["metadata"]["scope"] == "client" + # A campaign unsubscribe link is a CMP-channel contact event; the client + # callback contract carries client-level subscription changes made + # through the subscription APIs. + callback = ClientCallback.objects.filter(status=ClientCallbackStatus.PENDING).first() + assert callback is None + cmp_body = CmpCallback.objects.order_by("-id").first() + assert cmp_body is not None and cmp_body.event_type == "subscription.unsubscribed" def test_unsubscribe_invalid_token_and_invalid_scope_do_not_mutate(client, recipient): diff --git a/mailing/tests/test_transactional_api.py b/mailing/tests/test_transactional_api.py index 23345d7..a14328b 100644 --- a/mailing/tests/test_transactional_api.py +++ b/mailing/tests/test_transactional_api.py @@ -1,17 +1,21 @@ import json import time +from datetime import timedelta import pytest from django.contrib import admin from django.test import override_settings from django.urls import reverse from django.utils import timezone +from django.utils.dateparse import parse_datetime from mailing.admin import EmailEventAdmin, EmailTemplateAdmin, TransactionalMessageAdmin from mailing.models import ( Audience, CategoryPreference, Client, + ClientCallback, + ClientCallbackStatus, CmpCallback, CmpCallbackStatus, Contact, @@ -28,7 +32,9 @@ ) from mailing.queue_contracts import validate_transactional_email_message from mailing.services.auth import create_client_api_key +from mailing.services.client_callbacks import process_due_client_callbacks from mailing.services.cmp_callbacks import process_due_cmp_callbacks +from mailing.tests.callback_helpers import create_callback_endpoint, install_fake_opener pytestmark = pytest.mark.django_db(transaction=True) @@ -83,6 +89,18 @@ def template(api_client_record): ) +@pytest.fixture +def contact(): + return Contact.objects.create(email="reconcile@example.com") + + +@pytest.fixture +def other_client_record(organization): + other = Client.objects.create(organization=organization, name="Other Product", slug="other-product") + create_client_api_key(client=other, name="Other", raw_api_key="other-client-key") + return other + + def auth_headers(raw_key=API_KEY): return {"HTTP_AUTHORIZATION": f"Bearer {raw_key}"} @@ -817,7 +835,9 @@ def test_transactional_send_to_recipient_list_creates_per_member_messages( template, monkeypatch, ): + create_callback_endpoint(api_client_record) enqueued = [] + opener = install_fake_opener(monkeypatch) posts = collect_cmp_callbacks(monkeypatch) monkeypatch.setattr("mailing.services.transactional.enqueue_transactional_email", enqueued.append) allowed_contact = Contact.objects.create(email="allowed@example.com") @@ -888,21 +908,32 @@ def test_transactional_send_to_recipient_list_creates_per_member_messages( assert messages[1].last_error == "hard_bounce" assert len(enqueued) == 0 assert CmpCallback.objects.filter(status=CmpCallbackStatus.PENDING).count() == 2 + assert ClientCallback.objects.filter(status=ClientCallbackStatus.PENDING).count() == 2 process_due_cmp_callbacks() + process_due_client_callbacks() assert len(posts) == 2 - reasons_by_email = { - post["json"]["email"]: post["json"]["metadata"]["reason"] - for post in posts - } + assert len(opener.requests) == 2 + assert {post["json"]["event_type"] for post in posts} == {"transactional.skipped"} assert posts[0]["url"] == "https://cmp.example.com/api/datamailer/events" assert posts[0]["headers"]["Authorization"] == "Bearer secret" - assert {post["json"]["event_type"] for post in posts} == {"transactional.skipped"} assert {post["json"]["audience"] for post in posts} == {audience.slug} assert {post["json"]["client"] for post in posts} == {api_client_record.slug} + reasons_by_email = {post["json"]["email"]: post["json"]["metadata"]["reason"] for post in posts} assert reasons_by_email == { "allowed@example.com": "category_unsubscribe", "suppressed@example.com": "hard_bounce", } + bodies = [json.loads(request["body"].decode("utf-8")) for request in opener.requests] + assert {body["event_type"] for body in bodies} == {"delivery.suppressed"} + assert {body["reason_code"] for body in bodies} == {"category_unsubscribe", "hard_bounce"} + assert {body["template_key"] for body in bodies} == {template.key} + assert {body["client_reference"] for body in bodies} == { + messages[0].idempotency_key, + messages[1].idempotency_key, + } + joined = opener.requests[0]["body"].decode("utf-8") + opener.requests[1]["body"].decode("utf-8") + assert "allowed@example.com" not in joined + assert "suppressed@example.com" not in joined replay = post_recipient_list_transactional(client, recipient_list.key, payload) @@ -1744,3 +1775,145 @@ def test_transactional_models_are_available_in_admin(): assert isinstance(admin.site._registry[EmailTemplate], EmailTemplateAdmin) assert isinstance(admin.site._registry[TransactionalMessage], TransactionalMessageAdmin) assert isinstance(admin.site._registry[EmailEvent], EmailEventAdmin) + + +# --- Reconciliation: GET /api/transactional/messages?since= ------------------ + + +def get_reconciliation(django_client, since, raw_key=API_KEY): + return django_client.get( + reverse("mailing:api_transactional_messages_reconcile"), + data={"since": since} if since is not None else {}, + **auth_headers(raw_key), + ) + + +def seed_reconciliation_message(client_record, contact, template, *, key, status, **fields): + return TransactionalMessage.objects.create( + client=client_record, + contact=contact, + email=contact.email, + template=template, + template_key=template.key, + idempotency_key=key, + status=status, + **fields, + ) + + +def test_reconciliation_requires_auth(client): + response = client.get(reverse("mailing:api_transactional_messages_reconcile"), data={"since": "2026-01-01T00:00:00+00:00"}) + assert response.status_code == 401 + + +def test_reconciliation_requires_since(api_client_record, client): + response = get_reconciliation(client, None) + assert response.status_code == 400 + assert response.json()["error"]["fields"] == {"since": "required"} + + +def test_reconciliation_rejects_unparseable_since(api_client_record, client): + response = get_reconciliation(client, "not-a-date") + assert response.status_code == 400 + assert response.json()["error"]["fields"] == {"since": "must_be_iso_datetime"} + + +def test_reconciliation_reports_only_this_clients_recent_messages( + api_client_record, + other_client_record, + client, + contact, + template, +): + now = timezone.now() + # updated_at is an auto_now field: backdate through a queryset update. + bounced = seed_reconciliation_message( + api_client_record, + contact, + template, + key="reconcile-bounced", + status=TransactionalMessageStatus.BOUNCED, + ) + TransactionalMessage.objects.filter(pk=bounced.pk).update(updated_at=now - timedelta(minutes=3)) + old_message = seed_reconciliation_message( + api_client_record, + contact, + template, + key="reconcile-old", + status=TransactionalMessageStatus.SENT, + ) + TransactionalMessage.objects.filter(pk=old_message.pk).update(updated_at=now - timedelta(days=2)) + seed_reconciliation_message( + api_client_record, + contact, + template, + key="", + status=TransactionalMessageStatus.SENT, + ) + seed_reconciliation_message( + other_client_record, + contact, + template, + key="reconcile-other-client", + status=TransactionalMessageStatus.SENT, + ) + + response = get_reconciliation(client, (now - timedelta(hours=1)).isoformat()) + + assert response.status_code == 200 + rows = response.json()["messages"] + assert [row["id"] for row in rows] == [str(bounced.pk)] + assert rows[0]["client_reference"] == "reconcile-bounced" + assert rows[0]["status"] == "bounced" + assert rows[0]["template_key"] == template.key + assert rows[0]["template_version"] == 1 + assert rows[0]["reason_code"] == "hard_bounce" + assert parse_datetime(rows[0]["updated_at"]) is not None + + +def test_reconciliation_maps_statuses_for_the_package_vocabulary( + api_client_record, + client, + contact, + template, +): + now = timezone.now() + cases = { + "reconcile-queued": (TransactionalMessageStatus.QUEUED, "queued", ""), + "reconcile-sending": (TransactionalMessageStatus.SENDING, "retrying", ""), + "reconcile-sent": (TransactionalMessageStatus.SENT, "sent", ""), + "reconcile-delivered": (TransactionalMessageStatus.SENT, "delivered", ""), + "reconcile-skipped": (TransactionalMessageStatus.SKIPPED, "suppressed", "category_unsubscribe"), + "reconcile-skipped-raw": (TransactionalMessageStatus.SKIPPED, "suppressed", "suppressed"), + "reconcile-failed": (TransactionalMessageStatus.FAILED, "failed", "send_failed"), + "reconcile-complained": (TransactionalMessageStatus.COMPLAINED, "complained", "complaint"), + "reconcile-soft": (TransactionalMessageStatus.SENT, "sent", "soft_bounce"), + } + for index, (key, (status, _, _)) in enumerate(cases.items()): + fields = {} + if key == "reconcile-delivered": + fields["delivered_at"] = now + if key == "reconcile-skipped": + fields["metadata"] = {"reason": "category_unsubscribe"} + if key == "reconcile-skipped-raw": + fields["metadata"] = {"reason": "some raw provider note"} + if key == "reconcile-soft": + fields["last_error"] = "soft_bounce" + seed_reconciliation_message( + api_client_record, + contact, + template, + key=key, + status=status, + updated_at=now - timedelta(minutes=index + 1), + **fields, + ) + + response = get_reconciliation(client, (now - timedelta(hours=1)).isoformat()) + + assert response.status_code == 200 + rows = {row["client_reference"]: row for row in response.json()["messages"]} + assert set(rows) == set(cases) + for key, (_, expected_status, expected_reason) in cases.items(): + assert rows[key]["status"] == expected_status, key + assert rows[key]["reason_code"] == expected_reason, key diff --git a/mailing/tests/test_transactional_sender.py b/mailing/tests/test_transactional_sender.py index 165f10b..f00ddee 100644 --- a/mailing/tests/test_transactional_sender.py +++ b/mailing/tests/test_transactional_sender.py @@ -9,6 +9,7 @@ from mailing.models import ( Client, + ClientCallback, CmpCallback, CmpCallbackStatus, Contact, @@ -22,6 +23,7 @@ from mailing.services.cmp_callbacks import process_due_cmp_callbacks from mailing.services.transactional import build_transactional_queue_payload from mailing.sqs import records_from_messages +from mailing.tests.callback_helpers import create_callback_endpoint from mailing.workers import transactional_email_handler pytestmark = pytest.mark.django_db(transaction=True) @@ -399,6 +401,7 @@ def fail_if_called(): @override_settings(CMP_WEBHOOK_URL="https://cmp.example.com/api/datamailer/events", CMP_WEBHOOK_TOKEN="secret") def test_permanent_ses_failure_marks_failed_and_acknowledges(transactional_message, monkeypatch): posts = collect_cmp_callbacks(monkeypatch) + create_callback_endpoint(transactional_message.client) class PermanentSesClient: def send_email(self, **params): @@ -420,6 +423,8 @@ def send_email(self, **params): assert transactional_message.ses_message_id == "" assert transactional_message.last_error == "MessageRejected: Address rejected" assert event.metadata["reason"] == "ses_permanent_failure" + # A permanently failed send is CMP-channel only; the client callback + # contract announces it through reconciliation instead. assert CmpCallback.objects.filter(status=CmpCallbackStatus.PENDING).count() == 1 process_due_cmp_callbacks() assert len(posts) == 1 @@ -429,6 +434,7 @@ def send_email(self, **params): assert posts[0]["json"]["email"] == transactional_message.email assert posts[0]["json"]["client"] == transactional_message.client.slug assert posts[0]["json"]["metadata"]["reason"] == "ses_permanent_failure" + assert ClientCallback.objects.count() == 0 def test_mixed_batch_retries_only_invalid_and_transient_records(transactional_message, monkeypatch): diff --git a/mailing/tests/test_worker_status.py b/mailing/tests/test_worker_status.py index 9a2fc6e..032f8cd 100644 --- a/mailing/tests/test_worker_status.py +++ b/mailing/tests/test_worker_status.py @@ -7,11 +7,14 @@ from mailing.models import ( Audience, + CallbackEndpoint, Campaign, CampaignRecipient, CampaignRecipientStatus, CampaignStatus, Client, + ClientCallback, + ClientCallbackStatus, CmpCallback, CmpCallbackStatus, Contact, @@ -61,8 +64,12 @@ def test_sandbox_worker_statuses_include_systemd_state_and_local_backlog(setting assert statuses["cmp-callbacks"].badge_label == "Failed" assert statuses["cmp-callbacks"].badge_tone == "danger" assert statuses["cmp-callbacks"].detail == "failed; result=exit-code" + assert statuses["client-callbacks"].badge_label == "Failed" + assert statuses["client-callbacks"].badge_tone == "danger" + assert statuses["client-callbacks"].detail == "failed; result=exit-code" assert statuses["ses-webhooks"].backlog_count is None assert statuses["cmp-callbacks"].backlog_count == 1 + assert statuses["client-callbacks"].backlog_count == 1 assert statuses["recipient-list-imports"].backlog_count == 1 @@ -85,6 +92,9 @@ def test_worker_status_api_returns_staff_only_json_status(client, settings, monk assert workers["cmp-callbacks"]["alive"] is False assert workers["cmp-callbacks"]["status"] == "failed" assert workers["cmp-callbacks"]["detail"] == "failed; result=exit-code" + assert workers["client-callbacks"]["alive"] is False + assert workers["client-callbacks"]["status"] == "failed" + assert workers["client-callbacks"]["detail"] == "failed; result=exit-code" assert workers["ses-webhooks"]["backlog"] == {"label": "SQS backlog", "count": None} assert workers["recipient-list-imports"]["backlog"] == {"label": "Pending import jobs", "count": 1} @@ -98,7 +108,7 @@ def test_worker_status_api_requires_staff(client): def _fake_systemd_run(args, **kwargs): service_name = args[2] - if service_name == "relay-cmp-callbacks-worker.service": + if service_name in ("relay-cmp-callbacks-worker.service", "relay-client-callbacks-worker.service"): stdout = "\n".join( [ "LoadState=loaded", @@ -163,6 +173,11 @@ def _create_worker_backlog(): audience=audience, event_type=EmailEventType.UNSUBSCRIBE, ) + endpoint = CallbackEndpoint.objects.create( + client=client, + url="https://callback.example.com/hooks", + signing_secret="callback-signing-secret", + ) CmpCallback.objects.create( email_event=event, contact=contact, @@ -175,6 +190,18 @@ def _create_worker_backlog(): status=CmpCallbackStatus.PENDING, next_attempt_at=timezone.now(), ) + ClientCallback.objects.create( + email_event=event, + client=client, + endpoint=endpoint, + event_id="00000000-0000-5000-8000-000000000001", + event_type="subscription.changed", + payload={}, + body="{}", + body_hash="0" * 64, + status=ClientCallbackStatus.PENDING, + next_attempt_at=timezone.now(), + ) RecipientListImportJob.objects.create( client=client, audience=audience, diff --git a/mailing/urls.py b/mailing/urls.py index 0023c79..26cbc6e 100644 --- a/mailing/urls.py +++ b/mailing/urls.py @@ -178,6 +178,11 @@ name="api_transactional_template_test_send", ), path("api/transactional/send", views.api_transactional_send, name="api_transactional_send"), + path( + "api/transactional/messages", + views.api_transactional_messages_reconcile, + name="api_transactional_messages_reconcile", + ), path( "api/transactional/messages/", views.api_transactional_message_status, diff --git a/mailing/views.py b/mailing/views.py index 72489ae..6266718 100644 --- a/mailing/views.py +++ b/mailing/views.py @@ -26,10 +26,12 @@ ) from mailing.models import ( Audience, + CallbackEndpoint, Campaign, CampaignStatus, Client, ClientApiKey, + ClientCallback, CmpCallback, EmailEvent, EmailEventType, @@ -50,6 +52,7 @@ get_contact_preferences_for_client, get_contact_status_for_client, get_transactional_message_status_for_client, + get_transactional_messages_since_for_client, get_transactional_template_for_client, preview_campaign_for_client, queue_campaign_for_client, @@ -342,7 +345,10 @@ def transactional_message_detail(request, message_id): {"event": event, "context": event_context(event), "metadata_summary": metadata_summary(event.metadata)} for event in events ] - callback_rows = CmpCallback.objects.filter( + callback_rows = ClientCallback.objects.filter( + transactional_message=message, + ).order_by("sequence") + cmp_callback_rows = CmpCallback.objects.filter( email_event__transactional_message=message, ).order_by("-created_at", "-id") return render( @@ -353,6 +359,7 @@ def transactional_message_detail(request, message_id): "badge": Badge(message.get_status_display(), delivery_tone(message.status)), "event_rows": event_rows, "callback_rows": callback_rows, + "cmp_callback_rows": cmp_callback_rows, "metadata_summary": metadata_summary(message.metadata), }, ) @@ -856,6 +863,11 @@ def client_detail(request, client_id): raw_api_key_context = request.session.pop("operator_raw_api_key") key_form = ClientApiKeyForm(client=client) api_keys = client_api_keys_for_detail(client) + client_callbacks = ( + ClientCallback.objects.filter(client=client) + .select_related("endpoint") + .order_by("-created_at", "-id")[:10] + ) cmp_callbacks = ( CmpCallback.objects.filter(client=client) .select_related("contact", "email_event") @@ -876,6 +888,8 @@ def client_detail(request, client_id): "revoked_key_count": sum(1 for api_key in api_keys if api_key.revoked_at is not None), "key_form": key_form, "raw_api_key_context": raw_api_key_context, + "client_callbacks": client_callbacks, + "callback_endpoint": CallbackEndpoint.objects.filter(client=client).first(), "cmp_callbacks": cmp_callbacks, "mailchimp_status": mailchimp_status_payload(client), "mailchimp_syncs": mailchimp_syncs, @@ -1417,6 +1431,22 @@ def api_transactional_message_status(request, message_id): return JsonResponse(payload, status=200) +def api_transactional_messages_reconcile(request): + if request.method != "GET": + return method_not_allowed_response(["GET"]) + + client, error_response = authenticate_api_request(request) + if error_response: + return error_response + + try: + payload = get_transactional_messages_since_for_client(request.GET.get("since"), client) + except ApiValidationError as exc: + return validation_error_response(exc) + + return JsonResponse(payload, status=200) + + @csrf_exempt def api_recipient_list(request, list_key): if request.method not in {"GET", "PUT"}: diff --git a/relay/settings.py b/relay/settings.py index fc79b04..c65b716 100644 --- a/relay/settings.py +++ b/relay/settings.py @@ -259,6 +259,10 @@ def float_env(name, *, default): CMP_WEBHOOK_URL = os.environ.get("CMP_WEBHOOK_URL", "").strip() CMP_WEBHOOK_TOKEN = os.environ.get("CMP_WEBHOOK_TOKEN", "") CMP_WEBHOOK_TIMEOUT_SECONDS = float_env("CMP_WEBHOOK_TIMEOUT_SECONDS", default=3.0) +CLIENT_CALLBACK_TIMEOUT_SECONDS = float_env("CLIENT_CALLBACK_TIMEOUT_SECONDS", default=3.0) +CLIENT_CALLBACK_MAX_ATTEMPTS = int(float_env("CLIENT_CALLBACK_MAX_ATTEMPTS", default=8)) +CLIENT_CALLBACK_RETRY_BASE_SECONDS = float_env("CLIENT_CALLBACK_RETRY_BASE_SECONDS", default=60.0) +CLIENT_CALLBACK_RETRY_MAX_DELAY_SECONDS = float_env("CLIENT_CALLBACK_RETRY_MAX_DELAY_SECONDS", default=21600.0) MAILCHIMP_TIMEOUT_SECONDS = float_env("MAILCHIMP_TIMEOUT_SECONDS", default=5.0) SES_WEBHOOKS_SIGNATURE_MODE = os.environ.get( "SES_WEBHOOKS_SIGNATURE_MODE", diff --git a/scripts/deploy_relay_sandbox.sh b/scripts/deploy_relay_sandbox.sh index 7c52a9c..fb56e91 100755 --- a/scripts/deploy_relay_sandbox.sh +++ b/scripts/deploy_relay_sandbox.sh @@ -245,6 +245,7 @@ set_memory_args() { relay-scheduler) memory_args=(--memory 96m) ;; relay-ses-ingress) memory_args=(--memory 128m) ;; relay-cmp-callbacks) memory_args=(--memory 96m) ;; + relay-client-callbacks) memory_args=(--memory 96m) ;; relay-recipient-imports) memory_args=(--memory 128m) ;; esac } @@ -333,6 +334,9 @@ fi set_memory_args relay-cmp-callbacks replace_container relay-cmp-callbacks "${app_container_args[@]}" \ python manage.py process_cmp_callbacks --batch-size 25 --idle-sleep 5 +set_memory_args relay-client-callbacks +replace_container relay-client-callbacks "${app_container_args[@]}" \ + python manage.py process_client_callbacks --batch-size 25 --idle-sleep 5 set_memory_args relay-recipient-imports replace_container relay-recipient-imports "${app_container_args[@]}" \ python manage.py process_recipient_list_imports --batch-size 10 --idle-sleep 5 @@ -356,6 +360,7 @@ if [[ "$environment" == sandbox ]]; then relay-ses-ingress relay-inbound-ingress relay-cmp-callbacks + relay-client-callbacks relay-recipient-imports relay-caddy ) @@ -366,6 +371,7 @@ else relay-scheduler relay-ses-ingress relay-cmp-callbacks + relay-client-callbacks relay-recipient-imports ) fi diff --git a/templates/mailing/operator/client_detail.html b/templates/mailing/operator/client_detail.html index 312170c..ceaded7 100644 --- a/templates/mailing/operator/client_detail.html +++ b/templates/mailing/operator/client_detail.html @@ -78,6 +78,7 @@

Integration summary

Slug{{ client.slug }}
Default sender ID{{ client.default_sender_id|default:"-" }}
CMP webhook URL{{ client.cmp_webhook_url|default:"Global/default" }}
+
Callback endpoint{% if callback_endpoint %}{{ callback_endpoint.url }}{% if not callback_endpoint.enabled %} (disabled){% endif %}{% else %}-{% endif %}
Configured senders {% if client.sender_emails %} @@ -141,6 +142,57 @@

CMP callbacks

{% endif %} +
+
+
+

Client callbacks

+

Recent signed callback delivery attempts for this client.

+
+
+ {% if client_callbacks %} +
+ + + + + + + + + + + + + + {% for callback in client_callbacks %} + + + + + + + + + + {% endfor %} + +
EventContactStatusAttemptsNext retryDeliveredError
{{ callback.event_type }} + {% if callback.transactional_message_id %} + message {{ callback.transactional_message_id }} + {% elif callback.campaign_recipient_id %} + recipient {{ callback.campaign_recipient_id }} + {% else %} + - + {% endif %} + {{ callback.get_status_display }}{{ callback.attempt_count }}/{{ callback.max_attempts }}{{ callback.next_attempt_at|date:"Y-m-d H:i"|default:"-" }}{{ callback.delivered_at|date:"Y-m-d H:i"|default:"-" }}{{ callback.last_error|default:"-" }}
+
+ {% else %} +
+ No client callbacks recorded. +
Delivery, suppression, unsubscribe, and transactional failure callbacks will appear here.
+
+ {% endif %} +
diff --git a/templates/mailing/operator/transactional_message_detail.html b/templates/mailing/operator/transactional_message_detail.html index e8520f5..663770d 100644 --- a/templates/mailing/operator/transactional_message_detail.html +++ b/templates/mailing/operator/transactional_message_detail.html @@ -126,6 +126,45 @@

Events

CMP Callbacks

+ {% if cmp_callback_rows %} +
+ + + + + + + + + + + + + {% for callback in cmp_callback_rows %} + + + + + + + + + {% endfor %} + +
EventStatusAttemptsNext retryDeliveredError
{{ callback.event_type }}{{ callback.get_status_display }}{{ callback.attempt_count }}/{{ callback.max_attempts }}{{ callback.next_attempt_at|date:"Y-m-d H:i"|default:"-" }}{{ callback.delivered_at|date:"Y-m-d H:i"|default:"-" }}{{ callback.last_error|default:"-" }}
+
+ {% else %} +
+ No CMP callbacks queued. +
Callbacks for transactional skipped and failed events will appear here.
+
+ {% endif %} +
+ +
+
+

Client Callbacks

+
{% if callback_rows %}
@@ -155,7 +194,7 @@

CMP Callbacks

{% else %}
- No CMP callbacks queued. + No client callbacks queued.
Callbacks for transactional skipped and failed events will appear here.
{% endif %} diff --git a/tests/fixtures/client_callback_contract_v1.json b/tests/fixtures/client_callback_contract_v1.json new file mode 100644 index 0000000..96c575d --- /dev/null +++ b/tests/fixtures/client_callback_contract_v1.json @@ -0,0 +1,21 @@ +{ + "contract": "relay-client-callback", + "contract_version": 1, + "note": "Deterministic verification fixture for Relay client callbacks. canonical_body is the exact bytes Relay signs and posts; expected_signature is HMAC-SHA256 over '.' with signing_secret.", + "signing_secret": "fixture-callback-signing-secret", + "rotated_signing_secret": "fixture-rotated-callback-secret", + "timestamp": "1788868800", + "canonical_body": "{\"client_reference\":\"registration-user-123\",\"contract_version\":1,\"event_id\":\"dd8b8f17-6d55-5094-ae45-83cd8ac37b96\",\"event_type\":\"delivery.delivered\",\"message_id\":\"1042\",\"sequence\":3,\"template_key\":\"registration-welcome\",\"timestamp\":\"2026-09-08T12:00:00+00:00\"}", + "expected_signature": "sha256=b5bafe45177dc8254cfd1aa44d196478126e1f34137eaa213e9a783996db315e", + "rotated_expected_signature": "sha256=738904b33e67b96277205c68d5d17cba424dd8a6575c257b8e741d24ece32324", + "event": { + "contract_version": 1, + "event_id": "dd8b8f17-6d55-5094-ae45-83cd8ac37b96", + "event_type": "delivery.delivered", + "timestamp": "2026-09-08T12:00:00+00:00", + "sequence": 3, + "message_id": "1042", + "client_reference": "registration-user-123", + "template_key": "registration-welcome" + } +}