diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/BackendRegistry.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/BackendRegistry.java new file mode 100644 index 0000000..71cebd9 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/BackendRegistry.java @@ -0,0 +1,66 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.spi.VelocityBackend; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * An immutable registry of the backends the engine can dispatch to, keyed by the backend name a + * feature definition binds to (FR-16). + * + *

Injected at engine construction (no Dagger here — DI lives in the service tier). A feature + * bound to an unknown backend is a configuration error surfaced immediately by {@link #backend}. + */ +public final class BackendRegistry { + + private final Map backends; + + /** + * Creates a registry from a name → backend map. + * + * @param backends the backends by name; copied defensively, must be non-empty with non-blank keys + */ + public BackendRegistry(final Map backends) { + Objects.requireNonNull(backends, "backends"); + if (backends.isEmpty()) { + throw new IllegalArgumentException("backend registry must not be empty"); + } + backends.forEach( + (name, backend) -> { + Objects.requireNonNull(name, "backend name"); + Objects.requireNonNull(backend, "backend for '" + name + "'"); + if (name.isBlank()) { + throw new IllegalArgumentException("backend name must not be blank"); + } + }); + this.backends = Map.copyOf(backends); + } + + /** + * Looks up a backend by name. + * + * @param name the backend name a definition binds to + * @return the backend + * @throws IllegalArgumentException if no backend is registered under {@code name} + */ + public VelocityBackend backend(final String name) { + Objects.requireNonNull(name, "name"); + final VelocityBackend backend = backends.get(name); + if (backend == null) { + throw new IllegalArgumentException( + "no backend registered under name '" + name + "'; known backends: " + backends.keySet()); + } + return backend; + } + + /** + * The registered backend names. + * + * @return the immutable set of names + */ + public Set names() { + return backends.keySet(); + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/CapabilityValidator.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/CapabilityValidator.java new file mode 100644 index 0000000..38d9d64 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/CapabilityValidator.java @@ -0,0 +1,90 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.core.model.CapabilityValidationResult; +import com.codeheadsystems.velocity.core.model.CapabilityValidationResult.Violation; +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.spi.model.AggregationType; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities; +import com.codeheadsystems.velocity.spi.model.Window; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; + +/** + * Validates feature definitions against their target backend's declared capabilities (FR-29). + * + *

The engine rejects an import/reload unless every definition is valid (atomic with FR-17); this + * validator returns all violations rather than throwing on the first, so the rejection can + * carry a complete, precise list. Fast-reject, not silent degradation (NFR-18): the engine never + * pretends a backend can serve a window/aggregation/retention it did not declare. + * + *

Checks, per definition, that: the aggregation is supported; every window is supported (matches + * a declared {@code WindowSpec}); the backend's {@code maxRetention} covers the largest window + * (FR-22a); and DISTINCT is only used where DISTINCT is supported. + */ +public final class CapabilityValidator { + + /** Creates a capability validator. */ + public CapabilityValidator() {} + + /** + * Validates one definition against a backend's capabilities. + * + * @param definition the definition to validate + * @param capabilities the target backend's declared capabilities + * @return the collected violations (empty when valid) + */ + public CapabilityValidationResult validate( + final FeatureDefinition definition, final BackendCapabilities capabilities) { + Objects.requireNonNull(definition, "definition"); + Objects.requireNonNull(capabilities, "capabilities"); + final List violations = new ArrayList<>(); + collect(definition, capabilities, violations); + return new CapabilityValidationResult(violations); + } + + private void collect( + final FeatureDefinition definition, + final BackendCapabilities capabilities, + final List violations) { + final AggregationType type = definition.aggregation().type(); + if (!capabilities.supportsAggregation(type)) { + violations.add( + new Violation( + definition.name(), + "aggregation " + + type + + " is not supported by backend '" + + definition.backend() + + "'")); + } + for (final Window window : definition.windows()) { + if (!capabilities.supportsWindow(window)) { + violations.add( + new Violation( + definition.name(), + "window " + + window.type() + + " " + + window.duration() + + " is not supported by backend '" + + definition.backend() + + "'")); + } + } + final var largest = definition.largestWindow(); + if (largest.compareTo(capabilities.maxRetention()) > 0) { + violations.add( + new Violation( + definition.name(), + "largest window " + + largest + + " exceeds backend '" + + definition.backend() + + "' maxRetention " + + capabilities.maxRetention() + + " (FR-22a)")); + } + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/DefinitionVersionHasher.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/DefinitionVersionHasher.java new file mode 100644 index 0000000..6808c30 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/DefinitionVersionHasher.java @@ -0,0 +1,99 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.core.model.SubjectSource; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.Window; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; + +/** + * Computes the deterministic {@code versionHash} stamped onto a {@code FeatureDefinitions} snapshot + * (FR-40). + * + *

The hash is a SHA-256 over a canonical, order-independent string form of the definitions: + * definitions are sorted by name and each is rendered to a stable field-delimited line, so the same + * set of definitions always hashes to the same value regardless of list order, and any change to + * any field changes the hash. A caller that sees a value stamped with a different hash than it last + * read knows the definition changed under it (FR-40, §15 R12). + */ +public final class DefinitionVersionHasher { + + private DefinitionVersionHasher() {} + + /** + * Computes the version hash of a definition set. + * + * @param definitions the definitions to hash (any order) + * @return a lowercase hex SHA-256 of the canonical form + */ + public static String hash(final List definitions) { + Objects.requireNonNull(definitions, "definitions"); + final List lines = new ArrayList<>(definitions.size()); + for (final FeatureDefinition definition : definitions) { + lines.add(canonical(definition)); + } + lines.sort(String::compareTo); + final String canonical = String.join("\n", lines); + return sha256Hex(canonical.getBytes(StandardCharsets.UTF_8)); + } + + private static String canonical(final FeatureDefinition definition) { + final StringBuilder builder = new StringBuilder(); + builder + .append(definition.name()) + .append('|') + .append(definition.subjectType()) + .append('|') + .append(subjectSource(definition.subjectSource())) + .append('|') + .append(definition.backend()) + .append('|') + .append(aggregation(definition.aggregation())) + .append('|') + .append(definition.distinctThreshold()) + .append('|'); + final List windows = new ArrayList<>(); + for (final Window window : definition.windows()) { + windows.add(window.type() + ":" + window.duration()); + } + windows.sort(String::compareTo); + builder.append(String.join(",", windows)); + return builder.toString(); + } + + private static String subjectSource(final SubjectSource source) { + return switch (source) { + case SubjectSource.Primary ignored -> "PRIMARY"; + case SubjectSource.FromDimension fromDimension -> + "FROM_DIMENSION:" + fromDimension.dimensionName(); + }; + } + + private static String aggregation(final Aggregation aggregation) { + return aggregation.type() + + (aggregation.dimension() == null ? "" : ":" + aggregation.dimension()); + } + + private static String sha256Hex(final byte[] bytes) { + final MessageDigest digest; + try { + digest = MessageDigest.getInstance("SHA-256"); + } catch (final NoSuchAlgorithmException e) { + // SHA-256 is a mandated JCA algorithm; its absence is an unrecoverable environment fault. + throw new IllegalStateException("SHA-256 not available", e); + } + final byte[] hashed = digest.digest(bytes); + final StringBuilder hex = new StringBuilder(hashed.length * 2); + for (final byte b : hashed) { + hex.append(Character.forDigit((b >> 4) & 0xF, 16)); + hex.append(Character.forDigit(b & 0xF, 16)); + } + return hex.toString(); + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/DimensionHasher.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/DimensionHasher.java new file mode 100644 index 0000000..59d121d --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/DimensionHasher.java @@ -0,0 +1,65 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.spi.model.DistinctMember; +import com.codeheadsystems.velocity.spi.model.Namespace; +import java.nio.charset.StandardCharsets; +import java.security.InvalidKeyException; +import java.util.Objects; +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; + +/** + * Hashes a raw DISTINCT dimension value into an opaque, fixed-width {@link DistinctMember} so raw + * PII (IPs, tokens, device IDs) never reaches a backend (FR-38, §15 R11). + * + *

The token is {@code HMAC-SHA256(salt, value)} — 32 bytes — with the per-namespace salt drawn + * from a {@link NamespaceSaltProvider}. The mapping is deterministic per {@code (namespace, value)} + * (so the same value counts once) and different values produce different members (so cardinality is + * preserved), while the backend only ever stores and counts the opaque token. Keyed hashing (not a + * bare digest) is what makes a low-entropy dimension resistant to offline brute-force when the + * salt lives in a separate trust domain (§15 R11); see {@link NamespaceSaltProvider}. + */ +public final class DimensionHasher { + + private static final String HMAC_SHA256 = "HmacSHA256"; + + private final NamespaceSaltProvider saltProvider; + + /** + * Creates a hasher backed by the given salt provider. + * + * @param saltProvider the per-namespace salt source + */ + public DimensionHasher(final NamespaceSaltProvider saltProvider) { + this.saltProvider = Objects.requireNonNull(saltProvider, "saltProvider"); + } + + /** + * Hashes a raw dimension value into an opaque distinct member. + * + * @param namespace the namespace (selects the salt) + * @param value the raw dimension value + * @return the 32-byte keyed-hash member + */ + public DistinctMember hash(final Namespace namespace, final String value) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(value, "value"); + final Mac mac = newMac(saltProvider.salt(namespace)); + final byte[] token = mac.doFinal(value.getBytes(StandardCharsets.UTF_8)); + return new DistinctMember(token); + } + + private static Mac newMac(final byte[] salt) { + try { + final Mac mac = Mac.getInstance(HMAC_SHA256); + mac.init(new SecretKeySpec(salt, HMAC_SHA256)); + return mac; + } catch (final java.security.NoSuchAlgorithmException e) { + // HmacSHA256 is a mandated JCA algorithm; its absence is an unrecoverable environment fault. + throw new IllegalStateException("HmacSHA256 not available", e); + } catch (final InvalidKeyException e) { + throw new IllegalStateException("invalid HMAC salt", e); + } + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FanOutResolver.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FanOutResolver.java new file mode 100644 index 0000000..8533161 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FanOutResolver.java @@ -0,0 +1,166 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.core.model.FanOutResult; +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.core.model.FeatureDefinitions; +import com.codeheadsystems.velocity.core.model.SubjectSource; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.ApplyStatus; +import com.codeheadsystems.velocity.spi.model.FailureCode; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.FeatureResult; +import com.codeheadsystems.velocity.spi.model.Intent; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.spi.model.Intent.DistinctIntent; +import com.codeheadsystems.velocity.spi.model.Intent.SumIntent; +import com.codeheadsystems.velocity.spi.model.Namespace; +import com.codeheadsystems.velocity.spi.model.PerFeature; +import com.codeheadsystems.velocity.spi.model.Subject; +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import org.jspecify.annotations.Nullable; + +/** + * Resolves one {@code record()} event into backend-neutral write intents grouped by backend + * (FR-18). + * + *

For each definition in the namespace's snapshot the resolver decides the target {@link + * Subject} (the event's own subject for a {@link SubjectSource.Primary primary} source when the + * type matches; a dimension-derived subject for a {@link SubjectSource.FromDimension} source when + * that dimension is present), builds the SPI {@link Feature} and the aggregation-specific {@link + * Intent}, keyed-hashing DISTINCT dimension values via the {@link DimensionHasher} (FR-38). A + * single event therefore fans out across multiple subjects and backends. A definition that applies + * to the event's subject but cannot be written — a SUM with no value, or a DISTINCT whose dimension + * is absent — becomes a {@code SKIPPED} outcome rather than being silently dropped. + */ +public final class FanOutResolver { + + private final DimensionHasher dimensionHasher; + + /** + * Creates a resolver. + * + * @param dimensionHasher the keyed hasher for DISTINCT dimension values (FR-38) + */ + public FanOutResolver(final DimensionHasher dimensionHasher) { + this.dimensionHasher = Objects.requireNonNull(dimensionHasher, "dimensionHasher"); + } + + /** + * Resolves the fan-out of one event. + * + * @param namespace the namespace (selects DISTINCT salt) + * @param primarySubject the event's own subject + * @param dimensions the event's dimensions (used for dimension-derived subjects and DISTINCT) + * @param valueCents the event's SUM value in integer cents, or null if absent + * @param definitions the namespace's current definition snapshot + * @return the intents grouped by backend plus any skipped-with-reason outcomes + */ + public FanOutResult resolve( + final Namespace namespace, + final Subject primarySubject, + final Map dimensions, + final @Nullable BigDecimal valueCents, + final FeatureDefinitions definitions) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(primarySubject, "primarySubject"); + Objects.requireNonNull(dimensions, "dimensions"); + Objects.requireNonNull(definitions, "definitions"); + + final Map> byBackend = new LinkedHashMap<>(); + final List skipped = new ArrayList<>(); + + for (final FeatureDefinition definition : definitions.definitions()) { + final Subject subject = targetSubject(definition, primarySubject, dimensions); + if (subject == null) { + continue; // The definition does not apply to this event (FR-4 no-op). + } + final Feature feature = + new Feature(definition.name(), definition.aggregation(), definition.windows()); + final Intent intent = + buildIntent(namespace, definition, feature, subject, dimensions, valueCents, skipped); + if (intent != null) { + byBackend.computeIfAbsent(definition.backend(), unused -> new ArrayList<>()).add(intent); + } + } + return new FanOutResult(byBackend, skipped); + } + + private static @Nullable Subject targetSubject( + final FeatureDefinition definition, + final Subject primarySubject, + final Map dimensions) { + return switch (definition.subjectSource()) { + case SubjectSource.Primary ignored -> + primarySubject.type().equals(definition.subjectType()) ? primarySubject : null; + case SubjectSource.FromDimension fromDimension -> { + final String value = dimensions.get(fromDimension.dimensionName()); + yield present(value) ? new Subject(definition.subjectType(), value) : null; + } + }; + } + + private @Nullable Intent buildIntent( + final Namespace namespace, + final FeatureDefinition definition, + final Feature feature, + final Subject subject, + final Map dimensions, + final @Nullable BigDecimal valueCents, + final List skipped) { + final Aggregation aggregation = definition.aggregation(); + return switch (aggregation.type()) { + case COUNT -> new CountIntent(feature, subject); + case SUM -> { + if (valueCents == null) { + skipped.add( + skip(feature, "SUM feature '" + feature.name() + "' requires an event value")); + yield null; + } + if (valueCents.scale() != 0) { + skipped.add( + skip( + feature, + "SUM feature '" + + feature.name() + + "' requires integer cents (scale 0), was scale " + + valueCents.scale())); + yield null; + } + yield new SumIntent(feature, subject, valueCents); + } + case DISTINCT -> { + // The aggregation's dimension is the value counted; may differ from a FromDimension + // subject. + final String dimensionName = Objects.requireNonNull(aggregation.dimension()); + final String value = dimensions.get(dimensionName); + if (!present(value)) { + skipped.add( + skip( + feature, + "DISTINCT feature '" + + feature.name() + + "' requires dimension '" + + dimensionName + + "'")); + yield null; + } + yield new DistinctIntent(feature, subject, dimensionHasher.hash(namespace, value)); + } + }; + } + + private static PerFeature skip(final Feature feature, final String reason) { + return new PerFeature( + feature, ApplyStatus.SKIPPED, FeatureResult.failure(FailureCode.VALIDATION, reason)); + } + + private static boolean present(final @Nullable String value) { + return value != null && !value.isBlank(); + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FeatureDefinitionProvider.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FeatureDefinitionProvider.java new file mode 100644 index 0000000..5c6c01a --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FeatureDefinitionProvider.java @@ -0,0 +1,24 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.core.model.FeatureDefinitions; +import com.codeheadsystems.velocity.spi.model.Namespace; + +/** + * Supplies the current {@link FeatureDefinitions} snapshot for a namespace (FR-16/FR-17). + * + *

The engine reads a whole snapshot per request. Hot-reload is expressed as an atomic swap of + * the snapshot behind this interface, so a reader never observes a half-applied definition set + * (FR-17). A namespace with no configured definitions returns an empty snapshot, not {@code null} + * (FR-4). + */ +public interface FeatureDefinitionProvider { + + /** + * The current snapshot for a namespace. + * + * @param namespace the namespace + * @return the current snapshot; empty (never null) for an unconfigured namespace + */ + FeatureDefinitions definitions(Namespace namespace); +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FeatureDefinitionYaml.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FeatureDefinitionYaml.java new file mode 100644 index 0000000..191c5cc --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/FeatureDefinitionYaml.java @@ -0,0 +1,182 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.core.model.SubjectSource; +import com.codeheadsystems.velocity.core.model.yaml.FeatureDefinitionDoc; +import com.codeheadsystems.velocity.core.model.yaml.FeatureDefinitionDoc.AggregationDoc; +import com.codeheadsystems.velocity.core.model.yaml.FeatureDefinitionDoc.SubjectSourceDoc; +import com.codeheadsystems.velocity.core.model.yaml.FeatureDefinitionDoc.WindowDoc; +import com.codeheadsystems.velocity.core.model.yaml.FeatureDefinitionsDoc; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.AggregationType; +import com.codeheadsystems.velocity.spi.model.Namespace; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowType; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import org.jspecify.annotations.Nullable; +import tools.jackson.databind.DeserializationFeature; +import tools.jackson.dataformat.yaml.YAMLMapper; + +/** + * Imports and exports feature definitions as round-trippable YAML (FR-28). + * + *

The YAML document is the canonical external representation of a namespace's feature + * configuration, so it can be versioned in source control and moved between environments. The wire + * shape ({@link FeatureDefinitionsDoc}) is a plain Jackson DTO kept separate from the domain, so + * the domain stays serialization-neutral; this class maps between the two. Unknown fields are + * tolerated on read for forward compatibility. + * + *

Shape (durations are ISO-8601, e.g. {@code PT1H}): + * + * 

{@code
+ * namespace: acme
+ * definitions:
+ *   - name: card.count.1h
+ *     subjectType: card
+ *     subjectSource: { kind: PRIMARY }
+ *     backend: memory
+ *     aggregation: { type: COUNT }
+ *     windows:
+ *       - { duration: PT1H, type: TUMBLING }
+ *   - name: ip.count.1h
+ *     subjectType: ip
+ *     subjectSource: { kind: FROM_DIMENSION, dimension: ip }
+ *     backend: memory
+ *     aggregation: { type: COUNT }
+ *     windows:
+ *       - { duration: PT1M, type: TUMBLING }
+ * }
+ */ +public final class FeatureDefinitionYaml { + + private final YAMLMapper mapper; + + /** Creates a YAML codec with a forward-compatible mapper. */ + public FeatureDefinitionYaml() { + this.mapper = + YAMLMapper.builder() + .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) + .build(); + } + + /** + * Exports a namespace's definitions to YAML. + * + * @param namespace the namespace to stamp on the document + * @param definitions the definitions to export + * @return the YAML document + */ + public String export(final Namespace namespace, final List definitions) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(definitions, "definitions"); + final List docs = new ArrayList<>(definitions.size()); + for (final FeatureDefinition definition : definitions) { + docs.add(toDoc(definition)); + } + return mapper.writeValueAsString(new FeatureDefinitionsDoc(namespace.value(), docs)); + } + + /** + * Imports the definitions from a YAML document. + * + * @param yaml the YAML document + * @return the domain definitions + */ + public List importDefinitions(final String yaml) { + Objects.requireNonNull(yaml, "yaml"); + final FeatureDefinitionsDoc doc = mapper.readValue(yaml, FeatureDefinitionsDoc.class); + final List definitions = + doc.definitions() == null ? List.of() : doc.definitions(); + final List result = new ArrayList<>(definitions.size()); + for (final FeatureDefinitionDoc definition : definitions) { + result.add(fromDoc(definition)); + } + return result; + } + + /** + * The namespace declared in a YAML document, if any. + * + * @param yaml the YAML document + * @return the namespace, or empty if the document does not declare one + */ + public @Nullable Namespace importNamespace(final String yaml) { + Objects.requireNonNull(yaml, "yaml"); + final FeatureDefinitionsDoc doc = mapper.readValue(yaml, FeatureDefinitionsDoc.class); + final String namespace = doc.namespace(); + return namespace == null || namespace.isBlank() ? null : new Namespace(namespace); + } + + private static FeatureDefinitionDoc toDoc(final FeatureDefinition definition) { + final List windows = new ArrayList<>(definition.windows().size()); + for (final Window window : definition.windows()) { + windows.add(new WindowDoc(window.duration().toString(), window.type().name())); + } + final Aggregation aggregation = definition.aggregation(); + return new FeatureDefinitionDoc( + definition.name(), + definition.subjectType(), + toSubjectSourceDoc(definition.subjectSource()), + new AggregationDoc(aggregation.type().name(), aggregation.dimension()), + windows, + definition.backend(), + definition.distinctThreshold()); + } + + private static SubjectSourceDoc toSubjectSourceDoc(final SubjectSource source) { + return switch (source) { + case SubjectSource.Primary ignored -> new SubjectSourceDoc("PRIMARY", null); + case SubjectSource.FromDimension fromDimension -> + new SubjectSourceDoc("FROM_DIMENSION", fromDimension.dimensionName()); + }; + } + + private static FeatureDefinition fromDoc(final FeatureDefinitionDoc doc) { + final AggregationDoc aggregationDoc = + Objects.requireNonNull(doc.aggregation(), "aggregation is required"); + final List windowDocs = + Objects.requireNonNull(doc.windows(), "windows are required"); + final List windows = new ArrayList<>(windowDocs.size()); + for (final WindowDoc windowDoc : windowDocs) { + windows.add( + new Window( + Duration.parse(Objects.requireNonNull(windowDoc.duration(), "window duration")), + WindowType.valueOf(Objects.requireNonNull(windowDoc.type(), "window type")))); + } + return new FeatureDefinition( + Objects.requireNonNull(doc.name(), "name is required"), + Objects.requireNonNull(doc.subjectType(), "subjectType is required"), + fromSubjectSourceDoc(doc.subjectSource()), + fromAggregationDoc(aggregationDoc), + windows, + Objects.requireNonNull(doc.backend(), "backend is required"), + doc.distinctThreshold()); + } + + private static SubjectSource fromSubjectSourceDoc(final @Nullable SubjectSourceDoc doc) { + if (doc == null || doc.kind() == null || "PRIMARY".equals(doc.kind())) { + return SubjectSource.primary(); + } + if ("FROM_DIMENSION".equals(doc.kind())) { + return SubjectSource.fromDimension( + Objects.requireNonNull(doc.dimension(), "FROM_DIMENSION requires a dimension")); + } + throw new IllegalArgumentException("unknown subjectSource kind '" + doc.kind() + "'"); + } + + private static Aggregation fromAggregationDoc(final AggregationDoc doc) { + final AggregationType type = + AggregationType.valueOf(Objects.requireNonNull(doc.type(), "aggregation type is required")); + return switch (type) { + case COUNT -> Aggregation.count(); + case SUM -> Aggregation.sum(); + case DISTINCT -> + Aggregation.distinct( + Objects.requireNonNull(doc.dimension(), "DISTINCT requires a dimension")); + }; + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/InMemoryNamespaceSaltProvider.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/InMemoryNamespaceSaltProvider.java new file mode 100644 index 0000000..ec0c9f5 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/InMemoryNamespaceSaltProvider.java @@ -0,0 +1,70 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.spi.model.Namespace; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.security.SecureRandom; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * A development/test {@link NamespaceSaltProvider} that derives a per-namespace salt from a single + * in-process master secret. + * + *

Not for production (§15 R11). The master secret is co-resident with the + * process (and, in a test, with the distinct sets), which is exactly the custody posture R11 + * forbids. Production MUST use a KMS/secret-store-backed provider so the key lives in a separate + * trust domain. The salt is {@code SHA-256(master || namespace)}, so it is deterministic per + * namespace for a given master (reproducible hashing within a run) and distinct across namespaces. + */ +public final class InMemoryNamespaceSaltProvider implements NamespaceSaltProvider { + + private final byte[] master; + private final ConcurrentMap cache = new ConcurrentHashMap<>(); + + /** Creates a provider with a random 32-byte master secret (distinct per instance). */ + public InMemoryNamespaceSaltProvider() { + this(randomMaster()); + } + + /** + * Creates a provider with an explicit master secret — use a fixed secret for reproducible hashing + * across runs in tests. + * + * @param master the master secret bytes; copied defensively, must be non-empty + */ + public InMemoryNamespaceSaltProvider(final byte[] master) { + Objects.requireNonNull(master, "master"); + if (master.length == 0) { + throw new IllegalArgumentException("master secret must not be empty"); + } + this.master = master.clone(); + } + + @Override + public byte[] salt(final Namespace namespace) { + Objects.requireNonNull(namespace, "namespace"); + return cache.computeIfAbsent(namespace, this::derive).clone(); + } + + private byte[] derive(final Namespace namespace) { + final MessageDigest digest; + try { + digest = MessageDigest.getInstance("SHA-256"); + } catch (final NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 not available", e); + } + digest.update(master); + digest.update(namespace.value().getBytes(StandardCharsets.UTF_8)); + return digest.digest(); + } + + private static byte[] randomMaster() { + final byte[] bytes = new byte[32]; + new SecureRandom().nextBytes(bytes); + return bytes; + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/MutableFeatureDefinitionProvider.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/MutableFeatureDefinitionProvider.java new file mode 100644 index 0000000..50597a7 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/MutableFeatureDefinitionProvider.java @@ -0,0 +1,56 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.core.model.FeatureDefinitions; +import com.codeheadsystems.velocity.spi.model.Namespace; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * The default in-memory {@link FeatureDefinitionProvider}: a concurrent map of namespace → snapshot + * whose per-namespace snapshot is swapped atomically on {@link #reload} (FR-17). + * + *

Because each namespace's value is a single immutable {@link FeatureDefinitions} reference, a + * {@link #definitions} reader either sees the old snapshot in full or the new one in full — never a + * half-applied set — even while a concurrent {@code reload} runs. Loading definitions from an + * external store or file watcher (a non-primary hot-reload source, OQ-C) is deferred; such a source + * would drive {@link #reload}. + */ +public final class MutableFeatureDefinitionProvider implements FeatureDefinitionProvider { + + private final ConcurrentMap snapshots = new ConcurrentHashMap<>(); + + /** Creates an empty provider with no definitions configured for any namespace. */ + public MutableFeatureDefinitionProvider() {} + + @Override + public FeatureDefinitions definitions(final Namespace namespace) { + Objects.requireNonNull(namespace, "namespace"); + return snapshots.computeIfAbsent(namespace, MutableFeatureDefinitionProvider::emptySnapshot); + } + + /** + * Atomically replaces a namespace's definition set with a fresh snapshot (FR-17), recomputing the + * version hash (FR-40). + * + * @param namespace the namespace to reload + * @param definitions the new definition set (validated by the engine before this is called) + * @return the newly installed snapshot + */ + public FeatureDefinitions reload( + final Namespace namespace, final List definitions) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(definitions, "definitions"); + final FeatureDefinitions snapshot = + new FeatureDefinitions(namespace, definitions, DefinitionVersionHasher.hash(definitions)); + snapshots.put(namespace, snapshot); + return snapshot; + } + + private static FeatureDefinitions emptySnapshot(final Namespace namespace) { + return new FeatureDefinitions(namespace, List.of(), DefinitionVersionHasher.hash(List.of())); + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/NamespaceSaltProvider.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/NamespaceSaltProvider.java new file mode 100644 index 0000000..6d6dcfe --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/NamespaceSaltProvider.java @@ -0,0 +1,26 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.spi.model.Namespace; + +/** + * Supplies the per-namespace HMAC salt used to keyed-hash DISTINCT dimension values at rest (FR-38, + * §15 R11). + * + *

Key custody (§15 R11). In production the salt/key MUST live in a separate + * trust domain (a KMS or secret store), not co-resident with the distinct sets: a + * low-entropy dimension like an IPv4 (2³² space) is brute-forceable offline after a single + * datastore dump if the salt is dumped alongside it, defeating the "no raw PII at rest" guarantee. + * A KMS-backed implementation is a deferred follow-up; {@link InMemoryNamespaceSaltProvider} is a + * development/test default only. + */ +public interface NamespaceSaltProvider { + + /** + * The salt bytes for a namespace. + * + * @param namespace the namespace + * @return the salt; must be stable per namespace and never empty + */ + byte[] salt(Namespace namespace); +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/VelocityEngine.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/VelocityEngine.java new file mode 100644 index 0000000..79d3420 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/VelocityEngine.java @@ -0,0 +1,394 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import com.codeheadsystems.velocity.core.model.CapabilityValidationResult; +import com.codeheadsystems.velocity.core.model.CapabilityValidationResult.Violation; +import com.codeheadsystems.velocity.core.model.FanOutResult; +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.core.model.FeatureDefinitions; +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.DistinctStore; +import com.codeheadsystems.velocity.spi.SumStore; +import com.codeheadsystems.velocity.spi.VelocityBackend; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.ApplyContext; +import com.codeheadsystems.velocity.spi.model.ApplyResult; +import com.codeheadsystems.velocity.spi.model.ApplyStatus; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities; +import com.codeheadsystems.velocity.spi.model.FailureCode; +import com.codeheadsystems.velocity.spi.model.FeatureResult; +import com.codeheadsystems.velocity.spi.model.FeatureValue; +import com.codeheadsystems.velocity.spi.model.Intent; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.spi.model.Intent.DistinctIntent; +import com.codeheadsystems.velocity.spi.model.Intent.SumIntent; +import com.codeheadsystems.velocity.spi.model.Namespace; +import com.codeheadsystems.velocity.spi.model.PerFeature; +import com.codeheadsystems.velocity.spi.model.QueryContext; +import com.codeheadsystems.velocity.spi.model.QueryTuple; +import com.codeheadsystems.velocity.spi.model.Subject; +import com.codeheadsystems.velocity.spi.model.Window; +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import org.jspecify.annotations.Nullable; + +/** + * The embeddable engine and public API: turns a {@code record()} into backend-neutral intents, fans + * it out across every matching feature and backend, and merges the outcomes into one result stamped + * with the definition version it was computed under (FR-40). + * + *

The engine owns fan-out and routing only; windowing lives entirely in the backend (ADR 0003). + * It is constructed with a {@link BackendRegistry}, a {@link FeatureDefinitionProvider}, a {@link + * DimensionHasher}, and a {@link CapabilityValidator} (no Dagger — DI lives in the service tier). + * + *

Read-your-write rides on the backend: because the in-process backends return an apply's + * post-write value, {@link #record}'s result already reflects the write (ADR 0007). Every returned + * value is re-stamped with the current snapshot's {@code versionHash} (FR-40). + */ +public final class VelocityEngine { + + private final BackendRegistry backends; + private final FeatureDefinitionProvider definitionProvider; + private final CapabilityValidator capabilityValidator; + private final FanOutResolver fanOutResolver; + + /** + * Creates an engine. + * + * @param backends the registry of backends features bind to + * @param definitionProvider the source of per-namespace definition snapshots (atomic hot-reload) + * @param dimensionHasher the keyed hasher for DISTINCT dimension values (FR-38) + * @param capabilityValidator the validator run on reload (FR-29) + */ + public VelocityEngine( + final BackendRegistry backends, + final FeatureDefinitionProvider definitionProvider, + final DimensionHasher dimensionHasher, + final CapabilityValidator capabilityValidator) { + this.backends = Objects.requireNonNull(backends, "backends"); + this.definitionProvider = Objects.requireNonNull(definitionProvider, "definitionProvider"); + this.capabilityValidator = Objects.requireNonNull(capabilityValidator, "capabilityValidator"); + this.fanOutResolver = + new FanOutResolver(Objects.requireNonNull(dimensionHasher, "dimensionHasher")); + } + + /** + * Records an event, fanning it out to every matching feature across every affected subject and + * backend, and returns the aggregate post-write outcome (FR-18, ADR 0009). + * + *

An event that matches no definition is a no-op with an empty result, never an error (FR-4). + * Each per-feature entry is stamped with the current snapshot's definition version (FR-40). + * + * @param namespace the tenancy scope + * @param subject the event's primary subject + * @param dimensions the event's dimensions (empty if none) + * @param valueCents the event's SUM value in integer cents, or null if the event carries no value + * @return one {@link PerFeature} outcome per touched {@code (feature × window)} plus any skipped + * features + */ + public ApplyResult record( + final Namespace namespace, + final Subject subject, + final Map dimensions, + final @Nullable BigDecimal valueCents) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(subject, "subject"); + Objects.requireNonNull(dimensions, "dimensions"); + final FeatureDefinitions definitions = definitionProvider.definitions(namespace); + final FanOutResult fanOut = + fanOutResolver.resolve(namespace, subject, dimensions, valueCents, definitions); + final ApplyContext ctx = new ApplyContext(namespace, null); + final List outcomes = new ArrayList<>(); + for (final Map.Entry> group : fanOut.intentsByBackend().entrySet()) { + final VelocityBackend backend = backends.backend(group.getKey()); + outcomes.addAll(applyGroup(ctx, backend, group.getValue())); + } + outcomes.addAll(fanOut.skipped()); + return new ApplyResult(stampAll(outcomes, definitions.versionHash())); + } + + private List applyGroup( + final ApplyContext ctx, final VelocityBackend backend, final List intents) { + final List counts = new ArrayList<>(); + final List sums = new ArrayList<>(); + final List distincts = new ArrayList<>(); + for (final Intent intent : intents) { + switch (intent) { + case CountIntent count -> counts.add(count); + case SumIntent sum -> sums.add(sum); + case DistinctIntent distinct -> distincts.add(distinct); + } + } + final List outcomes = new ArrayList<>(); + if (!counts.isEmpty()) { + if (backend instanceof CountStore store) { + outcomes.addAll(store.applyCount(ctx, counts).perFeature()); + } else { + outcomes.addAll(unsupported(counts, "COUNT")); + } + } + if (!sums.isEmpty()) { + if (backend instanceof SumStore store) { + outcomes.addAll(store.applySum(ctx, sums).perFeature()); + } else { + outcomes.addAll(unsupported(sums, "SUM")); + } + } + if (!distincts.isEmpty()) { + if (backend instanceof DistinctStore store) { + outcomes.addAll(store.applyDistinct(ctx, distincts).perFeature()); + } else { + outcomes.addAll(unsupported(distincts, "DISTINCT")); + } + } + return outcomes; + } + + private static List unsupported( + final List intents, final String aggregation) { + final List failed = new ArrayList<>(intents.size()); + for (final Intent intent : intents) { + failed.add( + new PerFeature( + intent.feature(), + ApplyStatus.FAILED, + FeatureResult.failure( + FailureCode.INTERNAL, + "backend does not implement the " + + aggregation + + " mix-in for feature '" + + intent.feature().name() + + "'"))); + } + return failed; + } + + /** + * Queries a single feature's value over each of its windows for a subject. + * + * @param namespace the tenancy scope + * @param subject the subject to read + * @param featureName the feature to read + * @return one result per the feature's windows, in declaration order (each stamped with the + * current definition version) + * @throws IllegalArgumentException if the namespace has no such feature + */ + public List query( + final Namespace namespace, final Subject subject, final String featureName) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(subject, "subject"); + Objects.requireNonNull(featureName, "featureName"); + final FeatureDefinitions definitions = definitionProvider.definitions(namespace); + final FeatureDefinition definition = + definitions + .definition(featureName) + .orElseThrow( + () -> + new IllegalArgumentException( + "no feature '" + + featureName + + "' in namespace " + + namespace.value() + + " (version " + + definitions.versionHash() + + ")")); + return queryDefinition(namespace, subject, definition, definitions.versionHash()); + } + + /** + * Queries several features for a subject in one call. + * + * @param namespace the tenancy scope + * @param subject the subject to read + * @param featureNames the features to read + * @return a map of feature name → per-window results, in the request's iteration order + * @throws IllegalArgumentException if the namespace has no such feature + */ + public Map> query( + final Namespace namespace, final Subject subject, final List featureNames) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(subject, "subject"); + Objects.requireNonNull(featureNames, "featureNames"); + final Map> results = new LinkedHashMap<>(); + for (final String featureName : featureNames) { + results.put(featureName, query(namespace, subject, featureName)); + } + return results; + } + + private List queryDefinition( + final Namespace namespace, + final Subject subject, + final FeatureDefinition definition, + final String versionHash) { + final VelocityBackend backend = backends.backend(definition.backend()); + final Aggregation aggregation = definition.aggregation(); + final List tuples = new ArrayList<>(definition.windows().size()); + for (final Window window : definition.windows()) { + tuples.add(new QueryTuple(subject, aggregation, window)); + } + final QueryContext ctx = new QueryContext(namespace, null); + final List raw = + switch (aggregation.type()) { + case COUNT -> + backend instanceof CountStore store + ? store.queryCount(ctx, tuples) + : queryUnsupported(tuples, "COUNT"); + case SUM -> + backend instanceof SumStore store + ? store.querySum(ctx, tuples) + : queryUnsupported(tuples, "SUM"); + case DISTINCT -> + backend instanceof DistinctStore store + ? store.queryDistinct(ctx, tuples) + : queryUnsupported(tuples, "DISTINCT"); + }; + return stampResults(raw, versionHash); + } + + private static List queryUnsupported( + final List tuples, final String aggregation) { + final List failed = new ArrayList<>(tuples.size()); + for (int i = 0; i < tuples.size(); i++) { + failed.add( + FeatureResult.failure( + FailureCode.INTERNAL, "backend does not implement the " + aggregation + " mix-in")); + } + return failed; + } + + /** + * The declared capabilities of a registered backend (passthrough). + * + * @param backendName the backend name + * @return the backend's capabilities + */ + public BackendCapabilities capabilities(final String backendName) { + Objects.requireNonNull(backendName, "backendName"); + return backends.backend(backendName).capabilities(); + } + + /** + * Administratively purges stored state on every registered backend (FR-23). + * + * @param namespace the namespace to erase within + * @param subject the subject to erase; null erases the whole namespace + */ + public void purge(final Namespace namespace, final @Nullable Subject subject) { + Objects.requireNonNull(namespace, "namespace"); + for (final String name : backends.names()) { + backends.backend(name).purge(namespace, subject); + } + } + + /** + * Validates a candidate definition set against its target backends' capabilities (FR-29) without + * installing it. + * + * @param definitions the candidate definitions + * @return the collected violations (empty when the set is valid) + */ + public CapabilityValidationResult validateReload(final List definitions) { + Objects.requireNonNull(definitions, "definitions"); + final List violations = new ArrayList<>(); + for (final FeatureDefinition definition : definitions) { + final VelocityBackend backend = + backends.names().contains(definition.backend()) + ? backends.backend(definition.backend()) + : null; + if (backend == null) { + violations.add( + new Violation( + definition.name(), + "unknown backend '" + + definition.backend() + + "'; known backends: " + + backends.names())); + continue; + } + violations.addAll( + capabilityValidator.validate(definition, backend.capabilities()).violations()); + } + return new CapabilityValidationResult(violations); + } + + /** + * Validates and atomically installs a new definition set for a namespace (FR-17/FR-29). + * + *

Every definition is validated against its target backend's capabilities first; if any is + * invalid the whole reload is rejected and nothing is installed (atomic). On success the + * namespace's snapshot is swapped atomically and its version hash recomputed (FR-40). + * + * @param namespace the namespace to reload + * @param definitions the new definition set + * @return the newly installed snapshot + * @throws IllegalArgumentException if any definition is invalid (the message lists every + * violation) + * @throws UnsupportedOperationException if the configured provider does not support reload + */ + public FeatureDefinitions reload( + final Namespace namespace, final List definitions) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(definitions, "definitions"); + if (!(definitionProvider instanceof MutableFeatureDefinitionProvider mutable)) { + throw new UnsupportedOperationException( + "reload requires a MutableFeatureDefinitionProvider; configured provider is " + + definitionProvider.getClass().getName()); + } + final CapabilityValidationResult validation = validateReload(definitions); + if (!validation.valid()) { + final StringBuilder message = new StringBuilder("reload rejected for namespace "); + message.append(namespace.value()).append(':'); + for (final Violation violation : validation.violations()) { + message + .append("\n - ") + .append(violation.feature()) + .append(": ") + .append(violation.message()); + } + throw new IllegalArgumentException(message.toString()); + } + return mutable.reload(namespace, definitions); + } + + private static List stampAll( + final List outcomes, final String versionHash) { + final List stamped = new ArrayList<>(outcomes.size()); + for (final PerFeature outcome : outcomes) { + stamped.add( + new PerFeature( + outcome.feature(), outcome.status(), stamp(outcome.result(), versionHash))); + } + return stamped; + } + + private static List stampResults( + final List results, final String versionHash) { + final List stamped = new ArrayList<>(results.size()); + for (final FeatureResult result : results) { + stamped.add(stamp(result, versionHash)); + } + return stamped; + } + + private static FeatureResult stamp(final FeatureResult result, final String versionHash) { + if (result instanceof FeatureResult.Success success) { + final FeatureValue value = success.value(); + return FeatureResult.success( + new FeatureValue( + value.feature(), + value.window(), + value.value(), + value.exactness(), + value.readYourWriteLevel(), + versionHash, + value.windowBounds(), + value.asOf())); + } + return result; + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/CapabilityValidationResult.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/CapabilityValidationResult.java new file mode 100644 index 0000000..f646466 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/CapabilityValidationResult.java @@ -0,0 +1,47 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core.model; + +import java.util.List; +import java.util.Objects; + +/** + * The outcome of validating feature definitions against their target backends' capabilities + * (FR-29). + * + *

Validation collects all violations rather than failing on the first, so a reload is + * rejected with a complete, precise error list. A result with no violations is {@link #valid()}. + * + * @param violations every violation found; empty when all definitions are valid + */ +public record CapabilityValidationResult(List violations) { + + /** Stores an unmodifiable copy of the violations. */ + public CapabilityValidationResult { + Objects.requireNonNull(violations, "violations"); + violations = List.copyOf(violations); + } + + /** + * Whether every validated definition passed. + * + * @return {@code true} if there are no violations + */ + public boolean valid() { + return violations.isEmpty(); + } + + /** + * A single capability violation of one feature definition (FR-29). + * + * @param feature the offending feature's name + * @param message the precise, human-readable reason + */ + public record Violation(String feature, String message) { + + /** Validates both components are present. */ + public Violation { + Objects.requireNonNull(feature, "feature"); + Objects.requireNonNull(message, "message"); + } + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FanOutResult.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FanOutResult.java new file mode 100644 index 0000000..7ba022b --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FanOutResult.java @@ -0,0 +1,44 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core.model; + +import com.codeheadsystems.velocity.spi.model.Intent; +import com.codeheadsystems.velocity.spi.model.PerFeature; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** + * The resolved fan-out of one {@code record()} event (FR-18): the backend-neutral write intents + * grouped by their target backend, plus the outcomes for features that matched the event but could + * not be applied. + * + *

A single event fans out across multiple subjects and backends (FR-18); {@code + * intentsByBackend} keys the intents by backend name so the engine can dispatch each group to its + * backend and sub-group by aggregation. {@code skipped} carries the pre-computed {@link PerFeature} + * outcomes for definitions that applied to this event's subject but were skipped — a SUM whose + * value was missing, or a DISTINCT whose dimension was absent — so they surface in the aggregate + * result rather than being silently dropped. + * + * @param intentsByBackend the write intents grouped by target backend name; defensively copied + * @param skipped the pre-computed outcomes for matched-but-not-applied features; defensively copied + */ +public record FanOutResult(Map> intentsByBackend, List skipped) { + + /** Stores unmodifiable copies of both collections. */ + public FanOutResult { + Objects.requireNonNull(intentsByBackend, "intentsByBackend"); + Objects.requireNonNull(skipped, "skipped"); + intentsByBackend = Map.copyOf(intentsByBackend); + skipped = List.copyOf(skipped); + } + + /** + * Whether this fan-out produced no intents and no skipped outcomes (FR-4: no matching + * definition). + * + * @return {@code true} if nothing matched the event + */ + public boolean isEmpty() { + return intentsByBackend.isEmpty() && skipped.isEmpty(); + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FeatureDefinition.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FeatureDefinition.java new file mode 100644 index 0000000..5fe0ae4 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FeatureDefinition.java @@ -0,0 +1,143 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core.model; + +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.AggregationType; +import com.codeheadsystems.velocity.spi.model.Window; +import java.util.List; +import java.util.Objects; +import org.jspecify.annotations.Nullable; + +/** + * The configuration for a single feature (FR-16): what event pattern feeds it, its subject, its + * aggregation, the windows it is tracked over, and the backend that provides it. + * + *

A feature is bound to exactly one backend (FR-16/FR-29). Definitions are data, not code — they + * are imported/exported as YAML (FR-28) and hot-reloaded at runtime (FR-17). + * + * @param name the stable, non-blank feature name (e.g. {@code card.count.1h}) + * @param subjectType the non-blank subject type this feature keys on (e.g. {@code card}, {@code + * ip}) + * @param subjectSource where the fan-out subject comes from — the event's primary subject or a + * dimension-derived one (FR-18) + * @param aggregation what the feature aggregates (COUNT/SUM/DISTINCT) + * @param windows the non-empty set of windows the feature is tracked over + * @param backend the non-blank name of the backend that provides this feature (FR-16) + * @param distinctThreshold the per-{@code (subject, bucket)} exact→HLL threshold for DISTINCT + * features (ADR 0006); {@code null} means "use the backend default", and it MUST be null for + * non-DISTINCT features + */ +public record FeatureDefinition( + String name, + String subjectType, + SubjectSource subjectSource, + Aggregation aggregation, + List windows, + String backend, + @Nullable Long distinctThreshold) { + + /** Validates the definition's invariants and stores an unmodifiable copy of the windows. */ + public FeatureDefinition { + Objects.requireNonNull(name, "name"); + Objects.requireNonNull(subjectType, "subjectType"); + Objects.requireNonNull(subjectSource, "subjectSource"); + Objects.requireNonNull(aggregation, "aggregation"); + Objects.requireNonNull(windows, "windows"); + Objects.requireNonNull(backend, "backend"); + if (name.isBlank()) { + throw new IllegalArgumentException("feature name must not be blank"); + } + if (subjectType.isBlank()) { + throw new IllegalArgumentException("subjectType must not be blank"); + } + if (backend.isBlank()) { + throw new IllegalArgumentException("backend must not be blank"); + } + if (windows.isEmpty()) { + throw new IllegalArgumentException("feature '" + name + "' must declare at least one window"); + } + if (aggregation.type() != AggregationType.DISTINCT && distinctThreshold != null) { + throw new IllegalArgumentException( + "distinctThreshold is only valid for DISTINCT features, was set on '" + name + "'"); + } + windows = List.copyOf(windows); + } + + /** + * A COUNT feature over the event's own subject. + * + * @param name the feature name + * @param subjectType the subject type + * @param backend the backend name + * @param windows the windows + * @return a primary-subject COUNT definition + */ + public static FeatureDefinition count( + final String name, + final String subjectType, + final String backend, + final List windows) { + return new FeatureDefinition( + name, subjectType, SubjectSource.primary(), Aggregation.count(), windows, backend, null); + } + + /** + * A SUM feature over the event's own subject. + * + * @param name the feature name + * @param subjectType the subject type + * @param backend the backend name + * @param windows the windows + * @return a primary-subject SUM definition + */ + public static FeatureDefinition sum( + final String name, + final String subjectType, + final String backend, + final List windows) { + return new FeatureDefinition( + name, subjectType, SubjectSource.primary(), Aggregation.sum(), windows, backend, null); + } + + /** + * A DISTINCT feature over the given dimension, keyed on the event's own subject. + * + * @param name the feature name + * @param subjectType the subject type + * @param dimension the dimension whose distinct values are counted + * @param backend the backend name + * @param windows the windows + * @return a primary-subject DISTINCT definition with the backend-default threshold + */ + public static FeatureDefinition distinct( + final String name, + final String subjectType, + final String dimension, + final String backend, + final List windows) { + return new FeatureDefinition( + name, + subjectType, + SubjectSource.primary(), + Aggregation.distinct(dimension), + windows, + backend, + null); + } + + /** + * The largest window duration this feature is tracked over — the retention floor its backend must + * satisfy (FR-22a). + * + * @return the longest window duration + */ + public java.time.Duration largestWindow() { + java.time.Duration largest = windows.get(0).duration(); + for (final Window window : windows) { + if (window.duration().compareTo(largest) > 0) { + largest = window.duration(); + } + } + return largest; + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FeatureDefinitions.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FeatureDefinitions.java new file mode 100644 index 0000000..d05c2b2 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/FeatureDefinitions.java @@ -0,0 +1,67 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core.model; + +import com.codeheadsystems.velocity.spi.model.Namespace; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; + +/** + * An immutable, namespace-scoped snapshot of a namespace's feature definitions (FR-17), stamped + * with a deterministic {@code versionHash} (FR-40). + * + *

Hot-reload swaps a whole {@code FeatureDefinitions} atomically (FR-17): a reader either sees + * the old snapshot in full or the new one in full, never a half-applied set. The {@code + * versionHash} is the value stamped onto every {@link + * com.codeheadsystems.velocity.spi.model.FeatureValue} the snapshot produces, so a caller can + * detect that a definition changed under a running rule (FR-40). + * + * @param namespace the namespace this snapshot is for + * @param definitions the definitions in this snapshot; defensively copied, feature names must be + * unique + * @param versionHash the deterministic version stamp computed from {@code definitions} (FR-40) + */ +public record FeatureDefinitions( + Namespace namespace, List definitions, String versionHash) { + + /** Validates uniqueness of feature names and stores an unmodifiable copy of the definitions. */ + public FeatureDefinitions { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(definitions, "definitions"); + Objects.requireNonNull(versionHash, "versionHash"); + definitions = List.copyOf(definitions); + final Map byName = new LinkedHashMap<>(); + for (final FeatureDefinition definition : definitions) { + if (byName.putIfAbsent(definition.name(), definition) != null) { + throw new IllegalArgumentException( + "duplicate feature name '" + definition.name() + "' in namespace " + namespace.value()); + } + } + } + + /** + * Looks up a definition by feature name. + * + * @param featureName the feature name + * @return the matching definition, or empty if this snapshot has no such feature + */ + public Optional definition(final String featureName) { + for (final FeatureDefinition definition : definitions) { + if (definition.name().equals(featureName)) { + return Optional.of(definition); + } + } + return Optional.empty(); + } + + /** + * Whether this snapshot has no definitions (e.g. an unconfigured namespace). + * + * @return {@code true} if there are no definitions + */ + public boolean isEmpty() { + return definitions.isEmpty(); + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/SubjectSource.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/SubjectSource.java new file mode 100644 index 0000000..fa977a8 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/SubjectSource.java @@ -0,0 +1,58 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core.model; + +import java.util.Objects; + +/** + * Where a feature's fan-out subject comes from for a given event (D5, FR-18). + * + *

A single {@code record()} can update a feature keyed on the event's own subject and + * features keyed on a dimension-derived subject (e.g. also counting per-{@code ip}). This sealed + * type names those two origins so the {@code FanOutResolver} can decide, per definition, which + * {@link com.codeheadsystems.velocity.spi.model.Subject subject} an event contributes to. + */ +public sealed interface SubjectSource permits SubjectSource.Primary, SubjectSource.FromDimension { + + /** + * The event's own primary subject (the subject passed to {@code record()}). The definition + * applies only when that subject's type matches the definition's {@code subjectType}. + */ + record Primary() implements SubjectSource {} + + /** + * A subject derived from one of the event's dimensions (FR-18). The definition applies only when + * the named dimension is present on the event; the derived subject is {@code (subjectType, + * dimensionValue)}. + * + * @param dimensionName the non-blank dimension whose value becomes the subject value + */ + record FromDimension(String dimensionName) implements SubjectSource { + + /** Validates the dimension name is present and non-blank. */ + public FromDimension { + Objects.requireNonNull(dimensionName, "dimensionName"); + if (dimensionName.isBlank()) { + throw new IllegalArgumentException("dimensionName must not be blank"); + } + } + } + + /** + * The event's own primary subject. + * + * @return a {@link Primary} source + */ + static SubjectSource primary() { + return new Primary(); + } + + /** + * A subject derived from the named dimension. + * + * @param dimensionName the dimension whose value becomes the subject value + * @return a {@link FromDimension} source + */ + static SubjectSource fromDimension(String dimensionName) { + return new FromDimension(dimensionName); + } +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/package-info.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/package-info.java new file mode 100644 index 0000000..ca848b9 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/package-info.java @@ -0,0 +1,16 @@ +// SPDX-License-Identifier: BSD-3-Clause + +/** + * The value types of the {@code velocity-core} engine: feature definitions, the namespace-scoped + * definition snapshot, subject-source model, and validation/fan-out result records. + * + *

These are plain, immutable records/sealed types that validate their invariants in compact + * constructors — the engine's configuration and result vocabulary, kept separate from the engine + * logic (which lives in the parent {@code com.codeheadsystems.velocity.core} package and + * is tested to the coverage gate). This package is coverage-exempt (the {@code **}/{@code model/**} + * carve-out); no engine logic is parked here. + */ +@NullMarked +package com.codeheadsystems.velocity.core.model; + +import org.jspecify.annotations.NullMarked; diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/FeatureDefinitionDoc.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/FeatureDefinitionDoc.java new file mode 100644 index 0000000..1a8cc86 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/FeatureDefinitionDoc.java @@ -0,0 +1,58 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core.model.yaml; + +import com.fasterxml.jackson.annotation.JsonInclude; +import java.util.List; +import org.jspecify.annotations.Nullable; + +/** + * The YAML wire shape of one feature definition (FR-28). + * + * @param name the feature name + * @param subjectType the subject type + * @param subjectSource where the fan-out subject comes from (defaults to primary when absent) + * @param aggregation the aggregation (type + optional distinct dimension) + * @param windows the windows the feature is tracked over + * @param backend the backend that provides this feature + * @param distinctThreshold the DISTINCT exact→HLL threshold; null uses the backend default + */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public record FeatureDefinitionDoc( + @Nullable String name, + @Nullable String subjectType, + @Nullable SubjectSourceDoc subjectSource, + @Nullable AggregationDoc aggregation, + @Nullable List windows, + @Nullable String backend, + @Nullable Long distinctThreshold) { + + /** + * The YAML shape of a subject source: {@code kind: PRIMARY} or {@code kind: FROM_DIMENSION} with + * a {@code dimension}. + * + * @param kind {@code PRIMARY} or {@code FROM_DIMENSION} + * @param dimension the dimension name (only for {@code FROM_DIMENSION}) + */ + @JsonInclude(JsonInclude.Include.NON_NULL) + public record SubjectSourceDoc(@Nullable String kind, @Nullable String dimension) {} + + /** + * The YAML shape of an aggregation: a {@code type} and, for {@code DISTINCT}, a {@code + * dimension}. + * + * @param type {@code COUNT}, {@code SUM}, or {@code DISTINCT} + * @param dimension the distinct dimension (only for {@code DISTINCT}) + */ + @JsonInclude(JsonInclude.Include.NON_NULL) + public record AggregationDoc(@Nullable String type, @Nullable String dimension) {} + + /** + * The YAML shape of a window: an ISO-8601 {@code duration} (e.g. {@code PT1H}) and a {@code + * type}. + * + * @param duration the ISO-8601 duration string + * @param type {@code SLIDING} or {@code TUMBLING} + */ + @JsonInclude(JsonInclude.Include.NON_NULL) + public record WindowDoc(@Nullable String duration, @Nullable String type) {} +} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/FeatureDefinitionsDoc.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/FeatureDefinitionsDoc.java new file mode 100644 index 0000000..4f98561 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/FeatureDefinitionsDoc.java @@ -0,0 +1,23 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core.model.yaml; + +import com.fasterxml.jackson.annotation.JsonInclude; +import java.util.List; +import org.jspecify.annotations.Nullable; + +/** + * The YAML wire shape of a namespace's feature definition set (FR-28). + * + *

This is the canonical external representation of feature configuration — deliberately a plain, + * Jackson-bound DTO separate from the domain {@code FeatureDefinition} so the domain types stay + * serialization-neutral (mirroring the {@code velocity-spi} model philosophy). {@code + * FeatureDefinitionYaml} maps between this shape and the domain. Unknown fields are tolerated on + * read for forward compatibility. + * + * @param namespace the namespace these definitions belong to; may be null in a namespace-agnostic + * document (the caller supplies the namespace on import) + * @param definitions the feature definitions + */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public record FeatureDefinitionsDoc( + @Nullable String namespace, @Nullable List definitions) {} diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/package-info.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/package-info.java new file mode 100644 index 0000000..869ca64 --- /dev/null +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/model/yaml/package-info.java @@ -0,0 +1,13 @@ +// SPDX-License-Identifier: BSD-3-Clause + +/** + * The Jackson-bound YAML wire shapes for feature-definition import/export (FR-28). + * + *

Kept deliberately separate from the domain {@code FeatureDefinition} so the domain stays + * serialization-neutral: {@code FeatureDefinitionYaml} maps between these DTOs and the domain. + * These are coverage-exempt DTOs under {@code **}/{@code model/**}. + */ +@NullMarked +package com.codeheadsystems.velocity.core.model.yaml; + +import org.jspecify.annotations.NullMarked; diff --git a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/package-info.java b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/package-info.java index 47546d0..6094bff 100644 --- a/velocity-core/src/main/java/com/codeheadsystems/velocity/core/package-info.java +++ b/velocity-core/src/main/java/com/codeheadsystems/velocity/core/package-info.java @@ -1,4 +1,29 @@ // SPDX-License-Identifier: BSD-3-Clause -/** The velocity-core engine: record()/query(), fan-out resolution, and feature-definition I/O. */ +/** + * The {@code velocity-core} engine: the embeddable heart that turns a {@code record()} into + * backend-neutral intents and back into per-feature results. + * + *

This package holds the engine logic — {@link + * com.codeheadsystems.velocity.core.VelocityEngine} (the public API), the {@link + * com.codeheadsystems.velocity.core.FanOutResolver fan-out resolver} (FR-18), the {@link + * com.codeheadsystems.velocity.core.CapabilityValidator capability validator} (FR-29), the {@link + * com.codeheadsystems.velocity.core.DimensionHasher keyed dimension hasher} (FR-38), the {@link + * com.codeheadsystems.velocity.core.FeatureDefinitionProvider definition provider} with atomic + * hot-reload (FR-17), the {@link com.codeheadsystems.velocity.core.BackendRegistry backend + * registry}, and {@link com.codeheadsystems.velocity.core.FeatureDefinitionYaml YAML I/O} (FR-28). + * The value types live in {@code com.codeheadsystems.velocity.core.model}. + * + *

Deferred follow-ups (this increment). Micrometer metrics wiring (NFR-11) is + * intentionally out — Micrometer stays a {@code compileOnly} dependency until the service tier + * wires a real {@code MeterRegistry}. A KMS-backed {@link + * com.codeheadsystems.velocity.core.NamespaceSaltProvider} is deferred (only an in-memory default + * ships here; production MUST source the salt from a separate trust domain per §15 R11). + * Non-primary hot-reload sources (loading definitions from a store/watcher, OQ-C) are deferred; the + * {@link com.codeheadsystems.velocity.core.MutableFeatureDefinitionProvider} provides the + * atomic-swap mechanism a source would drive. + */ +@NullMarked package com.codeheadsystems.velocity.core; + +import org.jspecify.annotations.NullMarked; diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/BackendRegistryTest.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/BackendRegistryTest.java new file mode 100644 index 0000000..753b2d5 --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/BackendRegistryTest.java @@ -0,0 +1,42 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.codeheadsystems.velocity.testkit.InMemoryVelocityBackend; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class BackendRegistryTest { + + private final InMemoryVelocityBackend backend = new InMemoryVelocityBackend(); + + @Test + void resolvesRegisteredBackend() { + final BackendRegistry registry = new BackendRegistry(Map.of("memory", backend)); + assertThat(registry.backend("memory")).isSameAs(backend); + assertThat(registry.names()).containsExactly("memory"); + } + + @Test + void unknownBackendThrowsWithKnownNames() { + final BackendRegistry registry = new BackendRegistry(Map.of("memory", backend)); + assertThatThrownBy(() -> registry.backend("redis")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("redis") + .hasMessageContaining("memory"); + } + + @Test + void emptyRegistryIsRejected() { + assertThatThrownBy(() -> new BackendRegistry(Map.of())) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void blankNameIsRejected() { + assertThatThrownBy(() -> new BackendRegistry(Map.of(" ", backend))) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/CapabilityValidatorTest.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/CapabilityValidatorTest.java new file mode 100644 index 0000000..6c2d850 --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/CapabilityValidatorTest.java @@ -0,0 +1,117 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.core.model.CapabilityValidationResult; +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.spi.model.AggregationType; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities.WindowSpec; +import com.codeheadsystems.velocity.spi.model.Exactness; +import com.codeheadsystems.velocity.spi.model.ReadYourWriteLevel; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowType; +import java.time.Duration; +import java.util.List; +import java.util.Set; +import org.junit.jupiter.api.Test; + +class CapabilityValidatorTest { + + private final CapabilityValidator validator = new CapabilityValidator(); + + private static BackendCapabilities caps( + final Set aggregations, + final List windows, + final Duration maxRetention) { + return new BackendCapabilities( + aggregations, + windows.stream().map(w -> new WindowSpec(w, Exactness.EXACT, w.duration())).toList(), + false, + Long.MAX_VALUE, + Long.MAX_VALUE, + maxRetention, + ReadYourWriteLevel.ATOMIC, + false, + false, + Integer.MAX_VALUE); + } + + @Test + void validDefinitionPasses() { + final CapabilityValidationResult result = + validator.validate( + Fixtures.cardCount(), + caps( + Set.of(AggregationType.COUNT), + List.of(Fixtures.TUMBLING_1M, Fixtures.TUMBLING_1H), + Duration.ofDays(30))); + assertThat(result.valid()).isTrue(); + assertThat(result.violations()).isEmpty(); + } + + @Test + void unsupportedWindowIsRejected() { + final CapabilityValidationResult result = + validator.validate( + Fixtures.cardCount(), + caps( + Set.of(AggregationType.COUNT), List.of(Fixtures.TUMBLING_1M), Duration.ofDays(30))); + assertThat(result.valid()).isFalse(); + assertThat(result.violations()) + .anySatisfy(v -> assertThat(v.message()).contains("PT1H").contains("not supported")); + } + + @Test + void unsupportedAggregationIsRejected() { + final CapabilityValidationResult result = + validator.validate( + Fixtures.cardSum(), + caps( + Set.of(AggregationType.COUNT), + List.of(Fixtures.TUMBLING_1M, Fixtures.TUMBLING_1H), + Duration.ofDays(30))); + assertThat(result.violations()) + .anySatisfy(v -> assertThat(v.message()).contains("SUM").contains("not supported")); + } + + @Test + void retentionTooShortIsRejected() { + final FeatureDefinition oneHour = + FeatureDefinition.count( + "card.count.1h", "card", Fixtures.BACKEND, List.of(Fixtures.TUMBLING_1H)); + final CapabilityValidationResult result = + validator.validate( + oneHour, + caps( + Set.of(AggregationType.COUNT), + List.of(Fixtures.TUMBLING_1H), + Duration.ofMinutes(30))); + assertThat(result.violations()) + .anySatisfy(v -> assertThat(v.message()).contains("maxRetention").contains("FR-22a")); + } + + @Test + void distinctOnCountOnlyBackendIsRejected() { + final CapabilityValidationResult result = + validator.validate( + Fixtures.cardDistinctIp(), + caps( + Set.of(AggregationType.COUNT), List.of(Fixtures.TUMBLING_1H), Duration.ofDays(30))); + assertThat(result.violations()).anySatisfy(v -> assertThat(v.message()).contains("DISTINCT")); + } + + @Test + void collectsAllViolationsNotJustTheFirst() { + // Wrong aggregation AND an unsupported window in the same definition. + final CapabilityValidationResult result = + validator.validate( + Fixtures.cardSum(), + caps( + Set.of(AggregationType.COUNT), + List.of(new Window(Duration.ofSeconds(5), WindowType.SLIDING)), + Duration.ofDays(30))); + assertThat(result.violations()).hasSizeGreaterThanOrEqualTo(2); + } +} diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/DefinitionVersionHasherTest.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/DefinitionVersionHasherTest.java new file mode 100644 index 0000000..8a250ed --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/DefinitionVersionHasherTest.java @@ -0,0 +1,43 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import java.util.List; +import org.junit.jupiter.api.Test; + +class DefinitionVersionHasherTest { + + @Test + void isDeterministic() { + assertThat(DefinitionVersionHasher.hash(Fixtures.allDefinitions())) + .isEqualTo(DefinitionVersionHasher.hash(Fixtures.allDefinitions())); + } + + @Test + void isOrderIndependent() { + final List forward = Fixtures.allDefinitions(); + final List reversed = + List.of( + Fixtures.ipCount(), + Fixtures.cardDistinctIp(), + Fixtures.cardSum(), + Fixtures.cardCount()); + assertThat(DefinitionVersionHasher.hash(forward)) + .isEqualTo(DefinitionVersionHasher.hash(reversed)); + } + + @Test + void changesWhenADefinitionChanges() { + assertThat(DefinitionVersionHasher.hash(List.of(Fixtures.cardCount()))) + .isNotEqualTo(DefinitionVersionHasher.hash(List.of(Fixtures.cardSum()))); + } + + @Test + void emptySetHashesStably() { + assertThat(DefinitionVersionHasher.hash(List.of())) + .isEqualTo(DefinitionVersionHasher.hash(List.of())) + .isNotBlank(); + } +} diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/DimensionHasherTest.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/DimensionHasherTest.java new file mode 100644 index 0000000..b61a4c4 --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/DimensionHasherTest.java @@ -0,0 +1,47 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.spi.model.DistinctMember; +import com.codeheadsystems.velocity.spi.model.Namespace; +import org.junit.jupiter.api.Test; + +class DimensionHasherTest { + + private static final Namespace ACME = new Namespace("acme"); + private final DimensionHasher hasher = + new DimensionHasher(new InMemoryNamespaceSaltProvider("fixed".getBytes(UTF_8))); + + @Test + void sameValueHashesDeterministically() { + assertThat(hasher.hash(ACME, "203.0.113.7")).isEqualTo(hasher.hash(ACME, "203.0.113.7")); + } + + @Test + void differentValuesProduceDifferentMembers() { + assertThat(hasher.hash(ACME, "1.1.1.1")).isNotEqualTo(hasher.hash(ACME, "2.2.2.2")); + } + + @Test + void tokenIsAFixedWidthHashNotTheRawValue() { + final DistinctMember member = hasher.hash(ACME, "203.0.113.7"); + assertThat(member.token()).hasSize(32); + // FR-38/R11: the raw dimension value is never what gets stored. + assertThat(member.token()).isNotEqualTo("203.0.113.7".getBytes(UTF_8)); + } + + @Test + void saltIsPerNamespace() { + assertThat(hasher.hash(ACME, "203.0.113.7")) + .isNotEqualTo(hasher.hash(new Namespace("beta"), "203.0.113.7")); + } + + @Test + void differentMasterSecretsProduceDifferentMembers() { + final DimensionHasher other = + new DimensionHasher(new InMemoryNamespaceSaltProvider("other".getBytes(UTF_8))); + assertThat(hasher.hash(ACME, "203.0.113.7")).isNotEqualTo(other.hash(ACME, "203.0.113.7")); + } +} diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/FeatureDefinitionYamlTest.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/FeatureDefinitionYamlTest.java new file mode 100644 index 0000000..237ffc7 --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/FeatureDefinitionYamlTest.java @@ -0,0 +1,95 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.core.model.SubjectSource; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.Namespace; +import java.util.List; +import org.junit.jupiter.api.Test; + +class FeatureDefinitionYamlTest { + + private static final Namespace ACME = new Namespace("acme"); + private final FeatureDefinitionYaml yaml = new FeatureDefinitionYaml(); + + private static FeatureDefinition distinctWithThreshold() { + return new FeatureDefinition( + "card.distinct.ip.capped", + "card", + SubjectSource.primary(), + Aggregation.distinct("ip"), + List.of(Fixtures.TUMBLING_1H), + Fixtures.BACKEND, + 5000L); + } + + @Test + void exportThenImportRoundTrips() { + final List definitions = + List.of( + Fixtures.cardCount(), + Fixtures.cardSum(), + Fixtures.cardDistinctIp(), + Fixtures.ipCount(), + distinctWithThreshold()); + + final String exported = yaml.export(ACME, definitions); + final List imported = yaml.importDefinitions(exported); + + assertThat(imported).isEqualTo(definitions); + assertThat(yaml.importNamespace(exported)).isEqualTo(ACME); + } + + @Test + void exportedYamlIsHumanReadable() { + final String exported = yaml.export(ACME, List.of(Fixtures.ipCount())); + assertThat(exported) + .contains("acme") + .contains("ip.count") + .contains("FROM_DIMENSION") + .contains("PT1H"); + } + + @Test + void subjectSourceDefaultsToPrimaryWhenAbsent() { + final String withoutSource = + """ + namespace: acme + definitions: + - name: card.count + subjectType: card + backend: memory + aggregation: + type: COUNT + windows: + - duration: PT1H + type: TUMBLING + """; + final List imported = yaml.importDefinitions(withoutSource); + assertThat(imported).hasSize(1); + assertThat(imported.get(0).subjectSource()).isInstanceOf(SubjectSource.Primary.class); + } + + @Test + void unknownFieldsAreToleratedForForwardCompatibility() { + final String withExtra = + """ + namespace: acme + futureField: ignored + definitions: + - name: card.count + subjectType: card + backend: memory + someNewKnob: 7 + aggregation: + type: COUNT + windows: + - duration: PT1M + type: TUMBLING + """; + assertThat(yaml.importDefinitions(withExtra)).hasSize(1); + } +} diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/Fixtures.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/Fixtures.java new file mode 100644 index 0000000..f55468b --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/Fixtures.java @@ -0,0 +1,60 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.spi.VelocityBackend; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowType; +import java.time.Duration; +import java.util.List; +import java.util.Map; + +/** Shared test fixtures for the engine tests: standard windows, the canonical definition set. */ +final class Fixtures { + + static final String BACKEND = "memory"; + static final Window TUMBLING_1M = new Window(Duration.ofMinutes(1), WindowType.TUMBLING); + static final Window TUMBLING_1H = new Window(Duration.ofHours(1), WindowType.TUMBLING); + + private Fixtures() {} + + static FeatureDefinition cardCount() { + return FeatureDefinition.count( + "card.count", "card", BACKEND, List.of(TUMBLING_1M, TUMBLING_1H)); + } + + static FeatureDefinition cardSum() { + return FeatureDefinition.sum("card.sum", "card", BACKEND, List.of(TUMBLING_1M, TUMBLING_1H)); + } + + static FeatureDefinition cardDistinctIp() { + return FeatureDefinition.distinct( + "card.distinct.ip", "card", "ip", BACKEND, List.of(TUMBLING_1H)); + } + + /** An {@code ip}-subject COUNT whose subject is derived from the event's {@code ip} dimension. */ + static FeatureDefinition ipCount() { + return new FeatureDefinition( + "ip.count", + "ip", + com.codeheadsystems.velocity.core.model.SubjectSource.fromDimension("ip"), + com.codeheadsystems.velocity.spi.model.Aggregation.count(), + List.of(TUMBLING_1M, TUMBLING_1H), + BACKEND, + null); + } + + static List allDefinitions() { + return List.of(cardCount(), cardSum(), cardDistinctIp(), ipCount()); + } + + static VelocityEngine engine( + final MutableFeatureDefinitionProvider provider, final VelocityBackend backend) { + final BackendRegistry registry = new BackendRegistry(Map.of(BACKEND, backend)); + final DimensionHasher hasher = + new DimensionHasher(new InMemoryNamespaceSaltProvider("test-master".getBytes(UTF_8))); + return new VelocityEngine(registry, provider, hasher, new CapabilityValidator()); + } +} diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/MutableFeatureDefinitionProviderTest.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/MutableFeatureDefinitionProviderTest.java new file mode 100644 index 0000000..cf00b8b --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/MutableFeatureDefinitionProviderTest.java @@ -0,0 +1,86 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.core.model.FeatureDefinitions; +import com.codeheadsystems.velocity.spi.model.Namespace; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +class MutableFeatureDefinitionProviderTest { + + private static final Namespace ACME = new Namespace("acme"); + private final MutableFeatureDefinitionProvider provider = new MutableFeatureDefinitionProvider(); + + @Test + void unknownNamespaceReturnsEmptySnapshot() { + final FeatureDefinitions snapshot = provider.definitions(ACME); + assertThat(snapshot.isEmpty()).isTrue(); + assertThat(snapshot.versionHash()).isNotBlank(); + } + + @Test + void reloadSwapsSnapshotAndChangesVersion() { + final String empty = provider.definitions(ACME).versionHash(); + provider.reload(ACME, List.of(Fixtures.cardCount())); + final FeatureDefinitions after = provider.definitions(ACME); + assertThat(after.versionHash()).isNotEqualTo(empty); + assertThat(after.definition("card.count")).isPresent(); + } + + @Test + void concurrentReadsAlwaysSeeAWholeSnapshot() throws InterruptedException { + final List setA = List.of(Fixtures.cardCount()); + final List setB = + List.of(Fixtures.cardCount(), Fixtures.cardSum(), Fixtures.cardDistinctIp()); + provider.reload(ACME, setA); + + final Set observedSizes = ConcurrentHashMap.newKeySet(); + final int readers = 6; + final CountDownLatch start = new CountDownLatch(1); + final CountDownLatch done = new CountDownLatch(readers + 1); + + final Thread writer = + new Thread( + () -> { + await(start); + for (int i = 0; i < 2000; i++) { + provider.reload(ACME, (i % 2 == 0) ? setA : setB); + } + done.countDown(); + }); + writer.start(); + + for (int r = 0; r < readers; r++) { + new Thread( + () -> { + await(start); + for (int i = 0; i < 5000; i++) { + observedSizes.add(provider.definitions(ACME).definitions().size()); + } + done.countDown(); + }) + .start(); + } + + start.countDown(); + assertThat(done.await(30, TimeUnit.SECONDS)).isTrue(); + // Every observation was a whole snapshot — only setA's or setB's size, never a torn count. + assertThat(observedSizes).isSubsetOf(setA.size(), setB.size()); + } + + private static void await(final CountDownLatch latch) { + try { + latch.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } +} diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/UnsupportedMixinTest.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/UnsupportedMixinTest.java new file mode 100644 index 0000000..64df328 --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/UnsupportedMixinTest.java @@ -0,0 +1,98 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.model.ApplyContext; +import com.codeheadsystems.velocity.spi.model.ApplyResult; +import com.codeheadsystems.velocity.spi.model.ApplyStatus; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities.WindowSpec; +import com.codeheadsystems.velocity.spi.model.Exactness; +import com.codeheadsystems.velocity.spi.model.FeatureResult; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.spi.model.Namespace; +import com.codeheadsystems.velocity.spi.model.QueryContext; +import com.codeheadsystems.velocity.spi.model.QueryTuple; +import com.codeheadsystems.velocity.spi.model.ReadYourWriteLevel; +import com.codeheadsystems.velocity.spi.model.Subject; +import java.math.BigDecimal; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.jspecify.annotations.Nullable; +import org.junit.jupiter.api.Test; + +/** + * When a definition is bound to a backend that does not implement the required aggregation mix-in + * (an inconsistent configuration the capability validator would normally catch), the engine emits a + * distinguishable FAILED outcome rather than throwing. + */ +class UnsupportedMixinTest { + + private static final Namespace ACME = new Namespace("acme"); + private static final Subject CARD = new Subject("card", "1234"); + + @Test + void sumIntentToCountOnlyBackendFailsGracefully() { + final BackendRegistry registry = new BackendRegistry(Map.of("fake", new CountOnlyBackend())); + final MutableFeatureDefinitionProvider provider = new MutableFeatureDefinitionProvider(); + // Install directly (bypassing engine.reload's validation) to reach the defensive path. + provider.reload( + ACME, + List.of(FeatureDefinition.sum("card.sum", "card", "fake", List.of(Fixtures.TUMBLING_1H)))); + final VelocityEngine engine = + new VelocityEngine( + registry, + provider, + new DimensionHasher(new InMemoryNamespaceSaltProvider("m".getBytes(UTF_8))), + new CapabilityValidator()); + + final ApplyResult result = engine.record(ACME, CARD, Map.of(), new BigDecimal("100")); + assertThat(result.perFeature()) + .allSatisfy(pf -> assertThat(pf.status()).isEqualTo(ApplyStatus.FAILED)) + .allSatisfy(pf -> assertThat(pf.result()).isInstanceOf(FeatureResult.Failure.class)); + + // The query path is symmetric: a Failure per requested window, never a thrown exception. + assertThat(engine.query(ACME, CARD, "card.sum")) + .allSatisfy(r -> assertThat(r).isInstanceOf(FeatureResult.Failure.class)); + } + + /** A backend that only implements COUNT, used to exercise the engine's missing-mix-in path. */ + private static final class CountOnlyBackend implements CountStore { + + @Override + public BackendCapabilities capabilities() { + return new BackendCapabilities( + Set.of(com.codeheadsystems.velocity.spi.model.AggregationType.COUNT), + List.of(new WindowSpec(Fixtures.TUMBLING_1H, Exactness.EXACT, Duration.ofHours(1))), + false, + Long.MAX_VALUE, + Long.MAX_VALUE, + Duration.ofDays(30), + ReadYourWriteLevel.ATOMIC, + false, + false, + Integer.MAX_VALUE); + } + + @Override + public ApplyResult applyCount(final ApplyContext ctx, final List intents) { + return new ApplyResult(List.of()); + } + + @Override + public List queryCount(final QueryContext ctx, final List tuples) { + return List.of(); + } + + @Override + public void purge(final Namespace namespace, final @Nullable Subject subject) { + // no-op + } + } +} diff --git a/velocity-core/src/test/java/com/codeheadsystems/velocity/core/VelocityEngineTest.java b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/VelocityEngineTest.java new file mode 100644 index 0000000..97ff006 --- /dev/null +++ b/velocity-core/src/test/java/com/codeheadsystems/velocity/core/VelocityEngineTest.java @@ -0,0 +1,228 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.core; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.codeheadsystems.velocity.core.model.FeatureDefinition; +import com.codeheadsystems.velocity.spi.model.ApplyResult; +import com.codeheadsystems.velocity.spi.model.ApplyStatus; +import com.codeheadsystems.velocity.spi.model.FeatureResult; +import com.codeheadsystems.velocity.spi.model.FeatureValue; +import com.codeheadsystems.velocity.spi.model.Namespace; +import com.codeheadsystems.velocity.spi.model.PerFeature; +import com.codeheadsystems.velocity.spi.model.Subject; +import com.codeheadsystems.velocity.testkit.InMemoryVelocityBackend; +import com.codeheadsystems.velocity.testkit.MutableClock; +import java.math.BigDecimal; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** End-to-end tests driving the real in-memory backend through the engine. */ +class VelocityEngineTest { + + private static final Namespace ACME = new Namespace("acme"); + private static final Subject CARD = new Subject("card", "1234"); + private static final Map DIMENSIONS = + Map.of("ip", "203.0.113.7", "merchant", "m-88"); + private static final BigDecimal VALUE = new BigDecimal("14950"); + + private MutableClock clock; + private InMemoryVelocityBackend backend; + private MutableFeatureDefinitionProvider provider; + private VelocityEngine engine; + + @BeforeEach + void setUp() { + clock = new MutableClock(Instant.parse("2026-03-15T12:30:30Z")); + backend = new InMemoryVelocityBackend(clock); + provider = new MutableFeatureDefinitionProvider(); + engine = Fixtures.engine(provider, backend); + engine.reload(ACME, Fixtures.allDefinitions()); + } + + @Test + void fanOut_updatesEveryMatchedFeatureAcrossSubjects() { + engine.record(ACME, CARD, DIMENSIONS, VALUE); + + assertThat(value(engine.query(ACME, CARD, "card.count"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("1"); + assertThat(value(engine.query(ACME, CARD, "card.sum"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("14950"); + assertThat(value(engine.query(ACME, CARD, "card.distinct.ip"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("1"); + // The ip.count feature fanned out to a *different* subject derived from the ip dimension. + final Subject ipSubject = new Subject("ip", "203.0.113.7"); + assertThat(value(engine.query(ACME, ipSubject, "ip.count"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("1"); + } + + @Test + void record_resultReflectsTheWrite_readYourWrite() { + final ApplyResult result = engine.record(ACME, CARD, DIMENSIONS, VALUE); + + final PerFeature cardCount1h = + result.perFeature().stream() + .filter( + pf -> + pf.feature().name().equals("card.count") + && pf.result() instanceof FeatureResult.Success success + && success.value().window().equals(Fixtures.TUMBLING_1H)) + .findFirst() + .orElseThrow(); + assertThat(cardCount1h.status()).isEqualTo(ApplyStatus.APPLIED); + final FeatureValue value = ((FeatureResult.Success) cardCount1h.result()).value(); + assertThat(value.value()).isEqualByComparingTo("1"); + // FR-40: every returned value is stamped with the definition version it was computed under. + assertThat(value.definitionVersionHash()) + .isEqualTo(provider.definitions(ACME).versionHash()) + .isNotNull(); + } + + @Test + void record_noMatchingDefinition_isEmptyNotError() { + // A subject whose type matches no definition and no dimension-derived fan-out applies. + final ApplyResult result = engine.record(ACME, new Subject("device", "d-1"), Map.of(), null); + assertThat(result.perFeature()).isEmpty(); + } + + @Test + void record_sumWithoutValue_isSkippedWithReason() { + final ApplyResult result = engine.record(ACME, CARD, DIMENSIONS, null); + + final PerFeature sum = + result.perFeature().stream() + .filter(pf -> pf.feature().name().equals("card.sum")) + .findFirst() + .orElseThrow(); + assertThat(sum.status()).isEqualTo(ApplyStatus.SKIPPED); + assertThat(sum.result()).isInstanceOf(FeatureResult.Failure.class); + // COUNT still applied even though SUM was skipped. + assertThat(value(engine.query(ACME, CARD, "card.count"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("1"); + } + + @Test + void record_distinctWithoutDimension_isSkippedWithReason() { + // The card.distinct.ip feature resolves its (primary card) subject, but the "ip" dimension it + // counts distinctly is absent, so no member can be hashed and the write is skipped — the + // DISTINCT twin of record_sumWithoutValue_isSkippedWithReason (FR-18 fan-out skip path). + final ApplyResult result = engine.record(ACME, CARD, Map.of(), null); + + final PerFeature distinct = + result.perFeature().stream() + .filter(pf -> pf.feature().name().equals("card.distinct.ip")) + .findFirst() + .orElseThrow(); + assertThat(distinct.status()).isEqualTo(ApplyStatus.SKIPPED); + assertThat(distinct.result()).isInstanceOf(FeatureResult.Failure.class); + // COUNT still applied even though DISTINCT was skipped. + assertThat(value(engine.query(ACME, CARD, "card.count"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("1"); + } + + @Test + void distinct_countsDistinctHashedValues() { + engine.record(ACME, CARD, Map.of("ip", "1.1.1.1"), null); + engine.record(ACME, CARD, Map.of("ip", "2.2.2.2"), null); + assertThat(value(engine.query(ACME, CARD, "card.distinct.ip"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("2"); + + // Recording the same ip again does not raise cardinality. + engine.record(ACME, CARD, Map.of("ip", "1.1.1.1"), null); + assertThat(value(engine.query(ACME, CARD, "card.distinct.ip"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("2"); + } + + @Test + void namespaces_areIsolated() { + final Namespace beta = new Namespace("beta"); + engine.reload(beta, Fixtures.allDefinitions()); + + engine.record(ACME, CARD, DIMENSIONS, VALUE); + engine.record(ACME, CARD, DIMENSIONS, VALUE); + engine.record(beta, CARD, DIMENSIONS, VALUE); + + assertThat(value(engine.query(ACME, CARD, "card.count"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("2"); + assertThat(value(engine.query(beta, CARD, "card.count"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("1"); + } + + @Test + void reload_isAtomicAndChangesVersionAndResolvableFeatures() { + final String before = provider.definitions(ACME).versionHash(); + + // Swap: keep card.count, drop card.sum, keep the distinct + ip features. + engine.reload( + ACME, List.of(Fixtures.cardCount(), Fixtures.cardDistinctIp(), Fixtures.ipCount())); + final String after = provider.definitions(ACME).versionHash(); + assertThat(after).isNotEqualTo(before); + + // The removed feature no longer resolves. + assertThatThrownBy(() -> engine.query(ACME, CARD, "card.sum")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("card.sum"); + + // A retained feature still works. + engine.record(ACME, CARD, DIMENSIONS, VALUE); + assertThat(value(engine.query(ACME, CARD, "card.count"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("1"); + } + + @Test + void reload_rejectsInvalidDefinitionSet_andDoesNotApply() { + final String before = provider.definitions(ACME).versionHash(); + final var badWindow = + FeatureDefinition.count( + "card.count.1d", + "card", + Fixtures.BACKEND, + List.of( + new com.codeheadsystems.velocity.spi.model.Window( + java.time.Duration.ofDays(1), + com.codeheadsystems.velocity.spi.model.WindowType.TUMBLING))); + + assertThatThrownBy(() -> engine.reload(ACME, List.of(Fixtures.cardCount(), badWindow))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("card.count.1d"); + + // Rejected atomically: the previous snapshot is untouched. + assertThat(provider.definitions(ACME).versionHash()).isEqualTo(before); + } + + @Test + void capabilities_isPassthrough() { + assertThat(engine.capabilities(Fixtures.BACKEND)).isSameAs(backend.capabilities()); + } + + @Test + void purge_erasesAcrossBackends() { + engine.record(ACME, CARD, DIMENSIONS, VALUE); + engine.purge(ACME, CARD); + assertThat(value(engine.query(ACME, CARD, "card.count"), Fixtures.TUMBLING_1H)) + .isEqualByComparingTo("0"); + } + + @Test + void query_batchReturnsPerFeatureResults() { + engine.record(ACME, CARD, DIMENSIONS, VALUE); + final Map> results = + engine.query(ACME, CARD, List.of("card.count", "card.sum")); + assertThat(results).containsOnlyKeys("card.count", "card.sum"); + assertThat(value(results.get("card.sum"), Fixtures.TUMBLING_1H)).isEqualByComparingTo("14950"); + } + + private static BigDecimal value( + final List results, + final com.codeheadsystems.velocity.spi.model.Window window) { + return results.stream() + .filter(r -> r instanceof FeatureResult.Success s && s.value().window().equals(window)) + .map(r -> ((FeatureResult.Success) r).value().value()) + .findFirst() + .orElseThrow(() -> new AssertionError("no success result for window " + window)); + } +}