flamingock · bercianor · Aug 22, 2026
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
@@ -15,33 +15,50 @@
  */
 package io.flamingock.store.sql;
 
+import io.flamingock.internal.common.core.audit.AuditPersistenceFactory;
+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.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;
+import io.flamingock.store.sql.internal.SqlAuditRepository;
 import io.flamingock.store.sql.internal.SqlLockService;
+import io.flamingock.store.sql.internal.SqlJournalEventStore;
 import io.flamingock.externalsystem.sql.api.SqlExternalSystem;
 
 import javax.sql.DataSource;
 
 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;
     private RunnerId runnerId;
-    private SqlAuditPersistence persistence;
     private SqlLockService lockService;
+    private SqlJournalEventStore journalEventStore;
+    private JournalEventSequencerFactory journalEventSequencerFactory;
+    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 boolean autoCreate = true;
 
-    private SqlAuditStore(DataSource dataSource) {
-        this.dataSource = dataSource;
+    private SqlAuditStore(SqlExternalSystem targetSystem) {
+        this.targetSystem = targetSystem;
+        this.dataSource = targetSystem.getDataSource();
     }
 
     /**
@@ -55,7 +72,7 @@ private SqlAuditStore(DataSource dataSource) {
      * @return a new audit store bound to the same SQL datasource as the target system
      */
     public static SqlAuditStore from(SqlExternalSystem targetSystem) {
-        return new SqlAuditStore(targetSystem.getDataSource());
+        return new SqlAuditStore(targetSystem);
     }
 
     @Override
@@ -73,6 +90,11 @@ public SqlAuditStore withLockRepositoryName(String lockRepositoryName) {
         return this;
     }
 
+    public SqlAuditStore withJournalRepositoryName(String journalRepositoryName) {
+        this.journalRepositoryName = journalRepositoryName;
+        return this;
+    }
+
     public SqlAuditStore withAutoCreate(boolean autoCreate) {
         this.autoCreate = autoCreate;
         return this;
@@ -82,24 +104,84 @@ public SqlAuditStore withAutoCreate(boolean autoCreate) {
     public void initialize(ContextResolver baseContext) {
         runnerId = baseContext.getRequiredDependencyValue(RunnerId.class);
         communityConfiguration = baseContext.getRequiredDependencyValue(CommunityConfigurable.class);
+        validate();
+        auditRepository = new SqlAuditRepository(dataSource, auditRepositoryName);
+        journalEventStore = new SqlJournalEventStore(
+                dataSource,
+                journalRepositoryName,
+                targetSystem.getTxWrapper());
+        journalEventSequencerFactory = new JournalEventSequencerFactory(journalEventStore);
+        auditRepository.initialize(autoCreate);
+
+        lockService = new SqlLockService(dataSource, lockRepositoryName);
+        lockService.initialize(autoCreate);
     }
 
     @Override
-    public synchronized CommunityAuditPersistence getPersistence() {
-        if (persistence == null) {
-            persistence = new SqlAuditPersistence(communityConfiguration, dataSource, auditRepositoryName, autoCreate);
+    public AuditPersistenceFactory<CommunityAuditPersistence> getPersistenceFactory() {
+        return stageId -> {
+            boolean journalEventsEnabled = isJournalEventsEnabled();
+            JournalEventSequencer journalEventSequencer = null;
+            if (journalEventsEnabled) {
+                journalEventStore.initialize(autoCreate);
+                journalEventSequencer = journalEventSequencerFactory.forStream(stageId);
+            }
+
+            SqlAuditPersistence persistence = new SqlAuditPersistence(
+                    communityConfiguration,
+                    auditRepository,
+                    journalEventStore,
+                    journalEventSequencer,
+                    targetSystem.getTxWrapper(),
+                    journalEventsEnabled);
             persistence.initialize(runnerId);
-        }
-        return persistence;
+            return persistence;
+        };
+    }
+
+    @Override
+    public synchronized AuditReader getAuditReader() {
+        return () -> auditRepository.getAuditHistory();
     }
 
 
     @Override
     public synchronized CommunityLockService getLockService() {
-        if (lockService == null) {
-            lockService = new SqlLockService(dataSource, lockRepositoryName);
-            lockService.initialize(autoCreate);
-        }
         return lockService;
     }
+
+    private void validate() {
+        if (targetSystem == null || dataSource == null) {
+            throw new FlamingockException("The 'SqlExternalSystem' and its 'DataSource' are required.");
+        }
+        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.");
+        }
+    }
+
+    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");
+        }
+    }
+
+    private static boolean isJournalEventsEnabled() {
+        try {
+            return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false);
+        } catch (RuntimeException exception) {
+            return false;
+        }
+    }
 }
