feat: jpa, messaging, notification, mongo, graphql 어댑터터 리펙토링

2026-08-18 10:59:56 +09:00
parent 2f5d2fc219
commit e98b56eb03
372 changed files with 25131 additions and 20357 deletions
@@ -16,7 +16,6 @@ import dev.caskeleton.application.notification.platform.security.SecretMaterialP
 import dev.caskeleton.application.notification.platform.template.TemplateVariableValidator;
 import java.time.Duration;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -83,9 +82,23 @@ public class NotificationPlatformAutoConfiguration {
    };
  }

-  /** Contact point protection, only once key material is available. */
+  /**
+   * Contact point protection.
+   *
+   * <p>{@code @ConditionalOnBean(SecretMaterialProvider.class)} used to guard this, to express
+   * "only once key material is available". It does not express that. This is a plain
+   * {@code @Configuration} imported by a root, so the condition is evaluated while configurations
+   * are still being parsed and answers according to what happens to be registered at that moment —
+   * and the secret provider is declared by a sibling configuration the same root imports. The
+   * answer was "absent", so the bean vanished and the application failed on an unsatisfied
+   * dependency several layers away, in the deployment that first assembled the platform for real.
+   *
+   * <p>Nothing is lost by dropping it. Both configurations are reached only through the same root
+   * under the same master switch, so the provider exists whenever this does — and a deployment
+   * missing key material already fails at the provider itself, at startup, naming the purpose whose
+   * key is missing. That is a better failure than a bean quietly not being created.
+   */
  @Bean
