Skip to content

feat(dynamodb): add journal event persistence #948

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bercianor wants to merge 3 commits into
base: master
Choose a base branch
from
Open

feat(dynamodb): add journal event persistence #948

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3a54575
feat(dynamodb): add journal event persistence
bercianor Aug 3, 2026
84cec64
fix: comments in the pr
bercianor Aug 4, 2026
e5f8c99
fix: remove reservation and others architectural changes
bercianor Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 91 additions & 10 deletions ...ck-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/DynamoDBAuditStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,50 +15,65 @@
*/
package io.flamingock.store.dynamodb;

import io.flamingock.internal.common.core.audit.AuditPersistenceFactory;
import io.flamingock.internal.common.core.context.ContextResolver;
import io.flamingock.internal.common.core.error.FlamingockException;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.common.core.pipeline.PipelineHelper;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.external.store.CommunityAuditStore;
import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence;
import io.flamingock.internal.core.external.store.lock.community.CommunityLockService;
import io.flamingock.internal.core.journal.JournalEventSequencer;
import io.flamingock.internal.core.journal.JournalEventSequencerFactory;
import io.flamingock.internal.util.Constants;
import io.flamingock.internal.util.FeatureFlag;
import io.flamingock.internal.util.TimeService;
import io.flamingock.internal.util.constants.CommunityPersistenceConstants;
import io.flamingock.internal.util.id.RunnerId;
import io.flamingock.store.dynamodb.internal.DynamoDBAuditPersistence;
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 static final String DEFAULT_JOURNAL_REPOSITORY_NAME = "flamingockJournalEvents";

private final DynamoDbClient client;
private final DynamoDBExternalSystem targetSystem;
private RunnerId runnerId;
private CommunityConfigurable communityConfiguration;
private DynamoDBAuditPersistence persistence;
private DynamoDBAuditPersistence legacyPersistence;
private DynamoDBLockService lockService;
private TransactionWrapper txWrapper;
private DynamoDBJournalEventStore journalEventStore;
private JournalEventSequencerFactory journalEventSequencerFactory;
private String auditRepositoryName = CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME;
private String lockRepositoryName = CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME;
private String journalRepositoryName = DEFAULT_JOURNAL_REPOSITORY_NAME;
private long readCapacityUnits = 5L;
private long writeCapacityUnits = 5L;
private boolean autoCreate = true;

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}.
* <p>
* 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
Expand All @@ -76,6 +91,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;
Expand All @@ -95,22 +115,71 @@ public DynamoDBAuditStore withAutoCreate(boolean autoCreate) {
public void initialize(ContextResolver baseContext) {
runnerId = baseContext.getRequiredDependencyValue(RunnerId.class);
communityConfiguration = baseContext.getRequiredDependencyValue(CommunityConfigurable.class);
txWrapper = targetSystem.getTxWrapper();
journalEventStore = new DynamoDBJournalEventStore(
client,
journalRepositoryName,
readCapacityUnits,
writeCapacityUnits);
journalEventSequencerFactory = new JournalEventSequencerFactory(journalEventStore);
this.validate();
}

@Override
public synchronized CommunityAuditPersistence getPersistence() {
if (persistence == null) {
persistence = new DynamoDBAuditPersistence(
if (legacyPersistence == null) {
if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS)) {
journalEventStore.initialize(autoCreate);
JournalEventSequencer journalEventSequencer =
journalEventSequencerFactory.forStream(PipelineHelper.LEGACY_STAGE_ID);
Comment thread
bercianor marked this conversation as resolved.
Outdated
legacyPersistence = new DynamoDBAuditPersistence(
client,
txWrapper,
journalEventStore,
journalEventSequencer,
journalEventSequencerFactory,
auditRepositoryName,
readCapacityUnits,
writeCapacityUnits,
autoCreate,
communityConfiguration);
} else {
legacyPersistence = new DynamoDBAuditPersistence(
client,
auditRepositoryName,
readCapacityUnits,
writeCapacityUnits,
autoCreate,
communityConfiguration);
}
legacyPersistence.initialize(runnerId);
}
return legacyPersistence;
}

@Override
public AuditPersistenceFactory<CommunityAuditPersistence> getPersistenceFactory() {
return stageId -> {
if (!FeatureFlag.isEnabled(Features.JOURNAL_EVENTS)) {
return getPersistence();
}

journalEventStore.initialize(autoCreate);
JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId);
DynamoDBAuditPersistence persistence = new DynamoDBAuditPersistence(
client,
txWrapper,
journalEventStore,
journalEventSequencer,
journalEventSequencerFactory,
auditRepositoryName,
readCapacityUnits,
writeCapacityUnits,
autoCreate,
communityConfiguration);
persistence.initialize(runnerId);
}
return persistence;
return persistence;
};
}

@Override
Expand Down Expand Up @@ -140,8 +209,20 @@ 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 (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.");
}
}
}
92 changes: 92 additions & 0 deletions ...itstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,19 @@
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.enhanced.dynamodb.model.TransactWriteItemsEnhancedRequest;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;

