Fix deserialization of subclasses

Previously we were deserializing step data based on the T provided by the compiler.
However, this could be lossy if the step data was actually a subclass of
T with additional fields and is semantically incorrect if we try to
deserialize it as the parent class.

To remedy this, we need to know what the actual class of the step data
was before it was memoized with the Inngest server. We are adding this
as an additional field `class` on the JSON when serializing, then
reading (and deleting) this field and using it with reflection to
deserialize using the proper class.

It might make sense to put the field `class` in some kind of
meta/extension section of the payload to avoid conflict, but it's highly
unlikely a field will legitimately be named class in the user code since
it is a reserved keyword in Java.

This strategy doesn't entirely work for schema evolution because it's
still possible the classes have changed to the point where
deserialization would fail. We should follow up here and probably have a
fallback mechanism of returning a generic hash if deserialization fails.

inngest/src/main/kotlin/com/inngest/Function.kt

@@ -1,6 +1,9 @@
 package com.inngest
 
 import com.beust.klaxon.Json
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.node.ObjectNode
 import java.util.function.BiFunction
 
 // TODO - Add an abstraction layer between the Function call response and the comm handler response
@@ -229,8 +232,9 @@ internal open class InternalInngestFunction(
             // NOTE - Currently this error could be caught in the user's own function
             // that wraps a
             // step.run() - how can we prevent that or warn?
+
             return StepResult(
-                data = e.data,
+                data = serializeStepData(e.data),
                 id = e.hashedId,
                 name = e.id,
                 op = OpCode.StepRun,
@@ -255,4 +259,23 @@ internal open class InternalInngestFunction(
         // TODO use URL objects for serveUrl instead of strings so we can fetch things like scheme
         return configBuilder.build(client.appId, serveUrl)
     }
+
+    private fun serializeStepData(stepData: Any?): JsonNode? {
+        if (stepData == null) {
+            return stepData
+        }
+
+        val mapper = ObjectMapper()
+        val jsonString = mapper.writeValueAsString(stepData)
+        val readOnlyJson = mapper.readTree(jsonString)
+
+        if (!readOnlyJson.isObject) {
+            // primitives can be serialized directly
+            return readOnlyJson
+        }
+
+        val writeableJson = mapper.readTree(jsonString) as ObjectNode
+        writeableJson.put("class", stepData.javaClass.name)
+        return writeableJson
+    }
 }

inngest/src/main/kotlin/com/inngest/State.kt

@@ -2,6 +2,7 @@ package com.inngest
 
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.node.ObjectNode
 import java.security.MessageDigest
 
 class StateNotFound : Throwable("State not found for id")
@@ -60,8 +61,7 @@ class State(
         val stepResult = node.path("steps").get(hashedId) ?: throw StateNotFound()
 
         if (stepResult.has(fieldName)) {
-            val dataNode = stepResult.get(fieldName)
-            return mapper.treeToValue(dataNode, type)
+            return deserializeStepData(stepResult.get(fieldName), type)
         } else if (stepResult.has("error")) {
             val error = mapper.treeToValue(stepResult.get("error"), StepError::class.java)
             throw error
@@ -71,4 +71,19 @@ class State(
         // TODO - Check the state is actually null
         return null
     }
+
+    private fun <T> deserializeStepData(
+        serializedStepData: JsonNode?,
+        type: Class<T>,
+    ): T? {
+        val mapper = ObjectMapper()
+        if (serializedStepData == null || !serializedStepData.isObject || !serializedStepData.has("class")) {
+            // null and primitives can be deserialized directly
+            return mapper.treeToValue(serializedStepData, type)
+        }
+
+        val writeableJson = serializedStepData as ObjectNode
+        val className = writeableJson.remove("class").asText()
+        return mapper.treeToValue(writeableJson, Class.forName(className)) as T
+    }
 }