From d3d2df1fc702d07abf3fe70ed20ebc7ba442fe2e Mon Sep 17 00:00:00 2001 From: Ned Wolpert Date: Sat, 18 Jul 2026 18:21:47 -0700 Subject: [PATCH] Add velocity-backend-jdbi: the Postgres tumbling reference backend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first real velocity-spi backend (v1 tumbling reference, P10/§8) — Postgres via JDBI, validated against a real Postgres with Testcontainers. - JdbiVelocityBackend implements CountStore/SumStore/DistinctStore + TumblingSupport + SeedSupport. Exact, ATOMIC read-your-write: each apply is one transaction whose per-bucket upsert (INSERT ... ON CONFLICT ... DO UPDATE ... RETURNING) returns the post-write value, so concurrent applies never lose an increment (NFR-6, acc. #1). - Backend owns bucketing + the clock (ADR 0003/FR-3): a tumbling window resolves to the aligned current bucket floor(nowMs/Dms)*Dms — byte-identical to velocity-testkit WindowMath, so a JDBI value equals the in-memory reference. Storage identity is (namespace, subject, aggregation[dimension + type-via-table], window, bucket) — NOT feature name (a QueryTuple carries none), matching InMemory's StoreKey. - Three per-aggregation tables (counts / sums NUMERIC(38,0) cents / distinct_members BYTEA, exact). Idempotent CREATE ... IF NOT EXISTS migrate(). Exact-only distinct (HLL degradation is a documented follow-up). Unsupported (sliding) window -> Failure(UNSUPPORTED_WINDOW) on apply and query, never a silent 0. HLL seed rejected. - No Dagger/Dropwizard/Jackson (a backend depends only on velocity-spi + JDBI/Postgres/ Hikari/slf4j). Tests (real Postgres via Testcontainers, Docker required): 33 tests, 0 skipped. Wires the tumbling-compatible SHARED TCK scenarios (TumblingScenarios, SeedSupportScenarios, CapabilityConformanceScenarios) + a 16-thread concurrency test asserting exact atomicity (acc. #1) + bespoke tumbling Count/Sum/Distinct/Purge tests. Coverage LINE 100% / BRANCH 97.4% (gate 80/70). Validated by the change-validator agent: VERDICT APPROVE (round-trip fix confirmed, concurrency atomic, bespoke tests NOT weaker than the shared scenarios). FOLLOW-UP (before velocity-backend-redis): parameterize the window-agnostic TCK scenarios (Count/Sum/Distinct/Purge currently hardcode a sliding window) so every backend runs the SHARED scenarios instead of bespoke ones — see PR body. Co-Authored-By: Claude Opus 4.8 (1M context) --- settings.gradle.kts | 2 +- velocity-backend-jdbi/build.gradle.kts | 49 ++ .../backend/jdbi/JdbiVelocityBackend.java | 658 ++++++++++++++++++ .../velocity/backend/jdbi/package-info.java | 31 + .../backend/jdbi/AbstractJdbiBackendTest.java | 27 + .../backend/jdbi/JdbiAggregationTest.java | 416 +++++++++++ .../backend/jdbi/JdbiConcurrencyTest.java | 91 +++ .../backend/jdbi/JdbiConformanceTckTest.java | 89 +++ .../backend/jdbi/PostgresSupport.java | 74 ++ 9 files changed, 1436 insertions(+), 1 deletion(-) create mode 100644 velocity-backend-jdbi/build.gradle.kts create mode 100644 velocity-backend-jdbi/src/main/java/com/codeheadsystems/velocity/backend/jdbi/JdbiVelocityBackend.java create mode 100644 velocity-backend-jdbi/src/main/java/com/codeheadsystems/velocity/backend/jdbi/package-info.java create mode 100644 velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/AbstractJdbiBackendTest.java create mode 100644 velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiAggregationTest.java create mode 100644 velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiConcurrencyTest.java create mode 100644 velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiConformanceTckTest.java create mode 100644 velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/PostgresSupport.java diff --git a/settings.gradle.kts b/settings.gradle.kts index 694570a..33a0274 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -58,7 +58,7 @@ include("velocity-api") // OpenAPI 3.1 document + shared API DTOs (source o include("velocity-testkit") // in-memory velocity-spi impl, conformance TCK scenarios, fixtures // Phase 2 — v1 reference backends (tumbling + sliding), the service tier, and the generated client. -// include("velocity-backend-jdbi") // Postgres/JDBI backend — v1 tumbling reference +include("velocity-backend-jdbi") // Postgres/JDBI backend — v1 tumbling reference // include("velocity-backend-redis") // Redis/Lettuce backend — v1 sliding / hot-path reference // include("velocity-dropwizard") // Dropwizard bundle, Jersey resources, Dagger wiring, API-key auth // include("velocity-client-java") // Java client generated from OpenAPI via openapi-generator diff --git a/velocity-backend-jdbi/build.gradle.kts b/velocity-backend-jdbi/build.gradle.kts new file mode 100644 index 0000000..1d152ba --- /dev/null +++ b/velocity-backend-jdbi/build.gradle.kts @@ -0,0 +1,49 @@ +plugins { + id("velocity.library-conventions") + id("velocity.test-conventions") + id("velocity.publish-conventions") +} + +description = + "velocity-backend-jdbi: the v1 Postgres/JDBI tumbling reference backend — an exact, atomic," + + " read-your-write VelocityBackend (COUNT/SUM/DISTINCT over tumbling windows) that passes the" + + " velocity-testkit conformance TCK against real Postgres via Testcontainers." + +// JDBI, HikariCP, PostgreSQL and Testcontainers ship as automatic modules (no module-info); the +// baseline -Xlint:all makes their `requires` a warning, which -Werror (library-conventions) turns +// fatal. Mirror the pk-auth JDBI module and silence just the automatic-module lints on both the +// main and test compiles. +tasks.named("compileJava") { + options.compilerArgs.addAll( + listOf("-Xlint:-requires-automatic", "-Xlint:-requires-transitive-automatic"), + ) +} +tasks.named("compileTestJava") { + options.compilerArgs.addAll( + listOf("-Xlint:-requires-automatic", "-Xlint:-requires-transitive-automatic"), + ) +} + +dependencies { + // The frozen contract this backend implements is on the api surface. + api(project(":velocity-spi")) + api(libs.jspecify) + + implementation(libs.jdbi.core) + implementation(libs.postgresql) + implementation(libs.hikaricp) + implementation(libs.slf4j.api) + + // JDBI's class files reference com.google.errorprone.annotations.concurrent.GuardedBy; without + // errorprone-annotations on the compile classpath javac emits an annotation-not-found warning + // that -Werror turns fatal. compileOnly so it is not broadcast as a runtime dependency. + compileOnly(libs.build.errorprone.annotations) + testCompileOnly(libs.build.errorprone.annotations) + + // The testkit drives the conformance TCK (*Scenarios) + MutableClock against this backend. + testImplementation(project(":velocity-testkit")) + testImplementation(libs.testcontainers.core) + testImplementation(libs.testcontainers.junit.jupiter) + testImplementation(libs.testcontainers.postgresql) + testRuntimeOnly(libs.logback.classic) +} diff --git a/velocity-backend-jdbi/src/main/java/com/codeheadsystems/velocity/backend/jdbi/JdbiVelocityBackend.java b/velocity-backend-jdbi/src/main/java/com/codeheadsystems/velocity/backend/jdbi/JdbiVelocityBackend.java new file mode 100644 index 0000000..1319f2f --- /dev/null +++ b/velocity-backend-jdbi/src/main/java/com/codeheadsystems/velocity/backend/jdbi/JdbiVelocityBackend.java @@ -0,0 +1,658 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.jdbi; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.DistinctStore; +import com.codeheadsystems.velocity.spi.SeedSupport; +import com.codeheadsystems.velocity.spi.SumStore; +import com.codeheadsystems.velocity.spi.TumblingSupport; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.AggregationType; +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.BucketValue; +import com.codeheadsystems.velocity.spi.model.DistinctMember; +import com.codeheadsystems.velocity.spi.model.Exactness; +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.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.ReadYourWriteLevel; +import com.codeheadsystems.velocity.spi.model.SeedAggregate; +import com.codeheadsystems.velocity.spi.model.SeedAggregate.CountValue; +import com.codeheadsystems.velocity.spi.model.SeedAggregate.ExactDistinct; +import com.codeheadsystems.velocity.spi.model.SeedAggregate.HllDistinct; +import com.codeheadsystems.velocity.spi.model.SeedAggregate.SumValue; +import com.codeheadsystems.velocity.spi.model.Subject; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowBounds; +import com.codeheadsystems.velocity.spi.model.WindowType; +import java.math.BigDecimal; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import javax.sql.DataSource; +import org.jdbi.v3.core.Handle; +import org.jdbi.v3.core.Jdbi; +import org.jspecify.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The Postgres/JDBI v1 tumbling reference {@link com.codeheadsystems.velocity.spi.VelocityBackend}. + * + *

Implements every aggregation mix-in — {@link CountStore}, {@link SumStore}, {@link + * DistinctStore} — over {@link TumblingSupport tumbling} windows, plus {@link SeedSupport}. Values + * are exact and produced under {@link ReadYourWriteLevel#ATOMIC} read-your-write: each apply is one + * transaction whose per-bucket upsert ({@code INSERT ... ON CONFLICT ... DO UPDATE ... RETURNING}) + * returns the post-write value atomically, so concurrent applies to the same bucket never lose an + * increment (NFR-6, acceptance #1). + * + *

Backend owns bucketing and the clock (ADR 0003, FR-3). A tumbling window of + * duration {@code D} resolves to the current aligned bucket {@code [floor(nowMs/Dms)·Dms, +Dms)}, + * matching {@code velocity-testkit}'s {@code WindowMath} so a JDBI-backed value equals the + * reference in-memory value. The clock is injectable; the default is {@link Clock#systemUTC()} and + * tests inject a controllable clock. + * + *

Storage identity mirrors the reference backend. A bucket is keyed by {@code + * (namespace, subject, aggregation, window, bucket_start)} — NOT by feature name: a {@link + * QueryTuple} carries no feature, so a read resolves to the same bucket a write of the same + * aggregation touched (exactly as the in-memory reference keys by {@code (namespace, subject, + * aggregation)}). The aggregation's DISTINCT {@code dimension} is part of the key; COUNT/SUM use an + * empty dimension. + * + *

Schema. Three per-aggregation tables, keyed by {@code (namespace, + * subject_type, subject_value, dimension, window_type, window_duration_ms, bucket_start_ms)}: + * {@code velocity_counts} (+ {@code count}), {@code velocity_sums} (+ {@code sum_cents + * NUMERIC(38,0)}), and {@code velocity_distinct_members} (+ opaque {@code member BYTEA}, the key + * extended by {@code member} so a re-applied member de-dupes and cardinality is a {@code + * COUNT(*)}). Call {@link #migrate()} once (idempotent {@code CREATE ... IF NOT EXISTS}) before + * use. + */ +public final class JdbiVelocityBackend + implements CountStore, SumStore, DistinctStore, TumblingSupport, SeedSupport { + + private static final Logger LOGGER = LoggerFactory.getLogger(JdbiVelocityBackend.class); + + private static final String KEY_COLUMNS = + "namespace, subject_type, subject_value, dimension, window_type, window_duration_ms," + + " bucket_start_ms"; + private static final String KEY_CONFLICT = + "(namespace, subject_type, subject_value, dimension, window_type, window_duration_ms," + + " bucket_start_ms)"; + private static final String KEY_PREDICATE = + "namespace = :ns AND subject_type = :st AND subject_value = :sv AND dimension = :dim AND" + + " window_type = :wt AND window_duration_ms = :wd AND bucket_start_ms = :bs"; + private static final String KEY_VALUES = ":ns, :st, :sv, :dim, :wt, :wd, :bs"; + + /** + * The idempotent schema DDL. Primary keys double as the upsert conflict target and the read + * lookup index; the distinct table's key is extended by {@code member} so re-applying a member is + * a no-op and {@code COUNT(*)} over the bucket-key prefix (served by the {@code _bucket} index) + * is the exact cardinality. + */ + private static final List SCHEMA_DDL = + List.of( + "CREATE TABLE IF NOT EXISTS velocity_counts (" + + " namespace TEXT NOT NULL," + + " subject_type TEXT NOT NULL," + + " subject_value TEXT NOT NULL," + + " dimension TEXT NOT NULL," + + " window_type TEXT NOT NULL," + + " window_duration_ms BIGINT NOT NULL," + + " bucket_start_ms BIGINT NOT NULL," + + " count BIGINT NOT NULL," + + " seeded BOOLEAN NOT NULL DEFAULT FALSE," + + " PRIMARY KEY " + + KEY_CONFLICT + + ")", + "CREATE TABLE IF NOT EXISTS velocity_sums (" + + " namespace TEXT NOT NULL," + + " subject_type TEXT NOT NULL," + + " subject_value TEXT NOT NULL," + + " dimension TEXT NOT NULL," + + " window_type TEXT NOT NULL," + + " window_duration_ms BIGINT NOT NULL," + + " bucket_start_ms BIGINT NOT NULL," + + " sum_cents NUMERIC(38,0) NOT NULL," + + " seeded BOOLEAN NOT NULL DEFAULT FALSE," + + " PRIMARY KEY " + + KEY_CONFLICT + + ")", + "CREATE TABLE IF NOT EXISTS velocity_distinct_members (" + + " namespace TEXT NOT NULL," + + " subject_type TEXT NOT NULL," + + " subject_value TEXT NOT NULL," + + " dimension TEXT NOT NULL," + + " window_type TEXT NOT NULL," + + " window_duration_ms BIGINT NOT NULL," + + " bucket_start_ms BIGINT NOT NULL," + + " member BYTEA NOT NULL," + + " seeded BOOLEAN NOT NULL DEFAULT FALSE," + + " PRIMARY KEY (namespace, subject_type, subject_value, dimension, window_type," + + " window_duration_ms, bucket_start_ms, member))", + "CREATE INDEX IF NOT EXISTS velocity_distinct_members_bucket" + + " ON velocity_distinct_members (namespace, subject_type, subject_value, dimension," + + " window_type, window_duration_ms, bucket_start_ms)"); + + /** The tumbling windows this backend supports, all EXACT. First entry is the smallest. */ + private static final List SUPPORTED_WINDOWS = + List.of( + new Window(Duration.ofMinutes(1), WindowType.TUMBLING), + new Window(Duration.ofHours(1), WindowType.TUMBLING), + new Window(Duration.ofDays(1), WindowType.TUMBLING)); + + private final Jdbi jdbi; + private final Clock clock; + private final BackendCapabilities capabilities; + + /** + * Creates a backend over a {@link DataSource} (e.g. a HikariCP pool) with the system UTC clock. + * + * @param dataSource the pooled Postgres data source + */ + public JdbiVelocityBackend(final DataSource dataSource) { + this(Jdbi.create(Objects.requireNonNull(dataSource, "dataSource")), Clock.systemUTC()); + } + + /** + * Creates a backend over a {@link Jdbi} handle-source with an injectable clock — the authority + * for every window edge (FR-3). Inject a controllable clock in tests. + * + * @param jdbi the JDBI instance bound to a pooled Postgres data source + * @param clock the backend clock + */ + public JdbiVelocityBackend(final Jdbi jdbi, final Clock clock) { + this.jdbi = Objects.requireNonNull(jdbi, "jdbi"); + this.clock = Objects.requireNonNull(clock, "clock"); + this.capabilities = buildCapabilities(); + } + + private static BackendCapabilities buildCapabilities() { + final List specs = new ArrayList<>(); + for (final Window window : SUPPORTED_WINDOWS) { + // Tumbling stores at bucket granularity == the window duration (current-bucket-only value). + specs.add(new WindowSpec(window, Exactness.EXACT, window.duration())); + } + return new BackendCapabilities( + Set.of(AggregationType.COUNT, AggregationType.SUM, AggregationType.DISTINCT), + specs, + /* distinctHllSliding= */ false, + // Exact-only distinct: HLL degradation (ADR 0006) is a documented follow-up, so the exact + // clamp and the HLL threshold are effectively unbounded — this backend never sketches. + /* distinctExactCardinalityClamp= */ Long.MAX_VALUE, + /* distinctHllThresholdDefault= */ Long.MAX_VALUE, + /* maxRetention= */ Duration.ofDays(90), + ReadYourWriteLevel.ATOMIC, + /* idempotencySupported= */ false, + /* seedSupported= */ true, + /* maxAtomicFanOut= */ Integer.MAX_VALUE); + } + + /** + * Creates the schema idempotently ({@code CREATE TABLE/INDEX IF NOT EXISTS}). Safe to call + * repeatedly and concurrently; call once before the first apply/query. + * + * @return this backend, for chaining + */ + public JdbiVelocityBackend migrate() { + jdbi.useTransaction( + handle -> { + for (final String ddl : SCHEMA_DDL) { + handle.execute(ddl); + } + }); + LOGGER.info("velocity-backend-jdbi schema migrated ({} statements)", SCHEMA_DDL.size()); + return this; + } + + /** + * The clock this backend reads time from; the authority for every window edge (FR-3). + * + * @return the injected clock + */ + public Clock clock() { + return clock; + } + + @Override + public BackendCapabilities capabilities() { + return capabilities; + } + + // ---- COUNT --------------------------------------------------------------------------------- + + @Override + public ApplyResult applyCount(final ApplyContext ctx, final List intents) { + return apply( + ctx, + intents, + (handle, key, intent) -> + BigDecimal.valueOf( + handle + .createQuery( + "INSERT INTO velocity_counts (" + + KEY_COLUMNS + + ", count) VALUES (" + + KEY_VALUES + + ", 1) ON CONFLICT " + + KEY_CONFLICT + + " DO UPDATE SET count = velocity_counts.count + 1 RETURNING count") + .bindMap(key.binding()) + .mapTo(Long.class) + .one())); + } + + @Override + public List queryCount(final QueryContext ctx, final List tuples) { + return query( + ctx, + tuples, + (handle, key) -> + handle + .createQuery("SELECT count FROM velocity_counts WHERE " + KEY_PREDICATE) + .bindMap(key.binding()) + .mapTo(Long.class) + .findOne() + .map(BigDecimal::valueOf) + .orElse(BigDecimal.ZERO)); + } + + // ---- SUM ----------------------------------------------------------------------------------- + + @Override + public ApplyResult applySum(final ApplyContext ctx, final List intents) { + return apply( + ctx, + intents, + (handle, key, intent) -> + handle + .createQuery( + "INSERT INTO velocity_sums (" + + KEY_COLUMNS + + ", sum_cents) VALUES (" + + KEY_VALUES + + ", :val) ON CONFLICT " + + KEY_CONFLICT + + " DO UPDATE SET sum_cents = velocity_sums.sum_cents + :val" + + " RETURNING sum_cents") + .bindMap(key.binding()) + .bind("val", ((SumIntent) intent).valueCents()) + .mapTo(BigDecimal.class) + .one()); + } + + @Override + public List querySum(final QueryContext ctx, final List tuples) { + return query( + ctx, + tuples, + (handle, key) -> + handle + .createQuery("SELECT sum_cents FROM velocity_sums WHERE " + KEY_PREDICATE) + .bindMap(key.binding()) + .mapTo(BigDecimal.class) + .findOne() + .orElse(BigDecimal.ZERO)); + } + + // ---- DISTINCT ------------------------------------------------------------------------------ + + @Override + public ApplyResult applyDistinct(final ApplyContext ctx, final List intents) { + return apply( + ctx, + intents, + (handle, key, intent) -> { + handle + .createUpdate( + "INSERT INTO velocity_distinct_members (" + + KEY_COLUMNS + + ", member) VALUES (" + + KEY_VALUES + + ", :member) ON CONFLICT DO NOTHING") + .bindMap(key.binding()) + .bind("member", ((DistinctIntent) intent).member().token()) + .execute(); + return cardinality(handle, key); + }); + } + + @Override + public List queryDistinct(final QueryContext ctx, final List tuples) { + return query(ctx, tuples, JdbiVelocityBackend::cardinality); + } + + private static BigDecimal cardinality(final Handle handle, final BucketKey key) { + return BigDecimal.valueOf( + handle + .createQuery("SELECT count(*) FROM velocity_distinct_members WHERE " + KEY_PREDICATE) + .bindMap(key.binding()) + .mapTo(Long.class) + .one()); + } + + // ---- shared apply / query ------------------------------------------------------------------ + + private ApplyResult apply( + final ApplyContext ctx, final List intents, final BucketWriter writer) { + Objects.requireNonNull(ctx, "ctx"); + Objects.requireNonNull(intents, "intents"); + final Namespace namespace = ctx.namespace(); + final Instant now = clock.instant(); + return new ApplyResult( + jdbi.inTransaction( + handle -> { + final List outcomes = new ArrayList<>(); + for (final Intent intent : intents) { + final Feature feature = intent.feature(); + for (final Window window : feature.windows()) { + if (!capabilities.supportsWindow(window)) { + outcomes.add( + new PerFeature( + feature, + ApplyStatus.FAILED, + FeatureResult.failure( + FailureCode.UNSUPPORTED_WINDOW, unsupported(window)))); + continue; + } + final BucketKey key = + keyFor(namespace, intent.subject(), feature.aggregation(), window, now); + final BigDecimal value = writer.write(handle, key, intent); + outcomes.add( + new PerFeature( + feature, + ApplyStatus.APPLIED, + FeatureResult.success(featureValue(feature, window, value, key, now)))); + } + } + return outcomes; + })); + } + + private List query( + final QueryContext ctx, final List tuples, final BucketReader reader) { + Objects.requireNonNull(ctx, "ctx"); + Objects.requireNonNull(tuples, "tuples"); + final Namespace namespace = ctx.namespace(); + final Instant now = clock.instant(); + return jdbi.withHandle( + handle -> { + final List results = new ArrayList<>(); + for (final QueryTuple tuple : tuples) { + final Window window = tuple.window(); + if (!capabilities.supportsWindow(window)) { + results.add( + FeatureResult.failure(FailureCode.UNSUPPORTED_WINDOW, unsupported(window))); + continue; + } + final BucketKey key = + keyFor(namespace, tuple.subject(), tuple.aggregation(), window, now); + final BigDecimal value = reader.read(handle, key); + results.add( + FeatureResult.success( + featureValue(syntheticFeature(tuple), window, value, key, now))); + } + return results; + }); + } + + // ---- SEED ---------------------------------------------------------------------------------- + + @Override + public void seed( + final Namespace namespace, + final Subject subject, + final Feature feature, + final List buckets) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(subject, "subject"); + Objects.requireNonNull(feature, "feature"); + Objects.requireNonNull(buckets, "buckets"); + jdbi.useTransaction( + handle -> { + for (final BucketValue bucket : buckets) { + final Duration span = Duration.between(bucket.bounds().start(), bucket.bounds().end()); + final Window window = matchingSupportedWindow(feature, span); + if (window == null) { + throw new IllegalArgumentException( + "seed bucket of " + + span + + " matches no supported window of feature '" + + feature.name() + + "'"); + } + final BucketKey key = + keyAt( + namespace, + subject, + feature.aggregation(), + window, + bucket.bounds().start(), + span.toMillis()); + seedBucket(handle, key, bucket.aggregate()); + } + }); + } + + private static void seedBucket( + final Handle handle, final BucketKey key, final SeedAggregate aggregate) { + switch (aggregate) { + case CountValue count -> + handle + .createUpdate( + "INSERT INTO velocity_counts (" + + KEY_COLUMNS + + ", count, seeded) VALUES (" + + KEY_VALUES + + ", :count, TRUE) ON CONFLICT " + + KEY_CONFLICT + + " DO UPDATE SET count = velocity_counts.count + :count, seeded = TRUE") + .bindMap(key.binding()) + .bind("count", count.count()) + .execute(); + case SumValue sum -> + handle + .createUpdate( + "INSERT INTO velocity_sums (" + + KEY_COLUMNS + + ", sum_cents, seeded) VALUES (" + + KEY_VALUES + + ", :val, TRUE) ON CONFLICT " + + KEY_CONFLICT + + " DO UPDATE SET sum_cents = velocity_sums.sum_cents + :val, seeded = TRUE") + .bindMap(key.binding()) + .bind("val", sum.cents()) + .execute(); + case ExactDistinct exact -> { + for (final DistinctMember member : exact.members()) { + handle + .createUpdate( + "INSERT INTO velocity_distinct_members (" + + KEY_COLUMNS + + ", member, seeded) VALUES (" + + KEY_VALUES + + ", :member, TRUE) ON CONFLICT DO NOTHING") + .bindMap(key.binding()) + .bind("member", member.token()) + .execute(); + } + } + case HllDistinct hll -> + throw new IllegalArgumentException( + "velocity-backend-jdbi is exact-only distinct; rejecting HLL seed " + + hll + + " (ADR 0005/0006)"); + } + } + + private @Nullable Window matchingSupportedWindow(final Feature feature, final Duration span) { + for (final Window window : feature.windows()) { + if (window.duration().equals(span) && capabilities.supportsWindow(window)) { + return window; + } + } + return null; + } + + // ---- PURGE --------------------------------------------------------------------------------- + + @Override + public void purge(final Namespace namespace, final @Nullable Subject subject) { + Objects.requireNonNull(namespace, "namespace"); + jdbi.useTransaction( + handle -> { + for (final String table : + List.of("velocity_counts", "velocity_sums", "velocity_distinct_members")) { + if (subject == null) { + handle + .createUpdate("DELETE FROM " + table + " WHERE namespace = :ns") + .bind("ns", namespace.value()) + .execute(); + } else { + handle + .createUpdate( + "DELETE FROM " + + table + + " WHERE namespace = :ns AND subject_type = :st AND subject_value = :sv") + .bind("ns", namespace.value()) + .bind("st", subject.type()) + .bind("sv", subject.value()) + .execute(); + } + } + }); + } + + // ---- helpers ------------------------------------------------------------------------------- + + private BucketKey keyFor( + final Namespace namespace, + final Subject subject, + final Aggregation aggregation, + final Window window, + final Instant now) { + final long durationMs = window.duration().toMillis(); + final long bucketStartMs = Math.floorDiv(now.toEpochMilli(), durationMs) * durationMs; + return keyAt( + namespace, subject, aggregation, window, Instant.ofEpochMilli(bucketStartMs), durationMs); + } + + private static BucketKey keyAt( + final Namespace namespace, + final Subject subject, + final Aggregation aggregation, + final Window window, + final Instant bucketStart, + final long durationMs) { + final long bucketStartMs = bucketStart.toEpochMilli(); + return new BucketKey( + namespace.value(), + subject.type(), + subject.value(), + dimensionOf(aggregation), + window.type().name(), + durationMs, + bucketStartMs, + new WindowBounds( + Instant.ofEpochMilli(bucketStartMs), Instant.ofEpochMilli(bucketStartMs + durationMs))); + } + + /** + * COUNT/SUM have no dimension; DISTINCT carries the named dimension. Empty string keys the row. + */ + private static String dimensionOf(final Aggregation aggregation) { + final String dimension = aggregation.dimension(); + return dimension == null ? "" : dimension; + } + + private static FeatureValue featureValue( + final Feature feature, + final Window window, + final BigDecimal value, + final BucketKey key, + final Instant now) { + return new FeatureValue( + feature, + window, + value, + Exactness.EXACT, + ReadYourWriteLevel.ATOMIC, + /* definitionVersionHash= */ null, + key.bounds(), + now); + } + + private static Feature syntheticFeature(final QueryTuple tuple) { + final Aggregation aggregation = tuple.aggregation(); + final String name = + "query:" + + aggregation.type() + + (aggregation.dimension() == null ? "" : ":" + aggregation.dimension()); + return new Feature(name, aggregation, List.of(tuple.window())); + } + + private static String unsupported(final Window window) { + return "window " + + window.type() + + " " + + window.duration() + + " not supported by velocity-backend-jdbi"; + } + + /** Writes one intent into its bucket row and returns the post-write value (read-your-write). */ + @FunctionalInterface + private interface BucketWriter { + BigDecimal write(Handle handle, BucketKey key, Intent intent); + } + + /** Reads the current value of a bucket (0/empty when absent). */ + @FunctionalInterface + private interface BucketReader { + BigDecimal read(Handle handle, BucketKey key); + } + + /** + * The fully-resolved storage key of a single tumbling bucket, plus the window bounds the value + * covers. {@link #binding()} exposes the seven key columns as named JDBI parameters. + */ + private record BucketKey( + String namespace, + String subjectType, + String subjectValue, + String dimension, + String windowType, + long windowDurationMs, + long bucketStartMs, + WindowBounds bounds) { + + Map binding() { + final Map map = new HashMap<>(); + map.put("ns", namespace); + map.put("st", subjectType); + map.put("sv", subjectValue); + map.put("dim", dimension); + map.put("wt", windowType); + map.put("wd", windowDurationMs); + map.put("bs", bucketStartMs); + return map; + } + } +} diff --git a/velocity-backend-jdbi/src/main/java/com/codeheadsystems/velocity/backend/jdbi/package-info.java b/velocity-backend-jdbi/src/main/java/com/codeheadsystems/velocity/backend/jdbi/package-info.java new file mode 100644 index 0000000..8ed6c81 --- /dev/null +++ b/velocity-backend-jdbi/src/main/java/com/codeheadsystems/velocity/backend/jdbi/package-info.java @@ -0,0 +1,31 @@ +// SPDX-License-Identifier: BSD-3-Clause + +/** + * The v1 Postgres/JDBI tumbling reference backend (requirements §8, acceptance #1). + * + *

{@link com.codeheadsystems.velocity.backend.jdbi.JdbiVelocityBackend} is a real + * Postgres-backed {@link com.codeheadsystems.velocity.spi.VelocityBackend} that implements the + * {@link com.codeheadsystems.velocity.spi.CountStore COUNT}, {@link + * com.codeheadsystems.velocity.spi.SumStore SUM} and {@link + * com.codeheadsystems.velocity.spi.DistinctStore DISTINCT} aggregation mix-ins over {@link + * com.codeheadsystems.velocity.spi.TumblingSupport tumbling} windows, plus {@link + * com.codeheadsystems.velocity.spi.SeedSupport seeding}. It is exact and atomic: + * every value is {@link com.codeheadsystems.velocity.spi.model.Exactness#EXACT EXACT} and produced + * under {@link com.codeheadsystems.velocity.spi.model.ReadYourWriteLevel#ATOMIC ATOMIC} + * read-your-write, via a Postgres {@code INSERT ... ON CONFLICT ... DO UPDATE ... RETURNING} upsert + * (ADR 0007). + * + *

Tumbling only. The backend owns bucketing and is the clock authority (ADR + * 0003, FR-3): a tumbling window's value is the current aligned bucket {@code + * [floor(nowMs/durationMs)·durationMs, +durationMs)} only. It declares no sliding windows and does + * not implement {@link com.codeheadsystems.velocity.spi.SlidingSupport}. + * + *

Distinct is exact-only. Members are stored as opaque {@code BYTEA} rows and + * cardinality is a {@code COUNT(*)}; HLL degradation (ADR 0005/0006) is a documented follow-up, so + * the backend declares an effectively-unbounded exact clamp and rejects an {@code HllDistinct} + * seed. + */ +@NullMarked +package com.codeheadsystems.velocity.backend.jdbi; + +import org.jspecify.annotations.NullMarked; diff --git a/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/AbstractJdbiBackendTest.java b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/AbstractJdbiBackendTest.java new file mode 100644 index 0000000..98f390c --- /dev/null +++ b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/AbstractJdbiBackendTest.java @@ -0,0 +1,27 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.jdbi; + +import com.codeheadsystems.velocity.testkit.MutableClock; +import org.jdbi.v3.core.Jdbi; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; + +/** + * Base for the real-Postgres integration tests: a fresh {@link JdbiVelocityBackend} over the shared + * Testcontainers Postgres, driven by a controllable {@link MutableClock}, with the store truncated + * before each test. Disabled where Docker is absent via {@code VELOCITY_SKIP_TESTCONTAINERS=1}. + */ +@DisabledIfEnvironmentVariable(named = "VELOCITY_SKIP_TESTCONTAINERS", matches = "1") +abstract class AbstractJdbiBackendTest { + + protected MutableClock clock; + protected JdbiVelocityBackend backend; + + @BeforeEach + void setUpBackend() { + final Jdbi jdbi = PostgresSupport.ready(); + PostgresSupport.truncate(); + clock = MutableClock.atNow(); + backend = new JdbiVelocityBackend(jdbi, clock); + } +} diff --git a/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiAggregationTest.java b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiAggregationTest.java new file mode 100644 index 0000000..4d32ef6 --- /dev/null +++ b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiAggregationTest.java @@ -0,0 +1,416 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.jdbi; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +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.BucketValue; +import com.codeheadsystems.velocity.spi.model.DistinctMember; +import com.codeheadsystems.velocity.spi.model.Exactness; +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.FeatureValue; +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.QueryContext; +import com.codeheadsystems.velocity.spi.model.QueryTuple; +import com.codeheadsystems.velocity.spi.model.ReadYourWriteLevel; +import com.codeheadsystems.velocity.spi.model.SeedAggregate; +import com.codeheadsystems.velocity.spi.model.Subject; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowBounds; +import com.codeheadsystems.velocity.spi.model.WindowType; +import com.codeheadsystems.velocity.testkit.WindowMath; +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +/** + * Backend-specific COUNT/SUM/DISTINCT/PURGE assertions against real Postgres, over TUMBLING windows + * — the tumbling counterpart of the sliding-based {@code *StoreScenarios} the tumbling-only backend + * cannot run. Same contract facets: exact windowed values, read-your-write, namespace/subject + * isolation, seed merge, and the distinguishable-failure path for an unsupported window. + */ +final class JdbiAggregationTest extends AbstractJdbiBackendTest { + + private static final Namespace NS_A = new Namespace("jdbi-ns-a"); + private static final Namespace NS_B = new Namespace("jdbi-ns-b"); + private static final Subject SUBJECT_A = new Subject("card", "subject-a"); + private static final Subject SUBJECT_B = new Subject("card", "subject-b"); + private static final Window TUMBLING_1M = new Window(Duration.ofMinutes(1), WindowType.TUMBLING); + private static final Window TUMBLING_1H = new Window(Duration.ofHours(1), WindowType.TUMBLING); + private static final Window SLIDING_1M = new Window(Duration.ofMinutes(1), WindowType.SLIDING); + private static final String DIMENSION = "ip"; + + private static ApplyContext apply(final Namespace ns) { + return new ApplyContext(ns, null); + } + + private static QueryContext query(final Namespace ns) { + return new QueryContext(ns, null); + } + + private static DistinctMember member(final String token) { + return new DistinctMember(token.getBytes(StandardCharsets.UTF_8)); + } + + private static void assertValue(final FeatureResult result, final long expected) { + assertThat(result).isInstanceOf(FeatureResult.Success.class); + final FeatureValue value = ((FeatureResult.Success) result).value(); + assertThat(value.value()).isEqualByComparingTo(BigDecimal.valueOf(expected)); + } + + private static FeatureValue successValue(final FeatureResult result) { + assertThat(result).isInstanceOf(FeatureResult.Success.class); + return ((FeatureResult.Success) result).value(); + } + + @Nested + final class Count { + + private final Feature feature = + new Feature("jdbi.count", Aggregation.count(), List.of(TUMBLING_1M)); + + private FeatureResult read(final Namespace ns, final Subject subject) { + return backend + .queryCount(query(ns), List.of(new QueryTuple(subject, Aggregation.count(), TUMBLING_1M))) + .get(0); + } + + @Test + void applyThenQueryReturnsExactCountFlaggedExactAtomic() { + backend.applyCount( + apply(NS_A), + List.of( + new CountIntent(feature, SUBJECT_A), + new CountIntent(feature, SUBJECT_A), + new CountIntent(feature, SUBJECT_A))); + + final FeatureResult result = read(NS_A, SUBJECT_A); + assertValue(result, 3); + assertThat(successValue(result).exactness()).isEqualTo(Exactness.EXACT); + assertThat(successValue(result).readYourWriteLevel()).isEqualTo(ReadYourWriteLevel.ATOMIC); + assertThat(successValue(result).asOf()).isEqualTo(clock.instant()); + } + + @Test + void applyResultReflectsWriteReadYourWrite() { + final ApplyResult first = + backend.applyCount(apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + assertThat(first.perFeature().get(0).status()).isEqualTo(ApplyStatus.APPLIED); + assertValue(first.perFeature().get(0).result(), 1); + + final ApplyResult second = + backend.applyCount(apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + assertValue(second.perFeature().get(0).result(), 2); + } + + @Test + void multiWindowFeatureEmitsOneAppliedPerWindow() { + final Feature multi = + new Feature("jdbi.count.multi", Aggregation.count(), List.of(TUMBLING_1M, TUMBLING_1H)); + final ApplyResult result = + backend.applyCount(apply(NS_A), List.of(new CountIntent(multi, SUBJECT_A))); + + assertThat(result.perFeature()).hasSize(2); + assertThat(result.perFeature()) + .allSatisfy(pf -> assertThat(pf.status()).isEqualTo(ApplyStatus.APPLIED)); + assertThat( + result.perFeature().stream().map(pf -> successValue(pf.result()).window()).toList()) + .containsExactlyInAnyOrder(TUMBLING_1M, TUMBLING_1H); + assertThat(result.perFeature()).allSatisfy(pf -> assertValue(pf.result(), 1)); + } + + @Test + void valuesIsolatedByNamespace() { + backend.applyCount( + apply(NS_A), + List.of(new CountIntent(feature, SUBJECT_A), new CountIntent(feature, SUBJECT_A))); + assertValue(read(NS_A, SUBJECT_A), 2); + assertValue(read(NS_B, SUBJECT_A), 0); + } + + @Test + void valuesIsolatedBySubject() { + backend.applyCount(apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + assertValue(read(NS_A, SUBJECT_A), 1); + assertValue(read(NS_A, SUBJECT_B), 0); + } + + @Test + void unsupportedWindowApplyIsDistinguishableFailureNotSilentZero() { + final Feature sliding = + new Feature("jdbi.count.sliding", Aggregation.count(), List.of(SLIDING_1M)); + final ApplyResult result = + backend.applyCount(apply(NS_A), List.of(new CountIntent(sliding, SUBJECT_A))); + assertThat(result.perFeature()).hasSize(1); + assertThat(result.perFeature().get(0).status()).isEqualTo(ApplyStatus.FAILED); + assertThat(result.perFeature().get(0).result()).isInstanceOf(FeatureResult.Failure.class); + assertThat(((FeatureResult.Failure) result.perFeature().get(0).result()).code()) + .isEqualTo(FailureCode.UNSUPPORTED_WINDOW); + } + + @Test + void mixedFeatureAppliesSupportedWindowAndFailsUnsupportedInSameCall() { + final Feature mixed = + new Feature("jdbi.count.mixed", Aggregation.count(), List.of(TUMBLING_1M, SLIDING_1M)); + final ApplyResult result = + backend.applyCount(apply(NS_A), List.of(new CountIntent(mixed, SUBJECT_A))); + assertThat(result.perFeature()).hasSize(2); + assertThat(result.perFeature().stream().map(pf -> pf.status()).toList()) + .containsExactlyInAnyOrder(ApplyStatus.APPLIED, ApplyStatus.FAILED); + // The supported window still recorded exactly one event. + assertValue(read(NS_A, SUBJECT_A), 1); + } + } + + @Nested + final class Sum { + + private final Feature feature = + new Feature("jdbi.sum", Aggregation.sum(), List.of(TUMBLING_1M)); + + private FeatureValue read() { + return successValue( + backend + .querySum( + query(NS_A), List.of(new QueryTuple(SUBJECT_A, Aggregation.sum(), TUMBLING_1M))) + .get(0)); + } + + @Test + void applyThenQueryReturnsExactSumCentsScaleZero() { + backend.applySum( + apply(NS_A), + List.of( + new SumIntent(feature, SUBJECT_A, BigDecimal.valueOf(150)), + new SumIntent(feature, SUBJECT_A, BigDecimal.valueOf(250)))); + final FeatureValue value = read(); + assertThat(value.value()).isEqualByComparingTo(BigDecimal.valueOf(400)); + assertThat(value.value().scale()).isZero(); + assertThat(value.asOf()).isEqualTo(clock.instant()); + } + + @Test + void applyResultReflectsRunningSum() { + final ApplyResult first = + backend.applySum( + apply(NS_A), List.of(new SumIntent(feature, SUBJECT_A, BigDecimal.valueOf(500)))); + assertThat(successValue(first.perFeature().get(0).result()).value()) + .isEqualByComparingTo(BigDecimal.valueOf(500)); + final ApplyResult second = + backend.applySum( + apply(NS_A), List.of(new SumIntent(feature, SUBJECT_A, BigDecimal.valueOf(125)))); + assertThat(successValue(second.perFeature().get(0).result()).value()) + .isEqualByComparingTo(BigDecimal.valueOf(625)); + } + + @Test + void bigDecimalCentsPreservedWithoutOverflow() { + final BigDecimal big = new BigDecimal("9000000000000000000"); + backend.applySum( + apply(NS_A), + List.of(new SumIntent(feature, SUBJECT_A, big), new SumIntent(feature, SUBJECT_A, big))); + assertThat(read().value()).isEqualByComparingTo(big.add(big)); + } + + @Test + void negativeValuesForRefunds() { + backend.applySum( + apply(NS_A), + List.of( + new SumIntent(feature, SUBJECT_A, BigDecimal.valueOf(500)), + new SumIntent(feature, SUBJECT_A, BigDecimal.valueOf(-200)))); + assertThat(read().value()).isEqualByComparingTo(BigDecimal.valueOf(300)); + } + + @Test + void emptyWindowIsKnownZero() { + assertThat(read().value()).isEqualByComparingTo(BigDecimal.ZERO); + } + } + + @Nested + final class Distinct { + + private final Feature feature = + new Feature("jdbi.distinct", Aggregation.distinct(DIMENSION), List.of(TUMBLING_1M)); + + private FeatureResult read(final Subject subject) { + return backend + .queryDistinct( + query(NS_A), + List.of(new QueryTuple(subject, Aggregation.distinct(DIMENSION), TUMBLING_1M))) + .get(0); + } + + @Test + void applyThenQueryReturnsCardinalityFlaggedExact() { + backend.applyDistinct( + apply(NS_A), + List.of( + new DistinctIntent(feature, SUBJECT_A, member("alice")), + new DistinctIntent(feature, SUBJECT_A, member("bob")), + new DistinctIntent(feature, SUBJECT_A, member("carol")))); + final FeatureResult result = read(SUBJECT_A); + assertValue(result, 3); + assertThat(successValue(result).exactness()).isEqualTo(Exactness.EXACT); + } + + @Test + void deDupesRepeatedMembers() { + backend.applyDistinct( + apply(NS_A), + List.of( + new DistinctIntent(feature, SUBJECT_A, member("alice")), + new DistinctIntent(feature, SUBJECT_A, member("alice")), + new DistinctIntent(feature, SUBJECT_A, member("bob")))); + assertValue(read(SUBJECT_A), 2); + } + + @Test + void applyResultReflectsCardinalityReadYourWrite() { + final ApplyResult first = + backend.applyDistinct( + apply(NS_A), List.of(new DistinctIntent(feature, SUBJECT_A, member("alice")))); + assertValue(first.perFeature().get(0).result(), 1); + final ApplyResult second = + backend.applyDistinct( + apply(NS_A), List.of(new DistinctIntent(feature, SUBJECT_A, member("bob")))); + assertValue(second.perFeature().get(0).result(), 2); + } + + @Test + void valuesIsolatedBySubject() { + backend.applyDistinct( + apply(NS_A), List.of(new DistinctIntent(feature, SUBJECT_A, member("alice")))); + assertValue(read(SUBJECT_B), 0); + } + } + + @Nested + final class Purge { + + private final Feature feature = + new Feature("jdbi.purge", Aggregation.count(), List.of(TUMBLING_1M)); + + private FeatureResult count(final Namespace ns, final Subject subject) { + return backend + .queryCount(query(ns), List.of(new QueryTuple(subject, Aggregation.count(), TUMBLING_1M))) + .get(0); + } + + @Test + void purgeSubjectClearsThatSubjectOnly() { + backend.applyCount(apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + backend.applyCount(apply(NS_A), List.of(new CountIntent(feature, SUBJECT_B))); + + backend.purge(NS_A, SUBJECT_A); + + assertValue(count(NS_A, SUBJECT_A), 0); + assertValue(count(NS_A, SUBJECT_B), 1); + } + + @Test + void purgeNamespaceClearsEntireNamespace() { + backend.applyCount(apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + backend.applyCount(apply(NS_A), List.of(new CountIntent(feature, SUBJECT_B))); + backend.applyCount(apply(NS_B), List.of(new CountIntent(feature, SUBJECT_A))); + + backend.purge(NS_A, null); + + assertValue(count(NS_A, SUBJECT_A), 0); + assertValue(count(NS_A, SUBJECT_B), 0); + assertValue(count(NS_B, SUBJECT_A), 1); + } + } + + @Nested + final class Seed { + + @Test + void seededSumBucketMergesWithRecorded() { + final Feature feature = new Feature("jdbi.seed.sum", Aggregation.sum(), List.of(TUMBLING_1M)); + final WindowBounds bucket = WindowMath.tumblingBucket(clock.instant(), 60_000); + backend.seed( + NS_A, + SUBJECT_A, + feature, + List.of(new BucketValue(bucket, new SeedAggregate.SumValue(BigDecimal.valueOf(1000))))); + backend.applySum( + apply(NS_A), List.of(new SumIntent(feature, SUBJECT_A, BigDecimal.valueOf(250)))); + + final FeatureValue value = + successValue( + backend + .querySum( + query(NS_A), + List.of(new QueryTuple(SUBJECT_A, Aggregation.sum(), TUMBLING_1M))) + .get(0)); + assertThat(value.value()).isEqualByComparingTo(BigDecimal.valueOf(1250)); + } + + @Test + void seededExactDistinctBucketMergesWithRecorded() { + final Feature feature = + new Feature("jdbi.seed.distinct", Aggregation.distinct(DIMENSION), List.of(TUMBLING_1M)); + final WindowBounds bucket = WindowMath.tumblingBucket(clock.instant(), 60_000); + backend.seed( + NS_A, + SUBJECT_A, + feature, + List.of( + new BucketValue( + bucket, + new SeedAggregate.ExactDistinct(List.of(member("alice"), member("bob")))))); + // Recording an overlapping member ("bob") plus a new one ("carol") de-dupes against the seed. + backend.applyDistinct( + apply(NS_A), + List.of( + new DistinctIntent(feature, SUBJECT_A, member("bob")), + new DistinctIntent(feature, SUBJECT_A, member("carol")))); + + assertValue( + backend + .queryDistinct( + query(NS_A), + List.of(new QueryTuple(SUBJECT_A, Aggregation.distinct(DIMENSION), TUMBLING_1M))) + .get(0), + 3); + } + + @Test + void seedRejectsUnsupportedAggregationWindowSpan() { + final Feature feature = + new Feature("jdbi.seed.bad", Aggregation.count(), List.of(TUMBLING_1M)); + final WindowBounds start = WindowMath.tumblingBucket(clock.instant(), 60_000); + final WindowBounds badBucket = new WindowBounds(start.start(), start.start().plusSeconds(90)); + assertThatThrownBy( + () -> + backend.seed( + NS_A, + SUBJECT_A, + feature, + List.of(new BucketValue(badBucket, new SeedAggregate.CountValue(1))))) + .isInstanceOf(IllegalArgumentException.class); + } + } + + @Test + void dataSourceConstructorProducesUsableBackend() { + final JdbiVelocityBackend viaDs = + new JdbiVelocityBackend(PostgresSupport.dataSource()).migrate(); + assertThat(viaDs.capabilities().seedSupported()).isTrue(); + assertThat(viaDs.clock()).isNotNull(); + } +} diff --git a/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiConcurrencyTest.java b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiConcurrencyTest.java new file mode 100644 index 0000000..7c02881 --- /dev/null +++ b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiConcurrencyTest.java @@ -0,0 +1,91 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.jdbi; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.ApplyContext; +import com.codeheadsystems.velocity.spi.model.Feature; +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.Subject; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowType; +import java.math.BigDecimal; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +/** + * Acceptance #1: the Postgres upsert is atomic under concurrent load. {@code N} threads each apply + * the same COUNT feature {@code M} times against real Postgres; the final bucket count must be + * exactly {@code N*M}, never lost to a read-modify-write race (NFR-6). This is what the {@code + * INSERT ... ON CONFLICT DO UPDATE SET count = count + 1} row-locking upsert buys. + */ +final class JdbiConcurrencyTest extends AbstractJdbiBackendTest { + + private static final Namespace NS = new Namespace("jdbi-concurrency"); + private static final Subject SUBJECT = new Subject("card", "hot-subject"); + private static final Window TUMBLING_1H = new Window(Duration.ofHours(1), WindowType.TUMBLING); + + @Test + void concurrentApplyToSameBucketIsExactlyAtomic() throws Exception { + // A 1-hour tumbling bucket so every apply lands in the SAME bucket for the whole test run. + final Feature feature = + new Feature("jdbi.concurrent.count", Aggregation.count(), List.of(TUMBLING_1H)); + final int threads = 16; + final int perThread = 50; + final int expected = threads * perThread; + + final CountDownLatch ready = new CountDownLatch(threads); + final CountDownLatch fire = new CountDownLatch(1); + try (ExecutorService pool = Executors.newFixedThreadPool(threads)) { + final List> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + futures.add( + pool.submit( + () -> { + ready.countDown(); + await(fire); + for (int i = 0; i < perThread; i++) { + backend.applyCount( + new ApplyContext(NS, null), List.of(new CountIntent(feature, SUBJECT))); + } + })); + } + assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue(); + fire.countDown(); + for (final Future future : futures) { + future.get(30, TimeUnit.SECONDS); + } + } + + final FeatureResult result = + backend + .queryCount( + new QueryContext(NS, null), + List.of(new QueryTuple(SUBJECT, Aggregation.count(), TUMBLING_1H))) + .get(0); + assertThat(result).isInstanceOf(FeatureResult.Success.class); + assertThat(((FeatureResult.Success) result).value().value()) + .isEqualByComparingTo(BigDecimal.valueOf(expected)); + } + + 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-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiConformanceTckTest.java b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiConformanceTckTest.java new file mode 100644 index 0000000..2c582d4 --- /dev/null +++ b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/JdbiConformanceTckTest.java @@ -0,0 +1,89 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.jdbi; + +import com.codeheadsystems.velocity.testkit.tck.CapabilityConformanceScenarios; +import com.codeheadsystems.velocity.testkit.tck.SeedSupportScenarios; +import com.codeheadsystems.velocity.testkit.tck.TumblingScenarios; +import org.junit.jupiter.api.Test; + +/** + * Drives the {@code velocity-testkit} conformance TCK against real Postgres for the scenarios that + * apply to a tumbling-only backend (ADR 0004): {@link TumblingScenarios}, {@link + * SeedSupportScenarios} and {@link CapabilityConformanceScenarios}. This proves the JDBI backend + * honors the same frozen contract as the reference in-memory backend. + * + *

{@code SlidingScenarios} is intentionally NOT run — this backend declares no sliding windows. + * The aggregation suites ({@code Count/Sum/Distinct/PurgeStoreScenarios}) hardcode a sliding window + * in their fixtures, so they cannot run against a tumbling-only backend; the equivalent tumbling + * assertions live in {@link JdbiAggregationTest}. + */ +final class JdbiConformanceTckTest extends AbstractJdbiBackendTest { + + private TumblingScenarios tumbling() { + return new TumblingScenarios(backend, clock); + } + + private SeedSupportScenarios seed() { + return new SeedSupportScenarios<>(backend, clock); + } + + private CapabilityConformanceScenarios capability() { + return new CapabilityConformanceScenarios(backend); + } + + // ---- TumblingScenarios ---- + + @Test + void tumblingAccumulatesThenResetsAtBoundary() { + tumbling().tumblingAccumulatesThenResetsAtBoundary(); + } + + // ---- SeedSupportScenarios (acceptance #16) ---- + + @Test + void seededBucketMergesIdenticallyWithRecorded() { + seed().seededBucketMergesIdenticallyWithRecorded(); + } + + @Test + void hllDistinctSeedRejected() { + seed().hllDistinctSeedRejected(); + } + + @Test + void seedWithUnsupportedWindowRejected() { + seed().seedWithUnsupportedWindowRejected(); + } + + // ---- CapabilityConformanceScenarios (ADR 0004) ---- + + @Test + void declaredAggregationsMatchImplementedMixins() { + capability().declaredAggregationsMatchImplementedMixins(); + } + + @Test + void declaredWindowMarkersMatchWindowSpecs() { + capability().declaredWindowMarkersMatchWindowSpecs(); + } + + @Test + void seedSupportFlagMatchesMixin() { + capability().seedSupportFlagMatchesMixin(); + } + + @Test + void distinctHllSlidingIsFalse() { + capability().distinctHllSlidingIsFalse(); + } + + @Test + void unsupportedWindowQueryIsDistinguishableFailure() { + capability().unsupportedWindowQueryIsDistinguishableFailure(); + } + + @Test + void supportedEmptyWindowReturnsKnownZero() { + capability().supportedEmptyWindowReturnsKnownZero(); + } +} diff --git a/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/PostgresSupport.java b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/PostgresSupport.java new file mode 100644 index 0000000..4713cdc --- /dev/null +++ b/velocity-backend-jdbi/src/test/java/com/codeheadsystems/velocity/backend/jdbi/PostgresSupport.java @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.jdbi; + +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; +import java.time.Clock; +import javax.sql.DataSource; +import org.jdbi.v3.core.Jdbi; +import org.testcontainers.containers.PostgreSQLContainer; +import org.testcontainers.utility.DockerImageName; + +/** + * Shared Testcontainers Postgres fixture — one container per JVM, reused across every integration + * test class. Builds a HikariCP pool over the container's JDBC URL, a {@link Jdbi} over that pool, + * and migrates the {@link JdbiVelocityBackend} schema once. {@link #truncate()} resets state + * between tests so each starts from a clean store. + */ +final class PostgresSupport { + + private static final PostgreSQLContainer CONTAINER = + new PostgreSQLContainer<>(DockerImageName.parse("postgres:16-alpine")) + .withDatabaseName("velocity") + .withUsername("velocity") + .withPassword("velocity-test") + .withReuse(true); + + private static HikariDataSource dataSource; + private static Jdbi jdbi; + + private PostgresSupport() {} + + /** Lazily starts the container, builds the pool + Jdbi, and migrates the schema once. */ + static synchronized Jdbi ready() { + if (jdbi != null) { + return jdbi; + } + if (!CONTAINER.isRunning()) { + CONTAINER.start(); + } + final HikariConfig cfg = new HikariConfig(); + cfg.setJdbcUrl(CONTAINER.getJdbcUrl()); + cfg.setUsername(CONTAINER.getUsername()); + cfg.setPassword(CONTAINER.getPassword()); + cfg.setMaximumPoolSize(8); + dataSource = new HikariDataSource(cfg); + jdbi = Jdbi.create(dataSource); + // Migrate through the backend's own idempotent DDL path (Clock is irrelevant to migration). + new JdbiVelocityBackend(jdbi, Clock.systemUTC()).migrate(); + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + if (dataSource != null) { + dataSource.close(); + } + })); + return jdbi; + } + + /** The pooled data source, exposed so a test can drive the {@link DataSource} constructor. */ + static DataSource dataSource() { + ready(); + return dataSource; + } + + /** Truncates every velocity table so a test starts from an empty store. */ + static void truncate() { + ready(); + jdbi.useHandle( + handle -> + handle.execute( + "TRUNCATE TABLE velocity_counts, velocity_sums, velocity_distinct_members")); + } +}