Dvc 4209 event queue local bucketing methods

src/main/java/com/devcycle/sdk/server/common/model/User.java

@@ -12,16 +12,20 @@
 
 package com.devcycle.sdk.server.common.model;
 
+import com.devcycle.sdk.server.local.utils.LongTimestampDeserializer;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NonNull;
+import lombok.*;
+import lombok.extern.jackson.Jacksonized;
 
 @Data
 @Builder
+@Jacksonized
+@NoArgsConstructor
+@AllArgsConstructor
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public class User {
 
@@ -33,53 +37,70 @@ public class User {
   private String userId;
 
   @Schema(description = "User's email used to identify the user on the dashboard / target audiences")
+  @JsonProperty("email")
   private String email;
 
   @Schema(description = "User's name used to identify the user on the dashboard / target audiences")
+  @JsonProperty("name")
   private String name;
 
   @Schema(description = "User's language in ISO 639-1 format")
+  @JsonProperty("language")
   private String language;
 
   @Schema(description = "User's country in ISO 3166 alpha-2 format")
+  @JsonProperty("country")
   private String country;
 
   @Schema(description = "App Version of the running application")
+  @JsonProperty("appVersion")
   private String appVersion;
 
   @Schema(description = "App Build number of the running application")
+  @JsonProperty("appBuild")
   private String appBuild;
 
   @Schema(description = "User's custom data to target the user with, data will be logged to DevCycle for use in dashboard.")
+  @JsonProperty("customData")
   private Object customData;
 
   @Schema(description = "User's custom data to target the user with, data will not be logged to DevCycle only used for feature bucketing.")
+  @JsonProperty("privateCustomData")
   private Object privateCustomData;
 
   @Schema(description = "Date the user was created, Unix epoch timestamp format")
+  @JsonProperty("createdDate")
+  @JsonDeserialize(using = LongTimestampDeserializer.class)
   private Long createdDate;
 
   @Schema(description = "Date the user was last seen, Unix epoch timestamp format")
+  @JsonProperty("lastSeenDate")
+  @JsonDeserialize(using = LongTimestampDeserializer.class)
   private Long lastSeenDate;
 
   @Schema(description = "Platform the SDK is running on")
   @Builder.Default
+  @JsonProperty("platform")
   private String platform = "Java";
 
   @Schema(description = "Version of the platform the SDK is running on")
   @Builder.Default
+  @JsonProperty("platformVersion")
   private String platformVersion = System.getProperty("java.version");
 
   @Schema(description = "User's device model")
   @Builder.Default
+  @JsonProperty("deviceModel")
   private String deviceModel = "";
 
   @Schema(description = "DevCycle SDK type")
   @Builder.Default
+  @JsonProperty("sdkType")
   private PlatformData.SdkTypeEnum sdkType = PlatformData.SdkTypeEnum.SERVER;
 
   @Schema(description = "DevCycle SDK Version")
   @Builder.Default
+  @JsonProperty("sdkVersion")
   private String sdkVersion = "1.1.0";
 
   public PlatformData getPlatformData() {
@@ -90,4 +111,4 @@ public PlatformData getPlatformData() {
             .sdkVersion(sdkVersion)
             .build();
   }
-}
+}

src/main/java/com/devcycle/sdk/server/local/api/DVCLocalClient.java

@@ -5,7 +5,7 @@
 
 import com.devcycle.sdk.server.common.model.*;
 import com.devcycle.sdk.server.local.bucketing.LocalBucketing;
-import com.devcycle.sdk.server.local.configManager.EnvironmentConfigManager;
+import com.devcycle.sdk.server.local.managers.EnvironmentConfigManager;
 import com.devcycle.sdk.server.local.model.BucketedUserConfig;
 import com.devcycle.sdk.server.local.model.DVCLocalOptions;
 import com.fasterxml.jackson.annotation.JsonInclude;

src/main/java/com/devcycle/sdk/server/local/bucketing/LocalBucketing.java

@@ -7,14 +7,16 @@
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.Collection;
+import java.text.SimpleDateFormat;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.devcycle.sdk.server.local.model.BucketedUserConfig;
+import com.devcycle.sdk.server.local.model.EventPayload;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import com.fasterxml.jackson.databind.SerializationFeature;
 import io.github.kawamuray.wasmtime.*;
 import io.github.kawamuray.wasmtime.Module;
 
@@ -61,6 +63,11 @@ private Collection<Extern> setImportsOnLinker() {
         });
         linker.define("env", "abort", Extern.fromFunc(abortFn));
 
