Fix/graph import streams

datahub-api-model/build.gradle

@@ -52,6 +52,11 @@ dependencies {
 
     compileOnly 'org.projectlombok:lombok'
     annotationProcessor 'org.projectlombok:lombok'
+
+    // Test-only: the module ships the constraint annotations but no provider, so a test that
+    // asserts a constraint actually fires needs an implementation to run them. testImplementation,
+    // so nothing framework-shaped reaches the published artifact.
+    testImplementation 'org.springframework.boot:spring-boot-starter-validation'
 }
 
 // Library module: no main class, so don't build an executable boot jar.

datahub-api-model/src/main/java/ai/intellistream/datahub/api/responses/DataWrapper.java

@@ -2,9 +2,11 @@
 package ai.intellistream.datahub.api.responses;
 
 import ai.intellistream.datahub.models.policy.PolicyWarning;
+import ai.intellistream.datahub.models.validation.FieldLimits;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonRootName;
 import jakarta.validation.Valid;
+import jakarta.validation.constraints.Size;
 import tools.jackson.dataformat.xml.annotation.JacksonXmlElementWrapper;
 import io.swagger.v3.oas.annotations.media.Schema;
 
@@ -15,8 +17,12 @@
 @Schema(name="DataWrapper", description="DataWrapper with items")
 public class DataWrapper<T> {
 
+    // Bounded on the request side: every create/update/delete endpoint that is not GraphDataWrapper
+    // binds this envelope, and an unbounded items[] is how a caller turns many small entities into
+    // bulk storage. Responses are never bean-validated, so paging is unaffected.
     @JacksonXmlElementWrapper(useWrapping = false)
     @Valid
+    @Size(max = FieldLimits.BATCH_ITEMS_MAX)
     private Collection<T> items = new ArrayList<>();
 
     /**

datahub-api-model/src/main/java/ai/intellistream/datahub/api/responses/DatapointString.java

@@ -1,9 +1,11 @@
 // SPDX-License-Identifier: Apache-2.0
 package ai.intellistream.datahub.api.responses;
 
+import ai.intellistream.datahub.models.validation.FieldLimits;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Size;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
@@ -42,6 +44,7 @@ public class DatapointString {
     private String timestamp;
 
     @NotBlank
+    @Size(max = FieldLimits.DATAPOINT_VALUE_MAX)
     @Schema(description = """
             The data point value as a string, interpreted per the timeseries `valueType`:
             - `BIGINT` — whole number, max 8 bytes. No fractional part.

datahub-api-model/src/main/java/ai/intellistream/datahub/api/responses/DatapointsCollection.java

@@ -3,7 +3,10 @@
 import tools.jackson.databind.annotation.JsonSerialize;
 import ai.intellistream.datahub.json.ToStringSerializer;
 
+import ai.intellistream.datahub.models.validation.FieldLimits;
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.Size;
 import lombok.Data;
 import lombok.Getter;
 import lombok.Setter;
@@ -24,6 +27,10 @@ public class DatapointsCollection {
             example = "a4545_well_pump_pressure_a")
     private String externalId;
 
+    // @Valid so the per-datapoint constraints actually cascade: without it neither @NotBlank nor the
+    // value-length cap on DatapointString was ever evaluated on an insert.
+    @Valid
+    @Size(max = FieldLimits.DATAPOINTS_PER_COLLECTION_MAX)
     private List<DatapointString> datapoints;
 
     @Schema(description = "If more than 100 000 datapoints are returned, a cursor hash value will be returned",

datahub-api-model/src/main/java/ai/intellistream/datahub/models/EventModel.java

@@ -6,6 +6,8 @@
 import ai.intellistream.datahub.helpers.datetime.DateTimeHandler;
 import com.fasterxml.jackson.annotation.*;
 import io.swagger.v3.oas.annotations.media.Schema;
+import ai.intellistream.datahub.models.validation.BoundedMetadata;
+import ai.intellistream.datahub.models.validation.FieldLimits;
 import ai.intellistream.datahub.models.validation.ForbiddenValues;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
@@ -53,12 +55,14 @@ public class EventModel extends AbstractResource{
     /**
      * Entity specific metadata. A key-value store.
      */
+    @BoundedMetadata
     @Schema(description = "Event metadata, additional fields you can bind information with.", example = "{\"definition\": \"SAP ORDER PLACED\"}")
     private Map<String, String> metadata = new HashMap<>();
 
     /**
      * The description of the event.
      */
+    @Size(max = FieldLimits.DESCRIPTION_MAX)
     @Schema(description = "The description of the event.", example = "This event was caused by....")
     private String description;
 
@@ -100,6 +104,7 @@ public class EventModel extends AbstractResource{
      * externalId, or both; the API resolves the missing side and always returns both. One list
      * rather than two parallel ones is deliberate — parallel id/externalId lists drift.
      */
+    @Size(max = FieldLimits.RELATED_RESOURCES_MAX)
     @Schema(description = "Resources that this event has a relation to. Supply id, externalId or both; both are returned.",
             example = "[{\"id\": 34, \"externalId\": \"sensor_abc\"}]")
     private List<IdCollection> relatedResources = new ArrayList<>();

datahub-api-model/src/main/java/ai/intellistream/datahub/models/GeoLocation.java

@@ -3,8 +3,10 @@
 
 import ai.intellistream.datahub.json.GeoLocationDeserializer;
 import ai.intellistream.datahub.json.GeoLocationSerializer;
+import ai.intellistream.datahub.models.validation.FieldLimits;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import jakarta.validation.constraints.AssertTrue;
+import jakarta.validation.constraints.Size;
 import tools.jackson.databind.annotation.JsonDeserialize;
 import tools.jackson.databind.annotation.JsonSerialize;
 import tools.jackson.databind.json.JsonMapper;
@@ -38,6 +40,7 @@ public class GeoLocation {
             "Polygon", "MultiPolygon", "GeometryCollection");
 
     /** The raw GeoJSON geometry, stored verbatim. */
+    @Size(max = FieldLimits.GEOJSON_MAX_CHARS)
     private String json;
 
     public GeoLocation() {

datahub-api-model/src/main/java/ai/intellistream/datahub/models/NodeModel.java

@@ -3,6 +3,8 @@
 
 import ai.intellistream.datahub.helpers.text.ExternalIds;
 import ai.intellistream.datahub.json.ToStringSerializer;
+import ai.intellistream.datahub.models.validation.BoundedMetadata;
+import ai.intellistream.datahub.models.validation.FieldLimits;
 import ai.intellistream.datahub.models.validation.ForbiddenValues;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -60,9 +62,11 @@ public abstract class NodeModel extends AbstractResource {
     @Schema(description = "The name of the object.", example = "klp pipe ws-a1212-dl")
     private String name;
 
+    @BoundedMetadata
     @Schema(description = "Entity specific metadata. A key-value store.", example = "{\"work_order\": \"wo-sap-12344\"}")
     private Map<String, String> metadata = new HashMap<>();
 
+    @Size(max = FieldLimits.DESCRIPTION_MAX)
     @Schema(description = "The description of the object.", example = "Water stream pipe")
     private String description;
 
@@ -101,8 +105,9 @@ public abstract class NodeModel extends AbstractResource {
      */
     @NotNull
     @Size(min = 1, message = "resource.needs.at.least.one.label")
+    @Size(max = FieldLimits.LABELS_MAX, message = "resource.too.many.labels")
     @Schema(description = "A list of the labels associated with this node.", example = "[\"resource\", \"PIPE\"]")
-    private List<String> labels = new ArrayList<>();
+    private List<@Size(max = FieldLimits.LABEL_LENGTH_MAX) String> labels = new ArrayList<>();
 
     /**
      * The unified node-centric relation encoding: the nodes this one is connected to, each with its
@@ -111,6 +116,7 @@ public abstract class NodeModel extends AbstractResource {
      * {@code relationsFrom}). Populated where the graph is loaded (see {@code ResourceNetwork}); empty
      * otherwise.
      */
+    @Size(max = FieldLimits.RELATED_RESOURCES_MAX)
     @Schema(description = "Nodes this node is connected to, with relationship type and direction.",
             example = "[{\"id\": 34, \"externalId\": \"sensor_abc\", \"relationshipType\": \"PUBLISHES_DATA_TO\", \"direction\": \"OUTBOUND\"}]")
     private List<RelatedNode> relatedResources = new ArrayList<>();

datahub-api-model/src/main/java/ai/intellistream/datahub/models/RelForm.java

@@ -4,8 +4,11 @@
 import ai.intellistream.datahub.json.ToStringSerializer;
 
 import ai.intellistream.datahub.helpers.text.TextValidator;
+import ai.intellistream.datahub.models.validation.BoundedMetadata;
+import ai.intellistream.datahub.models.validation.FieldLimits;
 import ai.intellistream.datahub.validation.resources.RelationshipTypeNotNull;
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.Size;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.Setter;
@@ -41,11 +44,13 @@ public class RelForm {
     @JsonSerialize(using = ToStringSerializer.class)
     private Long relationshipTypeId;
 
+    @BoundedMetadata
     private HashMap<String, String> metadata = new HashMap<>();
 
     @JsonSerialize(using = ToStringSerializer.class)
     private Long dataSetId;
 
+    @Size(max = FieldLimits.DESCRIPTION_MAX)
     private String description;
 
     public void setName(String name){

datahub-api-model/src/main/java/ai/intellistream/datahub/models/validation/BoundedMetadata.java

@@ -0,0 +1,32 @@
+// SPDX-License-Identifier: Apache-2.0
+package ai.intellistream.datahub.models.validation;
+
+import jakarta.validation.Constraint;
+import jakarta.validation.Payload;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Bounds a metadata map: how many entries it may hold, and how long each key and value may be.
+ * Defaults come from {@link FieldLimits}.
+ */
+@Constraint(validatedBy = BoundedMetadataValidator.class)
+@Target({ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.ANNOTATION_TYPE})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface BoundedMetadata {
+
+    String message() default "metadata.too.large";
+
+    Class<?>[] groups() default {};
+
+    Class<? extends Payload>[] payload() default {};
+
+    int maxEntries() default FieldLimits.METADATA_MAX_ENTRIES;
+
+    int maxKeyLength() default FieldLimits.METADATA_KEY_MAX;
+
+    int maxValueLength() default FieldLimits.METADATA_VALUE_MAX;
+}

datahub-api-model/src/main/java/ai/intellistream/datahub/models/validation/BoundedMetadataValidator.java

@@ -0,0 +1,48 @@
+// SPDX-License-Identifier: Apache-2.0
+package ai.intellistream.datahub.models.validation;
+
+import jakarta.validation.ConstraintValidator;
+import jakarta.validation.ConstraintValidatorContext;
+
+import java.util.Map;
+
+public class BoundedMetadataValidator implements ConstraintValidator<BoundedMetadata, Map<String, String>> {
+
+    private int maxEntries;
+    private int maxKeyLength;
+    private int maxValueLength;
+
+    @Override
+    public void initialize(BoundedMetadata constraintAnnotation) {
+        this.maxEntries = constraintAnnotation.maxEntries();
+        this.maxKeyLength = constraintAnnotation.maxKeyLength();
+        this.maxValueLength = constraintAnnotation.maxValueLength();
+    }
+
+    @Override
+    public boolean isValid(Map<String, String> value, ConstraintValidatorContext context) {
+        if (value == null || value.isEmpty()) {
+            return true;
+        }
+
+        if (value.size() > maxEntries) {
+            return fail(context, "metadata.too.many.entries");
+        }
+
+        for (Map.Entry<String, String> entry : value.entrySet()) {
+            if (entry.getKey() != null && entry.getKey().length() > maxKeyLength) {
+                return fail(context, "metadata.key.too.long");
+            }
+            if (entry.getValue() != null && entry.getValue().length() > maxValueLength) {
+                return fail(context, "metadata.value.too.long");
+            }
+        }
+        return true;
+    }
+
+    private boolean fail(ConstraintValidatorContext context, String messageKey) {
+        context.disableDefaultConstraintViolation();
+        context.buildConstraintViolationWithTemplate(messageKey).addConstraintViolation();
+        return false;
+    }
+}

datahub-api-model/src/main/java/ai/intellistream/datahub/models/validation/EventFields.java

@@ -111,6 +111,9 @@ public boolean validateFields(){
             }
         }
 
+        SizeRules.checkLength("Event", "event.description.max.length.error", "Description",
+                this.description.getSet(), FieldLimits.DESCRIPTION_MAX, errors);
+
         if(this.metadata.getSet() != null){
             Map<String, String> metadata = this.metadata.getSet();
             if(metadata.containsKey("")){
@@ -119,6 +122,13 @@ public boolean validateFields(){
             }
         }
 
+        SizeRules.checkMetadata("Event", "event", this.metadata, errors);
+
+        SizeRules.checkCount("Event", "event.related.resources.too.many", "Related resources",
+                this.relatedResources.getSet(), FieldLimits.RELATED_RESOURCES_MAX, errors);
+        SizeRules.checkCount("Event", "event.related.resources.too.many", "Related resources",
+                this.relatedResources.getAdd(), FieldLimits.RELATED_RESOURCES_MAX, errors);
+
         // An entry with neither side is unresolvable, so reject it here rather than letting the
         // service raise it — the caller gets one validation response with every bad field in it.
         Stream.of(this.relatedResources.getSet(), this.relatedResources.getAdd(), this.relatedResources.getRemove())

datahub-api-model/src/main/java/ai/intellistream/datahub/models/validation/FieldLimits.java

@@ -0,0 +1,55 @@
+// SPDX-License-Identifier: Apache-2.0
+package ai.intellistream.datahub.models.validation;
+
+/**
+ * Size ceilings for the free-form fields of the wire contract, in one place so the bean-validation
+ * annotations, the hand-written update validators and the OpenAPI schema documentation cannot drift
+ * apart.
+ *
+ * <p>These bound what a single entity can carry. Without them a caller can use {@code description}
+ * or {@code metadata} as file storage: both were unbounded at every layer, and events are not even
+ * subject to the incidental Postgres ceilings that nodes inherit from their indexes.
+ */
+public final class FieldLimits {
+
+    private FieldLimits() {
+    }
+
+    /** Long enough for a substantial prose description, short enough to be useless as a file. */
+    public static final int DESCRIPTION_MAX = 10_000;
+
+    /** Metadata is a tag store, not a document store. */
+    public static final int METADATA_MAX_ENTRIES = 256;
+
+    /** Well under the {@code node_metadata.key varchar(1024)} column. */
+    public static final int METADATA_KEY_MAX = 128;
+
+    /** Keeps key+value under the ~2.7 KB {@code (node_id, key, value)} btree index-tuple ceiling. */
+    public static final int METADATA_VALUE_MAX = 1_024;
+
+    public static final int RELATED_RESOURCES_MAX = 100;
+
+    public static final int LABELS_MAX = 64;
+
+    public static final int LABEL_LENGTH_MAX = 512;
+
+    /** A detailed polygon fits; a payload does not. */
+    public static final int GEOJSON_MAX_CHARS = 65_536;
+
+    /** Numeric strings and status codes fit. */
+    public static final int DATAPOINT_VALUE_MAX = 64;
+
+    /** Items in one {@code DataWrapper}. Matches the Java SDK's default ingest batch size. */
+    public static final int BATCH_ITEMS_MAX = 10_000;
+
+    /** Datapoints in one collection, for numeric series. */
+    public static final int DATAPOINTS_PER_COLLECTION_MAX = 100_000;
+
+    /**
+     * Datapoints in one collection for TEXT/MIXED series. Tighter than the numeric cap because a
+     * text batch is the one shape that can approach Pulsar's 5 MB per-message ceiling: this keeps a
+     * worst-case collection at roughly 640 KB of values. Enforced in the service, once the series'
+     * value type is known — bean validation runs before the series is resolved.
+     */
+    public static final int TEXT_DATAPOINTS_PER_COLLECTION_MAX = 10_000;
+}

datahub-api-model/src/main/java/ai/intellistream/datahub/models/validation/ResourceFields.java

@@ -13,8 +13,11 @@
 import ai.intellistream.datahub.validation.FieldValidationError;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Stream;
 
 /**
  * Resource fields mapping for updating the object
@@ -84,6 +87,9 @@ public boolean validateFields(){
             }
         }
 
+        SizeRules.checkLength("Resource", "resource.description.max.length.error", "Description",
+                this.description.getSet(), FieldLimits.DESCRIPTION_MAX, errors);
+
         if(this.metadata.getSet() != null){
             Map<String, String> metadata = this.metadata.getSet();
             if(metadata.containsKey("")){
@@ -92,6 +98,19 @@ public boolean validateFields(){
             }
         }
 
+        SizeRules.checkMetadata("Resource", "resource", this.metadata, errors);
+
+        SizeRules.checkCount("Resource", "resource.too.many.labels", "Labels",
+                this.labels.getSet(), FieldLimits.LABELS_MAX, errors);
+        SizeRules.checkCount("Resource", "resource.too.many.labels", "Labels",
+                this.labels.getAdd(), FieldLimits.LABELS_MAX, errors);
+
+        Stream.of(this.labels.getSet(), this.labels.getAdd())
+                .filter(Objects::nonNull)
+                .flatMap(Collection::stream)
+                .forEach(label -> SizeRules.checkLength("Resource", "resource.label.max.length.error",
+                        "Label", label, FieldLimits.LABEL_LENGTH_MAX, errors));
+
         if(this.geoLocation.getSet() != null && !this.geoLocation.getSet().isValidGeoJson()){
             errors.add(
                     new FieldValidationError(