diff --git a/...ngock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/AuditEntryMapper.java b/...ngock-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/AuditEntryMapper.java
@@ -0,0 +1,146 @@
+/*
+ * 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.api.RecoveryStrategy;
+import io.flamingock.internal.common.core.audit.AuditEntry;
+import io.flamingock.internal.common.core.audit.AuditTxType;
+
+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;
+
+/**
+ * Binds and reads the typed, flattened SQL representation of an {@link AuditEntry}.
+ */
+final class AuditEntryMapper {
+
+    private static final List<String> 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<String> columnNames() {
+        return COLUMN_NAMES;
+    }
+
+    static void bind(PreparedStatement statement, AuditEntry auditEntry, int firstColumn) throws SQLException {
+        bind(statement, auditEntry, firstColumn, Types.BOOLEAN);
+    }
+
+    static void bind(PreparedStatement statement,
+                     AuditEntry auditEntry,
+                     int firstColumn,
+                     int nullableBooleanType) throws SQLException {
+        int column = firstColumn;
+        statement.setString(column++, auditEntry.getExecutionId());
+        statement.setString(column++, auditEntry.getStageId());
+        statement.setString(column++, auditEntry.getChangeId());
+        statement.setString(column++, auditEntry.getAuthor());
+        if (auditEntry.getCreatedAt() == null) {
+            statement.setNull(column++, Types.TIMESTAMP);
+        } else {
+            statement.setTimestamp(column++, Timestamp.valueOf(auditEntry.getCreatedAt()));
+        }
+        statement.setString(column++, auditEntry.getState() == null ? null : auditEntry.getState().name());
+        statement.setString(column++, auditEntry.getClassName());
+        statement.setString(column++, auditEntry.getMethodName());
+        statement.setString(column++, auditEntry.getSourceFile());
+        statement.setString(column++, auditEntry.getMetadata() == null ? null : auditEntry.getMetadata().toString());
+        statement.setLong(column++, auditEntry.getExecutionMillis());
+        statement.setString(column++, auditEntry.getExecutionHostname());
+        statement.setString(column++, auditEntry.getErrorTrace());
+        statement.setString(column++, auditEntry.getType() == null ? null : auditEntry.getType().name());
+        statement.setString(column++, auditEntry.getTxType() == null ? null : auditEntry.getTxType().name());
+        statement.setString(column++, auditEntry.getTargetSystemId());
+        statement.setString(column++, auditEntry.getOrder());
+        statement.setString(column++, auditEntry.getRecoveryStrategy() == null ? null : auditEntry.getRecoveryStrategy().name());
+        setNullableBoolean(statement, column++, auditEntry.getTransactionFlag(), nullableBooleanType);
+        setNullableBoolean(statement, column, auditEntry.getSystemChange(), nullableBooleanType);
+    }
+
+    static AuditEntry fromResultSet(ResultSet resultSet) throws SQLException {
+        Timestamp createdAt = resultSet.getTimestamp(columnName(4));
+        return new AuditEntry(
+                resultSet.getString(columnName(0)),
+                resultSet.getString(columnName(1)),
+                resultSet.getString(columnName(2)),
+                resultSet.getString(columnName(3)),
+                createdAt == null ? null : createdAt.toLocalDateTime(),
+                enumValue(AuditEntry.Status.class, resultSet.getString(columnName(5))),
+                enumValue(AuditEntry.ChangeType.class, resultSet.getString(columnName(13))),
+                resultSet.getString(columnName(6)),
+                resultSet.getString(columnName(7)),
+                resultSet.getString(columnName(8)),
+                resultSet.getLong(columnName(10)),
+                resultSet.getString(columnName(11)),
+                resultSet.getString(columnName(9)),
+                readBoolean(resultSet, columnName(19)),
+                resultSet.getString(columnName(12)),
+                AuditTxType.fromString(resultSet.getString(columnName(14))),
+                resultSet.getString(columnName(15)),
+                resultSet.getString(columnName(16)),
+                enumValue(RecoveryStrategy.class, resultSet.getString(columnName(17))),
+                readNullableBoolean(resultSet, columnName(18)));
+    }
+
+    private static String columnName(int index) {
+        return COLUMN_NAMES.get(index);
+    }
+
+    private static void setNullableBoolean(PreparedStatement statement,
+                                            int column,
+                                            Boolean value,
+                                            int nullableBooleanType) throws SQLException {
+        if (value == null) {
+            statement.setNull(column, nullableBooleanType);
+        } else {
+            statement.setBoolean(column, value);
+        }
+    }
+
+    private static boolean readBoolean(ResultSet resultSet, String column) throws SQLException {
+        Boolean value = readNullableBoolean(resultSet, column);
+        return value != null && value;
+    }
+
+    private static Boolean readNullableBoolean(ResultSet resultSet, String column) throws SQLException {
+        Object value = resultSet.getObject(column);
+        if (value == null) {
+            return null;
+        }
+        if (value instanceof Boolean) {
+            return (Boolean) value;
+        }
+        if (value instanceof Number) {
+            return ((Number) value).intValue() != 0;
+        }
+        return Boolean.valueOf(value.toString());
+    }
+
+    private static <E extends Enum<E>> E enumValue(Class<E> type, String value) {
+        return value == null ? null : Enum.valueOf(type, value);
+    }
+}
diff --git a/...-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/JournalEventConstants.java b/...-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/JournalEventConstants.java
@@ -0,0 +1,63 @@
+/*
+ * 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");
+        }
+    }
+}