+        Func seedFn = WasmFunctions.wrap(store, F64, () -> {
+            return System.currentTimeMillis() * Math.random();
+        });
+        linker.define("env", "seed", Extern.fromFunc(seedFn));
+
         return Arrays.asList(Extern.fromFunc(dateNowFn), Extern.fromFunc(consoleLogFn), Extern.fromFunc(abortFn));
     }
 
@@ -143,5 +150,75 @@ public BucketedUserConfig generateBucketedConfig(String token, String user) thro
         BucketedUserConfig config = objectMapper.readValue(bucketedConfigString, BucketedUserConfig.class);
         return config;
     }
+
+    public void initEventQueue(String token, String options) {
+        int tokenAddress = newWasmString(token);
+        int optionsAddress = newWasmString(options);
+
+        Func initEventQueuePtr = linker.get(store, "", "initEventQueue").get().func();
+        WasmFunctions.Consumer2<Integer, Integer> fn = WasmFunctions.consumer(store, initEventQueuePtr, I32, I32);
+        fn.accept(tokenAddress, optionsAddress);
+    }
+
+    public void queueEvent(String token, String user, String event) {
+        int tokenAddress = newWasmString(token);
+        int userAddress = newWasmString(user);
+        int eventAddress = newWasmString(event);
+
+        Func queueEventPtr = linker.get(store, "", "queueEvent").get().func();
+        WasmFunctions.Consumer3<Integer, Integer, Integer> fn = WasmFunctions.consumer(store, queueEventPtr, I32, I32, I32);
+        fn.accept(tokenAddress, userAddress, eventAddress);
+    }
+
+    public void queueAggregateEvent(String token, String event, String variableVariationMap) {
+        int tokenAddress = newWasmString(token);
+        int eventAddress = newWasmString(event);
+        int variableVariationMapAddress = newWasmString(variableVariationMap);
+
+        Func queueAggregateEventPtr = linker.get(store, "", "queueAggregateEvent").get().func();
+        WasmFunctions.Consumer3<Integer, Integer, Integer> fn = WasmFunctions.consumer(store, queueAggregateEventPtr, I32, I32, I32);
+        fn.accept(tokenAddress, eventAddress, variableVariationMapAddress);
+    }
+
+    public EventPayload[] flushEventQueue(String token) throws JsonProcessingException {
+        int tokenAddress = newWasmString(token);
+
+        Func flushEventQueuePtr = linker.get(store, "", "flushEventQueue").get().func();
+        WasmFunctions.Function1<Integer, Integer> fn = WasmFunctions.func(
+                store, flushEventQueuePtr, I32, I32);
+
+        int resultAddress = fn.call(tokenAddress);
+        String flushPayloadsStr = readWasmString(resultAddress);
+
+        ObjectMapper objectMapper = new ObjectMapper();
+        objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+
+        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ"); //2022-09-08T20:16:31.741Z
+        df.setTimeZone(TimeZone.getTimeZone("UTC"));
+        objectMapper.setDateFormat(df);
+
+        EventPayload[] payloads = objectMapper.readValue(flushPayloadsStr, EventPayload[].class);
+
+        return payloads;
+    }
+
+    public void onPayloadFailure(String token, String payloadId, boolean retryable) {
+        // TODO - passing int value doesn't work yet (only addresses)
+//        int tokenAddress = newWasmString(token);
+//        int payloadIdAddress = newWasmString(payloadId);
+//
+//        Func onPayloadFailurePtr = linker.get(store, "", "onPayloadFailure").get().func();
+//        WasmFunctions.Consumer3<Integer, Integer, Integer> fn = WasmFunctions.consumer(store, onPayloadFailurePtr, I32, I32, I32);
+//        fn.accept(tokenAddress, payloadIdAddress, retryable ? 1 : 0);
+    }
+
+    public void onPayloadSuccess(String token, String payloadId) {
+        int tokenAddress = newWasmString(token);
+        int payloadIdAddress = newWasmString(payloadId);
+
+        Func onPayloadSuccessPtr = linker.get(store, "", "onPayloadSuccess").get().func();
+        WasmFunctions.Consumer2<Integer, Integer> fn = WasmFunctions.consumer(store, onPayloadSuccessPtr, I32, I32);
+        fn.accept(tokenAddress, payloadIdAddress);
+    }
 }
 

src/main/java/com/devcycle/sdk/server/local/configManager/EnvironmentConfigManager.java → src/main/java/com/devcycle/sdk/server/local/managers/EnvironmentConfigManager.java

