initial draft

common/util/build.gradle.kts

@@ -35,6 +35,7 @@ publishing {
         create<MavenPublication>("common-util") {
             artifactId = "common-util"
             from(components["java"])
+            artifacts.forEach { a -> println(a.file) }
         }
     }
 }

common/util/src/main/java/org/eclipse/dataspaceconnector/common/reflection/ReflectionUtil.java

@@ -33,13 +33,14 @@ public class ReflectionUtil {
     private static final String CLOSING_BRACKET = "]";
 
     /**
-     * Utility function to get value of a field from an object. For field names currently the dot notation and array indexers are supported:
+     * Utility function to get value of a field from an object. For field names currently the dot notation and array
+     * indexers are supported:
      * <pre>
      *     someObject.someValue
      *     someObject[2].someValue //someObject must impement the List interface
      * </pre>
      *
-     * @param object       The object
+     * @param object The object
      * @param propertyName The name of the field
      * @return The field's value.
      * @throws ReflectionException if the field does not exist or is not accessible
@@ -53,6 +54,9 @@ public static <T> T getFieldValue(String propertyName, Object object) {
             var field = propertyName.substring(0, dotIx);
             var rest = propertyName.substring(dotIx + 1);
             object = getFieldValue(field, object);
+            if (object == null) {
+                return null;
+            }
             return getFieldValue(rest, object);
         } else if (propertyName.matches(ARRAY_INDEXER_REGEX)) { //array indexer
             var openingBracketIx = propertyName.indexOf(OPENING_BRACKET);
@@ -82,10 +86,10 @@ public static <T> T getFieldValue(String propertyName, Object object) {
 
 
     /**
-     * Utility function to get value of a field from an object. Essentially the same as {@link ReflectionUtil#getFieldValue(String, Object)}
-     * but it does not throw an exception
+     * Utility function to get value of a field from an object. Essentially the same as
+     * {@link ReflectionUtil#getFieldValue(String, Object)} but it does not throw an exception
      *
-     * @param object       The object
+     * @param object The object
      * @param propertyName The name of the field
      * @return The field's value. Returns null if the field does not exist or is inaccessible.
      */
@@ -118,9 +122,10 @@ public static <T> Comparator<T> propertyComparator(boolean isAscending, String p
 
 
     /**
-     * Gets a field with a given name from all declared fields of a class including supertypes. Will include protected and private fields.
+     * Gets a field with a given name from all declared fields of a class including supertypes. Will include protected
+     * and private fields.
      *
-     * @param clazz     The class of the object
+     * @param clazz The class of the object
      * @param fieldName The fieldname
      * @return A field with the given name, null if the field does not exist
      */

core/defaults/src/main/java/org/eclipse/dataspaceconnector/core/defaults/negotiationstore/InMemoryContractNegotiationStore.java

@@ -36,11 +36,11 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static java.lang.String.format;
 import static java.util.stream.Collectors.toList;
 
 /**
- * An in-memory, threadsafe process store.
- * This implementation is intended for testing purposes only.
+ * An in-memory, threadsafe process store. This implementation is intended for testing purposes only.
  */
 public class InMemoryContractNegotiationStore implements ContractNegotiationStore {
 
@@ -124,6 +124,14 @@ public void delete(String processId) {
         return lockManager.readLock(() -> agreementQueryResolver.query(getAgreements(), querySpec));
     }
 
+    @Override
+    public Stream<ContractNegotiation> getNegotiationsWithAgreementOnAsset(String assetId) {
+        var filter = format("contractAgreement.assetId = %s", assetId);
+        var query = QuerySpec.Builder.newInstance().filter(filter).build();
+
+        return queryNegotiations(query);
+    }
+
     @Override
     public @NotNull List<ContractNegotiation> nextForState(int state, int max) {
         return lockManager.readLock(() -> {

core/defaults/src/test/java/org/eclipse/dataspaceconnector/core/defaults/negotiationstore/InMemoryContractNegotiationStoreTest.java

@@ -20,6 +20,7 @@
 import org.eclipse.dataspaceconnector.spi.query.SortOrder;
 import org.eclipse.dataspaceconnector.spi.types.domain.contract.agreement.ContractAgreement;
 import org.eclipse.dataspaceconnector.spi.types.domain.contract.negotiation.ContractNegotiation;
+import org.eclipse.dataspaceconnector.spi.types.domain.contract.negotiation.ContractNegotiationStates;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -28,6 +29,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.UUID;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -320,6 +322,41 @@ void queryAgreements_verifySorting_invalidProperty() {
         assertThat(store.queryAgreements(query)).isEmpty();
     }
 
+    @Test
+    void getNegotiationsWithAgreementOnAsset_negotiationWithAgreement() {
+        var agreement = createAgreementBuilder().id("contract1").build();
+        var negotiation = createNegotiationBuilder("negotiation1").contractAgreement(agreement).build();
+        var assetId = agreement.getAssetId();
+
+        store.save(negotiation);
+
+        var result = store.getNegotiationsWithAgreementOnAsset(assetId).collect(Collectors.toList());
+
+        assertThat(result).hasSize(1).usingRecursiveFieldByFieldElementComparator().containsOnly(negotiation);
+    }
+
+    @Test
+    void getNegotiationsWithAgreementOnAsset_negotiationWithoutAgreement() {
+        var assetId = UUID.randomUUID().toString();
+        var negotiation = ContractNegotiation.Builder.newInstance()
+                .type(ContractNegotiation.Type.CONSUMER)
+                .id("negotiation1")
+                .contractAgreement(null)
+                .correlationId("corr-negotiation1")
+                .state(ContractNegotiationStates.REQUESTED.code())
+                .counterPartyAddress("consumer")
+                .counterPartyId("consumerId")
+                .protocol("ids-multipart")
+                .build();
+
+        store.save(negotiation);
+
+        var result = store.getNegotiationsWithAgreementOnAsset(assetId).collect(Collectors.toList());
+
+        assertThat(result).isEmpty();
+        assertThat(store.queryAgreements(QuerySpec.none())).isEmpty();
+    }
+
     @NotNull
     private ContractNegotiation requestingNegotiation() {
         var negotiation = createNegotiation(UUID.randomUUID().toString());

extensions/api/data-management/asset/src/main/java/org/eclipse/dataspaceconnector/api/datamanagement/asset/service/AssetServiceImpl.java

@@ -51,13 +51,23 @@ public Collection<Asset> query(QuerySpec query) {
         return transactionContext.execute(() -> index.queryAssets(query).collect(toList()));
     }
 
+    @Override
+    public ServiceResult<Asset> create(Asset asset, DataAddress dataAddress) {
+        return transactionContext.execute(() -> {
+            if (findById(asset.getId()) == null) {
+                loader.accept(asset, dataAddress);
+                return ServiceResult.success(asset);
+            } else {
+                return ServiceResult.conflict(format("Asset %s cannot be created because it already exist", asset.getId()));
+            }
+        });
+    }
+
     @Override
     public ServiceResult<Asset> delete(String assetId) {
         return transactionContext.execute(() -> {
-            var filter = format("contractAgreement.assetId = %s", assetId);
-            var query = QuerySpec.Builder.newInstance().filter(filter).build();
 
-            var negotiationsOnAsset = contractNegotiationStore.queryNegotiations(query);
+            var negotiationsOnAsset = contractNegotiationStore.getNegotiationsWithAgreementOnAsset(assetId);
             if (negotiationsOnAsset.findAny().isPresent()) {
                 return ServiceResult.conflict(format("Asset %s cannot be deleted as it is referenced by at least one contract agreement", assetId));
             }
@@ -70,16 +80,4 @@ public ServiceResult<Asset> delete(String assetId) {
             return ServiceResult.success(deleted);
         });
     }
-
-    @Override
-    public ServiceResult<Asset> create(Asset asset, DataAddress dataAddress) {
-        return transactionContext.execute(() -> {
-            if (findById(asset.getId()) == null) {
-                loader.accept(asset, dataAddress);
-                return ServiceResult.success(asset);
-            } else {
-                return ServiceResult.conflict(format("Asset %s cannot be created because it already exist", asset.getId()));
-            }
-        });
-    }
 }

extensions/azure/cosmos/contract-negotiation-store-cosmos/src/main/java/org/eclipse/dataspaceconnector/contract/negotiation/store/CosmosContractNegotiationStore.java

@@ -42,6 +42,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static java.lang.String.format;
 import static java.util.Optional.ofNullable;
 import static net.jodah.failsafe.Failsafe.with;
 
@@ -162,6 +163,14 @@ public void delete(String negotiationId) {
                 .filter(Objects::nonNull);
     }
 
+    @Override
+    public Stream<ContractNegotiation> getNegotiationsWithAgreementOnAsset(String assetId) {
+        var filter = format("contractAgreement.assetId = %s", assetId);
+        var query = QuerySpec.Builder.newInstance().filter(filter).build();
+
+        return queryNegotiations(query);
+    }
+
     @Override
     public @NotNull List<ContractNegotiation> nextForState(int state, int max) {
 

extensions/azure/cosmos/contract-negotiation-store-cosmos/src/test/java/org/eclipse/dataspaceconnector/contract/negotiation/store/CosmosContractNegotiationStoreIntegrationTest.java

@@ -544,6 +544,41 @@ void queryAgreements_verifySorting_invalidProperty() {
         assertThat(agreements).isEmpty();
     }
 
+    @Test
+    void getNegotiationsWithAgreementOnAsset_negotiationWithAgreement() {
+        var agreement = generateAgreementBuilder().id("contract1").build();
+        var negotiation = generateNegotiationBuilder("negotiation1").contractAgreement(agreement).build();
+        var assetId = agreement.getAssetId();
+
+        store.save(negotiation);
+
+        var result = store.getNegotiationsWithAgreementOnAsset(assetId).collect(Collectors.toList());
+
+        assertThat(result).hasSize(1).usingRecursiveFieldByFieldElementComparator().containsOnly(negotiation);
+    }
+
+    @Test
+    void getNegotiationsWithAgreementOnAsset_negotiationWithoutAgreement() {
+        var assetId = UUID.randomUUID().toString();
+        var negotiation = ContractNegotiation.Builder.newInstance()
+                .type(ContractNegotiation.Type.CONSUMER)
+                .id("negotiation1")
+                .contractAgreement(null)
+                .correlationId("corr-negotiation1")
+                .state(ContractNegotiationStates.REQUESTED.code())
+                .counterPartyAddress("consumer")
+                .counterPartyId("consumerId")
+                .protocol("ids-multipart")
+                .build();
+
+        store.save(negotiation);
+
+        var result = store.getNegotiationsWithAgreementOnAsset(assetId).collect(Collectors.toList());
+
+        assertThat(result).isEmpty();
+        assertThat(store.queryAgreements(QuerySpec.none())).isEmpty();
+    }
+
     private ContractNegotiationDocument toDocument(Object object) {
         var json = typeManager.writeValueAsString(object);
         return typeManager.readValue(json, ContractNegotiationDocument.class);

extensions/sql/contract-negotiation-store-sql/src/main/java/org/eclipse/dataspaceconnector/sql/contractnegotiation/store/SqlContractNegotiationStore.java

@@ -191,6 +191,25 @@ public void delete(String negotiationId) {
         });
     }
 
+    @Override
+    public Stream<ContractNegotiation> getNegotiationsWithAgreementOnAsset(String assetId) {
+        var statement = "\n" +
+                "SELECT *\n" +
+                "FROM edc_contract_negotiation\n" +
+                "INNER JOIN edc_contract_agreement eca on edc_contract_negotiation.contract_agreement_id = eca.agreement_id\n" +
+                "WHERE edc_contract_negotiation.contract_agreement_id in (SELECT agreement_id\n" +
+                "                                                         FROM edc_contract_agreement\n" +
+                "                                                         WHERE asset_id = ?);\n";
+
+        return transactionContext.execute(() -> {
+            try (var connection = getConnection()) {
+                return executeQuery(connection, this::mapContractNegotiation, statement, assetId).stream();
+            } catch (SQLException e) {
+                throw new EdcPersistenceException(e);
+            }
+        });
+    }
+
     @Override
     public @NotNull List<ContractNegotiation> nextForState(int state, int max) {
         return transactionContext.execute(() -> {

extensions/sql/contract-negotiation-store-sql/src/test/java/org/eclipse/dataspaceconnector/sql/contractnegotiation/store/SqlContractNegotiationStoreTest.java

@@ -401,6 +401,43 @@ void queryNegotiations() {
                 .allMatch(i -> i > 4 && i < 15);
     }
 
+    @Test
+    void getNegotiationsWithAgreementOnAsset_negotiationWithAgreement() {
+        var agreement = createContract("contract1");
+        var negotiation = createNegotiation("negotiation1", agreement);
+        var assetId = agreement.getAssetId();
+
+        store.save(negotiation);
+
+        var result = store.getNegotiationsWithAgreementOnAsset(assetId).collect(Collectors.toList());
+
+        assertThat(result).hasSize(1).usingRecursiveFieldByFieldElementComparator().containsOnly(negotiation);
+
+    }
+
+    @Test
+    void getNegotiationsWithAgreementOnAsset_negotiationWithoutAgreement() {
+        var assetId = UUID.randomUUID().toString();
+        var negotiation = ContractNegotiation.Builder.newInstance()
+                .type(ContractNegotiation.Type.CONSUMER)
+                .id("negotiation1")
+                .contractAgreement(null)
+                .correlationId("corr-negotiation1")
+                .state(ContractNegotiationStates.REQUESTED.code())
+                .counterPartyAddress("consumer")
+                .counterPartyId("consumerId")
+                .protocol("ids-multipart")
+                .build();
+
+        store.save(negotiation);
+
+        var result = store.getNegotiationsWithAgreementOnAsset(assetId).collect(Collectors.toList());
+
+        assertThat(result).isEmpty();
+        assertThat(store.queryAgreements(QuerySpec.none())).isEmpty();
+
+    }
+
     @Test
     @DisplayName("Verify that paging is used")
     void queryNegotiations_withAgreement() {

spi/contract-spi/src/main/java/org/eclipse/dataspaceconnector/spi/contract/negotiation/store/ContractNegotiationStore.java

@@ -49,7 +49,8 @@ public interface ContractNegotiationStore extends StateEntityStore<ContractNegot
     ContractAgreement findContractAgreement(String contractId);
 
     /**
-     * Persists a contract negotiation. This follows UPSERT semantics, so if the object didn't exit before, it's created.
+     * Persists a contract negotiation. This follows UPSERT semantics, so if the object didn't exit before, it's
+     * created.
      */
     void save(ContractNegotiation negotiation);
 
@@ -59,8 +60,8 @@ public interface ContractNegotiationStore extends StateEntityStore<ContractNegot
     void delete(String negotiationId);
 
     /**
-     * Finds all contract negotiations that are covered by a specific {@link QuerySpec}. If no {@link QuerySpec#getSortField()}
-     * is specified, results are not explicitly sorted.
+     * Finds all contract negotiations that are covered by a specific {@link QuerySpec}. If no
+     * {@link QuerySpec#getSortField()} is specified, results are not explicitly sorted.
      * <p>
      * The general order of precedence of the query parameters is:
      * <pre>
@@ -85,12 +86,20 @@ public interface ContractNegotiationStore extends StateEntityStore<ContractNegot
 
 
     /**
-     * Finds all contract agreement that are covered by a specific {@link QuerySpec}.
-     * If no {@link QuerySpec#getSortField()} is specified, results are not explicitly sorted.
+     * Finds all contract agreement that are covered by a specific {@link QuerySpec}. If no
+     * {@link QuerySpec#getSortField()} is specified, results are not explicitly sorted.
      *
      * @param querySpec The query spec, e.g. paging, filtering, etc.
      * @return a stream of ContractAgreement, cannot be null.
      */
     @NotNull
     Stream<ContractAgreement> queryAgreements(QuerySpec querySpec);
+
+    /**
+     * Finds all negotiations, that have agreements targeting the given asset
+     *
+     * @param assetId The asset for which the negotiations + agreements are wanted.
+     * @return A stream of contract negotiations, or empty
+     */
+    Stream<ContractNegotiation> getNegotiationsWithAgreementOnAsset(String assetId);
 }