WIP: start eventing from timestamp

akka-javasdk-testkit/src/main/java/akka/javasdk/testkit/EntitySerializationChecker.java

@@ -4,18 +4,28 @@
 
 package akka.javasdk.testkit;
 
-import akka.javasdk.impl.serialization.JsonSerializer;
+import akka.javasdk.impl.serialization.Serializer;
 import akka.runtime.sdk.spi.BytesPayload;
 import java.lang.reflect.Type;
 
+/**
+ * Internal helper to verify serialization/deserialization of entity events, state, commands, and
+ * responses. Supports both JSON and Protobuf serialization formats.
+ */
 final class EntitySerializationChecker {
 
-  private static JsonSerializer jsonSerializer = new JsonSerializer();
+  private static Serializer serializer = new Serializer();
 
   static void verifySerDer(Object object, Object entity) {
     try {
-      BytesPayload bytesPayload = jsonSerializer.toBytes(object);
-      jsonSerializer.fromBytes(bytesPayload);
+      BytesPayload bytesPayload = serializer.toBytes(object);
+      // For protobuf, we need to deserialize with the expected type
+      // For JSON with type hints, fromBytes(bytesPayload) works
+      if (serializer.isProtobuf(bytesPayload)) {
+        serializer.fromBytes(object.getClass(), bytesPayload);
+      } else {
+        serializer.fromBytes(bytesPayload);
+      }
     } catch (Exception e) {
       fail(object, entity, e);
     }
@@ -24,8 +34,8 @@ static void verifySerDer(Object object, Object entity) {
   /** different deserialization for responses, state, and commands */
   static void verifySerDerWithExpectedType(Class<?> expectedClass, Object object, Object entity) {
     try {
-      BytesPayload bytesPayload = jsonSerializer.toBytes(object);
-      jsonSerializer.fromBytes(expectedClass, bytesPayload);
+      BytesPayload bytesPayload = serializer.toBytes(object);
+      serializer.fromBytes(expectedClass, bytesPayload);
     } catch (Exception e) {
       fail(object, entity, e);
     }
@@ -34,8 +44,8 @@ static void verifySerDerWithExpectedType(Class<?> expectedClass, Object object,
   /** different deserialization for responses, state, and commands */
   static void verifySerDerWithExpectedType(Type expectedType, Object object, Object entity) {
     try {
-      BytesPayload bytesPayload = jsonSerializer.toBytes(object);
-      jsonSerializer.fromBytes(expectedType, bytesPayload);
+      BytesPayload bytesPayload = serializer.toBytes(object);
+      serializer.fromBytes(expectedType, bytesPayload);
     } catch (Exception e) {
       fail(object, entity, e);
     }

akka-javasdk-testkit/src/main/java/akka/javasdk/testkit/EventSourcedEntityEffectsRunner.java

@@ -13,30 +13,44 @@
 import akka.javasdk.testkit.impl.EventSourcedResultImpl;
 import akka.javasdk.testkit.impl.TestKitEventSourcedEntityCommandContext;
 import akka.javasdk.testkit.impl.TestKitEventSourcedEntityEventContext;
+import com.google.protobuf.GeneratedMessageV3;
 import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import scala.jdk.javaapi.CollectionConverters;
 
 /** Extended by generated code, not meant for user extension */
 abstract class EventSourcedEntityEffectsRunner<S, E> {
 
   private final Class<?> stateClass;
+  private final List<Class<? extends GeneratedMessageV3>> allowedProtoEventTypes;
   protected EventSourcedEntity<S, E> entity;
   private S _state;
   private boolean deleted = false;
   private List<E> events = new ArrayList<>();
 
+  @SuppressWarnings("unchecked")
+  private static List<Class<? extends GeneratedMessageV3>> getProtoEventTypes(
+      Class<?> entityClass) {
+    return new ArrayList<>(
+        (List<Class<? extends GeneratedMessageV3>>)
+            (List<?>) CollectionConverters.asJava(Reflect.protoEventTypes(entityClass)));
+  }
+
   public EventSourcedEntityEffectsRunner(EventSourcedEntity<S, E> entity) {
     this.entity = entity;
     this.stateClass = Reflect.eventSourcedEntityStateType(entity.getClass());
+    this.allowedProtoEventTypes = getProtoEventTypes(entity.getClass());
     this._state = entity.emptyState();
   }
 
   public EventSourcedEntityEffectsRunner(EventSourcedEntity<S, E> entity, S initialState) {
     this.entity = entity;
     this.stateClass = Reflect.eventSourcedEntityStateType(entity.getClass());
+    this.allowedProtoEventTypes = getProtoEventTypes(entity.getClass());
     verifySerDerWithExpectedType(stateClass, initialState, entity);
     this._state = initialState;
   }
@@ -45,6 +59,7 @@ public EventSourcedEntityEffectsRunner(EventSourcedEntity<S, E> entity, List<E>
     this.entity = entity;
     this._state = entity.emptyState();
     this.stateClass = Reflect.eventSourcedEntityStateType(entity.getClass());
+    this.allowedProtoEventTypes = getProtoEventTypes(entity.getClass());
     entity._internalSetCurrentState(this._state, false);
     // NB: updates _state
     playEventsForEntity(initialEvents);
@@ -80,6 +95,30 @@ public List<E> getAllEvents() {
     return events;
   }
 
+  /**
+   * Validates that events are of allowed types when @ProtoEventTypes is used. Throws
+   * IllegalArgumentException if an event is not one of the declared types.
+   */
+  private void validateProtoEventTypes(List<E> events) {
+    if (!allowedProtoEventTypes.isEmpty()) {
+      for (E event : events) {
+        Class<?> eventClass = event.getClass();
+        boolean isAllowed =
+            allowedProtoEventTypes.stream()
+                .anyMatch(allowed -> allowed.isAssignableFrom(eventClass));
+        if (!isAllowed) {
+          String allowedTypesStr =
+              allowedProtoEventTypes.stream().map(Class::getName).collect(Collectors.joining(", "));
+          throw new IllegalArgumentException(
+              String.format(
+                  "Event Sourced Entity [%s] tried to persist event of type [%s] "
+                      + "which is not declared in @ProtoEventTypes. Allowed types are: [%s]",
+                  entity.getClass().getName(), eventClass.getName(), allowedTypesStr));
+        }
+      }
+    }
+  }
+
   /**
    * creates a command context to run the commands, then creates an event context to run the events,
    * and finally, creates a command context to run the side effects. It cleans each context after
@@ -100,7 +139,10 @@ protected <R> EventSourcedResult<R> interpretEffects(
       entity._internalSetCommandContext(Optional.of(commandContext));
       entity._internalSetCurrentState(this._state, this.deleted);
       effectExecuted = effect.get();
-      this.events.addAll(EventSourcedResultImpl.eventsOf(effectExecuted));
+      // Validate proto event types before adding events
+      List<E> newEvents = EventSourcedResultImpl.eventsOf(effectExecuted);
+      validateProtoEventTypes(newEvents);
+      this.events.addAll(newEvents);
     } finally {
       entity._internalSetCommandContext(Optional.empty());
     }

akka-javasdk-testkit/src/main/java/akka/javasdk/testkit/EventingTestKit.java

@@ -7,7 +7,7 @@
 import akka.actor.typed.ActorSystem;
 import akka.annotation.InternalApi;
 import akka.javasdk.Metadata;
-import akka.javasdk.impl.serialization.JsonSerializer;
+import akka.javasdk.impl.serialization.Serializer;
 import akka.javasdk.testkit.impl.EventingTestKitImpl;
 import akka.javasdk.testkit.impl.OutgoingMessagesImpl;
 import akka.javasdk.testkit.impl.TestKitMessageImpl;
@@ -20,7 +20,7 @@ public interface EventingTestKit {
   /** INTERNAL API */
   @InternalApi
   static EventingTestKit start(
-      ActorSystem<?> system, String host, int port, JsonSerializer serializer) {
+      ActorSystem<?> system, String host, int port, Serializer serializer) {
     return EventingTestKitImpl.start(system, host, port, serializer);
   }
 
@@ -213,9 +213,9 @@ interface OutgoingMessages {
   }
 
   class MessageBuilder {
-    private final JsonSerializer serializer;
+    private final Serializer serializer;
 
-    public MessageBuilder(JsonSerializer serializer) {
+    public MessageBuilder(Serializer serializer) {
       this.serializer = serializer;
     }
 

akka-javasdk-testkit/src/main/java/akka/javasdk/testkit/SerializationTestkit.java

@@ -4,35 +4,46 @@
 
 package akka.javasdk.testkit;
 
-import akka.javasdk.impl.serialization.JsonSerializer;
+import akka.javasdk.impl.serialization.Serializer;
 import akka.runtime.sdk.spi.BytesPayload;
 import akka.util.ByteString;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import java.io.IOException;
 
-/** Helper class for serializing and deserializing objects for testing schema migration. */
+/**
+ * Helper class for serializing and deserializing objects for testing schema migration. Supports
+ * both JSON and Protobuf serialization formats.
+ */
 public final class SerializationTestkit {
 
   private record SerializedPayload(String contentType, byte[] bytes) {}
 
-  private static final JsonSerializer jsonSerializer = new JsonSerializer();
+  private static final Serializer serializer = new Serializer();
 
+  /**
+   * Serialize a value to bytes. Automatically uses protobuf format for protobuf messages and JSON
+   * format for other types.
+   */
   public static <T> byte[] serialize(T value) {
-    BytesPayload bytesPayload = jsonSerializer.toBytes(value);
+    BytesPayload bytesPayload = serializer.toBytes(value);
     SerializedPayload serializedPayload =
         new SerializedPayload(bytesPayload.contentType(), bytesPayload.bytes().toArray());
     try {
-      return jsonSerializer.objectMapper().writeValueAsBytes(serializedPayload);
+      return serializer.json().objectMapper().writeValueAsBytes(serializedPayload);
     } catch (JsonProcessingException e) {
       throw new RuntimeException("Unexpected serialization error", e);
     }
   }
 
+  /**
+   * Deserialize bytes to the expected type. Automatically detects the content type and uses the
+   * appropriate deserializer.
+   */
   public static <T> T deserialize(Class<T> valueClass, byte[] bytes) {
     try {
       SerializedPayload serializedPayload =
-          jsonSerializer.objectMapper().readValue(bytes, SerializedPayload.class);
-      return jsonSerializer.fromBytes(
+          serializer.json().objectMapper().readValue(bytes, SerializedPayload.class);
+      return serializer.fromBytes(
           valueClass,
           new BytesPayload(
               ByteString.fromArray(serializedPayload.bytes), serializedPayload.contentType));

akka-javasdk-testkit/src/main/java/akka/javasdk/testkit/TestKit.java

@@ -30,7 +30,7 @@
 import akka.javasdk.impl.client.ComponentClientImpl;
 import akka.javasdk.impl.grpc.GrpcClientProviderImpl;
 import akka.javasdk.impl.http.HttpClientImpl;
-import akka.javasdk.impl.serialization.JsonSerializer;
+import akka.javasdk.impl.serialization.Serializer;
 import akka.javasdk.impl.timer.TimerSchedulerImpl;
 import akka.javasdk.keyvalueentity.KeyValueEntity;
 import akka.javasdk.testkit.EventingTestKit.IncomingMessages;
@@ -689,7 +689,7 @@ private void startEventingTestkit() {
       // after discovery happens
       eventingTestKit =
           EventingTestKit.start(
-              runtimeActorSystem, "0.0.0.0", eventingTestKitPort, new JsonSerializer());
+              runtimeActorSystem, "0.0.0.0", eventingTestKitPort, new Serializer());
     }
   }
 
@@ -774,7 +774,7 @@ public SpiSettings getSettings() {
       final Sdk.StartupContext startupContext =
           runner.started().toCompletableFuture().get(20, TimeUnit.SECONDS);
       final ComponentClients componentClients = startupContext.componentClients();
-      final JsonSerializer serializer = startupContext.serializer();
+      final Serializer serializer = startupContext.serializer();
       dependencyProvider =
           Optional.ofNullable(startupContext.dependencyProvider().getOrElse(() -> null));
       sanitizer = startupContext.sanitizer();

akka-javasdk-testkit/src/main/scala/akka/javasdk/testkit/impl/EventingTestKitImpl.scala

@@ -32,7 +32,7 @@ import akka.javasdk.JsonSupport
 import akka.javasdk.Metadata.{ MetadataEntry => SdkMetadataEntry }
 import akka.javasdk.impl.AnySupport
 import akka.javasdk.impl.MetadataImpl
-import akka.javasdk.impl.serialization.JsonSerializer
+import akka.javasdk.impl.serialization.Serializer
 import akka.javasdk.testkit.EventingTestKit
 import akka.javasdk.testkit.EventingTestKit.IncomingMessages
 import akka.javasdk.testkit.EventingTestKit.OutgoingMessages
@@ -79,7 +79,7 @@ object EventingTestKitImpl {
    * The returned testkit can be used to expect and emit events to the proxy as if they came from an actual pub/sub
    * event backend.
    */
-  def start(system: ActorSystem[_], host: String, port: Int, serializer: JsonSerializer): EventingTestKit = {
+  def start(system: ActorSystem[_], host: String, port: Int, serializer: Serializer): EventingTestKit = {
 
     // Create service handlers
     val service = new EventingTestServiceImpl(system, host, port, serializer)
@@ -161,7 +161,7 @@ object EventingTestKitImpl {
  * Implements the EventingTestKit protocol originally defined in proxy
  * protocols/testkit/src/main/protobuf/eventing_test_backend.proto
  */
-final class EventingTestServiceImpl(system: ActorSystem[_], val host: String, var port: Int, serializer: JsonSerializer)
+final class EventingTestServiceImpl(system: ActorSystem[_], val host: String, var port: Int, serializer: Serializer)
     extends EventingTestKit {
 
   private val log = LoggerFactory.getLogger(classOf[EventingTestServiceImpl])
@@ -297,7 +297,7 @@ final class EventingTestServiceImpl(system: ActorSystem[_], val host: String, va
   }
 }
 
-private[testkit] class IncomingMessagesImpl(val sourcesHolder: ActorRef, val serializer: JsonSerializer)
+private[testkit] class IncomingMessagesImpl(val sourcesHolder: ActorRef, val serializer: Serializer)
     extends IncomingMessages {
 
   def addSourceProbe(runningSourceProbe: RunningSourceProbe): Unit = {
@@ -344,9 +344,7 @@ private[testkit] class IncomingMessagesImpl(val sourcesHolder: ActorRef, val ser
     "Publishing a delete message is supported only for ValueEntity messages.")
 }
 
-private[testkit] class VeIncomingMessagesImpl(
-    override val sourcesHolder: ActorRef,
-    override val serializer: JsonSerializer)
+private[testkit] class VeIncomingMessagesImpl(override val sourcesHolder: ActorRef, override val serializer: Serializer)
     extends IncomingMessagesImpl(sourcesHolder, serializer) {
 
   override def publishDelete(subject: String): Unit = {
@@ -362,7 +360,7 @@ private[testkit] class VeIncomingMessagesImpl(
 
 private[testkit] class OutgoingMessagesImpl(
     private[testkit] val destinationProbe: TestProbe,
-    protected val serializer: JsonSerializer)
+    protected val serializer: Serializer)
     extends OutgoingMessages {
   import EventingTestKitImpl.metadataToSpi
 
@@ -474,7 +472,7 @@ private[testkit] object TestKitMessageImpl {
     TestKitMessageImpl[ByteString](m.payload, metadata).asInstanceOf[TestKitMessage[ByteString]]
   }
 
-  def defaultMetadata(message: Any, subject: String, serializer: JsonSerializer): SdkMetadata = {
+  def defaultMetadata(message: Any, subject: String, serializer: Serializer): SdkMetadata = {
     val (contentType, ceType) = message match {
       case pbMsg: GeneratedMessageV3 =>
         val desc = pbMsg.getDescriptorForType

akka-javasdk-testkit/src/test/java/akka/javasdk/testkit/eventsourced/ProtoEventSourcedEntity.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (C) 2021-2026 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.javasdk.testkit.eventsourced;
+
+import akka.javasdk.annotations.Component;
+import akka.javasdk.annotations.ProtoEventTypes;
+import akka.javasdk.eventsourcedentity.EventSourcedEntity;
+import com.google.protobuf.Duration;
+import com.google.protobuf.GeneratedMessageV3;
+import com.google.protobuf.StringValue;
+import com.google.protobuf.Timestamp;
+
+/**
+ * Event sourced entity with @ProtoEventTypes for testing validation in the testkit. Uses standard
+ * Google protobuf types so no custom proto files are needed.
+ */
+@Component(id = "test-proto-entity")
+@ProtoEventTypes({StringValue.class, Duration.class})
+public class ProtoEventSourcedEntity extends EventSourcedEntity<String, GeneratedMessageV3> {
+
+  @Override
+  public String emptyState() {
+    return "";
+  }
+
+  @Override
+  public String applyEvent(GeneratedMessageV3 event) {
+    return switch (event) {
+      case StringValue sv -> sv.getValue();
+      case Duration d -> String.valueOf(d.getSeconds());
+      default -> throw new IllegalArgumentException("Unknown event type: " + event.getClass());
+    };
+  }
+
+  /** Command handler that persists an allowed event type (StringValue). */
+  public Effect<String> persistAllowedEvent(String value) {
+    return effects().persist(StringValue.of(value)).thenReply(__ -> "ok");
+  }
+
+  /** Command handler that persists another allowed event type (Duration). */
+  public Effect<String> persistAllowedEventDuration(long seconds) {
+    return effects()
+        .persist(Duration.newBuilder().setSeconds(seconds).build())
+        .thenReply(__ -> "ok");
+  }
+
+  /**
+   * Command handler that tries to persist an event type NOT in @ProtoEventTypes. Timestamp is NOT
+   * in the annotation, so this should fail with validation error.
+   */
+  public Effect<String> persistNotAllowedEvent(long seconds) {
+    return effects()
+        .persist(Timestamp.newBuilder().setSeconds(seconds).build())
+        .thenReply(__ -> "should not reach here");
+  }
+}