From 96d0e74c1fb2cac19f5e2784942cd0d1ebf4b1a5 Mon Sep 17 00:00:00 2001
From: Ben Papillon <benpapillon@gmail.com>
Date: Fri, 4 Sep 2026 16:51:30 -0700
Subject: [PATCH 1/2] tolerate unknown datastream envelope fields

The datastream server added stream_id to every message; Jackson's default
FAIL_ON_UNKNOWN_PROPERTIES made the client drop every message, so no flag,
company or user update was ever applied. Ignore unknown fields on the
envelope types and expose stream_id for replay.
---
 .../api/datastream/DataStreamMessages.java      | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/src/main/java/com/schematic/api/datastream/DataStreamMessages.java b/src/main/java/com/schematic/api/datastream/DataStreamMessages.java
index 73c0205d..fccfd4d1 100644
--- a/src/main/java/com/schematic/api/datastream/DataStreamMessages.java
+++ b/src/main/java/com/schematic/api/datastream/DataStreamMessages.java
@@ -1,5 +1,6 @@
 package com.schematic.api.datastream;
 
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.JsonNode;
 import java.util.Map;
@@ -88,6 +89,8 @@ public static MessageType fromString(String value) {
     }
 
     /** Request message sent to the datastream server. */
+    // The server may add envelope fields (e.g. stream_id); never reject a message for one.
+    @JsonIgnoreProperties(ignoreUnknown = true)
     public static class DataStreamReq {
         @JsonProperty("action")
         private final Action action;
@@ -118,6 +121,8 @@ public Map<String, String> getKeys() {
     }
 
     /** Wrapper for request messages. */
+    // The server may add envelope fields (e.g. stream_id); never reject a message for one.
+    @JsonIgnoreProperties(ignoreUnknown = true)
     public static class DataStreamBaseReq {
         @JsonProperty("data")
         private final DataStreamReq data;
@@ -132,6 +137,8 @@ public DataStreamReq getData() {
     }
 
     /** Response message received from the datastream server. */
+    // The server may add envelope fields (e.g. stream_id); never reject a message for one.
+    @JsonIgnoreProperties(ignoreUnknown = true)
     public static class DataStreamResp {
         @JsonProperty("data")
         private JsonNode data;
@@ -145,8 +152,16 @@ public static class DataStreamResp {
         @JsonProperty("message_type")
         private String messageType;
 
+        /** Server-side stream ID of the message; absent on snapshots. Recorded for replay on reconnect. */
+        @JsonProperty("stream_id")
+        private String streamId;
+
         public DataStreamResp() {}
 
+        public String getStreamId() {
+            return streamId;
+        }
+
         public JsonNode getData() {
             return data;
         }
@@ -173,6 +188,8 @@ public String getMessageType() {
     }
 
     /** Error message from the datastream server. */
+    // The server may add envelope fields (e.g. stream_id); never reject a message for one.
+    @JsonIgnoreProperties(ignoreUnknown = true)
     public static class DataStreamError {
         @JsonProperty("error")
         private String error;

From aea8a7be91b929a98952ae5fce9971688e7364ab Mon Sep 17 00:00:00 2001
From: Ben Papillon <benpapillon@gmail.com>
Date: Fri, 4 Sep 2026 17:02:04 -0700
Subject: [PATCH 2/2] chore(tests): cover datastream envelope stream_id and
 unknown fields

---
 .../datastream/DataStreamMessagesTest.java    | 35 +++++++++++++++++++
 1 file changed, 35 insertions(+)
 create mode 100644 src/test/java/com/schematic/api/datastream/DataStreamMessagesTest.java

diff --git a/src/test/java/com/schematic/api/datastream/DataStreamMessagesTest.java b/src/test/java/com/schematic/api/datastream/DataStreamMessagesTest.java
new file mode 100644
index 00000000..4e8c6dc0
--- /dev/null
+++ b/src/test/java/com/schematic/api/datastream/DataStreamMessagesTest.java
@@ -0,0 +1,35 @@
+package com.schematic.api.datastream;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.schematic.api.datastream.DataStreamMessages.DataStreamError;
+import com.schematic.api.datastream.DataStreamMessages.DataStreamResp;
+import org.junit.jupiter.api.Test;
+
+public class DataStreamMessagesTest {
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    @Test
+    public void responseParsesStreamIdAndIgnoresUnknownEnvelopeFields() throws Exception {
+        String json = "{\"data\":{\"id\":\"flag_1\"},\"entity_id\":\"flag_1\",\"entity_type\":\"rulesengine.Flag\","
+                + "\"message_type\":\"full\",\"stream_id\":\"1725494400000-0\",\"some_future_field\":true}";
+
+        DataStreamResp resp = mapper.readValue(json, DataStreamResp.class);
+
+        assertEquals("flag_1", resp.getEntityId());
+        assertEquals("rulesengine.Flag", resp.getEntityType());
+        assertEquals("full", resp.getMessageType());
+        assertEquals("1725494400000-0", resp.getStreamId());
+        assertEquals("flag_1", resp.getData().get("id").asText());
+    }
+
+    @Test
+    public void errorIgnoresUnknownEnvelopeFields() throws Exception {
+        String json = "{\"error\":\"not found\",\"entity_type\":\"rulesengine.Company\",\"stream_id\":\"x\"}";
+
+        DataStreamError err = mapper.readValue(json, DataStreamError.class);
+
+        assertEquals("not found", err.getError());
+    }
+}