From f23d70673b44c65680c267551db6cae9c3ba4c0b Mon Sep 17 00:00:00 2001 From: Paul Latzelsperger Date: Wed, 12 Aug 2026 17:53:43 +0200 Subject: [PATCH 1/3] feat(events): publish certificate-exchange status changes to NATS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Certo held CX-0135 exchange state and emitted nothing, so the only way to observe an exchange reaching FULFILLED or REJECTED was to poll its management API. It now publishes those status changes as CloudEvents onto the platform's shared edc-events JetStream stream, in the same envelope shape and subject space the EDC runtimes use, so existing consumers of that stream see certo's events alongside the connector's. Subjects are events.certificate.exchange.; types follow the CX-0000 §2.3 reverse-DNS convention already used by CcmEvents. Both are bound together in one catalogue enum (ExchangeEventType) so a consumer routing on the subject and switching on the type can never see them disagree, and a test asserts the mapping is total over FulfillmentStatus and AcceptanceStatus -- a status that maps to nothing would publish nothing, silently. Emission hangs off Spring Data's @DomainEvents rather than the ~13 service call sites: every status mutation already funnels through the aggregates' transition methods, and save() is already called after each one, so recording there means a transition cannot be missed by forgetting a publish call. The domain records facts; it does not publish them. Two things that would each have silently lost events: - fallbackExecution = true on the @TransactionalEventListener is required, not defensive. pollAcceptance/publish (provider) and initiateRequest/pollRequest/retrieve (consumer) are declared @Transactional(NOT_SUPPORTED) and commit through an inner TransactionTemplate, so by delivery time there is no surrounding transaction and Spring would discard those events -- losing the provider-initiated publish path entirely. - The consumer's updateFulfillment mirrors whatever the provider last reported and is called on every poll, so it records an event only when the status actually changes. AFTER_COMMIT means a rolled-back transition announces nothing. Publish failures are logged and swallowed, matching the EDC bridge: the state change has already committed, and failing the API call would misreport work that was done. The CloudEvents `source` is the emitting application's hostname (the pod name under Kubernetes), as in every other platform producer; the emitting tenant travels in the CX-0000 §2.1.2 sourcebpn extension and in the payload's participantContextId. Off by default at the application level (certo.events.nats.enabled), so local runs and the test suite need no broker; the chart turns it on. The chart reuses the platform's shared `edc-events` NATS identity rather than requiring a user of its own, which depends on core-platform-distribution provisioning it and on certo's ServiceAccount being listed there. Co-Authored-By: Claude Opus 5 --- build.gradle.kts | 9 + charts/certo/templates/deployment.yaml | 63 ++++++ charts/certo/values.yaml | 44 +++++ .../CertificateExchangeStatusChanged.java | 51 +++++ .../certo/common/event/ExchangeEventType.java | 93 +++++++++ .../certo/common/event/ExchangePhase.java | 11 ++ .../certo/common/event/ExchangeRole.java | 15 ++ .../common/event/nats/NKeyAuthHandler.java | 55 ++++++ .../common/event/nats/NatsConfiguration.java | 135 +++++++++++++ .../common/event/nats/NatsEventPublisher.java | 104 ++++++++++ .../common/event/nats/NatsProperties.java | 49 +++++ .../domain/ConsumerCertificateExchange.java | 81 ++++++++ .../domain/ProviderCertificateExchange.java | 79 ++++++++ src/main/resources/application.yaml | 20 ++ .../common/event/ExchangeEventTypeTest.java | 78 ++++++++ .../event/nats/NatsEventPublisherTest.java | 96 +++++++++ .../NatsEventPublishingIntegrationTest.java | 183 ++++++++++++++++++ ...ConsumerCertificateExchangeEventsTest.java | 97 ++++++++++ ...ProviderCertificateExchangeEventsTest.java | 157 +++++++++++++++ 19 files changed, 1420 insertions(+) create mode 100644 src/main/java/org/metaform/certo/common/event/CertificateExchangeStatusChanged.java create mode 100644 src/main/java/org/metaform/certo/common/event/ExchangeEventType.java create mode 100644 src/main/java/org/metaform/certo/common/event/ExchangePhase.java create mode 100644 src/main/java/org/metaform/certo/common/event/ExchangeRole.java create mode 100644 src/main/java/org/metaform/certo/common/event/nats/NKeyAuthHandler.java create mode 100644 src/main/java/org/metaform/certo/common/event/nats/NatsConfiguration.java create mode 100644 src/main/java/org/metaform/certo/common/event/nats/NatsEventPublisher.java create mode 100644 src/main/java/org/metaform/certo/common/event/nats/NatsProperties.java create mode 100644 src/test/java/org/metaform/certo/common/event/ExchangeEventTypeTest.java create mode 100644 src/test/java/org/metaform/certo/common/event/nats/NatsEventPublisherTest.java create mode 100644 src/test/java/org/metaform/certo/common/event/nats/NatsEventPublishingIntegrationTest.java create mode 100644 src/test/java/org/metaform/certo/consumer/domain/ConsumerCertificateExchangeEventsTest.java create mode 100644 src/test/java/org/metaform/certo/provider/domain/ProviderCertificateExchangeEventsTest.java diff --git a/build.gradle.kts b/build.gradle.kts index c60d5f9..439dbb4 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -33,6 +33,11 @@ dependencies { runtimeOnly("com.h2database:h2") runtimeOnly("org.postgresql:postgresql") + // NATS/JetStream client for publishing certificate-exchange events onto the platform's shared + // `edc-events` stream. Version-matched to the platform's EDC events-nats bridge and the CX-VE + // onboarding API, so all three speak to the same server through the same client. + implementation("io.nats:jnats:2.25.3") + // HTTP client used by the consumer to retrieve certificates from a provider's data plane. implementation("com.squareup.okhttp3:okhttp:4.12.0") // Failsafe: retry with exponential backoff around outbound OkHttp calls — the same retry library @@ -55,6 +60,10 @@ dependencies { testImplementation("com.squareup.okhttp3:mockwebserver:4.12.0") // MockMvc security integration (auto-applies the filter chain) for the management-API auth tests. testImplementation("org.springframework.security:spring-security-test") + // A real NATS server for the event-publishing test: the @DomainEvents -> after-commit -> JetStream + // path spans Spring Data, the transaction manager and the NATS client, and only an end-to-end run + // proves it. Requires Docker; the test skips itself when none is available. + testImplementation("org.testcontainers:junit-jupiter:1.21.3") testRuntimeOnly("org.junit.platform:junit-platform-launcher") } diff --git a/charts/certo/templates/deployment.yaml b/charts/certo/templates/deployment.yaml index 32ea417..8958802 100644 --- a/charts/certo/templates/deployment.yaml +++ b/charts/certo/templates/deployment.yaml @@ -21,6 +21,42 @@ spec: imagePullSecrets: {{- toYaml . | nindent 8 }} {{- end }} + {{- if and .Values.events.nats.enabled .Values.natsAuth.enabled }} + initContainers: + # Logs into Vault with the pod's SA token and drops certo's NATS NKey seed on a pod-private + # in-memory volume. It only FETCHES an identity that must already have been provisioned — + # see the natsAuth notes in values.yaml, which the stock Core Platform Distribution does not + # do for certo. + # + # The retry is unbounded by design (Vault and its bootstrap come up asynchronously), so a + # missing role or seed leaves the pod in Init:0/1 rather than failing fast. Vault's error is + # deliberately NOT sent to /dev/null: "invalid role name" is exactly what you need to see here. + - name: fetch-nats-nkey + image: {{ .Values.natsAuth.image }} + command: [ "sh", "-ec" ] + args: + - | + export VAULT_ADDR={{ .Values.natsAuth.vaultUrl | quote }} + echo "Fetching the NATS NKey seed (Vault role {{ .Values.natsAuth.vaultRole }})..." + until VAULT_TOKEN=$(vault write -field=token auth/kubernetes/login \ + role={{ .Values.natsAuth.vaultRole }} \ + jwt=@/var/run/secrets/kubernetes.io/serviceaccount/token); do + echo "Vault login failed (Vault or its bootstrap not ready yet), retrying in 2 seconds..." + sleep 2 + done + export VAULT_TOKEN + until vault kv get -field=seed {{ .Values.natsAuth.vaultSecretPath }} > /vault/secrets/nats.nk; do + echo "Seed not in Vault yet (nats-auth-bootstrap still running?), retrying in 2 seconds..." + sleep 2 + done + # 0444 rather than 0400: init and app container may run as different UIDs; + # the volume is pod-private tmpfs either way. + chmod 0444 /vault/secrets/nats.nk + echo "NKey seed written to /vault/secrets/nats.nk" + volumeMounts: + - name: nats-nkey + mountPath: /vault/secrets + {{- end }} containers: - name: certo image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" @@ -48,6 +84,20 @@ spec: - name: CERTO_SECURITY_SIGLETBASEURL value: {{ .Values.sigletBaseUrl | quote }} {{- end }} + {{- if .Values.events.nats.enabled }} + - name: CERTO_EVENTS_NATS_ENABLED + value: "true" + - name: CERTO_EVENTS_NATS_URL + value: {{ .Values.events.nats.url | quote }} + - name: CERTO_EVENTS_NATS_STREAM + value: {{ .Values.events.nats.stream | quote }} + - name: CERTO_EVENTS_NATS_CREATESTREAM + value: {{ .Values.events.nats.createStream | quote }} + {{- if .Values.natsAuth.enabled }} + - name: CERTO_EVENTS_NATS_NKEYSEEDPATH + value: /vault/secrets/nats.nk + {{- end }} + {{- end }} {{- with .Values.extraEnv }} {{- toYaml . | nindent 12 }} {{- end }} @@ -63,7 +113,20 @@ spec: readinessProbe: tcpSocket: port: http + {{- if and .Values.events.nats.enabled .Values.natsAuth.enabled }} + volumeMounts: + - name: nats-nkey + mountPath: /vault/secrets + readOnly: true + {{- end }} {{- with .Values.resources }} resources: {{- toYaml . | nindent 12 }} {{- end }} + {{- if and .Values.events.nats.enabled .Values.natsAuth.enabled }} + volumes: + # Pod-private in-memory volume the NKey seed is delivered on (never hits disk) + - name: nats-nkey + emptyDir: + medium: Memory + {{- end }} diff --git a/charts/certo/values.yaml b/charts/certo/values.yaml index 294b55c..8cb1cb3 100644 --- a/charts/certo/values.yaml +++ b/charts/certo/values.yaml @@ -33,6 +33,50 @@ springProfile: prod # layer; point it at a mock siglet for dev/test. sigletBaseUrl: "" +# Certificate-exchange status changes (CX-0135 §2.1.3) published as CloudEvents on NATS JetStream, +# on subjects events.certificate.exchange.* — the same shape and stream the platform's EDC runtimes +# publish to, so consumers of `edc-events` see certo's events alongside the connector's. +# +# ON by default — emitting these events is part of what certo is for. Set `enabled: false` to run +# certo with no broker dependency at all (it then makes no NATS connection and the publisher bean +# does not exist). `url` must point at a reachable NATS server whenever this is on. +events: + nats: + enabled: true + url: nats://nats:4222 + # The stream events are expected to land on. Only used when createStream is set: a publish + # addresses a subject, and the server routes it to whichever stream captures that subject. + stream: edc-events + # Create the stream if missing. Standalone/dev only — in the platform the stream is owned by the + # nats-bootstrap job and certo's NATS user has no stream-management rights. + createStream: false + +# NKey authentication for the NATS connection. When enabled an init container logs into Vault with +# the pod's ServiceAccount token and drops the seed on a pod-private in-memory volume, exactly as the +# platform's other NATS clients do. +# +# ON by default, matching the platform's posture that every NATS client authenticates. +# +# It requires a certo NATS identity to already exist: a Vault k8s-auth role bound to this pod's +# ServiceAccount, a seed under `vaultSecretPath`, and a users.conf entry permitting +# `publish: ["events.>", "$JS.API.>"]`. Note the Core Platform Distribution provisions identities for +# its own components from a fixed list (controlplane, identityhub, issuerservice, cfm-agents, +# nats-admin) that does NOT yet include certo — against such a platform there is nothing to fetch, +# the init container retries indefinitely and the pod stays in Init:0/1. +# +# Set `enabled: false` where NATS runs without authentication; publishing then still works, over an +# unauthenticated connection. That is what the CX-VE install script does today. +natsAuth: + enabled: true + vaultUrl: http://vault:8200 + # Defaults to REUSING the platform's shared `edc-events` identity — certo publishes on events.>, + # which is that identity's profile — rather than requiring a user of its own. The deployment must + # add certo's ServiceAccount to that identity's serviceAccounts list on the platform side. + # Both are configurable so this also works against whatever else mints certo an identity. + vaultRole: nats-edc-events + vaultSecretPath: secret/nats/edc-events + image: hashicorp/vault:latest + database: url: jdbc:postgresql://postgres:5432/certo username: certo diff --git a/src/main/java/org/metaform/certo/common/event/CertificateExchangeStatusChanged.java b/src/main/java/org/metaform/certo/common/event/CertificateExchangeStatusChanged.java new file mode 100644 index 0000000..71da732 --- /dev/null +++ b/src/main/java/org/metaform/certo/common/event/CertificateExchangeStatusChanged.java @@ -0,0 +1,51 @@ +package org.metaform.certo.common.event; + +import com.fasterxml.jackson.annotation.JsonInclude; +import org.metaform.certo.common.model.StatusError; + +import java.time.OffsetDateTime; +import java.util.List; + +/** + * A status change on one phase of a {@code Certificate Exchange} — the {@code data} payload of the + * CloudEvent published to NATS. + * + *

Recorded by the exchange aggregates themselves (see {@code ProviderCertificateExchange} and + * {@code ConsumerCertificateExchange}) and drained by Spring Data on {@code save()}, so it is a + * statement of fact about committed state rather than an intent. + * + * @param role which aggregate observed the change; a single Certo may hold both sides + * @param phase Fulfillment or Acceptance (CX-0135 §2.1.3) + * @param eventType the catalogue entry — carries the subject and CloudEvents type + * @param exchangeId the exchange this change belongs to + * @param participantContextId the tenant that owns the exchange; resolves the CloudEvents source/sourcebpn + * @param counterpartyBpn the other party's BPN (the consumer's, seen from the provider, and vice versa) + * @param counterpartyDid the other party's DID + * @param certificateId null while the certificate identity is still unknown (a pending request) + * @param revision the certificate revision, null alongside an unknown {@code certificateId} + * @param previousStatus the status being left, or null when the exchange was just opened — + * which is what distinguishes "opened in state X" from "transitioned to X" + * @param status the status now in effect + * @param errors CX-0135 §4.4.4 error details accompanying the new status, if any + * @param occurredAt when the change was recorded + */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public record CertificateExchangeStatusChanged( + ExchangeRole role, + ExchangePhase phase, + ExchangeEventType eventType, + String exchangeId, + String participantContextId, + String counterpartyBpn, + String counterpartyDid, + String certificateId, + Integer revision, + String previousStatus, + String status, + List errors, + OffsetDateTime occurredAt) { + + public CertificateExchangeStatusChanged { + errors = errors == null ? null : List.copyOf(errors); + } +} diff --git a/src/main/java/org/metaform/certo/common/event/ExchangeEventType.java b/src/main/java/org/metaform/certo/common/event/ExchangeEventType.java new file mode 100644 index 0000000..6c3cb8a --- /dev/null +++ b/src/main/java/org/metaform/certo/common/event/ExchangeEventType.java @@ -0,0 +1,93 @@ +package org.metaform.certo.common.event; + +import org.metaform.certo.common.model.AcceptanceStatus; +import org.metaform.certo.common.model.FulfillmentStatus; + +import java.util.EnumMap; +import java.util.Map; + +/** + * The published catalogue of certificate-exchange events: one entry per CX-0135 §2.1.3 status, + * binding it to the NATS subject it is published on and the CloudEvents {@code type} it carries. + * + *

Subject and type live together here on purpose — they are the two halves of this component's + * public contract, and a consumer routing on one while switching on the other must never see them + * disagree. {@code ExchangeEventTypeTest} asserts the mapping is total over both status enums, so a + * status added to {@link FulfillmentStatus} or {@link AcceptanceStatus} fails the build here rather + * than silently emitting nothing. + * + *

The {@code events.} prefix is not cosmetic: it is the subject space the platform's + * {@code edc-events} JetStream stream captures ({@code subjects: "events.>"}), and the only one the + * NATS permission matrix grants publishers. Leaves are lowerCamelCase, matching the EDC event bridge + * (e.g. {@code events.transfer.process.deprovisioningRequested}). Types follow the CX-0000 §2.3 + * reverse-DNS convention already used by {@link org.metaform.certo.common.cloudevent.CcmEvents}. + */ +public enum ExchangeEventType { + + REQUESTED(ExchangePhase.FULFILLMENT, "requested", "CertificateExchangeRequested"), + ACKNOWLEDGED(ExchangePhase.FULFILLMENT, "acknowledged", "CertificateExchangeAcknowledged"), + CERTIFICATION_REQUESTED(ExchangePhase.FULFILLMENT, "certificationRequested", "CertificateExchangeCertificationRequested"), + FULFILLED(ExchangePhase.FULFILLMENT, "fulfilled", "CertificateExchangeFulfilled"), + DECLINED(ExchangePhase.FULFILLMENT, "declined", "CertificateExchangeDeclined"), + FAILED(ExchangePhase.FULFILLMENT, "failed", "CertificateExchangeFailed"), + + RETRIEVED(ExchangePhase.ACCEPTANCE, "retrieved", "CertificateExchangeRetrieved"), + ACCEPTED(ExchangePhase.ACCEPTANCE, "accepted", "CertificateExchangeAccepted"), + REJECTED(ExchangePhase.ACCEPTANCE, "rejected", "CertificateExchangeRejected"), + ERRORED(ExchangePhase.ACCEPTANCE, "errored", "CertificateExchangeErrored"); + + /** Subject namespace of every event in this catalogue; {@code events.certificate.exchange.>} takes them all. */ + public static final String SUBJECT_PREFIX = "events.certificate.exchange."; + private static final String TYPE_PREFIX = "org.catena-x.ccm."; + private static final String TYPE_SUFFIX = ".v1"; + + private static final Map BY_FULFILLMENT = new EnumMap<>(FulfillmentStatus.class); + private static final Map BY_ACCEPTANCE = new EnumMap<>(AcceptanceStatus.class); + + static { + for (var status : FulfillmentStatus.values()) { + BY_FULFILLMENT.put(status, valueOf(status.name())); + } + for (var status : AcceptanceStatus.values()) { + BY_ACCEPTANCE.put(status, valueOf(status.name())); + } + } + + private final ExchangePhase phase; + private final String subject; + private final String type; + + ExchangeEventType(ExchangePhase phase, String subjectLeaf, String typeName) { + this.phase = phase; + this.subject = SUBJECT_PREFIX + subjectLeaf; + this.type = TYPE_PREFIX + typeName + TYPE_SUFFIX; + } + + /** + * The catalogue entry for a Fulfillment status. Total by construction — the static initializer + * resolves every {@link FulfillmentStatus} constant by name and fails class initialization if one + * has no counterpart here. + */ + public static ExchangeEventType of(FulfillmentStatus status) { + return BY_FULFILLMENT.get(status); + } + + /** The catalogue entry for an Acceptance status. Total by construction — see {@link #of(FulfillmentStatus)}. */ + public static ExchangeEventType of(AcceptanceStatus status) { + return BY_ACCEPTANCE.get(status); + } + + public ExchangePhase phase() { + return phase; + } + + /** The NATS subject this event is published on. */ + public String subject() { + return subject; + } + + /** The CloudEvents {@code type} attribute (CX-0000 §2.3 reverse-DNS). */ + public String type() { + return type; + } +} diff --git a/src/main/java/org/metaform/certo/common/event/ExchangePhase.java b/src/main/java/org/metaform/certo/common/event/ExchangePhase.java new file mode 100644 index 0000000..a7e5a3f --- /dev/null +++ b/src/main/java/org/metaform/certo/common/event/ExchangePhase.java @@ -0,0 +1,11 @@ +package org.metaform.certo.common.event; + +/** + * The phase of the CX-0135 §2.1.3 exchange state machine a status change belongs to. + */ +public enum ExchangePhase { + /** Provider-owned: REQUESTED / ACKNOWLEDGED / CERTIFICATION_REQUESTED / FULFILLED / DECLINED / FAILED. */ + FULFILLMENT, + /** Consumer-owned: RETRIEVED / ACCEPTED / REJECTED / ERRORED. */ + ACCEPTANCE +} diff --git a/src/main/java/org/metaform/certo/common/event/ExchangeRole.java b/src/main/java/org/metaform/certo/common/event/ExchangeRole.java new file mode 100644 index 0000000..c2601df --- /dev/null +++ b/src/main/java/org/metaform/certo/common/event/ExchangeRole.java @@ -0,0 +1,15 @@ +package org.metaform.certo.common.event; + +/** + * Which side of a {@code Certificate Exchange} observed a status change. + * + *

