Merge feature/phase-a-templates (with develop) into feature/phase-c-tracking

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
fix: review findings on tracking, and a template API for other qbits
2026-10-14 07:21:37 -05:00 · 2026-10-13 23:00:26 -05:00 · 2026-10-13 22:28:42 -05:00 · 2026-10-13 22:14:50 -05:00 · 2026-10-13 21:37:12 -05:00
46 changed files with 4859 additions and 90 deletions
@@ -6,6 +6,10 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and
 ## [Unreleased]

 ### Added
+- Phase (c): authenticated SNS, Postmark, SendGrid and Mailgun webhooks, durable inbox deduplication and Mailgun nonce replay protection; configurable Javalin routes.
+- Atomic scheduled event processing, per-recipient event history, monotonic delivery state, category-aware suppression, optional daily complaint alerts and FAILED inbox Reprocess with Review/Result.
+- One-click unsubscribe and escaped self-contained light/dark confirmation pages using the existing envelope-bound signer; account/security tokens refused.
+- Event-based home totals, PostgreSQL/H2 tracking DDL, provider setup instructions, test-only SNS certificate/key and real HTTP, crypto, parser, concurrency, retry and privacy tests. Preferences stays deferred until core #1162 reaches an RC.
 - Phase (a), part 2: versioned Markdown/HTML/MJML templates, typed variable schemas, escaped blocks/loops/fallbacks, branded light/dark layout and contrast validation.
 - New Version, Activate Version and version-bound Approve For PHI actions, edit permissions, immutable history and parent-scoped version selectors.
 - Sender-scoped record variables, classification stand-in, HMAC snapshot tokens, AES-GCM bodies, encrypted query displays and protected resend.
@@ -4,9 +4,9 @@ One email service for a QQQ system: a durable outbox (`email_message`) with retr
 letters, idempotency keys, rate limits and provider health; SMTP, Amazon SES, Postmark,
 SendGrid, Mailgun and Gmail SMTP transports; versioned Markdown, HTML and MJML templates with
 dark-mode-safe branding; classified variables and PHI protection; preview widgets; a
-real-data home; and the core messaging bridge. Every send goes through the outbox.
+real-data home; delivery events, suppression and one-click unsubscribe; and the core messaging bridge. Every send goes through the outbox.

-The [design](docs/DESIGN.md) also covers delivery tracking, unsubscribe, Preferences, the
+The [design](docs/DESIGN.md) also specifies the deferred Preferences UI (core #1162),
 consumer migrations and inbound mail. Preview uses the standard widget endpoint; the
 dedicated preview API arrives with QQQ RC.6. Attachments are refused until
 qbit-object-storage is wired in.
@@ -45,6 +45,8 @@ in a static variable.
 |---|---|---|
 | `email_message` | Messages | Qbit only; people use record actions |
 | `email_message_attempt` | Attempts on a message | Qbit only |
+| `email_event` | Events on a message | Qbit only; readable with `email.view` |
+| `email_webhook_inbox` | Inbox on a provider | Qbit only; readable with `email.view`; Reprocess requires `email.manage` |
 | `email_template` | Templates | `email.manage`; template edit grants also apply |
 | `email_template_version` | Versions on a template | Draft edits only; lifecycle uses record actions |
 | `email_brand` | Brands | `email.manage`; template edit grants also apply |
@@ -68,7 +70,7 @@ insert/update/delete capabilities. Labels contain no status decorations.
 ## Providers and credentials

 See the [provider setup guide](docs/PROVIDERS.md) for row fields, DNS/domain setup,
-limits, Gmail app passwords and the webhook configuration planned for phase (c).
+limits, Gmail app passwords and provider webhook setup.

 | Kind | Transport | Status |
 |---|---|---|
@@ -127,8 +129,8 @@ names and STARTTLS. GMAIL_API is refused with
 HTTP transports use fixed HTTPS endpoints, 10-second connect and 30-second request
 timeouts, and no redirects. HTTP 429/5xx retry with optional Retry-After; account
 and credential failures pause the provider. Tracking is explicitly controlled by
-`trackOpens` / `trackClicks`, off by default. Outbound tracking settings do not
-install the event webhook receiver; that arrives in phase (c).
+`trackOpens` / `trackClicks`, off by default. The producer registers the phase-(c)
+receiver idempotently when the provided Javalin middleware is present.

 ## Sending

@@ -241,6 +243,35 @@ sanitizer removes iframes and style blocks, so this preview is not an exact mail
 emulator. Draft MJML is compiled in memory for preview when a compiler is configured;
 preview never stores the result or changes the version.

+### Templates from another qbit
+
+A qbit that sends its own mail ships the template with its code instead of asking an
+administrator to create it. Describe it with `EmailTemplateDefinition` and call
+`EmailTemplates.ensure` at startup or before first use:
+
+```java
+boolean created = EmailTemplates.ensure(new EmailTemplateDefinition()
+   .withName("order_ready")
+   .withLabel("Order ready")
+   .withCategory(EmailCategory.TRANSACTIONAL)
+   .withSubject("Order {{order_number}} is ready")
+   .withBody("Hi {{customer_name}}, order {{order_number}} is ready for pickup.")
+   .withVariable("order_number", "Order number", "STRING", true, "PUBLIC")
+   .withVariable("customer_name", "Customer name", "STRING", true, null)
+   .withSampleVariables("{\"order_number\":\"A-1\",\"customer_name\":\"Pat\"}")
+   .withOwnerQbit("orders")
+   .withEditPermissionName("orders.edit"));
+```
+
+`kind` defaults to MARKDOWN (MJML and HTML are also accepted), `withTextBody` is
+optional, and `withVariablesSchema(json)` replaces `withVariable` when you already have
+the JSON. When no template of that name exists, `ensure` validates the definition like
+Activate Version (syntax, schema, a render with the samples, MJML compile), creates the
+template and an ACTIVE version 1 as the qbit, and returns true. When one exists it does
+nothing and returns false, so people's edits and later versions are never overwritten;
+the unique name settles concurrent callers. An invalid definition throws
+`IllegalArgumentException` with plain reasons and writes nothing.
+
 ## PHI and PII

 `withRecordVariable` reads through `QueryAction` in the sender's QContext; it does not
@@ -279,15 +310,17 @@ and content. Keys resolve through `secretResolver`; no secret values are stored.

 NOTIFICATION/MARKETING templates prepare footer links and headers using
 `QQQ_EMAIL_UNSUBSCRIBE_KEY` and `publicBaseUrl`; marketing refuses without them.
-Tokens bind the message, recipient-envelope hash and category. The phase-(c)
-unsubscribe route and Preferences are still pending: signed links are prepared here,
-but this qbit alone does not yet serve the unsubscribe destination. Do not deploy
-opt-out mail without that receiving handler.
+Tokens bind the message, recipient-envelope hash and category. The receiving route
+is implemented: GET only shows a light/dark confirmation page; explicit POST upserts
+category suppression and records UNSUBSCRIBED. Invalid or account/security tokens
+return 404. The legacy token covers every To/Cc/Bcc recipient, so use one mailbox
+per opt-out message for individual preferences. Preferences UI remains deferred
+until core #1162 ships in an RC.

 The home uses viewer-scoped counts and queries, UTC hour/day charts, provider health
 (admins), and 20 recent messages with masked recipients. No sample data or sampled
-counts are used. Bounce/complaint totals use dedicated event timestamps, populated
-by the future webhook phase or host integrations.
+counts are used. Delivered/bounced/complained totals count recipient events and
+use their occurred dates; sent/dead-letter totals count messages.

 ## Delivery and actions

@@ -331,12 +364,49 @@ Providers offer Send Test Email with a recipient input and a reported outcome.
 record bindings use `{"patient":{"table":"patient","id":123}}` in
 `recordVariablesJson`. Output includes `messageId`, `status`, and `reasons`.

-Hidden schedules are `email_dispatch` (15 s by default), `email_maintain` (300 s, lease
+Hidden schedules include `email_process_webhooks` (30 s), `email_dispatch` (15 s by default), `email_maintain` (300 s, lease
 recovery including disabled providers, and health probes), and `email_retention`
 (daily). Retention clears final-message bodies after 30 days, preserving `bodySha256`;
 message deletion is disabled by default. `finishedDate` records finalization independently
 of enqueue and later retention updates.

+## Delivery tracking
+
+Apply the new `email_webhook_inbox` and `email_event` DDL plus indexes for an existing
+phase-(a)/(b) database. Configure the authenticated HTTPS endpoints using
+[the provider guide](docs/PROVIDERS.md#event-webhooks). Verification supports SNS
+RSA v1/v2 with pinned topic and AWS certificate hosts, Postmark Basic auth,
+SendGrid ECDSA and Mailgun HMAC with durable nonce replay protection.
+
+The 1 MiB receiver persists before acknowledging. DB transactions serialize
+processing across replicas and commit event rows, message state and suppression
+atomically. Events in a payload are applied independently: matched events commit at
+once, and only unmatched events (or events whose message is still SENDING) retry the
+row with configured backoff. After six attempts they are skipped and the row finishes
+PROCESSED with a note ("N events did not match a message"). Rows that cannot be parsed
+or written fail after six attempts; **Reprocess** on FAILED inbox rows uses the
+standard Review/Result flow. An event whose recipient is not on the matched message
+is ignored, and an SES bounce or complaint without its recipient list is skipped.
+SNS notifications older than an hour or more than five minutes ahead are refused, and
+certificate fetch failures are cached for five minutes per URL. Disabled tracking is
+ignored. Per-event recipient keys deduplicate
+overlapping batches. All new service state is in the DB; certificate caches are in
+memory. No runtime filesystem or new dependencies are used.
+
+Hard bounce/complaint suppress ALL; provider unsubscribe suppresses the message's
+opt-out category. Message status never regresses from COMPLAINED to BOUNCED or
+DELIVERED. Optional `email.complaint` alerts warn after more than three complaints
+in 24 hours, with daily deduplication. Provider details mask addresses and are capped
+at 500 characters; raw authenticated inbox payloads are sensitive DB data.
+
+NOTIFICATION and MARKETING messages with unsubscribe configured require exactly one
+To recipient and no Cc/Bcc; send one message per recipient. The token binds that
+mailbox, and an unsubscribe POST suppresses only that address for the message's
+category. Multi-recipient requests are refused before suppression filtering.
+TRANSACTIONAL and SECURITY messages, and NOTIFICATION messages without unsubscribe
+configured, may still have multiple recipients. Existing single-recipient links
+remain valid; old multi-recipient links are rejected.
+
 ## Core messaging bridge

 The producer adds messaging provider `qbitEmail` (configurable), type `QBIT_EMAIL`.
@@ -375,6 +445,7 @@ All settings have fluent `withX` setters and getters.
 | `backendName` | Required existing backend |
 | `schedulerName` | Optional existing scheduler; required for automatic dispatch |
 | `publicBaseUrl` | Public app URL for signed unsubscribe links |
+| `webhookBasePath` | `/qqq/hooks/email`; absolute path outside `/qqq/v1` |
 | `mjmlCompiler` | Optional `MjmlCompilerInterface`; required for MJML activation |
 | `classificationOverrides` | Map of `table.field` to `EmailClassifications.Level` for RC.5 |
 | `defaultProviderName` | Optional named provider override; otherwise enabled default row |
@@ -1,6 +1,6 @@
 # qbit-email design

-**Status:** phase (a) implemented; later phases remain specification. Phases (a) to (d) build outbound; phase (e) is inbound (section 12, design only).
+**Status:** phases (a), (b) and (c) implemented, except Preferences (8.3), deferred until core #1162 ships in an RC. Phase (d) consumer migrations and phase (e) inbound remain specification.
 **Brief:** owner decisions of 2026-10-13 (providers, outbound first, both template kinds).
 **Owner decisions (2026-10-12): all recommendations accepted.** Every "Recommended" choice and
 every decision below stands; it is adjusted once it can be seen working.
@@ -205,7 +205,9 @@ error) and `email_event` (section 6).
  content), Cancel (queued or waiting), Send Now (waiting). Standard streamed-ETL Review and Result
  screens.
 - **Retention:** a daily job clears bodies after `bodyRetentionDays` and deletes final messages
-  older than `messageRetentionDays` (0 = keep forever; kof22 keeps).
+  older than `messageRetentionDays` (0 = keep forever; kof22 keeps). NOTIFICATION and MARKETING
+  messages are never deleted, because their unsubscribe links resolve through the message row
+  (8.2); at `messageRetentionDays` retention only clears their bodies and variable snapshot.

 ## 5. Templates and branding

@@ -283,11 +285,10 @@ id, or unsaved source, plus variables) returns subject, html, text and warnings.
  token snapshot, linking to the original message for the token binding.
 - The home follows the newer Alerts pattern: standard QQQ chart widgets plus inline,
  theme-neutral cards. All queries/counts use the viewer's session. Bounce/complaint
-  dates have dedicated columns for phase-(c) events; no dates or metrics are invented.
- Part 2 signs opt-out links and prepares headers; the receiving unsubscribe handler
-  remains phase (c). Without its key/base URL, notification omits opt-out links and
+  dates are populated by phase-(c) events; home totals count those recipient events.
+- Part 2 signs opt-out links and prepares headers; phase (c) now serves the receiver. Without its key/base URL, notification omits opt-out links and
  headers, while marketing refuses as specified in 8.2. Do not use opt-out delivery
-  without a receiving handler. The token's address hash covers the recipient envelope.
+  without a receiving handler. The token's address hash binds the single recipient mailbox.

 ## 6. Delivery tracking

@@ -312,6 +313,32 @@ DEFERRED, BOUNCED, COMPLAINED, OPENED, CLICKED, UNSUBSCRIBED, REJECTED), bounce_
 UNDETERMINED), recipient, occurred_date, detail (the provider's reason, truncated; never body
 content), url (clicks only). Opens and clicks are stored only when the provider row enables them.

+### 6.3 Phase (c) implementation
+- Inbox states: PENDING, PROCESSED, FAILED, IGNORED. Retry after configured backoff,
+  FAILED after six attempts, and a standard Reprocess action resets FAILED to PENDING.
+- Events in one payload are applied independently. Only unmatched events (or events whose
+  message is still SENDING) defer the row; on the sixth attempt they are skipped and the row is
+  PROCESSED with the note "N events did not match a message". Suppression applies only to a
+  recipient in the matched message's to, cc or bcc; an SES bounce or complaint without its
+  recipient list is skipped, never widened to all destinations. SNS notifications must carry a
+  Timestamp no older than 1 hour and no more than 5 minutes ahead, and certificate fetch
+  failures are cached for 5 minutes per URL.
+- A short DB transaction locks the common provider gate then the actual provider;
+  it rechecks the observed inbox attempt and commits all changes together. No worker
+  filesystem, extra claim table, in-process scheduler or network operation is needed
+  during processing. A crash rolls back the transaction and releases the claim.
+- `email_event.event_key` provides unique recipient/event deduplication across
+  overlapping batches, alongside inbox uniqueness on provider/event and provider/nonce.
+- Message correlation always stays within the authenticated provider. Unknown message
+  ids retry (including callbacks before SENT); unsupported event kinds are ignored.
+- Opens/clicks require current provider flags and are also dropped for PHI-approved
+  messages. Event details mask addresses before truncation to 500 characters.
+- The unsubscribe signer binds one mailbox. A valid POST suppresses only that address
+  for MARKETING/NOTIFICATION; the page never displays addresses. Legacy links remain
+  valid only for messages with one To recipient and no Cc/Bcc.
+- PostgreSQL and H2 definitions, crypto fixtures and real random-port Javalin tests
+  are included. Certificate caching is memory-only. Setup is in `PROVIDERS.md`.
+
 ## 7. PHI and PII

 ### 7.1 Classification
