diff --git a/CHANGELOG.md b/CHANGELOG.md index a5cf2bb..698007f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,13 +5,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and ## [Unreleased] +### Changed +- Built on QQQ 4.1.0-RC.7 (default QQQ profile). + ### Added +- `EmailJobEffectTarget` automatically registers the durable-jobs `email` target, with idempotent outbox handoff, refusal reasons, reconciliation and job-owner session enforcement; H2 integration coverage and payload/registration documentation. - `EmailApiReadiness.failedChecks()` and `failed(check)` with `EmailApiReadinessCheck` (NOT_INSTALLED, PROVIDER_MISSING, PROVIDER_INVALID, PHI_PROVIDER, PUBLIC_BASE_URL, UNSUBSCRIBE_KEY, BRAND_POSTAL_ADDRESS, BODY_KEY), so callers react to a specific prerequisite without parsing reason text. - `EmailApi.sendOnce` as the public durable-effect handoff, with API-only outcome/result types, committed enqueue/replay semantics and suppression/transaction refusal parity. Exact idempotency-key and RFC Message-ID lookups return `EmailApiMessage`, which now includes `rfcMessageId` across all lookup routes. Public requests support ordinary caller transactions; `EmailEffects`, `EmailEffectOutcome` and `EmailMessageRef` are deprecated internals. H2 parity, reconciliation, storage-failure and API-boundary coverage; no new dependencies. - Stable `com.kingsrook.qbits.email.api` contract for cross-qbit sends, create-once templates, metadata-only message lookups, enabled sender addresses, default provider selection and secret-safe readiness (including marketing and PHI prerequisites). JDK/QQQ-only signatures, complete request/status mapping tests and a reflection package-boundary guard; no new dependencies. - Documented both supported cross-qbit routes: core-only `SendMessageAction` via `qbitEmail`, and the public API for richer sends and lookups. - Persisted unique RFC Message-IDs, stable across delivery attempts, with SMTP/Gmail SMTP, SES raw MIME and Postmark/SendGrid/Mailgun request propagation (SES replaces the header downstream). -- Core-independent `EmailEffects` handoff, reference lookups by idempotency key or RFC Message-ID, refusal mapping and an RC.7 binding design sketch; offline retry and provider-contract tests. +- Core-independent `EmailEffects` handoff, reference lookups by idempotency key or RFC Message-ID and refusal mapping; offline retry and provider-contract tests. - Checkstyle now checks test sources as well as main sources; existing test imports and formatting corrected. - Provider sender addresses: child table `email_provider_sender_address` (validated lowercase address, display name, one default, enabled, note; unique per provider; domain bound to the provider's sending domain; admin-only), shown on the provider record. `EmailRequest.withFromAddressId`, `from_address_id` on messages, refusal of a from address that is not an enabled row of the resolved provider, and the provider `from_address` / `from_name` kept as the fallback. - Contingent possible-value filters on the new from-address pickers (send processes by `providerId`, Send Test Email by the acted-on provider, message `fromAddressId` by `providerId`) and on the attempt and event message pickers (by `providerId`) and the suppression source event (by `sourceMessageId`); the send processes also take `providerId`. diff --git a/README.md b/README.md index c7e5c1c..70d9b7f 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ qbit-object-storage is wired in. ## Install -Requires Java 21, QQQ **4.1.0-RC.5**, and a transactional RDBMS with the supplied unique +Requires Java 21, QQQ **4.1.0-RC.7**, and a transactional RDBMS with the supplied unique constraints. Apply [the PostgreSQL DDL](docs/ddl/email-postgres.sql) before producing the qbit. SQL timestamps store UTC values without a timezone, matching QQQ RC.5. The physical table names equal their metadata names; `backendName` selects the host's QQQ database backend, as it does in qbit-webhooks. @@ -588,23 +588,29 @@ PHI canary egress, encryption, preview widgets and home data. ## Durable jobs -`EmailEffects.send(request, effectIdempotencyKey)` hands a message to the committed -email outbox exactly once per key and returns an `EmailEffects.Result` containing -`EmailEffectOutcome` and an `EmailMessageRef`. `QUEUED_OR_DUPLICATE` means the outbox -accepted responsibility; `REFUSED` means a definite refusal, including suppression. -Use `findByIdempotencyKey` or `findByRfcMessageId` to reconcile a handoff. These are -trusted system reads returning only references. Database failures propagate. -Requests must not carry a caller transaction; the effect dispatcher calls after its -own chunk commit. Existing `EmailService` transaction support is unchanged. +Built on QQQ **4.1.0-RC.7**. Enable `DurableJobsMetaData` and define the jobs tables +**before** producing the email qbit: `EmailQBitProducer` automatically registers +`EmailJobEffectTarget` under target name `email`. If jobs are enabled later, call +`EmailJobEffectTarget.register(instance)` afterward. Registration is a no-op when +jobs are disabled. -Each new message stores a unique `rfcMessageId` once. SMTP (including the Gmail preset), -Postmark, SendGrid and Mailgun carry it on every attempt. SES receives it in the raw -request but replaces it downstream; every transport retains `X-QQQ-Email-Id`. -Message-ID enables correlation and receiver deduplication where supported; it does -not promise exactly-once inbox delivery. +Declare `new JobEffectDeclaration("welcomeEmail", "", "email", +JobEffectType.IDEMPOTENT_EXTERNAL)` in the process's `DurableProcessMetaData`, then +call `JobEffects.emit(input, "welcomeEmail", rowKey, payload)` in the backend step. +The payload accepts recipients, template or ad-hoc bodies, variables, routing and +record linkage; see the [complete payload contract and examples](docs/DURABLE-JOBS.md). -See [the handoff contract, provider references, schema upgrade and RC.7 binding sketch](docs/DURABLE-JOBS.md). -The build remains on released QQQ **4.1.0-RC.5**, without durable-jobs dependencies. +The target deduplicates handoffs by the effect key. SENT means the email outbox has +committed, while its dispatcher owns provider delivery and retries. Job effects +send under the job owner's current session and captured security scope; an owner +whose session cannot be established produces FAILED. Other exceptions produce +UNKNOWN and can safely be retried using the same key. Reconcile finds the existing +outbox row, including terminal audit rows. + +Each message retains its stored RFC Message-ID across provider attempts. SMTP, +Postmark, SendGrid and Mailgun carry it; SES replaces it downstream. This supports +correlation and receiver deduplication where available, without guaranteeing +exactly-once inbox delivery. See the [schema upgrade and provider notes](docs/DURABLE-JOBS.md). ## License diff --git a/docs/DURABLE-JOBS.md b/docs/DURABLE-JOBS.md index ee42f04..efb2330 100644 --- a/docs/DURABLE-JOBS.md +++ b/docs/DURABLE-JOBS.md @@ -1,8 +1,8 @@ # Durable job email effects -This implements the email-side contract for QRun-IO/qqq#1191, spec section 2.13, -without importing core jobs types. QQQ stays at released 4.1.0-RC.5. The -`EmailJobEffectTarget implements EffectTarget` binding is deferred until RC.7 ships. +QQQ 4.1.0-RC.7 supplies the durable jobs side-effects outbox. QBit Email registers +`com.kingsrook.qbits.email.outbox.EmailJobEffectTarget` as target `email` to hand +those effects to its own committed outbox. All provider sending remains asynchronous. ## Stored identity and schema upgrade @@ -86,44 +86,101 @@ existing outbox row (including an audit row); it must never create another email merely because that row is terminal. Retain message rows and idempotency keys for at least the maximum effect retry/reconciliation horizon. -## RC.7 binding design note (not compiled) +## Register, declare and emit -Mapping: `deliver -> QUEUED_OR_DUPLICATE => SENT(ref.id)`, `REFUSED => FAILED`; -`reconcile -> getMessageByIdempotencyKey present => FOUND(ref.id), absent => NOT_FOUND`. -Exceptions escape to core's retry/unknown handling. Payload decoding must preserve -the existing PHI/variable policies and must not attach a caller transaction. - -This 25-line sketch is documentation-only. Match core names and signatures to the -published RC.7 contract when it ships; `decode` represents payload conversion. +Enable durable jobs **before** producing the email qbit, so the producer can register +its target automatically: ```java -// Future RC.7 binding; no core jobs imports in the RC.5 implementation. -final class EmailJobEffectTarget implements EffectTarget -{ - // SENT acknowledges the committed qbit outbox row. - @Override - public DeliveryResult deliver(Effect effect) throws Exception - { - EmailApiRequest request = decode(effect.payload()); - EmailApiEffectResult result = EmailApi.sendOnce(request, effect.idempotencyKey()); - if(result.outcome() == EmailApiEffectOutcome.REFUSED) - { - return (DeliveryResult.failed("Email refused")); - } - return (DeliveryResult.sent(result.message().id().toString())); - } - - // Existence reconciles handoff independently of mailbox delivery. - @Override - public ReconcileResult reconcile(Effect effect) throws Exception - { - return (EmailApi.getMessageByIdempotencyKey(effect.idempotencyKey()) - .map(ref -> ReconcileResult.found(ref.id().toString())) - .orElseGet(ReconcileResult::notFound)); - } -} +instance.withSupplementalMetaData(new DurableJobsMetaData()); +new DurableJobsMetaDataProvider().defineAll(instance, "rdbms", null); +new EmailQBitProducer() + .withQBitConfig(new EmailQBitConfig().withBackendName("rdbms")) + .produce(instance); ``` +Apply core's jobs database schema as usual. `EmailJobEffectTarget.register(instance)` +is a no-op without `DurableJobsMetaData`. If jobs are enabled after producing email, +call this helper explicitly after enabling them. + +Declare the effect on the process's `DurableProcessMetaData` (retain any existing +durable settings). Replace `` with the process's real backend step name: + +```java +process.withSupplementalMetaData(new DurableProcessMetaData() + .withEffect(new JobEffectDeclaration("welcomeEmail", "", "email", + JobEffectType.IDEMPOTENT_EXTERNAL))); +``` + +Then emit from that step using its `RunBackendStepInput input` and a stable row key: + +```java +Map payload = new LinkedHashMap<>(); +payload.put("to", "person@example.com"); +payload.put("subject", "Welcome"); +payload.put("textBody", "Your account is ready."); +JobEffects.emit(input, "welcomeEmail", rowKey, payload); +``` + +The jobs classes above come from `com.kingsrook.qqq.backend.core.jobs` and its +`effects` package; `Map`, `LinkedHashMap` and `Serializable` are JDK types. +Core stores the effect with the chunk's progress and dispatches after commit. +Outside a durable job, `JobEffects.emit` calls the target immediately with a fresh +key, still only enqueueing email; a result other than SENT throws. + +Use **IDEMPOTENT_EXTERNAL**: qbit-email deduplicates by the effect's exact key, +while its own delivery outbox reuses the stored Message-ID on provider attempts. +This acknowledges one durable handoff, not exactly-once delivery to a mailbox. + +## Payload contract + +Unknown keys are ignored. Recognized values must have the following types: + +| Keys | Values | +|---|---| +| `to`, `cc`, `bcc` | A String mailbox or a list of String mailboxes; omitted means empty. Comma-separated strings are not split. | +| `subject`, `templateName` | Strings; ordinary template/ad-hoc validation applies. | +| `markdownBody`, `htmlBody`, `textBody` | String body alternatives. | +| `variables` | Map with String keys and serializable values; requires a template schema and retains the normal PHI/classification checks. | +| `category` | String enum name: `TRANSACTIONAL`, `SECURITY`, `NOTIFICATION`, or `MARKETING`. | +| `providerName`, `brandName`, `locale` | Strings; omitted locale retains `en`. | +| `from`, `replyTo` | String mailboxes, subject to ordinary sender validation. | +| `fromAddressId` | Integer identifying an enabled address for the selected provider. | +| `sendAfter` | ISO-8601 Instant String, for example `2026-10-14T12:00:00Z`. | +| `linkedTableName`, `linkedRecordId` | String table name; record id converted to String. | + +The target always sets `senderQbit = "jobs"`, `senderProcess = effectName`, and +uses `EffectRequest.idempotencyKey()`. Payload keys cannot override these or attach +a transaction. `recordVariables`, `securityValues` and `payloadRef` are not decoded; +provide the supported values directly in `payload`. + +## Identity and outcomes + +The dispatcher invokes the target as the system user. When `jobId` is present, the +target loads the job under core's system scope and builds its owner session through +`JobSessions.forJob`. Sending runs inside `QContext.withTemporaryContext` using that +owner session, with current permissions and the intersection of current and captured +security keys. The original dispatcher context is restored even after an exception. +Without a job id, sending retains the caller's session. + +- `QUEUED_OR_DUPLICATE` becomes `EffectResult.sent(outboxId)`; `providerRef` is the + String email row id, not the transport provider's reference. +- `REFUSED` becomes `EffectResult.failed(reasons)`, including suppressed replays. +- A missing job or an owner session rejected by authentication/permission validation + becomes FAILED with `The job's owner can no longer send it: ...`. +- Other exceptions, including payload decoding and database failures, become UNKNOWN. + Error text omits raw exception details to avoid exposing payloads or database values. + Core may resend IDEMPOTENT_EXTERNAL effects using the same key. +- Reconcile uses `EmailEffects.findByIdempotencyKey`: any existing row becomes FOUND + with its id, otherwise NOT_FOUND. Lookup failures propagate rather than falsely + reporting absence. Reconcile never creates mail or changes the owner session. + +The binding uses `EmailEffects` internally; other consuming qbits continue to use +`EmailApi`. The H2 target tests install core jobs metadata with `defineAll`, create +all columns of the `jobs_job` lookup table, and observe the owner through an insert +customizer because email messages have no built-in created-by/user field. No +production job-lookup test seam is needed. + Tests use H2, the existing GreenMail SMTP sink, fake SES and HTTP request seams; no vendor endpoints are called. The PostgreSQL migration is supplied for host deployment and is not executed by the offline Java test gate. diff --git a/pom.xml b/pom.xml index 5aeb48e..81b933a 100644 --- a/pom.xml +++ b/pom.xml @@ -241,7 +241,7 @@ true - 4.1.0-RC.5 + 4.1.0-RC.7 diff --git a/src/main/java/com/kingsrook/qbits/email/EmailQBitProducer.java b/src/main/java/com/kingsrook/qbits/email/EmailQBitProducer.java index f89cc91..523c9f2 100644 --- a/src/main/java/com/kingsrook/qbits/email/EmailQBitProducer.java +++ b/src/main/java/com/kingsrook/qbits/email/EmailQBitProducer.java @@ -197,6 +197,7 @@ public class EmailQBitProducer implements QBitMetaDataProducer output.add(new EmailOutboxMessagingProviderMetaData().withName(qBitConfig.getBridgeProviderName())); output.addSelfToInstance(qInstance); registerWebhooks(qInstance); + com.kingsrook.qbits.email.outbox.EmailJobEffectTarget.register(qInstance); return (new MetaDataProducerMultiOutput()); } diff --git a/src/main/java/com/kingsrook/qbits/email/outbox/EmailEffects.java b/src/main/java/com/kingsrook/qbits/email/outbox/EmailEffects.java index 6564a2f..880e602 100644 --- a/src/main/java/com/kingsrook/qbits/email/outbox/EmailEffects.java +++ b/src/main/java/com/kingsrook/qbits/email/outbox/EmailEffects.java @@ -42,8 +42,15 @@ public final class EmailEffects /******************************************************************************* ** A refusal can have a reference when suppression retained an audit row. *******************************************************************************/ - public record Result(EmailEffectOutcome outcome, EmailMessageRef message) + public record Result(EmailEffectOutcome outcome, EmailMessageRef message, List reasons) { + /******************************************************************************* + ** Preserve the original constructor for callers that only need the reference. + *******************************************************************************/ + public Result(EmailEffectOutcome outcome, EmailMessageRef message) + { + this(outcome, message, List.of()); + } } @@ -68,14 +75,15 @@ public final class EmailEffects || request.getTransaction() != null || request.getIdempotencyKey() != null && !request.getIdempotencyKey().equals(effectIdempotencyKey)) { - return (new Result(EmailEffectOutcome.REFUSED, null)); + return (new Result(EmailEffectOutcome.REFUSED, null, List.of("An email request without a caller transaction and a valid, matching effect key are required"))); } request.withIdempotencyKey(effectIdempotencyKey); EmailSendResult result = EmailService.send(request); EmailMessageRef message = result.messageId() == null ? null : find("id", result.messageId()).orElseThrow(() -> new QException("Queued email reference is unavailable")); boolean refused = result.status() == EmailSendResult.Status.REFUSED || result.status() == EmailSendResult.Status.SUPPRESSED || message != null && "SUPPRESSED".equals(message.status()); - return (new Result(refused ? EmailEffectOutcome.REFUSED : EmailEffectOutcome.QUEUED_OR_DUPLICATE, message)); + List reasons = message != null && "SUPPRESSED".equals(message.status()) ? List.of("All recipients are suppressed") : result.reasons(); + return (new Result(refused ? EmailEffectOutcome.REFUSED : EmailEffectOutcome.QUEUED_OR_DUPLICATE, message, reasons)); } diff --git a/src/main/java/com/kingsrook/qbits/email/outbox/EmailJobEffectTarget.java b/src/main/java/com/kingsrook/qbits/email/outbox/EmailJobEffectTarget.java new file mode 100644 index 0000000..0052013 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/email/outbox/EmailJobEffectTarget.java @@ -0,0 +1,200 @@ +/* + * 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.outbox; + + +import java.io.Serializable; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; +import com.kingsrook.qqq.backend.core.context.CapturedContext; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException; +import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaData; +import com.kingsrook.qqq.backend.core.jobs.JobSessions; +import com.kingsrook.qqq.backend.core.jobs.effects.EffectRequest; +import com.kingsrook.qqq.backend.core.jobs.effects.EffectResult; +import com.kingsrook.qqq.backend.core.jobs.effects.EffectTarget; +import com.kingsrook.qqq.backend.core.jobs.effects.ReconcileResult; +import com.kingsrook.qqq.backend.core.jobs.store.DurableJobStore; +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.model.session.QSession; + + +/******************************************************************************* + ** Durable jobs target for IDEMPOTENT_EXTERNAL email effects. SENT acknowledges + ** the committed email outbox, whose key deduplication makes retries safe. The + ** email dispatcher owns provider delivery and preserves the stored Message-ID. + ** Payload keys and types are documented in docs/DURABLE-JOBS.md. + *******************************************************************************/ +public class EmailJobEffectTarget implements EffectTarget +{ + public static final String TARGET_NAME = "email"; + + + + /******************************************************************************* + ** Register after enabling durable jobs; leave instances without jobs alone. + *******************************************************************************/ + public static void register(QInstance instance) + { + DurableJobsMetaData jobs = DurableJobsMetaData.of(instance); + if(jobs != null) + { + jobs.withEffectTarget(TARGET_NAME, new QCodeReference(EmailJobEffectTarget.class)); + } + } + + + + /******************************************************************************* + ** Enqueue under the job owner's current permissions and captured scope. + ** Other exceptions are uncertain; core may safely resend using the same key. + *******************************************************************************/ + @Override + public EffectResult deliver(EffectRequest r) + { + try + { + EmailRequest request = decode(r.payload()).withSender("jobs", r.effectName()); + EmailEffects.Result result; + if(r.jobId() == null) + { + result = EmailEffects.send(request, r.idempotencyKey()); + } + else + { + QInstance instance = QContext.getQInstance(); + QSession owner; + try + { + QRecord job = new DurableJobStore(DurableJobsMetaData.of(instance)).getJob(r.jobId()); + if(job == null) + { + return (EffectResult.failed("The job's owner can no longer send it: job not found")); + } + owner = JobSessions.forJob(instance, job); + } + catch(QPermissionDeniedException | QAuthenticationException e) + { + return (EffectResult.failed("The job's owner can no longer send it: owner session unavailable")); + } + AtomicReference handoff = new AtomicReference<>(); + QContext.withTemporaryContext(new CapturedContext(instance, owner), + () -> handoff.set(EmailEffects.send(request, r.idempotencyKey()))); + result = handoff.get(); + } + if(result.outcome() == EmailEffectOutcome.REFUSED) + { + return (EffectResult.failed(String.join("; ", result.reasons()))); + } + return (EffectResult.sent(String.valueOf(result.message().id()))); + } + catch(Exception e) + { + // Exception details can contain payloads or database values. + return (EffectResult.unknown("Email handoff could not be confirmed (" + e.getClass().getSimpleName() + ")")); + } + } + + + + /******************************************************************************* + ** Reconcile handoff by existence, including suppressed and terminal audit rows. + *******************************************************************************/ + @Override + public ReconcileResult reconcile(EffectRequest r) throws QException + { + return (EmailEffects.findByIdempotencyKey(r.idempotencyKey()) + .map(message -> ReconcileResult.found(String.valueOf(message.id()))) + .orElseGet(ReconcileResult::notFound)); + } + + + + /******************************************************************************* + ** Decode only supported values; identity, sender and transaction cannot be + ** overridden by payload keys. Variables retain normal template/PHI validation. + *******************************************************************************/ + static EmailRequest decode(Map payload) + { + Map values = payload == null ? Map.of() : payload; + EmailRequest request = new EmailRequest() + .withTo(addresses(values.get("to"))) + .withCc(addresses(values.get("cc"))) + .withBcc(addresses(values.get("bcc"))) + .withFrom((String) values.get("from")) + .withFromAddressId((Integer) values.get("fromAddressId")) + .withReplyTo((String) values.get("replyTo")) + .withSubject((String) values.get("subject")) + .withTemplateName((String) values.get("templateName")) + .withMarkdownBody((String) values.get("markdownBody")) + .withHtmlBody((String) values.get("htmlBody")) + .withTextBody((String) values.get("textBody")) + .withProviderName((String) values.get("providerName")) + .withBrandName((String) values.get("brandName")) + .withLinkedRecord((String) values.get("linkedTableName"), values.get("linkedRecordId")); + if(values.get("locale") != null) + { + request.withLocale((String) values.get("locale")); + } + if(values.get("category") != null) + { + request.withCategory(EmailCategory.valueOf((String) values.get("category"))); + } + if(values.get("sendAfter") != null) + { + request.withSendAfter(Instant.parse((String) values.get("sendAfter"))); + } + if(values.get("variables") instanceof Map variables) + { + variables.forEach((key, value) -> request.withVariable((String) key, (Serializable) value)); + } + else if(values.get("variables") != null) + { + throw (new IllegalArgumentException("variables must be a map")); + } + return (request); + } + + + + /******************************************************************************* + ** Accept one mailbox or a list of mailboxes, without comma splitting. + *******************************************************************************/ + private static List addresses(Object value) + { + if(value == null) + { + return (List.of()); + } + if(value instanceof String address) + { + return (List.of(address)); + } + return (((List) value).stream().map(String.class::cast).toList()); + } +} diff --git a/src/test/java/com/kingsrook/qbits/email/outbox/EmailJobEffectTargetTest.java b/src/test/java/com/kingsrook/qbits/email/outbox/EmailJobEffectTargetTest.java new file mode 100644 index 0000000..ef9abe7 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/email/outbox/EmailJobEffectTargetTest.java @@ -0,0 +1,494 @@ +/* + * 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.outbox; + + +import java.io.Serializable; +import java.sql.Connection; +import java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import com.kingsrook.qbits.email.EmailQBitConfig; +import com.kingsrook.qbits.email.EmailQBitProducer; +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.actions.customizers.TableCustomizerInterface; +import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException; +import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaData; +import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaDataProvider; +import com.kingsrook.qqq.backend.core.jobs.DurableJobsTables; +import com.kingsrook.qqq.backend.core.jobs.effects.EffectRequest; +import com.kingsrook.qqq.backend.core.jobs.effects.EffectResult; +import com.kingsrook.qqq.backend.core.jobs.effects.JobEffectType; +import com.kingsrook.qqq.backend.core.jobs.effects.ReconcileResult; +import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput; +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.model.metadata.tables.QTableMetaData; +import com.kingsrook.qqq.backend.core.model.session.QSession; +import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession; +import com.kingsrook.qqq.backend.core.model.session.QUser; +import com.kingsrook.qqq.backend.core.modules.authentication.implementations.FullyAnonymousAuthenticationModule; +import com.kingsrook.qqq.backend.module.rdbms.jdbc.ConnectionManager; +import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + + +/******************************************************************************* + ** Real H2 outbox and jobs lookup; no transport or mocked owner lookup. + *******************************************************************************/ +class EmailJobEffectTargetTest +{ + private final EmailJobEffectTarget target = new EmailJobEffectTarget(); + private static QSession observedSession; + + + + /******************************************************************************* + ** Install a local outbox and provider without dispatching any messages. + *******************************************************************************/ + @BeforeEach + void setup() throws Exception + { + H2Support.setup(config -> + { + }); + H2Support.provider(25); + observedSession = null; + } + + + + /******************************************************************************* + ** Restore per-test context and hook state. + *******************************************************************************/ + @AfterEach + void cleanup() + { + QContext.clear(); + observedSession = null; + } + + + + /******************************************************************************* + ** A normal effect contains only serializable payload values. + *******************************************************************************/ + private EffectRequest request(String key, Integer jobId, Map payload) + { + return (new EffectRequest(key, "welcomeEmail", JobEffectType.IDEMPOTENT_EXTERNAL, jobId, 1, "row", payload, null, 1)); + } + + + + /******************************************************************************* + ** Valid ad-hoc email payload. + *******************************************************************************/ + private Map payload() + { + return (new LinkedHashMap<>(Map.of("to", "person@example.com", "subject", "Welcome", "textBody", "Hello"))); + } + + + + /******************************************************************************* + ** Replays acknowledge the same committed row and ignore payload identity keys. + *******************************************************************************/ + @Test + void deliversOnceAndReconciles() throws Exception + { + var payload = payload(); + payload.put("senderQbit", "spoofed"); + payload.put("senderProcess", "spoofed"); + payload.put("idempotencyKey", "spoofed"); + payload.put("transaction", "ignored"); + var request = request("same", null, payload); + assertThat(target.reconcile(request)).isEqualTo(ReconcileResult.notFound()); + var first = target.deliver(request); + assertThat(first.outcome()).isEqualTo(EffectResult.Outcome.SENT); + assertThat(target.deliver(request)).isEqualTo(first); + var rows = H2Support.all(EmailTables.MESSAGE); + assertThat(rows).hasSize(1); + var row = rows.getFirst(); + assertThat(first.providerRef()).isEqualTo(row.getValueString("id")); + assertThat(row.getValueString("senderQbit")).isEqualTo("jobs"); + assertThat(row.getValueString("senderProcess")).isEqualTo("welcomeEmail"); + assertThat(row.getValueString("idempotencyKey")).isEqualTo("same"); + assertThat(target.reconcile(request)).isEqualTo(ReconcileResult.found(first.providerRef())); + SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", row.getValueInteger("id")).withValue("status", "DEAD_LETTER")); + assertThat(target.reconcile(request)).isEqualTo(ReconcileResult.found(first.providerRef())); + } + + + + /******************************************************************************* + ** Suppression is a permanent refusal on every attempt, but reconciles by row. + *******************************************************************************/ + @Test + void suppressedRecipientFails() throws Exception + { + H2Support.insert(EmailTables.SUPPRESSION, new QRecord().withValue("address", "person@example.com") + .withValue("scope", "ALL").withValue("reason", "HARD_BOUNCE")); + var request = request("suppressed", null, payload()); + assertThat(target.deliver(request)).isEqualTo(EffectResult.failed("All recipients are suppressed")); + assertThat(target.deliver(request).outcome()).isEqualTo(EffectResult.Outcome.FAILED); + assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1); + assertThat(target.reconcile(request).outcome()).isEqualTo(ReconcileResult.Outcome.FOUND); + } + + + + /******************************************************************************* + ** Service validation reasons reach the effect without bypassing template rules. + *******************************************************************************/ + @Test + void validationRefusalRetainsReasons() throws Exception + { + var payload = payload(); + payload.put("variables", new LinkedHashMap<>(Map.of("name", "Person"))); + assertThat(target.deliver(request("variables", null, payload))) + .isEqualTo(EffectResult.failed("Variables require a template with a schema")); + assertThat(target.deliver(request("empty", null, null)).outcome()).isEqualTo(EffectResult.Outcome.FAILED); + assertThat(target.deliver(request(" ", null, payload())).message()).contains("valid, matching effect key"); + assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty(); + } + + + + /******************************************************************************* + ** A database failure remains uncertain and does not disclose raw diagnostics. + *******************************************************************************/ + @Test + void storageFailureIsUnknown() throws Exception + { + var request = request("collision", null, payload()); + var sent = target.deliver(request); + SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", Integer.valueOf(sent.providerRef())).withValue("idempotencyKey", "changed")); + var result = target.deliver(request); + assertThat(result.outcome()).isEqualTo(EffectResult.Outcome.UNKNOWN); + assertThat(result.message()).doesNotContain("person@example.com", "INSERT", "Hello"); + assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1); + } + + + + /******************************************************************************* + ** Malformed supported values follow the specified safe same-key retry policy. + *******************************************************************************/ + @Test + void decodingExceptionsAreUnknown() throws Exception + { + for(var invalid : List.of(Map.entry("to", 4), Map.entry("category", "invalid"), Map.entry("variables", "invalid"))) + { + var payload = payload(); + payload.put(invalid.getKey(), (Serializable) invalid.getValue()); + assertThat(target.deliver(request("bad", null, payload)).outcome()).isEqualTo(EffectResult.Outcome.UNKNOWN); + } + assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty(); + } + + + + /******************************************************************************* + ** Every documented optional value reaches the request without losing types. + *******************************************************************************/ + @Test + void decodesSupportedPayload() + { + var payload = payload(); + payload.put("to", new ArrayList<>(List.of("one@example.com", "two@example.com"))); + payload.put("cc", "cc@example.com"); + payload.put("bcc", new ArrayList<>(List.of("bcc@example.com"))); + for(String key : List.of("from", "replyTo", "templateName", "markdownBody", "htmlBody", "providerName", "brandName", "linkedTableName")) + { + payload.put(key, key); + } + payload.put("fromAddressId", 7); + payload.put("locale", "fr"); + payload.put("category", "SECURITY"); + payload.put("sendAfter", "2026-10-14T12:00:00Z"); + payload.put("linkedRecordId", 42); + payload.put("variables", new LinkedHashMap<>(Map.of("count", 3, "active", true))); + var decoded = EmailJobEffectTarget.decode(payload); + assertThat(decoded.getTo()).containsExactly("one@example.com", "two@example.com"); + assertThat(decoded.getCc()).containsExactly("cc@example.com"); + assertThat(decoded.getBcc()).containsExactly("bcc@example.com"); + assertThat(decoded.getFrom()).isEqualTo("from"); + assertThat(decoded.getFromAddressId()).isEqualTo(7); + assertThat(decoded.getReplyTo()).isEqualTo("replyTo"); + assertThat(decoded.getSubject()).isEqualTo("Welcome"); + assertThat(decoded.getTemplateName()).isEqualTo("templateName"); + assertThat(decoded.getMarkdownBody()).isEqualTo("markdownBody"); + assertThat(decoded.getHtmlBody()).isEqualTo("htmlBody"); + assertThat(decoded.getTextBody()).isEqualTo("Hello"); + assertThat(decoded.getProviderName()).isEqualTo("providerName"); + assertThat(decoded.getBrandName()).isEqualTo("brandName"); + assertThat(decoded.getLocale()).isEqualTo("fr"); + assertThat(decoded.getCategory()).isEqualTo(EmailCategory.SECURITY); + assertThat(decoded.getSendAfter()).isEqualTo(Instant.parse("2026-10-14T12:00:00Z")); + assertThat(decoded.getLinkedTableName()).isEqualTo("linkedTableName"); + assertThat(decoded.getLinkedRecordId()).isEqualTo("42"); + assertThat(decoded.getVariables()).containsEntry("count", 3).containsEntry("active", true); + assertThat(EmailJobEffectTarget.decode(Map.of()).getLocale()).isEqualTo("en"); + } + + + + /******************************************************************************* + ** Use core's jobs metadata and a real H2 jobs_job table with all its columns. + ** Only the lookup table needs physical storage; no jobs worker runs here. + *******************************************************************************/ + private void enableJobs() throws Exception + { + QInstance instance = QContext.getQInstance(); + instance.withSupplementalMetaData(new DurableJobsMetaData()); + new DurableJobsMetaDataProvider().defineAll(instance, "rdbms", null); + instance.getAuthentication().withModuleCodeReference(new QCodeReference(OwnerAuthentication.class)); + QTableMetaData table = instance.getTable(DurableJobsTables.TABLE_JOB); + List columns = new ArrayList<>(); + for(var field : table.getFields().values()) + { + String sqlType = switch(field.getType()) + { + case INTEGER -> "INTEGER"; + case BOOLEAN -> "BOOLEAN"; + case DATE_TIME -> "TIMESTAMP"; + default -> "VARCHAR(10000)"; + }; + columns.add("\"" + field.getName() + "\" " + sqlType); + } + try(Connection connection = new ConnectionManager().getConnection((RDBMSBackendMetaData) instance.getBackend("rdbms"))) + { + connection.createStatement().execute("CREATE TABLE jobs_job (" + String.join(", ", columns) + ")"); + } + instance.getTable(EmailTables.MESSAGE).withCustomizer(TableCustomizers.PRE_INSERT_RECORD, new QCodeReference(ObserveOwner.class)); + QContext.init(instance, new QSystemUserSession()); + } + + + + /******************************************************************************* + ** Seed just the job identity and captured scope through JDBC, like a stored job. + *******************************************************************************/ + private void job(String userId) throws Exception + { + try(Connection connection = new ConnectionManager().getConnection((RDBMSBackendMetaData) QContext.getQInstance().getBackend("rdbms")); + var statement = connection.prepareStatement("INSERT INTO jobs_job (\"id\", \"userId\", \"credentialScopeJson\", \"clientChannel\") VALUES (1, ?, ?, ?)")) + { + statement.setString(1, userId); + statement.setString(2, "{\"tenant\":[\"allowed\"]}"); + statement.setString(3, "assistant"); + statement.executeUpdate(); + } + } + + + + /******************************************************************************* + ** The actual enqueue observes the owner and narrowed scope, then restores system. + *******************************************************************************/ + @Test + void sendsAsJobOwner() throws Exception + { + enableJobs(); + job("owner"); + QSession dispatcher = QContext.getQSession(); + assertThat(target.deliver(request("owner", 1, payload())).outcome()).isEqualTo(EffectResult.Outcome.SENT); + assertThat(observedSession.getUser().getIdReference()).isEqualTo("owner"); + assertThat(observedSession.getClientChannel()).isEqualTo("assistant"); + assertThat(observedSession.getSecurityKeyValues().get("tenant")).containsExactly("allowed"); + assertThat(QContext.getQSession()).isSameAs(dispatcher); + assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1); + } + + + + /******************************************************************************* + ** A missing owner fails permanently instead of sending as the system user. + *******************************************************************************/ + @Test + void missingOwnerFails() throws Exception + { + enableJobs(); + job(null); + assertOwnerFailure(); + } + + + + /******************************************************************************* + ** Authentication refusal fails permanently. + *******************************************************************************/ + @Test + void authenticationFailureFails() throws Exception + { + enableJobs(); + job("denied"); + assertOwnerFailure(); + } + + + + /******************************************************************************* + ** Core rejects a session for someone other than the persisted job owner. + *******************************************************************************/ + @Test + void wrongOwnerFails() throws Exception + { + enableJobs(); + job("wrong"); + assertOwnerFailure(); + } + + + + /******************************************************************************* + ** A deleted job cannot fall back to the dispatcher identity. + *******************************************************************************/ + @Test + void missingJobFails() throws Exception + { + enableJobs(); + assertOwnerFailure(); + } + + + + /******************************************************************************* + ** Refusal happens before the outbox sees the request and preserves the context. + *******************************************************************************/ + private void assertOwnerFailure() throws Exception + { + QSession dispatcher = QContext.getQSession(); + var result = target.deliver(request("denied", 1, payload())); + assertThat(result.outcome()).isEqualTo(EffectResult.Outcome.FAILED); + assertThat(result.message()).startsWith("The job's owner can no longer send it:"); + assertThat(QContext.getQSession()).isSameAs(dispatcher); + assertThat(observedSession).isNull(); + assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty(); + } + + + + /******************************************************************************* + ** A failed send also restores the system session after using the owner context. + *******************************************************************************/ + @Test + void ownerContextRestoredAfterException() throws Exception + { + enableJobs(); + job("owner"); + var request = request("collision", 1, payload()); + var sent = target.deliver(request); + SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", Integer.valueOf(sent.providerRef())).withValue("idempotencyKey", "changed")); + QSession dispatcher = QContext.getQSession(); + assertThat(target.deliver(request).outcome()).isEqualTo(EffectResult.Outcome.UNKNOWN); + assertThat(QContext.getQSession()).isSameAs(dispatcher); + assertThat(observedSession.getUser().getIdReference()).isEqualTo("owner"); + } + + + + /******************************************************************************* + ** A failed lookup must never be interpreted as proof of no previous handoff. + *******************************************************************************/ + @Test + void reconcileStorageFailurePropagates() throws Exception + { + try(Connection connection = new ConnectionManager().getConnection((RDBMSBackendMetaData) QContext.getQInstance().getBackend("rdbms"))) + { + connection.createStatement().execute("ALTER TABLE email_message RENAME TO unavailable_message"); + } + assertThatThrownBy(() -> target.reconcile(request("missing", null, payload()))).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QException.class); + } + + + + /******************************************************************************* + ** The helper is optional, and producing email after jobs registers automatically. + *******************************************************************************/ + @Test + void registersOnlyWhenEnabled() throws Exception + { + QInstance disabled = QContext.getQInstance(); + EmailJobEffectTarget.register(disabled); + assertThat(DurableJobsMetaData.of(disabled)).isNull(); + QInstance enabled = new QInstance().withSupplementalMetaData(new DurableJobsMetaData()); + enabled.addBackend(H2Support.backend()); + new DurableJobsMetaDataProvider().defineAll(enabled, "rdbms", null); + new EmailQBitProducer().withQBitConfig(new EmailQBitConfig().withBackendName("rdbms")).produce(enabled); + assertThat(DurableJobsMetaData.of(enabled).getEffectTargets()).containsOnlyKeys("email"); + assertThat(DurableJobsMetaData.of(enabled).getEffectTargets().get("email").getName()).isEqualTo(EmailJobEffectTarget.class.getName()); + EmailJobEffectTarget.register(enabled); + assertThat(DurableJobsMetaData.of(enabled).getEffectTargets()).hasSize(1); + } + + + + /******************************************************************************* + ** Test authentication supplies current permissions and a wider live scope. + *******************************************************************************/ + public static class OwnerAuthentication extends FullyAnonymousAuthenticationModule + { + /******************************************************************************* + ** Simulate an active owner, revoked authentication, or mismatched identity. + *******************************************************************************/ + @Override + public QSession createAutomatedSessionForUser(QInstance instance, Serializable userId) throws QAuthenticationException + { + if("denied".equals(userId)) + { + throw (new QAuthenticationException("Access revoked")); + } + return (H2Support.admin().withUser(new QUser().withIdReference("wrong".equals(userId) ? "someone-else" : String.valueOf(userId))) + .withSecurityKeyValues(Map.of("tenant", List.of("allowed", "new")))); + } + } + + + + /******************************************************************************* + ** Observe the session inside the real send without adding production test seams. + *******************************************************************************/ + public static class ObserveOwner implements TableCustomizerInterface + { + /******************************************************************************* + ** Email messages have no built-in user field, so observe at the insert hook. + *******************************************************************************/ + @Override + public List preInsertOrUpdate(AbstractActionInput input, List records, boolean preview, Optional> oldRecords) + { + observedSession = QContext.getQSession(); + return (records); + } + } +}