Fix keyArgument cast bug

gateway/src/main/java/com/bakdata/quick/gateway/fetcher/KeyFieldFetcher.java

@@ -17,17 +17,13 @@
 package com.bakdata.quick.gateway.fetcher;
 
 import com.bakdata.quick.common.exception.BadArgumentException;
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.protobuf.DynamicMessage;
-import com.google.protobuf.InvalidProtocolBufferException;
-import com.google.protobuf.util.JsonFormat;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import graphql.schema.DataFetcher;
 import graphql.schema.DataFetchingEnvironment;
-import java.io.UncheckedIOException;
-import java.nio.charset.StandardCharsets;
+import java.io.IOException;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -102,32 +98,27 @@ public Object get(final DataFetchingEnvironment environment) {
 
     // TODO: Fix this
     private Stream<K> findKeyArgument(final DataFetchingEnvironment environment) {
-        final String parentJson;
+        final JsonNode parentJson;
         try {
             parentJson = this.extractJson(environment);
-        } catch (final JsonProcessingException | InvalidProtocolBufferException e) {
+        } catch (final IOException e) {
             throw new RuntimeException("Could not convert source for extracting key", e);
         }
         // parse json and try to find the value for our argument
         // if it is an array, we need to resolve each one
-        try {
-            final JsonNode parent = this.objectMapper.readTree(parentJson);
-            final JsonNode node = parent.findValue(this.argument);
+        final JsonNode node = parentJson.findValue(this.argument);
+        if (node == null) {
+            throw new IllegalArgumentException(
+                String.format("Field + %s could not be found in source.", this.argument));
+        }
+        if (node.isArray()) {
+            final Iterator<JsonNode> elements = node.elements();
+            return (Stream<K>) StreamSupport
+                .stream(Spliterators.spliteratorUnknownSize(elements, 0), false)
+                .map(KeyFieldFetcher::extractCorrectType);
+        } else {
             final Object typedNode = extractCorrectType(node);
-            if (node == null) {
-                throw new IllegalArgumentException(
-                    String.format("Field + %s could not be found in source.", this.argument));
-            }
-            if (node.isArray()) {
-                final Iterator<JsonNode> elements = node.elements();
-                return (Stream<K>) StreamSupport
-                    .stream(Spliterators.spliteratorUnknownSize(elements, 0), false)
-                    .map(KeyFieldFetcher::extractCorrectType);
-            } else {
-                return (Stream<K>) Stream.of(typedNode);
-            }
-        } catch (final JsonProcessingException e) {
-            throw new UncheckedIOException("Could not process json: " + parentJson, e);
+            return (Stream<K>) Stream.of(typedNode);
         }
     }
 
@@ -149,30 +140,21 @@ private static Object extractCorrectType(final JsonNode jsonNode) {
         }
     }
 
-    private String extractJson(final DataFetchingEnvironment environment) throws JsonProcessingException,
-        InvalidProtocolBufferException {
+    private JsonNode extractJson(final DataFetchingEnvironment environment) throws IOException {
         // TODO work on JSON everywhere:
         //  1. Do not convert back to real types in MirrorDataFetcherClient
         //  2. Immediately convert to JSON in SubscriptionFetcher
         if (environment.getSource() instanceof GenericRecord) {
             final GenericRecord record = environment.getSource();
-            return new String(this.converter.convertToJson(record), StandardCharsets.UTF_8);
+            return this.objectMapper.readTree(this.converter.convertToJson(record));
         }
 
         if (environment.getSource() instanceof DynamicMessage) {
             final DynamicMessage source = environment.getSource();
-            return JsonFormat.printer().includingDefaultValueFields().print(source);
+            return this.objectMapper.readTree(source.toByteArray());
         }
 
         final Map<String, Object> value = environment.getSource();
-        return this.objectMapper.writeValueAsString(value);
-    }
-
-    private String valueAsString(final JsonNode node) {
-        try {
-            return node.isTextual() ? node.textValue() : this.objectMapper.writeValueAsString(node);
-        } catch (final JsonProcessingException e) {
-            throw new UncheckedIOException("Could not process json: " + node, e);
-        }
+        return this.objectMapper.valueToTree(value);
     }
 }

gateway/src/main/java/com/bakdata/quick/gateway/fetcher/QueryKeyArgumentFetcher.java

@@ -56,5 +56,4 @@ public T get(final DataFetchingEnvironment environment) {
         }
         return value;
     }
-
 }

gateway/src/test/java/com/bakdata/quick/gateway/GraphQLQueryExecutionTest.java

@@ -90,7 +90,8 @@ void shouldExecuteQueryWithSingleField(final TestInfo testInfo) throws IOExcepti
         final TestClientSupplier testClientSupplier = new TestClientSupplier();
         final GraphQL graphQL = this.getGraphQL(schemaPath, testClientSupplier);
         final DataFetcherClient<String, ?> dataFetcherClient = testClientSupplier.getClient("purchase-topic");