-  @ConditionalOnBean(SecretMaterialProvider.class)
  @ConditionalOnMissingBean
  public ContactPointProtector notificationContactPointProtector(SecretMaterialProvider secrets) {
    return new AesGcmContactPointProtector(secrets);
@@ -0,0 +1,211 @@
+package dev.caskeleton.adapter.outbound.notification.platform.autoconfigure;
+
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.JavaMailSenderSmtpDispatch;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpDispatch;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpFailureClassifier;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpMimeMessageFactory;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpProviderProperties;
+import dev.caskeleton.application.notification.platform.dispatch.AttachmentIntegrityGuard;
+import dev.caskeleton.application.notification.platform.provider.AttachmentResolver;
+import dev.caskeleton.application.notification.platform.security.ContactPointProtector;
+import jakarta.mail.Session;
+import java.time.Clock;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.mail.javamail.JavaMailSender;
+import org.springframework.mail.javamail.JavaMailSenderImpl;
+
+/**
+ * The SMTP provider family, assembled (NTF-INT-001).
+ *
+ * <p>{@code NotificationPlatformProviderConfig} injects {@code List<ProviderRuntimeAssembler>} and
+ * nothing in the composition contributed one, so the list was always empty. Assembly then produced
+ * a registry with no runtimes and {@code SERVING} accepted requests that could never find a route —
+ * which from outside is a platform silently dropping notifications. {@code
+ * SmtpProviderRuntimeAssembler} existed in main source throughout; the only thing that ever
+ * constructed it was its own unit test.
+ *
+ * <p>It lives in this leaf rather than in the composition root because the root must not carry
+ * {@code jakarta.mail} or {@code JavaMailSender} on its own compile classpath: the mail starter is
+ * an implementation detail of the family this package assembles, and a composition root that
+ * imports it to wire one provider has taken ownership of that provider's transport.
+ *
+ * <p>Gated on {@code spring.mail.host}, not on the notification switch alone. A deployment that
+ * turns the platform on without configuring a relay contributes no SMTP assembler, and assembly
+ * then refuses any profile that declares {@code type: SMTP} by naming the missing family — a better
+ * failure than a mail sender pointed at a default nobody chose.
+ *
+ * <p>The relay's address is Spring's, not a second copy. Host, port, credentials and transport
+ * security come from {@code spring.mail.*} through the auto-configured {@link JavaMailSender}. One
+ * relay described in two places is a defect this repository has already paid for: {@code
+ * app.jpa-platform.datasource.*} validated a pool that {@code spring.datasource.hikari.*} had
+ * built. {@link NotificationSmtpSettings} carries only what {@code spring.mail} has no word for —
+ * the envelope sender, the per-attempt timeouts and the dispatch concurrency.
+ */
+@Configuration(proxyBeanMethods = false)
+// The master switch is not repeated here. This configuration is reachable only through
+// NotificationPlatformRuntimeConfig, which the root imports under that switch, so "off" stays a
+// structural fact rather than a condition each new configuration has to remember. What this one
+// owns
+// is the narrower question: is a relay configured at all.
+@ConditionalOnProperty(name = "spring.mail.host")
+@EnableConfigurationProperties(NotificationSmtpSettings.class)
+public class NotificationSmtpProviderConfig {
+
+  /** The send itself, over the mail sender Spring configured. */
+  @Bean
+  @ConditionalOnMissingBean
+  public SmtpDispatch notificationSmtpDispatch(JavaMailSender sender) {
+    return new JavaMailSenderSmtpDispatch(sender);
+  }
+
+  /**
+   * The MIME factory, on the sender's own session.
+   *
+   * <p>The session is taken from {@link JavaMailSenderImpl} rather than created here, so the
+   * message is built with the same properties it will be sent with. A second session would let the
+   * factory and the transport disagree about encoding or TLS, and the disagreement would only show
+   * as a malformed message at a recipient.
+   */
+  @Bean
+  @ConditionalOnMissingBean
+  public SmtpMimeMessageFactory notificationSmtpMimeMessageFactory(JavaMailSender sender) {
+    if (sender instanceof JavaMailSenderImpl impl) {
+      return new SmtpMimeMessageFactory(impl.getSession());
+    }
+    return new SmtpMimeMessageFactory(Session.getInstance(new java.util.Properties()));
+  }
+
+  /** SMTP reply code to retry decision. */
+  @Bean
+  @ConditionalOnMissingBean
+  public SmtpFailureClassifier notificationSmtpFailureClassifier() {
+    return new SmtpFailureClassifier();
+  }
+
+  /**
+   * Attachment resolution, refused by default.
+   *
+   * <p>Attachments come from the file server or object storage capability, and a deployment that
+   * enabled neither has no way to read one. The default refuses rather than returning null: a null
+   * resolution reaches the integrity guard, which then compares a digest against nothing, and the
+   * message goes out without the attachment it claimed to carry.
+   */
+  @Bean
+  @ConditionalOnMissingBean
+  public AttachmentResolver notificationAttachmentResolver() {
+    return (reference, context) -> {
+      throw new IllegalStateException(
+          "an attachment was requested but no attachment capability is bound; enable the file "
+              + "server or object storage capability, or submit a notification without attachments");
+    };
+  }
+
+  /** Digest check between what a template referenced and what was read. */
+  @Bean
+  @ConditionalOnMissingBean
+  public AttachmentIntegrityGuard notificationAttachmentIntegrityGuard(
+      AttachmentResolver resolver) {
+    return new AttachmentIntegrityGuard(resolver);
+  }
+
+  /**
+   * Where an SMTP send runs.
+   *
+   * <p>Bounded, and it aborts rather than queues without limit: an SMTP relay that has stopped
+   * answering turns an unbounded queue into retained heap and a dispatch worker that never notices.
+   * {@code CallerRunsPolicy} would instead block the dispatch loop on the slow relay, which stalls
+   * every other channel too.
+   */
+  @Bean(name = "notificationSmtpExecutor", destroyMethod = "shutdown")
+  @ConditionalOnMissingBean(name = "notificationSmtpExecutor")
+  public Executor notificationSmtpExecutor(NotificationSmtpSettings settings) {
+    ThreadPoolExecutor executor =
+        new ThreadPoolExecutor(
+            1,
+            settings.maxConcurrency(),
+            60L,
+            TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(settings.maxConcurrency() * 4),
+            runnable -> {
+              Thread thread = new Thread(runnable, "notification-smtp");
+              thread.setDaemon(true);
+              return thread;
+            },
+            new ThreadPoolExecutor.AbortPolicy());
+    executor.allowCoreThreadTimeOut(true);
+    return executor;
+  }
+
+  /**
+   * The assembler itself.
+   *
+   * @param dispatch the send
+   * @param mimeFactory the MIME builder
+   * @param classifier the outcome classifier
+   * @param protector redacts contact points on the way out
+   * @param notificationSmtpExecutor where a send runs
+   * @param attachmentGuard the attachment integrity check
+   * @param settings the parts of the relay description Spring's own mail properties do not carry
+   * @param sender the auto-configured sender, read for host and port
+   * @param clock the clock the rate limiter measures against
+   * @return the SMTP assembler, contributed into the assembly's list
+   */
+  @Bean
+  @ConditionalOnMissingBean(name = "smtpProviderRuntimeAssembler")
+  public ProviderRuntimeAssembler smtpProviderRuntimeAssembler(
+      SmtpDispatch dispatch,
+      SmtpMimeMessageFactory mimeFactory,
+      SmtpFailureClassifier classifier,
+      ContactPointProtector protector,
+      Executor notificationSmtpExecutor,
+      AttachmentIntegrityGuard attachmentGuard,
+      NotificationSmtpSettings settings,
+      JavaMailSender sender,
+      Clock clock) {
+    return new SmtpProviderRuntimeAssembler(
+        dispatch,
+        mimeFactory,
+        classifier,
+        protector,
+        notificationSmtpExecutor,
+        attachmentGuard,
+        transport(settings, sender),
+        clock);
+  }
+
+  /**
+   * The relay's address, read from Spring's mail configuration.
+   *
+   * <p>{@code SmtpProviderProperties.TlsMode} has two members and neither is plaintext, so the type
+   * refuses an unencrypted relay by construction rather than by a validator somebody has to
+   * remember to run. That is why the mode is a setting rather than something derived from {@code
+   * spring.mail.properties.mail.smtp.starttls.enable}: a derived mode would silently become
+   * "whatever that flag happened to say", including off.
+   */
+  private static SmtpProviderProperties transport(
+      NotificationSmtpSettings settings, JavaMailSender sender) {
+    String host = settings.host();
+    int port = settings.port();
+    if (sender instanceof JavaMailSenderImpl impl) {
+      host = impl.getHost() == null ? host : impl.getHost();
+      port = impl.getPort() > 0 ? impl.getPort() : port;
+    }
+    return new SmtpProviderProperties(
+        host,
+        port,
+        settings.tlsMode(),
+        settings.senderIdentity(),
+        settings.connectTimeout(),
+        settings.readTimeout(),
+        settings.writeTimeout(),
+        settings.maxConcurrency());
+  }
+}
@@ -0,0 +1,60 @@
+package dev.caskeleton.adapter.outbound.notification.platform.autoconfigure;
+
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpProviderProperties;
+import java.time.Duration;
+import java.util.Objects;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.bind.DefaultValue;
+
+/**
+ * The parts of an SMTP relay description that {@code spring.mail.*} has no word for.
+ *
+ * <p>Host, port, username, password and the mail properties stay Spring's. Repeating them here
+ * would be one relay described twice, which is the shape of a defect this repository has already
+ * paid for — {@code app.jpa-platform.datasource.*} validated a pool that {@code
+ * spring.datasource.hikari.*} had built, so the validation passed while describing something that
+ * was not running.
+ *
+ * <p>{@code host} and {@code port} are present all the same, as a fallback only: {@link
+ * NotificationSmtpProviderConfig} reads them from the configured sender and falls back here when
+ * the sender is not the standard implementation. They are never the primary source.
+ *
+ * @param host fallback relay host, used only when the mail sender cannot be read
+ * @param port fallback relay port, same
+ * @param tlsMode transport security; the type has no plaintext member, deliberately
+ * @param senderIdentity the envelope sender every message is sent as
+ * @param connectTimeout how long a connection attempt may take
+ * @param readTimeout how long a reply may take
+ * @param writeTimeout how long a write may take
+ * @param maxConcurrency how many sends may run at once
+ */
+@ConfigurationProperties("ca-skeleton.notification.platform.smtp")
+public record NotificationSmtpSettings(
+    @DefaultValue("localhost") String host,
+    @DefaultValue("587") int port,
+    @DefaultValue("STARTTLS_REQUIRED") SmtpProviderProperties.TlsMode tlsMode,
+    @DefaultValue("no-reply@example.invalid") String senderIdentity,
+    @DefaultValue("5s") Duration connectTimeout,
+    @DefaultValue("10s") Duration readTimeout,
+    @DefaultValue("10s") Duration writeTimeout,
+    @DefaultValue("4") int maxConcurrency) {
+
+  public NotificationSmtpSettings {
+    Objects.requireNonNull(tlsMode, "tlsMode");
+    Objects.requireNonNull(senderIdentity, "senderIdentity");
+    Objects.requireNonNull(connectTimeout, "connectTimeout");
+    Objects.requireNonNull(readTimeout, "readTimeout");
+    Objects.requireNonNull(writeTimeout, "writeTimeout");
+    // The sender identity has a default that cannot deliver on purpose. example.invalid is reserved
+    // by RFC 2606 and resolves nowhere, so a deployment that forgot to set one gets a bounce it can
+    // trace rather than mail that appears to come from a real address it does not own.
+    if (senderIdentity.isBlank() || senderIdentity.indexOf('@') <= 0) {
+      throw new IllegalArgumentException(
+          "ca-skeleton.notification.platform.smtp.sender-identity must be an email address");
+    }
+    if (maxConcurrency < 1) {
+      throw new IllegalArgumentException(
+          "ca-skeleton.notification.platform.smtp.max-concurrency must be at least 1");
+    }
+  }
+}
@@ -0,0 +1,164 @@
+package dev.caskeleton.adapter.outbound.notification.platform.autoconfigure;
+
+import dev.caskeleton.adapter.outbound.notification.platform.dispatch.ProviderAttemptLimiter;
+import dev.caskeleton.adapter.outbound.notification.platform.dispatch.ProviderRuntime;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpDispatch;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpFailureClassifier;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpMimeMessageFactory;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpNotificationProviderAdapter;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpProviderProperties;
+import dev.caskeleton.application.notification.platform.api.ProviderId;
+import dev.caskeleton.application.notification.platform.api.ProviderProfileId;
+import dev.caskeleton.application.notification.platform.api.routing.Channel;
+import dev.caskeleton.application.notification.platform.dispatch.AttachmentIntegrityGuard;
+import dev.caskeleton.application.notification.platform.provider.ProviderCapabilities;
+import dev.caskeleton.application.notification.platform.provider.ProviderProfileSnapshot;
+import dev.caskeleton.application.notification.platform.security.ContactPointProtector;
+import java.time.Clock;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.Executor;
+
+/**
+ * The first production assembler: one configured SMTP profile becomes one working provider
+ * (NTF-INT-001).
+ *
+ * <p>{@code NotificationPlatformProviderConfig} assembles {@code List<ProviderRuntimeAssembler>}
+ * and production main source contained no implementation of that interface. So {@code SERVING}
+ * could not work in production whatever an operator configured: profiles bound, validation passed,
+ * and the runtime registry was built empty — requests reached durable acceptance and then found no
+ * eligible route, which reads from outside as the platform dropping notifications.
+ *
+ * <p>SMTP is the reference family because every piece above the wire already existed and was
+ * tested: the adapter, the MIME factory, the failure classifier. What was missing was the send
+ * itself ({@code SmtpDispatch} had no implementation) and this — the step that turns a profile into
+ * a runtime.
+ *
+ * <p><b>Where the relay's address comes from.</b> Host, port, credentials and TLS come from
+ * Spring's own {@code spring.mail.*} through the injected {@code JavaMailSender}, not from a second
+ * description on the provider profile. One relay described twice is the defect this repository has
+ * already paid for elsewhere — {@code app.jpa-platform.datasource.*} validated a pool that {@code
+ * spring.datasource.hikari.*} had built. The profile owns what is per-profile: its timeout, its
+ * concurrency and its rate.
+ *
+ * <p>Contributed as {@link AssembledProvider#dispatchOnly}, honestly: SMTP has no callback adapter,
+ * no provider-event projector and no status-query capability. Claiming any of them would produce a
+ * profile that fails on the first provider event rather than at startup.
+ */
+public final class SmtpProviderRuntimeAssembler implements ProviderRuntimeAssembler {
+
+  private final SmtpDispatch dispatch;
+  private final SmtpMimeMessageFactory mimeFactory;
+  private final SmtpFailureClassifier classifier;
+  private final ContactPointProtector protector;
+  private final Executor executor;
+  private final AttachmentIntegrityGuard attachmentGuard;
+  private final SmtpProviderProperties transport;
+  private final Clock clock;
+
+  /**
+   * Creates the assembler.
+   *
+   * @param dispatch the send, over the configured mail sender
+   * @param mimeFactory builds the MIME message
+   * @param classifier turns an SMTP outcome into a retry decision
+   * @param protector redacts contact points on the way out
+   * @param executor where a send runs
+   * @param attachmentGuard the attachment integrity check
+   * @param transport the relay's address and transport security, from {@code spring.mail.*}
+   * @param clock the clock the rate limiter measures against
+   */
+  public SmtpProviderRuntimeAssembler(
+      SmtpDispatch dispatch,
+      SmtpMimeMessageFactory mimeFactory,
+      SmtpFailureClassifier classifier,
+      ContactPointProtector protector,
+      Executor executor,
+      AttachmentIntegrityGuard attachmentGuard,
+      SmtpProviderProperties transport,
+      Clock clock) {
+    this.dispatch = Objects.requireNonNull(dispatch, "dispatch");
+    this.mimeFactory = Objects.requireNonNull(mimeFactory, "mimeFactory");
+    this.classifier = Objects.requireNonNull(classifier, "classifier");
+    this.protector = Objects.requireNonNull(protector, "protector");
+    this.executor = Objects.requireNonNull(executor, "executor");
+    this.attachmentGuard = Objects.requireNonNull(attachmentGuard, "attachmentGuard");
+    this.transport = Objects.requireNonNull(transport, "transport");
+    this.clock = Objects.requireNonNull(clock, "clock");
+  }
+
+  @Override
+  public ProviderType type() {
+    return ProviderType.SMTP;
+  }
+
+  @Override
+  public AssembledProvider assemble(
+      String profileId, NotificationPlatformSettings.Provider profile) {
+    Objects.requireNonNull(profileId, "profileId");
+    Objects.requireNonNull(profile, "profile");
+
+    SmtpProviderProperties properties =
+        new SmtpProviderProperties(
+            transport.host(),
+            transport.port(),
+            transport.tlsMode(),
+            transport.senderIdentity(),
+            transport.connectTimeout(),
+            profile.timeout(),
+            profile.timeout(),
+            profile.maxConcurrency());
+
+    SmtpNotificationProviderAdapter adapter =
+        new SmtpNotificationProviderAdapter(
+            dispatch, mimeFactory, classifier, protector, properties, executor, attachmentGuard);
+
+    ProviderRuntime runtime =
+        new ProviderRuntime(
+            snapshot(profileId, profile),
+            adapter,
+            new ProviderAttemptLimiter(profile.maxConcurrency(), profile.ratePerSecond(), clock));
+
+    return AssembledProvider.dispatchOnly(runtime, Channel.EMAIL);
+  }
+
+  private static ProviderProfileSnapshot snapshot(
+      String profileId, NotificationPlatformSettings.Provider profile) {
+    return new ProviderProfileSnapshot(
+        new ProviderProfileId(profileId),
+        new ProviderId(ProviderType.SMTP.name().toLowerCase(java.util.Locale.ROOT)),
+        Channel.EMAIL,
+        profile.environment(),
+        // No rotation has happened yet for a freshly assembled profile. The generation is what a
+        // rotation increments; starting anywhere but the first would make the first rotation look
+        // like it had already occurred.
+        1L,
+        capabilities(),
+        Map.of());
+  }
+
+  /**
+   * What SMTP can actually do, stated rather than assumed.
+   *
+   * <p>No batch, no provider-side idempotency, no status callback, no status query, no delivery
+   * receipt, no native scheduling, cancel or collapse. SMTP is a protocol for handing a message to
+   * a relay; everything past that is the relay's business and invisible to the sender. A capability
+   * declared here that the protocol does not have is a promise the dispatch loop will act on.
+   */
+  private static ProviderCapabilities capabilities() {
+    return new ProviderCapabilities(
+        false,
+        false,
+        false,
+        false,
+        false,
+        false,
+        false,
+        false,
+        1,
+        25L * 1024 * 1024,
+        // A relay that has not accepted the message within this is not going to; the platform's own
+        // queue-age bound is what decides how long a request keeps being retried.
+        java.time.Duration.ofHours(24));
+  }
+}
@@ -96,10 +96,14 @@ public final class NotificationSchedulerWorker implements AutoCloseable {
              // The lease is left to expire rather than being released optimistically: a worker
              // that
              // failed mid-dispatch cannot prove what the provider did.
+              // The cause chain by type, never by message. "reason=NotificationValidationException"
+              // alone identified nothing — a dozen checks raise it — and a library's exception text
+              // can carry a recipient address, so the types are named and the messages are not.
              log.warn(
-                  "notification dispatch failed worker={} reason={}",
+                  "notification dispatch failed worker={} reason={} causes={}",
                  workerId,
-                  failure.getClass().getSimpleName());
+                  failure.getClass().getSimpleName(),
+                  causeChain(failure));
            } finally {
              globalConcurrency.release();
            }
