Refactor: replace policy with policyId on ContractAgreement

CHANGELOG.md

@@ -31,6 +31,7 @@ in the detailed section referring to by linking pull requests or issues.
 * Extract single `PolicyArchive` implementation (#1158)
 * Replace `accessPolicy` and `contractPolicy` with `accessPolicyId` and `contractPolicyId` on `ContractDefinition` (#1144)
 * All DMgmt Api methods now produce and consume `APPLICATION_JSON` (#1175)
+* Replace `policy` with `policyId` on `ContractAgreement` (#1220)
 
 #### Removed
 

core/contract/src/main/java/org/eclipse/dataspaceconnector/contract/ContractServiceExtension.java

@@ -146,7 +146,7 @@ private void registerServices(ServiceExtensionContext context) {
         var observable = new ContractNegotiationObservableImpl();
         context.registerService(ContractNegotiationObservable.class, observable);
 
-        context.registerService(PolicyArchive.class, new PolicyArchiveImpl(store));
+        context.registerService(PolicyArchive.class, new PolicyArchiveImpl(store, policyStore));
 
         consumerNegotiationManager = ConsumerContractNegotiationManagerImpl.Builder.newInstance()
                 .waitStrategy(waitStrategy)
@@ -159,6 +159,7 @@ private void registerServices(ServiceExtensionContext context) {
                 .telemetry(telemetry)
                 .executorInstrumentation(context.getService(ExecutorInstrumentation.class))
                 .store(store)
+                .policyStore(policyStore)
                 .batchSize(context.getSetting(NEGOTIATION_CONSUMER_STATE_MACHINE_BATCH_SIZE, 5))
                 .build();
 
@@ -173,6 +174,7 @@ private void registerServices(ServiceExtensionContext context) {
                 .telemetry(telemetry)
                 .executorInstrumentation(context.getService(ExecutorInstrumentation.class))
                 .store(store)
+                .policyStore(policyStore)
                 .batchSize(context.getSetting(NEGOTIATION_PROVIDER_STATE_MACHINE_BATCH_SIZE, 5))
                 .build();
 

core/contract/src/main/java/org/eclipse/dataspaceconnector/contract/negotiation/AbstractContractNegotiationManager.java

@@ -22,6 +22,7 @@
 import org.eclipse.dataspaceconnector.spi.contract.validation.ContractValidationService;
 import org.eclipse.dataspaceconnector.spi.message.RemoteMessageDispatcherRegistry;
 import org.eclipse.dataspaceconnector.spi.monitor.Monitor;
+import org.eclipse.dataspaceconnector.spi.policy.store.PolicyStore;
 import org.eclipse.dataspaceconnector.spi.retry.WaitStrategy;
 import org.eclipse.dataspaceconnector.spi.system.ExecutorInstrumentation;
 import org.eclipse.dataspaceconnector.spi.telemetry.Telemetry;
@@ -43,6 +44,7 @@ public abstract class AbstractContractNegotiationManager {
     protected ExecutorInstrumentation executorInstrumentation;
     protected int batchSize = 5;
     protected WaitStrategy waitStrategy = () -> 5000L;  // default wait five seconds
+    protected PolicyStore policyStore;
 
     public static class Builder<T extends AbstractContractNegotiationManager> {
 
@@ -109,6 +111,11 @@ public Builder<T> store(ContractNegotiationStore store) {
             return this;
         }
 
+        public Builder<T> policyStore(PolicyStore policyStore) {
+            manager.policyStore = policyStore;
+            return this;
+        }
+
         public T build() {
             Objects.requireNonNull(manager.validationService, "contractValidationService");
             Objects.requireNonNull(manager.monitor, "monitor");
@@ -119,11 +126,11 @@ public T build() {
             Objects.requireNonNull(manager.telemetry, "telemetry");
             Objects.requireNonNull(manager.executorInstrumentation, "executorInstrumentation");
             Objects.requireNonNull(manager.negotiationStore, "store");
+            Objects.requireNonNull(manager.policyStore, "policyStore");
             manager.commandProcessor = new CommandProcessor<>(manager.commandQueue, manager.commandRunner, manager.monitor);
 
             return manager;
         }
-
     }
 
 }

core/contract/src/main/java/org/eclipse/dataspaceconnector/contract/negotiation/ConsumerContractNegotiationManagerImpl.java

@@ -21,6 +21,7 @@
 import org.eclipse.dataspaceconnector.common.statemachine.StateMachine;
 import org.eclipse.dataspaceconnector.common.statemachine.StateProcessorImpl;
 import org.eclipse.dataspaceconnector.contract.common.ContractId;
+import org.eclipse.dataspaceconnector.policy.model.Policy;
 import org.eclipse.dataspaceconnector.spi.contract.negotiation.ConsumerContractNegotiationManager;
 import org.eclipse.dataspaceconnector.spi.contract.negotiation.observe.ContractNegotiationListener;
 import org.eclipse.dataspaceconnector.spi.iam.ClaimToken;
@@ -175,13 +176,13 @@ public StatusResult<ContractNegotiation> offerReceived(ClaimToken token, String
      * @param token Claim token of the consumer that send the contract request.
      * @param negotiationId Id of the ContractNegotiation.
      * @param agreement Agreement sent by provider.
-     * @param hash A hash of all previous contract offers.
+     * @param policy the policy
      * @return a {@link StatusResult}: FATAL_ERROR, if no match found for Id or no last
      *         offer found for negotiation; OK otherwise
      */
     @WithSpan
     @Override
-    public StatusResult<ContractNegotiation> confirmed(ClaimToken token, String negotiationId, ContractAgreement agreement, String hash) {
+    public StatusResult<ContractNegotiation> confirmed(ClaimToken token, String negotiationId, ContractAgreement agreement, Policy policy) {
         var negotiation = negotiationStore.find(negotiationId);
         if (negotiation == null) {
             return StatusResult.failure(FATAL_ERROR, format("ContractNegotiation with id %s not found", negotiationId));
@@ -214,6 +215,7 @@ public StatusResult<ContractNegotiation> confirmed(ClaimToken token, String nego
             // TODO: otherwise will fail. But should do it, since it's already confirmed? A duplicated message received shouldn't be an issue
             negotiation.transitionConfirmed();
         }
+        policyStore.save(policy);
         update(negotiation, l -> l.preConfirmed(negotiation));
         monitor.debug(String.format("[Consumer] ContractNegotiation %s is now in state %s.",
                 negotiation.getId(), ContractNegotiationStates.from(negotiation.getState())));
@@ -390,14 +392,15 @@ private boolean processConsumerApproving(ContractNegotiation negotiation) {
         }
         var definitionId = contractIdTokens[DEFINITION_PART];
 
+        var policy = lastOffer.getPolicy();
         var agreement = ContractAgreement.Builder.newInstance()
                 .id(ContractId.createContractId(definitionId))
                 .contractStartDate(Instant.now().getEpochSecond())
                 .contractEndDate(Instant.now().plus(365, ChronoUnit.DAYS).getEpochSecond()) // TODO Make configurable (issue #722)
                 .contractSigningDate(Instant.now().getEpochSecond())
                 .providerAgentId(String.valueOf(lastOffer.getProvider()))
                 .consumerAgentId(String.valueOf(lastOffer.getConsumer()))
-                .policy(lastOffer.getPolicy())
+                .policyId(policy.getUid())
                 .assetId(lastOffer.getAsset().getId())
                 .build();
 
@@ -407,6 +410,7 @@ private boolean processConsumerApproving(ContractNegotiation negotiation) {
                 .connectorAddress(negotiation.getCounterPartyAddress())
                 .contractAgreement(agreement)
                 .correlationId(negotiation.getId())
+                .policy(policy)
                 .build();
 
         // TODO protocol-independent response type?

core/contract/src/main/java/org/eclipse/dataspaceconnector/contract/negotiation/ProviderContractNegotiationManagerImpl.java

@@ -21,6 +21,7 @@
 import org.eclipse.dataspaceconnector.common.statemachine.StateMachine;
 import org.eclipse.dataspaceconnector.common.statemachine.StateProcessorImpl;
 import org.eclipse.dataspaceconnector.contract.common.ContractId;
+import org.eclipse.dataspaceconnector.policy.model.Policy;
 import org.eclipse.dataspaceconnector.spi.contract.negotiation.ProviderContractNegotiationManager;
 import org.eclipse.dataspaceconnector.spi.contract.negotiation.observe.ContractNegotiationListener;
 import org.eclipse.dataspaceconnector.spi.iam.ClaimToken;
@@ -370,6 +371,7 @@ private boolean processConfirming(ContractNegotiation negotiation) {
         var retrievedAgreement = negotiation.getContractAgreement();
 
         ContractAgreement agreement;
+        Policy policy;
         if (retrievedAgreement == null) {
             var lastOffer = negotiation.getLastContractOffer();
 
@@ -380,6 +382,7 @@ private boolean processConfirming(ContractNegotiation negotiation) {
             }
             var definitionId = contractIdTokens[DEFINITION_PART];
 
+            policy = lastOffer.getPolicy();
             //TODO move to own service
             agreement = ContractAgreement.Builder.newInstance()
                     .id(ContractId.createContractId(definitionId))
@@ -388,11 +391,12 @@ private boolean processConfirming(ContractNegotiation negotiation) {
                     .contractSigningDate(Instant.now().getEpochSecond())
                     .providerAgentId(String.valueOf(lastOffer.getProvider()))
                     .consumerAgentId(String.valueOf(lastOffer.getConsumer()))
-                    .policy(lastOffer.getPolicy())
+                    .policyId(policy.getUid())
                     .assetId(lastOffer.getAsset().getId())
                     .build();
         } else {
             agreement = retrievedAgreement;
+            policy = policyStore.findById(agreement.getPolicyId());
         }
 
         var request = ContractAgreementRequest.Builder.newInstance()
@@ -401,6 +405,7 @@ private boolean processConfirming(ContractNegotiation negotiation) {
                 .connectorAddress(negotiation.getCounterPartyAddress())
                 .contractAgreement(agreement)
                 .correlationId(negotiation.getCorrelationId())
+                .policy(policy)
                 .build();
 
         //TODO protocol-independent response type?

core/contract/src/main/java/org/eclipse/dataspaceconnector/contract/policy/PolicyArchiveImpl.java

@@ -17,22 +17,26 @@
 import org.eclipse.dataspaceconnector.policy.model.Policy;
 import org.eclipse.dataspaceconnector.spi.contract.negotiation.store.ContractNegotiationStore;
 import org.eclipse.dataspaceconnector.spi.policy.store.PolicyArchive;
+import org.eclipse.dataspaceconnector.spi.policy.store.PolicyStore;
 import org.eclipse.dataspaceconnector.spi.types.domain.contract.agreement.ContractAgreement;
 
 import java.util.Optional;
 
 public class PolicyArchiveImpl implements PolicyArchive {
     private final ContractNegotiationStore contractNegotiationStore;
+    private final PolicyStore policyStore;
 
-    public PolicyArchiveImpl(ContractNegotiationStore contractNegotiationStore) {
+    public PolicyArchiveImpl(ContractNegotiationStore contractNegotiationStore, PolicyStore policyStore) {
         this.contractNegotiationStore = contractNegotiationStore;
+        this.policyStore = policyStore;
     }
 
     @Override
     public Policy findPolicyForContract(String contractId) {
         return Optional.ofNullable(contractId)
                 .map(contractNegotiationStore::findContractAgreement)
-                .map(ContractAgreement::getPolicy)
+                .map(ContractAgreement::getPolicyId)
+                .map(policyStore::findById)
                 .orElse(null);
     }
 

core/contract/src/test/java/org/eclipse/dataspaceconnector/contract/negotiation/AbstractContractNegotiationIntegrationTest.java

@@ -30,7 +30,9 @@
 import org.eclipse.dataspaceconnector.spi.message.MessageContext;
 import org.eclipse.dataspaceconnector.spi.message.RemoteMessageDispatcher;
 import org.eclipse.dataspaceconnector.spi.message.RemoteMessageDispatcherRegistry;
+import org.eclipse.dataspaceconnector.spi.monitor.ConsoleMonitor;
 import org.eclipse.dataspaceconnector.spi.monitor.Monitor;
+import org.eclipse.dataspaceconnector.spi.policy.store.PolicyStore;
 import org.eclipse.dataspaceconnector.spi.response.StatusResult;
 import org.eclipse.dataspaceconnector.spi.types.domain.asset.Asset;
 import org.eclipse.dataspaceconnector.spi.types.domain.contract.agreement.ContractAgreementRequest;
@@ -69,6 +71,8 @@ public abstract class AbstractContractNegotiationIntegrationTest {
 
     protected ContractValidationService validationService;
 
+    protected final PolicyStore policyStore = mock(PolicyStore.class);
+
     protected String consumerNegotiationId;
 
     protected ClaimToken token;
@@ -84,7 +88,7 @@ void setUp() {
         validationService = mock(ContractValidationService.class);
 
         // Create a monitor that logs to the console
-        Monitor monitor = new FakeConsoleMonitor();
+        Monitor monitor = new ConsoleMonitor();
 
         // Create CommandQueue mock
         CommandQueue<ContractNegotiationCommand> queue = (CommandQueue<ContractNegotiationCommand>) mock(CommandQueue.class);
@@ -105,6 +109,7 @@ void setUp() {
                 .commandRunner(runner)
                 .observable(providerObservable)
                 .store(providerStore)
+                .policyStore(policyStore)
                 .build();
 
         consumerManager = ConsumerContractNegotiationManagerImpl.Builder.newInstance()
@@ -116,6 +121,7 @@ void setUp() {
                 .commandRunner(runner)
                 .observable(consumerObservable)
                 .store(consumerStore)
+                .policyStore(policyStore)
                 .build();
 
         countDownLatch = new CountDownLatch(2);
@@ -125,7 +131,7 @@ void setUp() {
      * Implementation of the ContractNegotiationListener that signals a CountDownLatch when the
      * confirmed state has been reached.
      */
-    protected class ConfirmedContractNegotiationListener implements ContractNegotiationListener {
+    protected static class ConfirmedContractNegotiationListener implements ContractNegotiationListener {
 
         private final CountDownLatch countDownLatch;
 
@@ -143,7 +149,7 @@ public void preConfirmed(ContractNegotiation negotiation) {
      * Implementation of the ContractNegotiationListener that signals a CountDownLatch when the
      * declined state has been reached.
      */
-    protected class DeclinedContractNegotiationListener implements ContractNegotiationListener {
+    protected static class DeclinedContractNegotiationListener implements ContractNegotiationListener {
 
         private final CountDownLatch countDownLatch;
 
@@ -180,7 +186,7 @@ public CompletableFuture<Object> send(RemoteMessage message) {
                 result = consumerManager.offerReceived(token, request.getCorrelationId(), request.getContractOffer(), "hash");
             } else if (message instanceof ContractAgreementRequest) {
                 var request = (ContractAgreementRequest) message;
-                result = consumerManager.confirmed(token, request.getCorrelationId(), request.getContractAgreement(), "hash");
+                result = consumerManager.confirmed(token, request.getCorrelationId(), request.getContractAgreement(), request.getPolicy());
             } else if (message instanceof ContractRejection) {
                 var request = (ContractRejection) message;
                 result = consumerManager.declined(token, request.getCorrelationId());
@@ -245,51 +251,6 @@ public CompletableFuture<Object> send(RemoteMessage message) {
         }
     }
 
-    /**
-     * Monitor implementation that prints to the console.
-     */
-    protected class FakeConsoleMonitor implements Monitor {
-        @Override
-        public void debug(String message, Throwable... errors) {
-            System.out.println("\u001B[34mDEBUG\u001B[0m - " + message);
-            if (errors != null && errors.length > 0) {
-                for (Throwable error : errors) {
-                    error.printStackTrace();
-                }
-            }
-        }
-
-        @Override
-        public void info(String message, Throwable... errors) {
-            System.out.println("\u001B[32mINFO\u001B[0m - " + message);
-            if (errors != null && errors.length > 0) {
-                for (Throwable error : errors) {
-                    error.printStackTrace();
-                }
-            }
-        }
-
-        @Override
-        public void warning(String message, Throwable... errors) {
-            System.out.println("\u001B[33mWARNING\u001B[0m - " + message);
-            if (errors != null && errors.length > 0) {
-                for (Throwable error : errors) {
-                    error.printStackTrace();
-                }
-            }
-        }
-
-        @Override
-        public void severe(String message, Throwable... errors) {
-            System.out.println("\u001B[31mSEVERE\u001B[0m - " + message);
-            if (errors != null && errors.length > 0) {
-                for (Throwable error : errors) {
-                    error.printStackTrace();
-                }
-            }
-        }
-    }
-
     /**
      * Creates the initial contract offer.
      *