-        final Purchase purchase = Purchase.builder().purchaseId("test").amount(5).productId("product").build();
+        final long productId = 123L;
+        final Purchase purchase = Purchase.builder().purchaseId("test").amount(5).productId(productId).build();
         when(dataFetcherClient.fetchResult("test")).thenAnswer(invocation -> purchase);
 
         final ExecutionResult executionResult = graphQL.execute(Files.readString(queryPath));
@@ -102,7 +103,7 @@ void shouldExecuteQueryWithSingleField(final TestInfo testInfo) throws IOExcepti
             .isNotNull()
             .containsEntry("purchaseId", "test")
             .containsEntry("amount", 5)
-            .containsEntry("productId", "product");
+            .containsEntry("productId", productId);
     }
 
     @Test
@@ -137,6 +138,47 @@ void shouldExecuteRange(final TestInfo testInfo) throws IOException {
             .satisfies(userRequest -> assertThat(userRequest.get(2).get("requests")).isEqualTo(8));
     }
 
+    @Test
+    void shouldExecuteQueryWithSingleFieldAndModification(final TestInfo testInfo) throws IOException {
+        final String name = testInfo.getTestMethod().orElseThrow().getName();
+        final Path schemaPath = workingDirectory.resolve(name + ".graphql");
+        final Path queryPath = workingDirectory.resolve(name + "Query.graphql");
+
+        final TestClientSupplier testClientSupplier = new TestClientSupplier();
+        final GraphQL graphQL = this.getGraphQL(schemaPath, testClientSupplier);
+
+        final DataFetcherClient<String, ?> purchaseClient = testClientSupplier.getClient("purchase-topic");
+        final DataFetcherClient<Long, ?> productClient = testClientSupplier.getClient("product-topic");
+
+        final long productId = 123L;
+        final Map<String, Object> purchase1 = this.mapper.convertValue(
+            Purchase.builder().purchaseId("purchase1").amount(5).productId(productId).build(),
+            OBJECT_TYPE_REFERENCE
+        );
+
+        final Product product1 = Product.builder()
+            .productId(productId)
+            .name("product-name")
+            .price(Price.builder().total(5).build())
+            .build();
+
+        when(purchaseClient.fetchResult("purchase1")).thenAnswer(invocation -> purchase1);
+        when(productClient.fetchResult(productId)).thenAnswer(invocation -> product1);
+
+        final ExecutionResult executionResult = graphQL.execute(Files.readString(queryPath));
+
+        assertThat(executionResult.getErrors()).isEmpty();
+
+        final Map<String, Map<String, Object>> data = executionResult.getData();
+        assertThat(data.get("findPurchase"))
+            .satisfies(purchase -> assertThat(purchase)
+                .containsEntry("purchaseId", "purchase1")
+                .containsEntry("productId", productId)
+                .extractingByKey("product")
+                .isNotNull()
+            );
+    }
+
     @Test
     void shouldExecuteListQueryWithSingleFieldAndModification(final TestInfo testInfo) throws IOException {
         final String name = testInfo.getTestMethod().orElseThrow().getName();
@@ -147,34 +189,37 @@ void shouldExecuteListQueryWithSingleFieldAndModification(final TestInfo testInf
         final GraphQL graphQL = this.getGraphQL(schemaPath, testClientSupplier);
 
         final DataFetcherClient<String, ?> purchaseClient = testClientSupplier.getClient("purchase-topic");
-        final DataFetcherClient<String, ?> productClient = testClientSupplier.getClient("product-topic");
+        final DataFetcherClient<Long, ?> productClient = testClientSupplier.getClient("product-topic");
+
+        final long productId1 = 123L;
+        final long productId2 = 456L;
 
         final List<?> purchases = List.of(
             this.mapper.convertValue(
-                Purchase.builder().purchaseId("purchase1").amount(5).productId("product1").build(),
+                Purchase.builder().purchaseId("purchase1").amount(5).productId(productId1).build(),
                 OBJECT_TYPE_REFERENCE
             ),
             this.mapper.convertValue(
-                Purchase.builder().purchaseId("purchase2").amount(1).productId("product2").build(),
+                Purchase.builder().purchaseId("purchase2").amount(1).productId(productId2).build(),
                 OBJECT_TYPE_REFERENCE
             )
         );
 
         final Product product1 = Product.builder()
-            .productId("product1")
+            .productId(productId1)
             .name("product-name")
             .price(Price.builder().total(5).build())
             .build();
 
         final Product product2 = Product.builder()
-            .productId("product2")
+            .productId(productId2)
             .name("product-name2")
             .price(Price.builder().total(1).build())
             .build();
 
         when(purchaseClient.fetchList()).thenAnswer(invocation -> purchases);
-        when(productClient.fetchResult("product1")).thenAnswer(invocation -> product1);
-        when(productClient.fetchResult("product2")).thenAnswer(invocation -> product2);
+        when(productClient.fetchResult(productId1)).thenAnswer(invocation -> product1);
+        when(productClient.fetchResult(productId2)).thenAnswer(invocation -> product2);
 
         final ExecutionResult executionResult = graphQL.execute(Files.readString(queryPath));
 
@@ -187,14 +232,14 @@ void shouldExecuteListQueryWithSingleFieldAndModification(final TestInfo testInf
             .anySatisfy(purchase ->
                 assertThat(purchase)
                     .containsEntry("purchaseId", "purchase1")
-                    .containsEntry("productId", "product1")
+                    .containsEntry("productId", productId1)
                     .extractingByKey("product")
                     .isNotNull()
             )
             .anySatisfy(purchase ->
                 assertThat(purchase)
                     .containsEntry("purchaseId", "purchase2")
-                    .containsEntry("productId", "product2")
+                    .containsEntry("productId", productId2)
                     .extractingByKey("product")
                     .isNotNull()
             );
@@ -330,13 +375,13 @@ private GraphQL getGraphQL(final Path schemaPath, final ClientSupplier clientSup
     private void registerTopics() {
         this.registryClient.register(
             "purchase-topic",
-            new TopicData("purchase-topic", TopicWriteType.MUTABLE, QuickTopicType.DOUBLE, QuickTopicType.AVRO,
+            new TopicData("purchase-topic", TopicWriteType.MUTABLE, QuickTopicType.STRING, QuickTopicType.AVRO,
                 "")
         ).blockingAwait();
 
         this.registryClient.register(
             "product-topic",
-            new TopicData("product-topic", TopicWriteType.MUTABLE, QuickTopicType.DOUBLE, QuickTopicType.PROTOBUF, "")
+            new TopicData("product-topic", TopicWriteType.MUTABLE, QuickTopicType.LONG, QuickTopicType.PROTOBUF, "")
         ).blockingAwait();
 
         this.registryClient.register(
@@ -371,14 +416,14 @@ private void registerTopics() {
     @Builder
     private static class Purchase {
         String purchaseId;
-        String productId;
+        Long productId;
         int amount;
     }
 
     @Value
     @Builder
     private static class Product {
-        String productId;
+        Long productId;
         String name;
         String description;
         Price price;

gateway/src/test/resources/schema/execution/shouldExecuteListQueryWithSingleFieldAndModification.graphql

@@ -4,7 +4,7 @@ type Query {
 
 type Purchase  {
     purchaseId: ID!,
-    productId: ID!,
+    productId: Long!,
     userId: ID!,
     amount: Int,
     price: Price,
@@ -13,7 +13,7 @@ type Purchase  {
 }
 
 type Product {
-    productId: ID!,
+    productId: Long!,
     name: String,
     description: String,
     price: Price,

gateway/src/test/resources/schema/execution/shouldExecuteQueryWithSingleField.graphql

@@ -4,7 +4,7 @@ type Query {
 
 type Purchase  {
     purchaseId: ID!,
-    productId: ID!,
+    productId: Long!,
     userId: ID!,
     amount: Int,
     price: Price,

gateway/src/test/resources/schema/execution/shouldExecuteQueryWithSingleFieldAndModification.graphql

@@ -4,7 +4,7 @@ type Query {
 
 type Purchase  {
     purchaseId: ID!,
-    productId: ID!,
+    productId: Long!,
     userId: ID!,
     amount: Int,
     price: Price,
@@ -13,7 +13,7 @@ type Purchase  {
 }
 
 type Product {
-    productId: ID!,
+    productId: Long!,
     name: String,
     description: String,
     price: Price,

gateway/src/test/resources/schema/execution/shouldExecuteQueryWithSingleFieldAndModificationQuery.graphql

@@ -1,31 +1,13 @@
-type Query {
-    findPurchase(purchaseId: ID): Purchase @topic(name: "purchase-topic", keyArgument: "purchaseId")
+{
+    findPurchase(purchaseId: "purchase1") {
+        purchaseId
+        productId
+        product {
+            productId
+            name
+            price {
+                total
+            }
+        }
+    }
 }
-
-type Purchase  {
-    purchaseId: ID!,
-    productId: ID!,
-    userId: ID!,
-    amount: Int,
-    price: Price,
-    infos: [String]
-    product: Product @topic(name: "product-topic" keyField: "productId")
-}
-
-type Product {
-    productId: ID!,
-    name: String,
-    description: String,
-    price: Price,
-    metadata: Metadata
-}
-
-type Price {
-    total: Float,
-    currency: String
-}
-
-type Metadata {
-    created_at: Int,
-    source: String
-}