@@ -108,6 +112,25 @@ public final class NotificationSchedulerWorker implements AutoCloseable {
    return claimed.size();
  }

+  /**
+   * The exception's cause chain, as type names only.
+   *
+   * @param failure the dispatch failure
+   * @return the chain, outermost first, bounded so a self-referential cause cannot loop
+   */
+  private static String causeChain(Throwable failure) {
+    StringBuilder chain = new StringBuilder();
+    Throwable current = failure.getCause();
+    for (int depth = 0; current != null && depth < 8; depth++) {
+      if (chain.length() > 0) {
+        chain.append('<');
+      }
+      chain.append(current.getClass().getSimpleName());
+      current = current.getCause();
+    }
+    return chain.length() == 0 ? "none" : chain.toString();
+  }
+
  /** Start the polling loop on a dedicated thread. */
  public void start() {
    if (!running.compareAndSet(false, true)) {
@@ -0,0 +1,54 @@
+package dev.caskeleton.adapter.outbound.notification.platform.provider.smtp;
+
+import jakarta.mail.internet.MimeMessage;
+import java.util.Objects;
+import java.util.Optional;
+import org.springframework.mail.MailException;
+import org.springframework.mail.javamail.JavaMailSender;
+
+/**
+ * The production SMTP send, over Spring's {@link JavaMailSender} (NTF-INT-001).
+ *
+ * <p>{@link SmtpDispatch} was an interface with no implementation. The adapter above it, its MIME
+ * factory and its failure classifier were all complete and unit-tested against fakes, so the SMTP
+ * family looked finished from every angle except the one that matters: nothing could send. That is
+ * the shape of the whole NTF-INT-001 finding — configuration bound, validation passed, and the
+ * runtime registry constructed empty.
+ *
+ * <p>Failures are translated rather than propagated. {@link SmtpDispatchException} is what {@code
+ * SmtpFailureClassifier} reads to decide retryable from permanent, and a raw {@link MailException}
+ * reaching the dispatch loop would be classified as an unknown failure — which this platform treats
+ * as ambiguous, and ambiguous means the attempt is not retried, because a message that may already
+ * have been delivered must not be sent twice.
+ */
+public final class JavaMailSenderSmtpDispatch implements SmtpDispatch {
+
+  private final JavaMailSender sender;
+
+  /**
+   * Creates the dispatch.
+   *
+   * @param sender the configured mail sender, whose host, port and credentials come from Spring's
+   *     own {@code spring.mail.*} — the namespace an operator already knows, rather than a second
+   *     description of one relay
+   */
+  public JavaMailSenderSmtpDispatch(JavaMailSender sender) {
+    this.sender = Objects.requireNonNull(sender, "sender");
+  }
+
+  @Override
+  public void send(MimeMessage message) {
+    Objects.requireNonNull(message, "message");
+    try {
+      sender.send(message);
+    } catch (MailException failure) {
+      // dataCommitted = true, deliberately and conservatively. Spring collapses every JavaMail
+      // outcome into MailException subtypes that do not say whether the body reached the server
+      // before the connection broke, and the platform reads this flag to decide whether a retry
+      // could duplicate a delivered message. Claiming "not committed" without evidence is the
+      // assumption that sends a notification twice; claiming "committed" costs a delivery that has
+      // to be reconciled, which is the failure this platform is built to survive.
+      throw new SmtpDispatchException("SMTP_TRANSPORT_FAILURE", Optional.empty(), true, failure);
+    }
+  }
+}
@@ -45,11 +45,16 @@ public final class SmtpMimeMessageFactory {
    Objects.requireNonNull(attachments, "attachments");

    if (!(submission.content().content() instanceof EmailContent email)) {
-      throw rejection();
+      // A static cause, not the value: which check fired is diagnostic, what it saw may be content.
+      throw rejection(
+          new IllegalStateException(
+              "rendered content for an SMTP submission is "
+                  + submission.content().content().getClass().getSimpleName()
+                  + ", not EmailContent"));
    }
-    requireHeaderSafe(recipientAddress);
-    requireHeaderSafe(fromAddress);
-    requireHeaderSafe(email.subject());
+    requireHeaderSafe(recipientAddress, "recipient");
+    requireHeaderSafe(fromAddress, "sender");
+    requireHeaderSafe(email.subject(), "subject");

    try {
      MimeMessage message = new MimeMessage(session);
@@ -71,25 +76,38 @@ public final class SmtpMimeMessageFactory {
            attachment.displayName(), () -> attachment.content(), attachment.contentType());
      }
      for (var header : email.options().approvedHeaders().entrySet()) {
-        requireHeaderSafe(header.getKey());
-        requireHeaderSafe(header.getValue());
+        requireHeaderSafe(header.getKey(), "approved header name");
+        requireHeaderSafe(header.getValue(), "approved header value");
        message.setHeader(header.getKey(), header.getValue());
      }
      return message;
    } catch (MessagingException failure) {
-      throw rejection();
+      throw rejection(failure);
    }
  }

-  private static void requireHeaderSafe(String value) {
+  private static void requireHeaderSafe(String value, String field) {
    if (value.indexOf('\r') >= 0 || value.indexOf('\n') >= 0 || value.indexOf('\0') >= 0) {
-      throw rejection();
+      // The field name, never the value: a header-injection attempt is exactly the payload that
+      // must
+      // not be echoed into a log.
+      throw rejection(new IllegalStateException(field + " contains a header separator"));
    }
  }

-  private static NotificationValidationException rejection() {
-    return new NotificationValidationException(
+  /**
+   * The rejection, carrying what caused it.
+   *
+   * <p>The descriptor is deliberately the same for every construction failure — it is what gets
+   * stored on the delivery row, and a per-check code there would be a cardinality problem. The
+   * cause is what tells an operator which check fired, and dropping it made an SMTP dispatch
+   * failure indistinguishable from any other: a lane saw "reason=NotificationValidationException"
+   * and nothing else.
+   */
+  private static NotificationValidationException rejection(Throwable cause) {
+    NotificationFailureDescriptor descriptor =
        NotificationFailureDescriptor.preDispatch(
-            NotificationFailureCode.VALIDATION_FAILED, FailureCategory.INVALID_PAYLOAD));
+            NotificationFailureCode.VALIDATION_FAILED, FailureCategory.INVALID_PAYLOAD);
+    return new NotificationValidationException(descriptor, cause);
  }
 }
@@ -0,0 +1,149 @@
+package dev.caskeleton.adapter.outbound.notification.platform.security;
+
+import dev.caskeleton.application.notification.platform.security.NotificationPayloadProtection;
+import dev.caskeleton.application.notification.platform.security.NotificationPayloadUnreadableException;
+import dev.caskeleton.application.notification.platform.security.SecretKeyMaterial;
+import dev.caskeleton.application.notification.platform.security.SecretMaterialProvider;
+import dev.caskeleton.application.notification.platform.security.SecretPurpose;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.security.GeneralSecurityException;
+import java.security.SecureRandom;
+import java.util.Objects;
+import javax.crypto.Cipher;
+import javax.crypto.spec.GCMParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+
+/**
+ * AES-GCM at rest, with the key id in the envelope (NTF-INT-007).
+ *
+ * <h2>The envelope</h2>
+ *
+ * <pre>
+ *   byte    version          always 1
+ *   byte    keyIdLength      1..255, UTF-8 bytes
+ *   byte[]  keyId
+ *   byte[12] nonce
+ *   byte[]  ciphertext + GCM tag
+ * </pre>
+ *
+ * <p><b>The key id is the whole point of having a format at all.</b> This repository's callback
+ * protection stores nonce and ciphertext and nothing else, so the day the active key changes, every
+ * row written under the previous one becomes unreadable and nothing in the row can say which key it
+ * needed. That is not a rotation story with a gap in it; it is the absence of one. Reading the id
+ * back out and asking the secret store for that specific key is what makes rotation a change of
+ * default rather than a data migration.
+ *
+ * <p>The version byte is here for the same reason and costs one byte: a format that cannot say
+ * which format it is can only ever be changed by rewriting every row first.
+ *
+ * <p>Authentication is not an add-on. GCM verifies the tag on decrypt, so a modified ciphertext
+ * fails rather than producing plausible-looking variables — which for a notification payload would
+ * mean rendering attacker-chosen content into a message a recipient trusts.
+ */
+public final class AesGcmNotificationPayloadProtection implements NotificationPayloadProtection {
+
+  /** The only version this class writes, and the only one it reads. */
+  static final byte VERSION = 1;
+
+  private static final int NONCE_BYTES = 12;
+  private static final int TAG_BITS = 128;
+  private static final int MAX_KEY_ID_BYTES = 255;
+
+  private final SecretMaterialProvider secrets;
+  private final SecureRandom random;
+
+  /**
+   * Creates the protection.
+   *
+   * @param secrets the key store, which owns the active key and every retired one
+   * @param random the nonce source
+   */
+  public AesGcmNotificationPayloadProtection(SecretMaterialProvider secrets, SecureRandom random) {
+    this.secrets = Objects.requireNonNull(secrets, "secrets");
+    this.random = Objects.requireNonNull(random, "random");
+  }
+
+  @Override
+  public byte[] protect(byte[] plaintext) {
+    Objects.requireNonNull(plaintext, "plaintext");
+    SecretKeyMaterial key = secrets.activeKey(SecretPurpose.PAYLOAD_ENCRYPTION);
+    byte[] keyId = key.keyId().getBytes(StandardCharsets.UTF_8);
+    if (keyId.length == 0 || keyId.length > MAX_KEY_ID_BYTES) {
+      throw new IllegalStateException(
+          "a payload encryption key id must be 1..255 UTF-8 bytes to fit the envelope; this one is "
+              + keyId.length);
+    }
+    byte[] nonce = new byte[NONCE_BYTES];
+    random.nextBytes(nonce);
+    try {
+      Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
+      cipher.init(
+          Cipher.ENCRYPT_MODE,
+          new SecretKeySpec(key.material(), "AES"),
+          new GCMParameterSpec(TAG_BITS, nonce));
+      // The header is authenticated, not merely prefixed: without this, the key id and version are
+      // attacker-editable, and an envelope could be redirected at a key of the attacker's choosing.
+      byte[] header = header(keyId);
+      cipher.updateAAD(header);
+      byte[] ciphertext = cipher.doFinal(plaintext);
+
+      return ByteBuffer.allocate(header.length + nonce.length + ciphertext.length)
+          .put(header)
+          .put(nonce)
+          .put(ciphertext)
+          .array();
+    } catch (GeneralSecurityException failure) {
+      // The message is deliberately shapeless: a failure here is about keys and providers, and
+      // anything derived from the plaintext would put caller content into a log line.
+      throw new IllegalStateException("notification payload encryption failed", failure);
+    }
+  }
+
+  @Override
+  public byte[] reveal(byte[] envelope) {
+    Objects.requireNonNull(envelope, "envelope");
+    if (envelope.length < 2) {
+      throw new NotificationPayloadUnreadableException(
+          "the stored payload is not an envelope", "unknown", null);
+    }
+    if (envelope[0] != VERSION) {
+      throw new NotificationPayloadUnreadableException(
+          "unsupported payload envelope version " + envelope[0], "unknown", null);
+    }
+    int keyIdLength = Byte.toUnsignedInt(envelope[1]);
+    int nonceStart = 2 + keyIdLength;
+    if (keyIdLength == 0 || envelope.length < nonceStart + NONCE_BYTES) {
+      throw new NotificationPayloadUnreadableException(
+          "the stored payload envelope is truncated", "unknown", null);
+    }
+    String keyId = new String(envelope, 2, keyIdLength, StandardCharsets.UTF_8);
+
+    try {
+      SecretKeyMaterial key = secrets.keyById(keyId);
+      Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
+      cipher.init(
+          Cipher.DECRYPT_MODE,
+          new SecretKeySpec(key.material(), "AES"),
+          new GCMParameterSpec(TAG_BITS, envelope, nonceStart, NONCE_BYTES));
+      cipher.updateAAD(envelope, 0, nonceStart);
+
+      int cipherStart = nonceStart + NONCE_BYTES;
+      return cipher.doFinal(envelope, cipherStart, envelope.length - cipherStart);
+    } catch (GeneralSecurityException | RuntimeException failure) {
+      // Every reason collapses into one type on purpose. An unknown key, a wrong key and a modified
+      // ciphertext are the same event to a caller — the payload cannot be read — and telling them
+      // apart in the message tells an attacker which of the three they achieved.
+      throw new NotificationPayloadUnreadableException(
+          "the stored notification payload could not be decrypted", keyId, failure);
+    }
+  }
+
+  private static byte[] header(byte[] keyId) {
+    return ByteBuffer.allocate(2 + keyId.length)
+        .put(VERSION)
+        .put((byte) keyId.length)
+        .put(keyId)
+        .array();
+  }
+}
@@ -28,7 +28,7 @@ public final class JacksonNotificationVariablesCodec implements NotificationVari
    try {
      return NotificationJsonMapper.mapper().writeValueAsString(new TreeMap<>(variables));
    } catch (JacksonException failure) {
-      throw rejection();
+      throw rejection(failure);
    }
  }