@@ -1,4 +1,4 @@
-package com.devcycle.sdk.server.local.configManager;
+package com.devcycle.sdk.server.local.managers;
 
 import java.io.IOException;
 import java.util.concurrent.Executors;

src/main/java/com/devcycle/sdk/server/local/managers/EventQueueManager.java

@@ -0,0 +1,21 @@
+package com.devcycle.sdk.server.local.managers;
+
+import com.devcycle.sdk.server.local.bucketing.LocalBucketing;
+
+public class EventQueueManager {
+
+    private final LocalBucketing localBucketing;
+
+    public EventQueueManager(LocalBucketing localBucketing) {
+        this.localBucketing = localBucketing;
+    }
+
+    // TODO copy https://github.com/DevCycleHQ/js-sdks/blob/629b1f0a0d2cbed36bfe364d69f1929f645046db/sdk/nodejs/src/eventQueueAS.ts#L53
+    // to fill in these functions, then call from DVCLocalClient
+
+    //flushEvents
+
+    //queueEvent
+
+    //queueAggregateEvent
+}

src/main/java/com/devcycle/sdk/server/local/model/BucketedUserConfig.java

@@ -3,6 +3,7 @@
 import com.devcycle.sdk.server.common.model.Feature;
 import com.devcycle.sdk.server.common.model.Variable;
 
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -14,4 +15,5 @@ public class BucketedUserConfig {
     public Map<String, Variable> internalVariables;
     public Map<String, Variable> variables;
     public List<Double> knownVariableKeys;
+    public Map<String, FeatureVariation> variableVariationMap;
 }

src/main/java/com/devcycle/sdk/server/local/model/EventPayload.java

@@ -0,0 +1,15 @@
+package com.devcycle.sdk.server.local.model;
+
+import com.devcycle.sdk.server.common.model.User;
+
+public class EventPayload {
+    public Record[] records;
+    public String payloadId;
+    public int eventCount;
+
+    public static class Record {
+        public User user;
+        public RequestEvent[] events;
+    }
+}
+

src/main/java/com/devcycle/sdk/server/local/model/FeatureVariation.java

@@ -0,0 +1,6 @@
+package com.devcycle.sdk.server.local.model;
+
+public class FeatureVariation {
+    public String _feature;
+    public String _variation;
+}

src/main/java/com/devcycle/sdk/server/local/model/RequestEvent.java

@@ -0,0 +1,50 @@
+
+package com.devcycle.sdk.server.local.model;
+
+import com.devcycle.sdk.server.local.utils.LongTimestampDeserializer;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.math.BigDecimal;
+import java.util.Map;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class RequestEvent {
+
+    @Schema(required = true, description = "Custom event type")
+    private String customType;
+
+    @Schema(description = "Event type")
+    @Builder.Default
+    private String type = "customEvent";
+
+    @Schema(description = "User ID")
+    private String user_id;
+
+    @Schema(description = "Custom event target / subject of event. Contextual to event type")
+    private String target;
+
+    @Schema(description = "Unix epoch time the event occurred according to client")
+    @JsonDeserialize(using = LongTimestampDeserializer.class)
+    private Long date;
+
+    @Schema(description = "Unix epoch time the event occurred according to client")
+    @JsonDeserialize(using = LongTimestampDeserializer.class)
+    private Long clientDate;
+
+    @Schema(description = "Value for numerical events. Contextual to event type")
+    private BigDecimal value;
+
+    @Schema(description = "Extra JSON metadata for event. Contextual to event type")
+    private Object metaData;
+
+    @Schema(description = "Feature variation map")
+    private Map<String, String> featureVars;
+}

src/main/java/com/devcycle/sdk/server/local/utils/LongTimestampDeserializer.java

@@ -0,0 +1,27 @@
+package com.devcycle.sdk.server.local.utils;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import com.fasterxml.jackson.databind.exc.InvalidFormatException;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.time.format.DateTimeParseException;
+
+public class LongTimestampDeserializer extends StdDeserializer<Long> {
+    public LongTimestampDeserializer() {
+        super(Long.class);
+    }
+    @Override
+    public Long deserialize(JsonParser parser, DeserializationContext ctxt)
+            throws IOException {
+        String timestamp = parser.getText();
+        try {
+            return Instant.parse(timestamp).toEpochMilli();
+        } catch (DateTimeParseException dtpe) {
+            throw new InvalidFormatException(
+                    parser, dtpe.getMessage(), timestamp, Long.class);
+        }
+    }
+}