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 f1904e546..ef5c995d6 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 @@ -15,50 +15,63 @@ */ package io.flamingock.store.dynamodb; +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.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.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; +import io.flamingock.store.dynamodb.internal.DynamoDBJournalEventStore; import io.flamingock.store.dynamodb.internal.DynamoDBLockService; import io.flamingock.externalsystem.dynamodb.api.DynamoDBExternalSystem; import software.amazon.awssdk.services.dynamodb.DynamoDbClient; public class DynamoDBAuditStore implements CommunityAuditStore { - private final DynamoDbClient client; + private final DynamoDBExternalSystem targetSystem; + private RunnerId runnerId; private CommunityConfigurable communityConfiguration; private DynamoDBAuditPersistence persistence; private DynamoDBLockService lockService; + 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 long readCapacityUnits = 5L; private long writeCapacityUnits = 5L; private boolean autoCreate = true; + private DynamoDBAuditRepository auditRepository; + private DynamoDBJournalEventStore journalEventStore; + private JournalEventSequencerFactory journalEventSequencerFactory; - private DynamoDBAuditStore(DynamoDbClient client) { - this.client = client; + private DynamoDBAuditStore(DynamoDBExternalSystem targetSystem) { + this.targetSystem = targetSystem; + this.client = targetSystem.getClient(); } /** * Creates a {@link DynamoDBAuditStore} using the same DynamoDB client * configured in the given {@link DynamoDBExternalSystem}. *

- * Only the underlying DynamoDB instance (client) is reused. - * No additional target-system configuration is carried over. + * The DynamoDB client and transaction wrapper are reused from the target system. * * @param targetSystem the target system from which to derive the client * @return a new audit store bound to the same DynamoDB instance as the target system */ public static DynamoDBAuditStore from(DynamoDBExternalSystem targetSystem) { - return new DynamoDBAuditStore(targetSystem.getClient()); + return new DynamoDBAuditStore(targetSystem); } @Override @@ -76,6 +89,11 @@ public DynamoDBAuditStore withLockRepositoryName(String lockRepositoryName) { return this; } + public DynamoDBAuditStore withJournalRepositoryName(String journalRepositoryName) { + this.journalRepositoryName = journalRepositoryName; + return this; + } + public DynamoDBAuditStore withReadCapacityUnits(long readCapacityUnits) { this.readCapacityUnits = readCapacityUnits; return this; @@ -95,34 +113,53 @@ public DynamoDBAuditStore withAutoCreate(boolean autoCreate) { public void initialize(ContextResolver baseContext) { runnerId = baseContext.getRequiredDependencyValue(RunnerId.class); communityConfiguration = baseContext.getRequiredDependencyValue(CommunityConfigurable.class); + auditRepository = new DynamoDBAuditRepository(client); + journalEventStore = new DynamoDBJournalEventStore( + client, + journalRepositoryName, + readCapacityUnits, + writeCapacityUnits + ); + journalEventSequencerFactory = new JournalEventSequencerFactory(journalEventStore); + + lockService = new DynamoDBLockService(client, TimeService.getDefault()); + lockService.initialize( + autoCreate, + lockRepositoryName, + readCapacityUnits, + writeCapacityUnits + ); this.validate(); } @Override - public synchronized CommunityAuditPersistence getPersistence() { - if (persistence == null) { + public AuditPersistenceFactory getPersistenceFactory() { + return stageId -> { + JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId); persistence = new DynamoDBAuditPersistence( - client, - auditRepositoryName, - readCapacityUnits, - writeCapacityUnits, - autoCreate, - communityConfiguration); + communityConfiguration, + auditRepository, + journalEventStore, + journalEventSequencer, + targetSystem.getTxWrapper(), + auditRepositoryName, + readCapacityUnits, + writeCapacityUnits, + autoCreate + ); persistence.initialize(runnerId); - } - return persistence; + return persistence; + }; + } + + @Override + public AuditReader getAuditReader() { + auditRepository.initialize(autoCreate, auditRepositoryName, readCapacityUnits, writeCapacityUnits); + return () -> auditRepository.getAuditHistory(); } @Override public synchronized CommunityLockService getLockService() { - if (lockService == null) { - lockService = new DynamoDBLockService(client, TimeService.getDefault()); - lockService.initialize( - autoCreate, - lockRepositoryName, - readCapacityUnits, - writeCapacityUnits); - } return lockService; } @@ -140,8 +177,29 @@ private void validate() { throw new FlamingockException("The 'lockRepositoryName' property is required."); } + if (journalRepositoryName == null || journalRepositoryName.trim().isEmpty()) { + throw new FlamingockException("The 'journalRepositoryName' property is required."); + } + + if (readCapacityUnits <= 0) { + throw new FlamingockException("The 'readCapacityUnits' property must be greater than zero."); + } + + if (writeCapacityUnits <= 0) { + throw new FlamingockException("The 'writeCapacityUnits' property must be greater than zero."); + } + 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."); + } } + } diff --git a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistence.java b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistence.java index 8286dd5f0..fcc8f8fe4 100644 --- a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistence.java +++ b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistence.java @@ -16,32 +16,60 @@ package io.flamingock.store.dynamodb.internal; import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.context.RuntimeContext; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.core.transaction.TransactionWrapper; import io.flamingock.internal.core.configuration.community.CommunityConfigurable; +import io.flamingock.internal.core.context.BasicRuntimeContext; import io.flamingock.internal.core.external.store.audit.community.AbstractCommunityAuditPersistence; +import io.flamingock.internal.core.journal.JournalEventSequencer; +import io.flamingock.internal.core.journal.JournalEventSequencerFactory; +import io.flamingock.internal.util.FeatureFlag; import io.flamingock.internal.util.Result; import io.flamingock.internal.util.id.RunnerId; -import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.enhanced.dynamodb.model.TransactWriteItemsEnhancedRequest; import java.util.List; public class DynamoDBAuditPersistence extends AbstractCommunityAuditPersistence { - private final DynamoDbClient client; + private final DynamoDBAuditRepository auditRepository; + private final DynamoDBJournalEventStore journalEventStore; + private JournalEventSequencer journalEventSequencer; + private final TransactionWrapper txWrapper; private final String auditTableName; private final long readCapacityUnits; private final long writeCapacityUnits; private final boolean autoCreate; - private DynamoDBAuditor auditor; - - public DynamoDBAuditPersistence(DynamoDbClient client, + /** + * Creates a persistence over explicitly supplied audit, journal and transaction collaborators. + * + * @param localConfiguration community configuration + * @param auditRepository repository for audits + * @param journalEventStore journal store receiving staged events + * @param journalEventSequencer sequencer for the stage journal stream + * @param txWrapper transaction wrapper shared with the target system + * @param auditTableName audit table name + * @param readCapacityUnits audit and journal read capacity + * @param writeCapacityUnits audit and journal write capacity + * @param autoCreate whether missing tables may be created + */ + public DynamoDBAuditPersistence(CommunityConfigurable localConfiguration, + DynamoDBAuditRepository auditRepository, + DynamoDBJournalEventStore journalEventStore, + JournalEventSequencer journalEventSequencer, + TransactionWrapper txWrapper, String auditTableName, long readCapacityUnits, long writeCapacityUnits, - boolean autoCreate, - CommunityConfigurable localConfiguration) { + boolean autoCreate) { super(localConfiguration); - this.client = client; + this.auditRepository = auditRepository; + this.journalEventStore = journalEventStore; + this.journalEventSequencer = journalEventSequencer; + this.txWrapper = txWrapper; this.auditTableName = auditTableName; this.readCapacityUnits = readCapacityUnits; this.writeCapacityUnits = writeCapacityUnits; @@ -50,21 +78,44 @@ public DynamoDBAuditPersistence(DynamoDbClient client, @Override protected void doInitialize(RunnerId runnerId) { - auditor = new DynamoDBAuditor(client); - auditor.initialize( + auditRepository.initialize( autoCreate, auditTableName, readCapacityUnits, writeCapacityUnits); + if (isJournalEventsEnabled()) { + journalEventStore.initialize(autoCreate); + } } @Override public List getAuditHistory() { - return auditor.getAuditHistory(); + return auditRepository.getAuditHistory(); } @Override public Result writeEntry(AuditEntry auditEntry) { - return auditor.writeEntry(auditEntry); + if (isJournalEventsEnabled()) { + RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId()); + Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> { + TransactWriteItemsEnhancedRequest.Builder builder = runtimeContext.getContext() + .getRequiredDependencyValue(TransactWriteItemsEnhancedRequest.Builder.class); + JournalEvent journalEvent = journalEventSequencer.newEvent(auditEntry); + journalEventStore.contributeToTransaction(builder, journalEvent); + return auditRepository.contributeToTransaction(builder, auditEntry); + }); + journalEventSequencer.confirm(); + return result; + } + return auditRepository.writeEntry(auditEntry); + } + + private static boolean isJournalEventsEnabled() { + try { + return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false); + } catch (RuntimeException exception) { + return false; + } } + } diff --git a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditRepository.java b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditRepository.java new file mode 100644 index 000000000..b6bc1bca0 --- /dev/null +++ b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditRepository.java @@ -0,0 +1,151 @@ +/* + * Copyright 2023 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.dynamodb.internal; + +import io.flamingock.internal.util.Result; +import io.flamingock.internal.util.dynamodb.entities.AuditEntryEntity; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.util.dynamodb.DynamoDBConstants; +import io.flamingock.internal.util.dynamodb.DynamoDBUtil; +import io.flamingock.internal.util.log.FlamingockLoggerFactory; +import org.slf4j.Logger; +import software.amazon.awssdk.enhanced.dynamodb.DynamoDbTable; +import software.amazon.awssdk.enhanced.dynamodb.TableSchema; +import software.amazon.awssdk.enhanced.dynamodb.model.PutItemEnhancedRequest; +import software.amazon.awssdk.enhanced.dynamodb.model.ScanEnhancedRequest; +import software.amazon.awssdk.enhanced.dynamodb.model.TransactWriteItemsEnhancedRequest; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.services.dynamodb.model.AttributeDefinition; +import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest; +import software.amazon.awssdk.services.dynamodb.model.KeySchemaElement; +import software.amazon.awssdk.services.dynamodb.model.KeyType; +import software.amazon.awssdk.services.dynamodb.model.ResourceNotFoundException; +import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType; +import software.amazon.awssdk.services.dynamodb.model.TableDescription; + +import java.util.List; +import java.util.stream.Collectors; + +import static java.util.Collections.emptyList; + +public class DynamoDBAuditRepository { + + private static final Logger logger = FlamingockLoggerFactory.getLogger("DynamoDBAuditRepository"); + + private final DynamoDBUtil dynamoDBUtil; + protected DynamoDbTable table; + + public DynamoDBAuditRepository(DynamoDbClient client) { + this.dynamoDBUtil = new DynamoDBUtil(client); + } + + public synchronized void initialize(Boolean autoCreate, + String tableName, + long readCapacityUnits, + long writeCapacityUnits) { + if (table != null) { + return; + } + if (autoCreate) { + dynamoDBUtil.createTable( + dynamoDBUtil.getAttributeDefinitions(DynamoDBConstants.AUDIT_LOG_PK, null), + dynamoDBUtil.getKeySchemas(DynamoDBConstants.AUDIT_LOG_PK, null), + dynamoDBUtil.getProvisionedThroughput(readCapacityUnits, writeCapacityUnits), + tableName, + emptyList(), + emptyList() + ); + } + validateSchema(tableName); + table = dynamoDBUtil.getEnhancedClient().table(tableName, TableSchema.fromBean(AuditEntryEntity.class)); + } + + private void validateSchema(String tableName) { + TableDescription description; + try { + description = dynamoDBUtil.getDynamoDBClient().describeTable( + DescribeTableRequest.builder().tableName(tableName).build()).table(); + } catch (ResourceNotFoundException exception) { + throw new IllegalStateException("DynamoDB audit table '" + tableName + + "' is missing or has an invalid schema", exception); + } + + boolean keyValid = description.keySchema() != null + && description.keySchema().size() == 1 + && hasKey(description.keySchema().get(0), DynamoDBConstants.AUDIT_LOG_PK, KeyType.HASH); + boolean attributeValid = description.attributeDefinitions() != null + && description.attributeDefinitions().stream() + .anyMatch(attribute -> hasAttribute(attribute, DynamoDBConstants.AUDIT_LOG_PK, ScalarAttributeType.S)); + if (!keyValid || !attributeValid) { + throw new IllegalStateException("DynamoDB audit table '" + tableName + + "' has an invalid key schema"); + } + } + + private boolean hasKey(KeySchemaElement key, String name, KeyType type) { + return key != null && name.equals(key.attributeName()) && type == key.keyType(); + } + + private boolean hasAttribute(AttributeDefinition attribute, String name, ScalarAttributeType type) { + return attribute != null && name.equals(attribute.attributeName()) && type == attribute.attributeType(); + } + + /** + * Appends an audit entry using the historical append key. + * + * @param auditEntry entry to append + * @return successful write result + */ + Result writeEntry(AuditEntry auditEntry) { + AuditEntryEntity entity = new AuditEntryEntity(auditEntry); + logger.debug("Saving audit entry with key {}", entity.getPartitionKey()); + table.putItem(PutItemEnhancedRequest.builder(AuditEntryEntity.class) + .item(entity) + .build()); + return Result.OK(); + } + + /** + * Stages a current-state audit write in a caller-owned DynamoDB transaction. + * + * @param builder transaction builder receiving the audit write + * @param auditEntry entry to stage + */ + Result contributeToTransaction(TransactWriteItemsEnhancedRequest.Builder builder, AuditEntry auditEntry) { + if (table == null) { + throw new IllegalStateException("DynamoDB audit writer is not initialized"); + } + AuditEntryEntity entity = new AuditEntryEntity(auditEntry); + entity.setPartitionKey(auditEntry.getChangeId()); + builder.addPutItem(table, PutItemEnhancedRequest.builder(AuditEntryEntity.class) + .item(entity) + .build()); + logger.debug("Staged current-state audit entry with key {}", entity.getPartitionKey()); + return Result.OK(); + } + + public List getAuditHistory() { + return table + .scan(ScanEnhancedRequest.builder() + .consistentRead(true) + .build() + ) + .items() + .stream() + .map(AuditEntryEntity::toAuditEntry) + .collect(Collectors.toList()); + } +} diff --git a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditor.java b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditor.java deleted file mode 100644 index 0ef4dac8f..000000000 --- a/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditor.java +++ /dev/null @@ -1,93 +0,0 @@ -/* - * Copyright 2023 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.dynamodb.internal; - -import io.flamingock.internal.common.core.audit.AuditWriter; -import io.flamingock.internal.core.external.store.audit.community.CommunityAuditReader; -import io.flamingock.internal.util.dynamodb.entities.AuditEntryEntity; -import io.flamingock.internal.common.core.audit.AuditEntry; -import io.flamingock.internal.util.Result; -import io.flamingock.internal.util.dynamodb.DynamoDBConstants; -import io.flamingock.internal.util.dynamodb.DynamoDBUtil; -import io.flamingock.internal.util.log.FlamingockLoggerFactory; -import org.slf4j.Logger; -import software.amazon.awssdk.enhanced.dynamodb.DynamoDbTable; -import software.amazon.awssdk.enhanced.dynamodb.TableSchema; -import software.amazon.awssdk.enhanced.dynamodb.model.PutItemEnhancedRequest; -import software.amazon.awssdk.enhanced.dynamodb.model.ScanEnhancedRequest; -import software.amazon.awssdk.services.dynamodb.DynamoDbClient; -import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException; - -import java.util.List; -import java.util.stream.Collectors; - -import static java.util.Collections.emptyList; - -public class DynamoDBAuditor implements AuditWriter, CommunityAuditReader { - - private static final Logger logger = FlamingockLoggerFactory.getLogger("DynamoAuditor"); - - private final DynamoDBUtil dynamoDBUtil; - protected DynamoDbTable table; - - public DynamoDBAuditor(DynamoDbClient client) { - this.dynamoDBUtil = new DynamoDBUtil(client); - } - - protected void initialize(Boolean autoCreate, String tableName, long readCapacityUnits, long writeCapacityUnits) { - if (autoCreate) { - dynamoDBUtil.createTable( - dynamoDBUtil.getAttributeDefinitions(DynamoDBConstants.AUDIT_LOG_PK, null), - dynamoDBUtil.getKeySchemas(DynamoDBConstants.AUDIT_LOG_PK, null), - dynamoDBUtil.getProvisionedThroughput(readCapacityUnits, writeCapacityUnits), - tableName, - emptyList(), - emptyList() - ); - } - table = dynamoDBUtil.getEnhancedClient().table(tableName, TableSchema.fromBean(AuditEntryEntity.class)); - } - - @Override - public Result writeEntry(AuditEntry auditEntry) { - AuditEntryEntity entity = new AuditEntryEntity(auditEntry); - logger.debug("Saving audit entry with key {}", entity.getPartitionKey()); - try { - table.putItem( - PutItemEnhancedRequest.builder(AuditEntryEntity.class) - .item(entity) - .build() - ); - } catch (ConditionalCheckFailedException ex) { - logger.warn("Error saving audit entry with key {}", entity.getPartitionKey(), ex); - throw ex; - } - return Result.OK(); - } - - @Override - public List getAuditHistory() { - return table - .scan(ScanEnhancedRequest.builder() - .consistentRead(true) - .build() - ) - .items() - .stream() - .map(AuditEntryEntity::toAuditEntry) - .collect(Collectors.toList()); - } -} 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 new file mode 100644 index 000000000..4ddcfdfc9 --- /dev/null +++ b/community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStore.java @@ -0,0 +1,388 @@ +/* + * 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.dynamodb.internal; + +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.core.journal.JournalEventStore; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.Result; +import io.flamingock.internal.util.dynamodb.DynamoDBUtil; +import io.flamingock.internal.util.dynamodb.entities.journal.DynamoDBJournalEventMapper; +import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventEntity; +import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventFieldConstants; +import io.flamingock.internal.util.log.FlamingockLoggerFactory; +import org.slf4j.Logger; +import software.amazon.awssdk.enhanced.dynamodb.DynamoDbIndex; +import software.amazon.awssdk.enhanced.dynamodb.DynamoDbTable; +import software.amazon.awssdk.enhanced.dynamodb.Expression; +import software.amazon.awssdk.enhanced.dynamodb.Key; +import software.amazon.awssdk.enhanced.dynamodb.TableSchema; +import software.amazon.awssdk.enhanced.dynamodb.model.Page; +import software.amazon.awssdk.enhanced.dynamodb.model.PageIterable; +import software.amazon.awssdk.enhanced.dynamodb.model.PutItemEnhancedRequest; +import software.amazon.awssdk.enhanced.dynamodb.model.QueryConditional; +import software.amazon.awssdk.enhanced.dynamodb.model.QueryEnhancedRequest; +import software.amazon.awssdk.enhanced.dynamodb.model.TransactWriteItemsEnhancedRequest; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.services.dynamodb.model.AttributeDefinition; +import software.amazon.awssdk.services.dynamodb.model.AttributeValue; +import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException; +import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest; +import software.amazon.awssdk.services.dynamodb.model.GlobalSecondaryIndexDescription; +import software.amazon.awssdk.services.dynamodb.model.GlobalSecondaryIndex; +import software.amazon.awssdk.services.dynamodb.model.KeySchemaElement; +import software.amazon.awssdk.services.dynamodb.model.KeyType; +import software.amazon.awssdk.services.dynamodb.model.Projection; +import software.amazon.awssdk.services.dynamodb.model.ProjectionType; +import software.amazon.awssdk.services.dynamodb.model.ResourceNotFoundException; +import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType; +import software.amazon.awssdk.services.dynamodb.model.TableDescription; +import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** + * DynamoDB implementation of the local journal ({@code flamingockJournalEvents}). + *

+ * Sibling of {@link DynamoDBAuditRepository}/{@link DynamoDBLockService}: it owns its own table and index setup. + *

+ * The table has a base key of {@code (streamId, streamSequence)} with two GSIs: + *

+ * Reads and acknowledgements are exposed through {@link JournalEventStore}. The append + * ({@link #contributeToTransaction(TransactWriteItemsEnhancedRequest.Builder, JournalEvent)}) deliberately is not: it stages a + * conditional put on the shared transaction builder so the event lands in the same transaction as the audit + * entry it mirrors, and a transaction-request builder has no place in a core interface. Only + * {@link DynamoDBAuditPersistence} — which owns that transaction boundary — calls it. + */ +public class DynamoDBJournalEventStore implements JournalEventStore { + + private static final Logger logger = FlamingockLoggerFactory.getLogger("DynamoDBJournal"); + + private final DynamoDBUtil dynamoDBUtil; + private final String tableName; + private final long readCapacityUnits; + private final long writeCapacityUnits; + + private DynamoDbTable table; + private DynamoDbIndex pendingEventsIndex; + private DynamoDbIndex eventIdIndex; + + public DynamoDBJournalEventStore(DynamoDbClient client, + String tableName, + long readCapacityUnits, + long writeCapacityUnits) { + this.dynamoDBUtil = new DynamoDBUtil(client); + this.tableName = tableName; + this.readCapacityUnits = readCapacityUnits; + this.writeCapacityUnits = writeCapacityUnits; + } + + /** + * Initializes the store, gated by the {@link Features#JOURNAL_EVENTS} feature flag: when the flag is off + * nothing happens (no table, no indexes). When it is on, {@code autoCreate} creates and waits for the + * configured table when needed; otherwise the manually configured table is checked for the required shape. + * + * @param autoCreate whether to create the table when missing + */ + protected synchronized void initialize(boolean autoCreate) { + if (!isJournalEventsEnabled() || table != null) { + return; + } + if (autoCreate) { + createTable(); + } + validateSchema(); + table = dynamoDBUtil.getEnhancedClient().table(tableName, TableSchema.fromBean(JournalEventEntity.class)); + pendingEventsIndex = table.index(JournalEventFieldConstants.PENDING_EVENTS_INDEX); + eventIdIndex = table.index(JournalEventFieldConstants.EVENT_ID_INDEX); + } + + /** + * Stages a conditional event append on the caller's transaction builder. No server call happens until + * the caller commits the transaction. If the {@code (streamId, streamSequence)} position is already + * occupied, the complete transaction is cancelled and the failure surfaces as a + * {@code DatabaseTransactionException} (mapped by {@code DynamoDBTxWrapper}). + * + * @param builder the shared {@code TransactWriteItemsEnhancedRequest} builder + * @param event the event to append + */ + Result contributeToTransaction(TransactWriteItemsEnhancedRequest.Builder builder, JournalEvent event) { + if (table == null) { + throw new IllegalStateException("DynamoDB journal store is not initialized"); + } + JournalEventEntity eventEntity = DynamoDBJournalEventMapper.toEntity(event); + builder.addPutItem(table, PutItemEnhancedRequest.builder(JournalEventEntity.class) + .item(eventEntity) + .conditionExpression(Expression.builder() + .expression("attribute_not_exists(" + JournalEventFieldConstants.KEY_STREAM_ID + ")") + .build()) + .build()); + logger.debug("Journal event staged for commit [eventId={} type={} stream={} sequence={}]", + event.getEventId(), event.getEventType(), event.getStreamId(), event.getStreamSequence()); + return Result.OK(); + } + + @Override + public Optional> getLastEventByStream(String streamId) { + if (table == null) { + return Optional.empty(); + } + PageIterable pages = table.query(lastEventQuery(streamId)); + for (Page page : pages) { + if (!page.items().isEmpty()) { + return Optional.of(DynamoDBJournalEventMapper.fromEntity(page.items().get(0))); + } + } + return Optional.empty(); + } + + static QueryEnhancedRequest lastEventQuery(String streamId) { + QueryConditional queryConditional = QueryConditional.keyEqualTo(Key.builder().partitionValue(streamId).build()); + return QueryEnhancedRequest.builder() + .queryConditional(queryConditional) + .scanIndexForward(false) + .limit(1) + .consistentRead(true) + .build(); + } + + @Override + public List> getUnacknowledgedEvents(int limit) { + if (pendingEventsIndex == null) { + return Collections.emptyList(); + } + + List> events = new ArrayList<>(); + Iterator> pages = pendingEventsIndex.query(pendingEventsQuery(limit)).iterator(); + if (!pages.hasNext()) { + return events; + } + for (JournalEventEntity entity : pages.next().items()) { + events.add(DynamoDBJournalEventMapper.fromEntity(entity)); + } + return events; + } + + static QueryEnhancedRequest pendingEventsQuery(int limit) { + return QueryEnhancedRequest.builder() + .queryConditional(QueryConditional.keyEqualTo( + Key.builder().partitionValue(JournalEventFieldConstants.PENDING_PARTITION_VALUE).build())) + .scanIndexForward(true) + .limit(limit) + .build(); + } + + @Override + public long acknowledgeEvents(Collection eventIds) { + if (eventIdIndex == null || eventIds == null || eventIds.isEmpty()) { + return 0L; + } + long acknowledged = 0L; + for (String eventId : new LinkedHashSet<>(eventIds)) { + if (eventId == null || eventId.trim().isEmpty()) { + continue; + } + QueryConditional queryConditional = QueryConditional.keyEqualTo(Key.builder().partitionValue(eventId).build()); + for (Page page : eventIdIndex.query(queryConditional)) { + for (JournalEventEntity entity : page.items()) { + if (removePendingAttributes(entity)) { + acknowledged++; + } + } + } + } + return acknowledged; + } + + /** + * Drops both pending attributes, which removes the item from the sparse pending GSI. Conditioned on both + * attributes existing so re-acknowledging an already acknowledged event is a no-op. + * + * @return {@code true} when the item was actually updated + */ + private boolean removePendingAttributes(JournalEventEntity entity) { + Map key = new HashMap<>(); + key.put(JournalEventFieldConstants.KEY_STREAM_ID, AttributeValue.builder().s(entity.getStreamId()).build()); + key.put(JournalEventFieldConstants.KEY_STREAM_SEQUENCE, + AttributeValue.builder().n(String.valueOf(entity.getStreamSequence())).build()); + try { + dynamoDBUtil.getDynamoDBClient().updateItem(UpdateItemRequest.builder() + .tableName(tableName) + .key(key) + .updateExpression("REMOVE " + JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY + ", " + + JournalEventFieldConstants.KEY_PENDING_ORDER_KEY) + .conditionExpression("attribute_exists(" + JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY + + ") AND attribute_exists(" + JournalEventFieldConstants.KEY_PENDING_ORDER_KEY + ")") + .build()); + return true; + } catch (ConditionalCheckFailedException ex) { + logger.debug("Journal event already acknowledged [eventId={} stream={} sequence={}]", + entity.getEventId(), entity.getStreamId(), entity.getStreamSequence()); + return false; + } + } + + private void createTable() { + List attributeDefinitions = new ArrayList<>(); + attributeDefinitions.add(AttributeDefinition.builder() + .attributeName(JournalEventFieldConstants.KEY_STREAM_ID) + .attributeType(ScalarAttributeType.S) + .build()); + attributeDefinitions.add(AttributeDefinition.builder() + .attributeName(JournalEventFieldConstants.KEY_STREAM_SEQUENCE) + .attributeType(ScalarAttributeType.N) + .build()); + attributeDefinitions.add(AttributeDefinition.builder() + .attributeName(JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY) + .attributeType(ScalarAttributeType.S) + .build()); + attributeDefinitions.add(AttributeDefinition.builder() + .attributeName(JournalEventFieldConstants.KEY_PENDING_ORDER_KEY) + .attributeType(ScalarAttributeType.S) + .build()); + attributeDefinitions.add(AttributeDefinition.builder() + .attributeName(JournalEventFieldConstants.KEY_EVENT_ID) + .attributeType(ScalarAttributeType.S) + .build()); + + List globalSecondaryIndexes = new ArrayList<>(); + globalSecondaryIndexes.add(GlobalSecondaryIndex.builder() + .indexName(JournalEventFieldConstants.PENDING_EVENTS_INDEX) + .keySchema(Arrays.asList( + KeySchemaElement.builder() + .attributeName(JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY) + .keyType(KeyType.HASH) + .build(), + KeySchemaElement.builder() + .attributeName(JournalEventFieldConstants.KEY_PENDING_ORDER_KEY) + .keyType(KeyType.RANGE) + .build())) + .projection(Projection.builder().projectionType(ProjectionType.ALL).build()) + .provisionedThroughput(dynamoDBUtil.getProvisionedThroughput(readCapacityUnits, writeCapacityUnits)) + .build()); + globalSecondaryIndexes.add(GlobalSecondaryIndex.builder() + .indexName(JournalEventFieldConstants.EVENT_ID_INDEX) + .keySchema(Collections.singletonList(KeySchemaElement.builder() + .attributeName(JournalEventFieldConstants.KEY_EVENT_ID) + .keyType(KeyType.HASH) + .build())) + .projection(Projection.builder().projectionType(ProjectionType.KEYS_ONLY).build()) + .provisionedThroughput(dynamoDBUtil.getProvisionedThroughput(readCapacityUnits, writeCapacityUnits)) + .build()); + + dynamoDBUtil.createTable( + attributeDefinitions, + dynamoDBUtil.getKeySchemas(JournalEventFieldConstants.KEY_STREAM_ID, + JournalEventFieldConstants.KEY_STREAM_SEQUENCE), + dynamoDBUtil.getProvisionedThroughput(readCapacityUnits, writeCapacityUnits), + tableName, + Collections.emptyList(), + globalSecondaryIndexes); + } + + private void validateSchema() { + TableDescription description; + try { + description = dynamoDBUtil.getDynamoDBClient().describeTable( + DescribeTableRequest.builder().tableName(tableName).build()).table(); + } catch (ResourceNotFoundException exception) { + throw new IllegalStateException("DynamoDB journal table '" + tableName + + "' is missing or has an invalid schema", exception); + } + + boolean baseKeysValid = hasKeySchema(description.keySchema(), JournalEventFieldConstants.KEY_STREAM_ID, + KeyType.HASH.toString(), JournalEventFieldConstants.KEY_STREAM_SEQUENCE, KeyType.RANGE.toString()); + GlobalSecondaryIndexDescription pendingIndex = findIndex(description, JournalEventFieldConstants.PENDING_EVENTS_INDEX); + GlobalSecondaryIndexDescription eventIdIndex = findIndex(description, JournalEventFieldConstants.EVENT_ID_INDEX); + boolean indexesValid = pendingIndex != null + && hasKeySchema(pendingIndex.keySchema(), JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY, + KeyType.HASH.toString(), JournalEventFieldConstants.KEY_PENDING_ORDER_KEY, KeyType.RANGE.toString()) + && pendingIndex.projection() != null + && pendingIndex.projection().projectionType() == ProjectionType.ALL + && eventIdIndex != null + && hasKeySchema(eventIdIndex.keySchema(), JournalEventFieldConstants.KEY_EVENT_ID, KeyType.HASH.toString()) + && eventIdIndex.projection() != null + && eventIdIndex.projection().projectionType() == ProjectionType.KEYS_ONLY; + boolean attributesValid = hasAttribute(description, JournalEventFieldConstants.KEY_STREAM_ID, ScalarAttributeType.S) + && hasAttribute(description, JournalEventFieldConstants.KEY_STREAM_SEQUENCE, ScalarAttributeType.N) + && hasAttribute(description, JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY, ScalarAttributeType.S) + && hasAttribute(description, JournalEventFieldConstants.KEY_PENDING_ORDER_KEY, ScalarAttributeType.S) + && hasAttribute(description, JournalEventFieldConstants.KEY_EVENT_ID, ScalarAttributeType.S); + if (!baseKeysValid || !indexesValid || !attributesValid) { + throw new IllegalStateException("DynamoDB journal table '" + tableName + + "' has an invalid key or index schema"); + } + } + + private GlobalSecondaryIndexDescription findIndex(TableDescription description, String indexName) { + if (description.globalSecondaryIndexes() == null) { + return null; + } + for (GlobalSecondaryIndexDescription index : description.globalSecondaryIndexes()) { + if (indexName.equals(index.indexName())) { + return index; + } + } + return null; + } + + private boolean hasAttribute(TableDescription description, String name, ScalarAttributeType type) { + if (description.attributeDefinitions() == null) { + return false; + } + return description.attributeDefinitions().stream() + .anyMatch(attribute -> name.equals(attribute.attributeName()) && type == attribute.attributeType()); + } + + private boolean hasKeySchema(List schema, String... expected) { + if (schema == null || schema.size() * 2 != expected.length) { + return false; + } + for (int i = 0; i < schema.size(); i++) { + KeySchemaElement element = schema.get(i); + if (!expected[2 * i].equals(element.attributeName()) + || !expected[2 * i + 1].equals(element.keyType().toString())) { + return false; + } + } + return true; + } + + private static boolean isJournalEventsEnabled() { + try { + return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false); + } catch (RuntimeException exception) { + return false; + } + } +} diff --git a/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBAuditStoreJournalTest.java b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBAuditStoreJournalTest.java new file mode 100644 index 000000000..435a251f9 --- /dev/null +++ b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBAuditStoreJournalTest.java @@ -0,0 +1,273 @@ +/* + * 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.dynamodb; + +import io.flamingock.core.kit.audit.AuditEntryTestFactory; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.audit.AuditTxType; +import io.flamingock.internal.common.core.error.FlamingockException; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.core.configuration.community.CommunityConfiguration; +import io.flamingock.internal.core.context.SimpleContext; +import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.dynamodb.DynamoDBUtil; +import io.flamingock.internal.util.dynamodb.entities.journal.DynamoDBJournalEventMapper; +import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventEntity; +import io.flamingock.internal.util.id.RunnerId; +import io.flamingock.targetsystem.dynamodb.DynamoDBTargetSystem; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import software.amazon.awssdk.enhanced.dynamodb.TableSchema; +import software.amazon.awssdk.enhanced.dynamodb.model.ScanEnhancedRequest; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.services.dynamodb.model.DynamoDbException; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@Testcontainers +class DynamoDBAuditStoreJournalTest { + + private static final String STAGE_ONE = "stage-one"; + private static final String STAGE_TWO = "stage-two"; + + @Container + static final GenericContainer dynamoDBContainer = DynamoDBTestContainer.createContainer(); + + private DynamoDbClient client; + private String auditTableName; + private String lockTableName; + private String journalTableName; + + @BeforeEach + void setUp() { + client = DynamoDBTestContainer.createClient(dynamoDBContainer); + auditTableName = tableName("storeAudit"); + lockTableName = tableName("storeLock"); + journalTableName = tableName("storeJournal"); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + if (client != null) { + client.listTables().tableNames().forEach(tableName -> client.deleteTable( + software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest.builder() + .tableName(tableName) + .build())); + client.close(); + } + } + + @Test + @DisplayName("journal-enabled stores create independent persistence streams for each stage") + void journalEnabledStoreCreatesPerStagePersistenceStreams() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + SimpleContext context = newContext(); + DynamoDBAuditStore auditStore = initializeStore(context); + + CommunityAuditPersistence stageOnePersistence = auditStore.getPersistenceFactory().get(STAGE_ONE); + CommunityAuditPersistence stageTwoPersistence = auditStore.getPersistenceFactory().get(STAGE_TWO); + stageOnePersistence.writeEntry(auditEntry("change-one")); + stageTwoPersistence.writeEntry(auditEntry("change-two")); + + List> events = storedEvents(); + assertEquals(2, events.size()); + assertTrue(events.stream().anyMatch(event -> STAGE_ONE.equals(event.getStreamId()) + && event.getStreamSequence() == 1L + && "change-one".equals(event.getData().getChangeId()))); + assertTrue(events.stream().anyMatch(event -> STAGE_TWO.equals(event.getStreamId()) + && event.getStreamSequence() == 1L + && "change-two".equals(event.getData().getChangeId()))); + } + + @Test + @DisplayName("journal-enabled store reseeds each requested stage independently") + void journalEnabledStoreReseedsEachRequestedStageIndependently() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + SimpleContext context = newContext(); + DynamoDBAuditStore auditStore = initializeStore(context); + + auditStore.getPersistenceFactory().get(STAGE_ONE).writeEntry(auditEntry("stage-one-first")); + auditStore.getPersistenceFactory().get(STAGE_TWO).writeEntry(auditEntry("stage-two-first")); + auditStore.getPersistenceFactory().get(STAGE_ONE).writeEntry(auditEntry("stage-one-second")); + + List> events = storedEvents(); + assertEquals(3, events.size()); + assertTrue(events.stream().anyMatch(event -> STAGE_ONE.equals(event.getStreamId()) + && event.getStreamSequence() == 1L + && "stage-one-first".equals(event.getData().getChangeId()))); + assertTrue(events.stream().anyMatch(event -> STAGE_ONE.equals(event.getStreamId()) + && event.getStreamSequence() == 2L + && "stage-one-second".equals(event.getData().getChangeId()))); + assertTrue(events.stream().anyMatch(event -> STAGE_TWO.equals(event.getStreamId()) + && event.getStreamSequence() == 1L + && "stage-two-first".equals(event.getData().getChangeId()))); + } + + @Test + @DisplayName("journal-enabled autoCreate completes an audit-only installation with a journal table") + void journalEnabledAutoCreatesJournalForAuditOnlyInstallation() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + SimpleContext context = newContext(); + DynamoDBAuditStore auditStore = initializeStore(context); + + assertTrue(auditStore.getAuditReader().getAuditHistory().isEmpty()); + assertTrue(client.listTables().tableNames().contains(auditTableName)); + + auditStore.getPersistenceFactory().get(STAGE_ONE).writeEntry(auditEntry("audit-only-change")); + auditStore.getPersistenceFactory().get(STAGE_TWO); + + assertTrue(client.listTables().tableNames().contains(journalTableName)); + List> events = storedEvents(); + assertEquals(1, events.size()); + assertEquals("audit-only-change", events.get(0).getData().getChangeId()); + } + + @Test + @DisplayName("non-positive capacities are rejected during DynamoDB setup") + void nonPositiveCapacitiesFailDynamoDbSetup() { + SimpleContext context = newContext(); + DynamoDBAuditStore invalidReadCapacity = DynamoDBAuditStore.from(initializedTargetSystem(context)) + .withAuditRepositoryName(auditTableName) + .withLockRepositoryName(lockTableName) + .withJournalRepositoryName(journalTableName) + .withReadCapacityUnits(0L); + DynamoDBAuditStore invalidWriteCapacity = DynamoDBAuditStore.from(initializedTargetSystem(context)) + .withAuditRepositoryName(auditTableName) + .withLockRepositoryName(lockTableName) + .withJournalRepositoryName(journalTableName) + .withWriteCapacityUnits(-1L); + + assertThrows(DynamoDbException.class, () -> invalidReadCapacity.initialize(context)); + assertThrows(DynamoDbException.class, () -> invalidWriteCapacity.initialize(context)); + assertTrue(client.listTables().tableNames().isEmpty()); + } + + @Test + @DisplayName("manual setup validates the audit table before the journal table") + void manualSetupValidatesAuditTableBeforeJournalSetup() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + SimpleContext context = newContext(); + DynamoDBAuditStore auditStore = DynamoDBAuditStore.from(initializedTargetSystem(context)) + .withAuditRepositoryName(auditTableName) + .withLockRepositoryName(lockTableName) + .withJournalRepositoryName(journalTableName) + .withAutoCreate(false); + auditStore.initialize(context); + + IllegalStateException exception = assertThrows( + IllegalStateException.class, + () -> auditStore.getPersistenceFactory().get(STAGE_ONE)); + + assertTrue(exception.getMessage().contains("audit table")); + assertFalse(client.listTables().tableNames().contains(journalTableName)); + } + + @Test + @DisplayName("journal-enabled persistence remains stage-aware while the reader stays available") + void journalEnabledPersistenceRemainsStageAwareAndKeepsReaderAvailable() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + SimpleContext context = newContext(); + DynamoDBAuditStore auditStore = initializeStore(context); + + CommunityAuditPersistence persistence = auditStore.getPersistenceFactory().get(STAGE_ONE); + persistence.writeEntry(auditEntry("deprecated-direct-write")); + + assertEquals(1, persistence.getAuditHistory().size()); + assertEquals(1, auditStore.getAuditReader().getAuditHistory().size(), + "journal-enabled history reads must use the independent audit reader"); + List> events = storedEvents(); + assertEquals(1, events.size()); + assertEquals(STAGE_ONE, events.get(0).getStreamId()); + assertEquals(1L, events.get(0).getStreamSequence()); + } + + @Test + @DisplayName("journal repository cannot reuse the audit or lock table name") + void journalRepositoryNameMustBeUnique() { + SimpleContext context = newContext(); + DynamoDBTargetSystem targetSystem = initializedTargetSystem(context); + DynamoDBAuditStore auditStore = DynamoDBAuditStore.from(targetSystem) + .withAuditRepositoryName(auditTableName) + .withLockRepositoryName(lockTableName) + .withJournalRepositoryName(auditTableName); + + assertThrows(FlamingockException.class, () -> auditStore.initialize(context)); + } + + private List> storedEvents() { + if (!client.listTables().tableNames().contains(journalTableName)) { + return new ArrayList<>(); + } + return new DynamoDBUtil(client) + .getEnhancedClient() + .table(journalTableName, TableSchema.fromBean(JournalEventEntity.class)) + .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) + .items() + .stream() + .map(DynamoDBJournalEventMapper::fromEntity) + .collect(Collectors.toList()); + } + + private DynamoDBAuditStore initializeStore(SimpleContext context) { + DynamoDBAuditStore auditStore = DynamoDBAuditStore.from(initializedTargetSystem(context)) + .withAuditRepositoryName(auditTableName) + .withLockRepositoryName(lockTableName) + .withJournalRepositoryName(journalTableName); + auditStore.initialize(context); + return auditStore; + } + + private DynamoDBTargetSystem initializedTargetSystem(SimpleContext context) { + DynamoDBTargetSystem targetSystem = new DynamoDBTargetSystem("dynamodb", client); + targetSystem.initialize(context); + return targetSystem; + } + + private SimpleContext newContext() { + SimpleContext context = new SimpleContext(); + context.addDependency(RunnerId.generate()); + context.addDependency(new CommunityConfiguration()); + return context; + } + + private static AuditEntry auditEntry(String changeId) { + return AuditEntryTestFactory.createTestAuditEntry( + changeId, + AuditEntry.Status.APPLIED, + AuditTxType.NON_TX, + (Class) null); + } + + private static String tableName(String prefix) { + return prefix + UUID.randomUUID().toString().replace("-", ""); + } +} diff --git a/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBExternalSystemContractTest.java b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBExternalSystemContractTest.java new file mode 100644 index 000000000..90e281841 --- /dev/null +++ b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBExternalSystemContractTest.java @@ -0,0 +1,35 @@ +/* + * 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.dynamodb; + +import io.flamingock.externalsystem.dynamodb.api.DynamoDBExternalSystem; +import io.flamingock.internal.common.core.transaction.TransactionalExternalSystem; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +class DynamoDBExternalSystemContractTest { + + @Test + @DisplayName("DynamoDB external systems expose the shared transactional contract") + void dynamoDBExternalSystemIsTransactional() { + assertTrue( + TransactionalExternalSystem.class.isAssignableFrom(DynamoDBExternalSystem.class), + "DynamoDBExternalSystem must extend TransactionalExternalSystem" + ); + } +} diff --git a/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBJournalFeatureFlagE2ETest.java b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBJournalFeatureFlagE2ETest.java new file mode 100644 index 000000000..55ff6c0ed --- /dev/null +++ b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBJournalFeatureFlagE2ETest.java @@ -0,0 +1,162 @@ +/* + * 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.dynamodb; + +import io.flamingock.common.test.pipeline.CodeChangeTestDefinition; +import io.flamingock.core.kit.TestKit; +import io.flamingock.core.kit.audit.AuditEntryExpectation; +import io.flamingock.core.kit.audit.AuditTestSupport; +import io.flamingock.dynamodb.kit.DynamoDBTestKit; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.constants.CommunityPersistenceConstants; +import io.flamingock.internal.util.dynamodb.DynamoDBUtil; +import io.flamingock.internal.util.dynamodb.entities.AuditEntryEntity; +import io.flamingock.internal.util.dynamodb.entities.journal.DynamoDBJournalEventMapper; +import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventEntity; +import io.flamingock.store.dynamodb.changes.audit._001__NonTxTransactionalFalseChange; +import io.flamingock.targetsystem.dynamodb.DynamoDBTargetSystem; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import software.amazon.awssdk.enhanced.dynamodb.TableSchema; +import software.amazon.awssdk.enhanced.dynamodb.model.ScanEnhancedRequest; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import static io.flamingock.core.kit.audit.AuditEntryExpectation.APPLIED; +import static io.flamingock.core.kit.audit.AuditEntryExpectation.STARTED; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * End-to-end coverage for the {@link Features#JOURNAL_EVENTS} gate through a complete runner execution. + */ +@Testcontainers +class DynamoDBJournalFeatureFlagE2ETest { + + private static final String JOURNAL_TABLE = "flamingockJournalEvents"; + private static final String DEFAULT_STAGE_NAME = "default-stage-name"; + + @Container + static final GenericContainer dynamoDBContainer = DynamoDBTestContainer.createContainer(); + + private DynamoDbClient client; + private TestKit testKit; + + @BeforeEach + void setUp() { + client = DynamoDBTestContainer.createClient(dynamoDBContainer); + testKit = DynamoDBTestKit.create( + client, + DynamoDBAuditStore.from(new DynamoDBTargetSystem("dynamodb", client))); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + testKit.cleanUp(); + } + + @Test + @DisplayName("journal disabled: the audit log retains every state transition") + void journalDisabledRetainsHistoricalAuditEntries() { + runPipeline( + STARTED("non-tx-transactional-false"), + APPLIED("non-tx-transactional-false")); + + assertEquals(Arrays.asList(AuditEntry.Status.APPLIED.name(), AuditEntry.Status.STARTED.name()), storedAuditRecords().stream() + .map(AuditEntryEntity::getState) + .sorted() + .collect(Collectors.toList())); + } + + @Test + @DisplayName("journal enabled: an audit-only installation transparently creates the journal") + void journalEnabledSplitsCurrentStateFromHistory() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + + runPipeline(APPLIED("non-tx-transactional-false")); + + assertTrue(client.listTables().tableNames().contains(CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME), + "the existing audit log must remain available"); + assertTrue(client.listTables().tableNames().contains(JOURNAL_TABLE)); + List auditRecords = storedAuditRecords(); + assertEquals(1, auditRecords.size(), "the audit table must retain only the current state when journal is enabled"); + assertEquals(AuditEntry.Status.APPLIED.name(), auditRecords.get(0).getState()); + List> events = storedEvents(); + + assertEquals(2, events.size(), "one event must be stored for each audit state transition"); + assertTrue(events.stream().allMatch(event -> DEFAULT_STAGE_NAME.equals(event.getStreamId())), + "journal events must use the pipeline stage as their stream"); + assertEquals(Arrays.asList(1L, 2L), events.stream() + .map(JournalEvent::getStreamSequence) + .sorted() + .collect(Collectors.toList()), + "journal stream sequences must be contiguous from one"); + assertEquals(Arrays.asList(AuditEntry.Status.STARTED, AuditEntry.Status.APPLIED), events.stream() + .map(event -> event.getData().getState()) + .sorted() + .collect(Collectors.toList()), + "the journal must retain both audit state transitions"); + } + + private void runPipeline(AuditEntryExpectation... expectedAudits) { + DynamoDBTargetSystem targetSystem = new DynamoDBTargetSystem("dynamodb", client); + AuditTestSupport.withTestKit(testKit) + .GIVEN_Changes(new CodeChangeTestDefinition( + _001__NonTxTransactionalFalseChange.class, + Collections.singletonList(DynamoDbClient.class))) + .WHEN(() -> testKit.createBuilder() + .setAuditStore(DynamoDBAuditStore.from(targetSystem)) + .addTargetSystem(targetSystem) + .build() + .run()) + .THEN_VerifyAuditSequenceStrict(expectedAudits) + .run(); + } + + private List> storedEvents() { + return new DynamoDBUtil(client) + .getEnhancedClient() + .table(JOURNAL_TABLE, TableSchema.fromBean(JournalEventEntity.class)) + .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) + .items() + .stream() + .map(DynamoDBJournalEventMapper::fromEntity) + .collect(Collectors.toList()); + } + + private List storedAuditRecords() { + return new DynamoDBUtil(client) + .getEnhancedClient() + .table(CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME, TableSchema.fromBean(AuditEntryEntity.class)) + .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) + .items() + .stream() + .collect(Collectors.toList()); + } +} diff --git a/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistenceJournalTest.java b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistenceJournalTest.java new file mode 100644 index 000000000..032d4d9cd --- /dev/null +++ b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistenceJournalTest.java @@ -0,0 +1,383 @@ +/* + * 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.dynamodb.internal; + +import io.flamingock.core.kit.audit.AuditEntryTestFactory; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.audit.AuditTxType; +import io.flamingock.internal.common.core.error.DatabaseTransactionException; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.core.journal.JournalEventType; +import io.flamingock.internal.core.configuration.community.CommunityConfiguration; +import io.flamingock.internal.core.journal.JournalEventSequencer; +import io.flamingock.internal.core.journal.JournalEventSequencerFactory; +import io.flamingock.internal.core.transaction.TransactionManager; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.dynamodb.DynamoDBUtil; +import io.flamingock.internal.util.dynamodb.entities.AuditEntryEntity; +import io.flamingock.internal.util.dynamodb.entities.journal.DynamoDBJournalEventMapper; +import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventEntity; +import io.flamingock.store.dynamodb.DynamoDBTestContainer; +import io.flamingock.targetsystem.dynamodb.DynamoDBTxWrapper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import software.amazon.awssdk.enhanced.dynamodb.TableSchema; +import software.amazon.awssdk.enhanced.dynamodb.model.ScanEnhancedRequest; +import software.amazon.awssdk.enhanced.dynamodb.model.TransactWriteItemsEnhancedRequest; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.services.dynamodb.model.DeleteItemRequest; +import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest; +import software.amazon.awssdk.services.dynamodb.model.AttributeValue; + +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Drives the DynamoDB persistence directly so the audit put and journal put can be verified at one transaction + * boundary without depending on a full pipeline. + */ +@Testcontainers +class DynamoDBAuditPersistenceJournalTest { + + private static final String STREAM_ID = "stage-under-test"; + + @Container + static final GenericContainer dynamoDBContainer = DynamoDBTestContainer.createContainer(); + + private DynamoDbClient client; + private DynamoDBTxWrapper txWrapper; + private DynamoDBJournalEventStore journalEventStore; + private String auditTableName; + private String journalTableName; + + @BeforeEach + void setUp() { + client = DynamoDBTestContainer.createClient(dynamoDBContainer); + auditTableName = tableName("journalAudit"); + journalTableName = tableName("journalEvents"); + txWrapper = new DynamoDBTxWrapper( + client, + new TransactionManager<>(TransactWriteItemsEnhancedRequest::builder)); + journalEventStore = new DynamoDBJournalEventStore(client, journalTableName, 5L, 5L); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + deleteTable(auditTableName); + deleteTable(journalTableName); + if (client != null) { + client.close(); + } + } + + @Test + @DisplayName("journal disabled keeps append audit records and does not create the journal table") + void journalDisabledKeepsAppendAuditPathAndCreatesNoJournalTable() { + DynamoDBAuditPersistence persistence = persistenceFor(newSequencer()); + AuditEntry started = auditEntry("change-1", AuditEntry.Status.STARTED); + AuditEntry applied = auditEntry("change-1", AuditEntry.Status.APPLIED); + + persistence.writeEntry(started); + persistence.writeEntry(applied); + + List records = storedAuditRecords(); + assertEquals(2, records.size(), "flag OFF must keep one append record per state transition"); + assertTrue(records.stream().map(AuditEntryEntity::getPartitionKey).collect(Collectors.toList()) + .contains(AuditEntryEntity.partitionKey(started.getExecutionId(), started.getChangeId(), started.getState()))); + assertTrue(records.stream().map(AuditEntryEntity::getPartitionKey).collect(Collectors.toList()) + .contains(AuditEntryEntity.partitionKey(applied.getExecutionId(), applied.getChangeId(), applied.getState()))); + assertFalse(tableExists(journalTableName), "flag OFF must not initialize the journal table"); + } + + @Test + @DisplayName("journal enabled commits one event with the audit record on the persistence stream") + void journalEnabledWritesEventAlongsideAuditRecord() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + DynamoDBAuditPersistence persistence = persistenceFor(newSequencer()); + AuditEntry entry = auditEntry("change-1", AuditEntry.Status.APPLIED); + + persistence.writeEntry(entry); + + assertEquals(1, persistence.getAuditHistory().size()); + List> events = storedEvents(); + assertEquals(1, events.size(), "exactly one event must be emitted per transition"); + JournalEvent event = events.get(0); + assertEquals(STREAM_ID, event.getStreamId()); + assertEquals(1L, event.getStreamSequence()); + assertEquals(JournalEventType.CHANGE_STATE, event.getEventType()); + assertFalse(event.isAcknowledged()); + assertEquals(entry.getChangeId(), event.getData().getChangeId()); + } + + @Test + @DisplayName("journal-enabled persistence creates the journal beside an existing audit table") + void journalEnabledCreatesJournalFromAuditOnlyInstallation() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + new DynamoDBAuditRepository(client).initialize(true, auditTableName, 5L, 5L); + + DynamoDBAuditPersistence persistence = persistenceFor(newSequencer()); + persistence.writeEntry(auditEntry("audit-only-change", AuditEntry.Status.APPLIED)); + + assertTrue(tableExists(auditTableName)); + assertTrue(tableExists(journalTableName)); + assertEquals(1, storedEvents().size()); + assertEquals("audit-only-change", storedEvents().get(0).getData().getChangeId()); + } + + @Test + @DisplayName("persistence uses current envelope time and retains historical imported timestamp") + void journalEnabledSeparatesEnvelopeAndImportedEntryTimes() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + DynamoDBAuditPersistence persistence = persistenceFor(newSequencer()); + LocalDateTime historical = LocalDateTime.of(2020, 1, 2, 3, 4, 5); + AuditEntry imported = importedAuditEntry("legacy-timestamp", historical); + + persistence.writeEntry(imported); + + JournalEvent event = storedEvents().get(0); + assertEquals(historical, event.getData().getCreatedAt()); + assertTrue(event.getOccurredAt().isAfter(historical.toInstant(ZoneOffset.UTC))); + } + + @Test + @DisplayName("journal enabled collapses successive states to one current audit record while retaining both events") + void journalEnabledKeepsCurrentStateAuditRecordAndJournalHistory() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + DynamoDBAuditPersistence persistence = persistenceFor(newSequencer()); + + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.STARTED)); + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED)); + + List auditRecords = persistence.getAuditHistory(); + assertEquals(1, auditRecords.size(), "flag ON must use the changeId-only current-state key"); + assertEquals(AuditEntry.Status.APPLIED, auditRecords.get(0).getState()); + + List> events = storedEvents(); + assertEquals(2, events.size(), "every state transition must remain in the journal"); + assertTrue(events.stream().anyMatch(event -> event.getStreamSequence() == 1L)); + assertTrue(events.stream().anyMatch(event -> event.getStreamSequence() == 2L)); + } + + @Test + @DisplayName("journal enabled keeps imported audits as current state while retaining every legacy event") + void journalEnabledKeepsImportedAuditAsCurrentStateAndJournalHistory() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + DynamoDBAuditPersistence persistence = persistenceFor(newSequencer()); + + persistence.writeEntry(legacyAuditEntry("legacy-change", AuditEntry.Status.STARTED)); + persistence.writeEntry(legacyAuditEntry("legacy-change", AuditEntry.Status.APPLIED)); + + List auditRecords = storedAuditRecords(); + assertEquals(1, auditRecords.size(), "flag ON must retain one current audit record for an imported change"); + assertEquals("legacy-change", auditRecords.get(0).getPartitionKey()); + assertEquals(AuditEntry.Status.APPLIED.name(), auditRecords.get(0).getState()); + + List> events = storedEvents(); + assertEquals(2, events.size(), "every imported state must remain in the journal"); + assertTrue(events.stream().allMatch(event -> "legacy-change".equals(event.getData().getChangeId()))); + } + + @Test + @DisplayName("a canceled transaction rolls back the audit record with the journal event") + void canceledTransactionRollsBackAuditAndJournalWrites() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + DynamoDBAuditPersistence persistence = persistenceFor(newSequencer()); + occupyStreamPosition(1L); + + assertThrows(DatabaseTransactionException.class, + () -> persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED))); + + assertTrue(persistence.getAuditHistory().isEmpty(), "the audit put must roll back with the canceled transaction"); + assertEquals(1, storedEvents().size(), "only the pre-existing stream-position occupant may remain"); + } + + @Test + @DisplayName("a canceled transaction does not confirm the sequence and the next retry reuses the position") + void canceledTransactionLeavesNoGapForRetry() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + JournalEventSequencer sequencer = newSequencer(); + DynamoDBAuditPersistence persistence = persistenceFor(sequencer); + occupyStreamPosition(1L); + + assertThrows(DatabaseTransactionException.class, + () -> persistence.writeEntry(auditEntry("failed-change", AuditEntry.Status.APPLIED))); + deleteStreamPosition(1L); + + persistence.writeEntry(auditEntry("successful-change", AuditEntry.Status.APPLIED)); + + List> events = storedEvents(); + assertEquals(1, events.size()); + assertEquals(1L, events.get(0).getStreamSequence(), + "confirm must be skipped after cancellation so the retry uses the unspent position"); + assertEquals("successful-change", events.get(0).getData().getChangeId()); + } + + private DynamoDBAuditPersistence persistenceFor(JournalEventSequencer sequencer) { + DynamoDBAuditPersistence persistence = new DynamoDBAuditPersistence( + new CommunityConfiguration(), + new DynamoDBAuditRepository(client), + journalEventStore, + sequencer, + txWrapper, + auditTableName, + 5L, + 5L, + true); + persistence.initialize(io.flamingock.internal.util.id.RunnerId.generate()); + return persistence; + } + + private JournalEventSequencer newSequencer() { + return new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID); + } + + private void occupyStreamPosition(long sequence) { + JournalEvent squatter = new JournalEvent<>( + "pre-existing-event", + JournalEventType.CHANGE_STATE, + JournalEvent.DEFAULT_VERSION, + STREAM_ID, + sequence, + Instant.now(), + auditEntry("pre-existing-change", AuditEntry.Status.APPLIED), + false); + new DynamoDBUtil(client) + .getEnhancedClient() + .table(journalTableName, TableSchema.fromBean(JournalEventEntity.class)) + .putItem(DynamoDBJournalEventMapper.toEntity(squatter)); + } + + private void deleteStreamPosition(long sequence) { + Map key = new HashMap<>(); + key.put("streamId", AttributeValue.builder().s(STREAM_ID).build()); + key.put("streamSequence", AttributeValue.builder().n(String.valueOf(sequence)).build()); + client.deleteItem(DeleteItemRequest.builder().tableName(journalTableName).key(key).build()); + } + + private List> storedEvents() { + if (!tableExists(journalTableName)) { + return new ArrayList<>(); + } + return new DynamoDBUtil(client) + .getEnhancedClient() + .table(journalTableName, TableSchema.fromBean(JournalEventEntity.class)) + .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) + .items() + .stream() + .map(DynamoDBJournalEventMapper::fromEntity) + .collect(Collectors.toList()); + } + + private List storedAuditRecords() { + if (!tableExists(auditTableName)) { + return new ArrayList<>(); + } + return new DynamoDBUtil(client) + .getEnhancedClient() + .table(auditTableName, TableSchema.fromBean(AuditEntryEntity.class)) + .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) + .items() + .stream() + .collect(Collectors.toList()); + } + + private boolean tableExists(String tableName) { + return client.listTables().tableNames().contains(tableName); + } + + private void deleteTable(String tableName) { + if (client != null && tableName != null && tableExists(tableName)) { + client.deleteTable(DeleteTableRequest.builder().tableName(tableName).build()); + } + } + + private static AuditEntry auditEntry(String changeId, AuditEntry.Status status) { + return AuditEntryTestFactory.createTestAuditEntry(changeId, status, AuditTxType.NON_TX, (Class) null); + } + + private static AuditEntry legacyAuditEntry(String changeId, AuditEntry.Status status) { + AuditEntry auditEntry = auditEntry(changeId, status); + return new AuditEntry( + auditEntry.getExecutionId(), + auditEntry.getStageId(), + auditEntry.getChangeId(), + auditEntry.getAuthor(), + auditEntry.getCreatedAt(), + auditEntry.getState(), + AuditEntry.ChangeType.MONGOCK_EXECUTION, + auditEntry.getClassName(), + auditEntry.getMethodName(), + auditEntry.getSourceFile(), + auditEntry.getExecutionMillis(), + auditEntry.getExecutionHostname(), + auditEntry.getMetadata(), + auditEntry.getSystemChange(), + auditEntry.getErrorTrace(), + auditEntry.getTxType(), + auditEntry.getTargetSystemId(), + auditEntry.getOrder(), + auditEntry.getRecoveryStrategy(), + auditEntry.getTransactionFlag()); + } + + private static AuditEntry importedAuditEntry(String changeId, LocalDateTime createdAt) { + AuditEntry source = legacyAuditEntry(changeId, AuditEntry.Status.APPLIED); + return new AuditEntry( + source.getExecutionId(), + source.getStageId(), + source.getChangeId(), + source.getAuthor(), + createdAt, + source.getState(), + source.getType(), + source.getClassName(), + source.getMethodName(), + source.getSourceFile(), + source.getExecutionMillis(), + source.getExecutionHostname(), + source.getMetadata(), + source.getSystemChange(), + source.getErrorTrace(), + source.getTxType(), + source.getTargetSystemId(), + source.getOrder(), + source.getRecoveryStrategy(), + source.getTransactionFlag()); + } + + private static String tableName(String prefix) { + return prefix + UUID.randomUUID().toString().replace("-", ""); + } +} diff --git a/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStoreTest.java b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStoreTest.java new file mode 100644 index 000000000..536fe187f --- /dev/null +++ b/community/flamingock-dynamodb-auditstore/src/test/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStoreTest.java @@ -0,0 +1,579 @@ +/* + * 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.dynamodb.internal; + +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.core.error.DatabaseTransactionException; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.core.journal.JournalEventType; +import io.flamingock.internal.core.context.BasicRuntimeContext; +import io.flamingock.internal.core.journal.JournalEventSequencer; +import io.flamingock.internal.core.journal.JournalEventSequencerFactory; +import io.flamingock.internal.core.transaction.TransactionManager; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.Result; +import io.flamingock.internal.util.dynamodb.DynamoDBUtil; +import io.flamingock.internal.util.dynamodb.entities.journal.DynamoDBJournalEventMapper; +import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventFieldConstants; +import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventEntity; +import io.flamingock.store.dynamodb.DynamoDBTestContainer; +import io.flamingock.targetsystem.dynamodb.DynamoDBTxWrapper; +import org.mockito.ArgumentMatchers; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import software.amazon.awssdk.enhanced.dynamodb.Expression; +import software.amazon.awssdk.enhanced.dynamodb.TableSchema; +import software.amazon.awssdk.enhanced.dynamodb.model.TransactWriteItemsEnhancedRequest; +import software.amazon.awssdk.enhanced.dynamodb.model.QueryEnhancedRequest; +import software.amazon.awssdk.enhanced.dynamodb.model.Page; +import software.amazon.awssdk.enhanced.dynamodb.model.PageIterable; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.services.dynamodb.model.AttributeValue; +import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest; +import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest; +import software.amazon.awssdk.services.dynamodb.model.DescribeTableResponse; +import software.amazon.awssdk.services.dynamodb.model.GlobalSecondaryIndexDescription; +import software.amazon.awssdk.services.dynamodb.model.KeyType; +import software.amazon.awssdk.services.dynamodb.model.GetItemRequest; + +import java.time.Instant; +import java.time.LocalDateTime; +import java.lang.reflect.Field; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@Testcontainers +class DynamoDBJournalEventStoreTest { + + private static final String TABLE_NAME = "flamingockJournalEvents"; + + @Container + static final GenericContainer dynamoDBContainer = DynamoDBTestContainer.createContainer(); + + private DynamoDbClient client; + private DynamoDBJournalEventStore store; + + @BeforeEach + void setUp() { + client = DynamoDBTestContainer.createClient(dynamoDBContainer); + deleteTableIfExists(TABLE_NAME); + store = new DynamoDBJournalEventStore(client, TABLE_NAME, 5L, 5L); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + deleteTableIfExists(TABLE_NAME); + } + + @Test + @DisplayName("Flag OFF: initialize must not create the journal table") + void flagOffInitializeDoesNotCreateJournalTable() { + // Given: feature flag off + FeatureFlag.remove(Features.JOURNAL_EVENTS); + + // When: store initializes with autoCreate + store.initialize(true); + + // Then: no journal table exists + assertFalse(tableExists(TABLE_NAME)); + } + + @Test + @DisplayName("Flag ON: initialize creates the journal table with both GSIs") + void flagOnInitializeCreatesTableWithBothIndexes() { + // Given: feature flag on + FeatureFlag.enable(Features.JOURNAL_EVENTS); + + // When: store initializes with autoCreate + store.initialize(true); + + // Then: table exists + assertTrue(tableExists(TABLE_NAME)); + + // And: both GSIs exist with the expected key schemas + DescribeTableResponse response = client.describeTable( + DescribeTableRequest.builder().tableName(TABLE_NAME).build()); + List indexes = response.table().globalSecondaryIndexes(); + assertEquals(2, indexes.size()); + + GlobalSecondaryIndexDescription pendingIndex = indexes.stream() + .filter(index -> JournalEventFieldConstants.PENDING_EVENTS_INDEX.equals(index.indexName())) + .findFirst() + .orElseThrow(() -> new AssertionError("PendingEventsIndex not found")); + assertEquals(2, pendingIndex.keySchema().size()); + assertEquals(KeyType.HASH, pendingIndex.keySchema().get(0).keyType()); + assertEquals(JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY, + pendingIndex.keySchema().get(0).attributeName()); + assertEquals(KeyType.RANGE, pendingIndex.keySchema().get(1).keyType()); + assertEquals(JournalEventFieldConstants.KEY_PENDING_ORDER_KEY, + pendingIndex.keySchema().get(1).attributeName()); + assertEquals(software.amazon.awssdk.services.dynamodb.model.ProjectionType.ALL, + pendingIndex.projection().projectionType()); + + GlobalSecondaryIndexDescription eventIdIndex = indexes.stream() + .filter(index -> JournalEventFieldConstants.EVENT_ID_INDEX.equals(index.indexName())) + .findFirst() + .orElseThrow(() -> new AssertionError("EventIdIndex not found")); + assertEquals(1, eventIdIndex.keySchema().size()); + assertEquals(KeyType.HASH, eventIdIndex.keySchema().get(0).keyType()); + assertEquals(JournalEventFieldConstants.KEY_EVENT_ID, eventIdIndex.keySchema().get(0).attributeName()); + assertEquals(software.amazon.awssdk.services.dynamodb.model.ProjectionType.KEYS_ONLY, + eventIdIndex.projection().projectionType()); + } + + @Test + @DisplayName("malformed journal flags fail closed without initializing the table") + void malformedJournalFlagDoesNotInitializeJournalTable() { + try (MockedStatic flags = Mockito.mockStatic(FeatureFlag.class)) { + flags.when(() -> FeatureFlag.ifEnabled( + ArgumentMatchers.eq(Features.JOURNAL_EVENTS), + ArgumentMatchers.any(Runnable.class))) + .thenThrow(new IllegalArgumentException("malformed feature value")); + flags.when(() -> FeatureFlag.isEnabled(Features.JOURNAL_EVENTS)) + .thenThrow(new IllegalArgumentException("malformed feature value")); + flags.when(() -> FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false)) + .thenThrow(new IllegalArgumentException("malformed feature value")); + + assertDoesNotThrow(() -> store.initialize(true)); + } + + assertFalse(tableExists(TABLE_NAME)); + } + + @Test + @DisplayName("unknown journal flags remain disabled") + void unknownJournalFlagDoesNotInitializeJournalTable() { + String unknownFlag = "unknown-journal-flag"; + FeatureFlag.enable(unknownFlag); + try { + store.initialize(true); + assertFalse(tableExists(TABLE_NAME)); + } finally { + FeatureFlag.remove(unknownFlag); + } + } + + @Test + @DisplayName("auto-create validates an existing journal table before binding") + void autoCreateValidatesExistingJournalSchema() { + DynamoDBUtil dynamoDBUtil = new DynamoDBUtil(client); + dynamoDBUtil.createTable( + dynamoDBUtil.getAttributeDefinitions("wrongKey", null), + dynamoDBUtil.getKeySchemas("wrongKey", null), + dynamoDBUtil.getProvisionedThroughput(5L, 5L), + TABLE_NAME, + Collections.emptyList(), + Collections.emptyList()); + FeatureFlag.enable(Features.JOURNAL_EVENTS); + + IllegalStateException exception = assertThrows(IllegalStateException.class, + () -> store.initialize(true)); + + assertTrue(exception.getMessage().contains("invalid key or index schema")); + } + + @Test + @DisplayName("an occupied position rejects the write and a later position accepts the retried event") + void occupiedPositionRejectsRetriedEventAndLaterPositionAccepted() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + commit(journalEvent("stage-1", 1L, "first-event")); + + assertThrows(DatabaseTransactionException.class, + () -> commit(journalEvent("stage-1", 1L, "retryable-event"))); + + commit(journalEvent("stage-1", 2L, "retryable-event")); + + assertEquals(2, store.getUnacknowledgedEvents(10).size()); + } + + @Test + @DisplayName("Flag ON with autoCreate false: initialize asserts an existing table without throwing") + void flagOnInitializeWithAutoCreateFalseAssertsExistingTable() { + // Given: flag on and table already created + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + + // When: a second store initializes with autoCreate=false + DynamoDBJournalEventStore assertStore = new DynamoDBJournalEventStore(client, TABLE_NAME, 5L, 5L); + + // Then: existence assertion passes without throwing + assertDoesNotThrow(() -> assertStore.initialize(false)); + } + + @Test + @DisplayName("Appending to an occupied stream position cancels the transaction and surfaces DatabaseTransactionException") + void occupiedStreamPositionCancelsTransaction() { + // Given: flag on, store initialized, and an event committed at (stage-1, 1) + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + commit(journalEvent("stage-1", 1L, "event-1")); + + // When: another event is appended at the same position + JournalEvent collision = journalEvent("stage-1", 1L, "event-2"); + + // Then: the transaction cancels and DatabaseTransactionException surfaces + assertThrows(DatabaseTransactionException.class, () -> commit(collision)); + } + + @Test + @DisplayName("distinct event IDs preserve deterministic pending position order") + void distinctEventIdsPreservePendingPositionOrder() throws InterruptedException { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + commit(Arrays.asList( + journalEvent("stage-2", 1L, "event-stage-2"), + journalEvent("stage-1", 2L, "event-stage-1-second"), + journalEvent("stage-1", 1L, "event-stage-1-first"))); + + List> events = awaitUnacknowledgedCount(3); + + assertEquals(3, events.size()); + assertEquals("stage-1", events.get(0).getStreamId()); + assertEquals(1L, events.get(0).getStreamSequence()); + assertEquals("stage-1", events.get(1).getStreamId()); + assertEquals(2L, events.get(1).getStreamSequence()); + assertEquals("stage-2", events.get(2).getStreamId()); + assertEquals(1L, events.get(2).getStreamSequence()); + } + + @Test + @DisplayName("getLastEventByStream returns the highest sequence per stream, desc, with full payload") + void getLastEventByStreamReturnsHighestSequence() throws InterruptedException { + // Given: flag on, store initialized, events 1..3 on stage-1 and 1 on stage-2 + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + commit(journalEvent("stage-1", 1L, "event-1")); + commit(journalEvent("stage-1", 2L, "event-2")); + commit(journalEvent("stage-1", 3L, "event-3")); + commit(journalEvent("stage-2", 1L, "event-4")); + + // When/Then: last event of stage-1 is sequence 3 with the full AuditEntry payload + Optional> last = store.getLastEventByStream("stage-1"); + assertTrue(last.isPresent()); + assertEquals(3L, last.get().getStreamSequence()); + assertEquals("event-3", last.get().getEventId()); + assertEquals(JournalEventType.CHANGE_STATE, last.get().getEventType()); + assertEquals("change-3", last.get().getData().getChangeId()); + + // And: last event of stage-2 is sequence 1 + Optional> other = store.getLastEventByStream("stage-2"); + assertTrue(other.isPresent()); + assertEquals(1L, other.get().getStreamSequence()); + + // And: a stream with no events returns empty + Optional> none = store.getLastEventByStream("empty-stream"); + assertFalse(none.isPresent()); + } + + @Test + @DisplayName("last-event query is strongly consistent for sequencer reseeding") + void lastEventQueryIsStronglyConsistent() { + QueryEnhancedRequest request = DynamoDBJournalEventStore.lastEventQuery("stage-1"); + + assertTrue(request.consistentRead()); + assertFalse(request.scanIndexForward()); + assertEquals(1, request.limit()); + } + + @Test + @DisplayName("pending query uses the sparse partition and ascending order") + void pendingQueryUsesSparsePartitionAndAscendingOrder() { + QueryEnhancedRequest request = DynamoDBJournalEventStore.pendingEventsQuery(3); + Expression keyCondition = request.queryConditional() + .expression(TableSchema.fromBean(JournalEventEntity.class), JournalEventFieldConstants.PENDING_EVENTS_INDEX); + + assertTrue(request.scanIndexForward()); + assertEquals(3, request.limit()); + assertTrue(request.filterExpression() == null); + assertTrue(keyCondition.expressionNames().containsValue(JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY)); + assertEquals(JournalEventFieldConstants.PENDING_PARTITION_VALUE, + keyCondition.expressionValues().values().iterator().next().s()); + } + + @Test + @DisplayName("mapper writes a constant pending partition and collision-safe position order key") + void mapperWritesCollisionSafePendingOrderKey() { + JournalEventEntity singleCharacter = DynamoDBJournalEventMapper.toEntity( + journalEvent("A", 1L, "same-event-id")); + JournalEventEntity longerPrefix = DynamoDBJournalEventMapper.toEntity( + journalEvent("AA", 1L, "same-event-id")); + + assertEquals(JournalEventFieldConstants.PENDING_PARTITION_VALUE, + singleCharacter.getPendingPartitionKey()); + assertEquals("0041!0000000000000001", singleCharacter.getPendingOrderKey()); + assertEquals("00410041!0000000000000001", longerPrefix.getPendingOrderKey()); + assertTrue(singleCharacter.getPendingOrderKey().compareTo(longerPrefix.getPendingOrderKey()) < 0); + } + + @Test + @DisplayName("mapper omits both pending keys for an acknowledged event") + void mapperOmitsBothPendingKeysWhenAcknowledged() { + JournalEvent source = journalEvent("stage-1", 1L, "event-1"); + JournalEvent acknowledged = new JournalEvent<>( + source.getEventId(), + source.getEventType(), + source.getEventVersion(), + source.getStreamId(), + source.getStreamSequence(), + source.getOccurredAt(), + source.getData(), + true); + + JournalEventEntity entity = DynamoDBJournalEventMapper.toEntity(acknowledged); + + assertNull(entity.getPendingPartitionKey()); + assertNull(entity.getPendingOrderKey()); + } + + @Test + @DisplayName("pending order key keeps positive sequence ordering with fixed-width hexadecimal") + void pendingOrderKeyKeepsSequenceOrdering() { + String lowerSequence = DynamoDBJournalEventMapper.pendingOrderKey("stage-1", 2L); + String higherSequence = DynamoDBJournalEventMapper.pendingOrderKey("stage-1", 16L); + + assertEquals("00730074006100670065002D0031!0000000000000002", lowerSequence); + assertEquals("00730074006100670065002D0031!0000000000000010", higherSequence); + assertTrue(lowerSequence.compareTo(higherSequence) < 0); + } + + @Test + @DisplayName("pending order key rejects values exceeding DynamoDB sort-key size") + void pendingOrderKeyRejectsOversizedStreamId() { + String oversizedStreamId = new String(new char[252]).replace('\0', 'a'); + + assertThrows(IllegalArgumentException.class, + () -> DynamoDBJournalEventMapper.pendingOrderKey(oversizedStreamId, 1L)); + } + + @Test + @DisplayName("a new sequencer after a committed event starts at the next sequence") + void newSequencerAfterCommittedEventStartsAtNextSequence() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + commit(journalEvent("stage-1", 1L, "event-1")); + + JournalEventSequencer sequencer = new JournalEventSequencerFactory(store).forStream("stage-1"); + JournalEvent nextEvent = sequencer.newEvent(journalEvent("stage-1", 2L, "event-2").getData()); + + assertEquals(2L, nextEvent.getStreamSequence()); + } + + @Test + @DisplayName("getUnacknowledgedEvents returns position-ordered bounded events") + void getUnacknowledgedEventsQueriesPendingIndexSortedAndLimited() throws InterruptedException { + // Given: flag on, store initialized, 3 events on stage-1 and 2 on stage-2 + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + commit(journalEvent("stage-1", 1L, "event-1")); + commit(journalEvent("stage-1", 2L, "event-2")); + commit(journalEvent("stage-1", 3L, "event-3")); + commit(journalEvent("stage-2", 1L, "event-4")); + commit(journalEvent("stage-2", 2L, "event-5")); + + // When: a positive limit is requested + List> limited = store.getUnacknowledgedEvents(2); + + // Then: the native bounded query preserves pending-index order + assertEquals(2, limited.size()); + assertEquals("stage-1", limited.get(0).getStreamId()); + assertEquals(1L, limited.get(0).getStreamSequence()); + assertEquals("stage-1", limited.get(1).getStreamId()); + assertEquals(2L, limited.get(1).getStreamSequence()); + } + + @Test + @DisplayName("getUnacknowledgedEvents consumes only the native bounded page") + void getUnacknowledgedEventsDoesNotFollowContinuationPages() throws Exception { + DynamoDBJournalEventStore requestStore = new DynamoDBJournalEventStore( + Mockito.mock(DynamoDbClient.class), TABLE_NAME, 5L, 5L); + @SuppressWarnings("unchecked") + software.amazon.awssdk.enhanced.dynamodb.DynamoDbIndex pendingIndex = + Mockito.mock(software.amazon.awssdk.enhanced.dynamodb.DynamoDbIndex.class); + Field pendingIndexField = DynamoDBJournalEventStore.class.getDeclaredField("pendingEventsIndex"); + pendingIndexField.setAccessible(true); + pendingIndexField.set(requestStore, pendingIndex); + + Map continuation = Collections.singletonMap( + JournalEventFieldConstants.KEY_STREAM_ID, AttributeValue.builder().s("stage-1").build()); + Page page = Page.create(Collections.singletonList( + DynamoDBJournalEventMapper.toEntity(journalEvent("stage-1", 1L, "event-1"))), continuation); + Mockito.when(pendingIndex.query(ArgumentMatchers.any(QueryEnhancedRequest.class))) + .thenReturn(PageIterable.create(() -> Collections.singletonList(page).iterator())); + + List> events = requestStore.getUnacknowledgedEvents(3); + + assertEquals(Collections.singletonList("event-1"), events.stream() + .map(JournalEvent::getEventId) + .collect(Collectors.toList())); + } + + @Test + @DisplayName("acknowledgement with blank IDs returns 0 without exception") + void acknowledgementWithBlankIdsReturnsZero() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + + assertEquals(0L, store.acknowledgeEvents(Arrays.asList(null, "", " "))); + } + + @Test + @DisplayName("acknowledgeEvents resolves via EventIdIndex and drops the pending attribute") + void acknowledgeEventsRemovesPendingAttribute() throws InterruptedException { + // Given: flag on, store initialized, 2 unacknowledged events on stage-1 + FeatureFlag.enable(Features.JOURNAL_EVENTS); + store.initialize(true); + commit(journalEvent("stage-1", 1L, "event-1")); + commit(journalEvent("stage-1", 2L, "event-2")); + + // When: only event-2 is acknowledged (GSI reads are eventually consistent, so tolerate retry) + long acknowledged = awaitAcknowledgedCount(Collections.singletonList("event-2"), 1L); + + // Then: exactly one event was updated + assertEquals(1L, acknowledged); + + // And: event-2 is absent from the unacknowledged set once consistent + List> remaining = awaitUnacknowledgedCount(1); + assertEquals("event-1", remaining.get(0).getEventId()); + + // And: empty collection acknowledges nothing + assertEquals(0L, store.acknowledgeEvents(Collections.emptyList())); + + // And: re-acknowledging an already acknowledged event updates nothing + assertEquals(0L, store.acknowledgeEvents(Collections.singletonList("event-2"))); + + Map key = new HashMap<>(); + key.put(JournalEventFieldConstants.KEY_STREAM_ID, + AttributeValue.builder().s("stage-1").build()); + key.put(JournalEventFieldConstants.KEY_STREAM_SEQUENCE, + AttributeValue.builder().n("2").build()); + Map acknowledgedItem = client.getItem(GetItemRequest.builder() + .tableName(TABLE_NAME) + .key(key) + .consistentRead(true) + .build()) + .item(); + assertFalse(acknowledgedItem.containsKey(JournalEventFieldConstants.KEY_PENDING_PARTITION_KEY)); + assertFalse(acknowledgedItem.containsKey(JournalEventFieldConstants.KEY_PENDING_ORDER_KEY)); + } + + private void commit(JournalEvent event) { + commit(Collections.singletonList(event)); + } + + private void commit(List> events) { + DynamoDBTxWrapper txWrapper = new DynamoDBTxWrapper( + client, + new TransactionManager<>(TransactWriteItemsEnhancedRequest::builder)); + txWrapper.wrapInTransaction(new BasicRuntimeContext("session-" + UUID.randomUUID()), ctx -> { + TransactWriteItemsEnhancedRequest.Builder builder = ctx.getContext() + .getRequiredDependencyValue(TransactWriteItemsEnhancedRequest.Builder.class); + for (JournalEvent event : events) { + store.contributeToTransaction(builder, event); + } + return Result.OK(); + }); + } + + private JournalEvent journalEvent(String streamId, long sequence, String eventId) { + AuditEntry auditEntry = new AuditEntry( + "execution-1", + streamId, + "change-" + sequence, + "author", + LocalDateTime.of(2026, 1, 1, 0, 0), + AuditEntry.Status.APPLIED, + AuditEntry.ChangeType.STANDARD_CODE, + "com.example.Change", + "apply", + "Source.java", + 150L, + "host-1", + "metadata", + false, + "no-error", + AuditTxType.NON_TX, + "dynamodb", + "1", + RecoveryStrategy.MANUAL_INTERVENTION, + true); + return new JournalEvent<>(eventId, JournalEventType.CHANGE_STATE, streamId, sequence, Instant.now(), auditEntry); + } + + private List> awaitUnacknowledgedCount(int expected) throws InterruptedException { + return awaitUnacknowledgedCount(10, expected); + } + + private List> awaitUnacknowledgedCount(int limit, int expected) + throws InterruptedException { + List> events = Collections.emptyList(); + for (int attempt = 0; attempt < 50; attempt++) { + events = store.getUnacknowledgedEvents(limit); + if (events.size() == expected) { + return events; + } + Thread.sleep(100); + } + return events; + } + + private long awaitAcknowledgedCount(Collection eventIds, long expected) throws InterruptedException { + for (int attempt = 0; attempt < 50; attempt++) { + long acknowledged = store.acknowledgeEvents(eventIds); + if (acknowledged == expected) { + return acknowledged; + } + Thread.sleep(100); + } + return store.acknowledgeEvents(eventIds); + } + + private boolean tableExists(String tableName) { + return client.listTables().tableNames().contains(tableName); + } + + private void deleteTableIfExists(String tableName) { + if (tableExists(tableName)) { + client.deleteTable(DeleteTableRequest.builder().tableName(tableName).build()); + } + } + +} 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 d301b3b80..ac7fcd71e 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 @@ -171,12 +171,6 @@ public AuditReader getAuditReader() { return () -> auditRepository.getAuditHistory(); } - - @Override - public CommunityAuditPersistence getPersistence() { - throw new RuntimeException("getPersistence shouldn´t be called at MongodbSync "); - } - @Override public synchronized CommunityLockService getLockService() { return lockService; diff --git a/core/flamingock-core/src/main/java/io/flamingock/internal/core/external/store/AuditStore.java b/core/flamingock-core/src/main/java/io/flamingock/internal/core/external/store/AuditStore.java index 3dcbe1201..dfb2f2dda 100644 --- a/core/flamingock-core/src/main/java/io/flamingock/internal/core/external/store/AuditStore.java +++ b/core/flamingock-core/src/main/java/io/flamingock/internal/core/external/store/AuditStore.java @@ -21,15 +21,18 @@ import io.flamingock.internal.common.core.audit.AuditWriter; import io.flamingock.internal.common.core.context.ContextInitializable; import io.flamingock.internal.common.core.audit.AuditPersistence; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.util.FeatureFlag; import java.util.Collections; import java.util.Set; public interface AuditStore extends ExternalSystem, ContextInitializable { - //This will be replaced since we need to have a factory @Deprecated - PERSISTENCE getPersistence(); + default PERSISTENCE getPersistence() { + throw new IllegalStateException("Audit stores must provide persistence through getPersistenceFactory(stageId)"); + } default AuditReader getAuditReader() { return getPersistence(); @@ -37,6 +40,9 @@ default AuditReader getAuditReader() { //TODO temporally default, until we implement the other DB stores default AuditPersistenceFactory getPersistenceFactory() { + if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false)) { + throw new IllegalStateException("Journal-enabled audit stores require getPersistenceFactory(stageId)"); + } return stageId -> getPersistence(); } diff --git a/core/flamingock-core/src/test/java/io/flamingock/internal/core/external/store/AuditStoreTest.java b/core/flamingock-core/src/test/java/io/flamingock/internal/core/external/store/AuditStoreTest.java new file mode 100644 index 000000000..97b514f96 --- /dev/null +++ b/core/flamingock-core/src/test/java/io/flamingock/internal/core/external/store/AuditStoreTest.java @@ -0,0 +1,60 @@ +/* + * 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.core.external.store; + +import io.flamingock.internal.common.core.audit.AuditPersistence; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.util.FeatureFlag; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; + +class AuditStoreTest { + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + } + + @Test + void defaultPersistenceFactoryUsesLegacyPersistenceWhenJournalEventsAreDisabled() { + AuditStore auditStore = mock(AuditStore.class, CALLS_REAL_METHODS); + AuditPersistence persistence = mock(AuditPersistence.class); + doReturn(persistence).when(auditStore).getPersistence(); + + assertSame(persistence, auditStore.getPersistenceFactory().get("stage")); + } + + @Test + void defaultPersistenceFactoryFailsFastWhenJournalEventsAreEnabled() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + AuditStore auditStore = mock(AuditStore.class, CALLS_REAL_METHODS); + + assertThrows(IllegalStateException.class, auditStore::getPersistenceFactory); + } + + @Test + void defaultPersistenceRequiresStageAwareFactory() { + AuditStore auditStore = mock(AuditStore.class, CALLS_REAL_METHODS); + + assertThrows(IllegalStateException.class, auditStore::getPersistence); + } +} diff --git a/core/target-systems/flamingock-dynamodb-externalsystem-api/build.gradle.kts b/core/target-systems/flamingock-dynamodb-externalsystem-api/build.gradle.kts index 1e99c314d..fb0e972c4 100644 --- a/core/target-systems/flamingock-dynamodb-externalsystem-api/build.gradle.kts +++ b/core/target-systems/flamingock-dynamodb-externalsystem-api/build.gradle.kts @@ -1,6 +1,6 @@ val coreApiVersion: String by extra dependencies { - implementation("io.flamingock:flamingock-core-api:${coreApiVersion}") + api(project(":core:flamingock-core-commons")) //General compileOnly("software.amazon.awssdk:dynamodb-enhanced:2.25.29") @@ -12,4 +12,4 @@ java { toolchain { languageVersion.set(JavaLanguageVersion.of(8)) } -} \ No newline at end of file +} diff --git a/core/target-systems/flamingock-dynamodb-externalsystem-api/src/main/java/io/flamingock/externalsystem/dynamodb/api/DynamoDBExternalSystem.java b/core/target-systems/flamingock-dynamodb-externalsystem-api/src/main/java/io/flamingock/externalsystem/dynamodb/api/DynamoDBExternalSystem.java index 749453e12..641fdf7eb 100644 --- a/core/target-systems/flamingock-dynamodb-externalsystem-api/src/main/java/io/flamingock/externalsystem/dynamodb/api/DynamoDBExternalSystem.java +++ b/core/target-systems/flamingock-dynamodb-externalsystem-api/src/main/java/io/flamingock/externalsystem/dynamodb/api/DynamoDBExternalSystem.java @@ -15,9 +15,9 @@ */ package io.flamingock.externalsystem.dynamodb.api; -import io.flamingock.api.external.ExternalSystem; +import io.flamingock.internal.common.core.transaction.TransactionalExternalSystem; import software.amazon.awssdk.services.dynamodb.DynamoDbClient; -public interface DynamoDBExternalSystem extends ExternalSystem { +public interface DynamoDBExternalSystem extends TransactionalExternalSystem { DynamoDbClient getClient(); } diff --git a/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterJournalTest.java b/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterJournalTest.java new file mode 100644 index 000000000..d7549f9c9 --- /dev/null +++ b/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterJournalTest.java @@ -0,0 +1,216 @@ +/* + * 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.importer.mongock.dynamodb; + +import io.flamingock.api.annotations.EnableFlamingock; +import io.flamingock.api.annotations.Stage; +import io.flamingock.common.test.mongock.MongockChangeEntry; +import io.flamingock.common.test.mongock.MongockChangeState; +import io.flamingock.common.test.mongock.MongockChangeType; +import io.flamingock.core.kit.TestKit; +import io.flamingock.dynamodb.kit.DynamoDBTableFactory; +import io.flamingock.dynamodb.kit.DynamoDBTestKit; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.constants.CommunityPersistenceConstants; +import io.flamingock.internal.util.dynamodb.DynamoDBUtil; +import io.flamingock.internal.util.dynamodb.entities.AuditEntryEntity; +import io.flamingock.internal.util.dynamodb.entities.journal.DynamoDBJournalEventMapper; +import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventEntity; +import io.flamingock.store.dynamodb.DynamoDBAuditStore; +import io.flamingock.support.mongock.annotations.MongockSupport; +import io.flamingock.targetsystem.dynamodb.DynamoDBTargetSystem; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.enhanced.dynamodb.TableSchema; +import software.amazon.awssdk.enhanced.dynamodb.model.ScanEnhancedRequest; +import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; + +import java.net.URI; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.util.Arrays; +import java.util.Date; +import java.util.List; +import java.util.stream.Collectors; + +import static io.flamingock.internal.common.core.metadata.Constants.DEFAULT_MONGOCK_ORIGIN; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@Testcontainers +@MongockSupport(targetSystem = "dynamodb-target-system") +@EnableFlamingock(stages = {@Stage(location = "io.flamingock.importer.mongock.dynamodb.changes")}) +class DynamoDBImporterJournalTest { + + private static final String LEGACY_STAGE_ID = "flamingock-legacy-stage"; + private static final String IMPORTED_BEFORE_CHANGE_ID = "create-users-table_before"; + private static final String IMPORTED_CHANGE_ID = "create-users-table"; + private static final Instant HISTORICAL_TIMESTAMP = Instant.parse("2025-06-19T05:43:57.132Z"); + + @Container + static final GenericContainer dynamoDBContainer = new GenericContainer<>("amazon/dynamodb-local:latest") + .withExposedPorts(8000); + + private DynamoDbClient client; + private DynamoDBTargetSystem targetSystem; + private DynamoDBMongockTestHelper mongockTestHelper; + private TestKit testKit; + + @BeforeEach + void setUp() { + String endpoint = String.format("http://%s:%d", + dynamoDBContainer.getHost(), + dynamoDBContainer.getMappedPort(8000)); + client = DynamoDbClient.builder() + .endpointOverride(URI.create(endpoint)) + .region(Region.US_EAST_1) + .httpClient(UrlConnectionHttpClient.builder().build()) + .credentialsProvider(StaticCredentialsProvider.create( + AwsBasicCredentials.create("dummy", "dummy"))) + .build(); + + DynamoDBTableFactory.createMongockTable(client, DEFAULT_MONGOCK_ORIGIN); + mongockTestHelper = new DynamoDBMongockTestHelper(client, DEFAULT_MONGOCK_ORIGIN); + targetSystem = new DynamoDBTargetSystem("dynamodb-target-system", client); + testKit = DynamoDBTestKit.create( + client, + DynamoDBAuditStore.from(targetSystem)); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + mongockTestHelper.reset(); + testKit.cleanUp(); + client.close(); + } + + @Test + @DisplayName("journal-enabled imports retain current audit records and publish full payloads on the legacy stream") + void journalEnabledImportKeepsCurrentAuditRecordsAndUsesLegacyStream() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + mongockTestHelper.write(new MongockChangeEntry( + "legacy-execution-before", + IMPORTED_BEFORE_CHANGE_ID, + "mongock", + Date.from(Instant.parse("2025-06-19T05:43:57.094Z")), + MongockChangeState.EXECUTED, + MongockChangeType.BEFORE_EXECUTION, + "io.mongock.examples.mongodb.standalone.mondogb.sync.migration.initializer.ClientInitializerChangeUnit", + "beforeExecution", + "legacy-before-metadata", + 25L, + "legacy-host", + null, + false, + null)); + mongockTestHelper.write(new MongockChangeEntry( + "legacy-execution", + IMPORTED_CHANGE_ID, + "mongock", + Date.from(HISTORICAL_TIMESTAMP), + MongockChangeState.EXECUTED, + MongockChangeType.EXECUTION, + "io.mongock.examples.mongodb.standalone.mondogb.sync.migration.initializer.ClientInitializerChangeUnit", + "apply", + "legacy-metadata", + 23L, + "legacy-host", + null, + false, + null)); + + Instant runStartedAt = Instant.now(); + testKit.createBuilder() + .addTargetSystem(targetSystem) + .build() + .run(); + Instant runFinishedAt = Instant.now(); + + List importedAuditRecords = storedAuditRecords().stream() + .filter(entry -> Arrays.asList(IMPORTED_BEFORE_CHANGE_ID, IMPORTED_CHANGE_ID) + .contains(entry.getChangeId())) + .collect(Collectors.toList()); + assertEquals(2, importedAuditRecords.size()); + assertTrue(importedAuditRecords.stream().anyMatch(entry -> + IMPORTED_BEFORE_CHANGE_ID.equals(entry.getPartitionKey()))); + assertTrue(importedAuditRecords.stream().anyMatch(entry -> + IMPORTED_CHANGE_ID.equals(entry.getPartitionKey())), + "flag ON must use the changeId-only current-state key for imported audit entries"); + + List> importedEvents = storedEvents().stream() + .filter(event -> Arrays.asList(IMPORTED_BEFORE_CHANGE_ID, IMPORTED_CHANGE_ID) + .contains(event.getData().getChangeId())) + .collect(Collectors.toList()); + assertEquals(2, importedEvents.size()); + assertTrue(importedEvents.stream().allMatch(event -> LEGACY_STAGE_ID.equals(event.getStreamId()))); + assertTrue(importedEvents.stream().allMatch(event -> + !event.getData().getStageId().equals(event.getStreamId()))); + assertEquals(Arrays.asList(1L, 2L), importedEvents.stream() + .map(JournalEvent::getStreamSequence) + .sorted() + .collect(Collectors.toList())); + + JournalEvent importedEvent = importedEvents.stream() + .filter(event -> IMPORTED_CHANGE_ID.equals(event.getData().getChangeId())) + .findFirst() + .orElseThrow(() -> new AssertionError("Expected the imported execution event")); + AuditEntry importedEntry = importedEvent.getData(); + assertEquals(LocalDateTime.ofInstant(HISTORICAL_TIMESTAMP, ZoneId.systemDefault()), + importedEntry.getCreatedAt()); + assertEquals("legacy-execution", importedEntry.getExecutionId()); + assertEquals("legacy-metadata", importedEntry.getMetadata()); + assertEquals(23L, importedEntry.getExecutionMillis()); + assertEquals("legacy-host", importedEntry.getExecutionHostname()); + assertTrue(!importedEvent.getOccurredAt().isBefore(runStartedAt)); + assertTrue(!importedEvent.getOccurredAt().isAfter(runFinishedAt)); + } + + private List storedAuditRecords() { + return new DynamoDBUtil(client) + .getEnhancedClient() + .table(CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME, + TableSchema.fromBean(AuditEntryEntity.class)) + .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) + .items() + .stream() + .collect(Collectors.toList()); + } + + private List> storedEvents() { + return new DynamoDBUtil(client) + .getEnhancedClient() + .table("flamingockJournalEvents", TableSchema.fromBean(JournalEventEntity.class)) + .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) + .items() + .stream() + .map(DynamoDBJournalEventMapper::fromEntity) + .collect(Collectors.toList()); + } +} diff --git a/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterTest.java b/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterTest.java index d4818fbcf..056843e54 100644 --- a/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterTest.java +++ b/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterTest.java @@ -23,8 +23,10 @@ import io.flamingock.dynamodb.kit.DynamoDBTableFactory; import io.flamingock.dynamodb.kit.DynamoDBTestKit; import io.flamingock.internal.common.core.response.data.ErrorInfo; +import io.flamingock.internal.common.core.feature.Features; import io.flamingock.internal.core.operation.StagedExecuteOperationException; import io.flamingock.internal.core.builder.runner.Runner; +import io.flamingock.internal.util.FeatureFlag; import io.flamingock.support.mongock.annotations.MongockSupport; import io.flamingock.targetsystem.dynamodb.DynamoDBTargetSystem; import org.junit.jupiter.api.AfterEach; @@ -99,6 +101,7 @@ void setUp() { @AfterEach void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); // DynamoDB local doesn't need explicit cleanup between tests // Tables are automatically cleaned by Testcontainers on restart mongockTestHelper.reset(); diff --git a/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/AuditEntryEntity.java b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/AuditEntryEntity.java index 172dab756..637929ef1 100644 --- a/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/AuditEntryEntity.java +++ b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/AuditEntryEntity.java @@ -181,7 +181,7 @@ public void setSourceFile(String sourceFile) { @DynamoDbAttribute(AuditEntryFieldConstants.KEY_METADATA) public String getMetadata() { - return metadata.toString(); + return metadata != null ? metadata.toString() : null; } public void setMetadata(Object metadata) { @@ -208,7 +208,7 @@ public void setExecutionHostname(String executionHostname) { @DynamoDbAttribute(AuditEntryFieldConstants.KEY_ERROR_TRACE) public String getErrorTrace() { - return errorTrace.toString(); + return errorTrace != null ? errorTrace.toString() : null; } public void setErrorTrace(Object errorTrace) { diff --git a/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/DynamoDBJournalEventMapper.java b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/DynamoDBJournalEventMapper.java new file mode 100644 index 000000000..d51c0e51e --- /dev/null +++ b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/DynamoDBJournalEventMapper.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.internal.util.dynamodb.entities.journal; + +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.core.journal.JournalEventType; +import io.flamingock.internal.util.JsonObjectMapper; +import io.flamingock.internal.util.dynamodb.entities.AuditEntryEntity; + +import java.time.Instant; + +/** + * Maps a {@link JournalEvent} carrying an {@link AuditEntry} payload to/from a + * {@link JournalEventEntity}. + *

+ * The {@code data} payload is embedded as the JSON serialization of an {@link AuditEntryEntity}, + * so the audit representation stays single-sourced. {@code occurredAt} is stored as an ISO-8601 + * string. {@code pendingPartitionKey} and {@code pendingOrderKey} are present while the event is + * unacknowledged and absent once acknowledged. + *

+ * Only {@link JournalEventType#CHANGE_STATE} events carry an {@link AuditEntry} payload today. + * Other event types carry different payloads and are not yet implemented, so this mapper rejects + * them rather than silently mis-mapping their data as an audit entry. + */ +public final class DynamoDBJournalEventMapper { + + /** The only event type whose {@code data} is an {@link AuditEntry} and is supported for now. */ + private static final JournalEventType SUPPORTED_EVENT_TYPE = JournalEventType.CHANGE_STATE; + private static final int MAX_DYNAMODB_SORT_KEY_BYTES = 1024; + private static final char[] HEX_DIGITS = "0123456789ABCDEF".toCharArray(); + + private DynamoDBJournalEventMapper() { + } + + public static JournalEventEntity toEntity(JournalEvent event) { + requireSupportedType(event.getEventType()); + JournalEventEntity entity = new JournalEventEntity(); + entity.setEventId(event.getEventId()); + entity.setEventType(event.getEventType().name()); + entity.setEventVersion(event.getEventVersion()); + entity.setStreamId(event.getStreamId()); + entity.setStreamSequence(event.getStreamSequence()); + entity.setOccurredAt(event.getOccurredAt().toString()); + if (event.isAcknowledged()) { + entity.setPendingPartitionKey(null); + entity.setPendingOrderKey(null); + } else { + entity.setPendingPartitionKey(JournalEventFieldConstants.PENDING_PARTITION_VALUE); + entity.setPendingOrderKey(pendingOrderKey(event.getStreamId(), event.getStreamSequence())); + } + entity.setPayload(serializePayload(event.getData())); + return entity; + } + + public static JournalEvent fromEntity(JournalEventEntity entity) { + JournalEventType eventType = JournalEventType.valueOf(entity.getEventType()); + requireSupportedType(eventType); + AuditEntry data = deserializePayload(entity.getPayload()); + Instant occurredAt = Instant.parse(entity.getOccurredAt()); + boolean partitionKeyMissing = entity.getPendingPartitionKey() == null; + boolean orderKeyMissing = entity.getPendingOrderKey() == null; + if (partitionKeyMissing != orderKeyMissing) { + throw new IllegalStateException("Journal event pending keys must be present together"); + } + boolean acknowledged = partitionKeyMissing; + return new JournalEvent<>( + entity.getEventId(), + eventType, + entity.getEventVersion() != null ? entity.getEventVersion() : JournalEvent.DEFAULT_VERSION, + entity.getStreamId(), + entity.getStreamSequence(), + occurredAt, + data, + acknowledged); + } + + /** + * Encodes a journal position so DynamoDB's lexicographic sort order matches stream position order. + * Java UTF-16 code units are encoded independently to avoid delimiter collisions, and the positive + * sequence is rendered as a fixed-width hexadecimal value. + * + * @param streamId stream identifier + * @param streamSequence positive stream sequence + * @return collision-safe pending index sort key + */ + public static String pendingOrderKey(String streamId, Long streamSequence) { + if (streamId == null || streamSequence == null) { + throw new IllegalArgumentException("streamId and streamSequence are required"); + } + if (streamSequence < 0) { + throw new IllegalArgumentException("streamSequence must be non-negative"); + } + + StringBuilder encoded = new StringBuilder(streamId.length() * 4 + 17); + for (char character : streamId.toCharArray()) { + encoded.append(HEX_DIGITS[(character >>> 12) & 0x0F]); + encoded.append(HEX_DIGITS[(character >>> 8) & 0x0F]); + encoded.append(HEX_DIGITS[(character >>> 4) & 0x0F]); + encoded.append(HEX_DIGITS[character & 0x0F]); + } + encoded.append('!'); + encoded.append(String.format("%016X", streamSequence)); + if (encoded.length() > MAX_DYNAMODB_SORT_KEY_BYTES) { + throw new IllegalArgumentException("pendingOrderKey exceeds DynamoDB's 1024-byte sort-key limit"); + } + return encoded.toString(); + } + + private static String serializePayload(AuditEntry auditEntry) { + try { + return JsonObjectMapper.DEFAULT_INSTANCE.writeValueAsString(new AuditEntryEntity(auditEntry)); + } catch (Exception e) { + throw new IllegalStateException("Failed to serialize journal event payload", e); + } + } + + private static AuditEntry deserializePayload(String payload) { + try { + return JsonObjectMapper.DEFAULT_INSTANCE.readValue(payload, AuditEntryEntity.class).toAuditEntry(); + } catch (Exception e) { + throw new IllegalStateException("Failed to deserialize journal event payload", e); + } + } + + private static void requireSupportedType(JournalEventType eventType) { + if (eventType != SUPPORTED_EVENT_TYPE) { + throw new UnsupportedOperationException( + "DynamoDBJournalEventMapper only supports " + SUPPORTED_EVENT_TYPE + " events (AuditEntry payload); " + + "event type " + eventType + " is not yet implemented"); + } + } +} diff --git a/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/JournalEventEntity.java b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/JournalEventEntity.java new file mode 100644 index 000000000..721ef4e80 --- /dev/null +++ b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/JournalEventEntity.java @@ -0,0 +1,125 @@ +/* + * 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.util.dynamodb.entities.journal; + +import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbBean; +import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbPartitionKey; +import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSecondaryPartitionKey; +import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSecondarySortKey; +import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSortKey; + +/** + * DynamoDB persistence representation of a {@code JournalEvent} carrying an {@code AuditEntry} payload. + *

+ * Base key: {@code (streamId, streamSequence)}. {@code pendingPartitionKey} and {@code pendingOrderKey} are + * sparse attributes: they are only present while the event has not been acknowledged, which is what makes + * the event visible in {@code PendingEventsIndex}. The payload is the JSON serialization of the embedded + * {@code AuditEntryEntity} + * ({@code AuditEntry} itself has no no-arg constructor, so it cannot be a {@code @DynamoDbBean}). + */ +@DynamoDbBean +public class JournalEventEntity { + + private String streamId; + private Long streamSequence; + private String pendingPartitionKey; + private String pendingOrderKey; + private String eventId; + private String eventType; + private String occurredAt; + private Integer eventVersion; + private String payload; + + public JournalEventEntity() { + } + + @DynamoDbPartitionKey + public String getStreamId() { + return streamId; + } + + public void setStreamId(String streamId) { + this.streamId = streamId; + } + + @DynamoDbSortKey + public Long getStreamSequence() { + return streamSequence; + } + + public void setStreamSequence(Long streamSequence) { + this.streamSequence = streamSequence; + } + + @DynamoDbSecondaryPartitionKey(indexNames = JournalEventFieldConstants.PENDING_EVENTS_INDEX) + public String getPendingPartitionKey() { + return pendingPartitionKey; + } + + public void setPendingPartitionKey(String pendingPartitionKey) { + this.pendingPartitionKey = pendingPartitionKey; + } + + @DynamoDbSecondarySortKey(indexNames = JournalEventFieldConstants.PENDING_EVENTS_INDEX) + public String getPendingOrderKey() { + return pendingOrderKey; + } + + public void setPendingOrderKey(String pendingOrderKey) { + this.pendingOrderKey = pendingOrderKey; + } + + @DynamoDbSecondaryPartitionKey(indexNames = JournalEventFieldConstants.EVENT_ID_INDEX) + public String getEventId() { + return eventId; + } + + public void setEventId(String eventId) { + this.eventId = eventId; + } + + public String getEventType() { + return eventType; + } + + public void setEventType(String eventType) { + this.eventType = eventType; + } + + public String getOccurredAt() { + return occurredAt; + } + + public void setOccurredAt(String occurredAt) { + this.occurredAt = occurredAt; + } + + public Integer getEventVersion() { + return eventVersion; + } + + public void setEventVersion(Integer eventVersion) { + this.eventVersion = eventVersion; + } + + public String getPayload() { + return payload; + } + + public void setPayload(String payload) { + this.payload = payload; + } +} 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 new file mode 100644 index 000000000..6b55747cb --- /dev/null +++ b/utils/dynamodb-util/src/main/java/io/flamingock/internal/util/dynamodb/entities/journal/JournalEventFieldConstants.java @@ -0,0 +1,45 @@ +/* + * 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.util.dynamodb.entities.journal; + +/** + * Attribute and index names of the DynamoDB journal events table + * ({@code flamingockJournalEvents}). + *

+ * The base key is {@code (streamId, streamSequence)}; the sparse pending GSI + * ({@link #PENDING_EVENTS_INDEX}) carries only items with a {@code pendingPartitionKey} and + * {@code pendingOrderKey}, + * and the non-unique eventId GSI ({@link #EVENT_ID_INDEX}) serves acknowledgement + * lookups only. Event identity is enforced transactionally by a reserved item in this table. + */ +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"; + public static final String KEY_PENDING_ORDER_KEY = "pendingOrderKey"; + public static final String KEY_EVENT_ID = "eventId"; + + public static final String PENDING_PARTITION_VALUE = "pending"; + + public static final String PENDING_EVENTS_INDEX = "PendingEventsIndex"; + public static final String EVENT_ID_INDEX = "EventIdIndex"; + + private JournalEventFieldConstants() { + } +}