@@ -39,13 +39,14 @@ public final class JacksonNotificationVariablesCodec implements NotificationVari
      return NotificationJsonMapper.mapper()
          .readValue(payload, new TypeReference<TreeMap<String, Object>>() {});
    } catch (JacksonException failure) {
-      throw rejection();
+      throw rejection(failure);
    }
  }

-  private static NotificationValidationException rejection() {
+  private static NotificationValidationException rejection(Throwable cause) {
    return new NotificationValidationException(
        NotificationFailureDescriptor.preDispatch(
-            NotificationFailureCode.VALIDATION_FAILED, FailureCategory.INVALID_PAYLOAD));
+            NotificationFailureCode.VALIDATION_FAILED, FailureCategory.INVALID_PAYLOAD),
+        cause);
  }
 }
@@ -0,0 +1,133 @@
+package dev.caskeleton.adapter.outbound.notification.platform.autoconfigure;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpFailureClassifier;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpMimeMessageFactory;
+import dev.caskeleton.adapter.outbound.notification.platform.provider.smtp.SmtpProviderProperties;
+import dev.caskeleton.application.notification.platform.api.routing.Channel;
+import dev.caskeleton.application.notification.platform.dispatch.AttachmentIntegrityGuard;
+import jakarta.mail.Session;
+import java.time.Clock;
+import java.time.Duration;
+import java.util.Properties;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * One configured profile becomes one working provider (NTF-INT-001).
+ *
+ * <p>Before this assembler existed, {@code NotificationPlatformProviderConfig} collected {@code
+ * List<ProviderRuntimeAssembler>} and production main source implemented the interface nowhere. A
+ * fully configured SMTP profile therefore produced no runtime, no route and no error: requests
+ * reached durable acceptance and then found nothing eligible to send them, which reads from outside
+ * as the platform silently dropping notifications.
+ */
+class SmtpProviderRuntimeAssemblerTest {
+
+  @Test
+  @DisplayName("the assembler serves the SMTP family on the email channel")
+  void theAssemblerServesSmtpOnEmail() {
+    assertThat(assembler().type()).isEqualTo(ProviderType.SMTP);
+    assertThat(ProviderType.SMTP.channel()).isEqualTo(Channel.EMAIL);
+  }
+
+  @Test
+  @DisplayName("a configured profile assembles into a runtime bound to its own profile id")
+  void aConfiguredProfileAssemblesIntoARuntime() {
+    AssembledProvider assembled = assembler().assemble("primary-email", profile());
+
+    assertThat(assembled.channel()).isEqualTo(Channel.EMAIL);
+    assertThat(assembled.runtime().profile().profileId().value()).isEqualTo("primary-email");
+    assertThat(assembled.runtime().profile().providerId().value()).isEqualTo("smtp");
+    assertThat(assembled.runtime().profile().environment()).isEqualTo("local");
+  }
+
+  @Test
+  @DisplayName("SMTP contributes dispatch and claims nothing it cannot do")
+  void smtpClaimsNothingItCannotDo() {
+    AssembledProvider assembled = assembler().assemble("primary-email", profile());
+
+    assertThat(assembled.callback())
+        .as("SMTP has no callback adapter; claiming one fails on the first provider event instead")
+        .isEmpty();
+    assertThat(assembled.projector()).isEmpty();
+    assertThat(assembled.reconciliation())
+        .as("handing a message to a relay is the end of what the sender can observe")
+        .isEmpty();
+
+    var capabilities = assembled.runtime().profile().capabilities();
+    assertThat(capabilities.statusCallback()).isFalse();
+    assertThat(capabilities.statusQuery()).isFalse();
+    assertThat(capabilities.providerIdempotency())
+        .as("a capability declared here is a promise the dispatch loop acts on")
+        .isFalse();
+  }
+
+  @Test
+  @DisplayName("the per-profile bounds come from the profile, not from the shared transport")
+  void thePerProfileBoundsComeFromTheProfile() {
+    AssembledProvider assembled = assembler().assemble("primary-email", profile());
+
+    assertThat(assembled.runtime().profile().credentialGeneration())
+        .as("a freshly assembled profile has had no rotation; starting elsewhere fakes one")
+        .isEqualTo(1L);
+  }
+
+  private static SmtpProviderRuntimeAssembler assembler() {
+    Session session = Session.getInstance(new Properties());
+    return new SmtpProviderRuntimeAssembler(
+        message -> {
+          throw new UnsupportedOperationException("assembly only; nothing is sent here");
+        },
+        new SmtpMimeMessageFactory(session),
+        new SmtpFailureClassifier(),
+        protector(),
+        Runnable::run,
+        new AttachmentIntegrityGuard((reference, context) -> null),
+        new SmtpProviderProperties(
+            "localhost",
+            1025,
+            // The type cannot express plaintext: STARTTLS_REQUIRED and IMPLICIT_TLS are the
+            // only members, which is the transport refusing an unencrypted relay by
+            // construction rather than by a validator somebody has to remember to run.
+            SmtpProviderProperties.TlsMode.STARTTLS_REQUIRED,
+            "no-reply@example.test",
+            Duration.ofSeconds(2),
+            Duration.ofSeconds(5),
+            Duration.ofSeconds(5),
+            4),
+        Clock.systemUTC());
+  }
+
+  private static NotificationPlatformSettings.Provider profile() {
+    return new NotificationPlatformSettings.Provider(
+        "SMTP", true, true, "local", "smtp-local", "", "", "", Duration.ofSeconds(10), 4, 8);
+  }
+
+  /** Not a lambda: the protector has three operations, and only one of them is exercised here. */
+  private static dev.caskeleton.application.notification.platform.security.ContactPointProtector
+      protector() {
+    return new dev.caskeleton.application.notification.platform.security.ContactPointProtector() {
+      @Override
+      public dev.caskeleton.application.notification.platform.security.ProtectedContactPoint
+          protect(
+              dev.caskeleton.application.notification.platform.contact.ContactPointValue value) {
+        throw new UnsupportedOperationException("assembly only");
+      }
+
+      @Override
+      public dev.caskeleton.application.notification.platform.contact.ContactPointValue reveal(
+          dev.caskeleton.application.notification.platform.security.ProtectedContactPoint value,
+          dev.caskeleton.application.notification.platform.security.AccessContext context) {
+        throw new UnsupportedOperationException("assembly only");
+      }
+
+      @Override
+      public String fingerprint(
+          dev.caskeleton.application.notification.platform.contact.ContactPointValue value) {
+        throw new UnsupportedOperationException("assembly only");
+      }
+    };
+  }
+}
@@ -0,0 +1,195 @@
+package dev.caskeleton.adapter.outbound.notification.platform.security;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import dev.caskeleton.application.notification.platform.security.NotificationPayloadUnreadableException;
+import dev.caskeleton.application.notification.platform.security.SecretKeyMaterial;
+import dev.caskeleton.application.notification.platform.security.SecretMaterialProvider;
+import dev.caskeleton.application.notification.platform.security.SecretPurpose;
+import java.nio.charset.StandardCharsets;
+import java.security.SecureRandom;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The at-rest envelope, and the rotation it exists to make possible (NTF-INT-007).
+ *
+ * <p>The accept path stored template variables verbatim — caller content that can be a reset code,
+ * an order total or an address. What decides whether an encryption layer is real is not that it
+ * encrypts; it is whether the day the key changes is a change of default or a data migration.
+ */
+class AesGcmNotificationPayloadProtectionTest {
+
+  private static final byte[] PAYLOAD =
+      "{\"code\":\"481516\",\"total\":\"1250.00\"}".getBytes(StandardCharsets.UTF_8);
+
+  @Test
+  @DisplayName("a payload survives the round trip unchanged")
+  void aPayloadSurvivesTheRoundTrip() {
+    Keys keys = new Keys("payload-2026-08");
+
+    var protection = new AesGcmNotificationPayloadProtection(keys, new SecureRandom());
+    byte[] envelope = protection.protect(PAYLOAD);
+
+    assertThat(protection.reveal(envelope)).isEqualTo(PAYLOAD);
+  }
+
+  @Test
+  @DisplayName("the stored bytes contain no part of the plaintext")
+  void theStoredBytesRevealNothing() {
+    Keys keys = new Keys("payload-2026-08");
+
+    byte[] envelope =
+        new AesGcmNotificationPayloadProtection(keys, new SecureRandom()).protect(PAYLOAD);
+
+    assertThat(new String(envelope, StandardCharsets.UTF_8))
+        .as("a column an operator can read is a column an incident can read")
+        .doesNotContain("481516")
+        .doesNotContain("1250.00");
+  }
+
+  @Test
+  @DisplayName("two encryptions of one payload differ")
+  void twoEncryptionsDiffer() {
+    var protection = new AesGcmNotificationPayloadProtection(new Keys("k1"), new SecureRandom());
+
+    assertThat(protection.protect(PAYLOAD))
+        .as("a deterministic ciphertext tells an observer which two requests carried one payload")
+        .isNotEqualTo(protection.protect(PAYLOAD));
+  }
+
+  @Test
+  @DisplayName("a row written under a retired key is still readable after rotation")
+  void aRetiredKeyStillReads() {
+    Keys keys = new Keys("payload-2026-08");
+    var beforeRotation = new AesGcmNotificationPayloadProtection(keys, new SecureRandom());
+    byte[] oldRow = beforeRotation.protect(PAYLOAD);
+
+    keys.rotateTo("payload-2026-09");
+
+    assertThat(new AesGcmNotificationPayloadProtection(keys, new SecureRandom()).reveal(oldRow))
+        .as(
+            "this is the whole reason the envelope carries a key id. Without one, rotation is a "
+                + "one-way door: every row written under the previous key becomes unreadable and "
+                + "nothing in the row can say which key it needed")
+        .isEqualTo(PAYLOAD);
+  }
+
+  @Test
+  @DisplayName("a new row uses the new key, so rotation actually rotates")
+  void aNewRowUsesTheNewKey() {
+    Keys keys = new Keys("payload-2026-08");
+    keys.rotateTo("payload-2026-09");
+
+    byte[] envelope =
+        new AesGcmNotificationPayloadProtection(keys, new SecureRandom()).protect(PAYLOAD);
+
+    assertThat(keyIdOf(envelope)).isEqualTo("payload-2026-09");
+  }
+
+  @Test
+  @DisplayName("a modified ciphertext is refused rather than decrypted into something plausible")
+  void aModifiedCiphertextIsRefused() {
+    Keys keys = new Keys("k1");
+    var protection = new AesGcmNotificationPayloadProtection(keys, new SecureRandom());
+    byte[] envelope = protection.protect(PAYLOAD);
+    envelope[envelope.length - 1] ^= 0x01;
+
+    assertThatThrownBy(() -> protection.reveal(envelope))
+        .as("rendering attacker-chosen variables into a message a recipient trusts is the risk")
+        .isInstanceOf(NotificationPayloadUnreadableException.class);
+  }
+
+  @Test
+  @DisplayName("an edited key id is refused, because the header is authenticated")
+  void anEditedKeyIdIsRefused() {
+    Keys keys = new Keys("k1");
+    keys.add("k2");
+    var protection = new AesGcmNotificationPayloadProtection(keys, new SecureRandom());
+    byte[] envelope = protection.protect(PAYLOAD);
+    envelope[2] = 'k';
+    envelope[3] = '2';
+
+    assertThatThrownBy(() -> protection.reveal(envelope))
+        .as(
+            "the header is passed as AAD, so redirecting an envelope at another key fails the tag "
+                + "rather than being attempted")
+        .isInstanceOf(NotificationPayloadUnreadableException.class);
+  }
+
+  @Test
+  @DisplayName("an unknown key names itself in the failure")
+  void anUnknownKeyNamesItself() {
+    Keys keys = new Keys("payload-2026-08");
+    byte[] envelope =
+        new AesGcmNotificationPayloadProtection(keys, new SecureRandom()).protect(PAYLOAD);
+    Keys emptied = new Keys("payload-2026-09");
+
+    assertThatThrownBy(
+            () ->
+                new AesGcmNotificationPayloadProtection(emptied, new SecureRandom())
+                    .reveal(envelope))
+        .isInstanceOf(NotificationPayloadUnreadableException.class)
+        .as("the operator's next question is always which key is missing")
+        .hasMessageContaining("payload-2026-08");
+  }
+
+  @Test
+  @DisplayName("a truncated or foreign envelope is refused, not misread")
+  void aTruncatedEnvelopeIsRefused() {
+    var protection = new AesGcmNotificationPayloadProtection(new Keys("k1"), new SecureRandom());
+
+    assertThatCode(() -> protection.reveal(new byte[] {1}))
+        .isInstanceOf(NotificationPayloadUnreadableException.class);
+    assertThatCode(
+            () -> protection.reveal("plaintext row from before".getBytes(StandardCharsets.UTF_8)))
+        .as("a pre-migration plaintext row must fail loudly rather than decode into nonsense")
+        .isInstanceOf(NotificationPayloadUnreadableException.class);
+  }
+
+  private static String keyIdOf(byte[] envelope) {
+    return new String(envelope, 2, Byte.toUnsignedInt(envelope[1]), StandardCharsets.UTF_8);
+  }
+
+  /** A key store with history, which is what rotation needs and what the envelope addresses. */
+  private static final class Keys implements SecretMaterialProvider {
+
+    private final Map<String, SecretKeyMaterial> byId = new LinkedHashMap<>();
+    private String activeId;
+
+    Keys(String activeId) {
+      add(activeId);
+      this.activeId = activeId;
+    }
+
+    void add(String keyId) {
+      byte[] material = new byte[32];
+      Arrays.fill(material, (byte) keyId.hashCode());
+      byId.put(keyId, new SecretKeyMaterial(keyId, SecretPurpose.PAYLOAD_ENCRYPTION, material));
+    }
+
+    void rotateTo(String keyId) {
+      add(keyId);
+      activeId = keyId;
+    }
+
+    @Override
+    public SecretKeyMaterial activeKey(SecretPurpose purpose) {
+      return byId.get(activeId);
+    }
+
+    @Override
+    public SecretKeyMaterial keyById(String keyId) {
+      SecretKeyMaterial key = byId.get(keyId);
+      if (key == null) {
+        throw new IllegalStateException("no key " + keyId);
+      }
+      return key;
+    }
+  }
+}