@@ -371,6 +398,7 @@ MANUAL), source_message_id, source_event_id, note, created_date, expires_date (o
  respects HARD_BOUNCE and COMPLAINT.

 ### 8.2 One-click unsubscribe (RFC 8058, RFC 2369)
+- NOTIFICATION or MARKETING messages carrying an unsubscribe link/header must have exactly one To recipient and no Cc/Bcc; send one message per recipient.
 - MARKETING and NOTIFICATION messages carry
  `List-Unsubscribe: <https://{publicBaseUrl}/qqq/hooks/email/unsubscribe/{token}>` and
  `List-Unsubscribe-Post: List-Unsubscribe=One-Click`, and the footer links to the same URL.
@@ -379,11 +407,11 @@ MANUAL), source_message_id, source_event_id, note, created_date, expires_date (o
  MARKETING sends are refused. Tokens do not expire (RFC 8058 expects the link to keep working).
 - `POST` with `List-Unsubscribe=One-Click` records the suppression and returns 200. `GET` returns a
  minimal page with one "Unsubscribe" button that POSTs, because mail scanners prefetch GETs; then
-  "You are unsubscribed from {category}", with a link to Preferences.
+  "You are unsubscribed from {category}". A Preferences link awaits core #1162.
 - An `UNSUBSCRIBED` event is recorded on the message.

 ### 8.3 Preferences (core #1162)
-The qbit registers a Preferences section `email` ("Email") whose provider
+**Deferred: core #1162 is not in an RC.** The planned qbit registers a Preferences section `email` ("Email") whose provider
 (`EmailPreferenceProvider implements UserPreferenceProviderInterface`) maps the signed-in user's
 address (`UserPreferenceRequestContext.getUserEmailAddress()`, never an address from the request)
 to `email_suppression` rows: one toggle per opt-out-able category (NOTIFICATION,
@@ -455,10 +483,11 @@ section's description explains the split with qbit-alerts' Notifications section
 | `email_webhook_inbox` | relation (on the provider) | the qbit |
 | `email_inbound`, `email_inbound_attachment` | phase (e) | the qbit |

-Postgres DDL in `docs/ddl/email-postgres.sql` (phase a).
+Postgres DDL in `docs/ddl/email-postgres.sql` (phases a-c).

 ## 14. Configuration (`EmailQBitConfig`)
-backendName (required), schedulerName (required to send), publicBaseUrl (unsubscribe links),
+backendName (required), schedulerName (required for automatic processing), publicBaseUrl (unsubscribe links),
+webhookBasePath (`/qqq/hooks/email`, shared by tracking and unsubscribe),
 defaultProviderName, bridgeProviderName (`qbitEmail`), secretResolver, allowedEnvNames,
 dispatchIntervalSeconds (15), batchSize (100), maxAttempts (6), retryMinutes, leaseSeconds (120),
 bodyRetentionDays (30), messageRetentionDays (0), mjmlCompiler, classificationOverrides, appName
@@ -469,12 +498,12 @@ and appLabel, securityFields and recordSecurityLocks (multi-tenant hosts).
 |---|---|---|
 | (a) | tables, outbox and dispatcher, SMTP and SES, Markdown/HTML/MJML templates and versions, brands and the layout, preview, EmailService, the process, the core bridge, PHI refusal and redaction, the app home | GreenMail SMTP sink end to end; SES through a mocked `SesV2Client`; the claim race (two runners, each message sent once); retry, dead letter, idempotency; rendering (variables, escaping, `#if`/`#each`, plain text, dark-mode contrast); PHI refusal; snapshot redaction; the canary test; producer standards (names, nav, permissions) |
 | (b) | Postmark, SendGrid, Mailgun, Gmail SMTP | HTTP mocks per provider (success, 4xx permanent, 429/5xx transient); the Gmail preset. HTML/MJML moved into part (a), part 2 at owner request. |
-| (c) | webhook route and inbox, events, suppression, unsubscribe, Preferences section | signature verification per provider (valid, tampered, stale, missing secret); the SNS certificate host rule; suppression on hard bounce and complaint; one-click POST and the GET page; category rules |
+| (c) | webhook route and inbox, events, suppression, unsubscribe (Preferences deferred) | signature verification per provider (valid, tampered, stale, missing secret); the SNS certificate host rule; suppression on hard bounce and complaint; one-click POST and the GET page; category rules |
 | (d) | consumer migrations | each consumer's own tests; a contract test of `EmailService` |
 | (e) | inbound | design only for now |

 Dependencies: phase (a) builds on QQQ 4.1.0-RC.5 with the classification stand-in, and moves to
-RC.6 for `DataClassification` and Preferences.
+later RCs for `DataClassification` and Preferences when their APIs are released.

 ## 16. Owner steps (outside code)
 - **PMG (kof22):** none; the relay already accepts the cluster.
@@ -191,25 +191,75 @@ It would require a Google Cloud project, OAuth consent, client credentials and
 refresh-token management. It adds no needed system-mail capability over the app
 password preset in this phase and is built only on request (DESIGN 3.6).

-## Event webhooks: phase (c)
+## Event webhooks

-Event ingestion is **not installed by phase (b)**. After phase (c) is deployed,
-configure provider event delivery at:
+Phase (c) installs authenticated provider receivers when the host includes Javalin.
+Use `https://<app>{base}/{kind}/{providerName}`, where `base` is
+`EmailQBitConfig.webhookBasePath` (default `/qqq/hooks/email`), `kind` is `ses`,
+`postmark`, `sendgrid` or `mailgun`, and `providerName` is the row's unique name.
+The row must be enabled and its kind must match. The configurable base must start
+with `/`, have no trailing slash, and stay outside `/qqq/v1`.

-`https://<app>/qqq/hooks/email/{kind}/{providerName}`
-
-Use the configured provider name and provider-kind route defined by phase (c).
-Do not expect delivery/bounce/complaint/open/click updates until the receiver
-exists. Tracking options can operate on the provider side now.
-
-| Provider | Event setup to prepare |
+| Provider | Setup and events |
 |---|---|
-| Postmark | Server/stream event webhooks with Basic auth; `POSTMARK_WEBHOOK_PASSWORD` |
-| SendGrid | Signed Event Webhook; verification key in `SENDGRID_WEBHOOK_PUBLIC_KEY` |
-| Mailgun | Regional domain event webhooks; `MAILGUN_WEBHOOK_SIGNING_KEY`, separate from the sending key |
-| SES | Configuration-set events to SNS, HTTP subscription and `sns_topic_arn`; signature/topic verification arrives in phase (c) |
-| SMTP / Gmail SMTP | No native provider event webhook in this design |
+| SES | Send configuration-set Delivery, Bounce, Complaint, Reject and DeliveryDelay events to SNS; enable Open/Click only when desired. Set `sns_topic_arn` to that exact topic. Subscribe the HTTPS `/ses/{providerName}` URL with raw delivery **off**, so SNS sends its signed JSON envelope. No webhook secret env is used. Verified subscription confirmations are fetched automatically and retained as IGNORED inbox rows. |
+| Postmark | Configure server/message-stream webhooks for Delivery, Bounce, SpamComplaint and SubscriptionChange; optionally Open and Click. Use HTTP Basic username **`postmark`** and the password resolved from `webhook_secret_env_name` (normally `POSTMARK_WEBHOOK_PASSWORD`). Configure those credentials in Postmark's webhook settings/URL; keep the stored provider row limited to the environment name. |
+| SendGrid | Enable the Signed Event Webhook for delivered, bounce, dropped, deferred, spamreport and unsubscribe (group_unsubscribe is also accepted); optionally open/click. Store the base64 DER verification public key in `SENDGRID_WEBHOOK_PUBLIC_KEY`, referenced by `webhook_secret_env_name`. Preserve both Twilio signature/timestamp headers through the proxy. Batches are authenticated over timestamp plus the unchanged raw body. |
+| Mailgun | Configure regional domain webhooks for delivered, permanent/temporary failed, complained and unsubscribed; optionally opened/clicked. Set `webhook_secret_env_name=MAILGUN_WEBHOOK_SIGNING_KEY` to the account's HTTP webhook signing key, **not** its sending API key. Preserve the JSON `signature` object and `event-data`. |
+| SMTP / Gmail SMTP | No native event receiver. One-click unsubscribe remains available for opt-out mail. |

-Webhook environment names are allowlisted now so rows can be prepared; outbound
-sends never resolve them. Verification, durable event processing and unsubscribe
-handling arrive in later phases.
+SNS accepts SignatureVersion 1 and 2, pins TopicArn, and fetches certificates only
+from HTTPS `sns.<region>.amazonaws.com` or `.amazonaws.com.cn` hosts with `.pem`
+paths. Certificates are cached only in memory; redirects are disabled. Confirmation
+URLs use the same HTTPS host restriction. Canonicalization follows
+[AWS's signature rules](https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message-verify-message-signature.html).
+SNS retries can be old, so freshness is enforced only for SendGrid and Mailgun:
+timestamps must be within five minutes in either direction. Keep host clocks synchronized.
+Postmark authenticates the request with Basic auth and supplies no signed timestamp.
+
+Requests are capped at 1 MiB, including chunked requests. Responses are JSON:
+200 `received` or `duplicate`, 401 invalid verification/Mailgun nonce replay,
+404 missing/disabled/mismatched provider, 413 oversized input, and 503 unavailable
+verification configuration, SNS fetch or database. No request bodies, recipient
+addresses or credential values are logged. Raw payloads are stored in the DB inbox
+as sensitive data, readable with `email.view`.
+
+The hidden `email_process_webhooks` schedule runs every 30 seconds. One bounded
+transaction applies each inbox row, including events, message state and suppressions.
+The common provider lock serializes suppression upserts across replicas/providers;
+no network I/O occurs under that lock. Failed attempts back off with `retryMinutes`
+and become FAILED after six attempts. Use the inbox's **Reprocess** Review/Result
+action (`email.manage`) after resolving the cause. Unmatched messages retry, including
+a webhook racing the transport's SENT commit. Unknown/disabled tracking events are
+IGNORED. Per-event recipient keys prevent overlapping SendGrid batches from duplicating
+history. Mailgun's `(provider_id, nonce)` uniqueness is distinct from event deduplication:
+a stored nonce is refused with 401 even if its event is already known.
+
+Hard bounces and complaints upsert ALL suppression; complaints take precedence over
+hard-bounce reasons. Unsubscribe events suppress only MARKETING or NOTIFICATION for
+the matched message. Delivery advances SENT to DELIVERED; hard bounce and complaint
+advance delivery state without reversing a later complaint. Opens/clicks require the
+corresponding provider flag and are never retained for PHI-approved messages.
+Provider reason text is address-masked and limited to 500 characters; only clicks
+retain URLs. More than three complaints in 24 hours raises optional WARNING alert
+`email.complaint`, deduplicated per UTC day by core's alert sink.
+
+## One-click unsubscribe
+
+Set `publicBaseUrl` to the public HTTPS app origin and supply
+`QQQ_EMAIL_UNSUBSCRIBE_KEY` through the secret resolver. The configured webhook base
+also controls newly signed unsubscribe URLs. Keep the key and published path stable:
+tokens have no expiry, but cannot resolve after their message row is deleted.
+
+GET `{base}/unsubscribe/{token}` displays a self-contained, escaped light/dark form
+and never changes state. POST `List-Unsubscribe=One-Click` returns an empty 200;
+the page's form POST returns confirmation HTML. Both atomically upsert category
+suppression and record an UNSUBSCRIBED event. Repeated POSTs do not duplicate events.
+Invalid tokens return a generic 404. TRANSACTIONAL and SECURITY tokens are neither
+issued nor accepted.
+
+The existing v1 token binds the message, category and **complete To/Cc/Bcc envelope
+hash**, rather than one mailbox. Following that link opts out every mailbox in the
+bound envelope for that category; the page reveals none of those addresses. Send
+one recipient per opt-out message when individual preference control is needed.
+Preferences UI remains deferred until core #1162 ships in an RC.
@@ -1,4 +1,4 @@
-- Phase (a), parts 1 and 2. Apply before producing the qbit. Host security columns must be added by the host.
+-- Phases (a), (b) and (c). Apply before producing the qbit. Host security columns must be added by the host.

 CREATE TABLE email_provider (
   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
@@ -184,3 +184,42 @@ CREATE TABLE email_suppression (
 CREATE INDEX email_message_due ON email_message (status, next_attempt_date);
 CREATE INDEX email_attempt_provider_start ON email_message_attempt (provider_id, start_date);
 CREATE INDEX email_attempt_provider_end ON email_message_attempt (provider_id, end_date, id);
+
+CREATE TABLE email_webhook_inbox (
+   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   create_date TIMESTAMP,
+   modify_date TIMESTAMP,
+   provider_id INTEGER NOT NULL,
+   kind VARCHAR(500) NOT NULL,
+   external_event_id VARCHAR(500) NOT NULL,
+   nonce VARCHAR(500),
+   status VARCHAR(500) NOT NULL,
+   attempts INTEGER NOT NULL DEFAULT 0,
+   next_attempt_date TIMESTAMP,
+   last_error VARCHAR(500),
+   payload_json TEXT,
+   payload_sha256 VARCHAR(500),
+   received_date TIMESTAMP,
+   processed_date TIMESTAMP,
+   UNIQUE (provider_id, external_event_id),
+   UNIQUE (provider_id, nonce)
+);
+
+CREATE TABLE email_event (
+   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   create_date TIMESTAMP,
+   modify_date TIMESTAMP,
+   message_id INTEGER NOT NULL,
+   provider_id INTEGER,
+   event_type VARCHAR(500) NOT NULL,
+   bounce_type VARCHAR(500),
+   recipient VARCHAR(500),
+   occurred_date TIMESTAMP,
+   detail VARCHAR(500),
+   url TEXT,
+   event_key VARCHAR(500) NOT NULL UNIQUE
+);
+CREATE INDEX email_inbox_due ON email_webhook_inbox (status, next_attempt_date);
+CREATE INDEX email_event_message ON email_event (message_id);
+CREATE INDEX email_event_type_date ON email_event (event_type, occurred_date);
+CREATE INDEX email_message_provider_id ON email_message (provider_id, provider_message_id);
@@ -46,6 +46,7 @@ public class EmailQBitConfig implements QBitConfig
   private String backendName;
   private String schedulerName;
   private String publicBaseUrl;
+   private String webhookBasePath = "/qqq/hooks/email";
   private String defaultProviderName;
   private String bridgeProviderName = "qbitEmail";
   private Function<String, String> secretResolver = System::getenv;
@@ -89,6 +90,11 @@ public class EmailQBitConfig implements QBitConfig
   @Override
   public void validate(QInstance instance, List<String> errors)
   {
+      if(webhookBasePath == null || !webhookBasePath.matches("/(?:[A-Za-z0-9_-]+/)*[A-Za-z0-9_-]+")
+         || webhookBasePath.equals("/qqq/v1") || webhookBasePath.startsWith("/qqq/v1/"))
+      {
+         errors.add("webhookBasePath must be an absolute path outside /qqq/v1, without a trailing slash");
+      }
      if(backendName == null || backendName.isBlank())
      {
         errors.add("backendName is required");
@@ -201,6 +207,23 @@ public class EmailQBitConfig implements QBitConfig
      return (publicBaseUrl);
   }

+   /*******************************************************************************
+    ** Public provider and unsubscribe route base.
+    *******************************************************************************/
+   public String getWebhookBasePath()
+   {
+      return (webhookBasePath);
+   }
+
+   /*******************************************************************************
+    ** Configure the route base and queue-time unsubscribe URLs together.
+    *******************************************************************************/
+   public EmailQBitConfig withWebhookBasePath(String webhookBasePath)
+   {
+      this.webhookBasePath = webhookBasePath;
+      return (this);
+   }
+

   /*******************************************************************************
    ** Fluent setter for publicBaseUrl.
@@ -195,6 +195,24 @@ public class EmailQBitProducer implements QBitMetaDataProducer<EmailQBitConfig>
      output.add(EmailApp.app(qBitConfig, tables));
      output.add(new EmailOutboxMessagingProviderMetaData().withName(qBitConfig.getBridgeProviderName()));
      output.addSelfToInstance(qInstance);
+      registerWebhooks(qInstance);
      return (new MetaDataProducerMultiOutput());
   }
+
+   /*******************************************************************************
+    ** Keep optional provided middleware linkage outside non-web applications.
+    *******************************************************************************/
+   private static void registerWebhooks(QInstance instance) throws QException
+   {
+      try
+      {
+         Class.forName("io.javalin.Javalin");
+         Class.forName("com.kingsrook.qqq.middleware.javalin.QJavalinMetaData");
+      }
+      catch(ClassNotFoundException e)
+      {
+         return;
+      }
+      com.kingsrook.qbits.email.webhook.EmailWebhookRouteProvider.register(instance);
+   }
 }
@@ -68,8 +68,21 @@ public final class EmailHome
      String[] fields = {"sentDate", "deliveredDate", "deadLetterDate", "bouncedDate", "complainedDate"};
      for(int i = 0; i < fields.length; i++)
      {
-         totals.add(count(EmailTables.MESSAGE, since(fields[i], now.minus(i < 3 ? 1 : 7, ChronoUnit.DAYS))
-            .withCriteria(fields[i], QCriteriaOperator.LESS_THAN_OR_EQUALS, now)));
+         String eventType = switch(i)
+         {
+            case 1 -> "DELIVERED";
+            case 3 -> "BOUNCED";
+            case 4 -> "COMPLAINED";
+            default -> null;
+         };
+         String field = eventType == null ? fields[i] : "occurredDate";
+         QQueryFilter filter = since(field, now.minus(i < 3 ? 1 : 7, ChronoUnit.DAYS))
+            .withCriteria(field, QCriteriaOperator.LESS_THAN_OR_EQUALS, now);
+         if(eventType != null)
+         {
+            filter.withCriteria("eventType", QCriteriaOperator.EQUALS, eventType);
+         }
+         totals.add(count(eventType == null ? EmailTables.MESSAGE : EmailTables.EVENT, filter));
      }
      totals.add(count(EmailTables.SUPPRESSION, new QQueryFilter()));
      List<QRecord> messages = query(EmailTables.MESSAGE, new QQueryFilter().withOrderBy(new QFilterOrderBy("createDate", false))
@@ -35,6 +35,7 @@ import com.kingsrook.qqq.backend.core.alerts.QAlerts;
 *******************************************************************************/
 public final class EmailAlerts
 {
+   public static final String COMPLAINT = "email.complaint";
   public static final String UNHEALTHY = "email.provider.unhealthy";
   public static final String DEAD_LETTER = "email.message.dead-letter";

@@ -50,7 +51,9 @@ public final class EmailAlerts
    *******************************************************************************/
   public static List<QAlertTypeMetaData> types()
   {
-      return (List.of(new QAlertTypeMetaData().withName(UNHEALTHY).withLabel("Email provider unhealthy").withDefaultSeverity(AlertSeverity.ERROR)
+      return (List.of(new QAlertTypeMetaData().withName(COMPLAINT).withLabel("Email complaints").withDefaultSeverity(AlertSeverity.WARNING)
+            .withSubjectTableName(EmailTables.EVENT).withVisibilityPermission("email.view").withSourceName("email"),
+         new QAlertTypeMetaData().withName(UNHEALTHY).withLabel("Email provider unhealthy").withDefaultSeverity(AlertSeverity.ERROR)
            .withSubjectTableName(EmailTables.PROVIDER).withVisibilityPermission("email.admin").withSourceName("email"),
         new QAlertTypeMetaData().withName(DEAD_LETTER).withLabel("Email dead letters").withDefaultSeverity(AlertSeverity.WARNING)
            .withSubjectTableName(EmailTables.MESSAGE).withVisibilityPermission("email.view").withSourceName("email")));
@@ -62,7 +65,11 @@ public final class EmailAlerts
   public static void raise(String type, Integer providerId)
   {
      String key = "provider:" + providerId + (DEAD_LETTER.equals(type) ? ":" + EmailRuntime.clock().instant().getEpochSecond() / 3600 : "");
-      QAlerts.raise(new AlertSignal(type).withDedupKey(key).withTitle(UNHEALTHY.equals(type) ? "Email provider unhealthy" : "Email message dead-lettered")
+      if(COMPLAINT.equals(type))
+      {
+         key = "complaints:" + EmailRuntime.clock().instant().getEpochSecond() / 86400;
+      }
+      QAlerts.raise(new AlertSignal(type).withDedupKey(key).withTitle(COMPLAINT.equals(type) ? "More than three email complaints in 24 hours" : UNHEALTHY.equals(type) ? "Email provider unhealthy" : "Email message dead-lettered")
         .withSummary("Review the Email tables for status and retry actions."));
   }

@@ -312,4 +312,88 @@ public final class EmailEnums
         return (name().charAt(0) + name().substring(1).toLowerCase(java.util.Locale.ROOT).replace('_', ' '));
      }
   }
+
+
+
+   /*******************************************************************************
+    ** InboxStatus values.
+    *******************************************************************************/
+   public enum InboxStatus implements PossibleValueEnum<String>
+   {
+      PENDING, PROCESSED, FAILED, IGNORED;
+
+      /*******************************************************************************
+       ** Stable stored identifier.
+       *******************************************************************************/
+      @Override
+      public String getPossibleValueId()
+      {
+         return (name());
+      }
+
+      /*******************************************************************************
+       ** Display label.
+       *******************************************************************************/
+      @Override
+      public String getPossibleValueLabel()
+      {
+         return (name().charAt(0) + name().substring(1).toLowerCase(java.util.Locale.ROOT).replace('_', ' '));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** EventType values.
+    *******************************************************************************/
+   public enum EventType implements PossibleValueEnum<String>
+   {
+      DELIVERED, DEFERRED, BOUNCED, COMPLAINED, OPENED, CLICKED, UNSUBSCRIBED, REJECTED;
+
+      /*******************************************************************************
+       ** Stable stored identifier.
+       *******************************************************************************/
+      @Override
+      public String getPossibleValueId()
+      {
+         return (name());
+      }
+
+      /*******************************************************************************
+       ** Display label.
+       *******************************************************************************/
+      @Override
+      public String getPossibleValueLabel()
+      {
+         return (name().charAt(0) + name().substring(1).toLowerCase(java.util.Locale.ROOT).replace('_', ' '));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** BounceType values.
+    *******************************************************************************/
+   public enum BounceType implements PossibleValueEnum<String>
+   {
+      HARD, SOFT, UNDETERMINED;
+
+      /*******************************************************************************
+       ** Stable stored identifier.
+       *******************************************************************************/
+      @Override
+      public String getPossibleValueId()
+      {
+         return (name());
+      }
+
+      /*******************************************************************************
+       ** Display label.
+       *******************************************************************************/
+      @Override
+      public String getPossibleValueLabel()
+      {
+         return (name().charAt(0) + name().substring(1).toLowerCase(java.util.Locale.ROOT).replace('_', ' '));
+      }
+   }
 }
@@ -59,9 +59,11 @@ public final class EmailTables
   public static final String VERSION = "email_template_version";
   public static final String BRAND = "email_brand";
   public static final String SUPPRESSION = "email_suppression";
-   public static final List<String> ALL = List.of(PROVIDER, MESSAGE, ATTEMPT, TEMPLATE, VERSION, BRAND, SUPPRESSION);
+   public static final String INBOX = "email_webhook_inbox";
+   public static final String EVENT = "email_event";
+   public static final List<String> ALL = List.of(PROVIDER, MESSAGE, ATTEMPT, TEMPLATE, VERSION, BRAND, SUPPRESSION, INBOX, EVENT);
   public static final List<String> MAIN = List.of(MESSAGE, TEMPLATE, BRAND, SUPPRESSION, PROVIDER);
-   public static final List<String> SYSTEM_WRITTEN = List.of(MESSAGE, ATTEMPT);
+   public static final List<String> SYSTEM_WRITTEN = List.of(MESSAGE, ATTEMPT, INBOX, EVENT);

   /*******************************************************************************
    ** Utility class.
@@ -95,6 +97,9 @@ public final class EmailTables
      result.add(QPossibleValueSource.newForEnum("email_suppression_scope", EmailEnums.SuppressionScope.values()));
      result.add(QPossibleValueSource.newForEnum("email_suppression_reason", EmailEnums.SuppressionReason.values()));
      result.add(QPossibleValueSource.newForEnum("email_category", EmailCategory.values()));
+      result.add(QPossibleValueSource.newForEnum("email_inbox_status", EmailEnums.InboxStatus.values()));
+      result.add(QPossibleValueSource.newForEnum("email_event_type", EmailEnums.EventType.values()));
+      result.add(QPossibleValueSource.newForEnum("email_bounce_type", EmailEnums.BounceType.values()));
      ALL.forEach(name -> result.add(QPossibleValueSource.newForTable(name)));
      for(QPossibleValueSource source : result)
      {
@@ -108,7 +113,7 @@ public final class EmailTables
   }

   /*******************************************************************************
-    ** The seven phase (a) tables.
+    ** Outbound delivery and tracking tables.
    *******************************************************************************/
   public static List<QTableMetaData> tables(EmailQBitConfig config)
   {
@@ -120,6 +125,8 @@ public final class EmailTables
      tables.add(templateversion(config));
      tables.add(brand(config));
      tables.add(suppression(config));
+      tables.add(inbox(config));
+      tables.add(event(config));
      for(QTableMetaData table : tables)
      {
         for(QFieldMetaData field : config.getSecurityFields())
@@ -204,6 +211,7 @@ public final class EmailTables
      table.withSection(SectionFactory.defaultT2("fromAddress", "fromName", "replyTo", "sendingDomain", "host", "port", "security", "region", "usernameEnvName", "secretEnvName", "webhookSecretEnvName", "snsTopicArn", "configurationSet", "messageStream", "maxPerMinute", "maxPerDay", "trackOpens", "trackClicks", "phiAllowed", "healthStatus", "lastSuccessDate", "lastFailureDate", "lastFailureMessage"));
      table.withSection(SectionFactory.defaultT3("createDate", "modifyDate"));
      table.withSection(SectionFactory.customT2("children", new QIcon("list")).withLabel("Messages").withWidgetName(PROVIDER + "_children"));
+      table.withSection(SectionFactory.customT2("inbox", new QIcon("list")).withLabel("Webhook Inbox").withWidgetName(INBOX + "_children"));
      return (table);
   }

@@ -259,6 +267,7 @@ public final class EmailTables
      table.withSection(SectionFactory.defaultT2("replyTo", "subject", "templateId", "templateVersionId", "brandId", "category", "locale", "variablesSnapshot", "htmlBody", "textBody", "phiApproved", "unsubscribeUrl", "bodySha256", "idempotencyKey", "providerId", "providerMessageId", "attemptCount", "nextAttemptDate", "leaseExpiresDate", "lastError", "finishedDate", "sentDate", "deliveredDate", "bouncedDate", "complainedDate", "deadLetterDate", "linkedTableName", "linkedRecordId", "senderQbit", "senderProcess", "suppressedAddresses", "sendAfter"));
      table.withSection(SectionFactory.defaultT3("createDate", "modifyDate"));
      table.withSection(SectionFactory.customT2("children", new QIcon("list")).withLabel("Attempts").withWidgetName(MESSAGE + "_children"));
+      table.withSection(SectionFactory.customT2("events", new QIcon("list")).withLabel("Events").withWidgetName(EVENT + "_children"));
      return (table);
   }

@@ -390,7 +399,7 @@ public final class EmailTables
         .withField(field("scope", QFieldType.STRING).withPossibleValueSourceName("email_suppression_scope").withIsRequired(true).withMaxLength(500))
         .withField(field("reason", QFieldType.STRING).withPossibleValueSourceName("email_suppression_reason").withIsRequired(true).withMaxLength(500))
         .withField(field("sourceMessageId", QFieldType.INTEGER))
-         .withField(field("sourceEventId", QFieldType.INTEGER))
+         .withField(field("sourceEventId", QFieldType.INTEGER).withPossibleValueSourceName(EVENT))
         .withField(field("note", QFieldType.TEXT))
         .withField(field("createdDate", QFieldType.DATE_TIME))
         .withField(field("expiresDate", QFieldType.DATE_TIME));
@@ -401,12 +410,59 @@ public final class EmailTables
      return (table);
   }

+
+   /*******************************************************************************
+    ** System-written webhook inbox records.
+    *******************************************************************************/
+   private static QTableMetaData inbox(EmailQBitConfig config)
+   {
+      QTableMetaData table = base(config, INBOX, "Webhook Inbox")
+         .withField(field("providerId", QFieldType.INTEGER).withIsRequired(true).withPossibleValueSourceName(PROVIDER))
+         .withField(field("kind", QFieldType.STRING).withMaxLength(500))
+         .withField(field("externalEventId", QFieldType.STRING).withMaxLength(500).withIsRequired(true))
+         .withField(field("nonce", QFieldType.STRING).withMaxLength(500))
+         .withField(field("status", QFieldType.STRING).withMaxLength(500).withIsRequired(true).withPossibleValueSourceName("email_inbox_status"))
+         .withField(field("attempts", QFieldType.INTEGER))
+         .withField(field("nextAttemptDate", QFieldType.DATE_TIME))
+         .withField(field("lastError", QFieldType.STRING).withMaxLength(500))
+         .withField(field("payloadJson", QFieldType.TEXT).withIsSensitive(true))
+         .withField(field("payloadSha256", QFieldType.STRING).withMaxLength(500))
+         .withField(field("receivedDate", QFieldType.DATE_TIME))
+         .withField(field("processedDate", QFieldType.DATE_TIME));
+      table.withUniqueKey(new UniqueKey("providerId", "externalEventId"));
+      table.withUniqueKey(new UniqueKey("providerId", "nonce"));
+      table.withSection(SectionFactory.defaultT1("id", "providerId", "kind", "externalEventId", "nonce", "status", "attempts", "nextAttemptDate", "lastError", "payloadJson", "payloadSha256", "receivedDate", "processedDate"));
+      table.withSection(SectionFactory.defaultT3("createDate", "modifyDate"));
+      return (table);
+   }
+
+   /*******************************************************************************
+    ** System-written event records.
+    *******************************************************************************/
+   private static QTableMetaData event(EmailQBitConfig config)
+   {
+      QTableMetaData table = base(config, EVENT, "Event")
+         .withField(field("messageId", QFieldType.INTEGER).withIsRequired(true).withPossibleValueSourceName(MESSAGE))
+         .withField(field("providerId", QFieldType.INTEGER).withPossibleValueSourceName(PROVIDER))
+         .withField(field("eventType", QFieldType.STRING).withMaxLength(500).withIsRequired(true).withPossibleValueSourceName("email_event_type"))
+         .withField(field("bounceType", QFieldType.STRING).withMaxLength(500).withPossibleValueSourceName("email_bounce_type"))
+         .withField(field("recipient", QFieldType.STRING).withMaxLength(500).withIsSensitive(true))
+         .withField(field("occurredDate", QFieldType.DATE_TIME))
+         .withField(field("detail", QFieldType.STRING).withMaxLength(500))
+         .withField(field("url", QFieldType.TEXT).withIsSensitive(true))
+         .withField(field("eventKey", QFieldType.STRING).withMaxLength(500).withIsRequired(true).withIsHidden(true));
+      table.withUniqueKey(new UniqueKey("eventKey"));
+      table.withSection(SectionFactory.defaultT1("id", "messageId", "providerId", "eventType", "bounceType", "recipient", "occurredDate", "detail", "url"));
+      table.withSection(SectionFactory.defaultT3("createDate", "modifyDate"));
+      return (table);
+   }
+
   /*******************************************************************************
    ** Relations used by child lists.
    *******************************************************************************/
   public static List<QJoinMetaData> joins()
   {
-      return (List.of(join(PROVIDER, MESSAGE, "providerId"), join(MESSAGE, ATTEMPT, "messageId"), join(TEMPLATE, VERSION, "templateId")));
+      return (List.of(join(PROVIDER, MESSAGE, "providerId"), join(MESSAGE, ATTEMPT, "messageId"), join(TEMPLATE, VERSION, "templateId"), join(PROVIDER, INBOX, "providerId"), join(MESSAGE, EVENT, "messageId")));
   }

   /*******************************************************************************
@@ -426,8 +482,8 @@ public final class EmailTables
      List<QWidgetMetaData> widgets = new ArrayList<>();
      for(QJoinMetaData join : joins())
      {
-         widgets.add(ChildRecordListRenderer.widgetMetaDataBuilder(join).withName(join.getLeftTable() + "_children")
-            .withLabel(join.getRightTable().equals(ATTEMPT) ? "Attempts" : join.getRightTable().equals(VERSION) ? "Versions" : "Messages")
+         widgets.add(ChildRecordListRenderer.widgetMetaDataBuilder(join).withName(List.of(INBOX, EVENT).contains(join.getRightTable()) ? join.getRightTable() + "_children" : join.getLeftTable() + "_children")
+            .withLabel(join.getRightTable().equals(INBOX) ? "Webhook Inbox" : join.getRightTable().equals(EVENT) ? "Events" : join.getRightTable().equals(ATTEMPT) ? "Attempts" : join.getRightTable().equals(VERSION) ? "Versions" : "Messages")
            .withMaxRows(50).getWidgetMetaData().withGridColumns(12).withPermissionRules(permissionRules()));
      }
      return (widgets);
@@ -36,6 +36,7 @@ import com.kingsrook.qbits.email.metadata.EmailTableCustomizers;
 import com.kingsrook.qbits.email.metadata.EmailTables;
 import com.kingsrook.qbits.email.permissions.EmailPermissionChecker;
 import com.kingsrook.qbits.email.permissions.EmailPermissions;
+import com.kingsrook.qbits.email.render.EmailUnsubscribeLinks;
 import com.kingsrook.qbits.email.support.EmailStore;
 import com.kingsrook.qbits.email.support.SystemWrites;
 import com.kingsrook.qbits.email.template.EmailTemplateRenderer;
@@ -187,6 +188,18 @@ public final class EmailService
      {
         return (refused("One or more email addresses are invalid"));
      }
+      try
+      {
+         String unsubscribe = EmailUnsubscribeLinks.create("pending", String.join(",", to), category.name());
+         if(unsubscribe != null || retained != null && retained.getValueString("unsubscribeUrl") != null)
+         {
+            EmailUnsubscribeLinks.recipient(String.join(",", to), String.join(",", cc), String.join(",", bcc));
+         }
+      }
+      catch(IllegalArgumentException e)
+      {
+         return (refused(e.getMessage()));
+      }
      if(to.isEmpty() && cc.isEmpty() && bcc.isEmpty())
      {
         return (refused("At least one recipient is required"));
@@ -214,9 +227,14 @@ public final class EmailService
      }
      boolean allSuppressed = to.isEmpty() && cc.isEmpty() && bcc.isEmpty();
      EmailRenderer.Rendered rendered;
+      QRecord adHocBrand = null;
      try
      {
         rendered = templateContent == null ? renderer.render(request) : new EmailRenderer.Rendered(templateContent.getValueString("htmlBody"), templateContent.getValueString("textBody"));
+         if(templateContent == null)
+         {
+            adHocBrand = com.kingsrook.qbits.email.render.EmailBrands.resolve(request.getBrandName(), null);
+         }
      }
      catch(IllegalArgumentException e)
      {
@@ -228,6 +246,7 @@ public final class EmailService
         .withValue("toAddresses", String.join(",", to)).withValue("ccAddresses", String.join(",", cc)).withValue("bccAddresses", String.join(",", bcc))
         .withValue("fromAddress", from).withValue("replyTo", replyTo).withValue("subject", request.getSubject())
         .withValue("htmlBody", rendered.html()).withValue("textBody", rendered.text()).withValue("bodySha256", digest(rendered))
+         .withValue("brandId", adHocBrand == null ? null : adHocBrand.getValue("id"))
         .withValue("category", category.name()).withValue("locale", request.getLocale()).withValue("idempotencyKey", request.getIdempotencyKey())
         .withValue("linkedTableName", request.getLinkedTableName()).withValue("linkedRecordId", request.getLinkedRecordId())
         .withValue("senderQbit", request.getSenderQbit()).withValue("senderProcess", request.getSenderProcess())
@@ -237,14 +256,6 @@ public final class EmailService
      {
         row.getValues().putAll(templateContent.getValues());
      }
-      try
-      {
-         com.kingsrook.qbits.email.render.EmailUnsubscribeLinks.create("pending", String.join(",", to), category.name());
-      }
-      catch(IllegalArgumentException e)
-      {
-         return (refused(e.getMessage()));
-      }
      for(var field : config.getSecurityFields())
      {
         row.setValue(field.getName(), request.getSecurityValues().get(field.getName()));
@@ -383,10 +394,10 @@ public final class EmailService
      try
      {
         String unsubscribe = row.getValueString("unsubscribeUrl");
-         if(unsubscribe == null && (templateRenderer != null || !Boolean.TRUE.equals(row.getValueBoolean("phiApproved"))))
+         if(unsubscribe == null && !"SUPPRESSED".equals(row.getValueString("status"))
+            && (templateRenderer != null || !Boolean.TRUE.equals(row.getValueBoolean("phiApproved"))))
         {
-            unsubscribe = com.kingsrook.qbits.email.render.EmailUnsubscribeLinks.create(inserted.getValueString("id"),
-               row.getValueString("toAddresses") + "," + row.getValueString("ccAddresses") + "," + row.getValueString("bccAddresses"), row.getValueString("category"));
+            unsubscribe = EmailUnsubscribeLinks.create(inserted.getValueString("id"), row.getValueString("toAddresses"), row.getValueString("category"));
         }
         QRecord content;
         if(templateRenderer != null)
@@ -86,6 +86,8 @@ public final class EmailProcesses
   public static final String SEND_TEMPLATED = "email_send_templated";
   public static final String DISPATCH = "email_dispatch";
   public static final String MAINTAIN = "email_maintain";
+   public static final String REPROCESS = "email_reprocess_webhook";
+   public static final String PROCESS_WEBHOOKS = "email_process_webhooks";
   public static final String RETENTION = "email_retention";

   /*******************************************************************************
@@ -102,7 +104,7 @@ public final class EmailProcesses
   {
      return (switch(name)
      {
-         case RESEND, CANCEL, SEND_NOW, NEW_VERSION, ACTIVATE_VERSION -> EmailPermissions.MANAGE;
+         case REPROCESS, RESEND, CANCEL, SEND_NOW, NEW_VERSION, ACTIVATE_VERSION -> EmailPermissions.MANAGE;
         case SEND_TEMPLATED, SEND_AD_HOC -> EmailPermissions.SEND;
         default -> EmailPermissions.ADMIN;
      });
@@ -136,6 +138,7 @@ public final class EmailProcesses
         action(SEND_NOW, "Send Now", EmailTables.MESSAGE), test, send, adHoc,
         action(NEW_VERSION, "New Version", EmailTables.TEMPLATE), action(ACTIVATE_VERSION, "Activate Version", EmailTables.VERSION),
         action(APPROVE_PHI, "Approve For PHI", EmailTables.VERSION),
+         action(REPROCESS, "Reprocess", EmailTables.INBOX), scheduled(config, PROCESS_WEBHOOKS, "Process Email Webhooks", 30),
         scheduled(config, DISPATCH, "Dispatch Email", config.getDispatchIntervalSeconds()),
         scheduled(config, MAINTAIN, "Maintain Email", 300), scheduled(config, RETENTION, "Email Retention", 86400)));
   }
@@ -175,6 +178,10 @@ public final class EmailProcesses
    *******************************************************************************/
   private static boolean eligible(String process, QRecord row)
   {
+      if(REPROCESS.equals(process))
+      {
+         return ("FAILED".equals(row.getValueString("status")));
+      }
      if(NEW_VERSION.equals(process))
      {
         return (true);
@@ -266,7 +273,7 @@ public final class EmailProcesses
         {
            Integer id = row.getValueInteger("id");
            QRecord current = EmailStore.get(NEW_VERSION.equals(process) ? EmailTables.TEMPLATE : List.of(ACTIVATE_VERSION, APPROVE_PHI).contains(process) ? EmailTables.VERSION
-               : SEND_TEST.equals(process) ? EmailTables.PROVIDER : EmailTables.MESSAGE, id, null);
+               : REPROCESS.equals(process) ? EmailTables.INBOX : SEND_TEST.equals(process) ? EmailTables.PROVIDER : EmailTables.MESSAGE, id, null);
            String result = "Skipped";
            if(current != null && eligible(process, current))
            {
@@ -283,6 +290,7 @@ public final class EmailProcesses
                     com.kingsrook.qbits.email.template.EmailVersions.approve(id);
                     yield "Approved for PHI";
                  }
+                  case REPROCESS -> com.kingsrook.qbits.email.tracking.EmailInboxProcessor.reprocess(id) ? "PENDING" : "Skipped";
                  case SEND_NOW -> new EmailDispatcher().sendNow(id).name();
                  case CANCEL -> cancel(id) ? "CANCELLED" : "Skipped";
                  case RESEND -> resend(current).status().name();
@@ -431,6 +439,7 @@ public final class EmailProcesses
      {
         switch(input.getProcessName())
         {
+            case PROCESS_WEBHOOKS -> new com.kingsrook.qbits.email.tracking.EmailInboxProcessor().runOnce();
            case RETENTION -> retain();
            case MAINTAIN -> new EmailDispatcher().maintain();
            default -> new EmailDispatcher().runOnce();
@@ -440,6 +449,8 @@ public final class EmailProcesses

   /*******************************************************************************
    ** Clear retained bodies and optionally delete final messages and their attempts.
+    ** NOTIFICATION and MARKETING messages are never deleted (their unsubscribe links must
+    ** keep working); at message retention only their bodies and variable snapshot are cleared.
    *******************************************************************************/
   public static void retain() throws QException
   {
@@ -464,18 +475,35 @@ public final class EmailProcesses
         {
            continue;
         }
-         if(config.getMessageRetentionDays() > 0 && finalDate.isBefore(now.minus(Duration.ofDays(config.getMessageRetentionDays()))))
+         boolean expired = config.getMessageRetentionDays() > 0 && finalDate.isBefore(now.minus(Duration.ofDays(config.getMessageRetentionDays())));
+         boolean optOut = List.of("NOTIFICATION", "MARKETING").contains(row.getValueString("category"));
+         if(expired && !optOut)
         {
            try(QBackendTransaction transaction = QBackendTransaction.openFor(new UpdateInput(EmailTables.MESSAGE)))
            {
+               if(row.getValueInteger("providerId") != null)
+               {
+                  com.kingsrook.qbits.email.tracking.EmailTrackingStore.lock(row.getValueInteger("providerId"), transaction);
+               }
+               SystemWrites.delete(EmailTables.EVENT, EmailStore.equal("messageId", row.getValueInteger("id")), transaction);
               SystemWrites.delete(EmailTables.ATTEMPT, EmailStore.equal("messageId", row.getValueInteger("id")), transaction);
               SystemWrites.delete(EmailTables.MESSAGE, EmailStore.equal("id", row.getValueInteger("id")), transaction);
               transaction.commit();
            }
         }
-         else if(finalDate.isBefore(now.minus(Duration.ofDays(config.getBodyRetentionDays()))) && (row.getValue("htmlBody") != null || row.getValue("textBody") != null))
+         else if(expired || finalDate.isBefore(now.minus(Duration.ofDays(config.getBodyRetentionDays()))))
         {
-            SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", row.getValue("id")).withValue("htmlBody", null).withValue("textBody", null));
+            // Notification and marketing rows are never deleted, so their unsubscribe links keep working; only content goes.
+            boolean clearSnapshot = expired && row.getValue("variablesSnapshot") != null;
+            if(row.getValue("htmlBody") != null || row.getValue("textBody") != null || clearSnapshot)
+            {
+               QRecord update = new QRecord().withValue("id", row.getValue("id")).withValue("htmlBody", null).withValue("textBody", null);
+               if(clearSnapshot)
+               {
+                  update.withValue("variablesSnapshot", null);
+               }
+               SystemWrites.updateOne(EmailTables.MESSAGE, update);
+            }
         }
      }
   }
@@ -27,13 +27,15 @@ import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.util.Base64;
 import java.util.HexFormat;
+import java.util.Locale;
 import java.util.Map;
 import com.kingsrook.qbits.email.EmailRuntime;
 import com.kingsrook.qbits.email.template.VariableSchema;
+import jakarta.mail.internet.InternetAddress;


 /*******************************************************************************
- ** Queue-time link signing from DESIGN 8.2. The receiving route is phase (c).
+ ** Queue-time link signing and verification from DESIGN 8.2.
 *******************************************************************************/
 public final class EmailUnsubscribeLinks
 {
@@ -45,39 +47,104 @@ public final class EmailUnsubscribeLinks
   }

   /*******************************************************************************
-    ** Bind the message, recipient-envelope hash and category without exposing PII.
+    ** Bind the message, single mailbox hash and category without exposing PII.
    ** No key means no link/header; marketing must refuse rather than omit opt-out.
    *******************************************************************************/
-   public static String create(String messageId, String recipients, String category)
+   public static String create(String messageId, String recipient, String category)
   {
      if(!"NOTIFICATION".equals(category) && !"MARKETING".equals(category))
      {
         return (null);
      }
+      String key;
      try
      {
-         String key = EmailRuntime.config().getSecretResolver().apply("QQQ_EMAIL_UNSUBSCRIBE_KEY");
-         String base = EmailRuntime.config().getPublicBaseUrl();
-         if(key == null || key.isBlank() || base == null || base.isBlank())
+         key = EmailRuntime.config().getSecretResolver().apply("QQQ_EMAIL_UNSUBSCRIBE_KEY");
+      }
+      catch(Exception e)
+      {
+         throw (new IllegalArgumentException("Opt-out email needs publicBaseUrl and QQQ_EMAIL_UNSUBSCRIBE_KEY"));
+      }
+      String base = EmailRuntime.config().getPublicBaseUrl();
+      if(key == null || key.isBlank() || base == null || base.isBlank())
+      {
+         if("MARKETING".equals(category))
         {
-            if("MARKETING".equals(category))
-            {
-               throw (new IllegalArgumentException());
-            }
-            return (null);
+            throw (new IllegalArgumentException("Opt-out email needs publicBaseUrl and QQQ_EMAIL_UNSUBSCRIBE_KEY"));
         }
+         return (null);
+      }
+      String address = recipient(recipient, null, null);
+      try
+      {
         VariableSchema.url(base);
-         String hash = HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(recipients.getBytes(StandardCharsets.UTF_8)));
+         String hash = HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(address.getBytes(StandardCharsets.UTF_8)));
         String payload = Base64.getUrlEncoder().withoutPadding().encodeToString(VariableSchema.json(Map.of("messageId", messageId, "addressHash", hash, "category", category)).getBytes(StandardCharsets.UTF_8));
         Mac mac = Mac.getInstance("HmacSHA256");
         mac.init(new SecretKeySpec(key.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
         String signed = "v1." + payload;
         String signature = Base64.getUrlEncoder().withoutPadding().encodeToString(mac.doFinal(signed.getBytes(StandardCharsets.UTF_8)));
-         return (base.replaceAll("/+$", "") + "/qqq/hooks/email/unsubscribe/" + signed + "." + signature);
+         return (base.replaceAll("/+$", "") + EmailRuntime.config().getWebhookBasePath() + "/unsubscribe/" + signed + "." + signature);
      }
      catch(Exception e)
      {
         throw (new IllegalArgumentException("Opt-out email needs publicBaseUrl and QQQ_EMAIL_UNSUBSCRIBE_KEY"));
      }
   }
+
+   /*******************************************************************************
+    ** Exactly one To mailbox, canonicalized for signing and category suppression.
+    *******************************************************************************/
+   public static String recipient(String to, String cc, String bcc)
+   {
+      try
+      {
+         InternetAddress[] addresses = InternetAddress.parse(to == null ? "" : to, true);
+         if(addresses.length == 1 && !addresses[0].isGroup() && addresses[0].getAddress().contains("@")
+            && (cc == null || cc.isBlank()) && (bcc == null || bcc.isBlank()))
+         {
+            addresses[0].validate();
+            return (addresses[0].getAddress().toLowerCase(Locale.ROOT));
+         }
+      }
+      catch(jakarta.mail.internet.AddressException e)
+      {
+         // Malformed stored envelopes cannot authorize an opt-out.
+      }
+      throw (new IllegalArgumentException("Notification and marketing email with an unsubscribe link goes to one recipient per message; send one message per recipient."));
+   }
+
+   /*******************************************************************************
+    ** Verify the signed format before any lookup or side effect.
+    ** Invalid and non-opt-out categories are indistinguishable to callers.
+    *******************************************************************************/
+   public static org.json.JSONObject verify(String token)
+   {
+      try
+      {
+         if(token == null || token.length() > 4096)
+         {
+            return (null);
+         }
+         String[] parts = token.split("\\.", -1);
+         String key = EmailRuntime.config().getSecretResolver().apply("QQQ_EMAIL_UNSUBSCRIBE_KEY");
+         if(parts.length != 3 || !"v1".equals(parts[0]) || key == null || key.isBlank())
+         {
+            return (null);
+         }
+         Mac mac = Mac.getInstance("HmacSHA256");
+         mac.init(new SecretKeySpec(key.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+         if(!MessageDigest.isEqual(mac.doFinal((parts[0] + "." + parts[1]).getBytes(StandardCharsets.UTF_8)), Base64.getUrlDecoder().decode(parts[2])))
+         {
+            return (null);
+         }
+         var payload = new org.json.JSONObject(new String(Base64.getUrlDecoder().decode(parts[1]), StandardCharsets.UTF_8));
+         String category = payload.getString("category");
+         return (("MARKETING".equals(category) || "NOTIFICATION".equals(category)) ? payload : null);
+      }
+      catch(Exception e)
+      {
+         return (null);
+      }
+   }
 }
@@ -0,0 +1,315 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.template;
+
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.email.outbox.EmailCategory;
+
+
+/*******************************************************************************
+ ** A template another qbit ships with its own code: its name, content and
+ ** variable contract. Hand it to {@link EmailTemplates#ensure} at startup or on
+ ** first use; it is created once as an ACTIVE version 1 and never overwritten.
+ *******************************************************************************/
+public final class EmailTemplateDefinition
+{
+   private String name;
+   private String label;
+   private EmailCategory category;
+   private String kind = "MARKDOWN";
+   private String subject;
+   private String body;
+   private String textBody;
+   private String variablesSchema;
+   private final List<Map<String, Object>> variables = new ArrayList<>();
+   private String sampleVariables;
+   private String ownerQbit;
+   private String editPermissionName;
+
+   /*******************************************************************************
+    ** Unique snake_case template name, as used by EmailRequest.withTemplateName.
+    *******************************************************************************/
+   public EmailTemplateDefinition withName(String name)
+   {
+      this.name = name;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Human label shown in the template table.
+    *******************************************************************************/
+   public EmailTemplateDefinition withLabel(String label)
+   {
+      this.label = label;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Template category; it decides opt-out and PHI rules.
+    *******************************************************************************/
+   public EmailTemplateDefinition withCategory(EmailCategory category)
+   {
+      this.category = category;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** MARKDOWN (the default), MJML or HTML.
+    *******************************************************************************/
+   public EmailTemplateDefinition withKind(String kind)
+   {
+      this.kind = kind;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Subject source, which may use variables.
+    *******************************************************************************/
+   public EmailTemplateDefinition withSubject(String subject)
+   {
+      this.subject = subject;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Body source in the template's kind.
+    *******************************************************************************/
+   public EmailTemplateDefinition withBody(String body)
+   {
+      this.body = body;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Optional plain-text source; omitted, it is derived from the body.
+    *******************************************************************************/
+   public EmailTemplateDefinition withTextBody(String textBody)
+   {
+      this.textBody = textBody;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** The variables schema as the JSON list the template table stores. Use this or
+    ** withVariable, not both.
+    *******************************************************************************/
+   public EmailTemplateDefinition withVariablesSchema(String variablesSchema)
+   {
+      this.variablesSchema = variablesSchema;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Declare one variable (type: STRING, NUMBER, DATE, DATE_TIME, BOOLEAN, URL,
+    ** LIST or RECORD; classification: PUBLIC, CONFIDENTIAL, PII, PHI or null).
+    *******************************************************************************/
+   public EmailTemplateDefinition withVariable(String name, String label, String type, boolean required, String classification)
+   {
+      Map<String, Object> entry = new LinkedHashMap<>();
+      entry.put("name", name);
+      entry.put("label", label);
+      entry.put("type", type);
+      entry.put("required", required);
+      entry.put("classification", classification);
+      variables.add(entry);
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Synthetic sample values as a JSON object, used to validate and preview.
+    *******************************************************************************/
+   public EmailTemplateDefinition withSampleVariables(String sampleVariables)
+   {
+      this.sampleVariables = sampleVariables;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** The qbit that owns this template.
+    *******************************************************************************/
+   public EmailTemplateDefinition withOwnerQbit(String ownerQbit)
+   {
+      this.ownerQbit = ownerQbit;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Optional permission, besides the email permission, needed to edit the template.
+    *******************************************************************************/
+   public EmailTemplateDefinition withEditPermissionName(String editPermissionName)
+   {
+      this.editPermissionName = editPermissionName;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** The template name.
+    *******************************************************************************/
+   public String getName()
+   {
+      return (name);
+   }
+
+   /*******************************************************************************
+    ** The template label.
+    *******************************************************************************/
+   public String getLabel()
+   {
+      return (label);
+   }
+
+   /*******************************************************************************
+    ** The template category.
+    *******************************************************************************/
+   public EmailCategory getCategory()
+   {
+      return (category);
+   }
+
+   /*******************************************************************************
+    ** The template kind.
+    *******************************************************************************/
+   public String getKind()
+   {
+      return (kind);
+   }
+
+   /*******************************************************************************
+    ** The subject source.
+    *******************************************************************************/
+   public String getSubject()
+   {
+      return (subject);
+   }
+
+   /*******************************************************************************
+    ** The body source.
+    *******************************************************************************/
+   public String getBody()
+   {
+      return (body);
+   }
+
+   /*******************************************************************************
+    ** The optional plain-text source.
+    *******************************************************************************/
+   public String getTextBody()
+   {
+      return (textBody);
+   }
+
+   /*******************************************************************************
+    ** The schema JSON: the explicit text, or the typed variables as a JSON list.
+    *******************************************************************************/
+   public String getVariablesSchema()
+   {
+      if(variablesSchema != null && !variablesSchema.isBlank())
+      {
+         return (variablesSchema);
+      }
+      return (VariableSchema.json(variables));
+   }
+
+   /*******************************************************************************
+    ** The sample values JSON, defaulting to an empty object.
+    *******************************************************************************/
+   public String getSampleVariables()
+   {
+      return (sampleVariables == null || sampleVariables.isBlank() ? "{}" : sampleVariables);
+   }
+
+   /*******************************************************************************
+    ** The owning qbit.
+    *******************************************************************************/
+   public String getOwnerQbit()
+   {
+      return (ownerQbit);
+   }
+
+   /*******************************************************************************
+    ** The optional edit permission.
+    *******************************************************************************/
+   public String getEditPermissionName()
+   {
+      return (editPermissionName);
+   }
+
+   /*******************************************************************************
+    ** Plain-language reasons the definition is incomplete or malformed, before any
+    ** rendering check. Empty when it passes.
+    *******************************************************************************/
+   public List<String> problems()
+   {
+      List<String> reasons = new ArrayList<>();
+      if(name == null || !name.matches("[a-z][a-z0-9_]{0,99}"))
+      {
+         reasons.add("Name must be snake_case: lowercase letters, digits and underscores, starting with a letter");
+      }
+      if(label == null || label.isBlank())
+      {
+         reasons.add("Label is required");
+      }
+      if(category == null)
+      {
+         reasons.add("Category is required");
+      }
+      if(kind == null || !List.of("MARKDOWN", "MJML", "HTML").contains(kind))
+      {
+         reasons.add("Kind must be MARKDOWN, MJML or HTML");
+      }
+      if(subject == null || subject.isBlank())
+      {
+         reasons.add("Subject is required");
+      }
+      if(body == null || body.isBlank())
+      {
+         reasons.add("Body is required");
+      }
+      if(variablesSchema != null && !variablesSchema.isBlank() && !variables.isEmpty())
+      {
+         reasons.add("Declare variables with withVariablesSchema or withVariable, not both");
+      }
+      else
+      {
+         try
+         {
+            new VariableSchema(getVariablesSchema());
+         }
+         catch(IllegalArgumentException e)
+         {
+            reasons.add(e.getMessage());
+         }
+      }
+      try
+      {
+         VariableSchema.values(sampleVariables);
+      }
+      catch(IllegalArgumentException e)
+      {
+         reasons.add(e.getMessage());
+      }
+      return (reasons);
+   }
+}
@@ -0,0 +1,118 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.template;
+
+
+import java.util.List;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Entry point for other qbits to ship their templates. A template is created
+ ** once; after that this class never touches it, so people's edits are safe.
+ *******************************************************************************/
+public final class EmailTemplates
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private EmailTemplates()
+   {
+   }
+
+   /*******************************************************************************
+    ** Create the template and an ACTIVE version 1 when no template of that name
+    ** exists. The definition is validated like Activate Version (syntax, schema, a
+    ** render with the samples, MJML compile); an invalid one is refused with
+    ** plain reasons and nothing is written. Returns true only when this call
+    ** created the template; the unique name arbitrates concurrent callers.
+    *******************************************************************************/
+   public static boolean ensure(EmailTemplateDefinition definition) throws QException
+   {
+      List<String> reasons = definition.problems();
+      if(!reasons.isEmpty())
+      {
+         throw (new IllegalArgumentException("Template definition refused: " + String.join("; ", reasons)));
+      }
+      return (SystemWrites.call(() -> create(definition)));
+   }
+
+   /*******************************************************************************
+    ** The checked, transactional creation, run as the qbit.
+    *******************************************************************************/
+   private static boolean create(EmailTemplateDefinition definition) throws QException
+   {
+      if(exists(definition.getName()))
+      {
+         return (false);
+      }
+      QRecord template = new QRecord().withValue("name", definition.getName()).withValue("label", definition.getLabel())
+         .withValue("category", definition.getCategory().name()).withValue("kind", definition.getKind()).withValue("phiApproved", false)
+         .withValue("ownerQbit", definition.getOwnerQbit()).withValue("editPermissionName", definition.getEditPermissionName());
+      QRecord version = new QRecord().withValue("versionNumber", 1).withValue("status", "ACTIVE").withValue("locale", "en")
+         .withValue("subject", definition.getSubject()).withValue("bodySource", definition.getBody()).withValue("textSource", definition.getTextBody())
+         .withValue("variablesSchema", definition.getVariablesSchema()).withValue("sampleVariables", definition.getSampleVariables());
+      try
+      {
+         EmailVersions.validate(template, version);
+      }
+      catch(IllegalArgumentException e)
+      {
+         throw (new IllegalArgumentException("Template definition refused: " + e.getMessage()));
+      }
+      try(QBackendTransaction tx = QBackendTransaction.openFor(new InsertInput(EmailTables.TEMPLATE)))
+      {
+         QRecord stored = SystemWrites.insert(EmailTables.TEMPLATE, List.of(template), tx).getRecords().getFirst();
+         if(stored.getValue("id") == null || stored.getErrors() != null && !stored.getErrors().isEmpty())
+         {
+            tx.rollback();
+            if(exists(definition.getName()))
+            {
+               return (false);
+            }
+            throw (new QException("Could not create template " + definition.getName()));
+         }
+         version.withValue("templateId", stored.getValue("id"));
+         QRecord storedVersion = SystemWrites.insert(EmailTables.VERSION, List.of(version), tx).getRecords().getFirst();
+         if(storedVersion.getValue("id") == null || storedVersion.getErrors() != null && !storedVersion.getErrors().isEmpty())
+         {
+            throw (new QException("Could not create template version for " + definition.getName()));
+         }
+         EmailStore.update(EmailTables.TEMPLATE, new QRecord().withValue("id", stored.getValue("id")).withValue("activeVersionId", storedVersion.getValue("id")), tx);
+         tx.commit();
+         return (true);
+      }
+   }
+
+   /*******************************************************************************
+    ** Whether a template of that name is already present.
+    *******************************************************************************/
+   private static boolean exists(String name) throws QException
+   {
+      return (!SystemWrites.query(EmailTables.TEMPLATE, EmailStore.equal("name", name)).isEmpty());
+   }
+}
@@ -0,0 +1,302 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+
+/*******************************************************************************
+ ** Provider payloads become small recipient-level events, never message bodies.
+ *******************************************************************************/
+public final class EmailEventParser
+{
+   /*******************************************************************************
+    ** Provider identifiers remain scoped to the provider during database matching.
+    *******************************************************************************/
+   public record Event(String externalId, String messageId, String providerMessageId, String type, String bounceType,
+      String recipient, Instant occurred, String detail, String url)
+   {
+   }
+
+   /*******************************************************************************
+    ** Parse one provider request, including all SendGrid events and SES recipients.
+    *******************************************************************************/
+   public List<Event> parse(String kind, String body, Instant received)
+   {
+      List<Event> events = new ArrayList<>();
+      switch(kind)
+      {
+         case "SES" -> ses(new JSONObject(body), received, events);
+         case "POSTMARK" -> postmark(new JSONObject(body), received, events);
+         case "MAILGUN" -> mailgun(new JSONObject(body).getJSONObject("event-data"), received, events);
+         case "SENDGRID" ->
+         {
+            JSONArray batch = new JSONArray(body);
+            for(int i = 0; i < batch.length(); i++)
+            {
+               sendgrid(batch.getJSONObject(i), received, events);
+            }
+         }
+         default -> throw new IllegalArgumentException("Unsupported email event kind");
+      }
+      return (events);
+   }
+
+   /*******************************************************************************
+    ** SES recipients live under the specific event, with mail.destination fallback for non-suppressive events only.
+    *******************************************************************************/
+   private static void ses(JSONObject envelope, Instant received, List<Event> events)
+   {
+      if(!"Notification".equals(envelope.getString("Type")))
+      {
+         return;
+      }
+      JSONObject payload = new JSONObject(envelope.getString("Message"));
+      String rawType = payload.optString("eventType", payload.optString("notificationType"));
+      String type = switch(rawType)
+      {
+         case "Delivery" -> "DELIVERED";
+         case "Bounce" -> "BOUNCED";
+         case "Complaint" -> "COMPLAINED";
+         case "Reject" -> "REJECTED";
+         case "DeliveryDelay" -> "DEFERRED";
+         case "Open" -> "OPENED";
+         case "Click" -> "CLICKED";
+         default -> null;
+      };
+      if(type == null)
+      {
+         return;
+      }
+      JSONObject mail = payload.getJSONObject("mail");
+      String key = "DeliveryDelay".equals(rawType) ? "deliveryDelay" : rawType.toLowerCase(Locale.ROOT);
+      JSONObject data = object(payload, key);
+      String bounce = "BOUNCED".equals(type) ? switch(data.optString("bounceType"))
+      {
+         case "Permanent" -> "HARD";
+         case "Transient" -> "SOFT";
+         default -> "UNDETERMINED";
+      } : null;
+      String messageId = null;
+      JSONArray headers = mail.optJSONArray("headers");
+      if(headers != null)
+      {
+         for(int i = 0; i < headers.length(); i++)
+         {
+            JSONObject entry = headers.getJSONObject(i);
+            if("X-QQQ-Email-Id".equalsIgnoreCase(entry.optString("name")))
+            {
+               messageId = entry.optString("value", null);
+            }
+         }
+      }
+      JSONArray recipients = data.optJSONArray(switch(type)
+      {
+         case "BOUNCED" -> "bouncedRecipients";
+         case "COMPLAINED" -> "complainedRecipients";
+         case "DEFERRED" -> "delayedRecipients";
+         default -> "recipients";
+      });
+      if(recipients == null && ("BOUNCED".equals(type) || "COMPLAINED".equals(type)))
+      {
+         // Never widen a bounce or complaint to every destination; the event carries no recipient and is skipped.
+         events.add(new Event(envelope.getString("MessageId"), messageId, mail.getString("messageId"), type, bounce, null,
+            time(data.opt("timestamp"), received), null, null));
+         return;
+      }
+      if(recipients == null)
+      {
+         recipients = mail.getJSONArray("destination");
+      }
+      for(int i = 0; i < recipients.length(); i++)
+      {
+         Object recipient = recipients.get(i);
+         String address = recipient instanceof JSONObject row ? row.getString("emailAddress") : recipient.toString();
+         String reason = recipient instanceof JSONObject row ? row.optString("diagnosticCode", data.optString("bounceSubType", null))
+            : data.optString("reason", data.optString("smtpResponse", data.optString("delayType", data.optString("complaintFeedbackType", null))));
+         events.add(new Event(envelope.getString("MessageId"), messageId, mail.getString("messageId"), type, bounce, address,
+            time(data.opt("timestamp"), received), detail(reason), "CLICKED".equals(type) ? data.optString("link", null) : null));
+      }
+   }
+
+   /*******************************************************************************
+    ** Postmark subscription reactivation is informational, never an unsubscribe.
+    *******************************************************************************/
+   private static void postmark(JSONObject data, Instant received, List<Event> events)
+   {
+      String type = switch(data.getString("RecordType"))
+      {
+         case "Delivery" -> "DELIVERED";
+         case "Bounce" -> "BOUNCED";
+         case "SpamComplaint" -> "COMPLAINED";
+         case "Open" -> "OPENED";
+         case "Click" -> "CLICKED";
+         case "SubscriptionChange" -> data.optBoolean("SuppressSending", false) ? "UNSUBSCRIBED" : null;
+         default -> null;
+      };
+      if(type == null)
+      {
+         return;
+      }
+      String bounce = "BOUNCED".equals(type) ? switch(data.optString("Type"))
+      {
+         case "HardBounce", "SpamNotification", "SpamComplaint", "BadEmailAddress", "ManuallyDeactivated" -> "HARD";
+         case "SoftBounce", "Transient", "DnsError", "MailboxFull" -> "SOFT";
+         default -> "UNDETERMINED";
+      } : null;
+      Object date = data.opt("DeliveredAt");
+      for(String field : List.of("BouncedAt", "ReceivedAt", "ChangedAt"))
+      {
+         if(data.has(field))
+         {
+            date = data.get(field);
+         }
+      }
+      String id = data.has("ID") ? data.get("RecordType") + ":" + data.get("ID") : fingerprint(data);
+      events.add(new Event(id, null, data.optString("MessageID", null), type, bounce,
+         data.optString("Recipient", data.optString("Email", null)), time(date, received),
+         detail(data.optString("Description", data.optString("Details", data.optString("SuppressionReason", null)))),
+         "CLICKED".equals(type) ? data.optString("OriginalLink", null) : null));
+   }
+
+   /*******************************************************************************
+    ** SendGrid retains the provider message id prefix before its delivery suffix.
+    *******************************************************************************/
+   private static void sendgrid(JSONObject data, Instant received, List<Event> events)
+   {
+      String type = switch(data.getString("event"))
+      {
+         case "delivered" -> "DELIVERED";
+         case "bounce" -> "BOUNCED";
+         case "dropped" -> "REJECTED";
+         case "deferred" -> "DEFERRED";
+         case "spamreport" -> "COMPLAINED";
+         case "open" -> "OPENED";
+         case "click" -> "CLICKED";
+         case "unsubscribe", "group_unsubscribe" -> "UNSUBSCRIBED";
+         default -> null;
+      };
+      if(type == null)
+      {
+         return;
+      }
+      String bounce = "BOUNCED".equals(type) ? ("blocked".equals(data.optString("type")) ? "SOFT" : "HARD") : null;
+      String providerMessage = data.optString("sg_message_id", null);
+      if(providerMessage != null && providerMessage.contains("."))
+      {
+         providerMessage = providerMessage.substring(0, providerMessage.indexOf('.'));
+      }
+      events.add(new Event(data.optString("sg_event_id", fingerprint(data)),
+         data.optString("qqq_email_id", object(data, "custom_args").optString("qqq_email_id", null)), providerMessage, type, bounce,
+         data.optString("email", null), time(data.opt("timestamp"), received), detail(data.optString("reason", data.optString("response", null))),
+         "CLICKED".equals(type) ? data.optString("url", null) : null));
+   }
+
+   /*******************************************************************************
+    ** Mailgun MIME ids are normalized identically to the outbound transport.
+    *******************************************************************************/
+   private static void mailgun(JSONObject data, Instant received, List<Event> events)
+   {
+      String type = switch(data.getString("event"))
+      {
+         case "delivered" -> "DELIVERED";
+         case "failed" -> "BOUNCED";
+         case "complained" -> "COMPLAINED";
+         case "opened" -> "OPENED";
+         case "clicked" -> "CLICKED";
+         case "unsubscribed" -> "UNSUBSCRIBED";
+         default -> null;
+      };
+      if(type == null)
+      {
+         return;
+      }
+      String bounce = "BOUNCED".equals(type) ? switch(data.optString("severity"))
+      {
+         case "permanent" -> "HARD";
+         case "temporary" -> "SOFT";
+         default -> "UNDETERMINED";
+      } : null;
+      JSONObject variables = object(data, "user-variables");
+      String providerMessage = object(object(data, "message"), "headers").optString("message-id", null);
+      if(providerMessage != null)
+      {
+         providerMessage = providerMessage.replaceAll("^<|>$", "");
+      }
+      events.add(new Event(data.optString("id", fingerprint(data)), variables.optString("qqq_email_id", variables.optString("v:qqq_email_id", null)),
+         providerMessage, type, bounce, data.optString("recipient", null), time(data.opt("timestamp"), received),
+         detail(data.optString("reason", object(data, "delivery-status").optString("message", null))), "CLICKED".equals(type) ? data.optString("url", null) : null));
+   }
+
+   /*******************************************************************************
+    ** Optional nested objects contain no executable data.
+    *******************************************************************************/
+   private static JSONObject object(JSONObject parent, String key)
+   {
+      JSONObject value = parent.optJSONObject(key);
+      return (value == null ? new JSONObject() : value);
+   }
+
+   /*******************************************************************************
+    ** Fall back to receipt time only when the provider supplied no timestamp.
+    *******************************************************************************/
+   private static Instant time(Object value, Instant fallback)
+   {
+      if(value == null || JSONObject.NULL.equals(value))
+      {
+         return (fallback);
+      }
+      if(value instanceof Number number)
+      {
+         return (Instant.ofEpochMilli((long) (number.doubleValue() * 1000)));
+      }
+      return (OffsetDateTime.parse(value.toString()).toInstant());
+   }
+
+   /*******************************************************************************
+    ** Fallback identity for providers without a distinct event identifier.
+    *******************************************************************************/
+   private static String fingerprint(JSONObject data)
+   {
+      return (WebhookVerification.hash(data.toString().getBytes(StandardCharsets.UTF_8)));
+   }
+
+   /*******************************************************************************
+    ** Mask complete addresses before truncation, including quoted local parts.
+    *******************************************************************************/
+   public static String detail(String reason)
+   {
+      if(reason == null)
+      {
+         return (null);
+      }
+      String masked = reason.replaceAll("(?i)(?:\"[^\"]*\"|[^\\s<>(),;:]+)@[^\\s<>(),;:]+", "[address]");
+      return (masked.substring(0, Math.min(500, masked.length())));
+   }
+}
@@ -0,0 +1,333 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.Set;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qbits.email.delivery.EmailAlerts;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import jakarta.mail.internet.InternetAddress;
+
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** Bounded, replica-safe database transactions apply inbox rows exactly once.
+ ** No network or external effects occur while the row's provider is locked.
+ *******************************************************************************/
+public final class EmailInboxProcessor
+{
+   private static final QLogger LOG = QLogger.getLogger(EmailInboxProcessor.class);
+
+   private static final int MAX_ATTEMPTS = 6;
+
+   /*******************************************************************************
+    ** Poll due rows; transaction rollback releases claims after a worker crash.
+    *******************************************************************************/
+   public int runOnce() throws QException
+   {
+      int processed = 0;
+      for(QRecord row : SystemWrites.query(EmailTables.INBOX, EmailStore.equal("status", "PENDING")
+         .withCriteria("nextAttemptDate", QCriteriaOperator.LESS_THAN_OR_EQUALS, EmailRuntime.clock().instant())
+         .withOrderBy(new QFilterOrderBy("id")).withLimit(EmailRuntime.config().getBatchSize())))
+      {
+         if(process(row))
+         {
+            processed++;
+         }
+      }
+      return (processed);
+   }
+
+   /*******************************************************************************
+    ** Re-read eligibility after the lock, including the observed attempt number.
+    *******************************************************************************/
+   private static boolean due(QRecord current, QRecord observed)
+   {
+      return (current != null && "PENDING".equals(current.getValueString("status"))
+         && Objects.equals(current.getValue("attempts"), observed.getValue("attempts"))
+         && current.getValueInstant("nextAttemptDate") != null && !current.getValueInstant("nextAttemptDate").isAfter(EmailRuntime.clock().instant()));
+   }
+
+   /*******************************************************************************
+    ** All event, message, suppression and receipt changes commit together.
+    *******************************************************************************/
+   private boolean process(QRecord observed) throws QException
+   {
+      boolean complaint = false;
+      try(QBackendTransaction tx = QBackendTransaction.openFor(new UpdateInput(EmailTables.PROVIDER)))
+      {
+         QRecord provider = EmailTrackingStore.lock(observed.getValueInteger("providerId"), tx);
+         QRecord row = EmailStore.get(EmailTables.INBOX, observed.getValueInteger("id"), tx);
+         if(!due(row, observed))
+         {
+            return (false);
+         }
+         int applied = 0;
+         int unmatched = 0;
+         int skipped = 0;
+         for(EmailEventParser.Event event : new EmailEventParser().parse(row.getValueString("kind"), row.getValueString("payloadJson"), row.getValueInstant("receivedDate")))
+         {
+            if(("OPENED".equals(event.type()) && !Boolean.TRUE.equals(provider.getValueBoolean("trackOpens")))
+               || ("CLICKED".equals(event.type()) && !Boolean.TRUE.equals(provider.getValueBoolean("trackClicks"))))
+            {
+               continue;
+            }
+            QRecord message = match(event, provider.getValueInteger("id"), tx);
+            if(message == null || "SENDING".equals(message.getValueString("status")))
+            {
+               unmatched++;
+               continue;
+            }
+            String recipient = normalize(event.recipient());
+            if(recipient == null || !recipients(message).contains(recipient))
+            {
+               LOG.info("Ignoring an email event whose recipient is not on the message", logPair("messageId", message.getValue("id")));
+               skipped++;
+               continue;
+            }
+            if(Boolean.TRUE.equals(message.getValueBoolean("phiApproved")) && List.of("OPENED", "CLICKED").contains(event.type()))
+            {
+               continue;
+            }
+            String key = provider.getValueInteger("id") + ":" + WebhookVerification.hash((event.externalId() + "\n" + event.type() + "\n"
+               + recipient).getBytes(StandardCharsets.UTF_8));
+            QRecord stored = EmailTrackingStore.event(message, event, key, tx);
+            if(stored == null)
+            {
+               continue;
+            }
+            applied++;
+            updateMessage(message, stored, tx);
+            if("COMPLAINED".equals(event.type()) || ("BOUNCED".equals(event.type()) && "HARD".equals(event.bounceType())))
+            {
+               EmailTrackingStore.suppress(message, stored, "ALL", "COMPLAINED".equals(event.type()) ? "COMPLAINT" : "HARD_BOUNCE", tx);
+               complaint |= "COMPLAINED".equals(event.type());
+            }
+            else if("UNSUBSCRIBED".equals(event.type()))
+            {
+               EmailTrackingStore.suppress(message, stored, message.getValueString("category"), "UNSUBSCRIBE", tx);
+            }
+         }
+
+         int attempts = row.getValueInteger("attempts") + 1;
+         boolean retry = unmatched > 0 && attempts < MAX_ATTEMPTS;
+         List<String> notes = new ArrayList<>();
+         if(unmatched > 0)
+         {
+            notes.add(unmatched + (unmatched == 1 ? " event did not match a message" : " events did not match a message"));
+         }
+         if(skipped > 0)
+         {
+            notes.add(skipped + (skipped == 1 ? " event was skipped (recipient not on the message)" : " events were skipped (recipient not on the message)"));
+         }
+         EmailStore.update(EmailTables.INBOX, new QRecord().withValue("id", row.getValue("id"))
+            .withValue("status", retry ? "PENDING" : (applied == 0 && unmatched == 0 ? "IGNORED" : "PROCESSED")).withValue("attempts", attempts)
+            .withValue("processedDate", retry ? null : EmailRuntime.clock().instant())
+            .withValue("nextAttemptDate", retry ? EmailRuntime.clock().instant().plusSeconds(60L * EmailRuntime.config().retryMinutesAfterAttempt(attempts)) : null)
+            .withValue("lastError", notes.isEmpty() ? null : String.join("; ", notes)), tx);
+         tx.commit();
+      }
+      catch(Exception e)
+      {
+         fail(observed);
+         return (false);
+      }
+      if(complaint)
+      {
+         complaintAlert();
+      }
+      return (true);
+   }
+
+   /*******************************************************************************
+    ** Canonical mailbox, or null when the provider supplied none that is valid.
+    *******************************************************************************/
+   private static String normalize(String value)
+   {
+      if(value == null || value.isBlank())
+      {
+         return (null);
+      }
+      try
+      {
+         return (EmailTrackingStore.address(value));
+      }
+      catch(IllegalArgumentException e)
+      {
+         return (null);
+      }
+   }
+
+   /*******************************************************************************
+    ** The canonical to, cc and bcc mailboxes of a message; only these may be
+    ** suppressed by an event matched to it.
+    *******************************************************************************/
+   private static Set<String> recipients(QRecord message)
+   {
+      Set<String> set = new HashSet<>();
+      for(String field : List.of("toAddresses", "ccAddresses", "bccAddresses"))
+      {
+         String value = message.getValueString(field);
+         if(value == null || value.isBlank())
+         {
+            continue;
+         }
+         try
+         {
+            for(InternetAddress address : InternetAddress.parse(value, false))
+            {
+               set.add(address.getAddress().toLowerCase(Locale.ROOT));
+            }
+         }
+         catch(jakarta.mail.internet.AddressException e)
+         {
+            // A malformed stored envelope contributes no recipients.
+         }
+      }
+      return (set);
+   }
+
+   /*******************************************************************************
+    ** Match our id first, then the provider id; neither may cross providers.
+    *******************************************************************************/
+   private static QRecord match(EmailEventParser.Event event, Integer providerId, QBackendTransaction tx) throws QException
+   {
+      if(event.messageId() != null && event.messageId().matches("[0-9]{1,10}") && Long.parseLong(event.messageId()) <= Integer.MAX_VALUE)
+      {
+         QRecord message = EmailStore.get(EmailTables.MESSAGE, Integer.valueOf(event.messageId()), tx);
+         if(message != null && providerId.equals(message.getValueInteger("providerId")))
+         {
+            return (message);
+         }
+      }
+      if(event.providerMessageId() == null || event.providerMessageId().isBlank())
+      {
+         return (null);
+      }
+      List<QRecord> rows = SystemWrites.query(EmailTables.MESSAGE, EmailStore.equal("providerId", providerId)
+         .withCriteria("providerMessageId", QCriteriaOperator.EQUALS, event.providerMessageId()), tx);
+      return (rows.size() == 1 ? rows.getFirst() : null);
+   }
+
+   /*******************************************************************************
+    ** Complaint outranks hard bounce, which outranks delivery; never regress.
+    *******************************************************************************/
+   private static void updateMessage(QRecord message, QRecord event, QBackendTransaction tx) throws QException
+   {
+      String type = event.getValueString("eventType");
+      String status = message.getValueString("status");
+      String field = switch(type)
+      {
+         case "DELIVERED" -> "SENT".equals(status) ? "deliveredDate" : null;
+         case "BOUNCED" -> "HARD".equals(event.getValueString("bounceType")) && List.of("SENT", "DELIVERED").contains(status) ? "bouncedDate" : null;
+         case "COMPLAINED" -> List.of("SENT", "DELIVERED", "BOUNCED").contains(status) ? "complainedDate" : null;
+         default -> null;
+      };
+      if(field != null)
+      {
+         EmailStore.update(EmailTables.MESSAGE, new QRecord().withValue("id", message.getValue("id")).withValue("status", type)
+            .withValue(field, event.getValue("occurredDate")), tx);
+      }
+   }
+
+   /*******************************************************************************
+    ** After rollback, only the still-due observed attempt may consume retry budget.
+    *******************************************************************************/
+   private static void fail(QRecord observed) throws QException
+   {
+      try(QBackendTransaction tx = QBackendTransaction.openFor(new UpdateInput(EmailTables.PROVIDER)))
+      {
+         EmailTrackingStore.lock(observed.getValueInteger("providerId"), tx);
+         QRecord row = EmailStore.get(EmailTables.INBOX, observed.getValueInteger("id"), tx);
+         if(due(row, observed))
+         {
+            int attempts = row.getValueInteger("attempts") + 1;
+            EmailStore.update(EmailTables.INBOX, new QRecord().withValue("id", row.getValue("id")).withValue("attempts", attempts)
+               .withValue("status", attempts >= MAX_ATTEMPTS ? "FAILED" : "PENDING").withValue("lastError", "Email event processing failed; check message matching and payload validity.")
+               .withValue("nextAttemptDate", attempts >= MAX_ATTEMPTS ? null : EmailRuntime.clock().instant().plusSeconds(60L * EmailRuntime.config().retryMinutesAfterAttempt(attempts))), tx);
+            tx.commit();
+         }
+      }
+   }
+
+   /*******************************************************************************
+    ** Standard Reprocess action rechecks FAILED under the same transaction lock.
+    *******************************************************************************/
+   public static boolean reprocess(Integer id) throws QException
+   {
+      QRecord selected = EmailStore.get(EmailTables.INBOX, id, null);
+      if(selected == null)
+      {
+         return (false);
+      }
+      try(QBackendTransaction tx = QBackendTransaction.openFor(new UpdateInput(EmailTables.PROVIDER)))
+      {
+         EmailTrackingStore.lock(selected.getValueInteger("providerId"), tx);
+         QRecord row = EmailStore.get(EmailTables.INBOX, id, tx);
+         if(!"FAILED".equals(row.getValueString("status")))
+         {
+            return (false);
+         }
+         EmailStore.update(EmailTables.INBOX, new QRecord().withValue("id", id).withValue("status", "PENDING").withValue("attempts", 0)
+            .withValue("nextAttemptDate", EmailRuntime.clock().instant()).withValue("lastError", null).withValue("processedDate", null), tx);
+         tx.commit();
+         return (true);
+      }
+   }
+
+   /*******************************************************************************
+    ** Optional alerting cannot turn a committed event into a failed inbox row.
+    *******************************************************************************/
+   private static void complaintAlert()
+   {
+      try
+      {
+         Class.forName("com.kingsrook.qqq.backend.core.alerts.QAlerts");
+         if(SystemWrites.count(EmailTables.EVENT, EmailStore.equal("eventType", "COMPLAINED")
+            .withCriteria("occurredDate", QCriteriaOperator.GREATER_THAN_OR_EQUALS, EmailRuntime.clock().instant().minusSeconds(86400))
+            .withCriteria("occurredDate", QCriteriaOperator.LESS_THAN_OR_EQUALS, EmailRuntime.clock().instant())) > 3)
+         {
+            EmailAlerts.raise(EmailAlerts.COMPLAINT, 0);
+         }
+      }
+      catch(Exception | LinkageError e)
+      {
+         // Alert integration is optional; event data is already committed.
+      }
+   }
+}
@@ -0,0 +1,218 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Locale;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import jakarta.mail.internet.InternetAddress;
+
+
+/*******************************************************************************
+ ** DB-only receipt, event deduplication and suppression writes.
+ *******************************************************************************/
+public final class EmailTrackingStore
+{
+   /*******************************************************************************
+    ** Distinguish an authenticated retry from reuse of a Mailgun nonce.
+    *******************************************************************************/
+   public enum Receipt
+   {
+      RECEIVED, DUPLICATE, REPLAY
+   }
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private EmailTrackingStore()
+   {
+   }
+
+   /*******************************************************************************
+    ** Serialize address upserts across providers, then acquire the dispatcher lock.
+    ** Every tracking writer uses this order and does no network I/O while locked.
+    *******************************************************************************/
+   public static QRecord lock(Integer providerId, QBackendTransaction tx) throws QException
+   {
+      List<QRecord> providers = SystemWrites.query(EmailTables.PROVIDER, new QQueryFilter().withOrderBy(new QFilterOrderBy("id")).withLimit(1), tx);
+      if(providers.isEmpty())
+      {
+         throw (new QException("Email provider unavailable"));
+      }
+      EmailStore.lockProvider(providers.getFirst().getValueInteger("id"), tx);
+      QRecord provider = EmailStore.lockProvider(providerId, tx);
+      if(provider == null)
+      {
+         throw (new QException("Email provider unavailable"));
+      }
+      return (provider);
+   }
+
+   /*******************************************************************************
+    ** A provider lock avoids expected duplicate SQL errors (and payload logging).
+    ** Unique constraints remain the final database guard.
+    *******************************************************************************/
+   public static Receipt receive(QRecord provider, String externalId, String nonce, byte[] bytes, String note) throws QException
+   {
+      Integer providerId = provider.getValueInteger("id");
+      try(QBackendTransaction tx = QBackendTransaction.openFor(new UpdateInput(EmailTables.PROVIDER)))
+      {
+         EmailStore.lockProvider(providerId, tx);
+         QQueryFilter filter = EmailStore.equal("providerId", providerId);
+         if(nonce != null && SystemWrites.count(EmailTables.INBOX, filter.clone().withCriteria("nonce", QCriteriaOperator.EQUALS, nonce), tx) > 0)
+         {
+            return (Receipt.REPLAY);
+         }
+         if(SystemWrites.count(EmailTables.INBOX, filter.withCriteria("externalEventId", QCriteriaOperator.EQUALS, externalId), tx) > 0)
+         {
+            return (Receipt.DUPLICATE);
+         }
+         var now = EmailRuntime.clock().instant();
+         QRecord row = new QRecord().withValue("providerId", providerId).withValue("kind", provider.getValue("kind"))
+            .withValue("externalEventId", externalId).withValue("nonce", nonce).withValue("status", note == null ? "PENDING" : "IGNORED")
+            .withValue("attempts", 0).withValue("nextAttemptDate", note == null ? now : null).withValue("lastError", note)
+            .withValue("payloadJson", new String(bytes, StandardCharsets.UTF_8)).withValue("payloadSha256", WebhookVerification.hash(bytes))
+            .withValue("receivedDate", now).withValue("processedDate", note == null ? null : now);
+         copySecurity(provider, row);
+         insert(EmailTables.INBOX, row, tx);
+         tx.commit();
+         return (Receipt.RECEIVED);
+      }
+      catch(QException e)
+      {
+         // Re-query after rollback: a unique violation aborts a PostgreSQL transaction.
+         if(nonce != null && SystemWrites.count(EmailTables.INBOX, EmailStore.equal("providerId", providerId)
+            .withCriteria("nonce", QCriteriaOperator.EQUALS, nonce)) > 0)
+         {
+            return (Receipt.REPLAY);
+         }
+         if(SystemWrites.count(EmailTables.INBOX, EmailStore.equal("providerId", providerId)
+            .withCriteria("externalEventId", QCriteriaOperator.EQUALS, externalId)) > 0)
+         {
+            return (Receipt.DUPLICATE);
+         }
+         throw (new QException("Email inbox write failed"));
+      }
+   }
+
+   /*******************************************************************************
+    ** Fail atomically without exposing input values or database diagnostics.
+    *******************************************************************************/
+   public static QRecord insert(String table, QRecord row, QBackendTransaction tx) throws QException
+   {
+      List<QRecord> records = SystemWrites.insert(table, List.of(row), tx).getRecords();
+      if(records.isEmpty() || records.getFirst().getValueInteger("id") == null || (records.getFirst().getErrors() != null && !records.getFirst().getErrors().isEmpty()))
+      {
+         throw (new QException("Email tracking write failed"));
+      }
+      return (records.getFirst());
+   }
+
+   /*******************************************************************************
+    ** Preserve host tenant security on child and suppression records.
+    *******************************************************************************/
+   public static void copySecurity(QRecord source, QRecord target)
+   {
+      for(var field : EmailRuntime.config().getSecurityFields())
+      {
+         target.withValue(field.getName(), source.getValue(field.getName()));
+      }
+   }
+
+   /*******************************************************************************
+    ** Insert a recipient event once even across overlapping provider batches.
+    *******************************************************************************/
+   public static QRecord event(QRecord message, EmailEventParser.Event event, String key, QBackendTransaction tx) throws QException
+   {
+      if(SystemWrites.count(EmailTables.EVENT, EmailStore.equal("eventKey", key), tx) > 0)
+      {
+         return (null);
+      }
+      QRecord row = new QRecord().withValue("messageId", message.getValue("id")).withValue("providerId", message.getValue("providerId"))
+         .withValue("eventType", event.type()).withValue("bounceType", event.bounceType()).withValue("recipient", address(event.recipient()))
+         .withValue("occurredDate", event.occurred()).withValue("detail", event.detail()).withValue("url", "CLICKED".equals(event.type()) ? event.url() : null)
+         .withValue("eventKey", key);
+      copySecurity(message, row);
+      return (insert(EmailTables.EVENT, row, tx));
+   }
+
+   /*******************************************************************************
+    ** Upsert without weakening a complaint or shortening an existing suppression.
+    ** All callers hold the common tracking lock throughout their transaction.
+    *******************************************************************************/
+   public static void suppress(QRecord message, QRecord event, String scope, String reason, QBackendTransaction tx) throws QException
+   {
+      if("UNSUBSCRIBE".equals(reason) && !List.of("MARKETING", "NOTIFICATION").contains(scope == null ? "" : scope))
+      {
+         return;
+      }
+      String address = event.getValueString("recipient");
+      List<QRecord> existing = SystemWrites.query(EmailTables.SUPPRESSION, EmailStore.equal("address", address)
+         .withCriteria("scope", QCriteriaOperator.EQUALS, scope), tx);
+      QRecord row = new QRecord().withValue("address", address).withValue("scope", scope).withValue("reason", reason)
+         .withValue("sourceMessageId", message.getValue("id")).withValue("sourceEventId", event.getValue("id")).withValue("expiresDate", null);
+      copySecurity(message, row);
+      if(existing.isEmpty())
+      {
+         row.withValue("createdDate", EmailRuntime.clock().instant());
+         insert(EmailTables.SUPPRESSION, row, tx);
+      }
+      else
+      {
+         QRecord old = existing.getFirst();
+         if("COMPLAINT".equals(old.getValueString("reason")))
+         {
+            row.withValue("reason", "COMPLAINT");
+         }
+         row.withValue("id", old.getValue("id"));
+         EmailStore.update(EmailTables.SUPPRESSION, row, tx);
+      }
+   }
+
+   /*******************************************************************************
+    ** Canonical mailbox for event deduplication and suppression lookup.
+    *******************************************************************************/
+   public static String address(String value)
+   {
+      try
+      {
+         InternetAddress address = new InternetAddress(value, true);
+         address.validate();
+         return (address.getAddress().toLowerCase(Locale.ROOT));
+      }
+      catch(Exception e)
+      {
+         throw (new IllegalArgumentException("Email event recipient is invalid"));
+      }
+   }
+}
@@ -0,0 +1,132 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import java.nio.charset.StandardCharsets;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.render.EmailUnsubscribeLinks;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.template.EmailTemplateEngine;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import org.json.JSONObject;
+
+
+/*******************************************************************************
+ ** Stateless token validation and atomic, idempotent category opt-out.
+ *******************************************************************************/
+public final class EmailUnsubscribe
+{
+   /*******************************************************************************
+    ** Resolve the signed message and its single recipient, without writes.
+    *******************************************************************************/
+   public QRecord resolve(String token) throws QException
+   {
+      return (resolve(token, null));
+   }
+
+   /*******************************************************************************
+    ** Recheck the binding against the same transaction used for suppression.
+    *******************************************************************************/
+   private QRecord resolve(String token, QBackendTransaction transaction) throws QException
+   {
+      JSONObject payload = EmailUnsubscribeLinks.verify(token);
+      if(payload == null)
+      {
+         return (null);
+      }
+      try
+      {
+         QRecord message = EmailStore.get(EmailTables.MESSAGE, Integer.valueOf(payload.getString("messageId")), transaction);
+         if(message == null || !payload.getString("category").equals(message.getValueString("category")))
+         {
+            return (null);
+         }
+         String recipient = EmailUnsubscribeLinks.recipient(message.getValueString("toAddresses"), message.getValueString("ccAddresses"), message.getValueString("bccAddresses"));
+         String hash = payload.getString("addressHash");
+         // Safe legacy links remain valid only after the single-mailbox check above.
+         String legacyEnvelope = message.getValueString("toAddresses") + "," + message.getValueString("ccAddresses") + "," + message.getValueString("bccAddresses");
+         boolean matches = WebhookVerification.equal(hash, WebhookVerification.hash(recipient.getBytes(StandardCharsets.UTF_8)))
+            || WebhookVerification.equal(hash, WebhookVerification.hash(legacyEnvelope.getBytes(StandardCharsets.UTF_8)));
+         return (matches ? message : null);
+      }
+      catch(IllegalArgumentException | org.json.JSONException e)
+      {
+         return (null);
+      }
+   }
+
+   /*******************************************************************************
+    ** Opt out only the signed mailbox; reject changed or multi-recipient envelopes.
+    ** Never display those addresses on the public page or emit them to logs.
+    *******************************************************************************/
+   public void unsubscribe(String token) throws QException
+   {
+      QRecord initial = resolve(token);
+      if(initial == null)
+      {
+         throw (new IllegalArgumentException("Invalid unsubscribe link"));
+      }
+      try(QBackendTransaction tx = QBackendTransaction.openFor(new UpdateInput(EmailTables.PROVIDER)))
+      {
+         EmailTrackingStore.lock(initial.getValueInteger("providerId"), tx);
+         QRecord message = resolve(token, tx);
+         if(message == null || !initial.getValueInteger("providerId").equals(message.getValueInteger("providerId")))
+         {
+            throw (new IllegalArgumentException("Invalid unsubscribe link"));
+         }
+         String recipient = EmailUnsubscribeLinks.recipient(message.getValueString("toAddresses"), message.getValueString("ccAddresses"), message.getValueString("bccAddresses"));
+         String key = "unsubscribe:" + message.getValueInteger("id") + ":" + WebhookVerification.hash(recipient.getBytes(StandardCharsets.UTF_8));
+         var event = new EmailEventParser.Event(key, message.getValueString("id"), null, "UNSUBSCRIBED", null, recipient, EmailRuntime.clock().instant(), null, null);
+         QRecord stored = EmailTrackingStore.event(message, event, key, tx);
+         if(stored == null)
+         {
+            stored = com.kingsrook.qbits.email.support.SystemWrites.query(EmailTables.EVENT, EmailStore.equal("eventKey", key), tx).getFirst();
+         }
+         EmailTrackingStore.suppress(message, stored, message.getValueString("category"), "UNSUBSCRIBE", tx);
+         tx.commit();
+      }
+   }
+
+   /*******************************************************************************
+    ** Self-contained accessible page; an empty form action posts to this same URL.
+    *******************************************************************************/
+   public String page(QRecord message, boolean confirmed) throws QException
+   {
+      QRecord brand = message.getValueInteger("brandId") == null ? null : EmailStore.get(EmailTables.BRAND, message.getValueInteger("brandId"), null);
+      String label = brand == null || brand.getValueString("label") == null ? EmailRuntime.config().getAppLabel() : brand.getValueString("label");
+      String category = "MARKETING".equals(message.getValueString("category")) ? "Marketing" : "Notification";
+      String question = confirmed ? "You are unsubscribed from " + category + " email from " + label + "."
+         : "Unsubscribe from " + category + " email from " + label + "?";
+      return ("<!doctype html><html lang=\"en\"><head><meta charset=\"utf-8\"><meta name=\"viewport\" content=\"width=device-width,initial-scale=1\">"
+         + "<meta name=\"color-scheme\" content=\"light dark\"><title>Email preferences</title><style>"
+         + ":root{color-scheme:light dark}*{box-sizing:border-box}body{margin:0;padding:20px;font:18px/1.5 system-ui;background:#fff;color:#172033}"
+         + "main{max-width:560px;margin:40px auto;overflow-wrap:anywhere}h1{font-size:1.5rem}button{max-width:100%;padding:12px 20px;font:inherit;background:#174ea6;color:#fff;border:0;border-radius:6px;cursor:pointer}"
+         + "@media(prefers-color-scheme:dark){body{background:#111827;color:#f9fafb}button{background:#a9c7ff;color:#111827}}</style></head><body><main><p>"
+         + EmailTemplateEngine.escape(label) + "</p><h1>" + EmailTemplateEngine.escape(question) + "</h1>"
+         + (confirmed ? "" : "<form method=\"post\"><input type=\"hidden\" name=\"confirm\" value=\"unsubscribe\"><button type=\"submit\">Unsubscribe</button></form>")
+         + "</main></body></html>");
+   }
+}
@@ -0,0 +1,65 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.HexFormat;
+import org.json.JSONObject;
+
+
+/*******************************************************************************
+ ** Authenticate Mailgun; the durable inbox separately arbitrates nonce replay.
+ *******************************************************************************/
+public final class MailgunVerifier
+{
+   /*******************************************************************************
+    ** Verify timestamp, token and hexadecimal HMAC without retaining the key.
+    *******************************************************************************/
+   public boolean verify(JSONObject payload, String secret, Instant now)
+   {
+      if(secret == null || secret.isBlank())
+      {
+         return (false);
+      }
+      try
+      {
+         JSONObject signed = payload.getJSONObject("signature");
+         String timestamp = signed.get("timestamp").toString();
+         String token = signed.getString("token");
+         if(token.isBlank() || token.length() > 500 || !WebhookVerification.fresh(timestamp, now))
+         {
+            return (false);
+         }
+         Mac mac = Mac.getInstance("HmacSHA256");
+         mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+         String expected = HexFormat.of().formatHex(mac.doFinal((timestamp + token).getBytes(StandardCharsets.UTF_8)));
+         return (WebhookVerification.equal(expected, signed.getString("signature")));
+      }
+      catch(Exception e)
+      {
+         return (false);
+      }
+   }
+}
@@ -0,0 +1,52 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+
+
+/*******************************************************************************
+ ** Postmark authenticates with fixed username postmark and a resolved password.
+ *******************************************************************************/
+public final class PostmarkVerifier
+{
+   /*******************************************************************************
+    ** Verify HTTP Basic credentials without logging or returning either value.
+    *******************************************************************************/
+   public boolean verify(String authorization, String secret)
+   {
+      if(secret == null || secret.isBlank() || authorization == null || !authorization.regionMatches(true, 0, "Basic ", 0, 6))
+      {
+         return (false);
+      }
+      try
+      {
+         String credentials = new String(Base64.getDecoder().decode(authorization.substring(6)), StandardCharsets.UTF_8);
+         return (WebhookVerification.equal("postmark:" + secret, credentials));
+      }
+      catch(IllegalArgumentException e)
+      {
+         return (false);
+      }
+   }
+}
@@ -0,0 +1,60 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import java.nio.charset.StandardCharsets;
+import java.security.KeyFactory;
+import java.security.Signature;
+import java.security.spec.X509EncodedKeySpec;
+import java.time.Instant;
+import java.util.Base64;
+
+
+/*******************************************************************************
+ ** Verify SendGrid's timestamp and raw body with its DER encoded EC public key.
+ *******************************************************************************/
+public final class SendGridVerifier
+{
+   /*******************************************************************************
+    ** The timestamp bytes precede the unchanged request body.
+    *******************************************************************************/
+   public boolean verify(byte[] body, String signature, String timestamp, String secret, Instant now)
+   {
+      if(secret == null || secret.isBlank() || signature == null || !WebhookVerification.fresh(timestamp, now))
+      {
+         return (false);
+      }
+      try
+      {
+         var key = KeyFactory.getInstance("EC").generatePublic(new X509EncodedKeySpec(Base64.getDecoder().decode(secret)));
+         Signature verifier = Signature.getInstance("SHA256withECDSA");
+         verifier.initVerify(key);
+         verifier.update(timestamp.getBytes(StandardCharsets.UTF_8));
+         verifier.update(body);
+         return (verifier.verify(Base64.getDecoder().decode(signature)));
+      }
+      catch(Exception e)
+      {
+         return (false);
+      }
+   }
+}
@@ -0,0 +1,219 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpRequest;
+import java.nio.charset.StandardCharsets;
+import java.security.Signature;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import com.kingsrook.qbits.email.transport.EmailHttp;
+import org.json.JSONObject;
+
+
+/*******************************************************************************
+ ** SNS verification with HTTPS AWS-only certificate retrieval, memory caching (including
+ ** a five-minute cache of fetch failures) and a freshness window on the signed Timestamp.
+ *******************************************************************************/
+public final class SnsVerifier
+{
+   private final EmailHttp http;
+   private static final Duration MAX_AGE = Duration.ofHours(1);
+   private static final Duration MAX_FUTURE = Duration.ofMinutes(5);
+   private static final Duration FAILURE_TTL = Duration.ofMinutes(5);
+
+   private final Map<String, X509Certificate> certificates = new ConcurrentHashMap<>();
+   private final Map<String, Instant> failures = new ConcurrentHashMap<>();
+
+   /*******************************************************************************
+    ** Default bounded HTTPS transport without redirects.
+    *******************************************************************************/
+   public SnsVerifier()
+   {
+      this(EmailHttp::sendJdk);
+   }
+
+   /*******************************************************************************
+    ** Inject HTTP for deterministic certificate and confirmation tests.
+    *******************************************************************************/
+   public SnsVerifier(EmailHttp http)
+   {
+      this.http = http;
+   }
+
+   /*******************************************************************************
+    ** Strict URL boundary applied before every network operation.
+    *******************************************************************************/
+   public static boolean allowedUrl(String value, boolean certificate)
+   {
+      try
+      {
+         URI uri = URI.create(value);
+         return ("https".equals(uri.getScheme()) && uri.getHost() != null
+            && uri.getHost().matches("^sns\\.[a-z0-9-]+\\.amazonaws\\.com(\\.cn)?$")
+            && uri.getUserInfo() == null && (uri.getPort() == -1 || uri.getPort() == 443) && uri.getFragment() == null
+            && (!certificate || (uri.getPath().endsWith(".pem") && uri.getQuery() == null)));
+      }
+      catch(RuntimeException e)
+      {
+         return (false);
+      }
+   }
+
+   /*******************************************************************************
+    ** AWS's fixed order includes the final newline and optional Subject only.
+    *******************************************************************************/
+   public static String canonical(JSONObject payload)
+   {
+      String type = payload.getString("Type");
+      List<String> fields;
+      if("Notification".equals(type))
+      {
+         fields = payload.has("Subject") ? List.of("Message", "MessageId", "Subject", "Timestamp", "TopicArn", "Type")
+            : List.of("Message", "MessageId", "Timestamp", "TopicArn", "Type");
+      }
+      else if("SubscriptionConfirmation".equals(type) || "UnsubscribeConfirmation".equals(type))
+      {
+         fields = List.of("Message", "MessageId", "SubscribeURL", "Timestamp", "Token", "TopicArn", "Type");
+      }
+      else
+      {
+         throw (new IllegalArgumentException("Unsupported SNS envelope"));
+      }
+      StringBuilder value = new StringBuilder();
+      for(String field : fields)
+      {
+         value.append(field).append('\n').append(payload.getString(field)).append('\n');
+      }
+      return (value.toString());
+   }
+
+   /*******************************************************************************
+    ** Authenticate the topic and signature; no confirmation before success.
+    *******************************************************************************/
+   public boolean verify(JSONObject payload, String topic, Instant now) throws IOException, InterruptedException
+   {
+      try
+      {
+         String url = payload.getString("SigningCertURL");
+         if(topic == null || topic.isBlank() || !topic.equals(payload.getString("TopicArn")) || !allowedUrl(url, true))
+         {
+            return (false);
+         }
+         String algorithm = switch(payload.getString("SignatureVersion"))
+         {
+            case "1" -> "SHA1withRSA";
+            case "2" -> "SHA256withRSA";
+            default -> null;
+         };
+         if(algorithm == null)
+         {
+            return (false);
+         }
+         Instant stamp = Instant.parse(payload.getString("Timestamp"));
+         if(stamp.isBefore(now.minus(MAX_AGE)) || stamp.isAfter(now.plus(MAX_FUTURE)))
+         {
+            return (false);
+         }
+         String canonical = canonical(payload);
+         X509Certificate certificate = certificates.get(url);
+         if(certificate == null || certificate.getNotAfter().toInstant().isBefore(now))
+         {
+            Instant failed = failures.get(url);
+            if(failed != null && failed.plus(FAILURE_TTL).isAfter(now))
+            {
+               throw (new IOException("SNS verification service unavailable"));
+            }
+            byte[] pem;
+            try
+            {
+               pem = get(url).getBytes(StandardCharsets.UTF_8);
+            }
+            catch(IOException e)
+            {
+               if(failures.size() >= 64)
+               {
+                  failures.clear();
+               }
+               failures.put(url, now);
+               throw (e);
+            }
+            certificate = (X509Certificate) CertificateFactory.getInstance("X.509").generateCertificate(new ByteArrayInputStream(pem));
+            certificate.checkValidity(Date.from(now));
+            if(certificates.size() >= 64)
+            {
+               certificates.clear();
+            }
+            certificates.put(url, certificate);
+         }
+         certificate.checkValidity(Date.from(now));
+         Signature verifier = Signature.getInstance(algorithm);
+         verifier.initVerify(certificate.getPublicKey());
+         verifier.update(canonical.getBytes(StandardCharsets.UTF_8));
+         return (verifier.verify(Base64.getDecoder().decode(payload.getString("Signature"))));
+      }
+      catch(IOException | InterruptedException e)
+      {
+         throw (e);
+      }
+      catch(Exception e)
+      {
+         return (false);
+      }
+   }
+
+   /*******************************************************************************
+    ** Caller invokes this only after verifying the signed confirmation envelope.
+    *******************************************************************************/
+   public void confirm(JSONObject payload) throws IOException, InterruptedException
+   {
+      String url = payload.getString("SubscribeURL");
+      if(!allowedUrl(url, false))
+      {
+         throw (new IllegalArgumentException("Invalid SNS confirmation URL"));
+      }
+      get(url);
+   }
+
+   /*******************************************************************************
+    ** Fixed request timeout, no redirects, and no provider response diagnostics.
+    *******************************************************************************/
+   private String get(String url) throws IOException, InterruptedException
+   {
+      var response = http.send(HttpRequest.newBuilder(URI.create(url)).timeout(Duration.ofSeconds(30)).GET().build());
+      if(response.statusCode() != 200 || response.body() == null || response.body().length() > 65536)
+      {
+         throw (new IOException("SNS verification service unavailable"));
+      }
+      return (response.body());
+   }
+}
@@ -0,0 +1,105 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.time.Instant;
+import java.util.HexFormat;
+import com.kingsrook.qbits.email.EmailQBitConfig;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Small shared verification primitives. Errors never retain request data.
+ *******************************************************************************/
+public final class WebhookVerification
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private WebhookVerification()
+   {
+   }
+
+   /*******************************************************************************
+    ** Resolve only an allowlisted webhook secret; null means unavailable.
+    *******************************************************************************/
+   public static String secret(QRecord provider)
+   {
+      EmailQBitConfig config = EmailRuntime.config();
+      String name = provider.getValueString("webhookSecretEnvName");
+      try
+      {
+         if(name != null && EmailQBitConfig.ENV_NAME.matcher(name).matches() && config.effectiveAllowedEnvNames(provider.getValueString("kind")).contains(name))
+         {
+            String value = config.getSecretResolver().apply(name);
+            return (value == null || value.isBlank() ? null : value);
+         }
+      }
+      catch(RuntimeException e)
+      {
+         return (null);
+      }
+      return (null);
+   }
+
+   /*******************************************************************************
+    ** Compare bytes without data-dependent early exits.
+    *******************************************************************************/
+   public static boolean equal(String expected, String actual)
+   {
+      return (expected != null && actual != null && MessageDigest.isEqual(expected.getBytes(StandardCharsets.UTF_8), actual.getBytes(StandardCharsets.UTF_8)));
+   }
+
+   /*******************************************************************************
+    ** Reject timestamps outside five minutes, including future timestamps.
+    *******************************************************************************/
+   public static boolean fresh(String timestamp, Instant now)
+   {
+      try
+      {
+         Instant time = Instant.ofEpochSecond(Long.parseLong(timestamp));
+         return (!time.isBefore(now.minusSeconds(300)) && !time.isAfter(now.plusSeconds(300)));
+      }
+      catch(RuntimeException e)
+      {
+         return (false);
+      }
+   }
+
+   /*******************************************************************************
+    ** Stable payload and event fingerprints, never payload diagnostics.
+    *******************************************************************************/
+   public static String hash(byte[] bytes)
+   {
+      try
+      {
+         return (HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes)));
+      }
+      catch(Exception e)
+      {
+         throw (new IllegalStateException("SHA-256 unavailable"));
+      }
+   }
+}
@@ -30,10 +30,10 @@ import java.time.Duration;


 /*******************************************************************************
- ** Package-visible HTTP seam; no provider SDK, credential cache or filesystem.
+ ** Shared HTTP seam; no provider SDK, credential cache or filesystem.
 *******************************************************************************/
@FunctionalInterface
-interface EmailHttp
+public interface EmailHttp
 {
   /*******************************************************************************
    ** Exchange one request without retries.
@@ -0,0 +1,292 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.webhook;
+
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Locale;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qbits.email.tracking.EmailTrackingStore;
+import com.kingsrook.qbits.email.tracking.EmailUnsubscribe;
+import com.kingsrook.qbits.email.tracking.MailgunVerifier;
+import com.kingsrook.qbits.email.tracking.PostmarkVerifier;
+import com.kingsrook.qbits.email.tracking.SendGridVerifier;
+import com.kingsrook.qbits.email.tracking.SnsVerifier;
+import com.kingsrook.qbits.email.tracking.WebhookVerification;
+import com.kingsrook.qqq.backend.core.context.CapturedContext;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
+import com.kingsrook.qqq.middleware.javalin.QJavalinMetaData;
+import com.kingsrook.qqq.middleware.javalin.QJavalinRouteProviderInterface;
+import io.javalin.config.JavalinConfig;
+import io.javalin.http.Context;
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+
+/*******************************************************************************
+ ** Authenticated provider ingress and signed public opt-out outside the UI API.
+ ** No bodies, tokens, addresses or exception diagnostics are logged here.
+ *******************************************************************************/
+public class EmailWebhookRouteProvider implements QJavalinRouteProviderInterface
+{
+   private static final int MAX_BODY = 1024 * 1024;
+   private QInstance instance;
+   private final SnsVerifier sns;
+
+   /*******************************************************************************
+    ** Middleware constructs the default route provider.
+    *******************************************************************************/
+   public EmailWebhookRouteProvider()
+   {
+      this(new SnsVerifier());
+   }
+
+   /*******************************************************************************
+    ** Inject the existing HTTP seam through the SNS verifier for local tests.
+    *******************************************************************************/
+   public EmailWebhookRouteProvider(SnsVerifier sns)
+   {
+      this.sns = sns;
+   }
+
+   /*******************************************************************************
+    ** Idempotent registration even when a producer is run repeatedly.
+    *******************************************************************************/
+   public static void register(QInstance instance)
+   {
+      QJavalinMetaData metadata = QJavalinMetaData.ofOrWithNew(instance);
+      if(CollectionUtils.nonNullList(metadata.getAdditionalRouteProviderReferences()).stream()
+         .noneMatch(reference -> reference != null && EmailWebhookRouteProvider.class.getName().equals(reference.getName())))
+      {
+         metadata.withAdditionalRouteProviderReference(new QCodeReference(EmailWebhookRouteProvider.class));
+      }
+   }
+
+   /*******************************************************************************
+    ** Supplied by the host middleware.
+    *******************************************************************************/
+   @Override
+   public void setQInstance(QInstance instance)
+   {
+      this.instance = instance;
+   }
+
+   /*******************************************************************************
+    ** Specific unsubscribe routes precede the provider wildcard route.
+    *******************************************************************************/
+   @Override
+   public void acceptJavalinConfig(JavalinConfig config)
+   {
+      String base = EmailRuntime.config(instance).getWebhookBasePath();
+      config.routes.get(base + "/unsubscribe/{token}", this::unsubscribe);
+      config.routes.post(base + "/unsubscribe/{token}", this::unsubscribe);
+      config.routes.post(base + "/{kind}/{providerName}", this::receive);
+   }
+
+   /*******************************************************************************
+    ** Bound streaming input even when Content-Length is absent or dishonest.
+    *******************************************************************************/
+   private static byte[] body(Context context) throws java.io.IOException
+   {
+      if(context.req().getContentLengthLong() > MAX_BODY)
+      {
+         respond(context, 413, "error", "Payload too large");
+         return (null);
+      }
+      byte[] bytes = context.req().getInputStream().readNBytes(MAX_BODY + 1);
+      if(bytes.length > MAX_BODY)
+      {
+         respond(context, 413, "error", "Payload too large");
+         return (null);
+      }
+      return (bytes);
+   }
+
+   /*******************************************************************************
+    ** Authenticate before storing. Missing configuration and DB outages are retryable.
+    *******************************************************************************/
+   private void receive(Context context)
+   {
+      CapturedContext previous = QContext.capture();
+      try
+      {
+         QContext.init(instance, new com.kingsrook.qqq.backend.core.model.session.QSystemUserSession(instance));
+         byte[] bytes = body(context);
+         if(bytes == null)
+         {
+            return;
+         }
+         String kind = context.pathParam("kind").toUpperCase(Locale.ROOT);
+         List<QRecord> providers = SystemWrites.query(EmailTables.PROVIDER, EmailStore.equal("name", context.pathParam("providerName")));
+         if(providers.size() != 1 || !Boolean.TRUE.equals(providers.getFirst().getValueBoolean("enabled"))
+            || !kind.equals(providers.getFirst().getValueString("kind")) || !List.of("SES", "POSTMARK", "SENDGRID", "MAILGUN").contains(kind))
+         {
+            respond(context, 404, "error", "Provider not found");
+            return;
+         }
+         QRecord provider = providers.getFirst();
+         String secret = "SES".equals(kind) ? provider.getValueString("snsTopicArn") : WebhookVerification.secret(provider);
+         if(secret == null || secret.isBlank())
+         {
+            respond(context, 503, "error", "Webhook verification unavailable");
+            return;
+         }
+         String payload = new String(bytes, StandardCharsets.UTF_8);
+         JSONObject json = "SENDGRID".equals(kind) ? new JSONObject() : new JSONObject(payload);
+         boolean verified = switch(kind)
+         {
+            case "SES" -> sns.verify(json, secret, EmailRuntime.clock().instant());
+            case "POSTMARK" -> new PostmarkVerifier().verify(context.header("Authorization"), secret);
+            case "SENDGRID" -> new SendGridVerifier().verify(bytes, context.header("X-Twilio-Email-Event-Webhook-Signature"),
+               context.header("X-Twilio-Email-Event-Webhook-Timestamp"), secret, EmailRuntime.clock().instant());
+            case "MAILGUN" -> new MailgunVerifier().verify(json, secret, EmailRuntime.clock().instant());
+            default -> false;
+         };
+         if(!verified)
+         {
+            respond(context, 401, "error", "Webhook verification failed");
+            return;
+         }
+         String id = WebhookVerification.hash(bytes);
+         String nonce = null;
+         String note = null;
+         if("SES".equals(kind))
+         {
+            id = json.getString("MessageId");
+            if(!"Notification".equals(json.getString("Type")))
+            {
+               if(!SnsVerifier.allowedUrl(json.getString("SubscribeURL"), false))
+               {
+                  respond(context, 401, "error", "Webhook verification failed");
+                  return;
+               }
+               if("SubscriptionConfirmation".equals(json.getString("Type")))
+               {
+                  sns.confirm(json);
+               }
+               note = "SNS subscription control message verified";
+            }
+         }
+         else if("MAILGUN".equals(kind))
+         {
+            id = json.getJSONObject("event-data").getString("id");
+            nonce = json.getJSONObject("signature").getString("token");
+         }
+         else if("POSTMARK".equals(kind) && json.has("ID"))
+         {
+            id = json.getString("RecordType") + ":" + json.get("ID");
+         }
+         else if("SENDGRID".equals(kind))
+         {
+            new JSONArray(payload);
+         }
+         if(id.isBlank() || id.length() > 500)
+         {
+            respond(context, 401, "error", "Webhook verification failed");
+            return;
+         }
+         EmailTrackingStore.Receipt outcome = EmailTrackingStore.receive(provider, id, nonce, bytes, note);
+         respond(context, outcome == EmailTrackingStore.Receipt.REPLAY ? 401 : 200,
+            outcome == EmailTrackingStore.Receipt.REPLAY ? "error" : outcome == EmailTrackingStore.Receipt.DUPLICATE ? "duplicate" : "received", true);
+      }
+      catch(org.json.JSONException e)
+      {
+         respond(context, 401, "error", "Webhook verification failed");
+      }
+      catch(Exception e)
+      {
+         if(e instanceof InterruptedException)
+         {
+            Thread.currentThread().interrupt();
+         }
+         respond(context, 503, "error", "Email inbox unavailable");
+      }
+      finally
+      {
+         QContext.init(previous);
+      }
+   }
+
+   /*******************************************************************************
+    ** GET is read-only. Only an explicit one-click or page form POST opts out.
+    *******************************************************************************/
+   private void unsubscribe(Context context)
+   {
+      CapturedContext previous = QContext.capture();
+      try
+      {
+         QContext.init(instance, new com.kingsrook.qqq.backend.core.model.session.QSystemUserSession(instance));
+         context.header("Cache-Control", "no-store").header("Referrer-Policy", "no-referrer")
+            .header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; frame-ancestors 'none'");
+         byte[] bytes = body(context);
+         if(bytes == null)
+         {
+            return;
+         }
+         String token = context.pathParam("token");
+         EmailUnsubscribe service = new EmailUnsubscribe();
+         QRecord message = service.resolve(token);
+         if(message == null)
+         {
+            context.status(404).contentType("text/html; charset=utf-8").result("<!doctype html><html lang=\"en\"><meta charset=\"utf-8\"><title>Link not found</title><p>This link is unavailable.</p></html>");
+            return;
+         }
+         boolean post = "POST".equals(context.req().getMethod());
+         boolean oneClick = false;
+         if(post)
+         {
+            String form = new String(bytes, StandardCharsets.UTF_8);
+            oneClick = "List-Unsubscribe=One-Click".equals(form);
+            if(!oneClick && !"confirm=unsubscribe".equals(form))
+            {
+               respond(context, 400, "error", "Unsubscribe confirmation required");
+               return;
+            }
+            service.unsubscribe(token);
+         }
+         context.status(200).contentType(oneClick ? "text/plain" : "text/html; charset=utf-8").result(oneClick ? "" : service.page(message, post));
+      }
+      catch(Exception e)
+      {
+         respond(context, 503, "error", "Unsubscribe unavailable");
+      }
+      finally
+      {
+         QContext.init(previous);
+      }
+   }
+
+   /*******************************************************************************
+    ** Fixed JSON responses contain no request or credential data.
+    *******************************************************************************/
+   private static void respond(Context context, int status, String key, Object value)
+   {
+      context.status(status).contentType("application/json").result(new JSONObject().put(key, value).toString());
+   }
+}
@@ -583,6 +583,48 @@ class EmailDeliveryTest
            .isEqualTo(EmailSendResult.Status.QUEUED);
   }

+   /*******************************************************************************
+    ** Opt-out messages are one mailbox per send, before suppression can trim them.
+    *******************************************************************************/
+   @ParameterizedTest
+   @ValueSource(strings = {"NOTIFICATION", "MARKETING"})
+   void unsubscribeRequiresOneToRecipient(String category) throws Exception
+   {
+      EmailRuntime.config().withPublicBaseUrl("https://app.example.com").withSecretResolver(name -> "test-unsubscribe-key");
+      H2Support.insert(EmailTables.BRAND, new QRecord().withValue("name", "postal").withValue("isDefault", true).withValue("postalAddress", "123 Test Street"));
+      for(EmailRequest invalid : List.of(request().withTo(List.of("person@example.com", "other@example.com")),
+         request().withTo("person@example.com, other@example.com"), request().withCc("other@example.com"),
+         request().withBcc("other@example.com"), request().withTo(List.of()).withCc("other@example.com"),
+         request().withTo(List.of()).withBcc("other@example.com"), request().withTo(List.of()), request().withCc("person@example.com")))
+      {
+         var result = EmailService.send(invalid.withCategory(EmailCategory.valueOf(category)));
+         assertThat(result.status()).isEqualTo(EmailSendResult.Status.REFUSED);
+         assertThat(result.reasons()).containsExactly("Notification and marketing email with an unsubscribe link goes to one recipient per message; send one message per recipient.");
+      }
+      H2Support.insert(EmailTables.SUPPRESSION, new QRecord().withValue("address", "other@example.com").withValue("scope", "ALL").withValue("reason", "HARD_BOUNCE"));
+      assertThat(EmailService.send(request().withCc("other@example.com").withCategory(EmailCategory.valueOf(category))).status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Transactional mail and notification mail without opt-out retain multi-party delivery.
+    *******************************************************************************/
+   @Test
+   void multiRecipientWithoutUnsubscribeStillQueues() throws Exception
+   {
+      EmailRuntime.config().withPublicBaseUrl("https://app.example.com").withSecretResolver(name -> "test-unsubscribe-key");
+      var result = EmailService.send(request().withTo(List.of("person@example.com, other@example.com")).withCc("copy@example.com").withBcc("blind@example.com"));
+      assertThat(result.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+      assertThat(message(result.messageId()).getValueString("unsubscribeUrl")).isNull();
+      assertThat(new EmailDispatcher().runOnce().sent()).isEqualTo(1);
+      assertThat(mail.getReceivedMessages()).hasSize(4);
+      assertThat(mail.getReceivedMessages()[0].getAllRecipients()).hasSize(3);
+      EmailRuntime.config().withSecretResolver(name -> null);
+      result = EmailService.send(request().withCc("copy@example.com").withBcc("blind@example.com").withCategory(EmailCategory.NOTIFICATION));
+      assertThat(result.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+      assertThat(message(result.messageId()).getValueString("unsubscribeUrl")).isNull();
+   }
+
   /*******************************************************************************
    ** Rate Limits Do Not Consume Attempts And Send After Is Honored.
    *******************************************************************************/
@@ -91,6 +91,12 @@ class EmailHomePolishTest
         .withValue("sentDate", NOW.minusSeconds(86400)).withValue("deliveredDate", NOW.plusSeconds(1))
         .withValue("deadLetterDate", NOW).withValue("bouncedDate", NOW.minusSeconds(7 * 86400))
         .withValue("complainedDate", NOW.minusSeconds(7 * 86400 + 1)));
+      for(String type : List.of("DELIVERED", "BOUNCED", "COMPLAINED"))
+      {
+         SystemWrites.insertOne(EmailTables.EVENT, new QRecord().withValue("messageId", 1).withValue("eventType", type).withValue("eventKey", type)
+            .withValue("occurredDate", "DELIVERED".equals(type) ? NOW.plusSeconds(1)
+               : NOW.minusSeconds(7 * 86400 + ("COMPLAINED".equals(type) ? 1 : 0))));
+      }
      H2Support.insert(EmailTables.SUPPRESSION, new QRecord().withValue("address", "a@example.test").withValue("scope", "ALL").withValue("reason", "MANUAL"));
      var document = Jsoup.parse(EmailHome.overview().getHtml());
      var tiles = document.select("body > div > div > div");
@@ -73,7 +73,7 @@ class EmailMetadataTest
   void producerStandardsAndIntegrationFields() throws Exception
   {
      var instance = QContext.getQInstance();
-      assertThat(instance.getTables()).hasSize(7);
+      assertThat(instance.getTables()).hasSize(9);
      for(var table : instance.getTables().values())
      {
         assertThat(table.getName()).matches("email_[a-z_]+");
@@ -495,8 +495,8 @@ class EmailMetadataTest
      config.validate(QContext.getQInstance(), errors);
      assertThat(errors).isEmpty();
      var scheduled = EmailProcesses.processes(config).stream().filter(p -> p.getSchedule() != null).toList();
-      assertThat(scheduled).hasSize(3).allMatch(p -> p.getIsHidden());
-      assertThat(scheduled).extracting(p -> p.getSchedule().getRepeatSeconds()).containsExactly(15, 300, 86400);
+      assertThat(scheduled).hasSize(4).allMatch(p -> p.getIsHidden());
+      assertThat(scheduled).extracting(p -> p.getSchedule().getRepeatSeconds()).containsExactly(30, 15, 300, 86400);
   }

 }
@@ -0,0 +1,199 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email;
+
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.ZoneOffset;
+import java.util.List;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.outbox.EmailCategory;
+import com.kingsrook.qbits.email.outbox.EmailRequest;
+import com.kingsrook.qbits.email.outbox.EmailSendResult;
+import com.kingsrook.qbits.email.outbox.EmailService;
+import com.kingsrook.qbits.email.processes.EmailProcesses;
+import com.kingsrook.qbits.email.render.EmailUnsubscribeLinks;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qbits.email.template.EmailTemplateDefinition;
+import com.kingsrook.qbits.email.template.EmailTemplates;
+import com.kingsrook.qbits.email.template.EmailVersions;
+import com.kingsrook.qbits.email.tracking.EmailUnsubscribe;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.NOW;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.SECRET;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/*******************************************************************************
+ ** Retention of opt-out mail and the consumer-facing template API.
+ *******************************************************************************/
+class EmailPhaseCReviewTest
+{
+   /*******************************************************************************
+    ** Isolated database with opt-out links configured and message retention on.
+    *******************************************************************************/
+   @BeforeEach
+   void setup() throws Exception
+   {
+      H2Support.setup(c -> c.withPublicBaseUrl("https://app.example.com").withSecretResolver(name -> SECRET).withMessageRetentionDays(30));
+      H2Support.provider(25);
+      EmailRuntime.setClock(Clock.fixed(NOW, ZoneOffset.UTC));
+   }
+
+   /*******************************************************************************
+    ** Clear thread and clock state.
+    *******************************************************************************/
+   @AfterEach
+   void cleanup()
+   {
+      QContext.clear();
+      EmailRuntime.setClock(null);
+   }
+
+   /*******************************************************************************
+    ** An old final message with content, in the given category.
+    *******************************************************************************/
+   private static QRecord oldMessage(String category, String address) throws Exception
+   {
+      return (SystemWrites.insertOne(EmailTables.MESSAGE, new QRecord().withValue("providerId", H2Support.all(EmailTables.PROVIDER).getFirst().getValue("id"))
+         .withValue("status", "SENT").withValue("category", category)
+         .withValue("toAddresses", address).withValue("sentDate", NOW.minus(Duration.ofDays(100))).withValue("htmlBody", "<p>hi</p>")
+         .withValue("textBody", "hi").withValue("variablesSnapshot", "{\"name\":\"x\"}")));
+   }
+
+   /*******************************************************************************
+    ** Old marketing and notification mail is kept (content cleared) so its
+    ** unsubscribe link works; transactional mail is deleted.
+    *******************************************************************************/
+   @Test
+   void retentionKeepsOptOutMessagesAndTheirUnsubscribeLinks() throws Exception
+   {
+      QRecord marketing = oldMessage("MARKETING", "pat@example.com");
+      QRecord notification = oldMessage("NOTIFICATION", "sam@example.com");
+      QRecord transactional = oldMessage("TRANSACTIONAL", "lee@example.com");
+      String url = EmailUnsubscribeLinks.create(marketing.getValueString("id"), "pat@example.com", "MARKETING");
+      assertThat(url).isNotNull();
+      EmailProcesses.retain();
+      assertThat(EmailStore.get(EmailTables.MESSAGE, transactional.getValueInteger("id"), null)).isNull();
+      for(QRecord kept : List.of(marketing, notification))
+      {
+         QRecord row = EmailStore.get(EmailTables.MESSAGE, kept.getValueInteger("id"), null);
+         assertThat(row).isNotNull();
+         assertThat(row.getValueString("htmlBody")).isNull();
+         assertThat(row.getValueString("textBody")).isNull();
+         assertThat(row.getValueString("variablesSnapshot")).isNull();
+      }
+      new EmailUnsubscribe().unsubscribe(url.substring(url.lastIndexOf('/') + 1));
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1).allSatisfy(row ->
+      {
+         assertThat(row.getValueString("address")).isEqualTo("pat@example.com");
+         assertThat(row.getValueString("scope")).isEqualTo("MARKETING");
+      });
+      EmailProcesses.retain();
+      assertThat(EmailStore.get(EmailTables.MESSAGE, marketing.getValueInteger("id"), null)).isNotNull();
+   }
+
+   /*******************************************************************************
+    ** A complete definition used by the template tests.
+    *******************************************************************************/
+   private static EmailTemplateDefinition definition()
+   {
+      return (new EmailTemplateDefinition().withName("order_ready").withLabel("Order ready").withCategory(EmailCategory.TRANSACTIONAL)
+         .withSubject("Order {{order_number}} is ready").withBody("Hi {{customer_name}}, order {{order_number}} is ready.")
+         .withVariable("order_number", "Order number", "STRING", true, "PUBLIC")
+         .withVariable("customer_name", "Customer name", "STRING", true, null)
+         .withSampleVariables("{\"order_number\":\"A-1\",\"customer_name\":\"Pat\"}").withOwnerQbit("orders"));
+   }
+
+   /*******************************************************************************
+    ** Created once as an ACTIVE version 1 that sends; the second call changes nothing.
+    *******************************************************************************/
+   @Test
+   void ensureCreatesOnceAndIsIdempotent() throws Exception
+   {
+      assertThat(EmailTemplates.ensure(definition())).isTrue();
+      assertThat(EmailTemplates.ensure(definition())).isFalse();
+      QRecord template = H2Support.all(EmailTables.TEMPLATE).getFirst();
+      assertThat(H2Support.all(EmailTables.TEMPLATE)).hasSize(1);
+      assertThat(template.getValueString("ownerQbit")).isEqualTo("orders");
+      assertThat(template.getValueString("kind")).isEqualTo("MARKDOWN");
+      assertThat(template.getValueBoolean("phiApproved")).isFalse();
+      List<QRecord> versions = H2Support.all(EmailTables.VERSION);
+      assertThat(versions).hasSize(1);
+      assertThat(versions.getFirst().getValueString("status")).isEqualTo("ACTIVE");
+      assertThat(versions.getFirst().getValueInteger("versionNumber")).isEqualTo(1);
+      assertThat(versions.getFirst().getValueString("variablesSchema")).contains("order_number").contains("PUBLIC");
+      assertThat(template.getValue("activeVersionId")).isEqualTo(versions.getFirst().getValue("id"));
+      EmailSendResult result = EmailService.send(new EmailRequest().withTemplateName("order_ready").withTo("pat@example.com")
+         .withVariable("order_number", "A-1").withVariable("customer_name", "Pat"));
+      assertThat(result.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+   }
+
+   /*******************************************************************************
+    ** An edited template keeps its edits when the defining qbit calls ensure again.
+    *******************************************************************************/
+   @Test
+   void ensureDoesNotTouchAnEditedTemplate() throws Exception
+   {
+      assertThat(EmailTemplates.ensure(definition())).isTrue();
+      QRecord template = H2Support.all(EmailTables.TEMPLATE).getFirst();
+      QRecord draft = EmailVersions.newVersion(template.getValueInteger("id"));
+      new UpdateAction().execute(new UpdateInput(EmailTables.VERSION).withRecords(List.of(new QRecord().withValue("id", draft.getValue("id"))
+         .withValue("bodySource", "Edited by a person for {{order_number}}"))));
+      EmailVersions.activate(draft.getValueInteger("id"));
+      assertThat(EmailTemplates.ensure(definition().withBody("Changed again {{order_number}} {{customer_name}}"))).isFalse();
+      assertThat(H2Support.all(EmailTables.VERSION)).hasSize(2);
+      QRecord active = EmailStore.get(EmailTables.VERSION, EmailStore.get(EmailTables.TEMPLATE, template.getValueInteger("id"), null).getValueInteger("activeVersionId"), null);
+      assertThat(active.getValueInteger("versionNumber")).isEqualTo(2);
+      assertThat(active.getValueString("bodySource")).isEqualTo("Edited by a person for {{order_number}}");
+   }
+
+   /*******************************************************************************
+    ** Invalid definitions are refused with plain reasons and write nothing.
+    *******************************************************************************/
+   @Test
+   void ensureRefusesInvalidDefinitions() throws Exception
+   {
+      assertThatThrownBy(() -> EmailTemplates.ensure(new EmailTemplateDefinition().withName("Bad Name").withKind("PDF")))
+         .isInstanceOf(IllegalArgumentException.class)
+         .hasMessageContaining("Name must be snake_case").hasMessageContaining("Label is required").hasMessageContaining("Category is required")
+         .hasMessageContaining("Kind must be").hasMessageContaining("Subject is required").hasMessageContaining("Body is required");
+      assertThatThrownBy(() -> EmailTemplates.ensure(definition().withVariablesSchema("[]")))
+         .hasMessageContaining("not both");
+      assertThatThrownBy(() -> EmailTemplates.ensure(new EmailTemplateDefinition().withName("x").withLabel("X").withCategory(EmailCategory.SECURITY)
+         .withSubject("S").withBody("B").withVariablesSchema("{").withSampleVariables("[")))
+         .hasMessageContaining("Invalid variables schema").hasMessageContaining("Variables must be a JSON object");
+      assertThatThrownBy(() -> EmailTemplates.ensure(definition().withBody("Hi {{undeclared_thing}}")))
+         .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("refused");
+      assertThatThrownBy(() -> EmailTemplates.ensure(definition().withSampleVariables("{}")))
+         .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("refused");
+      assertThat(H2Support.all(EmailTables.TEMPLATE)).isEmpty();
+      assertThat(H2Support.all(EmailTables.VERSION)).isEmpty();
+   }
+}
@@ -157,6 +157,33 @@ class EmailTemplatesTest
      assertThat(row.getValueString("subject")).endsWith(result.messageId().toString());
   }

+   /*******************************************************************************
+    ** Template categories and retained resend links enforce the same mailbox boundary.
+    *******************************************************************************/
+   @Test
+   void unsubscribeTemplateCategoryAndResend() throws Exception
+   {
+      EmailRuntime.config().withSecretResolver(name -> "test-unsubscribe-key");
+      QRecord version = draft("notice", "MARKDOWN", "[]", "{}", "A notification");
+      SystemWrites.updateOne(EmailTables.TEMPLATE, new QRecord().withValue("id", version.getValue("templateId")).withValue("category", "NOTIFICATION"));
+      EmailVersions.activate(version.getValueInteger("id"));
+      var refused = EmailService.send(new EmailRequest().withTemplateName("notice").withTo("person@example.com").withBcc("other@example.com"));
+      assertThat(refused.status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(refused.reasons()).containsExactly("Notification and marketing email with an unsubscribe link goes to one recipient per message; send one message per recipient.");
+      var queued = EmailService.send(new EmailRequest().withTemplateName("notice").withTo("\"Person, One\" <PERSON@example.com>"));
+      assertThat(queued.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+      var resent = EmailService.resend(queued.messageId());
+      assertThat(resent.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+      QRecord message = EmailStore.get(EmailTables.MESSAGE, queued.messageId(), null);
+      String url = message.getValueString("unsubscribeUrl");
+      assertThat(EmailStore.get(EmailTables.MESSAGE, resent.messageId(), null).getValueString("unsubscribeUrl")).isEqualTo(url);
+      new com.kingsrook.qbits.email.tracking.EmailUnsubscribe().unsubscribe(url.substring(url.lastIndexOf('/') + 1));
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1).allSatisfy(row -> assertThat(row.getValueString("address")).isEqualTo("person@example.com"));
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", queued.messageId()).withValue("ccAddresses", "other@example.com"));
+      EmailRuntime.config().withSecretResolver(name -> null);
+      assertThat(EmailService.resend(queued.messageId()).status()).isEqualTo(EmailSendResult.Status.REFUSED);
+   }
+
   /*******************************************************************************
    ** Branded rendered templates reach a real SMTP sink with both alternatives.
    *******************************************************************************/
@@ -220,8 +247,15 @@ class EmailTemplatesTest
      SystemWrites.insertOne(EmailTables.ATTEMPT, new QRecord().withValue("messageId", 1).withValue("providerId", 1).withValue("attemptNumber", 1)
         .withValue("outcome", "SUCCEEDED").withValue("startDate", now.minusSeconds(10)).withValue("endDate", now));
      H2Support.insert(EmailTables.SUPPRESSION, new QRecord().withValue("address", "x@example.com").withValue("scope", "ALL").withValue("reason", "MANUAL"));
+      for(String type : List.of("DELIVERED", "BOUNCED", "COMPLAINED"))
+      {
+         SystemWrites.insertOne(EmailTables.EVENT, new QRecord().withValue("messageId", 1).withValue("eventType", type).withValue("eventKey", type)
+            .withValue("occurredDate", "DELIVERED".equals(type) ? now : now.minusSeconds(("BOUNCED".equals(type) ? 2 : 8) * 86400)));
+      }
+      SystemWrites.insertOne(EmailTables.EVENT, new QRecord().withValue("messageId", 1).withValue("eventType", "DELIVERED").withValue("eventKey", "second-delivery")
+         .withValue("occurredDate", now.minusSeconds(60)));
      var tiles = org.jsoup.Jsoup.parse(EmailHome.overview().getHtml()).select("body > div > div > div");
-      assertThat(tiles).extracting(org.jsoup.nodes.Element::text).containsExactly("1 Sent (24 h)", "1 Delivered (24 h)", "0 Dead letters (24 h)", "1 Bounced (7 d)", "0 Complaints (7 d)", "1 Suppressed addresses");
+      assertThat(tiles).extracting(org.jsoup.nodes.Element::text).containsExactly("1 Sent (24 h)", "2 Delivered (24 h)", "0 Dead letters (24 h)", "1 Bounced (7 d)", "0 Complaints (7 d)", "1 Suppressed addresses");
      assertThat(EmailHome.hourly().getChartData().getDatasets().getFirst().getData()).contains(1);
      assertThat(EmailHome.suppressions().getChartData().getDatasets().get(3).getData()).contains(1);
      QRecord defaultBrand = H2Support.insert(EmailTables.BRAND, new QRecord().withValue("name", "default").withValue("label", "Default").withValue("isDefault", true));
@@ -261,6 +295,9 @@ class EmailTemplatesTest
      EmailRuntime.config().withSecretResolver(name -> null);
      assertThat(com.kingsrook.qbits.email.render.EmailUnsubscribeLinks.create("1", "address", "NOTIFICATION")).isNull();
      assertThatThrownBy(() -> com.kingsrook.qbits.email.render.EmailUnsubscribeLinks.create("1", "address", "MARKETING")).hasMessageContaining("QQQ_EMAIL_UNSUBSCRIBE_KEY");
+      EmailRuntime.config().withSecretResolver(name -> { throw new IllegalArgumentException("resolver detail"); });
+      assertThatThrownBy(() -> com.kingsrook.qbits.email.render.EmailUnsubscribeLinks.create("1", "person@example.com", "NOTIFICATION"))
+         .hasMessage("Opt-out email needs publicBaseUrl and QQQ_EMAIL_UNSUBSCRIBE_KEY");
   }

   /*******************************************************************************
@@ -0,0 +1,167 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+import java.util.List;
+import org.json.JSONArray;
+import org.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.*;
+import static org.assertj.core.api.Assertions.*;
+
+/*******************************************************************************
+ ** Provider-specific normalization contracts use realistic fixture envelopes.
+ *******************************************************************************/
+class EmailEventParserTest
+{
+   /*******************************************************************************
+    ** All SES event types, with notificationType fallback and header correlation.
+    *******************************************************************************/
+   @ParameterizedTest
+   @CsvSource({"Delivery,DELIVERED", "Bounce,BOUNCED", "Complaint,COMPLAINED", "Reject,REJECTED", "DeliveryDelay,DEFERRED", "Open,OPENED", "Click,CLICKED"})
+   void sesTypes(String raw, String expected) throws Exception
+   {
+      JSONObject envelope = new JSONObject(resource("/webhooks/ses-bounce.json"));
+      JSONObject body = new JSONObject(envelope.getString("Message"));
+      body.remove("eventType");
+      body.put("notificationType", raw);
+      String field = "DeliveryDelay".equals(raw) ? "deliveryDelay" : raw.toLowerCase();
+      JSONObject data = new JSONObject().put("timestamp", NOW.toString()).put("link", "https://example.com/page").put("bounceType", "Transient");
+      if("Bounce".equals(raw) || "Complaint".equals(raw))
+      {
+         data.put("Bounce".equals(raw) ? "bouncedRecipients" : "complainedRecipients", new JSONArray()
+            .put(new JSONObject().put("emailAddress", "recipient@example.com")).put(new JSONObject().put("emailAddress", "second@example.com")));
+      }
+      body.put(field, data);
+      envelope.put("Message", body.toString());
+      var events = new EmailEventParser().parse("SES", envelope.toString(), NOW);
+      assertThat(events).hasSize(2).allSatisfy(event ->
+      {
+         assertThat(event.type()).isEqualTo(expected);
+         assertThat(event.occurred()).isEqualTo(NOW);
+         assertThat(event.messageId()).isEqualTo("1");
+         assertThat(event.url()).isEqualTo("CLICKED".equals(expected) ? "https://example.com/page" : null);
+      });
+      if("BOUNCED".equals(expected))
+      {
+         assertThat(events.getFirst().bounceType()).isEqualTo("SOFT");
+      }
+   }
+
+   /*******************************************************************************
+    ** Postmark mappings keep subscriber reactivation out of suppression.
+    *******************************************************************************/
+   @ParameterizedTest
+   @CsvSource({"Delivery,DELIVERED", "Bounce,BOUNCED", "SpamComplaint,COMPLAINED", "Open,OPENED", "Click,CLICKED", "SubscriptionChange,UNSUBSCRIBED"})
+   void postmarkTypes(String raw, String expected) throws Exception
+   {
+      JSONObject body = new JSONObject(resource("/webhooks/postmark-delivery.json")).put("RecordType", raw).put("SuppressSending", true)
+         .put("OriginalLink", "https://example.com/page").put("Type", "Transient").put("ChangedAt", NOW.toString());
+      var event = new EmailEventParser().parse("POSTMARK", body.toString(), NOW).getFirst();
+      assertThat(event.type()).isEqualTo(expected);
+      assertThat(event.occurred()).isEqualTo(NOW);
+      assertThat(event.providerMessageId()).isEqualTo("provider-message-1");
+      assertThat(event.url()).isEqualTo("CLICKED".equals(expected) ? "https://example.com/page" : null);
+      if("BOUNCED".equals(expected))
+      {
+         assertThat(event.bounceType()).isEqualTo("SOFT");
+      }
+   }
+
+   /*******************************************************************************
+    ** SendGrid event mappings retain reason and provider-message prefix correlation.
+    *******************************************************************************/
+   @ParameterizedTest
+   @CsvSource({"delivered,DELIVERED", "bounce,BOUNCED", "dropped,REJECTED", "deferred,DEFERRED", "spamreport,COMPLAINED", "open,OPENED", "click,CLICKED", "unsubscribe,UNSUBSCRIBED"})
+   void sendgridTypes(String raw, String expected) throws Exception
+   {
+      JSONObject body = new JSONArray(resource("/webhooks/sendgrid-batch.json")).getJSONObject(0).put("event", raw)
+         .put("custom_args", new JSONObject().put("qqq_email_id", "25")).put("url", "https://example.com/page");
+      var event = new EmailEventParser().parse("SENDGRID", new JSONArray().put(body).toString(), NOW).getFirst();
+      assertThat(event.type()).isEqualTo(expected);
+      assertThat(event.messageId()).isEqualTo("25");
+      assertThat(event.providerMessageId()).isEqualTo("provider-message-1");
+      assertThat(event.url()).isEqualTo("CLICKED".equals(expected) ? "https://example.com/page" : null);
+   }
+
+   /*******************************************************************************
+    ** Mailgun's nested MIME id and both documented variable spellings correlate.
+    *******************************************************************************/
+   @ParameterizedTest
+   @CsvSource({"delivered,DELIVERED", "failed,BOUNCED", "complained,COMPLAINED", "opened,OPENED", "clicked,CLICKED", "unsubscribed,UNSUBSCRIBED"})
+   void mailgunTypes(String raw, String expected) throws Exception
+   {
+      JSONObject body = new JSONObject(resource("/webhooks/mailgun-failed.json"));
+      body.getJSONObject("event-data").put("event", raw).put("severity", "temporary").put("url", "https://example.com/page")
+         .put("user-variables", new JSONObject().put("v:qqq_email_id", "25"));
+      var event = new EmailEventParser().parse("MAILGUN", body.toString(), NOW).getFirst();
+      assertThat(event.type()).isEqualTo(expected);
+      assertThat(event.messageId()).isEqualTo("25");
+      assertThat(event.providerMessageId()).isEqualTo("provider-message-1");
+      assertThat(event.url()).isEqualTo("CLICKED".equals(expected) ? "https://example.com/page" : null);
+      if("BOUNCED".equals(expected))
+      {
+         assertThat(event.bounceType()).isEqualTo("SOFT");
+      }
+   }
+
+   /*******************************************************************************
+    ** Unknown events are ignored; incomplete known events fail for bounded retry.
+    *******************************************************************************/
+   @Test
+   void unknownEventsAndMissingFields() throws Exception
+   {
+      var parser = new EmailEventParser();
+      assertThat(parser.parse("SES", "{\"Type\":\"SubscriptionConfirmation\"}", NOW)).isEmpty();
+      assertThat(parser.parse("SES", new JSONObject().put("Type", "Notification").put("Message", "{\"eventType\":\"Send\"}").toString(), NOW)).isEmpty();
+      assertThat(parser.parse("POSTMARK", "{\"RecordType\":\"Unknown\"}", NOW)).isEmpty();
+      assertThat(parser.parse("SENDGRID", "[{\"event\":\"processed\"}]", NOW)).isEmpty();
+      assertThat(parser.parse("MAILGUN", "{\"event-data\":{\"event\":\"accepted\"}}", NOW)).isEmpty();
+      assertThatThrownBy(() -> parser.parse("POSTMARK", "invalid", NOW)).isInstanceOf(RuntimeException.class);
+      assertThatThrownBy(() -> parser.parse("OTHER", "{}", NOW)).isInstanceOf(IllegalArgumentException.class);
+      for(String kind : List.of("SES", "POSTMARK", "SENDGRID", "MAILGUN"))
+      {
+         assertThatThrownBy(() -> parser.parse(kind, "{}", NOW)).isInstanceOf(RuntimeException.class);
+      }
+      assertThat(EmailEventParser.detail("550 \"name with spaces\"@example.com unavailable " + "x".repeat(600))).hasSize(500).doesNotContain("example.com", "name with spaces");
+      assertThat(EmailEventParser.detail("bad person+tag@example.com and other@example.org")).isEqualTo("bad [address] and [address]");
+   }
+
+   /*******************************************************************************
+    ** Explicit uncertain bounce types remain uncertain, not hard suppression.
+    *******************************************************************************/
+   @Test
+   void undeterminedBounceTypes() throws Exception
+   {
+      var parser = new EmailEventParser();
+      JSONObject ses = new JSONObject(resource("/webhooks/ses-bounce.json"));
+      JSONObject inner = new JSONObject(ses.getString("Message"));
+      inner.getJSONObject("bounce").put("bounceType", "Undetermined");
+      assertThat(parser.parse("SES", ses.put("Message", inner.toString()).toString(), NOW).getFirst().bounceType()).isEqualTo("UNDETERMINED");
+      JSONObject postmark = new JSONObject(resource("/webhooks/postmark-bounce.json")).put("Type", "Unknown");
+      assertThat(parser.parse("POSTMARK", postmark.toString(), NOW).getFirst().bounceType()).isEqualTo("UNDETERMINED");
+      JSONObject mailgun = new JSONObject(resource("/webhooks/mailgun-failed.json"));
+      mailgun.getJSONObject("event-data").put("severity", "unknown");
+      assertThat(parser.parse("MAILGUN", mailgun.toString(), NOW).getFirst().bounceType()).isEqualTo("UNDETERMINED");
+   }
+}
@@ -0,0 +1,448 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Clock;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qbits.email.H2Support;
+import com.kingsrook.qbits.email.ProcessRunner;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.processes.EmailProcesses;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import org.json.JSONArray;
+import org.json.JSONObject;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.*;
+import static org.assertj.core.api.Assertions.*;
+
+/*******************************************************************************
+ ** Durable processing, rollback, suppression and race acceptance on real H2.
+ *******************************************************************************/
+class EmailInboxProcessorTest
+{
+   private QInstance instance;
+
+   /*******************************************************************************
+    ** Use a fixed clock for retry and provider event times.
+    *******************************************************************************/
+   @BeforeEach
+   void setup() throws Exception
+   {
+      instance = H2Support.setup(c -> c.withSecretResolver(name -> SECRET));
+      EmailRuntime.setClock(Clock.fixed(NOW, ZoneOffset.UTC));
+   }
+
+   /*******************************************************************************
+    ** Restore host-independent runtime state.
+    *******************************************************************************/
+   @AfterEach
+   void teardown()
+   {
+      QContext.clear();
+      EmailRuntime.setClock(null);
+   }
+
+   /*******************************************************************************
+    ** Ingest the same authenticated payload boundary used by HTTP.
+    *******************************************************************************/
+   private static void receive(QRecord provider, String id, String body) throws Exception
+   {
+      assertThat(EmailTrackingStore.receive(provider, id, null, body.getBytes(StandardCharsets.UTF_8), null)).isEqualTo(EmailTrackingStore.Receipt.RECEIVED);
+   }
+
+   /*******************************************************************************
+    ** All provider fixtures produce hard-bounce suppression and masked diagnostics.
+    *******************************************************************************/
+   @ParameterizedTest
+   @CsvSource({"SES,ses-bounce.json,2", "POSTMARK,postmark-bounce.json,1", "SENDGRID,sendgrid-batch.json,2", "MAILGUN,mailgun-failed.json,1"})
+   void realisticFixtures(String kind, String fixture, int count) throws Exception
+   {
+      QRecord provider = provider(kind);
+      QRecord message = message(provider);
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", message.getValue("id")).withValue("toAddresses", "recipient@example.com, second@example.com"));
+      receive(provider, "fixture", resource("/webhooks/" + fixture));
+      assertThat(new EmailInboxProcessor().runOnce()).isEqualTo(1);
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(count);
+      assertThat(H2Support.all(EmailTables.INBOX).getFirst().getValueString("status")).isEqualTo("PROCESSED");
+      QRecord updated = EmailStore.get(EmailTables.MESSAGE, message.getValueInteger("id"), null);
+      assertThat(updated.getValueString("status")).isEqualTo("BOUNCED");
+      assertThat(updated.getValueInstant("bouncedDate")).isNotNull();
+      for(QRecord suppression : H2Support.all(EmailTables.SUPPRESSION))
+      {
+         assertThat(suppression.getValueString("scope")).isEqualTo("ALL");
+         assertThat(suppression.getValueString("reason")).isEqualTo("HARD_BOUNCE");
+         assertThat(suppression.getValueInteger("sourceMessageId")).isEqualTo(message.getValueInteger("id"));
+         assertThat(suppression.getValueInteger("sourceEventId")).isPositive();
+      }
+      assertThat(H2Support.all(EmailTables.EVENT)).allSatisfy(event -> assertThat(event.getValueString("detail")).doesNotContain("recipient@example.com", "second@example.com"));
+      assertThat(new EmailInboxProcessor().runOnce()).isZero();
+   }
+
+   /*******************************************************************************
+    ** Soft bounces do not suppress; complaint dominates all later delivery events.
+    *******************************************************************************/
+   @Test
+   void statusOrderSuppressionUpsertsAndOverlappingBatches() throws Exception
+   {
+      QRecord provider = provider("SENDGRID");
+      QRecord message = message(provider);
+      JSONObject event = new JSONObject().put("email", "Recipient@EXAMPLE.COM").put("qqq_email_id", message.getValueString("id"))
+         .put("timestamp", NOW.getEpochSecond()).put("event", "bounce").put("type", "blocked").put("sg_event_id", "soft");
+      receive(provider, "one", new JSONArray().put(event).toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).isEmpty();
+      assertThat(H2Support.all(EmailTables.MESSAGE).getFirst().getValueString("status")).isEqualTo("SENT");
+      event.put("type", "bounce").put("sg_event_id", "hard");
+      receive(provider, "two", new JSONArray().put(event).toString());
+      new EmailInboxProcessor().runOnce();
+      event.put("event", "spamreport").put("sg_event_id", "complaint");
+      receive(provider, "three", new JSONArray().put(event).toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.MESSAGE).getFirst().getValueString("status")).isEqualTo("COMPLAINED");
+      assertThat(H2Support.all(EmailTables.MESSAGE).getFirst().getValueInstant("complainedDate")).isEqualTo(NOW);
+      event.put("event", "delivered").put("sg_event_id", "late-delivery");
+      JSONArray overlap = new JSONArray(resource("/webhooks/sendgrid-batch.json")).put(event);
+      receive(provider, "four", overlap.toString());
+      new EmailInboxProcessor().runOnce();
+      receive(provider, "five", overlap.toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.INBOX).getLast().getValueString("status")).isEqualTo("IGNORED");
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(6);
+      assertThat(H2Support.all(EmailTables.MESSAGE).getFirst().getValueString("status")).isEqualTo("COMPLAINED");
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION).getFirst().getValueString("reason")).isEqualTo("COMPLAINT");
+   }
+
+   /*******************************************************************************
+    ** Provider unsubscribes cannot suppress account/security mail.
+    *******************************************************************************/
+   @ParameterizedTest
+   @ValueSource(strings = {"MARKETING", "NOTIFICATION", "TRANSACTIONAL", "SECURITY"})
+   void categoryRule(String category) throws Exception
+   {
+      QRecord provider = provider("POSTMARK");
+      QRecord message = message(provider);
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", message.getValue("id")).withValue("category", category));
+      JSONObject event = new JSONObject(resource("/webhooks/postmark-delivery.json")).put("RecordType", "SubscriptionChange").put("SuppressSending", true);
+      receive(provider, "unsubscribe", event.toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(1);
+      if(List.of("MARKETING", "NOTIFICATION").contains(category))
+      {
+         assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1);
+         assertThat(H2Support.all(EmailTables.SUPPRESSION).getFirst().getValueString("scope")).isEqualTo(category);
+      }
+      else
+      {
+         assertThat(H2Support.all(EmailTables.SUPPRESSION)).isEmpty();
+      }
+      receive(provider, "reactivation", event.put("SuppressSending", false).toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(1);
+   }
+
+   /*******************************************************************************
+    ** Tracking flags apply before matching, and protected mail never stores tracking.
+    *******************************************************************************/
+   @Test
+   void opensAndClicksOnlyWhenEnabled() throws Exception
+   {
+      QRecord provider = provider("SENDGRID");
+      QRecord message = message(provider);
+      JSONArray batch = new JSONArray();
+      for(String type : List.of("open", "click"))
+      {
+         batch.put(new JSONObject().put("event", type).put("email", "recipient@example.com").put("qqq_email_id", message.getValue("id"))
+            .put("sg_event_id", type).put("url", "https://example.com/link").put("timestamp", NOW.getEpochSecond()));
+      }
+      receive(provider, "off", batch.toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+      SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", provider.getValue("id")).withValue("trackOpens", true).withValue("trackClicks", true));
+      receive(provider, "on", batch.toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(2);
+      assertThat(H2Support.all(EmailTables.EVENT).getFirst().getValueString("url")).isNull();
+      assertThat(H2Support.all(EmailTables.EVENT).getLast().getValueString("url")).isEqualTo("https://example.com/link");
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", message.getValue("id")).withValue("phiApproved", true));
+      batch.getJSONObject(0).put("sg_event_id", "protected-open");
+      batch.getJSONObject(1).put("sg_event_id", "protected-click");
+      receive(provider, "protected", batch.toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(2);
+   }
+
+   /*******************************************************************************
+    ** Partial batch failure rolls back events, suppression and message status.
+    *******************************************************************************/
+   @Test
+   void retriesSixTimesThenReviewReprocess() throws Exception
+   {
+      QRecord provider = provider("SENDGRID");
+      QRecord message = message(provider);
+      receive(provider, "retry", "[{not json");
+      for(int i = 1; i <= 6; i++)
+      {
+         assertThat(new EmailInboxProcessor().runOnce()).isZero();
+         QRecord row = H2Support.all(EmailTables.INBOX).getFirst();
+         assertThat(row.getValueInteger("attempts")).isEqualTo(i);
+         assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+         assertThat(H2Support.all(EmailTables.SUPPRESSION)).isEmpty();
+         assertThat(H2Support.all(EmailTables.MESSAGE).getFirst().getValueString("status")).isEqualTo("SENT");
+         if(i < 6)
+         {
+            assertThat(row.getValueString("status")).isEqualTo("PENDING");
+            assertThat(new EmailInboxProcessor().runOnce()).isZero();
+            EmailRuntime.setClock(Clock.fixed(row.getValueInstant("nextAttemptDate"), ZoneOffset.UTC));
+         }
+      }
+      QRecord failed = H2Support.all(EmailTables.INBOX).getFirst();
+      assertThat(failed.getValueString("status")).isEqualTo("FAILED");
+      var lines = ProcessRunner.run(EmailProcesses.REPROCESS, failed.getValue("id"));
+      assertThat(lines.review().toString()).contains("will be processed");
+      assertThat(lines.result().toString()).contains("PENDING");
+      assertThat(EmailInboxProcessor.reprocess(failed.getValueInteger("id"))).isFalse();
+      assertThat(EmailInboxProcessor.reprocess(9999)).isFalse();
+      SystemWrites.updateOne(EmailTables.INBOX, new QRecord().withValue("id", failed.getValue("id")).withValue("payloadJson", resource("/webhooks/sendgrid-batch.json")));
+      new EmailProcesses.ScheduledStep().run(new RunBackendStepInput().withProcessName(EmailProcesses.PROCESS_WEBHOOKS), new RunBackendStepOutput());
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(2);
+   }
+
+   /*******************************************************************************
+    ** A foreign event never blocks a real hard bounce in the same SendGrid batch; the
+    ** foreign one retries, then the row finishes PROCESSED with a note, not FAILED.
+    *******************************************************************************/
+   @Test
+   void unmatchedEventDoesNotBlockMatchedBounce() throws Exception
+   {
+      QRecord provider = provider("SENDGRID");
+      QRecord message = message(provider);
+      JSONArray batch = new JSONArray(resource("/webhooks/sendgrid-batch.json"));
+      batch.put(new JSONObject().put("email", "stranger@example.com").put("timestamp", NOW.getEpochSecond()).put("event", "bounce")
+         .put("type", "bounce").put("sg_event_id", "foreign").put("sg_message_id", "someone-elses-message.filter009"));
+      receive(provider, "mixed", batch.toString());
+      for(int i = 1; i <= 6; i++)
+      {
+         assertThat(new EmailInboxProcessor().runOnce()).isEqualTo(1);
+         QRecord row = H2Support.all(EmailTables.INBOX).getFirst();
+         assertThat(row.getValueInteger("attempts")).isEqualTo(i);
+         assertThat(H2Support.all(EmailTables.EVENT)).hasSize(2);
+         assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1);
+         assertThat(EmailStore.get(EmailTables.MESSAGE, message.getValueInteger("id"), null).getValueString("status")).isEqualTo("BOUNCED");
+         if(i < 6)
+         {
+            assertThat(row.getValueString("status")).isEqualTo("PENDING");
+            assertThat(row.getValueString("lastError")).isEqualTo("1 event did not match a message");
+            EmailRuntime.setClock(Clock.fixed(row.getValueInstant("nextAttemptDate"), ZoneOffset.UTC));
+         }
+      }
+      QRecord done = H2Support.all(EmailTables.INBOX).getFirst();
+      assertThat(done.getValueString("status")).isEqualTo("PROCESSED");
+      assertThat(done.getValueString("lastError")).isEqualTo("1 event did not match a message");
+      assertThat(done.getValueInstant("processedDate")).isNotNull();
+      assertThat(done.getValue("nextAttemptDate")).isNull();
+   }
+
+   /*******************************************************************************
+    ** Only a recipient of the matched message can be suppressed, and an SES bounce
+    ** without bouncedRecipients is skipped rather than widened to all destinations.
+    *******************************************************************************/
+   @Test
+   void suppressionOnlyForRecipientsOfTheMessage() throws Exception
+   {
+      QRecord provider = provider("SENDGRID");
+      QRecord message = message(provider);
+      JSONObject event = new JSONObject().put("event", "bounce").put("type", "bounce").put("email", "intruder@example.com")
+         .put("qqq_email_id", message.getValueString("id")).put("timestamp", NOW.getEpochSecond()).put("sg_event_id", "intruder");
+      receive(provider, "intruder", new JSONArray().put(event).toString());
+      assertThat(new EmailInboxProcessor().runOnce()).isEqualTo(1);
+      assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).isEmpty();
+      QRecord row = H2Support.all(EmailTables.INBOX).getFirst();
+      assertThat(row.getValueString("status")).isEqualTo("IGNORED");
+      assertThat(row.getValueString("lastError")).contains("1 event was skipped");
+      assertThat(row.getValueString("lastError")).doesNotContain("intruder");
+
+      QRecord ses = provider("SES");
+      QRecord sesMessage = message(ses);
+      JSONObject envelope = new JSONObject(resource("/webhooks/ses-bounce.json"));
+      JSONObject body = new JSONObject(envelope.getString("Message"));
+      body.getJSONObject("mail").getJSONArray("headers").getJSONObject(0).put("value", sesMessage.getValueString("id"));
+      body.getJSONObject("bounce").remove("bouncedRecipients");
+      receive(ses, "no-recipients", envelope.put("Message", body.toString()).toString());
+      assertThat(new EmailInboxProcessor().runOnce()).isEqualTo(1);
+      assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).isEmpty();
+      assertThat(EmailStore.get(EmailTables.MESSAGE, sesMessage.getValueInteger("id"), null).getValueString("status")).isEqualTo("SENT");
+   }
+
+   /*******************************************************************************
+    ** Our custom id cannot match another provider's message; callbacks may race SENT.
+    *******************************************************************************/
+   @Test
+   void providerBoundaryAndSendingRace() throws Exception
+   {
+      QRecord provider = provider("SENDGRID");
+      QRecord message = message(provider("POSTMARK"));
+      JSONObject event = new JSONObject().put("event", "delivered").put("qqq_email_id", message.getValue("id"))
+         .put("email", "recipient@example.com").put("sg_event_id", "cross-provider");
+      receive(provider, "cross", new JSONArray().put(event).toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+      QRecord actual = message(provider);
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", actual.getValue("id")).withValue("status", "SENDING"));
+      receive(provider, "racing", new JSONArray().put(event.put("qqq_email_id", actual.getValue("id")).put("sg_event_id", "racing")).toString());
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", actual.getValue("id")).withValue("status", "SENT"));
+      EmailRuntime.setClock(Clock.fixed(NOW.plusSeconds(61), ZoneOffset.UTC));
+      new EmailInboxProcessor().runOnce();
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(1);
+      assertThat(EmailStore.get(EmailTables.MESSAGE, actual.getValueInteger("id"), null).getValueString("status")).isEqualTo("DELIVERED");
+   }
+
+   /*******************************************************************************
+    ** Custom arguments support the entire integer primary-key range.
+    *******************************************************************************/
+   @Test
+   void fullIntegerMessageId() throws Exception
+   {
+      QRecord provider = provider("SENDGRID");
+      QRecord message = SystemWrites.insertOne(EmailTables.MESSAGE, new QRecord().withValue("id", 2000000000)
+         .withValue("providerId", provider.getValue("id")).withValue("status", "SENT").withValue("category", "NOTIFICATION")
+         .withValue("toAddresses", "recipient@example.com"));
+      assertThat(message.getValueInteger("id")).isEqualTo(2000000000);
+      JSONObject event = new JSONObject().put("event", "delivered").put("qqq_email_id", "2000000000")
+         .put("email", "recipient@example.com").put("sg_event_id", "large-message-id");
+      receive(provider, "large", new JSONArray().put(event).toString());
+      assertThat(new EmailInboxProcessor().runOnce()).isEqualTo(1);
+      assertThat(H2Support.all(EmailTables.EVENT).getFirst().getValueInteger("messageId")).isEqualTo(2000000000);
+   }
+
+   /*******************************************************************************
+    ** Two processor replicas observe one receipt but only one applies it.
+    *******************************************************************************/
+   @Test
+   void concurrentProcessors() throws Exception
+   {
+      QRecord provider = provider("POSTMARK");
+      message(provider);
+      receive(provider, "race", resource("/webhooks/postmark-bounce.json"));
+      try(var workers = Executors.newFixedThreadPool(2))
+      {
+         var tasks = new ArrayList<java.util.concurrent.Future<Integer>>();
+         for(int i = 0; i < 2; i++)
+         {
+            tasks.add(workers.submit(() ->
+            {
+               QContext.init(instance, H2Support.admin());
+               try
+               {
+                  return (new EmailInboxProcessor().runOnce());
+               }
+               finally
+               {
+                  QContext.clear();
+               }
+            }));
+         }
+         assertThat(tasks.get(0).get(20, TimeUnit.SECONDS) + tasks.get(1).get(20, TimeUnit.SECONDS)).isEqualTo(1);
+      }
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(1);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1);
+   }
+
+   /*******************************************************************************
+    ** Core's optional alert sink receives a day-scoped warning after threshold.
+    *******************************************************************************/
+   @Test
+   void complaintAlertThreshold() throws Exception
+   {
+      QRecord provider = provider("POSTMARK");
+      message(provider);
+      Alerts.signals.clear();
+      com.kingsrook.qqq.backend.core.alerts.AlertsInstanceMetaData.ofOrWithNew(instance)
+         .withSink(new com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference(Alerts.class));
+      for(int i = 0; i < 5; i++)
+      {
+         JSONObject event = new JSONObject(resource("/webhooks/postmark-bounce.json")).put("RecordType", "SpamComplaint").put("ID", i);
+         receive(provider, "complaint-" + i, event.toString());
+         new EmailInboxProcessor().runOnce();
+         assertThat(Alerts.signals).hasSize(Math.max(0, i - 2));
+      }
+      assertThat(Alerts.signals).allSatisfy(signal ->
+      {
+         assertThat(signal.getAlertTypeName()).isEqualTo("email.complaint");
+         assertThat(signal.getDedupKey()).startsWith("complaints:");
+      });
+      assertThat(Alerts.signals.get(0).getDedupKey()).isEqualTo(Alerts.signals.get(1).getDedupKey());
+   }
+
+   /*******************************************************************************
+    ** Captures content-free core alert signals.
+    *******************************************************************************/
+   public static class Alerts implements com.kingsrook.qqq.backend.core.alerts.AlertSinkInterface
+   {
+      static final List<com.kingsrook.qqq.backend.core.alerts.AlertSignal> signals = new ArrayList<>();
+      /*******************************************************************************
+       ** Test adapter implementation.
+       *******************************************************************************/
+      @Override
+      public void raise(com.kingsrook.qqq.backend.core.alerts.AlertSignal signal)
+      {
+         signals.add(signal);
+      }
+
+      /*******************************************************************************
+       ** Test adapter implementation.
+       *******************************************************************************/
+      @Override
+      public void resolve(com.kingsrook.qqq.backend.core.alerts.AlertResolution resolution)
+      {
+      }
+
+      /*******************************************************************************
+       ** Test adapter implementation.
+       *******************************************************************************/
+      @Override
+      public void heartbeat(String source, java.time.Instant timestamp)
+      {
+      }
+
+   }
+}
@@ -0,0 +1,512 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyPairGenerator;
+import java.security.Signature;
+import java.time.Clock;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qbits.email.H2Support;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.render.EmailUnsubscribeLinks;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qbits.email.webhook.EmailWebhookRouteProvider;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.middleware.javalin.QJavalinMetaData;
+import io.javalin.Javalin;
+import org.json.JSONObject;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.*;
+import static org.assertj.core.api.Assertions.*;
+
+/*******************************************************************************
+ ** Real random-port HTTP exercises the exact route provider and database boundary.
+ *******************************************************************************/
+class EmailWebhookRoutesTest
+{
+   private Javalin server;
+   private HttpClient client;
+   private QInstance instance;
+   private final List<String> fetched = new java.util.concurrent.CopyOnWriteArrayList<>();
+   private static final String BASE = "/custom/email";
+
+   /*******************************************************************************
+    ** Test the configurable path and injected SNS HTTP seam.
+    *******************************************************************************/
+   @BeforeEach
+   void setup() throws Exception
+   {
+      instance = H2Support.setup(c -> c.withWebhookBasePath(BASE).withPublicBaseUrl("https://app.example.com").withSecretResolver(name -> SECRET));
+      EmailRuntime.setClock(Clock.fixed(NOW, ZoneOffset.UTC));
+      String pem = resource("/sns-test/test-only-certificate.pem");
+      var routes = new EmailWebhookRouteProvider(new SnsVerifier(request ->
+      {
+         fetched.add(request.uri().toString());
+         return (response(request, 200, request.uri().getPath().endsWith(".pem") ? pem : "confirmed"));
+      }));
+      routes.setQInstance(instance);
+      server = Javalin.create(routes::acceptJavalinConfig).start(0);
+      client = HttpClient.newHttpClient();
+   }
+
+   /*******************************************************************************
+    ** Release every server, connection and thread-local test context.
+    *******************************************************************************/
+   @AfterEach
+   void teardown()
+   {
+      server.stop();
+      client.close();
+      EmailRuntime.setClock(null);
+      QContext.clear();
+   }
+
+   /*******************************************************************************
+    ** HTTP request helper, with no application code bypass.
+    *******************************************************************************/
+   private HttpResponse<String> request(String path, String method, String body, String... headers) throws Exception
+   {
+      var builder = HttpRequest.newBuilder(URI.create("http://localhost:" + server.port() + BASE + path));
+      if(headers.length > 0)
+      {
+         builder.headers(headers);
+      }
+      builder.method(method, body == null ? HttpRequest.BodyPublishers.noBody() : HttpRequest.BodyPublishers.ofString(body));
+      return (client.send(builder.build(), HttpResponse.BodyHandlers.ofString()));
+   }
+
+   /*******************************************************************************
+    ** Fixed Basic auth used only by synthetic Postmark requests.
+    *******************************************************************************/
+   private static String auth()
+   {
+      return ("Basic " + Base64.getEncoder().encodeToString(("postmark:" + SECRET).getBytes(StandardCharsets.UTF_8)));
+   }
+
+   /*******************************************************************************
+    ** All status codes and streaming size caps before any authentication lookup.
+    *******************************************************************************/
+   @Test
+   void postmarkStatusesAndDedup() throws Exception
+   {
+      QRecord provider = provider("POSTMARK");
+      String body = resource("/webhooks/postmark-delivery.json");
+      assertThat(request("/postmark/postmark", "POST", body, "Authorization", auth()).body()).contains("received");
+      assertThat(request("/postmark/postmark", "POST", body, "Authorization", auth()).body()).contains("duplicate");
+      assertThat(H2Support.all(EmailTables.INBOX)).hasSize(1);
+      assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+      assertThat(request("/postmark/postmark", "POST", body).statusCode()).isEqualTo(401);
+      assertThat(request("/postmark/postmark", "POST", "invalid", "Authorization", auth()).statusCode()).isEqualTo(401);
+      assertThat(request("/mailgun/postmark", "POST", body).statusCode()).isEqualTo(404);
+      assertThat(request("/postmark/missing", "POST", body).statusCode()).isEqualTo(404);
+      assertThat(request("/postmark/postmark", "POST", "x".repeat(1024 * 1024 + 1)).statusCode()).isEqualTo(413);
+      var chunked = HttpRequest.newBuilder(URI.create("http://localhost:" + server.port() + BASE + "/postmark/postmark"))
+         .POST(HttpRequest.BodyPublishers.ofInputStream(() -> new java.io.ByteArrayInputStream(new byte[1024 * 1024 + 1]))).build();
+      assertThat(client.send(chunked, HttpResponse.BodyHandlers.ofString()).statusCode()).isEqualTo(413);
+      EmailRuntime.config().withSecretResolver(name -> null);
+      assertThat(request("/postmark/postmark", "POST", body).statusCode()).isEqualTo(503);
+      SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", provider.getValue("id")).withValue("enabled", false).withValue("isDefault", false));
+      assertThat(request("/postmark/postmark", "POST", body).statusCode()).isEqualTo(404);
+   }
+
+   /*******************************************************************************
+    ** Missing per-provider verification configuration is explicitly retryable.
+    *******************************************************************************/
+   @ParameterizedTest
+   @ValueSource(strings = {"SES", "SENDGRID", "MAILGUN"})
+   void missingSecretOrTopic(String kind) throws Exception
+   {
+      QRecord provider = provider(kind);
+      if("SES".equals(kind))
+      {
+         SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", provider.getValue("id")).withValue("snsTopicArn", null));
+      }
+      EmailRuntime.config().withSecretResolver(name -> null);
+      assertThat(request("/" + kind.toLowerCase() + "/" + kind.toLowerCase(), "POST", "{}").statusCode()).isEqualTo(503);
+   }
+
+   /*******************************************************************************
+    ** DB failure is 503, and idempotent registration never adds duplicate routes.
+    *******************************************************************************/
+   @Test
+   void unavailableStoreAndRegistration() throws Exception
+   {
+      provider("POSTMARK");
+      EmailWebhookRouteProvider.register(instance);
+      EmailWebhookRouteProvider.register(instance);
+      assertThat(QJavalinMetaData.ofOrWithNew(instance).getAdditionalRouteProviderReferences()).hasSize(1);
+      var removed = instance.getTables().remove(EmailTables.INBOX);
+      try
+      {
+         assertThat(request("/postmark/postmark", "POST", resource("/webhooks/postmark-delivery.json"), "Authorization", auth()).statusCode()).isEqualTo(503);
+      }
+      finally
+      {
+         instance.addTable(removed);
+      }
+   }
+
+   /*******************************************************************************
+    ** SNS verifies before confirming, with a durable ignored control message.
+    *******************************************************************************/
+   @Test
+   void snsNotificationConfirmationAndTopicMismatch() throws Exception
+   {
+      provider("SES");
+      EmailRuntime.setClock(Clock.fixed(certificateTime(), ZoneOffset.UTC));
+      JSONObject payload = signSns(new JSONObject(resource("/webhooks/ses-bounce.json")));
+      assertThat(request("/ses/ses", "POST", payload.toString()).statusCode()).isEqualTo(200);
+      assertThat(request("/ses/ses", "POST", payload.toString()).body()).contains("duplicate");
+      payload.put("MessageId", "subscription").put("Type", "SubscriptionConfirmation").put("Token", "test-token")
+         .put("SubscribeURL", "https://sns.us-east-1.amazonaws.com/?Action=ConfirmSubscription&Token=test");
+      signSns(payload);
+      assertThat(request("/ses/ses", "POST", payload.toString()).statusCode()).isEqualTo(200);
+      assertThat(H2Support.all(EmailTables.INBOX).get(1).getValueString("status")).isEqualTo("IGNORED");
+      assertThat(fetched).anyMatch(url -> url.contains("ConfirmSubscription"));
+      assertThat(request("/ses/ses", "POST", signSns(payload.put("SubscribeURL", "http://evil.example/")).toString()).statusCode()).isEqualTo(401);
+      assertThat(request("/ses/ses", "POST", signSns(payload.put("TopicArn", TOPIC + "other")).toString()).statusCode()).isEqualTo(401);
+   }
+
+   /*******************************************************************************
+    ** Signed raw SendGrid body is accepted and tampering leaves no extra receipt.
+    *******************************************************************************/
+   @Test
+   void sendgridSignedBatch() throws Exception
+   {
+      provider("SENDGRID");
+      var keys = KeyPairGenerator.getInstance("EC").generateKeyPair();
+      EmailRuntime.config().withSecretResolver(name -> Base64.getEncoder().encodeToString(keys.getPublic().getEncoded()));
+      String body = resource("/webhooks/sendgrid-batch.json");
+      String timestamp = Long.toString(NOW.getEpochSecond());
+      Signature signer = Signature.getInstance("SHA256withECDSA");
+      signer.initSign(keys.getPrivate());
+      signer.update((timestamp + body).getBytes(StandardCharsets.UTF_8));
+      String signature = Base64.getEncoder().encodeToString(signer.sign());
+      String[] headers = {"X-Twilio-Email-Event-Webhook-Signature", signature, "X-Twilio-Email-Event-Webhook-Timestamp", timestamp};
+      assertThat(request("/sendgrid/sendgrid", "POST", body, headers).statusCode()).isEqualTo(200);
+      assertThat(request("/sendgrid/sendgrid", "POST", body + " ", headers).statusCode()).isEqualTo(401);
+      assertThat(request("/sendgrid/sendgrid", "POST", body, headers).body()).contains("duplicate");
+   }
+
+   /*******************************************************************************
+    ** Mailgun replay is 401, but a new signed nonce with an old event id is 200.
+    *******************************************************************************/
+   @Test
+   void mailgunReplayAndDuplicateEvent() throws Exception
+   {
+      provider("MAILGUN");
+      JSONObject payload = signMailgun(new JSONObject(resource("/webhooks/mailgun-failed.json")), "nonce-one", NOW);
+      assertThat(request("/mailgun/mailgun", "POST", payload.toString()).body()).contains("received");
+      assertThat(request("/mailgun/mailgun", "POST", payload.toString()).statusCode()).isEqualTo(401);
+      signMailgun(payload, "nonce-two", NOW);
+      assertThat(request("/mailgun/mailgun", "POST", payload.toString()).body()).contains("duplicate");
+      signMailgun(payload, "stale", NOW.minusSeconds(301));
+      assertThat(request("/mailgun/mailgun", "POST", payload.toString()).statusCode()).isEqualTo(401);
+   }
+
+   /*******************************************************************************
+    ** Concurrent receipt dedup is durable across separate HTTP worker threads.
+    *******************************************************************************/
+   @Test
+   void concurrentDuplicates() throws Exception
+   {
+      provider("POSTMARK");
+      String body = resource("/webhooks/postmark-bounce.json");
+      try(var workers = Executors.newFixedThreadPool(2))
+      {
+         var first = workers.submit(() -> request("/postmark/postmark", "POST", body, "Authorization", auth()));
+         var second = workers.submit(() -> request("/postmark/postmark", "POST", body, "Authorization", auth()));
+         assertThat(first.get(20, TimeUnit.SECONDS).statusCode()).isEqualTo(200);
+         assertThat(second.get(20, TimeUnit.SECONDS).statusCode()).isEqualTo(200);
+         assertThat(H2Support.all(EmailTables.INBOX)).hasSize(1);
+      }
+   }
+
+   /*******************************************************************************
+    ** GET never opts out, page content is escaped and one-click returns no body.
+    *******************************************************************************/
+   @Test
+   void unsubscribeGetAndBothPosts() throws Exception
+   {
+      QRecord message = message(provider("POSTMARK"));
+      QRecord brand = H2Support.insert(EmailTables.BRAND, new QRecord().withValue("name", "brand").withValue("label", "<script>Brand & Co</script>"));
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", message.getValue("id")).withValue("brandId", brand.getValue("id")));
+      String token = token(message);
+      var get = request("/unsubscribe/" + token, "GET", null);
+      assertThat(get.statusCode()).isEqualTo(200);
+      assertThat(get.body()).contains("&lt;script&gt;Brand &amp; Co&lt;/script&gt;", "Unsubscribe from Marketing", "color-scheme:light dark", "prefers-color-scheme:dark", "<form method=\"post\">")
+         .doesNotContain("<script>", "recipient@example.com", "http://", "https://");
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).isEmpty();
+      assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+      assertThat(request("/unsubscribe/" + token, "POST", "anything").statusCode()).isEqualTo(400);
+      var click = request("/unsubscribe/" + token, "POST", "List-Unsubscribe=One-Click", "Content-Type", "application/x-www-form-urlencoded");
+      assertThat(click.statusCode()).isEqualTo(200);
+      assertThat(click.body()).isEmpty();
+      assertThat(request("/unsubscribe/" + token, "POST", "confirm=unsubscribe").body()).contains("You are unsubscribed").doesNotContain("<form");
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(1);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION).getFirst().getValueString("scope")).isEqualTo("MARKETING");
+      assertThat(H2Support.all(EmailTables.MESSAGE).getFirst().getValueString("status")).isEqualTo("SENT");
+      assertThat(request("/unsubscribe/" + token, "POST", "x".repeat(1024 * 1024 + 1)).statusCode()).isEqualTo(413);
+   }
+
+   /*******************************************************************************
+    ** Real enqueue uses the existing signer and preserves the ad-hoc brand label.
+    *******************************************************************************/
+   @Test
+   void queuedAdHocBrandAndSigner() throws Exception
+   {
+      provider("POSTMARK");
+      QRecord brand = H2Support.insert(EmailTables.BRAND, new QRecord().withValue("name", "chosen-brand").withValue("label", "Chosen Brand"));
+      var result = com.kingsrook.qbits.email.outbox.EmailService.send(new com.kingsrook.qbits.email.outbox.EmailRequest()
+         .withTo("recipient@example.com").withSubject("Notice").withMarkdownBody("A notice").withBrandName("chosen-brand")
+         .withCategory(com.kingsrook.qbits.email.outbox.EmailCategory.NOTIFICATION));
+      assertThat(result.status()).isEqualTo(com.kingsrook.qbits.email.outbox.EmailSendResult.Status.QUEUED);
+      QRecord stored = H2Support.all(EmailTables.MESSAGE).getFirst();
+      assertThat(stored.getValueInteger("brandId")).isEqualTo(brand.getValueInteger("id"));
+      String url = stored.getValueString("unsubscribeUrl");
+      String path = "/unsubscribe/" + url.substring(url.lastIndexOf('/') + 1);
+      assertThat(request(path, "GET", null).body()).contains("Notification email from Chosen Brand?");
+      assertThat(request(path, "POST", "List-Unsubscribe=One-Click").statusCode()).isEqualTo(200);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION).getFirst().getValueString("scope")).isEqualTo("NOTIFICATION");
+   }
+
+   /*******************************************************************************
+    ** One signed mailbox is suppressed, while repeat POST restores its opt-out.
+    *******************************************************************************/
+   @Test
+   void unsubscribeSingleRecipientAndRepeatedSuppression() throws Exception
+   {
+      QRecord provider = provider("POSTMARK");
+      QRecord message = message(provider);
+      message.withValue("toAddresses", "\"Recipient, One\" <Recipient@example.com>").withValue("category", "NOTIFICATION");
+      SystemWrites.updateOne(EmailTables.MESSAGE, message);
+      QRecord other = message(provider);
+      SystemWrites.updateOne(EmailTables.MESSAGE, other.withValue("toAddresses", "other@example.com"));
+      String signed = token(message);
+      assertThat(EmailUnsubscribeLinks.verify(signed).getString("addressHash"))
+         .isEqualTo(WebhookVerification.hash("recipient@example.com".getBytes(StandardCharsets.UTF_8)));
+      String path = "/unsubscribe/" + signed;
+      assertThat(request(path, "POST", "confirm=unsubscribe").statusCode()).isEqualTo(200);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1).allSatisfy(row ->
+      {
+         assertThat(row.getValueString("scope")).isEqualTo("NOTIFICATION");
+         assertThat(row.getValueString("address")).isEqualTo("recipient@example.com");
+      });
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(1);
+      SystemWrites.delete(EmailTables.SUPPRESSION, new com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter());
+      assertThat(request(path, "POST", "List-Unsubscribe=One-Click").statusCode()).isEqualTo(200);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1);
+      assertThat(H2Support.all(EmailTables.EVENT)).hasSize(1);
+   }
+
+   /*******************************************************************************
+    ** Legacy single-recipient links keep working; legacy multi-recipient links fail closed.
+    *******************************************************************************/
+   @Test
+   void legacyUnsubscribeRequiresSingleRecipient() throws Exception
+   {
+      QRecord message = message(provider("POSTMARK"));
+      String safe = legacyToken(message);
+      assertThat(request("/unsubscribe/" + safe, "GET", null).statusCode()).isEqualTo(200);
+      for(String field : List.of("toAddresses", "ccAddresses", "bccAddresses"))
+      {
+         QRecord changed = new QRecord().withValue("id", message.getValue("id")).withValue("toAddresses", "recipient@example.com")
+            .withValue("ccAddresses", null).withValue("bccAddresses", null)
+            .withValue(field, "toAddresses".equals(field) ? "recipient@example.com, other@example.com" : "other@example.com");
+         SystemWrites.updateOne(EmailTables.MESSAGE, changed);
+         changed.withValue("category", "MARKETING");
+         String legacy = legacyToken(changed);
+         for(String signed : List.of(safe, legacy))
+         {
+            assertThat(request("/unsubscribe/" + signed, "GET", null).statusCode()).isEqualTo(404);
+            assertThat(request("/unsubscribe/" + signed, "POST", "List-Unsubscribe=One-Click").statusCode()).isEqualTo(404);
+         }
+      }
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).isEmpty();
+      assertThat(H2Support.all(EmailTables.EVENT)).isEmpty();
+      SystemWrites.updateOne(EmailTables.MESSAGE, message.withValue("ccAddresses", null).withValue("bccAddresses", null));
+      assertThat(request("/unsubscribe/" + safe, "POST", "List-Unsubscribe=One-Click").statusCode()).isEqualTo(200);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).hasSize(1);
+   }
+
+   /*******************************************************************************
+    ** Reproduce the old signature without relying on the current signer's validation.
+    *******************************************************************************/
+   private static String legacyToken(QRecord message) throws Exception
+   {
+      String envelope = message.getValueString("toAddresses") + "," + message.getValueString("ccAddresses") + "," + message.getValueString("bccAddresses");
+      JSONObject payload = new JSONObject().put("messageId", message.getValueString("id")).put("category", message.getValueString("category"))
+         .put("addressHash", WebhookVerification.hash(envelope.getBytes(StandardCharsets.UTF_8)));
+      String signed = "v1." + Base64.getUrlEncoder().withoutPadding().encodeToString(payload.toString().getBytes(StandardCharsets.UTF_8));
+      javax.crypto.Mac mac = javax.crypto.Mac.getInstance("HmacSHA256");
+      mac.init(new javax.crypto.spec.SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+      return (signed + "." + Base64.getUrlEncoder().withoutPadding().encodeToString(mac.doFinal(signed.getBytes(StandardCharsets.UTF_8))));
+   }
+
+   /*******************************************************************************
+    ** Signer and receiver reject non-opt-out categories, including valid HMACs.
+    *******************************************************************************/
+   @Test
+   void invalidAndTransactionalTokens() throws Exception
+   {
+      QRecord message = message(provider("POSTMARK"));
+      String token = token(message);
+      assertThat(request("/unsubscribe/bad-token", "GET", null).statusCode()).isEqualTo(404);
+      assertThat(request("/unsubscribe/" + token + "x", "POST", "List-Unsubscribe=One-Click").statusCode()).isEqualTo(404);
+      for(String category : List.of("TRANSACTIONAL", "SECURITY"))
+      {
+         assertThat(EmailUnsubscribeLinks.create("1", "recipient@example.com", category)).isNull();
+         JSONObject data = EmailUnsubscribeLinks.verify(token).put("category", category);
+         String signed = "v1." + Base64.getUrlEncoder().withoutPadding().encodeToString(data.toString().getBytes(StandardCharsets.UTF_8));
+         javax.crypto.Mac mac = javax.crypto.Mac.getInstance("HmacSHA256");
+         mac.init(new javax.crypto.spec.SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+         String forged = signed + "." + Base64.getUrlEncoder().withoutPadding().encodeToString(mac.doFinal(signed.getBytes(StandardCharsets.UTF_8)));
+         assertThat(request("/unsubscribe/" + forged, "POST", "List-Unsubscribe=One-Click").statusCode()).isEqualTo(404);
+      }
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", message.getValue("id")).withValue("toAddresses", "changed@example.com"));
+      assertThat(request("/unsubscribe/" + token, "GET", null).statusCode()).isEqualTo(404);
+      assertThat(H2Support.all(EmailTables.SUPPRESSION)).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Logs at normal operational levels never contain addresses or payload bodies.
+    *******************************************************************************/
+   @Test
+   void logsContainNoAddresses() throws Exception
+   {
+      QRecord provider = provider("POSTMARK");
+      message(provider);
+      List<String> logs = new java.util.concurrent.CopyOnWriteArrayList<>();
+      var capture = new org.apache.logging.log4j.core.appender.AbstractAppender("tracking-privacy", null, null, true, null)
+      {
+         @Override
+         public void append(org.apache.logging.log4j.core.LogEvent event)
+         {
+            // Match log4j2-test.xml: core's explicitly OFF diagnostics are not output.
+            if(event.getLevel() == org.apache.logging.log4j.Level.OFF)
+            {
+               return;
+            }
+            logs.add(event.getMessage().getFormattedMessage());
+            if(event.getThrown() != null)
+            {
+               logs.add(event.getThrown().toString());
+            }
+         }
+      };
+      capture.start();
+      var context = (org.apache.logging.log4j.core.LoggerContext) org.apache.logging.log4j.LogManager.getContext(false);
+      var root = context.getConfiguration().getRootLogger();
+      var previousLevel = root.getLevel();
+      root.addAppender(capture, org.apache.logging.log4j.Level.ALL, null);
+      root.setLevel(org.apache.logging.log4j.Level.INFO);
+      context.updateLoggers();
+      try
+      {
+         String body = resource("/webhooks/postmark-bounce.json");
+         assertThat(request("/postmark/postmark", "POST", body, "Authorization", auth()).statusCode()).isEqualTo(200);
+         assertThat(request("/postmark/postmark", "POST", body, "Authorization", auth()).statusCode()).isEqualTo(200);
+         assertThat(request("/postmark/postmark", "POST", body).statusCode()).isEqualTo(401);
+         new EmailInboxProcessor().runOnce();
+         String broken = new JSONObject(body).put("ID", 999).put("MessageID", "missing").toString();
+         assertThat(request("/postmark/postmark", "POST", broken, "Authorization", auth()).statusCode()).isEqualTo(200);
+         new EmailInboxProcessor().runOnce();
+         assertThat(String.join("\n", logs)).doesNotContain("recipient@example.com", "sender@example.com", SECRET, "RecordType", "550 mailbox");
+         assertThat(H2Support.all(EmailTables.INBOX).getLast().getValueString("lastError")).doesNotContain("@", "missing");
+      }
+      finally
+      {
+         root.removeAppender(capture.getName());
+         root.setLevel(previousLevel);
+         context.updateLoggers();
+         capture.stop();
+      }
+   }
+
+   /*******************************************************************************
+    ** Path validation excludes UI API routes and route metacharacters.
+    *******************************************************************************/
+   @Test
+   void routeConfigurationValidation()
+   {
+      for(String path : List.of("relative", "/qqq/v1", "/qqq/v1/email", "/hooks/{param}", "//hooks", "/hooks/"))
+      {
+         List<String> errors = new ArrayList<>();
+         EmailRuntime.config().withWebhookBasePath(path).validate(instance, errors);
+         assertThat(errors).anyMatch(error -> error.contains("webhookBasePath"));
+      }
+      List<String> errors = new ArrayList<>();
+      EmailRuntime.config().withWebhookBasePath("/qqq/hooks/email").validate(instance, errors);
+      assertThat(errors).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Both system tables are readable by viewers and reject direct user writes.
+    *******************************************************************************/
+   @Test
+   void systemTablePermissions() throws Exception
+   {
+      QContext.getQSession().setPermissions(java.util.Set.of("email.view"));
+      var checker = new com.kingsrook.qbits.email.permissions.EmailPermissionChecker();
+      for(String table : List.of(EmailTables.EVENT, EmailTables.INBOX))
+      {
+         var metadata = instance.getTable(table);
+         checker.checkPermissionsThrowing(new com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput(table), metadata);
+         assertThatThrownBy(() -> checker.checkPermissionsThrowing(new com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput(table), metadata))
+            .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
+         assertThatThrownBy(() -> checker.checkPermissionsThrowing(new com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput(table), metadata))
+            .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
+         assertThatThrownBy(() -> checker.checkPermissionsThrowing(new com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput(table), metadata))
+            .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
+      }
+      assertThat(instance.getTable(EmailTables.PROVIDER).getSections()).anyMatch(section -> (EmailTables.INBOX + "_children").equals(section.getWidgetName()));
+      assertThat(instance.getTable(EmailTables.MESSAGE).getSections()).anyMatch(section -> (EmailTables.EVENT + "_children").equals(section.getWidgetName()));
+   }
+
+   /*******************************************************************************
+    ** Sign the same single To mailbox as enqueue.
+    *******************************************************************************/
+   private static String token(QRecord message)
+   {
+      String url = EmailUnsubscribeLinks.create(message.getValueString("id"), message.getValueString("toAddresses"), message.getValueString("category"));
+      assertThat(url).startsWith("https://app.example.com" + BASE + "/unsubscribe/");
+      return (url.substring(url.lastIndexOf('/') + 1));
+   }
+}
@@ -0,0 +1,248 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+import java.net.URI;
+import java.net.http.HttpHeaders;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyFactory;
+import java.security.PrivateKey;
+import java.security.Signature;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+import java.security.spec.PKCS8EncodedKeySpec;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import javax.net.ssl.SSLSession;
+import com.kingsrook.qbits.email.H2Support;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import org.json.JSONObject;
+
+/*******************************************************************************
+ ** Synthetic provider and crypto fixtures, always non-production.
+ *******************************************************************************/
+public final class TrackingTestSupport
+{
+   public static final Instant NOW = Instant.parse("2026-10-13T12:05:00Z");
+   public static final String SECRET = "test-only-webhook-secret";
+   public static final String TOPIC = "arn:aws:sns:us-east-1:123456789012:email-events";
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private TrackingTestSupport()
+   {
+   }
+
+   /*******************************************************************************
+    ** Load a synthetic JSON or test-only PEM fixture from the classpath.
+    *******************************************************************************/
+   public static String resource(String name) throws Exception
+   {
+      try(var stream = TrackingTestSupport.class.getResourceAsStream(name))
+      {
+         return (new String(stream.readAllBytes(), StandardCharsets.UTF_8));
+      }
+   }
+
+   /*******************************************************************************
+    ** Configure a fully valid provider row without resolving its sending key.
+    *******************************************************************************/
+   public static QRecord provider(String kind) throws Exception
+   {
+      return (H2Support.insert(EmailTables.PROVIDER, new QRecord().withValue("name", kind.toLowerCase(java.util.Locale.ROOT))
+         .withValue("kind", kind).withValue("enabled", true).withValue("isDefault", H2Support.all(EmailTables.PROVIDER).isEmpty())
+         .withValue("fromAddress", "sender@example.com").withValue("sendingDomain", "example.com").withValue("region", "SES".equals(kind) ? "us-east-1" : "US")
+         .withValue("secretEnvName", switch(kind)
+         {
+            case "POSTMARK" -> "POSTMARK_SERVER_TOKEN";
+            case "SENDGRID" -> "SENDGRID_API_KEY";
+            case "MAILGUN" -> "MAILGUN_API_KEY";
+            default -> null;
+         })
+         .withValue("webhookSecretEnvName", switch(kind)
+         {
+            case "POSTMARK" -> "POSTMARK_WEBHOOK_PASSWORD";
+            case "SENDGRID" -> "SENDGRID_WEBHOOK_PUBLIC_KEY";
+            case "MAILGUN" -> "MAILGUN_WEBHOOK_SIGNING_KEY";
+            default -> null;
+         })
+         .withValue("snsTopicArn", TOPIC)));
+   }
+
+   /*******************************************************************************
+    ** Persist an accepted outbox row through the same authorized scope as delivery.
+    *******************************************************************************/
+   public static QRecord message(QRecord provider) throws Exception
+   {
+      return (SystemWrites.insertOne(EmailTables.MESSAGE, new QRecord().withValue("providerId", provider.getValue("id"))
+         .withValue("providerMessageId", "provider-message-1").withValue("status", "SENT").withValue("category", "MARKETING")
+         .withValue("toAddresses", "recipient@example.com").withValue("sentDate", NOW.minusSeconds(600))));
+   }
+
+   /*******************************************************************************
+    ** Test certificate controls the verification date independent of execution day.
+    *******************************************************************************/
+   public static Instant certificateTime() throws Exception
+   {
+      try(var stream = TrackingTestSupport.class.getResourceAsStream("/sns-test/test-only-certificate.pem"))
+      {
+         var cert = (X509Certificate) CertificateFactory.getInstance("X.509").generateCertificate(stream);
+         return (cert.getNotBefore().toInstant().plusSeconds(60));
+      }
+   }
+
+   /*******************************************************************************
+    ** Sign independently using AWS's required order, not the implementation helper.
+    *******************************************************************************/
+   public static JSONObject signSns(JSONObject payload) throws Exception
+   {
+      return (signSns(payload, certificateTime()));
+   }
+
+   /*******************************************************************************
+    ** Sign with an explicit envelope Timestamp (the verifier enforces freshness).
+    *******************************************************************************/
+   public static JSONObject signSns(JSONObject payload, Instant timestamp) throws Exception
+   {
+      payload.put("Timestamp", timestamp.toString());
+      String pem = resource("/sns-test/test-only-private-key.pem").replaceAll("-----[^-]+-----", "").replaceAll("\\s", "");
+      PrivateKey key = KeyFactory.getInstance("RSA").generatePrivate(new PKCS8EncodedKeySpec(Base64.getDecoder().decode(pem)));
+      List<String> fields = "Notification".equals(payload.getString("Type")) ? List.of("Message", "MessageId", "Subject", "Timestamp", "TopicArn", "Type")
+         : List.of("Message", "MessageId", "SubscribeURL", "Timestamp", "Token", "TopicArn", "Type");
+      StringBuilder canonical = new StringBuilder();
+      for(String field : fields)
+      {
+         if(payload.has(field))
+         {
+            canonical.append(field).append('\n').append(payload.getString(field)).append('\n');
+         }
+      }
+      Signature signer = Signature.getInstance("1".equals(payload.getString("SignatureVersion")) ? "SHA1withRSA" : "SHA256withRSA");
+      signer.initSign(key);
+      signer.update(canonical.toString().getBytes(StandardCharsets.UTF_8));
+      return (payload.put("Signature", Base64.getEncoder().encodeToString(signer.sign())));
+   }
+
+   /*******************************************************************************
+    ** Sign Mailgun's timestamp and nonce with a clearly test-only HMAC key.
+    *******************************************************************************/
+   public static JSONObject signMailgun(JSONObject body, String nonce, Instant now) throws Exception
+   {
+      String timestamp = Long.toString(now.getEpochSecond());
+      Mac mac = Mac.getInstance("HmacSHA256");
+      mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+      body.put("signature", new JSONObject().put("timestamp", timestamp).put("token", nonce)
+         .put("signature", java.util.HexFormat.of().formatHex(mac.doFinal((timestamp + nonce).getBytes(StandardCharsets.UTF_8)))));
+      return (body);
+   }
+
+   /*******************************************************************************
+    ** Memory-only HTTP response for the existing EmailHttp seam.
+    *******************************************************************************/
+   public static HttpResponse<String> response(HttpRequest request, int status, String body)
+   {
+      return (new HttpResponse<>()
+      {
+         /*******************************************************************************
+          ** Test adapter implementation.
+          *******************************************************************************/
+         @Override
+         public int statusCode()
+         {
+            return (status);
+         }
+
+         /*******************************************************************************
+          ** Test adapter implementation.
+          *******************************************************************************/
+         @Override
+         public HttpRequest request()
+         {
+            return (request);
+         }
+
+         /*******************************************************************************
+          ** Test adapter implementation.
+          *******************************************************************************/
+         @Override
+         public Optional<HttpResponse<String>> previousResponse()
+         {
+            return (Optional.empty());
+         }
+
+         /*******************************************************************************
+          ** Test adapter implementation.
+          *******************************************************************************/
+         @Override
+         public HttpHeaders headers()
+         {
+            return (HttpHeaders.of(Map.of(), (a, b) -> true));
+         }
+
+         /*******************************************************************************
+          ** Test adapter implementation.
+          *******************************************************************************/
+         @Override
+         public String body()
+         {
+            return (body);
+         }
+
+         /*******************************************************************************
+          ** Test adapter implementation.
+          *******************************************************************************/
+         @Override
+         public Optional<SSLSession> sslSession()
+         {
+            return (Optional.empty());
+         }
+
+         /*******************************************************************************
+          ** Test adapter implementation.
+          *******************************************************************************/
+         @Override
+         public URI uri()
+         {
+            return (request.uri());
+         }
+
+         /*******************************************************************************
+          ** Test adapter implementation.
+          *******************************************************************************/
+         @Override
+         public java.net.http.HttpClient.Version version()
+         {
+            return (java.net.http.HttpClient.Version.HTTP_1_1);
+         }
+
+      });
+   }
+}
@@ -0,0 +1,231 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.tracking;
+
+import java.nio.charset.StandardCharsets;
+import java.security.KeyPairGenerator;
+import java.security.Signature;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.*;
+import static org.assertj.core.api.Assertions.*;
+
+/*******************************************************************************
+ ** Crypto boundaries use genuine signatures and a public test-only RSA fixture.
+ *******************************************************************************/
+class WebhookVerifiersTest
+{
+   /*******************************************************************************
+    ** Postmark's fixed user and constant-time credential verification.
+    *******************************************************************************/
+   @Test
+   void postmarkValidTamperedMissing()
+   {
+      var verifier = new PostmarkVerifier();
+      String auth = "Basic " + Base64.getEncoder().encodeToString(("postmark:" + SECRET).getBytes(StandardCharsets.UTF_8));
+      assertThat(verifier.verify(auth, SECRET)).isTrue();
+      assertThat(verifier.verify(auth, "wrong")).isFalse();
+      assertThat(verifier.verify(auth, null)).isFalse();
+      assertThat(verifier.verify(auth, " ")).isFalse();
+      assertThat(verifier.verify(null, SECRET)).isFalse();
+      assertThat(verifier.verify("Bearer abc", SECRET)).isFalse();
+      assertThat(verifier.verify("Basic ?", SECRET)).isFalse();
+      assertThat(verifier.verify("Basic " + Base64.getEncoder().encodeToString(("other:" + SECRET).getBytes(StandardCharsets.UTF_8)), SECRET)).isFalse();
+   }
+
+   /*******************************************************************************
+    ** Genuine ECDSA validates unchanged bytes, timestamp freshness and DER key.
+    *******************************************************************************/
+   @Test
+   void sendgridValidTamperedStaleMissing() throws Exception
+   {
+      var keys = KeyPairGenerator.getInstance("EC").generateKeyPair();
+      String publicKey = Base64.getEncoder().encodeToString(keys.getPublic().getEncoded());
+      byte[] body = "[{\"event\":\"delivered\"}]".getBytes(StandardCharsets.UTF_8);
+      String timestamp = Long.toString(NOW.getEpochSecond());
+      Signature signer = Signature.getInstance("SHA256withECDSA");
+      signer.initSign(keys.getPrivate());
+      signer.update(timestamp.getBytes(StandardCharsets.UTF_8));
+      signer.update(body);
+      String signed = Base64.getEncoder().encodeToString(signer.sign());
+      var verifier = new SendGridVerifier();
+      assertThat(verifier.verify(body, signed, timestamp, publicKey, NOW)).isTrue();
+      assertThat(verifier.verify("tampered".getBytes(StandardCharsets.UTF_8), signed, timestamp, publicKey, NOW)).isFalse();
+      assertThat(verifier.verify(body, signed, timestamp, publicKey, NOW.plusSeconds(301))).isFalse();
+      assertThat(verifier.verify(body, signed, timestamp, publicKey, NOW.minusSeconds(301))).isFalse();
+      assertThat(verifier.verify(body, signed, "9223372036854775807", publicKey, NOW)).isFalse();
+      assertThat(verifier.verify(body, signed, timestamp, null, NOW)).isFalse();
+      assertThat(verifier.verify(body, null, timestamp, publicKey, NOW)).isFalse();
+      assertThat(verifier.verify(body, "?", timestamp, publicKey, NOW)).isFalse();
+      assertThat(verifier.verify(body, signed, timestamp, "?", NOW)).isFalse();
+   }
+
+   /*******************************************************************************
+    ** Mailgun validates the signed nonce, not its unsigned event body.
+    *******************************************************************************/
+   @Test
+   void mailgunValidTamperedStaleMissing() throws Exception
+   {
+      var verifier = new MailgunVerifier();
+      JSONObject data = signMailgun(new JSONObject(), "test-nonce", NOW);
+      assertThat(verifier.verify(data, SECRET, NOW)).isTrue();
+      assertThat(verifier.verify(data, SECRET, NOW.plusSeconds(301))).isFalse();
+      assertThat(verifier.verify(data, SECRET, NOW.minusSeconds(301))).isFalse();
+      assertThat(verifier.verify(data, null, NOW)).isFalse();
+      assertThat(verifier.verify(data, "wrong", NOW)).isFalse();
+      data.getJSONObject("signature").put("token", "changed");
+      assertThat(verifier.verify(data, SECRET, NOW)).isFalse();
+      assertThat(verifier.verify(new JSONObject(), SECRET, NOW)).isFalse();
+      data.getJSONObject("signature").put("token", "");
+      assertThat(verifier.verify(data, SECRET, NOW)).isFalse();
+   }
+
+   /*******************************************************************************
+    ** Both SNS algorithms, optional Subject and certificate caching.
+    *******************************************************************************/
+   @ParameterizedTest
+   @ValueSource(strings = {"1", "2"})
+   void snsSignatureAndCache(String version) throws Exception
+   {
+      String pem = resource("/sns-test/test-only-certificate.pem");
+      AtomicInteger requests = new AtomicInteger();
+      var verifier = new SnsVerifier(request ->
+      {
+         requests.incrementAndGet();
+         return (response(request, 200, pem));
+      });
+      JSONObject data = signSns(new JSONObject(resource("/webhooks/ses-bounce.json")).put("SignatureVersion", version));
+      Instant now = certificateTime();
+      assertThat(verifier.verify(data, TOPIC, now)).isTrue();
+      assertThat(verifier.verify(signSns(data.put("Subject", "Test subject")), TOPIC, now)).isTrue();
+      assertThat(requests.get()).isEqualTo(1);
+      assertThat(verifier.verify(data, TOPIC + "other", now)).isFalse();
+      assertThat(verifier.verify(data, null, now)).isFalse();
+      data.put("Message", "tampered");
+      assertThat(verifier.verify(data, TOPIC, now)).isFalse();
+      data.put("SignatureVersion", "3");
+      assertThat(verifier.verify(data, TOPIC, now)).isFalse();
+   }
+
+   /*******************************************************************************
+    ** Notifications outside the one hour past / five minute future window are refused
+    ** before any certificate fetch.
+    *******************************************************************************/
+   @Test
+   void snsTimestampWindow() throws Exception
+   {
+      AtomicInteger requests = new AtomicInteger();
+      String pem = resource("/sns-test/test-only-certificate.pem");
+      var verifier = new SnsVerifier(request ->
+      {
+         requests.incrementAndGet();
+         return (response(request, 200, pem));
+      });
+      Instant now = certificateTime();
+      JSONObject base = new JSONObject(resource("/webhooks/ses-bounce.json"));
+      assertThat(verifier.verify(signSns(new JSONObject(base.toString()), now.minusSeconds(3601)), TOPIC, now)).isFalse();
+      assertThat(verifier.verify(signSns(new JSONObject(base.toString()), now.plusSeconds(301)), TOPIC, now)).isFalse();
+      assertThat(requests.get()).isZero();
+      assertThat(verifier.verify(signSns(new JSONObject(base.toString()), now.minusSeconds(3599)), TOPIC, now)).isTrue();
+      assertThat(verifier.verify(signSns(new JSONObject(base.toString()), now.plusSeconds(299)), TOPIC, now)).isTrue();
+      assertThat(verifier.verify(signSns(new JSONObject(base.toString())).put("Timestamp", "yesterday"), TOPIC, now)).isFalse();
+   }
+
+   /*******************************************************************************
+    ** A failed certificate fetch is remembered per URL for five minutes.
+    *******************************************************************************/
+   @Test
+   void snsCachesCertificateFetchFailures() throws Exception
+   {
+      AtomicInteger requests = new AtomicInteger();
+      var verifier = new SnsVerifier(request ->
+      {
+         requests.incrementAndGet();
+         return (response(request, 503, "down"));
+      });
+      Instant now = certificateTime();
+      JSONObject data = signSns(new JSONObject(resource("/webhooks/ses-bounce.json")));
+      assertThatThrownBy(() -> verifier.verify(data, TOPIC, now)).isInstanceOf(java.io.IOException.class);
+      assertThatThrownBy(() -> verifier.verify(data, TOPIC, now.plusSeconds(299))).isInstanceOf(java.io.IOException.class);
+      assertThat(requests.get()).isEqualTo(1);
+      assertThatThrownBy(() -> verifier.verify(signSns(new JSONObject(data.toString()), now.plusSeconds(300)), TOPIC, now.plusSeconds(300))).isInstanceOf(java.io.IOException.class);
+      assertThat(requests.get()).isEqualTo(2);
+   }
+
+   /*******************************************************************************
+    ** Strict AWS URL host, scheme, path and authority before network access.
+    *******************************************************************************/
+   @ParameterizedTest
+   @ValueSource(strings = {"http://sns.us-east-1.amazonaws.com/cert.pem", "https://evil.example/cert.pem", "https://sns.us-east-1.amazonaws.com/cert.txt",
+      "https://sns.us-east-1.amazonaws.com.evil.example/cert.pem", "https://user@sns.us-east-1.amazonaws.com/cert.pem", "https://sns.us-east-1.amazonaws.com:8443/cert.pem",
+      "https://sns.us-east-1.amazonaws.com/cert.pem?x=1", "not a URI"})
+   void snsRejectsCertificateUrls(String url) throws Exception
+   {
+      var verifier = new SnsVerifier(request ->
+      {
+         throw (new AssertionError("Must not fetch untrusted URL"));
+      });
+      JSONObject data = new JSONObject(resource("/webhooks/ses-bounce.json")).put("SigningCertURL", url);
+      assertThat(verifier.verify(data, TOPIC, certificateTime())).isFalse();
+   }
+
+   /*******************************************************************************
+    ** Confirmations use AWS field order and only verified AWS URLs are fetched.
+    *******************************************************************************/
+   @Test
+   void snsConfirmationsAndFailures() throws Exception
+   {
+      String pem = resource("/sns-test/test-only-certificate.pem");
+      AtomicInteger confirms = new AtomicInteger();
+      var verifier = new SnsVerifier(request ->
+      {
+         if(request.uri().getPath().endsWith(".pem"))
+         {
+            return (response(request, 200, pem));
+         }
+         confirms.incrementAndGet();
+         return (response(request, 200, "confirmed"));
+      });
+      for(String type : List.of("SubscriptionConfirmation", "UnsubscribeConfirmation"))
+      {
+         JSONObject data = signSns(new JSONObject(resource("/webhooks/ses-bounce.json")).put("Type", type)
+            .put("SubscribeURL", "https://sns.us-east-1.amazonaws.com/?Action=ConfirmSubscription&Token=test")
+            .put("Token", "test-token"));
+         assertThat(verifier.verify(data, TOPIC, certificateTime())).isTrue();
+         verifier.confirm(data);
+         assertThatThrownBy(() -> verifier.confirm(data.put("SubscribeURL", "https://evil.example/"))).isInstanceOf(IllegalArgumentException.class);
+      }
+      assertThat(confirms.get()).isEqualTo(2);
+      assertThat(SnsVerifier.allowedUrl("https://sns.cn-north-1.amazonaws.com.cn/test.pem", true)).isTrue();
+      JSONObject data = signSns(new JSONObject(resource("/webhooks/ses-bounce.json")));
+      assertThat(new SnsVerifier(request -> response(request, 200, "bad cert")).verify(data, TOPIC, certificateTime())).isFalse();
+      assertThatThrownBy(() -> new SnsVerifier(request -> response(request, 503, "error")).verify(data, TOPIC, certificateTime())).isInstanceOf(java.io.IOException.class);
+      assertThat(verifier.verify(data, TOPIC, Instant.parse("2000-01-01T00:00:00Z"))).isFalse();
+      data.put("Type", "Unsupported");
+      assertThat(verifier.verify(data, TOPIC, certificateTime())).isFalse();
+   }
+}
@@ -1,4 +1,4 @@
-- Phase (a), parts 1 and 2. Apply before producing the qbit. Host security columns must be added by the host.
+-- Phases (a), (b) and (c). Apply before producing the qbit. Host security columns must be added by the host.

 CREATE TABLE email_provider (
   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
@@ -184,3 +184,42 @@ CREATE TABLE email_suppression (
 CREATE INDEX email_message_due ON email_message (status, next_attempt_date);
 CREATE INDEX email_attempt_provider_start ON email_message_attempt (provider_id, start_date);
 CREATE INDEX email_attempt_provider_end ON email_message_attempt (provider_id, end_date, id);
+
+CREATE TABLE email_webhook_inbox (
+   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   create_date TIMESTAMP,
+   modify_date TIMESTAMP,
+   provider_id INTEGER NOT NULL,
+   kind VARCHAR(500) NOT NULL,
+   external_event_id VARCHAR(500) NOT NULL,
+   nonce VARCHAR(500),
+   status VARCHAR(500) NOT NULL,
+   attempts INTEGER NOT NULL DEFAULT 0,
+   next_attempt_date TIMESTAMP,
+   last_error VARCHAR(500),
+   payload_json TEXT,
+   payload_sha256 VARCHAR(500),
+   received_date TIMESTAMP,
+   processed_date TIMESTAMP,
+   UNIQUE (provider_id, external_event_id),
+   UNIQUE (provider_id, nonce)
+);
+
+CREATE TABLE email_event (
+   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   create_date TIMESTAMP,
+   modify_date TIMESTAMP,
+   message_id INTEGER NOT NULL,
+   provider_id INTEGER,
+   event_type VARCHAR(500) NOT NULL,
+   bounce_type VARCHAR(500),
+   recipient VARCHAR(500),
+   occurred_date TIMESTAMP,
+   detail VARCHAR(500),
+   url TEXT,
+   event_key VARCHAR(500) NOT NULL UNIQUE
+);
+CREATE INDEX email_inbox_due ON email_webhook_inbox (status, next_attempt_date);
+CREATE INDEX email_event_message ON email_event (message_id);
+CREATE INDEX email_event_type_date ON email_event (event_type, occurred_date);
+CREATE INDEX email_message_provider_id ON email_message (provider_id, provider_message_id);
@@ -0,0 +1,13 @@
+# Test-only SNS certificate and private key
+
+These are public test fixtures, never production credentials. Generated at authoring
+time with OpenSSL; tests do not spawn OpenSSL or write certificate files.
+
+```
+openssl req -x509 -newkey rsa:2048 -nodes -sha256 -days 36500 \
+  -subj "/CN=TEST ONLY - SNS webhook verification fixture" \
+  -keyout test-only-private-key.pem -out test-only-certificate.pem
+```
+
+The fake EmailHttp serves this certificate for an AWS-shaped test URL. SNS tests
+sign with this key in memory. Never trust this certificate outside the tests.
@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDUTCCAjmgAwIBAgIUMxFivUctd7E0lQz94zgDEkGMm8EwDQYJKoZIhvcNAQEL
+BQAwNzE1MDMGA1UEAwwsVEVTVCBPTkxZIC0gU05TIHdlYmhvb2sgdmVyaWZpY2F0
+aW9uIGZpeHR1cmUwIBcNMjYxMDE0MDI0OTM4WhgPMjEyNjA5MjAwMjQ5MzhaMDcx
+NTAzBgNVBAMMLFRFU1QgT05MWSAtIFNOUyB3ZWJob29rIHZlcmlmaWNhdGlvbiBm
+aXh0dXJlMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAwDnXnAbhkNrq
+Hi05Rl33RojxWRN8N81yEISzyrp8zTBXg15Icp8dkLIEkTYrcZv5GDZ/ROE95nHG
+Ue9ODyK3ga8X4yyHZQEtc3F0SyCwwCg/sRIx2F217cnrr2jbF57mZBiUqcr0+naa
+ZyqpBsV4iBnZSIQchi0RWTkPI7kj0oJi9uWBOf7sVQBlKFyboTRloFLxnB4XdBQ5
+PYNiNT9pv8drCe3B9OcAu9odhRuI/fvws1C+zlDxqxlCvYGonUvGD9ob9x5NkuUg
+hABxyUYcZWgMzSBTcrL7vIZyTLsgEqRB2IZQDeqZhpcEc2ewrWvnTPzPVc9rzew1
+8hltQd4E2wIDAQABo1MwUTAdBgNVHQ4EFgQUIfc+4qImdzoXevD8eAK8iFynrTQw
+HwYDVR0jBBgwFoAUIfc+4qImdzoXevD8eAK8iFynrTQwDwYDVR0TAQH/BAUwAwEB
+/zANBgkqhkiG9w0BAQsFAAOCAQEAFaUW8M16oVhV/lERu+JXeasCtvLrudrufomH
+HybCP+YHbgee4RnIBDjKOzDZRpbjJmi/a7rJidtpdnD8Dn1lexB7wEDmgDdyB4oT
+4R62dKuOQgnL/RuDUUp3YYMZliSppSNQRgOYl7Islq3zmpYBCBAV9s4fiUHW8uBy
+ZYRQbF49xxPINpSreMJtrf+Pq9E1As4cCc3ViSS1ITsH4R515xJvQg64HQtunl8W
+rSzTfsAADCqqFapmdLQiHSJWIEXxbNv1B0vBSIJCQkNsjTceblgqYTAdK1Kufsw
+s5ZWUbQHAIXODa3VRvaislvH+ygHAtIBi1kyV33c2AiuPPVuqg==
+-----END CERTIFICATE-----
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDAOdecBuGQ2uoe
+LTlGXfdGiPFZE3w3zXIQhLPKunzNMFeDXkhynx2QsgSRNitxm/kYNn9E4T3mccZR
+704PIreBrxfjLIdlAS1zcXRLILDAKD+xEjHYXbXtyeuvaNsXnuZkGJSpyvT6dppn
+KqkGxXiIGdlIhByGLRFZOQ8juSPSgmL25YE5/uxVAGUoXJuhNGWgUvGcHhd0FDk9
+g2I1P2m/x2sJ7cH05wC72h2FG4j9+/CzUL7OUPGrGUK9gaidS8YP2hv3Hk2S5SCE
+AHHJRhxlaAzNIFNysvu8hnJMuyASpEHYhlAN6pmGlwRzZ7Cta+dM/M9Vz2vN7DXy
+GW1B3gTbAgMBAAECggEATrQgHEkQ9+855iscBTyZ7u5I6vC4blyWtD6xInMigByn
+6CilPxD42PKaPh5QwU5cOdauWzRh1MWGrqttaXMMcXOQMzpbITbPNjIgrFXxKGwS
+T9ykPfN4dqiy7/ndopkzuY366F4p2ZbP9yeXVydpB0iWF0PILgOr070x/q1iBmEX
+zY/rTgSO3xhS6fJOtZdCFvSY7Geg8pPYJ1pAYBaz2nPL1WYXN4M0qpllBLKBwwnf
+i6PRxLAyATLpOJgmI9sGmbfvu7yKRIAvVItR+2r5v03Fh7brKEOq9YmmzY7Hx62L
+VWsr6CMzUrsUm7R54epPeBBvhZ0AIWAf9gIDemvmUQKBgQDz4lms/uFjn6h9aoNX
+Xk/AdZGKp44d3vXF7X3MrKB10zPPg13n/Wwp0DrGP08aZMMbWevROIHfzToTFHkW
+sYiwmazF9y0+6tf0BmV8EuR7vJtIhKwTJAgdFbbTylXUdNc+5rh+O6IoImlpCD75
+rQfPe3fezlxzYmC43Q68WfRsYwKBgQDJxoRl8/a0jNx4ZFZzYfxk/uG5/HdY+dSS
+SecJvj9M6CTDbA8k8wBWgdsgEgvB5qCjTe7Bci9dagDcL4qUsAl4H05I09VVIbUd
+3h7JCWwVFr/Dvn6L7tqeJXlq5yDPfYDPvbe2EK+x/Zq/QTVtPYFqtlH83U78cU4N
+3g9mP1aDKQKBgCQU+O9/RhlXJbZoa8YhTetLeZzp7Ty+H96IpHQDljnYZXHuHZaV
+B8un4woxk6J0BemmBB2QHqbd1hm8ja2t1mtKUSV4zFjipdCUoa8ZnHQyE0dHWa37
+RrgG3mW3KbulDbNgryr955/Id4/yiF6oGt+3hKSSI1tu0qCHFAcJx//BAoGAJE4j
+5HRRH4t43/Ct3XT0BPgXhW93yfqp1OiXOZmRqjTa6I646t0OYoJa+xpEI09r9/Y9
+z3DMNPKt1PB47uf9zC9V1qTG7dpyzDz7LREU8XZ2arIL8wG3NF1py2Jq9OqQEtEA
+DTJTaqNHBMKm6lD2yEyl0+m+HI4GjVg3TR54YoECgYEA3kOMd1NuVDNHQU6qRD54
+oGvym00jaCYCH1mQvS7VWOkXfmT6ZnKw+qRLcoUGTtP76pcWNYQQMyUDfbdtQp1u
+gdrlS/zASsQhuDaKrCiuwOIlTvlx7InRi2MqpN+ztus8hZzyBHg9EHUWGykZ7rRh
+XtLanuf3Tn3MvyiHNaVPyQs=
+-----END PRIVATE KEY-----
@@ -0,0 +1,23 @@
+{
+  "signature": {
+    "timestamp": "1791892800",
+    "token": "test-mailgun-nonce",
+    "signature": "signed-in-test"
+  },
+  "event-data": {
+    "id": "mailgun-event-1",
+    "event": "failed",
+    "severity": "permanent",
+    "timestamp": 1791892800,
+    "recipient": "recipient@example.com",
+    "message": {
+      "headers": {
+        "message-id": "<provider-message-1>"
+      }
+    },
+    "delivery-status": {
+      "message": "550 recipient@example.com unavailable"
+    },
+    "user-variables": {}
+  }
+}
@@ -0,0 +1,9 @@
+{
+  "RecordType": "Bounce",
+  "ID": 10001,
+  "MessageID": "provider-message-1",
+  "Type": "HardBounce",
+  "Email": "recipient@example.com",
+  "BouncedAt": "2026-10-13T12:01:00Z",
+  "Description": "550 mailbox recipient@example.com is unavailable"
+}
@@ -0,0 +1,7 @@
+{
+  "RecordType": "Delivery",
+  "MessageID": "provider-message-1",
+  "Recipient": "recipient@example.com",
+  "DeliveredAt": "2026-10-13T12:00:00Z",
+  "Details": "250 accepted"
+}
@@ -0,0 +1,19 @@
+[
+  {
+    "email": "recipient@example.com",
+    "timestamp": 1791892800,
+    "event": "delivered",
+    "sg_event_id": "sg-event-1",
+    "sg_message_id": "provider-message-1.filter001",
+    "response": "250 accepted"
+  },
+  {
+    "email": "recipient@example.com",
+    "timestamp": 1791892860,
+    "event": "bounce",
+    "type": "bounce",
+    "sg_event_id": "sg-event-2",
+    "sg_message_id": "provider-message-1.filter001",
+    "reason": "550 recipient@example.com does not exist"
+  }
+]
@@ -0,0 +1,10 @@
+{
+  "Type": "Notification",
+  "MessageId": "sns-envelope-1",
+  "TopicArn": "arn:aws:sns:us-east-1:123456789012:email-events",
+  "Timestamp": "2026-10-13T12:00:00Z",
+  "SignatureVersion": "2",
+  "SigningCertURL": "https://sns.us-east-1.amazonaws.com/test-only.pem",
+  "Signature": "signed-in-test",
+  "Message": "{\"eventType\": \"Bounce\", \"mail\": {\"messageId\": \"provider-message-1\", \"destination\": [\"recipient@example.com\", \"second@example.com\"], \"headers\": [{\"name\": \"X-QQQ-Email-Id\", \"value\": \"1\"}]}, \"bounce\": {\"bounceType\": \"Permanent\", \"bounceSubType\": \"General\", \"timestamp\": \"2026-10-13T12:01:00Z\", \"bouncedRecipients\": [{\"emailAddress\": \"recipient@example.com\", \"diagnosticCode\": \"smtp; 550 recipient@example.com unavailable\"}, {\"emailAddress\": \"second@example.com\", \"diagnosticCode\": \"smtp; 550 second@example.com unavailable\"}]}}"
+}
Author	SHA1	Message	Date
james.maesandClaude Opus 5.5	f8f03b1416	Merge feature/phase-a-templates (with develop) into feature/phase-c-tracking Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-14 07:21:37 -05:00
james.maesandClaude Opus 5.5	40d54226b2	fix: review findings on tracking, and a template API for other qbits - Webhook inbox rows apply matched events independently; unmatched events retry, then are skipped with a note instead of failing the row. - Notification and marketing messages are never deleted by retention, so unsubscribe links keep working; their bodies are still cleared. - Events only suppress recipients of the matched message; an SES bounce without recipients no longer suppresses every destination. - SNS notifications outside a 1 h / +5 min window are refused, and failed certificate fetches are cached for 5 minutes. - EmailTemplateDefinition and EmailTemplates.ensure let other qbits register their templates once without overwriting edits. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-13 23:00:26 -05:00
james.maesandClaude Opus 5.5	4988a0740f	Merge feature/phase-a-templates into feature/phase-c-tracking Keeps the polished home layout with phase (c)'s event-backed metrics. Also fixes unsubscribe privacy: a message with an unsubscribe link goes to exactly one recipient, its token binds that address, and the handler suppresses only that address (DESIGN 8.2). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-13 22:28:42 -05:00
james.maesandClaude Opus 5.5	6af5062aa7	feat: delivery tracking, suppression from events, and one-click unsubscribe Phase (c) of docs/DESIGN.md: the email_webhook_inbox and email_event tables; a /qqq/hooks/email route provider with SES (SNS signature and certificate host rules, subscription confirmation), Postmark (basic auth), SendGrid (ECDSA) and Mailgun (HMAC with replay protection) verification into a durable inbox; a claim-safe inbox processor that maps provider events to email_event rows, message status and suppressions; RFC 8058 one-click unsubscribe with a confirmation page; the Reprocess action; and event-backed home metrics. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>	2026-10-13 22:14:50 -05:00
james.maes	17c7dc1140	Merge remote-tracking branch 'origin/feature/phase-b-providers' into feature/phase-c-tracking # Conflicts: # CHANGELOG.md # README.md	2026-10-13 21:37:12 -05:00