From e8b2d77915980c437ceb2cf59c0389ab733ab363 Mon Sep 17 00:00:00 2001 From: davidfrigolet Date: Tue, 15 Sep 2026 17:01:43 +0100 Subject: [PATCH] refactor(sql,core): centralize journal persistence, dedupe SQL journal code Extracts SQL-specific Journal Event DDL/DML generation into the sibling flamingock-sql-util module, deletes the "flamingockJournalEvents" literal duplicated three times (SQL, DynamoDB, MongoDB stores) in favor of the new CommunityPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME, and centralizes the per-stage journal initialization each community audit store was reimplementing. - SqlAuditStore/DynamoDBAuditStore/MongoDBSyncAuditStore no longer carry local copies of SqlJournalDialectHelper/JournalEventConstants or the default journal-store-name literal; they consume the shared versions from flamingock-sql-util and flamingock-general-util instead - JournalEventSequencerFactory (core) gains initializeForStage(stageId, autoCreate): feature-flag-gates, initializes the journal store, and returns the stage sequencer in one call, replacing the near-duplicated three-way copy of this logic - JournalEventStore.initialize(boolean) is now part of the interface contract rather than an implementation-specific method - fixes MongoDBSyncAuditStore never calling journalEventStore.initialize() per stage: its initialize method was protected in a different package and unreachable from the store, so the call was silently skipped; it is now public and wired through the new centralized init path - bumps generalUtilVersion/sqlVersion to the versions carrying the moved code (pending release of the two companion PRs below) --- build.gradle.kts | 4 +- .../store/dynamodb/DynamoDBAuditStore.java | 10 +- .../internal/DynamoDBJournalEventStore.java | 1 + .../mongodb/sync/MongoDBSyncAuditStore.java | 4 +- .../MongoDBSyncJournalEventStore.java | 3 +- .../flamingock/store/sql/SqlAuditStore.java | 46 +-- .../store/sql/internal/AuditEntryMapper.java | 13 +- .../sql/internal/JournalEventConstants.java | 63 ---- .../sql/internal/SqlAuditRepository.java | 5 +- .../sql/internal/SqlJournalDialectHelper.java | 329 ------------------ .../sql/internal/SqlJournalEventMapper.java | 16 +- .../sql/internal/SqlJournalEventStore.java | 15 +- .../internal/SqlJournalDialectHelperTest.java | 253 -------------- .../internal/SqlJournalEventMapperTest.java | 1 + .../SqlJournalEventStoreJdbcTest.java | 1 + .../journal/JournalEventSequencerFactory.java | 33 +- .../core/journal/JournalEventStore.java | 6 + .../journal/JournalEventFieldConstants.java | 2 - .../JournalEventPersistenceConstants.java | 28 -- 19 files changed, 86 insertions(+), 747 deletions(-) delete mode 100644 community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/JournalEventConstants.java delete mode 100644 community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalDialectHelper.java delete mode 100644 community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalDialectHelperTest.java delete mode 100644 utils/mongodb-util/src/main/java/io/flamingock/internal/common/mongodb/journal/JournalEventPersistenceConstants.java diff --git a/build.gradle.kts b/build.gradle.kts index f5e411b56..c3f67f956 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -22,10 +22,10 @@ allprojects { val declaredVersion = "1.5.0-SNAPSHOT" version = VersionManager.resolveVersion(declaredVersion, project.hasProperty("release")) - extra["generalUtilVersion"] = "1.6.0" + extra["generalUtilVersion"] = "1.7.0-SNAPSHOT" extra["templateApiVersion"] = "1.3.4" extra["coreApiVersion"] = "1.3.3" - extra["sqlVersion"] = "1.3.2" + extra["sqlVersion"] = "1.4.0-SNAPSHOT" extra["mongodbTemplateVersion"] = "1.3.2" repositories { diff --git a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/DynamoDBAuditStore.java b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/DynamoDBAuditStore.java index 4474d68ab..58c2bfc23 100644 --- a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/DynamoDBAuditStore.java +++ b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/DynamoDBAuditStore.java @@ -19,7 +19,6 @@ import io.flamingock.internal.common.core.audit.AuditReader; import io.flamingock.internal.common.core.context.ContextResolver; import io.flamingock.internal.common.core.error.FlamingockException; -import io.flamingock.internal.common.core.feature.Features; import io.flamingock.internal.core.configuration.community.CommunityConfigurable; import io.flamingock.internal.core.external.store.CommunityAuditStore; import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence; @@ -27,10 +26,8 @@ import io.flamingock.internal.core.journal.JournalEventSequencer; import io.flamingock.internal.core.journal.JournalEventSequencerFactory; import io.flamingock.internal.util.Constants; -import io.flamingock.internal.util.FeatureFlag; import io.flamingock.internal.util.TimeService; import io.flamingock.internal.util.constants.CommunityPersistenceConstants; -import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventFieldConstants; import io.flamingock.internal.util.id.RunnerId; import io.flamingock.store.dynamodb.internal.DynamoDBAuditPersistence; import io.flamingock.store.dynamodb.internal.DynamoDBAuditRepository; @@ -50,7 +47,7 @@ public class DynamoDBAuditStore implements CommunityAuditStore { private final DynamoDbClient client; private String auditRepositoryName = CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME; private String lockRepositoryName = CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME; - private String journalRepositoryName = JournalEventFieldConstants.DEFAULT_JOURNAL_REPOSITORY_NAME; + private String journalRepositoryName = CommunityPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME; private long readCapacityUnits = 5L; private long writeCapacityUnits = 5L; private boolean autoCreate = true; @@ -139,10 +136,7 @@ public void initialize(ContextResolver baseContext) { public AuditPersistenceFactory getPersistenceFactory() { return stageId -> { auditRepository.initialize(autoCreate); - if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false)) { - journalEventStore.initialize(autoCreate); - } - JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId); + JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.initializeForStage(stageId, autoCreate); persistence = new DynamoDBAuditPersistence( communityConfiguration, auditRepository, diff --git a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStore.java b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStore.java index 4a1609c22..33da16c4c 100644 --- a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStore.java +++ b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStore.java @@ -114,6 +114,7 @@ public DynamoDBJournalEventStore(DynamoDbClient client, * * @param autoCreate whether to create the table when missing */ + @Override public synchronized void initialize(boolean autoCreate) { if (!isJournalEventsEnabled() || table != null) { return; diff --git a/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/MongoDBSyncAuditStore.java b/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/MongoDBSyncAuditStore.java index ac7fcd71e..cca321278 100644 --- a/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/MongoDBSyncAuditStore.java +++ b/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/MongoDBSyncAuditStore.java @@ -45,9 +45,9 @@ import java.util.List; import java.util.Set; -import static io.flamingock.internal.common.mongodb.journal.JournalEventPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME; import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME; import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME; +import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME; public class MongoDBSyncAuditStore implements CommunityAuditStore { @@ -152,7 +152,7 @@ public void initialize(ContextResolver baseContext) { @Override public AuditPersistenceFactory getPersistenceFactory() { return stageId -> { - JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId); + JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.initializeForStage(stageId, autoCreate); persistence = new MongoDBSyncAuditPersistence( communityConfiguration, auditRepository, diff --git a/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncJournalEventStore.java b/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncJournalEventStore.java index 6f42ba3bd..3433fc5a2 100644 --- a/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncJournalEventStore.java +++ b/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncJournalEventStore.java @@ -84,7 +84,8 @@ public MongoDBSyncJournalEventStore(MongoDatabase database, .withWriteConcern(writeConcern); } - protected void initialize(boolean autoCreate) { + @Override + public void initialize(boolean autoCreate) { CollectionInitializator initializer = new CollectionInitializator<>( new MongoDBSyncCollectionHelper(collection), () -> new MongoDBDocumentHelper(new Document()), diff --git a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/SqlAuditStore.java b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/SqlAuditStore.java index a1754fd1e..dd7145942 100644 --- a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/SqlAuditStore.java +++ b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/SqlAuditStore.java @@ -19,15 +19,14 @@ import io.flamingock.internal.common.core.audit.AuditReader; import io.flamingock.internal.common.core.context.ContextResolver; import io.flamingock.internal.common.core.error.FlamingockException; -import io.flamingock.internal.common.core.feature.Features; import io.flamingock.internal.core.configuration.community.CommunityConfigurable; import io.flamingock.internal.core.external.store.CommunityAuditStore; import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence; import io.flamingock.internal.core.external.store.lock.community.CommunityLockService; import io.flamingock.internal.core.journal.JournalEventSequencer; import io.flamingock.internal.core.journal.JournalEventSequencerFactory; +import io.flamingock.internal.common.sql.journal.SqlJournalConstants; import io.flamingock.internal.util.Constants; -import io.flamingock.internal.util.FeatureFlag; import io.flamingock.internal.util.constants.CommunityPersistenceConstants; import io.flamingock.internal.util.id.RunnerId; import io.flamingock.store.sql.internal.SqlAuditPersistence; @@ -40,9 +39,6 @@ public class SqlAuditStore implements CommunityAuditStore { - private static final String SQL_IDENTIFIER_PATTERN = "[A-Za-z][A-Za-z0-9_]*"; - private static final String DEFAULT_JOURNAL_REPOSITORY_NAME = "flamingockJournalEvents"; - private final SqlExternalSystem targetSystem; private final DataSource dataSource; private CommunityConfigurable communityConfiguration; @@ -53,7 +49,7 @@ public class SqlAuditStore implements CommunityAuditStore { private SqlAuditRepository auditRepository; private String auditRepositoryName = CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME; private String lockRepositoryName = CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME; - private String journalRepositoryName = DEFAULT_JOURNAL_REPOSITORY_NAME; + private String journalRepositoryName = CommunityPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME; private boolean autoCreate = true; private SqlAuditStore(SqlExternalSystem targetSystem) { @@ -120,12 +116,8 @@ public void initialize(ContextResolver baseContext) { @Override public AuditPersistenceFactory getPersistenceFactory() { return stageId -> { - boolean journalEventsEnabled = isJournalEventsEnabled(); - JournalEventSequencer journalEventSequencer = null; - if (journalEventsEnabled) { - journalEventStore.initialize(autoCreate); - journalEventSequencer = journalEventSequencerFactory.forStream(stageId); - } + JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.initializeForStage(stageId, autoCreate); + boolean journalEventsEnabled = journalEventSequencer != null; SqlAuditPersistence persistence = new SqlAuditPersistence( communityConfiguration, @@ -157,31 +149,25 @@ private void validate() { validateRepositoryName(auditRepositoryName, "auditRepositoryName"); validateRepositoryName(lockRepositoryName, "lockRepositoryName"); validateRepositoryName(journalRepositoryName, "journalRepositoryName"); - if (auditRepositoryName.trim().equalsIgnoreCase(lockRepositoryName.trim())) { - throw new FlamingockException("The 'auditRepositoryName' and 'lockRepositoryName' properties must not be the same."); - } - if (journalRepositoryName.trim().equalsIgnoreCase(auditRepositoryName.trim())) { - throw new FlamingockException("The 'journalRepositoryName' and 'auditRepositoryName' properties must not be the same."); - } - if (journalRepositoryName.trim().equalsIgnoreCase(lockRepositoryName.trim())) { - throw new FlamingockException("The 'journalRepositoryName' and 'lockRepositoryName' properties must not be the same."); - } + validateDistinct(auditRepositoryName, "auditRepositoryName", lockRepositoryName, "lockRepositoryName"); + validateDistinct(journalRepositoryName, "journalRepositoryName", auditRepositoryName, "auditRepositoryName"); + validateDistinct(journalRepositoryName, "journalRepositoryName", lockRepositoryName, "lockRepositoryName"); } private void validateRepositoryName(String repositoryName, String propertyName) { - if (repositoryName == null || repositoryName.trim().isEmpty()) { - throw new FlamingockException(propertyName + " must not be blank"); - } - if (!repositoryName.matches(SQL_IDENTIFIER_PATTERN)) { - throw new FlamingockException(propertyName + " must be a simple SQL identifier"); + try { + SqlJournalConstants.validateIdentifier(repositoryName, propertyName); + } catch (IllegalArgumentException exception) { + throw new FlamingockException(exception.getMessage()); } } - private static boolean isJournalEventsEnabled() { + private void validateDistinct(String firstName, String firstField, String secondName, String secondField) { try { - return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false); - } catch (RuntimeException exception) { - return false; + SqlJournalConstants.validateDistinct(firstName, firstField, secondName, secondField); + } catch (IllegalArgumentException exception) { + throw new FlamingockException("The '" + firstField + "' and '" + secondField + + "' properties must not be the same."); } } } diff --git a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/AuditEntryMapper.java b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/AuditEntryMapper.java index 45d3af6a7..af00b5313 100644 --- a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/AuditEntryMapper.java +++ b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/AuditEntryMapper.java @@ -18,14 +18,13 @@ import io.flamingock.api.RecoveryStrategy; import io.flamingock.internal.common.core.audit.AuditEntry; import io.flamingock.internal.common.core.audit.AuditTxType; +import io.flamingock.internal.common.sql.journal.SqlAuditColumnNames; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Timestamp; import java.sql.Types; -import java.util.Arrays; -import java.util.Collections; import java.util.List; /** @@ -33,17 +32,11 @@ */ final class AuditEntryMapper { - private static final List COLUMN_NAMES = Collections.unmodifiableList(Arrays.asList( - "execution_id", "stage_id", "change_id", "author", "created_at", "state", "invoked_class", - "invoked_method", "source_file", "metadata", "execution_millis", "execution_hostname", - "error_trace", "type", "tx_strategy", "target_system_id", "change_order", "recovery_strategy", - "transaction_flag", "system_change")); - private AuditEntryMapper() { } static List columnNames() { - return COLUMN_NAMES; + return SqlAuditColumnNames.columnNames(); } static void bind(PreparedStatement statement, AuditEntry auditEntry, int firstColumn) throws SQLException { @@ -107,7 +100,7 @@ static AuditEntry fromResultSet(ResultSet resultSet) throws SQLException { } private static String columnName(int index) { - return COLUMN_NAMES.get(index); + return SqlAuditColumnNames.columnNames().get(index); } private static void setNullableBoolean(PreparedStatement statement, diff --git a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/JournalEventConstants.java b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/JournalEventConstants.java deleted file mode 100644 index 6cd07ee6b..000000000 --- a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/JournalEventConstants.java +++ /dev/null @@ -1,63 +0,0 @@ -/* - * Copyright 2026 Flamingock (https://www.flamingock.io) - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.flamingock.store.sql.internal; - -/** - * SQL names used by the relational Journal Event store. - */ -final class JournalEventConstants { - - static final String EVENT_ID = "event_id"; - static final String EVENT_TYPE = "event_type"; - static final String EVENT_VERSION = "event_version"; - static final String STREAM_ID = "stream_id"; - static final String STREAM_SEQUENCE = "stream_sequence"; - static final String OCCURRED_AT = "occurred_at"; - static final String ACKNOWLEDGED = "acknowledged"; - - static final String PENDING_EVENTS_INDEX = "pending_events"; - static final String EVENT_ID_INDEX = "event_id"; - - private JournalEventConstants() { - } - - /** - * Validates a configured SQL identifier before it is interpolated into DDL or DML. - * - * @param value identifier to validate - * @param fieldName configuration field containing the identifier - */ - static void validateIdentifier(String value, String fieldName) { - if (value == null || value.trim().isEmpty()) { - throw new IllegalArgumentException(fieldName + " must not be blank"); - } - if (!value.matches("[A-Za-z][A-Za-z0-9_]*")) { - throw new IllegalArgumentException(fieldName + " must be a simple SQL identifier"); - } - } - - /** - * Ensures that two configured SQL resources cannot address the same table. - */ - static void validateDistinct(String firstName, - String firstField, - String secondName, - String secondField) { - if (firstName.trim().equalsIgnoreCase(secondName.trim())) { - throw new IllegalArgumentException(firstField + " and " + secondField + " must not be the same"); - } - } -} diff --git a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlAuditRepository.java b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlAuditRepository.java index 8fd977b3c..01df11d9c 100644 --- a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlAuditRepository.java +++ b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlAuditRepository.java @@ -18,6 +18,7 @@ import io.flamingock.internal.common.core.audit.AuditEntry; import io.flamingock.internal.common.sql.SqlDialect; import io.flamingock.internal.common.sql.dialectHelpers.SqlAuditorDialectHelper; +import io.flamingock.internal.common.sql.journal.SqlJournalConstants; import io.flamingock.internal.util.Result; import javax.sql.DataSource; @@ -35,7 +36,7 @@ public SqlAuditRepository(DataSource dataSource, String auditTableName) { if (dataSource == null) { throw new IllegalArgumentException("dataSource must not be null"); } - JournalEventConstants.validateIdentifier(auditTableName, "auditTableName"); + SqlJournalConstants.validateIdentifier(auditTableName, "auditTableName"); this.dataSource = dataSource; this.auditTableName = auditTableName; } @@ -113,7 +114,7 @@ Result save(Connection connection, AuditEntry auditEntry) { if (auditEntry == null) { throw new IllegalArgumentException("auditEntry must not be null"); } - JournalEventConstants.validateIdentifier(auditTableName, "auditTableName"); + SqlJournalConstants.validateIdentifier(auditTableName, "auditTableName"); if (auditEntry.getChangeId() == null || auditEntry.getChangeId().trim().isEmpty()) { throw new IllegalArgumentException("changeId must not be blank"); } diff --git a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalDialectHelper.java b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalDialectHelper.java deleted file mode 100644 index a61d0fa5c..000000000 --- a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalDialectHelper.java +++ /dev/null @@ -1,329 +0,0 @@ -/* - * Copyright 2026 Flamingock (https://www.flamingock.io) - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.flamingock.store.sql.internal; - -import io.flamingock.internal.common.sql.SqlDialect; - -import java.nio.charset.StandardCharsets; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import java.sql.Types; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; - -/** - * Provides portable Journal Event SQL without relying on vendor-specific upsert or pagination syntax. - */ -public final class SqlJournalDialectHelper { - - private static final String INDEX_PREFIX = "idx_"; - private static final int INDEX_HASH_LENGTH = 8; - - private final SqlDialect sqlDialect; - - public SqlJournalDialectHelper(SqlDialect sqlDialect) { - if (sqlDialect == null) { - throw new IllegalArgumentException("sqlDialect must not be null"); - } - this.sqlDialect = sqlDialect; - } - - public SqlDialect getSqlDialect() { - return sqlDialect; - } - - int getMaximumIndexNameLength() { - switch (sqlDialect) { - case ORACLE: - return 30; - case POSTGRESQL: - return 63; - default: - return 128; - } - } - - List getIndexNames(String tableName) { - JournalEventConstants.validateIdentifier(tableName, "tableName"); - return Collections.unmodifiableList(Arrays.asList( - indexName(tableName, JournalEventConstants.PENDING_EVENTS_INDEX), - indexName(tableName, JournalEventConstants.EVENT_ID_INDEX))); - } - - List getColumnDefinitions() { - List auditColumnNames = AuditEntryMapper.columnNames(); - return Collections.unmodifiableList(Arrays.asList( - new ColumnDefinition(JournalEventConstants.EVENT_ID, ColumnType.VARCHAR, 255, false), - new ColumnDefinition(JournalEventConstants.EVENT_TYPE, ColumnType.VARCHAR, 32, false), - new ColumnDefinition(JournalEventConstants.EVENT_VERSION, ColumnType.INTEGER, 0, false), - new ColumnDefinition(JournalEventConstants.STREAM_ID, ColumnType.VARCHAR, 255, false), - new ColumnDefinition(JournalEventConstants.STREAM_SEQUENCE, ColumnType.LONG, 19, false), - new ColumnDefinition(JournalEventConstants.OCCURRED_AT, ColumnType.TIMESTAMP, 0, false), - new ColumnDefinition(JournalEventConstants.ACKNOWLEDGED, ColumnType.BOOLEAN, 0, false), - new ColumnDefinition(auditColumnNames.get(0), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(1), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(2), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(3), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(4), ColumnType.TIMESTAMP, 0, true), - new ColumnDefinition(auditColumnNames.get(5), ColumnType.VARCHAR, 64, true), - new ColumnDefinition(auditColumnNames.get(6), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(7), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(8), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(9), ColumnType.TEXT, 2048, true), - new ColumnDefinition(auditColumnNames.get(10), ColumnType.LONG, 19, true), - new ColumnDefinition(auditColumnNames.get(11), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(12), ColumnType.TEXT, 2048, true), - new ColumnDefinition(auditColumnNames.get(13), ColumnType.VARCHAR, 64, true), - new ColumnDefinition(auditColumnNames.get(14), ColumnType.VARCHAR, 64, true), - new ColumnDefinition(auditColumnNames.get(15), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(16), ColumnType.VARCHAR, 255, true), - new ColumnDefinition(auditColumnNames.get(17), ColumnType.VARCHAR, 64, true), - new ColumnDefinition(auditColumnNames.get(18), ColumnType.BOOLEAN, 0, true), - new ColumnDefinition(auditColumnNames.get(19), ColumnType.BOOLEAN, 0, true))); - } - - int getBooleanJdbcType() { - switch (sqlDialect) { - case MYSQL: - case MARIADB: - return Types.TINYINT; - case POSTGRESQL: - case H2: - case FIREBIRD: - case INFORMIX: - return Types.BOOLEAN; - case SQLITE: - return Types.INTEGER; - case SQLSERVER: - case SYBASE: - return Types.BIT; - case ORACLE: - return Types.NUMERIC; - case DB2: - default: - return Types.SMALLINT; - } - } - - public String getCreateTableSqlString(String tableName) { - JournalEventConstants.validateIdentifier(tableName, "tableName"); - StringBuilder sql = new StringBuilder("CREATE TABLE ") - .append(tableName) - .append(" ("); - List definitions = getColumnDefinitions(); - for (int i = 0; i < definitions.size(); i++) { - if (i > 0) { - sql.append(", "); - } - ColumnDefinition definition = definitions.get(i); - sql.append(definition.name) - .append(' ') - .append(sqlType(definition)); - if (!definition.nullable) { - sql.append(" NOT NULL"); - } - } - return sql.append(", PRIMARY KEY (") - .append(JournalEventConstants.STREAM_ID) - .append(", ") - .append(JournalEventConstants.STREAM_SEQUENCE) - .append(")") - .append(')') - .toString(); - } - - public List getCreateIndexSqlStrings(String tableName) { - List indexNames = getIndexNames(tableName); - return Collections.unmodifiableList(Arrays.asList( - String.format("CREATE INDEX %s ON %s (%s, %s, %s)", - indexNames.get(0), tableName, JournalEventConstants.ACKNOWLEDGED, - JournalEventConstants.STREAM_ID, JournalEventConstants.STREAM_SEQUENCE), - String.format("CREATE INDEX %s ON %s (%s)", - indexNames.get(1), tableName, JournalEventConstants.EVENT_ID))); - } - - public String getInsertSqlString(String tableName) { - JournalEventConstants.validateIdentifier(tableName, "tableName"); - StringBuilder columns = new StringBuilder(); - StringBuilder placeholders = new StringBuilder(); - for (ColumnDefinition definition : getColumnDefinitions()) { - if (columns.length() > 0) { - columns.append(", "); - placeholders.append(", "); - } - columns.append(definition.name); - placeholders.append("?"); - } - return String.format("INSERT INTO %s (%s) VALUES (%s)", tableName, columns, placeholders); - } - - private String sqlType(ColumnDefinition definition) { - switch (definition.type) { - case VARCHAR: - return getVarcharType(definition.size); - case INTEGER: - return "INTEGER"; - case LONG: - return getLongType(); - case TIMESTAMP: - return getTimestampType(); - case BOOLEAN: - return getBooleanType(); - case TEXT: - return getTextType(); - default: - throw new IllegalArgumentException("Unsupported Journal column type: " + definition.type); - } - } - - public String getLastEventSqlString(String tableName) { - JournalEventConstants.validateIdentifier(tableName, "tableName"); - return String.format( - "SELECT * FROM %s WHERE stream_id = ? ORDER BY stream_sequence DESC", - tableName); - } - - public String getUnacknowledgedEventsSqlString(String tableName) { - JournalEventConstants.validateIdentifier(tableName, "tableName"); - return String.format( - "SELECT * FROM %s WHERE acknowledged = ? ORDER BY stream_id ASC, stream_sequence ASC", - tableName); - } - - public String getAcknowledgeSqlString(String tableName) { - JournalEventConstants.validateIdentifier(tableName, "tableName"); - return String.format( - "UPDATE %s SET acknowledged = ? WHERE event_id = ? AND acknowledged = ?", - tableName); - } - - private String indexName(String tableName, String suffix) { - String naturalName = INDEX_PREFIX + tableName + "_" + suffix; - int maximumLength = getMaximumIndexNameLength(); - if (naturalName.length() <= maximumLength) { - return naturalName; - } - - String hash = hash(tableName); - int tableLength = maximumLength - INDEX_PREFIX.length() - suffix.length() - hash.length() - 2; - if (tableLength < 1) { - throw new IllegalArgumentException("Table name cannot produce a valid SQL index name"); - } - return INDEX_PREFIX + tableName.substring(0, tableLength) + "_" + suffix + "_" + hash; - } - - private static String hash(String value) { - try { - byte[] digest = MessageDigest.getInstance("SHA-256") - .digest(value.getBytes(StandardCharsets.UTF_8)); - StringBuilder result = new StringBuilder(INDEX_HASH_LENGTH); - for (int i = 0; i < INDEX_HASH_LENGTH / 2; i++) { - result.append(String.format("%02x", digest[i])); - } - return result.toString(); - } catch (NoSuchAlgorithmException exception) { - throw new IllegalStateException("SHA-256 is unavailable", exception); - } - } - - private String getVarcharType(int length) { - return (sqlDialect == SqlDialect.ORACLE ? "VARCHAR2(" : "VARCHAR(") + length + ")"; - } - - private String getLongType() { - return sqlDialect == SqlDialect.ORACLE ? "NUMBER(19)" : "BIGINT"; - } - - private String getTimestampType() { - switch (sqlDialect) { - case SQLSERVER: - case SYBASE: - return "DATETIME"; - case INFORMIX: - return "DATETIME YEAR TO FRACTION(3)"; - default: - return "TIMESTAMP"; - } - } - - private String getTextType() { - switch (sqlDialect) { - case MYSQL: - case MARIADB: - case POSTGRESQL: - case SQLSERVER: - case SYBASE: - case SQLITE: - return "TEXT"; - case INFORMIX: - return "LVARCHAR(2048)"; - case ORACLE: - return "VARCHAR2(4000)"; - case DB2: - case FIREBIRD: - case H2: - default: - return "VARCHAR(4000)"; - } - } - - private String getBooleanType() { - switch (sqlDialect) { - case MYSQL: - case MARIADB: - return "TINYINT(1)"; - case POSTGRESQL: - case H2: - case FIREBIRD: - case INFORMIX: - return "BOOLEAN"; - case SQLITE: - return "INTEGER"; - case SQLSERVER: - case SYBASE: - return "BIT"; - case ORACLE: - return "NUMBER(1)"; - case DB2: - default: - return "SMALLINT"; - } - } - - enum ColumnType { - VARCHAR, - INTEGER, - LONG, - TIMESTAMP, - BOOLEAN, - TEXT - } - - static final class ColumnDefinition { - final String name; - final ColumnType type; - final int size; - final boolean nullable; - - ColumnDefinition(String name, ColumnType type, int size, boolean nullable) { - this.name = name; - this.type = type; - this.size = size; - this.nullable = nullable; - } - } -} diff --git a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalEventMapper.java b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalEventMapper.java index 392088b99..4eeef10aa 100644 --- a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalEventMapper.java +++ b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalEventMapper.java @@ -19,6 +19,8 @@ import io.flamingock.internal.common.core.journal.JournalEvent; import io.flamingock.internal.common.core.journal.JournalEventType; import io.flamingock.internal.common.sql.SqlDialect; +import io.flamingock.internal.common.sql.journal.SqlJournalConstants; +import io.flamingock.internal.common.sql.dialectHelpers.SqlJournalDialectHelper; import java.sql.PreparedStatement; import java.sql.ResultSet; @@ -54,25 +56,25 @@ void bind(PreparedStatement statement, JournalEvent event) throws SQ JournalEvent fromResultSet(ResultSet resultSet) throws SQLException { JournalEventType eventType = JournalEventType.valueOf( - resultSet.getString(JournalEventConstants.EVENT_TYPE)); + resultSet.getString(SqlJournalConstants.EVENT_TYPE)); if (eventType != JournalEventType.CHANGE_STATE) { throw new UnsupportedOperationException("Unsupported SQL Journal Event type: " + eventType); } - Timestamp occurredAt = resultSet.getTimestamp(JournalEventConstants.OCCURRED_AT); + Timestamp occurredAt = resultSet.getTimestamp(SqlJournalConstants.OCCURRED_AT); if (occurredAt == null) { throw new SQLException("Journal event occurred_at must not be null"); } return new JournalEvent<>( - resultSet.getString(JournalEventConstants.EVENT_ID), + resultSet.getString(SqlJournalConstants.EVENT_ID), eventType, - resultSet.getInt(JournalEventConstants.EVENT_VERSION), - resultSet.getString(JournalEventConstants.STREAM_ID), - resultSet.getLong(JournalEventConstants.STREAM_SEQUENCE), + resultSet.getInt(SqlJournalConstants.EVENT_VERSION), + resultSet.getString(SqlJournalConstants.STREAM_ID), + resultSet.getLong(SqlJournalConstants.STREAM_SEQUENCE), occurredAt.toInstant(), AuditEntryMapper.fromResultSet(resultSet), - resultSet.getBoolean(JournalEventConstants.ACKNOWLEDGED)); + resultSet.getBoolean(SqlJournalConstants.ACKNOWLEDGED)); } private static void requireSupportedEvent(JournalEvent event) { diff --git a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalEventStore.java b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalEventStore.java index 02981c8cf..2f21204d4 100644 --- a/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalEventStore.java +++ b/community/flamingock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlJournalEventStore.java @@ -21,6 +21,8 @@ import io.flamingock.internal.common.core.transaction.TransactionWrapper; import io.flamingock.internal.common.sql.SqlDialect; import io.flamingock.internal.common.sql.SqlDialectFactory; +import io.flamingock.internal.common.sql.journal.SqlJournalConstants; +import io.flamingock.internal.common.sql.dialectHelpers.SqlJournalDialectHelper; import io.flamingock.internal.core.context.BasicRuntimeContext; import io.flamingock.internal.core.journal.JournalEventStore; @@ -70,7 +72,7 @@ public SqlJournalEventStore(DataSource dataSource, String tableName, Transaction if (dataSource == null) { throw new IllegalArgumentException("dataSource must not be null"); } - JournalEventConstants.validateIdentifier(tableName, "tableName"); + SqlJournalConstants.validateIdentifier(tableName, "tableName"); if (txWrapper == null) { throw new IllegalArgumentException("txWrapper must not be null"); } @@ -84,6 +86,7 @@ public SqlJournalEventStore(DataSource dataSource, String tableName, Transaction * * @param autoCreate whether the table and indexes may be created when missing */ + @Override public synchronized void initialize(boolean autoCreate) { try (Connection connection = dataSource.getConnection()) { dialectHelper = new SqlJournalDialectHelper(SqlDialectFactory.getSqlDialect(connection)); @@ -326,9 +329,9 @@ private void validateIndexes(DatabaseMetaData metadata) throws SQLException { Map indexes = readIndexes(metadata); List names = dialectHelper.getIndexNames(tableName); List> expectedColumns = new ArrayList<>(); - expectedColumns.add(asList(JournalEventConstants.ACKNOWLEDGED, - JournalEventConstants.STREAM_ID, JournalEventConstants.STREAM_SEQUENCE)); - expectedColumns.add(asList(JournalEventConstants.EVENT_ID)); + expectedColumns.add(asList(SqlJournalConstants.ACKNOWLEDGED, + SqlJournalConstants.STREAM_ID, SqlJournalConstants.STREAM_SEQUENCE)); + expectedColumns.add(asList(SqlJournalConstants.EVENT_ID)); for (int i = 0; i < names.size(); i++) { IndexMetadata index = indexes.get(names.get(i).toLowerCase(Locale.ROOT)); @@ -355,8 +358,8 @@ private void validatePrimaryKey(DatabaseMetaData metadata) throws SQLException { } } if (primaryKeyColumns.size() != 2 - || !JournalEventConstants.STREAM_ID.equalsIgnoreCase(primaryKeyColumns.get((short) 1)) - || !JournalEventConstants.STREAM_SEQUENCE.equalsIgnoreCase(primaryKeyColumns.get((short) 2))) { + || !SqlJournalConstants.STREAM_ID.equalsIgnoreCase(primaryKeyColumns.get((short) 1)) + || !SqlJournalConstants.STREAM_SEQUENCE.equalsIgnoreCase(primaryKeyColumns.get((short) 2))) { throw new IllegalStateException("SQL journal table '" + tableName + "' must have primary key (stream_id, stream_sequence)"); } diff --git a/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalDialectHelperTest.java b/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalDialectHelperTest.java deleted file mode 100644 index fd5581b73..000000000 --- a/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalDialectHelperTest.java +++ /dev/null @@ -1,253 +0,0 @@ -/* - * Copyright 2026 Flamingock (https://www.flamingock.io) - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.flamingock.store.sql.internal; - -import io.flamingock.internal.common.sql.SqlDialect; -import org.junit.jupiter.api.DisplayName; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.EnumSource; - -import java.lang.reflect.Modifier; -import java.util.Arrays; -import java.util.List; -import java.util.Locale; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; - -class SqlJournalDialectHelperTest { - - private static final String TABLE_NAME = "flamingockJournalEvents"; - - @ParameterizedTest(name = "{0} journal schema is typed and portable") - @EnumSource(SqlDialect.class) - @DisplayName("generates the journal schema and indexes for every supported SQL dialect") - void generatesTypedSchemaAndIndexesForEveryDialect(SqlDialect dialect) { - SqlJournalDialectHelper helper = new SqlJournalDialectHelper(dialect); - String ddl = helper.getCreateTableSqlString(TABLE_NAME).toUpperCase(); - List indexSql = helper.getCreateIndexSqlStrings(TABLE_NAME); - - assertTrue(ddl.contains("EVENT_ID")); - assertTrue(ddl.contains("EVENT_TYPE")); - assertTrue(ddl.contains("EVENT_VERSION")); - assertTrue(ddl.contains("STREAM_ID")); - assertTrue(ddl.contains("STREAM_SEQUENCE")); - assertTrue(ddl.contains("OCCURRED_AT")); - assertTrue(ddl.contains("ACKNOWLEDGED")); - assertTrue(ddl.contains("CREATED_AT")); - assertTrue(ddl.contains("PRIMARY KEY")); - assertFalse(ddl.contains("JSON"), "journal payloads must not use JSON columns"); - assertFalse(ddl.contains("CLOB"), "journal payloads must not use CLOB columns"); - - assertEquals(2, countOccurrences(ddl, "STREAM_ID"), - "stream_id must appear as a column and as both composite-key references"); - assertEquals(2, indexSql.size(), "pending and event-id indexes complement the composite primary key"); - assertTrue(indexSql.stream().allMatch(sql -> sql.toUpperCase().contains("CREATE INDEX"))); - assertNotNull(helper.getSqlDialect()); - assertTrue(helper.getIndexNames(TABLE_NAME).stream() - .allMatch(name -> name.length() <= helper.getMaximumIndexNameLength())); - - List definitionNames = columnNames(helper.getColumnDefinitions()); - assertTrue(Arrays.asList("event_id", "stream_id", "stream_sequence", "occurred_at", "acknowledged") - .stream().allMatch(definitionNames::contains)); - assertEquals(definitionNames, insertColumnNames(helper.getInsertSqlString(TABLE_NAME))); - } - - @Test - @DisplayName("keeps Journal schema names separate from the ordered audit payload names") - void keepsMinimalNameOwnershipBoundaries() throws Exception { - List expectedAuditColumns = Arrays.asList( - "execution_id", "stage_id", "change_id", "author", "created_at", "state", - "invoked_class", "invoked_method", "source_file", "metadata", "execution_millis", - "execution_hostname", "error_trace", "type", "tx_strategy", "target_system_id", - "change_order", "recovery_strategy", "transaction_flag", "system_change"); - - assertEquals(expectedAuditColumns, AuditEntryMapper.columnNames()); - assertEquals(20, AuditEntryMapper.columnNames().size()); - assertThrows(UnsupportedOperationException.class, - () -> AuditEntryMapper.columnNames().add("unexpected_column")); - assertFalse(Arrays.stream(AuditEntryMapper.class.getDeclaredFields()) - .anyMatch(field -> expectedAuditColumns.contains(field.getName().toLowerCase(Locale.ROOT)))); - - assertFalse(Arrays.stream(JournalEventConstants.class.getDeclaredFields()) - .anyMatch(field -> expectedAuditColumns.contains(field.getName().toLowerCase(Locale.ROOT)))); - assertFalse(Modifier.isPublic(JournalEventConstants.class.getModifiers())); - assertTrue(Modifier.isPublic(SqlJournalDialectHelper.class.getModifiers())); - assertTrue(Modifier.isFinal(SqlJournalDialectHelper.class.getModifiers())); - assertClassIsAbsent("io.flamingock.store.sql.internal.SqlAuditColumnConstants"); - assertClassIsAbsent("io.flamingock.store.sql.internal.JournalEventPersistenceConstants"); - } - - @ParameterizedTest(name = "{0} uses the exact journal type policy") - @EnumSource(SqlDialect.class) - @DisplayName("uses exact portable types and capacities") - void usesExactPortableTypes(SqlDialect dialect) { - SqlJournalDialectHelper helper = new SqlJournalDialectHelper(dialect); - String ddl = helper.getCreateTableSqlString(TABLE_NAME).toUpperCase(Locale.ROOT); - - assertTrue(ddl.contains("EVENT_ID " + varcharType(dialect, 255) + " NOT NULL")); - assertTrue(ddl.contains("EVENT_TYPE " + varcharType(dialect, 32) + " NOT NULL")); - assertTrue(ddl.contains("EVENT_VERSION INTEGER NOT NULL")); - assertTrue(ddl.contains("STREAM_ID " + varcharType(dialect, 255) + " NOT NULL")); - assertTrue(ddl.contains("STREAM_SEQUENCE " + longType(dialect) + " NOT NULL")); - assertTrue(ddl.contains("OCCURRED_AT " + timestampType(dialect) + " NOT NULL")); - assertTrue(ddl.contains("ACKNOWLEDGED " + booleanType(dialect) + " NOT NULL")); - assertTrue(ddl.contains("PRIMARY KEY (STREAM_ID, STREAM_SEQUENCE)")); - assertTrue(ddl.contains("METADATA " + textType(dialect))); - assertTrue(ddl.contains("ERROR_TRACE " + textType(dialect))); - assertFalse(ddl.contains("CLOB")); - assertTrue(ddl.contains("TRANSACTION_FLAG " + booleanType(dialect))); - assertTrue(ddl.contains("SYSTEM_CHANGE " + booleanType(dialect))); - } - - @Test - @DisplayName("keeps the required typed Journal definitions and text capacity policy") - void keepsTypedColumnDefinitionsAndTextCapacity() { - SqlJournalDialectHelper helper = new SqlJournalDialectHelper(SqlDialect.H2); - - assertEquals(Arrays.asList( - "event_id", "event_type", "event_version", "stream_id", "stream_sequence", "occurred_at", - "acknowledged", "execution_id", "stage_id", "change_id", "author", "created_at", "state", - "invoked_class", "invoked_method", "source_file", "metadata", "execution_millis", - "execution_hostname", "error_trace", "type", "tx_strategy", "target_system_id", - "change_order", "recovery_strategy", "transaction_flag", "system_change"), - columnNames(helper.getColumnDefinitions())); - assertEquals(27, helper.getColumnDefinitions().size()); - assertEquals(SqlJournalDialectHelper.ColumnType.TEXT, helper.getColumnDefinitions().get(16).type); - assertEquals(2048, helper.getColumnDefinitions().get(16).size); - assertEquals(SqlJournalDialectHelper.ColumnType.TEXT, helper.getColumnDefinitions().get(19).type); - assertEquals(2048, helper.getColumnDefinitions().get(19).size); - assertTrue(helper.getColumnDefinitions().get(25).nullable); - assertTrue(helper.getColumnDefinitions().get(26).nullable); - } - - private static List columnNames(List definitions) { - List names = new java.util.ArrayList<>(); - for (SqlJournalDialectHelper.ColumnDefinition definition : definitions) { - names.add(definition.name); - } - return names; - } - - private static List insertColumnNames(String insertSql) { - int start = insertSql.indexOf('(') + 1; - int end = insertSql.indexOf(") VALUES"); - return Arrays.asList(insertSql.substring(start, end).split(", ")); - } - - @Test - @DisplayName("derives deterministic table-scoped index names within dialect limits") - void derivesDeterministicTableScopedIndexNames() { - SqlJournalDialectHelper helper = new SqlJournalDialectHelper(SqlDialect.ORACLE); - String tableName = "journalEventsWithAnIntentionallyVeryLongTableNameForOracle"; - - List first = helper.getIndexNames(tableName); - List second = helper.getIndexNames(tableName); - - assertEquals(first, second); - assertEquals(2, first.stream().distinct().count()); - assertTrue(first.stream().allMatch(name -> name.length() <= 30)); - assertTrue(first.stream().allMatch(name -> name.startsWith("idx_"))); - assertTrue(helper.getCreateIndexSqlStrings(tableName).stream() - .allMatch(sql -> first.stream().anyMatch(sql::contains))); - - String shortTableName = "customJournalEvents"; - assertEquals(Arrays.asList( - "idx_customJournalEvents_pending_events", - "idx_customJournalEvents_event_id"), - new SqlJournalDialectHelper(SqlDialect.H2).getIndexNames(shortTableName)); - } - - private static void assertClassIsAbsent(String className) { - assertThrows(ClassNotFoundException.class, () -> Class.forName(className)); - } - - private static String varcharType(SqlDialect dialect, int size) { - return (dialect == SqlDialect.ORACLE ? "VARCHAR2(" : "VARCHAR(") + size + ")"; - } - - private static String longType(SqlDialect dialect) { - return dialect == SqlDialect.ORACLE ? "NUMBER(19)" : "BIGINT"; - } - - private static String timestampType(SqlDialect dialect) { - if (dialect == SqlDialect.SQLSERVER || dialect == SqlDialect.SYBASE) { - return "DATETIME"; - } - if (dialect == SqlDialect.INFORMIX) { - return "DATETIME YEAR TO FRACTION(3)"; - } - return "TIMESTAMP"; - } - - private static String booleanType(SqlDialect dialect) { - switch (dialect) { - case MYSQL: - case MARIADB: - return "TINYINT(1)"; - case POSTGRESQL: - case H2: - case FIREBIRD: - case INFORMIX: - return "BOOLEAN"; - case SQLITE: - return "INTEGER"; - case SQLSERVER: - case SYBASE: - return "BIT"; - case ORACLE: - return "NUMBER(1)"; - case DB2: - default: - return "SMALLINT"; - } - } - - private static String textType(SqlDialect dialect) { - switch (dialect) { - case MYSQL: - case MARIADB: - case POSTGRESQL: - case SQLSERVER: - case SYBASE: - case SQLITE: - return "TEXT"; - case INFORMIX: - return "LVARCHAR(2048)"; - case ORACLE: - return "VARCHAR2(4000)"; - case DB2: - case FIREBIRD: - case H2: - default: - return "VARCHAR(4000)"; - } - } - - private static int countOccurrences(String value, String token) { - int count = 0; - int offset = 0; - while ((offset = value.indexOf(token, offset)) >= 0) { - count++; - offset += token.length(); - } - return count; - } -} diff --git a/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalEventMapperTest.java b/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalEventMapperTest.java index 22dc93ced..dde2c568b 100644 --- a/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalEventMapperTest.java +++ b/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalEventMapperTest.java @@ -21,6 +21,7 @@ import io.flamingock.internal.common.core.journal.JournalEvent; import io.flamingock.internal.common.core.journal.JournalEventType; import io.flamingock.internal.common.sql.SqlDialect; +import io.flamingock.internal.common.sql.dialectHelpers.SqlJournalDialectHelper; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; diff --git a/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalEventStoreJdbcTest.java b/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalEventStoreJdbcTest.java index 3d4b112b3..abc8563b3 100644 --- a/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalEventStoreJdbcTest.java +++ b/community/flamingock-sql-auditstore/src/test/java/io/flamingock/store/sql/internal/SqlJournalEventStoreJdbcTest.java @@ -26,6 +26,7 @@ import io.flamingock.internal.common.core.journal.JournalEventType; import io.flamingock.internal.common.core.transaction.TransactionWrapper; import io.flamingock.internal.common.sql.SqlDialect; +import io.flamingock.internal.common.sql.dialectHelpers.SqlJournalDialectHelper; import io.flamingock.internal.core.transaction.TransactionManager; import io.flamingock.targetsystem.sql.SqlTxWrapper; import org.h2.jdbcx.JdbcDataSource; diff --git a/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java b/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java index 7f47700b1..f29fabea8 100644 --- a/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java +++ b/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java @@ -15,18 +15,43 @@ */ package io.flamingock.internal.core.journal; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.util.FeatureFlag; + public class JournalEventSequencerFactory { - private final JournalEventReader journalEventReader; + private final JournalEventStore journalEventStore; - public JournalEventSequencerFactory(JournalEventReader journalEventReader) { - this.journalEventReader = journalEventReader; + public JournalEventSequencerFactory(JournalEventStore journalEventStore) { + this.journalEventStore = journalEventStore; } public JournalEventSequencer forStream(String streamId) { - long initialSequence = journalEventReader.getLastEventByStream(streamId) + long initialSequence = journalEventStore.getLastEventByStream(streamId) .map(e -> e.getStreamSequence() + 1) .orElse(1L); return new JournalEventSequencer(streamId, initialSequence); } + + /** + * Centralizes the per-stage journal initialization shared by every community audit store: + * when the {@link Features#JOURNAL_EVENTS} feature flag is on, ensures the journal store exists + * (creating it when {@code autoCreate}) and returns a sequencer for the given stage; otherwise + * returns {@code null} and leaves the journal store untouched. + */ + public JournalEventSequencer initializeForStage(String stageId, boolean autoCreate) { + if (!isJournalEventsEnabled()) { + return null; + } + journalEventStore.initialize(autoCreate); + return forStream(stageId); + } + + private static boolean isJournalEventsEnabled() { + try { + return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false); + } catch (RuntimeException exception) { + return false; + } + } } diff --git a/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventStore.java b/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventStore.java index c74dbbcdb..568623b29 100644 --- a/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventStore.java +++ b/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventStore.java @@ -32,6 +32,12 @@ */ public interface JournalEventStore extends JournalEventReader { + /** + * Ensures the underlying storage (table, collection, index, …) exists, creating it when + * {@code autoCreate} is {@code true}, or validating it otherwise. + */ + void initialize(boolean autoCreate); + /** * Marks the events with the given ids as acknowledged and returns how many were updated. */ diff --git a/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/JournalEventFieldConstants.java b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/JournalEventFieldConstants.java index 6b55747cb..98e033d84 100644 --- a/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/JournalEventFieldConstants.java +++ b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/JournalEventFieldConstants.java @@ -27,8 +27,6 @@ */ public final class JournalEventFieldConstants { - public static final String DEFAULT_JOURNAL_REPOSITORY_NAME = "flamingockJournalEvents"; - public static final String KEY_STREAM_ID = "streamId"; public static final String KEY_STREAM_SEQUENCE = "streamSequence"; public static final String KEY_PENDING_PARTITION_KEY = "pendingPartitionKey"; diff --git a/utils/mongodb-util/src/main/java/io/flamingock/internal/common/mongodb/journal/JournalEventPersistenceConstants.java b/utils/mongodb-util/src/main/java/io/flamingock/internal/common/mongodb/journal/JournalEventPersistenceConstants.java deleted file mode 100644 index 40c813278..000000000 --- a/utils/mongodb-util/src/main/java/io/flamingock/internal/common/mongodb/journal/JournalEventPersistenceConstants.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * Copyright 2026 Flamingock (https://www.flamingock.io) - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.flamingock.internal.common.mongodb.journal; - -/** - * Default persistence name for the local event buffer, mirroring the (external, un-extendable) - * {@code CommunityPersistenceConstants} defaults used for the audit and lock collections. - */ -public final class JournalEventPersistenceConstants { - - public static final String DEFAULT_JOURNAL_STORE_NAME = "flamingockJournalEvents"; - - private JournalEventPersistenceConstants() { - } -}