feat(events): publish certificate-exchange status changes to NATS

.github/workflows/build-and-test.yml

@@ -2,6 +2,7 @@ name: Build and test
 
 on:
   push:
+  pull_request: 
 
 permissions:
   contents: read

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")
 }
 

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 }}

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

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

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.
+ *
+ * <p>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 &sect;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 <b>null when the exchange was just opened</b> —
+ *                             which is what distinguishes "opened in state X" from "transitioned to X"
+ * @param status               the status now in effect
+ * @param errors               CX-0135 &sect;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<StatusError> errors,
+        OffsetDateTime occurredAt) {
+
+    public CertificateExchangeStatusChanged {
+        errors = errors == null ? null : List.copyOf(errors);
+    }
+}

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.
+ *
+ * <p>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.
+ *
+ * <p>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 &sect;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<FulfillmentStatus, ExchangeEventType> BY_FULFILLMENT = new EnumMap<>(FulfillmentStatus.class);
+    private static final Map<AcceptanceStatus, ExchangeEventType> 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 &sect;2.3 reverse-DNS). */
+    public String type() {
+        return type;
+    }
+}

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
+}

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.
+ *
+ * <p>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
+}