Send rejection response for out-of-scope GOPACS flex messages

ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java

@@ -44,9 +44,11 @@
 import org.lfenergy.shapeshifter.core.service.handler.UftpPayloadHandler;
 import org.lfenergy.shapeshifter.core.service.participant.ParticipantResolutionService;
 import org.lfenergy.shapeshifter.core.service.receiving.UftpReceivedMessageService;
+import org.lfenergy.shapeshifter.core.service.receiving.response.UftpValidationResponseCreator;
 import org.lfenergy.shapeshifter.core.service.sending.UftpSendMessageService;
 import org.lfenergy.shapeshifter.core.service.serialization.UftpSerializer;
 import org.lfenergy.shapeshifter.core.service.validation.UftpValidationService;
+import org.lfenergy.shapeshifter.core.service.validation.model.ValidationResult;
 import org.openremote.container.timer.TimerService;
 import org.openremote.container.web.CORSConfig;
 import org.openremote.container.web.WebApplication;
@@ -551,8 +553,9 @@ protected long calculatePower(FlexRequestISPType requestIsp) {
 
     /**
      * Only act on flex messages whose congestion point matches this handler's contracted EAN. Returns
-     * false (and logs a warning) for out-of-scope messages so they are dropped before any asset mutation
-     * or outbound response. See issue #28 for full per-contract/role scoping via the V3 contracts endpoint.
+     * false (and logs a warning) for out-of-scope messages so they are rejected (with a rejection
+     * response, but without asset mutation or FlexOffer). See issue #28 for full per-contract/role
+     * scoping via the V3 contracts endpoint.
      */
     protected boolean isWithinContractedScope(String messageType, String conversationId, String congestionPoint) {
         if (Objects.equals(toCongestionPoint(contractedEAN), toCongestionPoint(congestionPoint))) {
@@ -594,11 +597,20 @@ protected void processRawMessage(String transportXml) {
             // resolves any participant domain, so a validly signed flex message from a participant outside
             // this handler's contracted EAN would otherwise be applied to the asset. Full per-contract/role
             // scoping via the V3 contracts endpoint is tracked in issue #28.
-            // Out-of-scope messages are intentionally dropped here: the transport call still returns 200
-            // (signed envelope accepted) but no FlexRequestResponse/FlexOffer is sent in reply.
+            // Out-of-scope messages are not applied to the asset and get no FlexOffer, but UFTP still
+            // requires a response: reject with a reason instead of leaving the DSO's conversation dangling.
             if (payloadMessage instanceof FlexMessageType flexMessage
                     && !isWithinContractedScope(payloadMessage.getClass().getSimpleName(),
                             payloadMessage.getConversationID(), flexMessage.getCongestionPoint())) {
+                PayloadMessageType rejection = UftpValidationResponseCreator.getResponseForMessage(
+                        payloadMessage, ValidationResult.rejection("CongestionPoint not within contracted scope"));
+                UftpParticipant sender = new UftpParticipant(signedMessage);
+                UftpParticipant responder = new UftpParticipant(payloadMessage.getRecipientDomain(),
+                        UftpRoleInformation.getRecipientRoleBySenderRole(sender.role()));
+                // Send delayed so the HTTP 200 on the transport call goes out first, like the accepted path
+                scheduledExecutorService.schedule(() ->
+                        notifyNewOutgoingMessage(OutgoingUftpMessage.create(responder, rejection)),
+                        this.responseDelaySeconds, TimeUnit.SECONDS);
                 return;
             }
 

ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy

@@ -224,18 +224,33 @@ class GOPACSHandlerHttpTest extends Specification implements ManagerContainerTra
         }
     }
 
-    def "a validly-signed FlexRequest for a different congestion point is accepted but produces no broker delivery"() {
+    def "a validly-signed FlexRequest for a different congestion point is rejected via the broker without a FlexOffer"() {
+        given: "polling for the asynchronous outbound delivery"
+        def conditions = new PollingConditions(timeout: 10, delay: 0.2)
+
         when: "a signed out-of-scope FlexRequest (different EAN) is POSTed"
         Response response = postSignedAsDso(flexRequestXml("ean.999999999999999999"))
 
         then: "the transport call still succeeds (the signed envelope is accepted)"
         response.statusInfo.family == Response.Status.Family.SUCCESSFUL
         response.close()
 
-        and: "no message is ever delivered to the broker"
-        // The drop happens before any reply is scheduled; allow the (zero-delay) scheduler to run first.
+        and: "a rejected FlexRequestResponse with a reason is delivered to the broker as signed XML"
+        conditions.eventually {
+            wireMock.verify(postRequestedFor(urlPathEqualTo(BROKER_PATH))
+                    .withHeader("Authorization", equalTo("Bearer " + ACCESS_TOKEN))
+                    .withRequestBody(containing("SignedMessage")))
+            assert decodedBrokerPayloads().any {
+                it.contains("FlexRequestResponse") && it.contains('Result="Rejected"') && it.contains("RejectionReason=")
+            }
+        }
+
+        and: "no FlexOffer is ever delivered"
+        // Replies are scheduled with zero delay in this test; give any (erroneous) FlexOffer time to arrive.
         Thread.sleep(1000)
-        wireMock.verify(0, postRequestedFor(urlPathEqualTo(BROKER_PATH)))
+        def payloads = decodedBrokerPayloads()
+        payloads.size() == 1
+        !payloads.any { it.contains("FlexOffer") }
     }
 
     def "malformed transport XML is rejected with 400"() {

ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy

@@ -183,14 +183,41 @@ class GOPACSHandlerTest extends Specification {
         ((FlexOrderResponse) handler.sent[0]).result == AcceptedRejectedType.ACCEPTED
     }
 
-    def "a validly-signed FlexRequest for a different congestion point is dropped with no mutation and no reply"() {
+    def "a validly-signed FlexRequest for a different congestion point is rejected without mutation or FlexOffer"() {
         when: "a signed out-of-scope FlexRequest (different EAN) is processed"
         signAndProcess(flexRequestXml("ean.999999999999999999"))
 
-        then: "the asset is not mutated and nothing is sent back"
+        then: "the asset is not mutated"
         0 * assetPredictedDatapointService.updateValues(_, _, _)
         0 * assetProcessingService.sendAttributeEvent(_, _)
-        handler.sent.isEmpty()
+
+        and: "only a rejected FlexRequestResponse is sent back, no FlexOffer"
+        handler.sent.size() == 1
+        handler.sent[0] instanceof FlexRequestResponse
+        def response = (FlexRequestResponse) handler.sent[0]
+        response.result == AcceptedRejectedType.REJECTED
+        !response.rejectionReason.isBlank()
+        response.flexRequestMessageID == FLEX_REQUEST_MESSAGE_ID
+        response.conversationID == CONVERSATION_ID
+        response.senderDomain == AGR_DOMAIN
+        response.recipientDomain == DSO_DOMAIN
+    }
+
+    def "a validly-signed FlexOrder for a different congestion point is rejected without mutation"() {
+        when: "a signed out-of-scope FlexOrder (different EAN) is processed"
+        signAndProcess(flexOrderXml("ean.999999999999999999", [4000, 8000]))
+
+        then: "the asset is not mutated"
+        0 * assetPredictedDatapointService.updateValues(_, _, _)
+        0 * assetProcessingService.sendAttributeEvent(_, _)
+
+        and: "only a rejected FlexOrderResponse is sent back"
+        handler.sent.size() == 1
+        handler.sent[0] instanceof FlexOrderResponse
+        def response = (FlexOrderResponse) handler.sent[0]
+        response.result == AcceptedRejectedType.REJECTED
+        !response.rejectionReason.isBlank()
+        response.conversationID == CONVERSATION_ID
     }
 
     def "toCongestionPoint canonicalises an EAN to the GOPACS ean.<code> format (#input -> #expected)"() {