import java.util.List;
Expand All @@ -31,6 +40,10 @@ public class DynamoDBAuditPersistence extends AbstractCommunityAuditPersistence
private final long readCapacityUnits;
private final long writeCapacityUnits;
private final boolean autoCreate;
private final TransactionWrapper txWrapper;
private final DynamoDBJournalEventStore journalEventStore;
private final JournalEventSequencer journalEventSequencer;
private final JournalEventSequencerFactory journalEventSequencerFactory;

private DynamoDBAuditor auditor;

Expand All @@ -40,12 +53,70 @@ public DynamoDBAuditPersistence(DynamoDbClient client,
long writeCapacityUnits,
boolean autoCreate,
CommunityConfigurable localConfiguration) {
this(client, null, null, null, null, auditTableName, readCapacityUnits, writeCapacityUnits,
autoCreate, localConfiguration);
}

/**
* Creates a persistence that can atomically stage an audit record and its journal event.
*
* @param client DynamoDB client used by the audit and journal stores
* @param txWrapper transaction wrapper shared with the target system
* @param journalEventStore journal store receiving staged events
* @param journalEventSequencer per-stage sequencer for the journal stream
* @param journalEventSequencerFactory factory used to route imported events to their destination stage
* @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
* @param localConfiguration community configuration
*/
Comment thread
bercianor marked this conversation as resolved.
public DynamoDBAuditPersistence(DynamoDbClient client,
Comment thread
bercianor marked this conversation as resolved.
Outdated
TransactionWrapper txWrapper,
DynamoDBJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
JournalEventSequencerFactory journalEventSequencerFactory,
String auditTableName,
long readCapacityUnits,
long writeCapacityUnits,
boolean autoCreate,
CommunityConfigurable localConfiguration) {
super(localConfiguration);
this.client = client;
this.auditTableName = auditTableName;
this.readCapacityUnits = readCapacityUnits;
this.writeCapacityUnits = writeCapacityUnits;
this.autoCreate = autoCreate;
this.txWrapper = txWrapper;
this.journalEventStore = journalEventStore;
this.journalEventSequencer = journalEventSequencer;
this.journalEventSequencerFactory = journalEventSequencerFactory;
}

/**
* Creates a persistence with a fixed journal stream sequencer.
*
* @param client DynamoDB client used by the audit and journal stores
* @param txWrapper transaction wrapper shared with the target system
* @param journalEventStore journal store receiving staged events
* @param journalEventSequencer sequencer for the persistence stream
* @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
* @param localConfiguration community configuration
*/
public DynamoDBAuditPersistence(DynamoDbClient client,
TransactionWrapper txWrapper,
DynamoDBJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
String auditTableName,
long readCapacityUnits,
long writeCapacityUnits,
boolean autoCreate,
CommunityConfigurable localConfiguration) {
this(client, txWrapper, journalEventStore, journalEventSequencer, null, auditTableName,
readCapacityUnits, writeCapacityUnits, autoCreate, localConfiguration);
}

@Override
Expand All @@ -56,6 +127,9 @@ protected void doInitialize(RunnerId runnerId) {
auditTableName,
readCapacityUnits,
writeCapacityUnits);
if (journalEventStore != null) {
FeatureFlag.ifEnabled(Features.JOURNAL_EVENTS, () -> journalEventStore.initialize(autoCreate));
}
}

@Override
Expand All @@ -65,6 +139,24 @@ public List<AuditEntry> getAuditHistory() {

@Override
public Result writeEntry(AuditEntry auditEntry) {
if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS)) {
if (txWrapper == null || journalEventStore == null || journalEventSequencer == null) {
throw new IllegalStateException("Journal-enabled persistence requires transaction and journal wiring");
}
JournalEventSequencer sequencer = journalEventSequencer;
Comment thread
bercianor marked this conversation as resolved.
Outdated
RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
TransactWriteItemsEnhancedRequest.Builder builder = runtimeContext.getContext()
.getRequiredDependencyValue(TransactWriteItemsEnhancedRequest.Builder.class);
JournalEvent<AuditEntry> journalEvent = sequencer.newEvent(auditEntry);
journalEventStore.write(builder, journalEvent);
auditor.stageWrite(builder, auditEntry);
return Result.OK();
});
sequencer.confirm();
return result;
}
return auditor.writeEntry(auditEntry);
}

}
19 changes: 19 additions & 0 deletions ...amodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
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.ConditionalCheckFailedException;

Expand Down Expand Up @@ -78,6 +79,24 @@ public Result writeEntry(AuditEntry auditEntry) {
return Result.OK();
}

/**
* Stages a current-state audit write in a caller-owned DynamoDB transaction.
*
* <p>The change identifier is the partition key in journal mode, so successive state transitions replace
* the current audit record while the journal keeps the complete transition history.</p>
*
* @param builder the transaction builder that owns the audit write
* @param auditEntry the audit entry to stage
*/
void stageWrite(TransactWriteItemsEnhancedRequest.Builder builder, AuditEntry auditEntry) {
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());
}

@Override
public List<AuditEntry> getAuditHistory() {
return table
Expand Down
Loading
Loading