A single Certo instance may act as both provider and consumer (the Verification Environment + * deploys exactly one), so the same logical exchange produces events from both aggregates. Without + * this discriminator the two are indistinguishable on the wire. + */ +public enum ExchangeRole { + /** The provider's record ({@code ProviderCertificateExchange}) — authoritative for Fulfillment. */ + PROVIDER, + /** The consumer's record ({@code ConsumerCertificateExchange}) — authoritative for Acceptance. */ + CONSUMER +} diff --git a/src/main/java/org/metaform/certo/common/event/nats/NKeyAuthHandler.java b/src/main/java/org/metaform/certo/common/event/nats/NKeyAuthHandler.java new file mode 100644 index 0000000..b44b5a6 --- /dev/null +++ b/src/main/java/org/metaform/certo/common/event/nats/NKeyAuthHandler.java @@ -0,0 +1,55 @@ +package org.metaform.certo.common.event.nats; + +import io.nats.client.AuthHandler; +import io.nats.client.NKey; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; + +/** + * NATS {@link AuthHandler} that authenticates with an ed25519 NKey seed read from a file (the + * Vault-mounted {@code /vault/secrets/nats.nk}). The seed never leaves the process: the server sends + * a nonce and we return its signature. + * + *

The seed is held in memory as a char array for the lifetime of the connection, matching how the + * platform's other Java runtimes authenticate. + */ +public class NKeyAuthHandler implements AuthHandler { + + private final char[] seed; + + public NKeyAuthHandler(Path seedFile) { + try { + this.seed = new String(Files.readAllBytes(seedFile), StandardCharsets.UTF_8).trim().toCharArray(); + } catch (IOException e) { + throw new UncheckedIOException("Unable to read NATS NKey seed from " + seedFile, e); + } + } + + @Override + public char[] getID() { + try { + return NKey.fromSeed(seed).getPublicKey(); + } catch (Exception e) { + throw new IllegalStateException("Unable to derive public key from NKey seed", e); + } + } + + @Override + public byte[] sign(byte[] nonce) { + try { + return NKey.fromSeed(seed).sign(nonce); + } catch (Exception e) { + throw new IllegalStateException("Unable to sign NATS nonce with NKey seed", e); + } + } + + @Override + public char[] getJWT() { + // NKey auth (not JWT/creds-based), so no JWT is presented. + return null; + } +} diff --git a/src/main/java/org/metaform/certo/common/event/nats/NatsConfiguration.java b/src/main/java/org/metaform/certo/common/event/nats/NatsConfiguration.java new file mode 100644 index 0000000..f1ae71a --- /dev/null +++ b/src/main/java/org/metaform/certo/common/event/nats/NatsConfiguration.java @@ -0,0 +1,135 @@ +package org.metaform.certo.common.event.nats; + +import io.nats.client.Connection; +import io.nats.client.JetStream; +import io.nats.client.JetStreamApiException; +import io.nats.client.Nats; +import io.nats.client.Options; +import io.nats.client.api.RetentionPolicy; +import io.nats.client.api.StorageType; +import io.nats.client.api.StreamConfiguration; +import org.metaform.certo.common.event.ExchangeEventType; +import org.metaform.certo.common.pc.store.ParticipantContextStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +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 tools.jackson.databind.ObjectMapper; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.nio.file.Path; +import java.time.Duration; + +/** + * Wires the NATS connection and JetStream context when {@code certo.events.nats.enabled=true}. Left + * disabled (the default) the app starts with no broker dependency, and no publisher bean exists. + * + *

Connection options mirror the platform's other publishers (the EDC {@code events-nats} bridge and + * the CX-VE onboarding API): reconnect forever, so a NATS restart does not take the app down with it. + */ +@Configuration +@EnableConfigurationProperties(NatsProperties.class) +@ConditionalOnProperty(prefix = "certo.events.nats", name = "enabled", havingValue = "true") +public class NatsConfiguration { + + private static final Logger log = LoggerFactory.getLogger(NatsConfiguration.class); + /** NATS API error code for "stream name already in use". */ + private static final int ERR_STREAM_NAME_IN_USE = 10058; + + @Bean(destroyMethod = "close") + public Connection natsConnection(NatsProperties properties) throws IOException, InterruptedException { + var options = new Options.Builder() + .server(properties.url()) + .maxReconnects(-1) + .reconnectWait(Duration.ofSeconds(1)) + .pingInterval(Duration.ofSeconds(20)) + .maxPingsOut(5); + if (properties.hasNkeyAuth()) { + options.authHandler(new NKeyAuthHandler(Path.of(properties.nkeySeedPath()))); + log.info("Connecting to NATS at {} with NKey auth", properties.url()); + } else { + log.info("Connecting to NATS at {} without authentication", properties.url()); + } + return Nats.connect(options.build()); + } + + @Bean + public JetStream jetStream(Connection connection, NatsProperties properties) throws IOException, JetStreamApiException { + if (properties.createStream()) { + createStreamIfAbsent(connection, properties); + } + return connection.jetStream(); + } + + /** + * The publisher lives here rather than being component-scanned so that this class's + * {@code @ConditionalOnProperty} is the single switch: with publishing off there is no + * {@link JetStream} bean, and a scanned publisher would fail the context looking for one. + */ + @Bean + public NatsEventPublisher natsEventPublisher(JetStream jetStream, + ObjectMapper mapper, + ParticipantContextStore contextStore, + NatsProperties properties) { + var source = resolveSource(properties); + log.info("Publishing certificate-exchange events with CloudEvents source '{}'", source); + return new NatsEventPublisher(jetStream, mapper, contextStore, source); + } + + /** + * The CloudEvents {@code source} for every event this app emits: its hostname, matching the EDC + * runtimes (their events-nats bridge uses the injected {@code Hostname} service). In Kubernetes + * HOSTNAME is the pod name, which is what identifies the producing instance. + * + *

Overridable via {@code certo.events.nats.source} for deployments that want a stable logical + * name rather than a per-pod one. + */ + private static String resolveSource(NatsProperties properties) { + if (properties.source() != null && !properties.source().isBlank()) { + return properties.source(); + } + var fromEnv = System.getenv("HOSTNAME"); + if (fromEnv != null && !fromEnv.isBlank()) { + return fromEnv; + } + try { + return InetAddress.getLocalHost().getHostName(); + } catch (UnknownHostException e) { + // Never fail startup over a cosmetic attribute; "localhost" is what EDC's Hostname + // service defaults to as well. + log.warn("Could not resolve the local hostname for the CloudEvents source, using 'localhost'", e); + return "localhost"; + } + } + + /** + * Creates the stream when it is missing. Development convenience only: in the platform the + * {@code edc-events} stream is owned by the {@code nats-bootstrap} job and this publisher's NATS + * user has no stream-management rights, so {@code createStream} must stay false there. + * + *

Interest retention matches the platform's stream: a message is removed once every registered + * consumer has taken it. + */ + private void createStreamIfAbsent(Connection connection, NatsProperties properties) throws IOException, JetStreamApiException { + var config = StreamConfiguration.builder() + .name(properties.stream()) + .subjects(ExchangeEventType.SUBJECT_PREFIX + ">") + .storageType(StorageType.Memory) + .retentionPolicy(RetentionPolicy.Interest) + .build(); + try { + connection.jetStreamManagement().addStream(config); + log.info("Created NATS stream '{}'", properties.stream()); + } catch (JetStreamApiException e) { + if (e.getApiErrorCode() != ERR_STREAM_NAME_IN_USE) { + throw e; + } + log.debug("NATS stream '{}' already exists, leaving it as is", properties.stream()); + } + } +} diff --git a/src/main/java/org/metaform/certo/common/event/nats/NatsEventPublisher.java b/src/main/java/org/metaform/certo/common/event/nats/NatsEventPublisher.java new file mode 100644 index 0000000..591e802 --- /dev/null +++ b/src/main/java/org/metaform/certo/common/event/nats/NatsEventPublisher.java @@ -0,0 +1,104 @@ +package org.metaform.certo.common.event.nats; + +import io.nats.client.JetStream; +import io.nats.client.impl.Headers; +import org.metaform.certo.common.cloudevent.CcmEvents; +import org.metaform.certo.common.cloudevent.CloudEvent; +import org.metaform.certo.common.event.CertificateExchangeStatusChanged; +import org.metaform.certo.common.pc.domain.ParticipantContext; +import org.metaform.certo.common.pc.store.ParticipantContextStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; +import tools.jackson.databind.ObjectMapper; + +import java.util.UUID; + +/** + * Publishes {@link CertificateExchangeStatusChanged} to NATS JetStream as a CloudEvents 1.0 event in + * JSON structured mode — the same shape the EDC runtimes publish through their {@code events-nats} + * bridge, so a consumer of the platform's {@code edc-events} stream handles Certo's events the same + * way it handles the connector's. + * + *

Not a {@code @Component}: it is registered by {@link NatsConfiguration}, which carries the + * {@code certo.events.nats.enabled} condition. Component-scanning it would register the bean + * unconditionally and fail the context with a missing {@code JetStream} whenever publishing is off — + * which is the default. + * + *

Delivery semantics. The listener runs {@code AFTER_COMMIT}, so a transaction that rolls + * back announces nothing. {@code fallbackExecution = true} is load-bearing rather than defensive: + * several exchange operations — {@code ProviderExchangeService.pollAcceptance} and {@code publish}, + * {@code ConsumerExchangeService.initiateRequest}, {@code pollRequest} and {@code retrieve} — are + * declared {@code @Transactional(propagation = NOT_SUPPORTED)} and commit through an inner + * {@code TransactionTemplate}. By the time the event is delivered there is no surrounding transaction, + * and without this flag Spring would discard those events silently, losing the provider-initiated + * publish path entirely. + * + *

Failures are swallowed and logged, matching the EDC bridge. The state change has already + * committed; propagating a broker failure would fail an API call whose work was actually done. + */ +public class NatsEventPublisher { + + private static final Logger log = LoggerFactory.getLogger(NatsEventPublisher.class); + + private final JetStream jetStream; + private final ObjectMapper mapper; + private final ParticipantContextStore contextStore; + private final String source; + + public NatsEventPublisher(JetStream jetStream, ObjectMapper mapper, ParticipantContextStore contextStore, String source) { + this.jetStream = jetStream; + this.mapper = mapper; + this.contextStore = contextStore; + this.source = source; + } + + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true) + public void on(CertificateExchangeStatusChanged event) { + try { + var subject = event.eventType().subject(); + jetStream.publish(subject, headers(), mapper.writeValueAsBytes(envelope(event))); + log.debug("Published {} for exchange {}", subject, event.exchangeId()); + } catch (Exception e) { + // Deliberately broad: a publishing failure must never surface as a failed API call, and + // must never mask the fact that the transition itself committed. + log.error("Failed to publish {} for exchange {}", + event.eventType().subject(), event.exchangeId(), e); + } + } + + /** + * Wraps the payload in the CloudEvents envelope. + * + *

{@code source} identifies the EMITTING APPLICATION — certo's hostname — matching every other + * platform event producer (the EDC bridge builds it as {@code URI.create(hostname)} from its + * injected {@code Hostname} service). It deliberately does NOT carry the tenant: a consumer + * de-duplicating on {@code source} + {@code id} needs a stable per-producer value, and tools that + * group events by origin expect the service, not the participant. + * + *

The emitting tenant is identified by the CX-0000 §2.1.2-REQUIRED {@code sourcebpn} extension + * instead, read from its participant context, with {@code participantContextId} in the payload. + */ + // Package-private so the envelope contract can be asserted without standing up a broker. + CloudEvent envelope(CertificateExchangeStatusChanged event) { + var context = contextStore.find(event.participantContextId()).orElse(null); + return new CloudEvent<>( + CloudEvent.SPEC_VERSION, + event.eventType().type(), + source, + event.exchangeId(), + UUID.randomUUID().toString(), + event.occurredAt(), + CloudEvent.CONTENT_TYPE_JSON, + null, + // A tenant deleted between the transition committing and this listener running leaves + // nothing to read; the event still goes out, carrying participantContextId in the data. + context == null ? null : context.bpn(), + event); + } + + private static Headers headers() { + return new Headers().add("Content-Type", CcmEvents.CONTENT_TYPE); + } +} diff --git a/src/main/java/org/metaform/certo/common/event/nats/NatsProperties.java b/src/main/java/org/metaform/certo/common/event/nats/NatsProperties.java new file mode 100644 index 0000000..fe9bfc5 --- /dev/null +++ b/src/main/java/org/metaform/certo/common/event/nats/NatsProperties.java @@ -0,0 +1,49 @@ +package org.metaform.certo.common.event.nats; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Configuration for publishing certificate-exchange events to NATS JetStream. + * + *

Defaults target the platform: the shared {@code edc-events} stream (created by the platform's + * {@code nats-bootstrap} job with subjects {@code events.>}) and the Vault-delivered NKey seed at + * {@code /vault/secrets/nats.nk}. Disabled by default, so the app, the test suite and any + * deployment that has not opted in run with no broker dependency at all. + * + * @param enabled whether to connect to NATS and publish at all + * @param url NATS server URL + * @param stream the JetStream stream events are expected to land on; only used when + * {@code createStream} is set, since publishing addresses a subject, not a stream + * @param createStream create {@code stream} on startup if absent. For standalone development only — + * in the platform the stream is owned by the {@code nats-bootstrap} job, and the + * publisher's NATS user is not permitted to manage streams + * @param nkeySeedPath path to the ed25519 NKey seed; blank connects unauthenticated (local dev, or a + * cluster with NATS auth switched off) + * @param source CloudEvents {@code source} for emitted events — the emitting application, as + * in every other platform producer. Blank resolves to the hostname (the pod + * name under Kubernetes); set it for a stable logical name instead + */ +@ConfigurationProperties(prefix = "certo.events.nats") +public record NatsProperties( + boolean enabled, + String url, + String stream, + boolean createStream, + String nkeySeedPath, + String source +) { + + public NatsProperties { + if (url == null || url.isBlank()) { + url = "nats://localhost:4222"; + } + if (stream == null || stream.isBlank()) { + stream = "edc-events"; + } + } + + /** An NKey seed path is optional (absent → connect unauthenticated). */ + public boolean hasNkeyAuth() { + return nkeySeedPath != null && !nkeySeedPath.isBlank(); + } +} diff --git a/src/main/java/org/metaform/certo/consumer/domain/ConsumerCertificateExchange.java b/src/main/java/org/metaform/certo/consumer/domain/ConsumerCertificateExchange.java index 00a0982..73830ea 100644 --- a/src/main/java/org/metaform/certo/consumer/domain/ConsumerCertificateExchange.java +++ b/src/main/java/org/metaform/certo/consumer/domain/ConsumerCertificateExchange.java @@ -7,8 +7,13 @@ import jakarta.persistence.Enumerated; import jakarta.persistence.Id; import jakarta.persistence.Table; +import jakarta.persistence.Transient; import jakarta.persistence.Version; import org.jetbrains.annotations.NotNull; +import org.metaform.certo.common.event.CertificateExchangeStatusChanged; +import org.metaform.certo.common.event.ExchangeEventType; +import org.metaform.certo.common.event.ExchangePhase; +import org.metaform.certo.common.event.ExchangeRole; import org.metaform.certo.common.util.Validations; import org.metaform.certo.common.model.AcceptanceStatus; import org.metaform.certo.common.model.FulfillmentStatus; @@ -17,7 +22,12 @@ import org.metaform.certo.common.persistence.StatusErrorListConverter; import org.metaform.certo.common.web.ApiException; import org.metaform.certo.consumer.spi.RetrievedCertificate; +import org.springframework.data.domain.AfterDomainEventPublication; +import org.springframework.data.domain.DomainEvents; +import java.time.OffsetDateTime; +import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.Objects; @@ -96,6 +106,14 @@ public class ConsumerCertificateExchange { @Version private long version; + /** + * Status changes recorded since this aggregate was last saved, drained by Spring Data's + * {@code @DomainEvents} support on {@code save()}. Never persisted, and never populated on an + * entity loaded from the database — only a mutation made in this JVM records one. + */ + @Transient + private final transient List domainEvents = new ArrayList<>(); + protected ConsumerCertificateExchange() { // for JPA } @@ -112,6 +130,7 @@ public ConsumerCertificateExchange(String exchangeId, String certificateId, Inte this.participantContextId = Validations.requireNonBlank(participantContextId, "participantContextId"); this.providerBpn = Validations.requireNonBlank(providerBpn, "providerBpn"); this.providerDid = Validations.requireNonBlank(providerDid, "providerDid"); + recordFulfillmentChange(null); } /** @@ -120,11 +139,18 @@ public ConsumerCertificateExchange(String exchangeId, String certificateId, Inte * once the backend issues the certificate); it is adopted here when the provider reports one. */ public void updateFulfillment(FulfillmentStatus status, String certificateId, List errors) { + var previous = fulfillmentStatus; this.fulfillmentStatus = status; this.fulfillmentErrors = errors; if (certificateId != null) { this.certificateId = certificateId; } + // Only a real change is an event. Unlike the provider's transitionFulfillment (which rejects a + // no-op transition outright), this method mirrors whatever the provider last reported and is + // called on every poll — most of which report the status unchanged. + if (previous != status) { + recordFulfillmentChange(previous); + } } /** @@ -137,9 +163,64 @@ public void transitionAcceptance(AcceptanceStatus status, List erro throw ApiException.conflict("Illegal acceptance transition " + acceptanceStatus + " -> " + status + " for exchange " + exchangeId); } + var previous = acceptanceStatus; this.acceptanceStatus = status; this.acceptanceErrors = errors; this.acceptanceReported = false; + recordAcceptanceChange(previous); + } + + // --- domain events ------------------------------------------------------------------------- + + /** + * Records the Fulfillment status now in effect (mirrored from the provider). {@code previous} is + * null when the exchange was just opened. + */ + private void recordFulfillmentChange(FulfillmentStatus previous) { + domainEvents.add(new CertificateExchangeStatusChanged( + ExchangeRole.CONSUMER, + ExchangePhase.FULFILLMENT, + ExchangeEventType.of(fulfillmentStatus), + exchangeId, + participantContextId, + providerBpn, + providerDid, + certificateId, + certificateId == null ? null : revision, + previous == null ? null : previous.name(), + fulfillmentStatus.name(), + fulfillmentErrors, + OffsetDateTime.now())); + } + + /** Records the Acceptance verdict now in effect; {@code previous} is null for the first one recorded. */ + private void recordAcceptanceChange(AcceptanceStatus previous) { + domainEvents.add(new CertificateExchangeStatusChanged( + ExchangeRole.CONSUMER, + ExchangePhase.ACCEPTANCE, + ExchangeEventType.of(acceptanceStatus), + exchangeId, + participantContextId, + providerBpn, + providerDid, + certificateId, + certificateId == null ? null : revision, + previous == null ? null : previous.name(), + acceptanceStatus.name(), + acceptanceErrors, + OffsetDateTime.now())); + } + + /** Drained by Spring Data on {@code save()}; the NATS publisher listens after commit. */ + @DomainEvents + Collection domainEvents() { + return List.copyOf(domainEvents); + } + + /** Clears the recorded events after publication so a second {@code save()} does not re-publish them. */ + @AfterDomainEventPublication + void clearDomainEvents() { + domainEvents.clear(); } /** Whether this exchange's recorded acceptance still needs (re-)reporting to the provider. */ diff --git a/src/main/java/org/metaform/certo/provider/domain/ProviderCertificateExchange.java b/src/main/java/org/metaform/certo/provider/domain/ProviderCertificateExchange.java index 20323bd..6acf856 100644 --- a/src/main/java/org/metaform/certo/provider/domain/ProviderCertificateExchange.java +++ b/src/main/java/org/metaform/certo/provider/domain/ProviderCertificateExchange.java @@ -11,17 +11,26 @@ import jakarta.persistence.Id; import jakarta.persistence.JoinColumn; import jakarta.persistence.Table; +import jakarta.persistence.Transient; import jakarta.persistence.UniqueConstraint; import jakarta.persistence.Version; import org.jetbrains.annotations.NotNull; +import org.metaform.certo.common.event.CertificateExchangeStatusChanged; +import org.metaform.certo.common.event.ExchangeEventType; +import org.metaform.certo.common.event.ExchangePhase; +import org.metaform.certo.common.event.ExchangeRole; import org.metaform.certo.common.util.Validations; import org.metaform.certo.common.model.AcceptanceStatus; import org.metaform.certo.common.model.FulfillmentStatus; import org.metaform.certo.common.model.StatusError; import org.metaform.certo.common.persistence.StatusErrorListConverter; import org.metaform.certo.common.web.ApiException; +import org.springframework.data.domain.AfterDomainEventPublication; +import org.springframework.data.domain.DomainEvents; import java.time.OffsetDateTime; +import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.Objects; @@ -121,6 +130,15 @@ public class ProviderCertificateExchange { @Version private long version; + /** + * Status changes recorded since this aggregate was last saved, drained by Spring Data's + * {@code @DomainEvents} support on {@code save()} (see {@link #domainEvents()}). Transient in every + * sense: never persisted, and never populated on an entity loaded from the database — only a + * mutation made in this JVM records one. + */ + @Transient + private final transient List domainEvents = new ArrayList<>(); + protected ProviderCertificateExchange() { // for JPA } @@ -151,6 +169,7 @@ public ProviderCertificateExchange(String exchangeId, this.counterpartyDid = Validations.requireNonBlank(counterpartyDid, "counterpartyDid"); this.fulfillmentStatus = Objects.requireNonNull(initialStatus, "fulfillmentStatus"); this.fulfillmentErrors = copyOrNull(initialErrors); + recordFulfillmentChange(null); } /** @@ -187,9 +206,11 @@ public void transitionFulfillment(FulfillmentStatus to, List errors throw ApiException.conflict("Illegal fulfillment transition " + fulfillmentStatus + " -> " + to + " for exchange " + exchangeId); } + var previous = fulfillmentStatus; this.fulfillmentStatus = to; this.fulfillmentErrors = copyOrNull(errors); releaseLiveKeyIfTerminal(); + recordFulfillmentChange(previous); } /** @@ -221,9 +242,11 @@ public void recordAcceptance(AcceptanceStatus to, List errors) { throw ApiException.conflict("Illegal acceptance transition " + acceptanceStatus + " -> " + to + " for exchange " + exchangeId); } + var previous = acceptanceStatus; this.acceptanceStatus = to; this.acceptanceErrors = copyOrNull(errors); releaseLiveKeyIfTerminal(); + recordAcceptanceChange(previous); } /** Nulls the live-request key once the exchange is no longer live, so the unique constraint stops guarding it. */ @@ -262,6 +285,62 @@ public boolean isLive() { return !fulfillmentStatus.isTerminal() && (acceptanceStatus == null || !acceptanceStatus.isTerminal()); } + // --- domain events ------------------------------------------------------------------------- + + /** + * Records the Fulfillment status now in effect. {@code previous} is null when the exchange was just + * opened, which is how a consumer tells "opened in state X" from "transitioned to X". + */ + private void recordFulfillmentChange(FulfillmentStatus previous) { + domainEvents.add(new CertificateExchangeStatusChanged( + ExchangeRole.PROVIDER, + ExchangePhase.FULFILLMENT, + ExchangeEventType.of(fulfillmentStatus), + exchangeId, + participantContextId, + counterpartyBpn, + counterpartyDid, + certificateId, + certificateId == null ? null : revision, + previous == null ? null : previous.name(), + fulfillmentStatus.name(), + fulfillmentErrors, + OffsetDateTime.now())); + } + + /** Records the Acceptance status now in effect; {@code previous} is null for the first verdict recorded. */ + private void recordAcceptanceChange(AcceptanceStatus previous) { + domainEvents.add(new CertificateExchangeStatusChanged( + ExchangeRole.PROVIDER, + ExchangePhase.ACCEPTANCE, + ExchangeEventType.of(acceptanceStatus), + exchangeId, + participantContextId, + counterpartyBpn, + counterpartyDid, + certificateId, + certificateId == null ? null : revision, + previous == null ? null : previous.name(), + acceptanceStatus.name(), + acceptanceErrors, + OffsetDateTime.now())); + } + + /** + * Drained by Spring Data on {@code save()} and handed to the {@code ApplicationEventPublisher}. The + * NATS publisher listens after commit, so a transaction that rolls back announces nothing. + */ + @DomainEvents + Collection domainEvents() { + return List.copyOf(domainEvents); + } + + /** Clears the recorded events after publication so a second {@code save()} does not re-publish them. */ + @AfterDomainEventPublication + void clearDomainEvents() { + domainEvents.clear(); + } + // --- accessors ----------------------------------------------------------------------------- public String exchangeId() { diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index 9ed49e6..8214574 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -40,6 +40,26 @@ certo: # GET /tokens/{participantContextId}/{flowId}. A siglet is required (point at a mock siglet for dev/test). security: siglet-base-url: http://localhost:8090 + # Certificate-exchange status changes (CX-0135 §2.1.3) published as CloudEvents onto a NATS JetStream + # stream, the way the platform's EDC runtimes publish theirs — subjects events.certificate.exchange.*, + # so a consumer of the shared `edc-events` stream sees them alongside the connector's. OFF by default: + # with `enabled: false` no connection is made and the publisher bean does not exist, so certo runs + # exactly as before unless a deployment opts in. + events: + nats: + enabled: ${CERTO_EVENTS_NATS_ENABLED:false} + url: ${CERTO_EVENTS_NATS_URL:nats://localhost:4222} + stream: ${CERTO_EVENTS_NATS_STREAM:edc-events} + # Standalone development only: in the platform the stream is owned by the nats-bootstrap job and + # certo's NATS user has no stream-management rights. + create-stream: ${CERTO_EVENTS_NATS_CREATESTREAM:false} + # Vault-delivered ed25519 seed, e.g. /vault/secrets/nats.nk. Blank connects unauthenticated. + nkey-seed-path: ${CERTO_EVENTS_NATS_NKEYSEEDPATH:} + # CloudEvents `source`: the emitting APPLICATION, matching the EDC runtimes (whose events-nats + # bridge uses their Hostname service). Blank resolves to $HOSTNAME — the pod name in + # Kubernetes — or the local hostname. Set it for a stable logical name instead of a per-pod + # one. The emitting TENANT is carried by the sourcebpn extension, not here. + source: ${CERTO_EVENTS_NATS_SOURCE:} # Outbound OkHttp client, shared across all adapters (routed through RetryingHttpClient). Timeouts are in # seconds (0 disables a given timeout). Retry (Failsafe, like EDC's EdcHttpClient) retries a transient # IOException or 5xx with exponential backoff. Every outbound call is idempotent so retry is safe: reads; diff --git a/src/test/java/org/metaform/certo/common/event/ExchangeEventTypeTest.java b/src/test/java/org/metaform/certo/common/event/ExchangeEventTypeTest.java new file mode 100644 index 0000000..2fbca34 --- /dev/null +++ b/src/test/java/org/metaform/certo/common/event/ExchangeEventTypeTest.java @@ -0,0 +1,78 @@ +package org.metaform.certo.common.event; + +import org.junit.jupiter.api.Test; +import org.metaform.certo.common.model.AcceptanceStatus; +import org.metaform.certo.common.model.FulfillmentStatus; + +import java.util.Arrays; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Guards the published contract of the event catalogue. These are cheap assertions protecting an + * expensive mistake: a status that maps to nothing publishes nothing, silently, and the gap only + * shows up as a consumer that never fires. + */ +class ExchangeEventTypeTest { + + @Test + void everyFulfillmentStatusHasACatalogueEntry() { + for (var status : FulfillmentStatus.values()) { + assertThat(ExchangeEventType.of(status)) + .as("no event catalogue entry for FulfillmentStatus.%s", status) + .isNotNull() + .extracting(ExchangeEventType::phase) + .isEqualTo(ExchangePhase.FULFILLMENT); + } + } + + @Test + void everyAcceptanceStatusHasACatalogueEntry() { + for (var status : AcceptanceStatus.values()) { + assertThat(ExchangeEventType.of(status)) + .as("no event catalogue entry for AcceptanceStatus.%s", status) + .isNotNull() + .extracting(ExchangeEventType::phase) + .isEqualTo(ExchangePhase.ACCEPTANCE); + } + } + + @Test + void catalogueCoversExactlyTheTwoStatusEnums() { + var statusNames = Stream.concat( + Arrays.stream(FulfillmentStatus.values()).map(Enum::name), + Arrays.stream(AcceptanceStatus.values()).map(Enum::name)) + .toList(); + // Nothing extra either: an entry with no backing status is dead contract surface. + assertThat(Arrays.stream(ExchangeEventType.values()).map(Enum::name)) + .containsExactlyInAnyOrderElementsOf(statusNames); + } + + @Test + void subjectsAreUniqueAndNamespaced() { + var subjects = Arrays.stream(ExchangeEventType.values()).map(ExchangeEventType::subject).toList(); + assertThat(subjects).doesNotHaveDuplicates().allMatch(s -> s.startsWith("events.certificate.exchange.")); + // The events. prefix is what the platform's edc-events stream captures and what the NATS + // permission matrix grants; losing it means the events are published nowhere observable. + assertThat(subjects).allMatch(s -> s.startsWith("events.")); + } + + @Test + void typesAreUniqueAndFollowTheCx0000Convention() { + var types = Arrays.stream(ExchangeEventType.values()).map(ExchangeEventType::type).toList(); + assertThat(types).doesNotHaveDuplicates() + .allMatch(t -> t.startsWith("org.catena-x.ccm.")) + .allMatch(t -> t.endsWith(".v1")); + } + + @Test + void subjectAndTypeAgreeOnTheStatusTheyName() { + assertThat(ExchangeEventType.of(FulfillmentStatus.CERTIFICATION_REQUESTED)) + .returns("events.certificate.exchange.certificationRequested", ExchangeEventType::subject) + .returns("org.catena-x.ccm.CertificateExchangeCertificationRequested.v1", ExchangeEventType::type); + assertThat(ExchangeEventType.of(AcceptanceStatus.REJECTED)) + .returns("events.certificate.exchange.rejected", ExchangeEventType::subject) + .returns("org.catena-x.ccm.CertificateExchangeRejected.v1", ExchangeEventType::type); + } +} diff --git a/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublisherTest.java b/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublisherTest.java new file mode 100644 index 0000000..afda0ac --- /dev/null +++ b/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublisherTest.java @@ -0,0 +1,96 @@ +package org.metaform.certo.common.event.nats; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.metaform.certo.common.cloudevent.CloudEvent; +import org.metaform.certo.common.event.CertificateExchangeStatusChanged; +import org.metaform.certo.common.event.ExchangeEventType; +import org.metaform.certo.common.event.ExchangePhase; +import org.metaform.certo.common.event.ExchangeRole; +import org.metaform.certo.common.model.FulfillmentStatus; +import org.metaform.certo.common.pc.domain.ParticipantContext; +import org.metaform.certo.testsupport.InMemoryParticipantContextStore; + +import java.time.OffsetDateTime; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * The CloudEvents envelope Certo puts on the wire. {@code source} and {@code sourcebpn} come from the + * emitting tenant rather than from the process, so they are worth pinning: CX-0000 §2.1.2 makes + * {@code sourcebpn} REQUIRED, and a consumer that de-duplicates on {@code source}+{@code id} depends + * on both being right. + */ +class NatsEventPublisherTest { + + private InMemoryParticipantContextStore contexts; + private NatsEventPublisher publisher; + + @BeforeEach + void setUp() { + contexts = new InMemoryParticipantContextStore(); + contexts.save(new ParticipantContext("pctx-1", "BPNL0000000001AB", "urn:bpn:BPNL0000000001AB", "did:web:provider")); + // JetStream and the mapper are only exercised by the publish path, not by envelope construction. + publisher = new NatsEventPublisher(null, null, contexts, "certo-7d56645cc7-2bzkl"); + } + + private static CertificateExchangeStatusChanged change(String participantContextId) { + return new CertificateExchangeStatusChanged( + ExchangeRole.PROVIDER, + ExchangePhase.FULFILLMENT, + ExchangeEventType.of(FulfillmentStatus.FULFILLED), + "ex-1", + participantContextId, + "BPNL-CONSUMER", + "did:web:consumer", + "cert-1", + 2, + "ACKNOWLEDGED", + "FULFILLED", + null, + OffsetDateTime.parse("2026-08-12T10:15:30Z")); + } + + @Test + void envelopeCarriesTheTenantIdentityAndCatalogueType() { + CloudEvent envelope = publisher.envelope(change("pctx-1")); + + assertThat(envelope.specVersion()).isEqualTo("1.0"); + assertThat(envelope.type()).isEqualTo("org.catena-x.ccm.CertificateExchangeFulfilled.v1"); + // source is the emitting APPLICATION, as in every other platform producer -- not the tenant. + assertThat(envelope.source()).isEqualTo("certo-7d56645cc7-2bzkl"); + // The tenant is carried by sourcebpn (CX-0000 §2.1.2) and participantContextId in the data. + assertThat(envelope.sourceBpn()).isEqualTo("BPNL0000000001AB"); + assertThat(envelope.dataContentType()).isEqualTo("application/json"); + assertThat(envelope.time()).isEqualTo(OffsetDateTime.parse("2026-08-12T10:15:30Z")); + assertThat(envelope.data().exchangeId()).isEqualTo("ex-1"); + } + + @Test + void subjectAttributeNamesTheExchange() { + // CloudEvents `subject` identifies the thing the event is about; the NATS subject (the routing + // key) is a separate concept and comes from the catalogue. + assertThat(publisher.envelope(change("pctx-1")).subject()).isEqualTo("ex-1"); + } + + @Test + void eachEnvelopeGetsItsOwnId() { + var first = publisher.envelope(change("pctx-1")); + var second = publisher.envelope(change("pctx-1")); + + assertThat(first.id()).isNotBlank().isNotEqualTo(second.id()); + } + + @Test + void aDeletedTenantStillProducesAnIdentifiableEvent() { + // The tenant can be removed between the transition committing and this listener running. + // Dropping the event would lose a committed fact, so it goes out without a sourcebpn; the + // application source is unaffected, and participantContextId still names the tenant. + var envelope = publisher.envelope(change("pctx-gone")); + + assertThat(envelope.source()).isEqualTo("certo-7d56645cc7-2bzkl"); + assertThat(envelope.sourceBpn()).isNull(); + assertThat(envelope.data().participantContextId()).isEqualTo("pctx-gone"); + assertThat(envelope.data().exchangeId()).isEqualTo("ex-1"); + } +} diff --git a/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublishingIntegrationTest.java b/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublishingIntegrationTest.java new file mode 100644 index 0000000..61e2ff4 --- /dev/null +++ b/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublishingIntegrationTest.java @@ -0,0 +1,183 @@ +package org.metaform.certo.common.event.nats; + +import io.nats.client.Connection; +import io.nats.client.JetStreamSubscription; +import io.nats.client.Nats; +import io.nats.client.api.StorageType; +import io.nats.client.api.StreamConfiguration; +import org.junit.jupiter.api.Test; +import org.metaform.certo.common.model.AcceptanceStatus; +import org.metaform.certo.common.model.FulfillmentStatus; +import org.metaform.certo.common.pc.domain.ParticipantContext; +import org.metaform.certo.common.pc.store.ParticipantContextStore; +import org.metaform.certo.provider.domain.ProviderCertificateExchange; +import org.metaform.certo.provider.store.ProviderCertificateExchangeStore; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; +import tools.jackson.databind.JsonNode; +import tools.jackson.databind.ObjectMapper; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * End-to-end proof that a committed status change reaches NATS. + * + *

The path under test spans four things a unit test cannot join up: the aggregate recording the + * change, Spring Data draining it on {@code save()}, the transaction manager releasing it after + * commit, and the JetStream client putting it on a subject the stream actually captures. It is also + * the only test that would catch a subject the stream does not match — a publish that silently goes + * nowhere. + * + *

Requires Docker. Testcontainers aborts (not fails) the class when none is available, so a + * Docker-less machine still gets a green build. + */ +@SpringBootTest +@Testcontainers +class NatsEventPublishingIntegrationTest { + + private static final String STREAM = "edc-events"; + private static final String SUBJECT_FILTER = "events.certificate.exchange.>"; + + @Container + @SuppressWarnings("resource") + static final GenericContainer NATS = new GenericContainer<>(DockerImageName.parse("nats:alpine")) + .withCommand("-js") + .withExposedPorts(4222) + .waitingFor(Wait.forLogMessage(".*Server is ready.*", 1)); + + @DynamicPropertySource + static void natsProperties(DynamicPropertyRegistry registry) { + registry.add("certo.events.nats.enabled", () -> true); + registry.add("certo.events.nats.url", () -> "nats://%s:%d".formatted(NATS.getHost(), NATS.getMappedPort(4222))); + registry.add("certo.events.nats.stream", () -> STREAM); + // The platform's nats-bootstrap job owns this stream in a real deployment; here the app creates it. + registry.add("certo.events.nats.create-stream", () -> true); + // Pin the CloudEvents source so it can be asserted; unset it resolves to this machine's hostname. + registry.add("certo.events.nats.source", () -> "certo-under-test"); + } + + @Autowired + private ProviderCertificateExchangeStore exchangeStore; + @Autowired + private ParticipantContextStore contextStore; + @Autowired + private ObjectMapper mapper; + @Autowired + private PlatformTransactionManager transactionManager; + + @Test + void aCommittedExchangeLifecycleIsPublishedInOrder() throws Exception { + var tenantId = "pctx-" + UUID.randomUUID(); + contextStore.save(new ParticipantContext(tenantId, "BPNL0000000009ZZ", "urn:bpn:BPNL0000000009ZZ", + "did:web:events-test-" + UUID.randomUUID())); + + try (var connection = Nats.connect("nats://%s:%d".formatted(NATS.getHost(), NATS.getMappedPort(4222)))) { + // Subscribe before producing: the stream retains on Interest policy, so a message published + // with no registered consumer is discarded rather than queued. + ensureStream(connection); + var subscription = connection.jetStream().subscribe(SUBJECT_FILTER); + + var exchangeId = "ex-" + UUID.randomUUID(); + var exchange = new ProviderCertificateExchange(exchangeId, tenantId, "cert-1", 1, + "BPNL-CONSUMER", "did:web:consumer", FulfillmentStatus.REQUESTED); + exchangeStore.save(exchange); + + var loaded = exchangeStore.findById(exchangeId).orElseThrow(); + loaded.transitionFulfillment(FulfillmentStatus.ACKNOWLEDGED, null); + exchangeStore.save(loaded); + + loaded = exchangeStore.findById(exchangeId).orElseThrow(); + loaded.transitionFulfillment(FulfillmentStatus.FULFILLED, null); + loaded.recordAcceptance(AcceptanceStatus.ACCEPTED, null); + exchangeStore.save(loaded); + + var received = new ArrayList(); + for (var i = 0; i < 4; i++) { + var message = subscription.nextMessage(Duration.ofSeconds(10)); + assertThat(message).as("expected 4 events, got %d", received.size()).isNotNull(); + message.ack(); + received.add(mapper.readTree(message.getData())); + } + + assertThat(received).extracting(node -> node.get("type").asString()).containsExactly( + "org.catena-x.ccm.CertificateExchangeRequested.v1", + "org.catena-x.ccm.CertificateExchangeAcknowledged.v1", + "org.catena-x.ccm.CertificateExchangeFulfilled.v1", + "org.catena-x.ccm.CertificateExchangeAccepted.v1"); + + // source names the emitting application (like every other platform producer); the tenant + // travels in sourcebpn and in the payload. + var first = received.getFirst(); + assertThat(first.get("source").asString()).isEqualTo("certo-under-test"); + assertThat(first.get("sourcebpn").asString()).isEqualTo("BPNL0000000009ZZ"); + assertThat(first.get("specversion").asString()).isEqualTo("1.0"); + assertThat(first.get("data").get("exchangeId").asString()).isEqualTo(exchangeId); + assertThat(first.get("data").get("previousStatus")).isNull(); + assertThat(received.get(1).get("data").get("previousStatus").asString()).isEqualTo("REQUESTED"); + assertThat(received.getLast().get("data").get("role").asString()).isEqualTo("PROVIDER"); + assertThat(received.getLast().get("data").get("phase").asString()).isEqualTo("ACCEPTANCE"); + + assertThat(subscription.nextMessage(Duration.ofMillis(500))) + .as("no further events expected") + .isNull(); + } + } + + @Test + void rolledBackWorkPublishesNothing() throws Exception { + try (var connection = Nats.connect("nats://%s:%d".formatted(NATS.getHost(), NATS.getMappedPort(4222)))) { + ensureStream(connection); + var subscription = connection.jetStream().subscribe(SUBJECT_FILTER); + drain(subscription); + + // Save inside a transaction that then rolls back. The aggregate records the change and + // Spring Data drains it, so the event IS raised — after-commit delivery is the only thing + // standing between a rolled-back transition and a consumer being told it happened. + new TransactionTemplate(transactionManager).execute(status -> { + exchangeStore.save(new ProviderCertificateExchange("ex-" + UUID.randomUUID(), "pctx-rollback", + "cert-1", 1, "BPNL-CONSUMER", "did:web:consumer", FulfillmentStatus.REQUESTED)); + status.setRollbackOnly(); + return null; + }); + + assertThat(subscription.nextMessage(Duration.ofSeconds(1))) + .as("a rolled-back transition must not be published") + .isNull(); + } + } + + /** Creates the stream if the app has not yet (bean init order is not guaranteed relative to the test). */ + private static void ensureStream(Connection connection) throws Exception { + var jsm = connection.jetStreamManagement(); + var existing = jsm.getStreamNames(); + if (!existing.contains(STREAM)) { + jsm.addStream(StreamConfiguration.builder() + .name(STREAM) + .subjects(SUBJECT_FILTER) + .storageType(StorageType.Memory) + .build()); + } + } + + /** Consumes anything left over from an earlier test so this one starts from a quiet subscription. */ + private static void drain(JetStreamSubscription subscription) throws Exception { + for (var message = subscription.nextMessage(Duration.ofMillis(200)); + message != null; + message = subscription.nextMessage(Duration.ofMillis(200))) { + message.ack(); + } + } +} diff --git a/src/test/java/org/metaform/certo/consumer/domain/ConsumerCertificateExchangeEventsTest.java b/src/test/java/org/metaform/certo/consumer/domain/ConsumerCertificateExchangeEventsTest.java new file mode 100644 index 0000000..7e5f2c3 --- /dev/null +++ b/src/test/java/org/metaform/certo/consumer/domain/ConsumerCertificateExchangeEventsTest.java @@ -0,0 +1,97 @@ +package org.metaform.certo.consumer.domain; + +import org.junit.jupiter.api.Test; +import org.metaform.certo.common.event.CertificateExchangeStatusChanged; +import org.metaform.certo.common.event.ExchangePhase; +import org.metaform.certo.common.event.ExchangeRole; +import org.metaform.certo.common.model.AcceptanceStatus; +import org.metaform.certo.common.model.FulfillmentStatus; +import org.metaform.certo.common.model.StatusError; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * The consumer aggregate mirrors the provider-owned Fulfillment status and owns the Acceptance + * verdict. Its Fulfillment mirror is written on every poll, so the interesting property here is that + * an unchanged status records nothing. + */ +class ConsumerCertificateExchangeEventsTest { + + private static ConsumerCertificateExchange newExchange(FulfillmentStatus initial) { + return new ConsumerCertificateExchange("ex-1", "cert-1", 3, true, initial, null, + "pctx-consumer", "BPNL-PROVIDER", "did:web:provider"); + } + + private static List eventsOf(ConsumerCertificateExchange exchange) { + return exchange.domainEvents().stream().map(CertificateExchangeStatusChanged.class::cast).toList(); + } + + @Test + void openingRecordsTheInitialStatusAsTheConsumerSide() { + assertThat(eventsOf(newExchange(FulfillmentStatus.REQUESTED))).singleElement().satisfies(event -> { + assertThat(event.role()).isEqualTo(ExchangeRole.CONSUMER); + assertThat(event.phase()).isEqualTo(ExchangePhase.FULFILLMENT); + assertThat(event.status()).isEqualTo("REQUESTED"); + assertThat(event.previousStatus()).isNull(); + // From the consumer's vantage point the counterparty is the provider. + assertThat(event.counterpartyBpn()).isEqualTo("BPNL-PROVIDER"); + assertThat(event.counterpartyDid()).isEqualTo("did:web:provider"); + }); + } + + @Test + void mirroringAnUnchangedStatusRecordsNothing() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + exchange.clearDomainEvents(); + + // pollRequest() calls updateFulfillment on every poll, whether or not the provider moved. + exchange.updateFulfillment(FulfillmentStatus.REQUESTED, null, null); + exchange.updateFulfillment(FulfillmentStatus.REQUESTED, null, null); + + assertThat(eventsOf(exchange)) + .as("a poll that reports no change must not publish an event") + .isEmpty(); + } + + @Test + void mirroringARealChangeRecordsIt() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + exchange.clearDomainEvents(); + + exchange.updateFulfillment(FulfillmentStatus.FULFILLED, "cert-7", null); + + assertThat(eventsOf(exchange)).singleElement().satisfies(event -> { + assertThat(event.previousStatus()).isEqualTo("REQUESTED"); + assertThat(event.status()).isEqualTo("FULFILLED"); + // The certificate identity the provider has just disclosed rides along. + assertThat(event.certificateId()).isEqualTo("cert-7"); + }); + } + + @Test + void acceptanceVerdictIsRecorded() { + var exchange = newExchange(FulfillmentStatus.FULFILLED); + exchange.clearDomainEvents(); + + exchange.transitionAcceptance(AcceptanceStatus.ERRORED, List.of(new StatusError("bad signature"))); + + assertThat(eventsOf(exchange)).singleElement().satisfies(event -> { + assertThat(event.phase()).isEqualTo(ExchangePhase.ACCEPTANCE); + assertThat(event.status()).isEqualTo("ERRORED"); + assertThat(event.errors()).containsExactly(new StatusError("bad signature")); + assertThat(event.eventType().subject()).isEqualTo("events.certificate.exchange.errored"); + }); + } + + @Test + void clearingPreventsRepublicationOnASecondSave() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + assertThat(exchange.domainEvents()).isNotEmpty(); + + exchange.clearDomainEvents(); + + assertThat(exchange.domainEvents()).isEmpty(); + } +} diff --git a/src/test/java/org/metaform/certo/provider/domain/ProviderCertificateExchangeEventsTest.java b/src/test/java/org/metaform/certo/provider/domain/ProviderCertificateExchangeEventsTest.java new file mode 100644 index 0000000..5b0ba0e --- /dev/null +++ b/src/test/java/org/metaform/certo/provider/domain/ProviderCertificateExchangeEventsTest.java @@ -0,0 +1,157 @@ +package org.metaform.certo.provider.domain; + +import org.junit.jupiter.api.Test; +import org.metaform.certo.common.event.CertificateExchangeStatusChanged; +import org.metaform.certo.common.event.ExchangePhase; +import org.metaform.certo.common.event.ExchangeRole; +import org.metaform.certo.common.model.AcceptanceStatus; +import org.metaform.certo.common.model.FulfillmentStatus; +import org.metaform.certo.common.model.StatusError; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * The provider aggregate records a status change for every mutation, which Spring Data drains on + * {@code save()}. These tests pin that down at the source — if a transition stops recording, no + * amount of correct plumbing downstream will publish it. + */ +class ProviderCertificateExchangeEventsTest { + + private static ProviderCertificateExchange newExchange(FulfillmentStatus initial) { + return new ProviderCertificateExchange("ex-1", "pctx-1", "cert-1", 3, "BPNL-CONSUMER", "did:web:consumer", initial); + } + + private static List eventsOf(ProviderCertificateExchange exchange) { + return exchange.domainEvents().stream().map(CertificateExchangeStatusChanged.class::cast).toList(); + } + + @Test + void openingRecordsTheInitialStatusWithNoPredecessor() { + var events = eventsOf(newExchange(FulfillmentStatus.REQUESTED)); + + assertThat(events).singleElement().satisfies(event -> { + assertThat(event.role()).isEqualTo(ExchangeRole.PROVIDER); + assertThat(event.phase()).isEqualTo(ExchangePhase.FULFILLMENT); + assertThat(event.status()).isEqualTo("REQUESTED"); + // The distinguishing mark of an opened exchange versus a transition into the same state. + assertThat(event.previousStatus()).isNull(); + assertThat(event.exchangeId()).isEqualTo("ex-1"); + assertThat(event.participantContextId()).isEqualTo("pctx-1"); + assertThat(event.counterpartyBpn()).isEqualTo("BPNL-CONSUMER"); + assertThat(event.counterpartyDid()).isEqualTo("did:web:consumer"); + assertThat(event.occurredAt()).isNotNull(); + }); + } + + @Test + void transitionRecordsBothEnds() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + exchange.clearDomainEvents(); + + exchange.transitionFulfillment(FulfillmentStatus.ACKNOWLEDGED, null); + + assertThat(eventsOf(exchange)).singleElement().satisfies(event -> { + assertThat(event.previousStatus()).isEqualTo("REQUESTED"); + assertThat(event.status()).isEqualTo("ACKNOWLEDGED"); + assertThat(event.eventType().subject()).isEqualTo("events.certificate.exchange.acknowledged"); + }); + } + + @Test + void terminalTransitionCarriesItsErrors() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + exchange.clearDomainEvents(); + + exchange.transitionFulfillment(FulfillmentStatus.DECLINED, List.of(new StatusError("not our site"))); + + assertThat(eventsOf(exchange)).singleElement().satisfies(event -> + assertThat(event.errors()).containsExactly(new StatusError("not our site"))); + } + + @Test + void fulfillRecordsOneEventCarryingTheNewlyBoundCertificate() { + var exchange = ProviderCertificateExchange.pending( + "ex-2", "pctx-1", "BPNL-CONSUMER", "did:web:consumer", "ISO9001", List.of("BPNS-1"), null); + exchange.clearDomainEvents(); + + exchange.fulfill("cert-9", 2); + + // fulfill() delegates to transitionFulfillment, so exactly one event — not one per mutator. + assertThat(eventsOf(exchange)).singleElement().satisfies(event -> { + assertThat(event.status()).isEqualTo("FULFILLED"); + assertThat(event.previousStatus()).isEqualTo("CERTIFICATION_REQUESTED"); + assertThat(event.certificateId()).isEqualTo("cert-9"); + assertThat(event.revision()).isEqualTo(2); + }); + } + + @Test + void pendingExchangeReportsNoCertificateIdentityYet() { + var exchange = ProviderCertificateExchange.pending( + "ex-3", "pctx-1", "BPNL-CONSUMER", "did:web:consumer", "ISO9001", List.of("BPNS-1"), null); + + assertThat(eventsOf(exchange)).singleElement().satisfies(event -> { + assertThat(event.status()).isEqualTo("CERTIFICATION_REQUESTED"); + assertThat(event.certificateId()).isNull(); + // Revision is meaningless without a certificate; it must not leak a default 0. + assertThat(event.revision()).isNull(); + }); + } + + @Test + void acceptanceIsRecordedOnTheAcceptancePhase() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + exchange.transitionFulfillment(FulfillmentStatus.ACKNOWLEDGED, null); + exchange.transitionFulfillment(FulfillmentStatus.FULFILLED, null); + exchange.clearDomainEvents(); + + exchange.recordAcceptance(AcceptanceStatus.ACCEPTED, null); + + assertThat(eventsOf(exchange)).singleElement().satisfies(event -> { + assertThat(event.phase()).isEqualTo(ExchangePhase.ACCEPTANCE); + assertThat(event.status()).isEqualTo("ACCEPTED"); + // First verdict recorded: no acceptance status preceded it. + assertThat(event.previousStatus()).isNull(); + assertThat(event.eventType().subject()).isEqualTo("events.certificate.exchange.accepted"); + }); + } + + @Test + void acceptanceTransitionCarriesThePreviousVerdict() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + exchange.transitionFulfillment(FulfillmentStatus.ACKNOWLEDGED, null); + exchange.transitionFulfillment(FulfillmentStatus.FULFILLED, null); + exchange.recordAcceptance(AcceptanceStatus.RETRIEVED, null); + exchange.clearDomainEvents(); + + exchange.recordAcceptance(AcceptanceStatus.REJECTED, List.of(new StatusError("wrong scope"))); + + assertThat(eventsOf(exchange)).singleElement().satisfies(event -> { + assertThat(event.previousStatus()).isEqualTo("RETRIEVED"); + assertThat(event.status()).isEqualTo("REJECTED"); + }); + } + + @Test + void everyTransitionAccumulatesUntilDrained() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + exchange.transitionFulfillment(FulfillmentStatus.ACKNOWLEDGED, null); + exchange.transitionFulfillment(FulfillmentStatus.FULFILLED, null); + + // A single save() must publish the whole trail, not just the last hop. + assertThat(eventsOf(exchange)).extracting(CertificateExchangeStatusChanged::status) + .containsExactly("REQUESTED", "ACKNOWLEDGED", "FULFILLED"); + } + + @Test + void clearingPreventsRepublicationOnASecondSave() { + var exchange = newExchange(FulfillmentStatus.REQUESTED); + assertThat(exchange.domainEvents()).isNotEmpty(); + + exchange.clearDomainEvents(); + + assertThat(exchange.domainEvents()).isEmpty(); + } +} From 4cbc07da3bed5bb43f33d767a9fa5196dc7193fe Mon Sep 17 00:00:00 2001 From: Paul Latzelsperger Date: Thu, 13 Aug 2026 08:56:02 +0200 Subject: [PATCH 2/3] add test --- gradle/wrapper/gradle-wrapper.properties | 2 +- .../NatsEventPublishingIntegrationTest.java | 99 +++++++++++++++++++ 2 files changed, 100 insertions(+), 1 deletion(-) diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 874636a..bb5d67e 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,6 +1,6 @@ #Thu Jun 04 16:10:23 CEST 2026 distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -distributionUrl=https\://services.gradle.org/distributions/gradle-9.5.1-bin.zip +distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists diff --git a/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublishingIntegrationTest.java b/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublishingIntegrationTest.java index 61e2ff4..0d9dfb9 100644 --- a/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublishingIntegrationTest.java +++ b/src/test/java/org/metaform/certo/common/event/nats/NatsEventPublishingIntegrationTest.java @@ -8,6 +8,7 @@ import org.junit.jupiter.api.Test; import org.metaform.certo.common.model.AcceptanceStatus; import org.metaform.certo.common.model.FulfillmentStatus; +import org.metaform.certo.common.model.StatusError; import org.metaform.certo.common.pc.domain.ParticipantContext; import org.metaform.certo.common.pc.store.ParticipantContextStore; import org.metaform.certo.provider.domain.ProviderCertificateExchange; @@ -27,7 +28,9 @@ import tools.jackson.databind.ObjectMapper; import java.time.Duration; +import java.time.OffsetDateTime; import java.util.ArrayList; +import java.util.List; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; @@ -136,6 +139,102 @@ void aCommittedExchangeLifecycleIsPublishedInOrder() throws Exception { } } + /** + * The consumer-initiated pull, which is the only flow that reaches the EARLY Fulfillment states. + * + *

The provider-initiated publish covered above constructs its exchange directly in + * {@code FULFILLED} ({@code ProviderExchangeService.publish}), so it never holds {@code REQUESTED}, + * {@code ACKNOWLEDGED} or {@code CERTIFICATION_REQUESTED} and correctly never announces them. A + * consumer asking for a certificate the provider does not yet hold opens a + * {@link ProviderCertificateExchange#pending} exchange instead, which walks the state machine + * properly — and that walk is what this asserts. + * + *

It also exercises {@code RETRIEVED}, the optional non-terminal acceptance status + * (CX-0135 §2.1.3). Note no certo code path sets it today — {@code ConsumerExchangeService.retrieve} + * fetches the certificate without recording acceptance — so it only occurs when a counterparty + * explicitly reports it. This test is the standing proof that the event works when the status does + * occur, rather than being a catalogue entry nothing can ever reach. + */ + @Test + void aConsumerInitiatedPullPublishesTheEarlyStates() throws Exception { + var tenantId = "pctx-" + UUID.randomUUID(); + contextStore.save(new ParticipantContext(tenantId, "BPNL0000000008YY", "urn:bpn:BPNL0000000008YY", + "did:web:events-pull-" + UUID.randomUUID())); + + try (var connection = Nats.connect("nats://%s:%d".formatted(NATS.getHost(), NATS.getMappedPort(4222)))) { + ensureStream(connection); + var subscription = connection.jetStream().subscribe(SUBJECT_FILTER); + drain(subscription); + + // The consumer asked for a certificate the provider does not hold yet: no certificate + // identity, and the request retained so a later issuance can be matched to it. + var exchangeId = "ex-" + UUID.randomUUID(); + var pending = ProviderCertificateExchange.pending(exchangeId, tenantId, + "BPNL-CONSUMER", "did:web:consumer", "ISO9001", List.of("BPNS-SITE-1"), + OffsetDateTime.parse("2026-08-12T09:00:00Z")); + exchangeStore.save(pending); + + // The certification authority issues; the certificate identity is bound at fulfil time. + var loaded = exchangeStore.findById(exchangeId).orElseThrow(); + loaded.fulfill("cert-pull-1", 4); + exchangeStore.save(loaded); + + // The consumer reports the optional RETRIEVED hop, then its terminal verdict. + loaded = exchangeStore.findById(exchangeId).orElseThrow(); + loaded.recordAcceptance(AcceptanceStatus.RETRIEVED, null); + exchangeStore.save(loaded); + + loaded = exchangeStore.findById(exchangeId).orElseThrow(); + loaded.recordAcceptance(AcceptanceStatus.REJECTED, List.of(new StatusError("scope does not cover BPNS-SITE-1"))); + exchangeStore.save(loaded); + + var received = new ArrayList(); + for (var i = 0; i < 4; i++) { + var message = subscription.nextMessage(Duration.ofSeconds(10)); + assertThat(message).as("expected 4 events, got %d", received.size()).isNotNull(); + message.ack(); + received.add(mapper.readTree(message.getData())); + } + + assertThat(received).extracting(node -> node.get("subject").asString()) + .as("every event belongs to the exchange under test") + .containsOnly(exchangeId); + + assertThat(received).extracting(node -> node.get("type").asString()).containsExactly( + "org.catena-x.ccm.CertificateExchangeCertificationRequested.v1", + "org.catena-x.ccm.CertificateExchangeFulfilled.v1", + "org.catena-x.ccm.CertificateExchangeRetrieved.v1", + "org.catena-x.ccm.CertificateExchangeRejected.v1"); + + // Opened, not transitioned — and with no certificate identity to report yet. + var opened = received.getFirst().get("data"); + assertThat(opened.get("previousStatus")).isNull(); + assertThat(opened.get("status").asString()).isEqualTo("CERTIFICATION_REQUESTED"); + assertThat(opened.get("certificateId")).isNull(); + assertThat(opened.get("revision")).isNull(); + + // Fulfilment binds the certificate the backend issued. + var fulfilled = received.get(1).get("data"); + assertThat(fulfilled.get("previousStatus").asString()).isEqualTo("CERTIFICATION_REQUESTED"); + assertThat(fulfilled.get("certificateId").asString()).isEqualTo("cert-pull-1"); + assertThat(fulfilled.get("revision").asInt()).isEqualTo(4); + + // The acceptance phase carries its own predecessor chain, independent of fulfilment. + var retrieved = received.get(2).get("data"); + assertThat(retrieved.get("phase").asString()).isEqualTo("ACCEPTANCE"); + assertThat(retrieved.get("previousStatus")).isNull(); + + var rejected = received.getLast().get("data"); + assertThat(rejected.get("previousStatus").asString()).isEqualTo("RETRIEVED"); + assertThat(rejected.get("errors").get(0).get("message").asString()) + .isEqualTo("scope does not cover BPNS-SITE-1"); + + assertThat(subscription.nextMessage(Duration.ofMillis(500))) + .as("no further events expected") + .isNull(); + } + } + @Test void rolledBackWorkPublishesNothing() throws Exception { try (var connection = Nats.connect("nats://%s:%d".formatted(NATS.getHost(), NATS.getMappedPort(4222)))) { From 5c8877803d8e30a7418a08a7952e33fc4ddbfb36 Mon Sep 17 00:00:00 2001 From: Paul Latzelsperger Date: Thu, 13 Aug 2026 09:05:13 +0200 Subject: [PATCH 3/3] trigger ci --- .github/workflows/build-and-test.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml index 6a657a0..5f836e6 100644 --- a/.github/workflows/build-and-test.yml +++ b/.github/workflows/build-and-test.yml @@ -2,6 +2,7 @@ name: Build and test on: push: + pull_request: permissions: contents: read