diff --git a/.gitignore b/.gitignore index 0789c14..123ffdc 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ target/ logs/ *.sqlite lib/ +dependency-reduced-pom.xml diff --git a/Makefile b/Makefile index e9098ff..20e1096 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ .PHONY: clean build format lint run-dev run-dev-pitch run-rti test verify clean-vendor refresh run-debug-portico -APP_JAR := target/hla-xapi-1.0-SNAPSHOT-jar-with-dependencies.jar +APP_JAR := target/hla-xapi-1.0-SNAPSHOT.jar PORTICO_REPO_URL ?= https://github.com/yetanalytics/portico.git PORTICO_REF ?= yet_patch_object_subs PORTICO_JAR ?= lib/maven-repository/org/porticoproject/portico/3.0.0-local/portico-3.0.0-local.jar diff --git a/README.md b/README.md index e300fb6..62719cd 100644 --- a/README.md +++ b/README.md @@ -58,8 +58,13 @@ Use Maven's `verify` lifecycle before opening a PR: make verify ``` -This runs unit tests, builds the package, and fails if linting or formatting checks do not pass. To run only -linting locally: +This runs unit tests, builds the package, and fails if linting or formatting checks do not pass. Note that the tests require docker to run. + +The PostgreSQL object-cache contract tests use Testcontainers and require a working Docker-compatible container +runtime. The test suite starts and removes its own PostgreSQL 17 container; no manually configured database is +needed. + +To run only linting locally: ```shell make lint diff --git a/doc/xapi-config.md b/doc/xapi-config.md index d5fd343..d56f0b7 100644 --- a/doc/xapi-config.md +++ b/doc/xapi-config.md @@ -226,7 +226,7 @@ The alias must exist in the trigger's `lookups` map. Each lookup alias is resolv ## Object Cache -The object cache stores the latest reflected values for subscribed HLA object attributes in SQLite. It is enabled when either: +The object cache stores the latest reflected values for subscribed HLA object attributes in SQLite or PostgreSQL. It is enabled when either: - a statement template contains a `query` injection, - a trigger defines `lookups` or uses `lookup` injections that reference cached object attributes, or @@ -252,16 +252,48 @@ Tracked object fields: - `attributes`: Top-level attribute names to subscribe to. - `allAttributes`: When `true`, expands to all top-level attributes for the class. +`HLA_OBJECT_CACHE_BACKEND` selects `sqlite` or `postgresql` case-insensitively. It defaults to `sqlite`. +Backend and connection settings are runtime configuration and cannot be set in the xAPI JSON file. + The cache decodes reflected values using the FOM and stores both top-level values and flattened nested values for fixed records and arrays. For example, reflecting `Position` can make `Position`, `Position.X`, and `Position.Y` available to query and lookup targets. -By default the SQLite database is `hla-object-cache.sqlite` in the working directory. It can be changed with: +### SQLite + +By default SQLite uses `hla-object-cache.sqlite` in the working directory. It can be changed with: ```shell HLA_OBJECT_CACHE_DB=/path/to/cache.sqlite make run-dev HLA_OBJECT_CACHE_JDBC_URL=jdbc:sqlite:/path/to/cache.sqlite make run-dev ``` -The cache starts fresh on initialization: the current schema drops and recreates object and FOM metadata tables when the cache opens. +### PostgreSQL + +For local development, start the included PostgreSQL 17 service: + +```shell +docker compose up -d --wait postgres +``` + +The service listens on `localhost:5432` and uses a named volume with database `hla_xapi`, username `hla_xapi`, and +password `hla_xapi_dev`. Stop it with `docker compose down`. To also delete its data and start with an empty database, +run `docker compose down -v`. + +Then provide the connection settings at runtime: + +```shell +HLA_OBJECT_CACHE_BACKEND=postgresql \ +HLA_OBJECT_CACHE_JDBC_URL=jdbc:postgresql://localhost:5432/hla_xapi \ +HLA_OBJECT_CACHE_USERNAME=hla_xapi \ +HLA_OBJECT_CACHE_PASSWORD=hla_xapi_dev \ +HLA_OBJECT_CACHE_SCHEMA=hla_object_cache \ +make run-dev +``` + +`HLA_OBJECT_CACHE_BACKEND=postgresql` selects PostgreSQL, and `HLA_OBJECT_CACHE_JDBC_URL` is then required. Username and password are optional when authentication is already present in the JDBC URL or supplied by the driver, but they must be supplied together through these variables when used. The schema defaults to `hla_object_cache` and must be a simple unquoted SQL identifier. + +The PostgreSQL account must be able to create and use the configured schema and create, drop, read, and write the cache tables. Assign a separate schema to each running adapter process; concurrent writers must not share one cache schema. + +Both backends start fresh on initialization. The cache drops and recreates only its five owned tables, then seeds the current FOM metadata. PostgreSQL does not drop the configured schema or any unrelated tables in it. ## LRS Configuration diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..351723b --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,19 @@ +services: + postgres: + image: postgres:17 + environment: + POSTGRES_DB: hla_xapi + POSTGRES_USER: hla_xapi + POSTGRES_PASSWORD: hla_xapi_dev + ports: + - "5432:5432" + volumes: + - postgres-data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U hla_xapi -d hla_xapi"] + interval: 5s + timeout: 5s + retries: 10 + +volumes: + postgres-data: diff --git a/pom.xml b/pom.xml index 878fd67..fc1830e 100644 --- a/pom.xml +++ b/pom.xml @@ -13,6 +13,7 @@ UTF-8 21 10.21.1 + 2.0.5 @@ -105,6 +106,23 @@ sqlite-jdbc 3.42.0.0 + + org.postgresql + postgresql + 42.7.13 + + + org.testcontainers + testcontainers-postgresql + ${testcontainers.version} + test + + + org.testcontainers + testcontainers-junit-jupiter + ${testcontainers.version} + test + org.springframework @@ -162,6 +180,11 @@ spotless-maven-plugin 2.43.0 + + org.apache.maven.plugins + maven-shade-plugin + 3.6.2 + org.apache.maven.plugins maven-checkstyle-plugin @@ -236,25 +259,34 @@ org.apache.maven.plugins - maven-assembly-plugin - 3.7.1 - - - - com.yetanalytics.hlaxapi.App - - - - jar-with-dependencies - - + maven-shade-plugin - make-assembly package - single + shade + + + + + + + com.yetanalytics.hlaxapi.App + + + + + + *:* + + META-INF/*.SF + META-INF/*.DSA + META-INF/*.RSA + + + + diff --git a/src/main/java/com/yetanalytics/hlaxapi/cache/JdbcObjectCacheStore.java b/src/main/java/com/yetanalytics/hlaxapi/cache/JdbcObjectCacheStore.java new file mode 100644 index 0000000..45d9304 --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/cache/JdbcObjectCacheStore.java @@ -0,0 +1,373 @@ +package com.yetanalytics.hlaxapi.cache; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.sql.Types; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.Optional; + +final class JdbcObjectCacheStore implements ObjectCacheStore { + + private static final int SCHEMA_VERSION = 1; + + private final ObjectCacheQueries queries; + private final ObjectMapper mapper = new ObjectMapper(); + private Connection connection; + + JdbcObjectCacheStore(Connection connection, ObjectCacheQueries queries, FomCatalog catalog) throws SQLException { + this.connection = Objects.requireNonNull(connection, "connection"); + this.queries = Objects.requireNonNull(queries, "queries"); + initialize(Objects.requireNonNull(catalog, "catalog")); + } + + @Override + public boolean isOpen() { + return connection != null; + } + + @Override + public CachedObject ensureObject( + String objectHandle, + String objectName, + FomCatalog.ObjectClassDef clazz) { + try (PreparedStatement statement = connection.prepareStatement(queries.upsertObject())) { + statement.setString(1, Objects.requireNonNull(objectHandle, "objectHandle")); + statement.setString(2, objectName); + statement.setInt(3, clazz.id()); + statement.setString(4, java.time.Instant.now().toString()); + statement.executeUpdate(); + return loadObject(objectHandle, clazz.localName()); + } catch (SQLException e) { + throw new IllegalStateException("Could not upsert object instance " + objectHandle, e); + } + } + + @Override + public void removeObject(String objectHandle, String removedAt) { + try (PreparedStatement statement = connection.prepareStatement(queries.removeObject())) { + statement.setString(1, removedAt); + statement.setString(2, objectHandle); + statement.executeUpdate(); + } catch (SQLException e) { + throw new IllegalStateException("Could not mark object removed: " + objectHandle, e); + } + } + + @Override + public Optional findCurrentValue(long instanceId, String pathKey) { + String normalizedPath = FomCatalog.wildcardArrayIndexes(pathKey); + try (PreparedStatement statement = connection.prepareStatement(queries.findCurrentValue())) { + statement.setLong(1, instanceId); + statement.setString(2, pathKey); + statement.setString(3, normalizedPath); + statement.setString(4, pathKey); + try (ResultSet resultSet = statement.executeQuery()) { + if (!resultSet.next()) { + return Optional.empty(); + } + return Optional.of(readCachedValue(resultSet)); + } + } catch (SQLException e) { + throw new IllegalStateException("Could not read current cached value", e); + } + } + + @Override + public Optional findCurrentValue(String objectHandle, String pathKey) { + try (PreparedStatement statement = connection.prepareStatement(queries.findObjectId())) { + statement.setString(1, objectHandle); + try (ResultSet resultSet = statement.executeQuery()) { + if (!resultSet.next()) { + return Optional.empty(); + } + return findCurrentValue(resultSet.getLong("id"), pathKey); + } + } catch (SQLException e) { + throw new IllegalStateException("Could not read object handle: " + objectHandle, e); + } + } + + @Override + public List currentObjects(FomCatalog.ObjectClassDef clazz) { + List objects = new ArrayList<>(); + try (PreparedStatement statement = connection.prepareStatement(queries.listCurrentObjects())) { + statement.setInt(1, clazz.id()); + try (ResultSet resultSet = statement.executeQuery()) { + while (resultSet.next()) { + objects.add(new CachedObject( + resultSet.getLong("id"), + resultSet.getString("object_handle"), + resultSet.getString("object_name"), + clazz.localName())); + } + } + } catch (SQLException e) { + throw new IllegalStateException( + "Could not list cached objects for class " + clazz.localName(), e); + } + return objects; + } + + @Override + public void replaceCurrentValues( + long instanceId, + FomCatalog.ObjectClassDef clazz, + String attributeName, + List values, + String observedAt, + long observedSequence) { + boolean autoCommit = currentAutoCommit(); + try { + connection.setAutoCommit(false); + deleteCurrentValues(instanceId, clazz.id(), attributeName); + for (DecodedAttributeValue value : values) { + Optional attributeId = attributeIdForPath(clazz, value.pathKey()); + if (attributeId.isPresent()) { + upsertCurrentValue( + instanceId, + attributeId.orElseThrow(), + value, + observedAt, + observedSequence); + } + } + connection.commit(); + } catch (SQLException | RuntimeException e) { + rollbackAfterReplacementFailure(e); + throw new IllegalStateException( + "Could not replace current object attribute " + attributeName, e); + } finally { + restoreAutoCommit(autoCommit); + } + } + + @Override + public Connection connection() { + return connection; + } + + @Override + public void close() { + if (connection == null) { + return; + } + try { + connection.close(); + connection = null; + } catch (SQLException e) { + throw new IllegalStateException("Could not close HLA object cache", e); + } + } + + private void initialize(FomCatalog catalog) throws SQLException { + executeStatements(queries.connectionSetupStatements()); + boolean autoCommit = connection.getAutoCommit(); + connection.setAutoCommit(false); + try { + executeStatements(queries.resetSchemaStatements()); + executeStatements(queries.createSchemaStatements()); + insertSchemaVersion(); + insertClasses(catalog); + insertAttributes(catalog); + if (queries.afterSeedFomMetadata().isPresent()) { + executeStatements(List.of(queries.afterSeedFomMetadata().orElseThrow())); + } + connection.commit(); + } catch (SQLException e) { + connection.rollback(); + throw e; + } finally { + connection.setAutoCommit(autoCommit); + } + } + + private void executeStatements(List statements) throws SQLException { + try (Statement statement = connection.createStatement()) { + for (String sql : statements) { + statement.execute(sql); + } + } + } + + private void insertSchemaVersion() throws SQLException { + try (PreparedStatement statement = connection.prepareStatement(queries.insertSchemaVersion())) { + statement.setInt(1, SCHEMA_VERSION); + statement.executeUpdate(); + } + } + + private void insertClasses(FomCatalog catalog) throws SQLException { + try (PreparedStatement statement = connection.prepareStatement(queries.insertClass())) { + for (FomCatalog.ObjectClassDef clazz : catalog.objectClasses()) { + statement.setInt(1, clazz.id()); + statement.setString(2, clazz.hlaName()); + statement.setString(3, clazz.localName()); + statement.setString(4, clazz.parentName()); + statement.addBatch(); + } + statement.executeBatch(); + } + } + + private void insertAttributes(FomCatalog catalog) throws SQLException { + try (PreparedStatement statement = connection.prepareStatement(queries.insertAttribute())) { + for (FomCatalog.ObjectClassDef clazz : catalog.objectClasses()) { + for (FomCatalog.FomAttribute attribute : clazz.attributes()) { + statement.setInt(1, attribute.id()); + statement.setInt(2, attribute.classId()); + statement.setString(3, attribute.attributeName()); + statement.setString(4, attribute.pathKey()); + statement.setString(5, attribute.dataType()); + statement.setString(6, attribute.primitiveType()); + statement.setInt(7, attribute.leaf() ? 1 : 0); + statement.addBatch(); + } + } + statement.executeBatch(); + } + } + + private CachedObject loadObject(String objectHandle, String className) throws SQLException { + try (PreparedStatement statement = connection.prepareStatement(queries.loadObject())) { + statement.setString(1, objectHandle); + try (ResultSet resultSet = statement.executeQuery()) { + if (!resultSet.next()) { + throw new SQLException("Object instance was not inserted: " + objectHandle); + } + return new CachedObject( + resultSet.getLong("id"), + objectHandle, + resultSet.getString("object_name"), + className); + } + } + } + + private void deleteCurrentValues(long instanceId, int classId, String attributeName) throws SQLException { + try (PreparedStatement statement = connection.prepareStatement(queries.deleteCurrentValues())) { + statement.setLong(1, instanceId); + statement.setInt(2, classId); + statement.setString(3, attributeName); + statement.executeUpdate(); + } + } + + private void upsertCurrentValue( + long instanceId, + int attributeId, + DecodedAttributeValue value, + String observedAt, + long observedSequence) throws SQLException { + try (PreparedStatement statement = connection.prepareStatement(queries.upsertCurrentValue())) { + Object objectValue = value.value(); + String valueType = CachedValue.valueType(objectValue); + statement.setLong(1, instanceId); + statement.setInt(2, attributeId); + statement.setString(3, valueType); + if (objectValue instanceof byte[] bytes) { + statement.setBytes(4, bytes); + statement.setNull(5, Types.VARCHAR); + } else { + statement.setNull(4, queries.binaryJdbcType()); + statement.setString(5, serializeJson(objectValue)); + } + statement.setBytes(6, value.rawBytes()); + statement.setString(7, observedAt); + statement.setLong(8, observedSequence); + statement.executeUpdate(); + } + } + + private Optional attributeIdForPath(FomCatalog.ObjectClassDef clazz, String pathKey) throws SQLException { + Optional existingId = attributeIdFromDatabase(clazz.id(), pathKey); + if (existingId.isPresent()) { + return existingId; + } + + String wildcardPath = FomCatalog.wildcardArrayIndexes(pathKey); + FomCatalog.FomAttribute template = clazz.attribute(wildcardPath).orElse(null); + if (template == null || wildcardPath.equals(pathKey)) { + return Optional.empty(); + } + + try (PreparedStatement statement = connection.prepareStatement(queries.insertDynamicAttribute())) { + statement.setInt(1, clazz.id()); + statement.setString(2, template.attributeName()); + statement.setString(3, pathKey); + statement.setString(4, template.dataType()); + statement.setString(5, template.primitiveType()); + statement.setInt(6, template.leaf() ? 1 : 0); + statement.executeUpdate(); + } + return attributeIdFromDatabase(clazz.id(), pathKey); + } + + private Optional attributeIdFromDatabase(int classId, String pathKey) throws SQLException { + try (PreparedStatement statement = connection.prepareStatement(queries.findAttributeId())) { + statement.setInt(1, classId); + statement.setString(2, pathKey); + try (ResultSet resultSet = statement.executeQuery()) { + if (resultSet.next()) { + return Optional.of(resultSet.getInt("id")); + } + } + } + return Optional.empty(); + } + + private void rollbackAfterReplacementFailure(Exception failure) { + try { + connection.rollback(); + } catch (SQLException rollbackError) { + failure.addSuppressed(rollbackError); + } + } + + private boolean currentAutoCommit() { + try { + return connection.getAutoCommit(); + } catch (SQLException e) { + throw new IllegalStateException("Could not read object cache auto-commit", e); + } + } + + private void restoreAutoCommit(boolean autoCommit) { + try { + connection.setAutoCommit(autoCommit); + } catch (SQLException e) { + throw new IllegalStateException("Could not restore object cache auto-commit", e); + } + } + + private String serializeJson(Object value) throws SQLException { + try { + return mapper.writeValueAsString(value); + } catch (JsonProcessingException e) { + throw new SQLException("Could not serialize cached value as JSON", e); + } + } + + private CachedValue readCachedValue(ResultSet resultSet) throws SQLException { + String valueType = resultSet.getString("value_type"); + byte[] rawBytes = resultSet.getBytes("raw_bytes"); + if ("blob".equals(valueType)) { + return new CachedValue(valueType, resultSet.getBytes("value_blob"), rawBytes); + } + String valueJson = resultSet.getString("value_json"); + if (valueJson == null) { + return new CachedValue(valueType, null, rawBytes); + } + try { + return new CachedValue(valueType, mapper.readValue(valueJson, Object.class), rawBytes); + } catch (JsonProcessingException e) { + throw new SQLException("Could not deserialize cached value JSON", e); + } + } +} diff --git a/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCache.java b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCache.java index 8746437..57b622f 100644 --- a/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCache.java +++ b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCache.java @@ -1,7 +1,5 @@ package com.yetanalytics.hlaxapi.cache; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; import com.yetanalytics.hlaxapi.FOMXML; import com.yetanalytics.hlaxapi.HLADecoderRegistry; import com.yetanalytics.hlaxapi.config.XapiConfig; @@ -9,13 +7,7 @@ import com.yetanalytics.hlaxapi.config.model.Target; import com.yetanalytics.hlaxapi.config.model.TrackedObject; import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.PreparedStatement; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; import java.time.Instant; -import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -27,43 +19,52 @@ public class ObjectCache implements AutoCloseable { - private static final int SCHEMA_VERSION = 1; - private final FomCatalog catalog; private final Map> subscriptions; private final HlaValueFlattener valueFlattener; private final CacheQueryService queryService; - private final ObjectMapper mapper = new ObjectMapper(); private final AtomicLong sequence = new AtomicLong(); - private Connection connection; + private ObjectCacheStore store; public ObjectCache(XapiConfig xapiConfig, FomCatalog catalog, FOMXML fomXml, HLADecoderRegistry decoderRegistry) { - this(xapiConfig, catalog, fomXml, decoderRegistry, defaultJdbcUrl()); + this(xapiConfig, catalog, fomXml, decoderRegistry, (ObjectCacheConnectionSettings) null); } + // Convenience overload for SQLite tests that supply a temporary database URL. ObjectCache( XapiConfig xapiConfig, FomCatalog catalog, FOMXML fomXml, HLADecoderRegistry decoderRegistry, String jdbcUrl) { + this( + xapiConfig, + catalog, + fomXml, + decoderRegistry, + ObjectCacheConnectionSettings.sqlite(jdbcUrl)); + } + + ObjectCache( + XapiConfig xapiConfig, + FomCatalog catalog, + FOMXML fomXml, + HLADecoderRegistry decoderRegistry, + ObjectCacheConnectionSettings settings) { this.catalog = Objects.requireNonNull(catalog, "catalog"); this.subscriptions = collectSubscriptions(xapiConfig); this.valueFlattener = new HlaValueFlattener(fomXml, decoderRegistry); this.queryService = new CacheQueryService(this); if (!subscriptions.isEmpty()) { - try { - this.connection = DriverManager.getConnection(Objects.requireNonNull(jdbcUrl, "jdbcUrl")); - initializeSchema(); - seedFomMetadata(); - } catch (SQLException e) { - throw new IllegalStateException("Could not initialize HLA object cache", e); - } + ObjectCacheConnectionSettings effectiveSettings = settings == null + ? ObjectCacheConnectionSettings.from(System.getenv()) + : settings; + this.store = ObjectCacheStoreFactory.open(effectiveSettings, catalog); } } public boolean isEnabled() { - return connection != null; + return store != null && store.isOpen(); } public Map> subscriptions() { @@ -116,7 +117,7 @@ public ValueResolution findValueResolution(CachedObject object, Target attrTarge public synchronized void discoverObject(String objectHandle, String objectName, String className) { if (isEnabled()) { FomCatalog.ObjectClassDef clazz = requireClass(className); - ensureObject(objectHandle, objectName, clazz.localName()); + store.ensureObject(objectHandle, objectName, clazz); } } @@ -129,86 +130,41 @@ public synchronized void reflectAttributeValue( return; } FomCatalog.ObjectClassDef clazz = requireClass(className); - CachedObject object = ensureObject(objectHandle, null, clazz.localName()); + CachedObject object = store.ensureObject(objectHandle, null, clazz); FomCatalog.FomAttribute topAttribute = clazz.attribute(attributeName) .orElseThrow(() -> new IllegalArgumentException( "No FOM attribute " + attributeName + " on object class " + className)); List values = valueFlattener.flatten(attributeName, topAttribute.dataType(), bytes); String observedAt = Instant.now().toString(); long observedSequence = sequence.incrementAndGet(); - for (DecodedAttributeValue value : values) { - attributeIdForPath(clazz, value.pathKey()).ifPresent(attributeId -> upsertCurrentValue( - object.id(), - attributeId, - value, - observedAt, - observedSequence)); - } + store.replaceCurrentValues( + object.id(), + clazz, + attributeName, + values, + observedAt, + observedSequence); } public synchronized void removeObject(String objectHandle) { if (!isEnabled()) { return; } - try (PreparedStatement statement = connection.prepareStatement( - "UPDATE object_instance SET removed_at = ? WHERE object_handle = ?")) { - statement.setString(1, Instant.now().toString()); - statement.setString(2, objectHandle); - statement.executeUpdate(); - } catch (SQLException e) { - throw new IllegalStateException("Could not mark object removed: " + objectHandle, e); - } + store.removeObject(objectHandle, Instant.now().toString()); } public synchronized Optional findCurrentValue(long instanceId, String pathKey) { if (!isEnabled()) { return Optional.empty(); } - String normalizedPath = FomCatalog.wildcardArrayIndexes(pathKey); - String sql = """ - SELECT c.value_type, c.value_json, c.value_blob, c.raw_bytes - FROM object_attribute_current c - JOIN fom_attribute a ON a.id = c.attribute_id - WHERE c.instance_id = ? AND (a.path_key = ? OR a.path_key = ?) - ORDER BY CASE WHEN a.path_key = ? THEN 0 ELSE 1 END - LIMIT 1 - """; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setLong(1, instanceId); - statement.setString(2, pathKey); - statement.setString(3, normalizedPath); - statement.setString(4, pathKey); - try (ResultSet resultSet = statement.executeQuery()) { - if (!resultSet.next()) { - return Optional.empty(); - } - return Optional.of(readCachedValue(resultSet)); - } - } catch (SQLException e) { - throw new IllegalStateException("Could not read current cached value", e); - } + return store.findCurrentValue(instanceId, pathKey); } public synchronized Optional findCurrentValue(String objectHandle, String pathKey) { if (!isEnabled()) { return Optional.empty(); } - String sql = """ - SELECT id - FROM object_instance - WHERE object_handle = ? - """; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setString(1, objectHandle); - try (ResultSet resultSet = statement.executeQuery()) { - if (!resultSet.next()) { - return Optional.empty(); - } - return findCurrentValue(resultSet.getLong("id"), pathKey); - } - } catch (SQLException e) { - throw new IllegalStateException("Could not read object handle: " + objectHandle, e); - } + return store.findCurrentValue(objectHandle, pathKey); } public synchronized List currentObjects(String className) { @@ -216,54 +172,20 @@ public synchronized List currentObjects(String className) { return List.of(); } FomCatalog.ObjectClassDef clazz = requireClass(className); - String sql = """ - SELECT id, object_handle, object_name - FROM object_instance - WHERE class_id = ? AND removed_at IS NULL - ORDER BY id - """; - List objects = new ArrayList<>(); - try (PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setInt(1, clazz.id()); - try (ResultSet resultSet = statement.executeQuery()) { - while (resultSet.next()) { - objects.add(new CachedObject( - resultSet.getLong("id"), - resultSet.getString("object_handle"), - resultSet.getString("object_name"), - clazz.localName())); - } - } - } catch (SQLException e) { - throw new IllegalStateException("Could not list cached objects for class " + className, e); - } - return objects; + return store.currentObjects(clazz); } Connection connection() { - return connection; + return store == null ? null : store.connection(); } @Override public synchronized void close() { - if (connection == null) { + if (store == null) { return; } - try { - connection.close(); - connection = null; - } catch (SQLException e) { - throw new IllegalStateException("Could not close HLA object cache", e); - } - } - - public static String defaultJdbcUrl() { - String configured = System.getenv("HLA_OBJECT_CACHE_JDBC_URL"); - if (configured != null && !configured.isBlank()) { - return configured; - } - String path = System.getenv().getOrDefault("HLA_OBJECT_CACHE_DB", "hla-object-cache.sqlite"); - return "jdbc:sqlite:" + path; + store.close(); + store = null; } private FomCatalog.ObjectClassDef requireClass(String className) { @@ -271,270 +193,6 @@ private FomCatalog.ObjectClassDef requireClass(String className) { .orElseThrow(() -> new IllegalArgumentException("No FOM object class " + className)); } - private CachedObject ensureObject(String objectHandle, String objectName, String className) { - FomCatalog.ObjectClassDef clazz = requireClass(className); - String sql = """ - INSERT INTO object_instance (object_handle, object_name, class_id, discovered_at) - VALUES (?, ?, ?, ?) - ON CONFLICT(object_handle) DO UPDATE SET - object_name = COALESCE(excluded.object_name, object_instance.object_name), - class_id = excluded.class_id, - removed_at = NULL - """; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setString(1, Objects.requireNonNull(objectHandle, "objectHandle")); - statement.setString(2, objectName); - statement.setInt(3, clazz.id()); - statement.setString(4, Instant.now().toString()); - statement.executeUpdate(); - return loadObject(objectHandle, clazz.localName()); - } catch (SQLException e) { - throw new IllegalStateException("Could not upsert object instance " + objectHandle, e); - } - } - - private CachedObject loadObject(String objectHandle, String className) throws SQLException { - try (PreparedStatement statement = connection.prepareStatement( - "SELECT id, object_name FROM object_instance WHERE object_handle = ?")) { - statement.setString(1, objectHandle); - try (ResultSet resultSet = statement.executeQuery()) { - if (!resultSet.next()) { - throw new SQLException("Object instance was not inserted: " + objectHandle); - } - return new CachedObject( - resultSet.getLong("id"), - objectHandle, - resultSet.getString("object_name"), - className); - } - } - } - - private void initializeSchema() throws SQLException { - try (Statement statement = connection.createStatement()) { - statement.execute("PRAGMA foreign_keys = OFF"); - statement.execute("DROP TABLE IF EXISTS object_attribute_current"); - statement.execute("DROP TABLE IF EXISTS object_instance"); - statement.execute("DROP TABLE IF EXISTS fom_attribute"); - statement.execute("DROP TABLE IF EXISTS fom_object_class"); - statement.execute("PRAGMA foreign_keys = ON"); - statement.execute(""" - CREATE TABLE IF NOT EXISTS fom_object_class ( - id INTEGER PRIMARY KEY, - hla_name TEXT NOT NULL, - local_name TEXT NOT NULL UNIQUE, - parent_name TEXT - ) - """); - statement.execute(""" - CREATE TABLE IF NOT EXISTS fom_attribute ( - id INTEGER PRIMARY KEY, - class_id INTEGER NOT NULL REFERENCES fom_object_class(id), - attribute_name TEXT NOT NULL, - path_key TEXT NOT NULL, - data_type TEXT NOT NULL, - primitive_type TEXT, - is_leaf INTEGER NOT NULL, - UNIQUE(class_id, path_key) - ) - """); - statement.execute(""" - CREATE TABLE IF NOT EXISTS object_instance ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - object_handle TEXT NOT NULL UNIQUE, - object_name TEXT, - class_id INTEGER NOT NULL REFERENCES fom_object_class(id), - discovered_at TEXT NOT NULL, - removed_at TEXT - ) - """); - statement.execute(""" - CREATE TABLE IF NOT EXISTS object_attribute_current ( - instance_id INTEGER NOT NULL REFERENCES object_instance(id), - attribute_id INTEGER NOT NULL REFERENCES fom_attribute(id), - value_type TEXT NOT NULL, - value_blob BLOB, - value_json TEXT, - raw_bytes BLOB, - observed_at TEXT NOT NULL, - sequence INTEGER NOT NULL, - PRIMARY KEY(instance_id, attribute_id) - ) - """); - statement.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_object_handle ON object_instance(object_handle)"); - statement.execute("CREATE INDEX IF NOT EXISTS idx_fom_attribute_class_path " - + "ON fom_attribute(class_id, path_key)"); - statement.execute("PRAGMA user_version = " + SCHEMA_VERSION); - } - } - - private void seedFomMetadata() throws SQLException { - boolean autoCommit = connection.getAutoCommit(); - connection.setAutoCommit(false); - try { - insertClasses(); - insertAttributes(); - connection.commit(); - } catch (SQLException e) { - connection.rollback(); - throw e; - } finally { - connection.setAutoCommit(autoCommit); - } - } - - private void insertClasses() throws SQLException { - String sql = """ - INSERT OR IGNORE INTO fom_object_class (id, hla_name, local_name, parent_name) - VALUES (?, ?, ?, ?) - """; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - for (FomCatalog.ObjectClassDef clazz : catalog.objectClasses()) { - statement.setInt(1, clazz.id()); - statement.setString(2, clazz.hlaName()); - statement.setString(3, clazz.localName()); - statement.setString(4, clazz.parentName()); - statement.addBatch(); - } - statement.executeBatch(); - } - } - - private void insertAttributes() throws SQLException { - String sql = """ - INSERT OR IGNORE INTO fom_attribute - (id, class_id, attribute_name, path_key, data_type, primitive_type, is_leaf) - VALUES (?, ?, ?, ?, ?, ?, ?) - """; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - for (FomCatalog.ObjectClassDef clazz : catalog.objectClasses()) { - for (FomCatalog.FomAttribute attribute : clazz.attributes()) { - statement.setInt(1, attribute.id()); - statement.setInt(2, attribute.classId()); - statement.setString(3, attribute.attributeName()); - statement.setString(4, attribute.pathKey()); - statement.setString(5, attribute.dataType()); - statement.setString(6, attribute.primitiveType()); - statement.setInt(7, attribute.leaf() ? 1 : 0); - statement.addBatch(); - } - } - statement.executeBatch(); - } - } - - private void upsertCurrentValue( - long instanceId, - int attributeId, - DecodedAttributeValue value, - String observedAt, - long observedSequence) { - String sql = """ - INSERT INTO object_attribute_current - (instance_id, attribute_id, value_type, value_blob, value_json, raw_bytes, observed_at, sequence) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(instance_id, attribute_id) DO UPDATE SET - value_type = excluded.value_type, - value_blob = excluded.value_blob, - value_json = excluded.value_json, - raw_bytes = excluded.raw_bytes, - observed_at = excluded.observed_at, - sequence = excluded.sequence - """; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - Object objectValue = value.value(); - String valueType = CachedValue.valueType(objectValue); - statement.setLong(1, instanceId); - statement.setInt(2, attributeId); - statement.setString(3, valueType); - if (objectValue instanceof byte[] bytes) { - statement.setBytes(4, bytes); - statement.setNull(5, java.sql.Types.VARCHAR); - } else { - statement.setNull(4, java.sql.Types.BLOB); - statement.setString(5, serializeJson(objectValue)); - } - statement.setBytes(6, value.rawBytes()); - statement.setString(7, observedAt); - statement.setLong(8, observedSequence); - statement.executeUpdate(); - } catch (SQLException e) { - throw new IllegalStateException("Could not upsert current object attribute", e); - } - } - - private Optional attributeIdForPath(FomCatalog.ObjectClassDef clazz, String pathKey) { - Optional existingId = attributeIdFromDatabase(clazz.id(), pathKey); - if (existingId.isPresent()) { - return existingId; - } - - String wildcardPath = FomCatalog.wildcardArrayIndexes(pathKey); - FomCatalog.FomAttribute template = clazz.attribute(wildcardPath).orElse(null); - if (template == null || wildcardPath.equals(pathKey)) { - return Optional.empty(); - } - - String sql = """ - INSERT OR IGNORE INTO fom_attribute - (class_id, attribute_name, path_key, data_type, primitive_type, is_leaf) - VALUES (?, ?, ?, ?, ?, ?) - """; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setInt(1, clazz.id()); - statement.setString(2, template.attributeName()); - statement.setString(3, pathKey); - statement.setString(4, template.dataType()); - statement.setString(5, template.primitiveType()); - statement.setInt(6, template.leaf() ? 1 : 0); - statement.executeUpdate(); - } catch (SQLException e) { - throw new IllegalStateException("Could not insert dynamic FOM attribute path " + pathKey, e); - } - return attributeIdFromDatabase(clazz.id(), pathKey); - } - - private Optional attributeIdFromDatabase(int classId, String pathKey) { - String sql = "SELECT id FROM fom_attribute WHERE class_id = ? AND path_key = ?"; - try (PreparedStatement statement = connection.prepareStatement(sql)) { - statement.setInt(1, classId); - statement.setString(2, pathKey); - try (ResultSet resultSet = statement.executeQuery()) { - if (resultSet.next()) { - return Optional.of(resultSet.getInt("id")); - } - } - } catch (SQLException e) { - throw new IllegalStateException("Could not read FOM attribute path " + pathKey, e); - } - return Optional.empty(); - } - - private String serializeJson(Object value) throws SQLException { - try { - return mapper.writeValueAsString(value); - } catch (JsonProcessingException e) { - throw new SQLException("Could not serialize cached value as JSON", e); - } - } - - private CachedValue readCachedValue(ResultSet resultSet) throws SQLException { - String valueType = resultSet.getString("value_type"); - byte[] rawBytes = resultSet.getBytes("raw_bytes"); - if ("blob".equals(valueType)) { - return new CachedValue(valueType, resultSet.getBytes("value_blob"), rawBytes); - } - String valueJson = resultSet.getString("value_json"); - if (valueJson == null) { - return new CachedValue(valueType, null, rawBytes); - } - try { - return new CachedValue(valueType, mapper.readValue(valueJson, Object.class), rawBytes); - } catch (JsonProcessingException e) { - throw new SQLException("Could not deserialize cached value JSON", e); - } - } - private Map> collectSubscriptions(XapiConfig xapiConfig) { Map> merged = new LinkedHashMap<>(); QueryReferenceCollector.collect(xapiConfig.statementTriggers) diff --git a/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheConnectionSettings.java b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheConnectionSettings.java new file mode 100644 index 0000000..d86535c --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheConnectionSettings.java @@ -0,0 +1,107 @@ +package com.yetanalytics.hlaxapi.cache; + +import com.yetanalytics.hlaxapi.config.model.ObjectCacheBackend; +import java.util.Map; +import java.util.Objects; +import java.util.regex.Pattern; + +record ObjectCacheConnectionSettings( + ObjectCacheBackend backend, + String jdbcUrl, + String username, + String password, + String schema) { + + static final String DEFAULT_SQLITE_PATH = "hla-object-cache.sqlite"; + static final String DEFAULT_POSTGRESQL_SCHEMA = "hla_object_cache"; + private static final Pattern SQL_IDENTIFIER = Pattern.compile("[A-Za-z_][A-Za-z0-9_]*"); + + static ObjectCacheConnectionSettings from(Map environment) { + Objects.requireNonNull(environment, "environment"); + ObjectCacheBackend backend = ObjectCacheBackend.fromString(environment.get("HLA_OBJECT_CACHE_BACKEND")); + return switch (backend) { + case SQLITE -> sqlite(environment); + case POSTGRESQL -> postgresql(environment); + }; + } + + static ObjectCacheConnectionSettings sqlite(String jdbcUrl) { + return validated(ObjectCacheBackend.SQLITE, jdbcUrl, null, null, null); + } + + static ObjectCacheConnectionSettings postgresql( + String jdbcUrl, + String username, + String password, + String schema) { + return validated(ObjectCacheBackend.POSTGRESQL, jdbcUrl, username, password, schema); + } + + static String defaultSqliteJdbcUrl(Map environment) { + String configured = trimmed(environment.get("HLA_OBJECT_CACHE_JDBC_URL")); + if (configured != null) { + return configured; + } + String path = trimmed(environment.get("HLA_OBJECT_CACHE_DB")); + return "jdbc:sqlite:" + (path == null ? DEFAULT_SQLITE_PATH : path); + } + + private static ObjectCacheConnectionSettings sqlite(Map environment) { + return validated( + ObjectCacheBackend.SQLITE, + defaultSqliteJdbcUrl(environment), + null, + null, + null); + } + + private static ObjectCacheConnectionSettings postgresql(Map environment) { + String jdbcUrl = trimmed(environment.get("HLA_OBJECT_CACHE_JDBC_URL")); + if (jdbcUrl == null) { + throw new IllegalArgumentException( + "HLA_OBJECT_CACHE_JDBC_URL is required for the PostgreSQL object cache backend"); + } + return validated( + ObjectCacheBackend.POSTGRESQL, + jdbcUrl, + environment.get("HLA_OBJECT_CACHE_USERNAME"), + environment.get("HLA_OBJECT_CACHE_PASSWORD"), + Objects.requireNonNullElse( + trimmed(environment.get("HLA_OBJECT_CACHE_SCHEMA")), + DEFAULT_POSTGRESQL_SCHEMA)); + } + + private static ObjectCacheConnectionSettings validated( + ObjectCacheBackend backend, + String jdbcUrl, + String username, + String password, + String schema) { + Objects.requireNonNull(backend, "backend"); + Objects.requireNonNull(jdbcUrl, "jdbcUrl"); + String requiredPrefix = switch (backend) { + case SQLITE -> "jdbc:sqlite:"; + case POSTGRESQL -> "jdbc:postgresql:"; + }; + if (!jdbcUrl.startsWith(requiredPrefix)) { + throw new IllegalArgumentException( + "Object cache JDBC URL must start with " + requiredPrefix + " for backend " + backend); + } + if ((username == null) != (password == null)) { + throw new IllegalArgumentException( + "HLA_OBJECT_CACHE_USERNAME and HLA_OBJECT_CACHE_PASSWORD must be provided together"); + } + if (backend == ObjectCacheBackend.POSTGRESQL + && (schema == null || !SQL_IDENTIFIER.matcher(schema).matches())) { + throw new IllegalArgumentException("Invalid PostgreSQL object cache schema: " + schema); + } + return new ObjectCacheConnectionSettings(backend, jdbcUrl, username, password, schema); + } + + private static String trimmed(String value) { + if (value == null || value.isBlank()) { + return null; + } + return value.trim(); + } +} diff --git a/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheQueries.java b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheQueries.java new file mode 100644 index 0000000..6e46ff4 --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheQueries.java @@ -0,0 +1,42 @@ +package com.yetanalytics.hlaxapi.cache; + +import java.util.List; +import java.util.Optional; + +interface ObjectCacheQueries { + int binaryJdbcType(); + + List connectionSetupStatements(); + + List resetSchemaStatements(); + + List createSchemaStatements(); + + String insertSchemaVersion(); + + String insertClass(); + + String insertAttribute(); + + Optional afterSeedFomMetadata(); + + String upsertObject(); + + String loadObject(); + + String removeObject(); + + String findCurrentValue(); + + String findObjectId(); + + String listCurrentObjects(); + + String deleteCurrentValues(); + + String upsertCurrentValue(); + + String insertDynamicAttribute(); + + String findAttributeId(); +} diff --git a/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheStore.java b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheStore.java new file mode 100644 index 0000000..94a5219 --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheStore.java @@ -0,0 +1,32 @@ +package com.yetanalytics.hlaxapi.cache; + +import java.sql.Connection; +import java.util.List; +import java.util.Optional; + +interface ObjectCacheStore extends AutoCloseable { + boolean isOpen(); + + CachedObject ensureObject(String objectHandle, String objectName, FomCatalog.ObjectClassDef clazz); + + void removeObject(String objectHandle, String removedAt); + + Optional findCurrentValue(long instanceId, String pathKey); + + Optional findCurrentValue(String objectHandle, String pathKey); + + List currentObjects(FomCatalog.ObjectClassDef clazz); + + void replaceCurrentValues( + long instanceId, + FomCatalog.ObjectClassDef clazz, + String attributeName, + List values, + String observedAt, + long observedSequence); + + Connection connection(); + + @Override + void close(); +} diff --git a/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheStoreFactory.java b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheStoreFactory.java new file mode 100644 index 0000000..faf9eaf --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/cache/ObjectCacheStoreFactory.java @@ -0,0 +1,44 @@ +package com.yetanalytics.hlaxapi.cache; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.util.Properties; + +final class ObjectCacheStoreFactory { + + private ObjectCacheStoreFactory() { + } + + static ObjectCacheStore open(ObjectCacheConnectionSettings settings, FomCatalog catalog) { + Connection connection = null; + try { + Properties properties = new Properties(); + if (settings.username() != null) { + properties.setProperty("user", settings.username()); + properties.setProperty("password", settings.password()); + } + connection = DriverManager.getConnection(settings.jdbcUrl(), properties); + ObjectCacheQueries queries = switch (settings.backend()) { + case SQLITE -> new SqliteObjectCacheQueries(); + case POSTGRESQL -> new PostgresqlObjectCacheQueries(settings.schema()); + }; + return new JdbcObjectCacheStore(connection, queries, catalog); + } catch (SQLException | RuntimeException e) { + closeAfterFailure(connection, e); + throw new IllegalStateException( + "Could not initialize " + settings.backend() + " HLA object cache", e); + } + } + + private static void closeAfterFailure(Connection connection, Exception failure) { + if (connection == null) { + return; + } + try { + connection.close(); + } catch (SQLException closeError) { + failure.addSuppressed(closeError); + } + } +} diff --git a/src/main/java/com/yetanalytics/hlaxapi/cache/PostgresqlObjectCacheQueries.java b/src/main/java/com/yetanalytics/hlaxapi/cache/PostgresqlObjectCacheQueries.java new file mode 100644 index 0000000..2859121 --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/cache/PostgresqlObjectCacheQueries.java @@ -0,0 +1,215 @@ +package com.yetanalytics.hlaxapi.cache; + +import java.util.List; +import java.util.Optional; + +final class PostgresqlObjectCacheQueries implements ObjectCacheQueries { + + private final String quotedSchema; + + PostgresqlObjectCacheQueries(String schema) { + this.quotedSchema = "\"" + schema + "\""; + } + + @Override + public int binaryJdbcType() { + return java.sql.Types.BINARY; + } + + @Override + public List connectionSetupStatements() { + return List.of( + "CREATE SCHEMA IF NOT EXISTS " + quotedSchema, + "SET search_path TO " + quotedSchema); + } + + @Override + public List resetSchemaStatements() { + return List.of( + "DROP TABLE IF EXISTS object_attribute_current", + "DROP TABLE IF EXISTS object_instance", + "DROP TABLE IF EXISTS fom_attribute", + "DROP TABLE IF EXISTS fom_object_class", + "DROP TABLE IF EXISTS object_cache_metadata"); + } + + @Override + public List createSchemaStatements() { + return List.of( + """ + CREATE TABLE object_cache_metadata ( + schema_version INTEGER NOT NULL + ) + """, + """ + CREATE TABLE fom_object_class ( + id INTEGER PRIMARY KEY, + hla_name TEXT NOT NULL, + local_name TEXT NOT NULL UNIQUE, + parent_name TEXT + ) + """, + """ + CREATE TABLE fom_attribute ( + id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + class_id INTEGER NOT NULL REFERENCES fom_object_class(id), + attribute_name TEXT NOT NULL, + path_key TEXT NOT NULL, + data_type TEXT NOT NULL, + primitive_type TEXT, + is_leaf INTEGER NOT NULL, + UNIQUE(class_id, path_key) + ) + """, + """ + CREATE TABLE object_instance ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + object_handle TEXT NOT NULL UNIQUE, + object_name TEXT, + class_id INTEGER NOT NULL REFERENCES fom_object_class(id), + discovered_at TEXT NOT NULL, + removed_at TEXT + ) + """, + """ + CREATE TABLE object_attribute_current ( + instance_id BIGINT NOT NULL REFERENCES object_instance(id), + attribute_id INTEGER NOT NULL REFERENCES fom_attribute(id), + value_type TEXT NOT NULL, + value_blob BYTEA, + value_json TEXT, + raw_bytes BYTEA, + observed_at TEXT NOT NULL, + sequence BIGINT NOT NULL, + PRIMARY KEY(instance_id, attribute_id) + ) + """); + } + + @Override + public String insertSchemaVersion() { + return "INSERT INTO object_cache_metadata (schema_version) VALUES (?)"; + } + + @Override + public String insertClass() { + return """ + INSERT INTO fom_object_class (id, hla_name, local_name, parent_name) + VALUES (?, ?, ?, ?) + ON CONFLICT(id) DO NOTHING + """; + } + + @Override + public String insertAttribute() { + return """ + INSERT INTO fom_attribute + (id, class_id, attribute_name, path_key, data_type, primitive_type, is_leaf) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO NOTHING + """; + } + + @Override + public Optional afterSeedFomMetadata() { + return Optional.of(""" + SELECT setval( + pg_get_serial_sequence('fom_attribute', 'id'), + COALESCE((SELECT MAX(id) FROM fom_attribute), 1), + true) + """); + } + + @Override + public String upsertObject() { + return """ + INSERT INTO object_instance (object_handle, object_name, class_id, discovered_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(object_handle) DO UPDATE SET + object_name = COALESCE(excluded.object_name, object_instance.object_name), + class_id = excluded.class_id, + removed_at = NULL + """; + } + + @Override + public String loadObject() { + return "SELECT id, object_name FROM object_instance WHERE object_handle = ?"; + } + + @Override + public String removeObject() { + return "UPDATE object_instance SET removed_at = ? WHERE object_handle = ?"; + } + + @Override + public String findCurrentValue() { + return """ + SELECT c.value_type, c.value_json, c.value_blob, c.raw_bytes + FROM object_attribute_current c + JOIN fom_attribute a ON a.id = c.attribute_id + WHERE c.instance_id = ? AND (a.path_key = ? OR a.path_key = ?) + ORDER BY CASE WHEN a.path_key = ? THEN 0 ELSE 1 END + LIMIT 1 + """; + } + + @Override + public String findObjectId() { + return "SELECT id FROM object_instance WHERE object_handle = ?"; + } + + @Override + public String listCurrentObjects() { + return """ + SELECT id, object_handle, object_name + FROM object_instance + WHERE class_id = ? AND removed_at IS NULL + ORDER BY id + """; + } + + @Override + public String deleteCurrentValues() { + return """ + DELETE FROM object_attribute_current + WHERE instance_id = ? + AND attribute_id IN ( + SELECT id + FROM fom_attribute + WHERE class_id = ? AND attribute_name = ? + ) + """; + } + + @Override + public String upsertCurrentValue() { + return """ + INSERT INTO object_attribute_current + (instance_id, attribute_id, value_type, value_blob, value_json, raw_bytes, observed_at, sequence) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(instance_id, attribute_id) DO UPDATE SET + value_type = excluded.value_type, + value_blob = excluded.value_blob, + value_json = excluded.value_json, + raw_bytes = excluded.raw_bytes, + observed_at = excluded.observed_at, + sequence = excluded.sequence + """; + } + + @Override + public String insertDynamicAttribute() { + return """ + INSERT INTO fom_attribute + (class_id, attribute_name, path_key, data_type, primitive_type, is_leaf) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(class_id, path_key) DO NOTHING + """; + } + + @Override + public String findAttributeId() { + return "SELECT id FROM fom_attribute WHERE class_id = ? AND path_key = ?"; + } +} diff --git a/src/main/java/com/yetanalytics/hlaxapi/cache/SqliteObjectCacheQueries.java b/src/main/java/com/yetanalytics/hlaxapi/cache/SqliteObjectCacheQueries.java new file mode 100644 index 0000000..08e883d --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/cache/SqliteObjectCacheQueries.java @@ -0,0 +1,200 @@ +package com.yetanalytics.hlaxapi.cache; + +import java.util.List; +import java.util.Optional; + +final class SqliteObjectCacheQueries implements ObjectCacheQueries { + + @Override + public int binaryJdbcType() { + return java.sql.Types.BLOB; + } + + @Override + public List connectionSetupStatements() { + return List.of("PRAGMA foreign_keys = ON"); + } + + @Override + public List resetSchemaStatements() { + return List.of( + "DROP TABLE IF EXISTS object_attribute_current", + "DROP TABLE IF EXISTS object_instance", + "DROP TABLE IF EXISTS fom_attribute", + "DROP TABLE IF EXISTS fom_object_class", + "DROP TABLE IF EXISTS object_cache_metadata"); + } + + @Override + public List createSchemaStatements() { + return List.of( + """ + CREATE TABLE object_cache_metadata ( + schema_version INTEGER NOT NULL + ) + """, + """ + CREATE TABLE fom_object_class ( + id INTEGER PRIMARY KEY, + hla_name TEXT NOT NULL, + local_name TEXT NOT NULL UNIQUE, + parent_name TEXT + ) + """, + """ + CREATE TABLE fom_attribute ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + class_id INTEGER NOT NULL REFERENCES fom_object_class(id), + attribute_name TEXT NOT NULL, + path_key TEXT NOT NULL, + data_type TEXT NOT NULL, + primitive_type TEXT, + is_leaf INTEGER NOT NULL, + UNIQUE(class_id, path_key) + ) + """, + """ + CREATE TABLE object_instance ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + object_handle TEXT NOT NULL UNIQUE, + object_name TEXT, + class_id INTEGER NOT NULL REFERENCES fom_object_class(id), + discovered_at TEXT NOT NULL, + removed_at TEXT + ) + """, + """ + CREATE TABLE object_attribute_current ( + instance_id INTEGER NOT NULL REFERENCES object_instance(id), + attribute_id INTEGER NOT NULL REFERENCES fom_attribute(id), + value_type TEXT NOT NULL, + value_blob BLOB, + value_json TEXT, + raw_bytes BLOB, + observed_at TEXT NOT NULL, + sequence INTEGER NOT NULL, + PRIMARY KEY(instance_id, attribute_id) + ) + """, + "PRAGMA user_version = 1"); + } + + @Override + public String insertSchemaVersion() { + return "INSERT INTO object_cache_metadata (schema_version) VALUES (?)"; + } + + @Override + public String insertClass() { + return """ + INSERT OR IGNORE INTO fom_object_class (id, hla_name, local_name, parent_name) + VALUES (?, ?, ?, ?) + """; + } + + @Override + public String insertAttribute() { + return """ + INSERT OR IGNORE INTO fom_attribute + (id, class_id, attribute_name, path_key, data_type, primitive_type, is_leaf) + VALUES (?, ?, ?, ?, ?, ?, ?) + """; + } + + @Override + public Optional afterSeedFomMetadata() { + return Optional.empty(); + } + + @Override + public String upsertObject() { + return """ + INSERT INTO object_instance (object_handle, object_name, class_id, discovered_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(object_handle) DO UPDATE SET + object_name = COALESCE(excluded.object_name, object_instance.object_name), + class_id = excluded.class_id, + removed_at = NULL + """; + } + + @Override + public String loadObject() { + return "SELECT id, object_name FROM object_instance WHERE object_handle = ?"; + } + + @Override + public String removeObject() { + return "UPDATE object_instance SET removed_at = ? WHERE object_handle = ?"; + } + + @Override + public String findCurrentValue() { + return """ + SELECT c.value_type, c.value_json, c.value_blob, c.raw_bytes + FROM object_attribute_current c + JOIN fom_attribute a ON a.id = c.attribute_id + WHERE c.instance_id = ? AND (a.path_key = ? OR a.path_key = ?) + ORDER BY CASE WHEN a.path_key = ? THEN 0 ELSE 1 END + LIMIT 1 + """; + } + + @Override + public String findObjectId() { + return "SELECT id FROM object_instance WHERE object_handle = ?"; + } + + @Override + public String listCurrentObjects() { + return """ + SELECT id, object_handle, object_name + FROM object_instance + WHERE class_id = ? AND removed_at IS NULL + ORDER BY id + """; + } + + @Override + public String deleteCurrentValues() { + return """ + DELETE FROM object_attribute_current + WHERE instance_id = ? + AND attribute_id IN ( + SELECT id + FROM fom_attribute + WHERE class_id = ? AND attribute_name = ? + ) + """; + } + + @Override + public String upsertCurrentValue() { + return """ + INSERT INTO object_attribute_current + (instance_id, attribute_id, value_type, value_blob, value_json, raw_bytes, observed_at, sequence) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(instance_id, attribute_id) DO UPDATE SET + value_type = excluded.value_type, + value_blob = excluded.value_blob, + value_json = excluded.value_json, + raw_bytes = excluded.raw_bytes, + observed_at = excluded.observed_at, + sequence = excluded.sequence + """; + } + + @Override + public String insertDynamicAttribute() { + return """ + INSERT OR IGNORE INTO fom_attribute + (class_id, attribute_name, path_key, data_type, primitive_type, is_leaf) + VALUES (?, ?, ?, ?, ?, ?) + """; + } + + @Override + public String findAttributeId() { + return "SELECT id FROM fom_attribute WHERE class_id = ? AND path_key = ?"; + } +} diff --git a/src/main/java/com/yetanalytics/hlaxapi/config/model/ObjectCacheBackend.java b/src/main/java/com/yetanalytics/hlaxapi/config/model/ObjectCacheBackend.java new file mode 100644 index 0000000..ecf4e20 --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/config/model/ObjectCacheBackend.java @@ -0,0 +1,19 @@ +package com.yetanalytics.hlaxapi.config.model; + +import java.util.Locale; + +public enum ObjectCacheBackend { + SQLITE, + POSTGRESQL; + + public static ObjectCacheBackend fromString(String value) { + if (value == null || value.isBlank()) { + return SQLITE; + } + return switch (value.trim().toLowerCase(Locale.ROOT)) { + case "sqlite" -> SQLITE; + case "postgresql" -> POSTGRESQL; + default -> throw new IllegalArgumentException("Unsupported object cache backend: " + value); + }; + } +} diff --git a/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCacheConnectionSettingsTest.java b/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCacheConnectionSettingsTest.java new file mode 100644 index 0000000..67d76c7 --- /dev/null +++ b/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCacheConnectionSettingsTest.java @@ -0,0 +1,97 @@ +package com.yetanalytics.hlaxapi.cache; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import com.yetanalytics.hlaxapi.config.model.ObjectCacheBackend; +import java.util.HashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class ObjectCacheConnectionSettingsTest { + + @Test + void defaultsToSqliteDatabasePath() { + ObjectCacheConnectionSettings settings = ObjectCacheConnectionSettings.from(Map.of()); + + assertEquals(ObjectCacheBackend.SQLITE, settings.backend()); + assertEquals("jdbc:sqlite:hla-object-cache.sqlite", settings.jdbcUrl()); + } + + @Test + void supportsSqlitePathAndJdbcUrlOverrides() { + assertEquals( + "jdbc:sqlite:/tmp/cache.sqlite", + ObjectCacheConnectionSettings.from(Map.of("HLA_OBJECT_CACHE_DB", "/tmp/cache.sqlite")) + .jdbcUrl()); + assertEquals( + "jdbc:sqlite::memory:", + ObjectCacheConnectionSettings.from(Map.of("HLA_OBJECT_CACHE_JDBC_URL", "jdbc:sqlite::memory:")) + .jdbcUrl()); + } + + @Test + void resolvesPostgresqlConnectionSettings() { + Map environment = Map.of( + "HLA_OBJECT_CACHE_BACKEND", " PostgreSQL ", + "HLA_OBJECT_CACHE_JDBC_URL", "jdbc:postgresql://localhost/cache", + "HLA_OBJECT_CACHE_USERNAME", "cache_user", + "HLA_OBJECT_CACHE_PASSWORD", "secret", + "HLA_OBJECT_CACHE_SCHEMA", "adapter_cache"); + + ObjectCacheConnectionSettings settings = ObjectCacheConnectionSettings.from(environment); + + assertEquals(ObjectCacheBackend.POSTGRESQL, settings.backend()); + assertEquals("jdbc:postgresql://localhost/cache", settings.jdbcUrl()); + assertEquals("cache_user", settings.username()); + assertEquals("secret", settings.password()); + assertEquals("adapter_cache", settings.schema()); + } + + @Test + void postgresqlRequiresJdbcUrl() { + assertThrows( + IllegalArgumentException.class, + () -> ObjectCacheConnectionSettings.from(Map.of("HLA_OBJECT_CACHE_BACKEND", "postgresql"))); + } + + @Test + void rejectsBackendUrlMismatch() { + assertThrows( + IllegalArgumentException.class, + () -> ObjectCacheConnectionSettings.from( + Map.of( + "HLA_OBJECT_CACHE_BACKEND", "postgresql", + "HLA_OBJECT_CACHE_JDBC_URL", "jdbc:sqlite:cache.sqlite"))); + } + + @Test + void rejectsIncompleteCredentialPair() { + Map environment = new HashMap<>(); + environment.put("HLA_OBJECT_CACHE_BACKEND", "postgresql"); + environment.put("HLA_OBJECT_CACHE_JDBC_URL", "jdbc:postgresql://localhost/cache"); + environment.put("HLA_OBJECT_CACHE_USERNAME", "cache_user"); + + assertThrows( + IllegalArgumentException.class, + () -> ObjectCacheConnectionSettings.from(environment)); + } + + @Test + void rejectsUnsafePostgresqlSchema() { + assertThrows( + IllegalArgumentException.class, + () -> ObjectCacheConnectionSettings.from( + Map.of( + "HLA_OBJECT_CACHE_BACKEND", "postgresql", + "HLA_OBJECT_CACHE_JDBC_URL", "jdbc:postgresql://localhost/cache", + "HLA_OBJECT_CACHE_SCHEMA", "cache;drop schema public"))); + } + + @Test + void rejectsUnknownBackend() { + assertThrows( + IllegalArgumentException.class, + () -> ObjectCacheConnectionSettings.from(Map.of("HLA_OBJECT_CACHE_BACKEND", "mysql"))); + } +} diff --git a/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCachePersistenceTest.java b/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCachePersistenceTest.java index 04ff66b..6c1ed37 100644 --- a/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCachePersistenceTest.java +++ b/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCachePersistenceTest.java @@ -7,6 +7,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import com.yetanalytics.hlaxapi.FOMXML; +import com.yetanalytics.hlaxapi.HLAEncodingTestSupport; import com.yetanalytics.hlaxapi.HLADecoderRegistry; import com.yetanalytics.hlaxapi.SimulationConfig; import com.yetanalytics.hlaxapi.config.XapiConfig; @@ -22,28 +23,30 @@ import hla.rti1516e.encoding.EncoderException; import hla.rti1516e.encoding.EncoderFactory; import hla.rti1516e.encoding.HLAfixedRecord; -import java.nio.file.Path; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.util.List; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.io.TempDir; import org.portico.impl.hla1516e.types.encoding.HLA1516eEncoderFactory; -final class ObjectCachePersistenceTest { +abstract class ObjectCachePersistenceTest { - private final EncoderFactory encoderFactory = new HLA1516eEncoderFactory(); - private final HLADecoderRegistry decoderRegistry = new HLADecoderRegistry(encoderFactory); - private final FOMXML fomXml = new FOMXML( + protected final EncoderFactory encoderFactory = new HLA1516eEncoderFactory(); + protected final HLADecoderRegistry decoderRegistry = new HLADecoderRegistry(encoderFactory); + protected final FOMXML fomXml = new FOMXML( new SimulationConfig(null, null, null, null, "config/HlaFedereplFOM.xml"), decoderRegistry); - private final FomCatalog catalog = new FomCatalog(fomXml); + protected final FomCatalog catalog = new FomCatalog(fomXml); + private final FOMXML dynamicArrayFomXml = new FOMXML( + new SimulationConfig(null, null, null, null, "src/test/resources/config/ObjectCacheTestFOM.xml"), + decoderRegistry); + private final FomCatalog dynamicArrayCatalog = new FomCatalog(dynamicArrayFomXml); @Test void initializesSchemaAndSeedsFomMetadata() throws SQLException { try (ObjectCache cache = newCache()) { - assertEquals(1, scalarLong(cache, "PRAGMA user_version")); + assertEquals(1, scalarLong(cache, "SELECT schema_version FROM object_cache_metadata")); assertTrue(count(cache, "SELECT COUNT(*) FROM fom_object_class") > 0); assertTrue(count(cache, "SELECT COUNT(*) FROM fom_attribute WHERE path_key = 'Position.X'") > 0); } @@ -84,6 +87,69 @@ void flattensFixedRecordValuesToNestedCurrentRows() { } } + @Test + void replacesDynamicArrayPathsWhenArrayShrinksAndReusesMetadata() throws SQLException { + byte[] hunger = encoded(encoderFactory.createHLAinteger32BE(75)); + + try (ObjectCache cache = newCache( + "dynamic-array", + enabledConfig(), + dynamicArrayCatalog, + dynamicArrayFomXml)) { + cache.discoverObject("object-1", "Rabbit One", "Rabbit"); + cache.reflectAttributeValue("object-1", "Rabbit", "Hunger", hunger); + cache.reflectAttributeValue( + "object-1", + "Rabbit", + "PositionHistory", + positionHistory(position(1, 2), position(3, 4))); + + assertEquals(2, ((List) cache.findCurrentValue("object-1", "PositionHistory") + .orElseThrow() + .value()) + .size()); + assertEquals(1, cache.findCurrentValue("object-1", "PositionHistory[0].X") + .orElseThrow() + .value()); + assertEquals(4, cache.findCurrentValue("object-1", "PositionHistory[1].Y") + .orElseThrow() + .value()); + assertEquals(5, currentValueCount(cache, "PositionHistory")); + long secondElementXId = attributeId(cache, "PositionHistory[1].X"); + + cache.reflectAttributeValue( + "object-1", + "Rabbit", + "PositionHistory", + positionHistory(position(9, 10))); + + assertEquals(1, ((List) cache.findCurrentValue("object-1", "PositionHistory") + .orElseThrow() + .value()) + .size()); + assertEquals(9, cache.findCurrentValue("object-1", "PositionHistory[0].X") + .orElseThrow() + .value()); + assertFalse(cache.findCurrentValue("object-1", "PositionHistory[1].X").isPresent()); + assertFalse(cache.findCurrentValue("object-1", "PositionHistory[1].Y").isPresent()); + assertEquals(3, currentValueCount(cache, "PositionHistory")); + assertEquals(75, cache.findCurrentValue("object-1", "Hunger").orElseThrow().value()); + assertEquals(secondElementXId, attributeId(cache, "PositionHistory[1].X")); + + cache.reflectAttributeValue( + "object-1", + "Rabbit", + "PositionHistory", + positionHistory(position(11, 12), position(13, 14))); + + assertEquals(13, cache.findCurrentValue("object-1", "PositionHistory[1].X") + .orElseThrow() + .value()); + assertEquals(5, currentValueCount(cache, "PositionHistory")); + assertEquals(secondElementXId, attributeId(cache, "PositionHistory[1].X")); + } + } + @Test void queryServiceEvaluatesCriteriaAndExcludesRemovedObjects() { try (ObjectCache cache = newCache()) { @@ -149,10 +215,8 @@ void queryServiceDistinguishesPresentNullFromMissingValue() { } @Test - void persistentCacheStartsFreshOnInitialization(@TempDir Path tempDir) throws SQLException { - String jdbcUrl = "jdbc:sqlite:" + tempDir.resolve("cache.sqlite"); - - try (ObjectCache cache = newCache(jdbcUrl)) { + void persistentCacheStartsFreshOnInitialization() throws SQLException { + try (ObjectCache cache = newCache("fresh-start")) { cache.discoverObject("object-1", "Rabbit One", "Rabbit"); cache.reflectAttributeValue("object-1", "Rabbit", "Hunger", encoded(encoderFactory.createHLAinteger32BE(75))); @@ -161,22 +225,28 @@ void persistentCacheStartsFreshOnInitialization(@TempDir Path tempDir) throws SQ assertEquals(1, count(cache, "SELECT COUNT(*) FROM object_attribute_current")); } - try (ObjectCache cache = newCache(jdbcUrl)) { + try (ObjectCache cache = newCache("fresh-start")) { assertEquals(0, count(cache, "SELECT COUNT(*) FROM object_instance")); assertEquals(0, count(cache, "SELECT COUNT(*) FROM object_attribute_current")); assertTrue(count(cache, "SELECT COUNT(*) FROM fom_object_class") > 0); } } - private ObjectCache newCache() { - return newCache("jdbc:sqlite::memory:"); + protected ObjectCache newCache() { + return newCache("default"); } - private ObjectCache newCache(String jdbcUrl) { - return new ObjectCache(enabledConfig(), catalog, fomXml, decoderRegistry, jdbcUrl); + protected ObjectCache newCache(String name) { + return newCache(name, enabledConfig(), catalog, fomXml); } - private XapiConfig enabledConfig() { + protected abstract ObjectCache newCache( + String name, + XapiConfig config, + FomCatalog cacheCatalog, + FOMXML cacheFomXml); + + protected XapiConfig enabledConfig() { TrackedObject trackedObject = new TrackedObject(); trackedObject.clazz = "Rabbit"; trackedObject.allAttributes = true; @@ -187,14 +257,18 @@ private XapiConfig enabledConfig() { return config; } - private byte[] position(int x, int y) { + protected byte[] position(int x, int y) { HLAfixedRecord record = encoderFactory.createHLAfixedRecord(); record.add(encoderFactory.createHLAinteger32BE(x)); record.add(encoderFactory.createHLAinteger32BE(y)); return encoded(record); } - private byte[] encoded(DataElement element) { + protected byte[] positionHistory(byte[]... positions) { + return HLAEncodingTestSupport.variableArray(positions); + } + + protected byte[] encoded(DataElement element) { try { return element.toByteArray(); } catch (EncoderException e) { @@ -202,18 +276,18 @@ private byte[] encoded(DataElement element) { } } - private long count(ObjectCache cache, String sql) throws SQLException { + protected long count(ObjectCache cache, String sql) throws SQLException { return scalarLong(cache, sql); } - private long scalarLong(ObjectCache cache, String sql) throws SQLException { + protected long scalarLong(ObjectCache cache, String sql) throws SQLException { try (PreparedStatement statement = cache.connection().prepareStatement(sql); ResultSet resultSet = statement.executeQuery()) { return resultSet.next() ? resultSet.getLong(1) : 0L; } } - private byte[] rawBytes(ObjectCache cache, String pathKey) throws SQLException { + protected byte[] rawBytes(ObjectCache cache, String pathKey) throws SQLException { String sql = """ SELECT c.raw_bytes FROM object_attribute_current c @@ -227,4 +301,29 @@ private byte[] rawBytes(ObjectCache cache, String pathKey) throws SQLException { } } } + + private long currentValueCount(ObjectCache cache, String attributeName) throws SQLException { + String sql = """ + SELECT COUNT(*) + FROM object_attribute_current c + JOIN fom_attribute a ON a.id = c.attribute_id + WHERE a.attribute_name = ? + """; + try (PreparedStatement statement = cache.connection().prepareStatement(sql)) { + statement.setString(1, attributeName); + try (ResultSet resultSet = statement.executeQuery()) { + return resultSet.next() ? resultSet.getLong(1) : 0L; + } + } + } + + private long attributeId(ObjectCache cache, String pathKey) throws SQLException { + try (PreparedStatement statement = cache.connection() + .prepareStatement("SELECT id FROM fom_attribute WHERE path_key = ?")) { + statement.setString(1, pathKey); + try (ResultSet resultSet = statement.executeQuery()) { + return resultSet.next() ? resultSet.getLong(1) : 0L; + } + } + } } diff --git a/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCacheTest.java b/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCacheTest.java index 1f152d9..9e08020 100644 --- a/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCacheTest.java +++ b/src/test/java/com/yetanalytics/hlaxapi/cache/ObjectCacheTest.java @@ -8,9 +8,9 @@ import com.yetanalytics.hlaxapi.HLADecoderRegistry; import com.yetanalytics.hlaxapi.SimulationConfig; import com.yetanalytics.hlaxapi.config.XapiConfig; -import com.yetanalytics.hlaxapi.config.model.ObjectCacheConfig; import com.yetanalytics.hlaxapi.config.model.ComparisonOperator; import com.yetanalytics.hlaxapi.config.model.Criterion; +import com.yetanalytics.hlaxapi.config.model.ObjectCacheConfig; import com.yetanalytics.hlaxapi.config.model.StatementTrigger; import com.yetanalytics.hlaxapi.config.model.Target; import com.yetanalytics.hlaxapi.config.model.TrackedObject; @@ -57,6 +57,13 @@ void disabledWhenNoQueryInjectionsAndDoesNotOpenSqlite(@TempDir Path tempDir) { } } + @Test + void disabledCacheDoesNotRequireConnectionSettings() { + try (ObjectCache cache = new ObjectCache(new XapiConfig(), catalog, fomXml, decoderRegistry)) { + assertFalse(cache.isEnabled()); + } + } + @Test void enabledWhenQueryInjectionsExistAndCanQueryReflectedValues(@TempDir Path tempDir) { Path databasePath = tempDir.resolve("enabled.sqlite"); diff --git a/src/test/java/com/yetanalytics/hlaxapi/cache/PostgresqlObjectCachePersistenceTest.java b/src/test/java/com/yetanalytics/hlaxapi/cache/PostgresqlObjectCachePersistenceTest.java new file mode 100644 index 0000000..89a9962 --- /dev/null +++ b/src/test/java/com/yetanalytics/hlaxapi/cache/PostgresqlObjectCachePersistenceTest.java @@ -0,0 +1,85 @@ +package com.yetanalytics.hlaxapi.cache; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.yetanalytics.hlaxapi.FOMXML; +import com.yetanalytics.hlaxapi.config.XapiConfig; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.postgresql.PostgreSQLContainer; + +@Testcontainers +final class PostgresqlObjectCachePersistenceTest extends ObjectCachePersistenceTest { + + @Container + private static final PostgreSQLContainer POSTGRESQL = new PostgreSQLContainer("postgres:17") + .withDatabaseName("hla_xapi_test") + .withUsername("hla_xapi") + .withPassword("hla_xapi_test") + .withTmpFs(Map.of("/var/lib/postgresql/data", "rw")); + + @Override + protected ObjectCache newCache( + String name, + XapiConfig config, + FomCatalog cacheCatalog, + FOMXML cacheFomXml) { + String schema = "hla_object_cache_test_" + name.replace('-', '_'); + ObjectCacheConnectionSettings settings = ObjectCacheConnectionSettings.postgresql( + POSTGRESQL.getJdbcUrl(), + POSTGRESQL.getUsername(), + POSTGRESQL.getPassword(), + schema); + return new ObjectCache(config, cacheCatalog, cacheFomXml, decoderRegistry, settings); + } + + @Test + void isolatesTablesInConfiguredSchemaAndUsesPostgresqlBinaryType() throws SQLException { + try (ObjectCache cache = newCache("schema-types")) { + assertEquals("hla_object_cache_test_schema_types", scalarString(cache, "SELECT current_schema()")); + assertEquals( + "bytea", + scalarString(cache, """ + SELECT data_type + FROM information_schema.columns + WHERE table_schema = current_schema() + AND table_name = 'object_attribute_current' + AND column_name = 'value_blob' + """)); + } + } + + @Test + void synchronizesAttributeIdentityAfterExplicitFomIds() throws SQLException { + try (ObjectCache cache = newCache("identity")) { + long seededMaximum = scalarLong(cache, "SELECT MAX(id) FROM fom_attribute"); + int rabbitClassId = catalog.objectClass("Rabbit").orElseThrow().id(); + try (PreparedStatement statement = cache.connection().prepareStatement(""" + INSERT INTO fom_attribute + (class_id, attribute_name, path_key, data_type, primitive_type, is_leaf) + VALUES (?, 'Synthetic', 'Synthetic[0]', 'HLAinteger32BE', 'HLAinteger32BE', 1) + """)) { + statement.setInt(1, rabbitClassId); + statement.executeUpdate(); + } + + assertTrue(scalarLong( + cache, + "SELECT id FROM fom_attribute WHERE path_key = 'Synthetic[0]'") + > seededMaximum); + } + } + + private String scalarString(ObjectCache cache, String sql) throws SQLException { + try (PreparedStatement statement = cache.connection().prepareStatement(sql); + ResultSet resultSet = statement.executeQuery()) { + return resultSet.next() ? resultSet.getString(1) : null; + } + } +} diff --git a/src/test/java/com/yetanalytics/hlaxapi/cache/SqliteObjectCachePersistenceTest.java b/src/test/java/com/yetanalytics/hlaxapi/cache/SqliteObjectCachePersistenceTest.java new file mode 100644 index 0000000..c80701d --- /dev/null +++ b/src/test/java/com/yetanalytics/hlaxapi/cache/SqliteObjectCachePersistenceTest.java @@ -0,0 +1,38 @@ +package com.yetanalytics.hlaxapi.cache; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import com.yetanalytics.hlaxapi.FOMXML; +import com.yetanalytics.hlaxapi.config.XapiConfig; +import java.nio.file.Path; +import java.sql.SQLException; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +final class SqliteObjectCachePersistenceTest extends ObjectCachePersistenceTest { + + @TempDir + private Path tempDir; + + @Override + protected ObjectCache newCache( + String name, + XapiConfig config, + FomCatalog cacheCatalog, + FOMXML cacheFomXml) { + return new ObjectCache( + config, + cacheCatalog, + cacheFomXml, + decoderRegistry, + "jdbc:sqlite:" + tempDir.resolve(name + ".sqlite")); + } + + @Test + void configuresSqliteSchemaVersionAndForeignKeys() throws SQLException { + try (ObjectCache cache = newCache()) { + assertEquals(1, scalarLong(cache, "PRAGMA user_version")); + assertEquals(1, scalarLong(cache, "PRAGMA foreign_keys")); + } + } +} diff --git a/src/test/resources/config/ObjectCacheTestFOM.xml b/src/test/resources/config/ObjectCacheTestFOM.xml new file mode 100644 index 0000000..a57fb93 --- /dev/null +++ b/src/test/resources/config/ObjectCacheTestFOM.xml @@ -0,0 +1,51 @@ + + + + + HLAobjectRoot + + Rabbit + + EntityId + HLAASCIIstring + + + Hunger + HLAinteger32BE + + + Position + GridPosition + + + PositionHistory + GridPositionHistory + + + + + + + + GridPositionHistory + GridPosition + Dynamic + HLAvariableArray + + + + + GridPosition + HLAfixedRecord + + X + HLAinteger32BE + + + Y + HLAinteger32BE + + + + + diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml new file mode 100644 index 0000000..68b0254 --- /dev/null +++ b/src/test/resources/logback-test.xml @@ -0,0 +1,16 @@ + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + + +