diff --git a/.gitignore b/.gitignore index 66ef0d3f3..f2195191f 100644 --- a/.gitignore +++ b/.gitignore @@ -103,3 +103,5 @@ test_*.db-shm # Generic SQLite artifacts *.db-wal *.db-shm +# Local Pi runtime state +.atl/ diff --git a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java index 538e54a86..3103546dc 100644 --- a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java +++ b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java @@ -21,22 +21,31 @@ import com.mongodb.reactivestreams.client.ClientSession; import com.mongodb.reactivestreams.client.MongoDatabase; import io.flamingock.externalsystem.mongodb.reactive.api.MongoDBReactiveExternalSystem; +import io.flamingock.internal.common.core.audit.AuditPersistenceFactory; +import io.flamingock.internal.common.core.audit.AuditReader; import io.flamingock.internal.common.core.context.ContextResolver; import io.flamingock.internal.common.core.error.FlamingockException; +import io.flamingock.internal.common.core.feature.Features; import io.flamingock.internal.core.configuration.community.CommunityConfigurable; import io.flamingock.internal.core.external.store.CommunityAuditStore; import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence; import io.flamingock.internal.core.external.store.lock.community.CommunityLockService; +import io.flamingock.internal.core.journal.JournalEventSequencer; +import io.flamingock.internal.core.journal.JournalEventSequencerFactory; import io.flamingock.internal.util.Constants; +import io.flamingock.internal.util.FeatureFlag; import io.flamingock.internal.util.TimeService; import io.flamingock.internal.util.id.RunnerId; import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveAuditPersistence; +import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveAuditRepository; +import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveJournalEventStore; import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveLockService; import java.util.Collections; import java.util.HashSet; import java.util.Set; +import static io.flamingock.internal.common.mongodb.journal.JournalEventPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME; import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME; import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME; @@ -46,15 +55,19 @@ public class MongoDBReactiveAuditStore implements CommunityAuditStore { protected RunnerId runnerId; private CommunityConfigurable communityConfiguration; - private MongoDBReactiveAuditPersistence persistence; + private CommunityAuditPersistence persistence; private MongoDBReactiveLockService lockService; private MongoDatabase database; private String auditRepositoryName = DEFAULT_AUDIT_STORE_NAME; private String lockRepositoryName = DEFAULT_LOCK_STORE_NAME; + private String journalRepositoryName = DEFAULT_JOURNAL_STORE_NAME; private ReadConcern readConcern = ReadConcern.MAJORITY; private ReadPreference readPreference = ReadPreference.primary(); private WriteConcern writeConcern = WriteConcern.MAJORITY.withJournal(true); private boolean autoCreate = true; + private MongoDBReactiveAuditRepository auditRepository; + private MongoDBReactiveJournalEventStore journalEventStore; + private JournalEventSequencerFactory journalEventSequencerFactory; private MongoDBReactiveAuditStore(MongoDBReactiveExternalSystem mongoDBTargetSystem) { this.mongoDBTargetSystem = mongoDBTargetSystem; @@ -89,6 +102,11 @@ public MongoDBReactiveAuditStore withLockRepositoryName(String lockRepositoryNam return this; } + public MongoDBReactiveAuditStore withJournalRepositoryName(String journalRepositoryName) { + this.journalRepositoryName = journalRepositoryName; + return this; + } + public MongoDBReactiveAuditStore withReadConcern(ReadConcern readConcern) { this.readConcern = readConcern; return this; @@ -114,40 +132,57 @@ public void initialize(ContextResolver baseContext) { runnerId = baseContext.getRequiredDependencyValue(RunnerId.class); communityConfiguration = baseContext.getRequiredDependencyValue(CommunityConfigurable.class); database = mongoDBTargetSystem.getMongoDatabase(); - this.validate(); + this.validate(); + + auditRepository = new MongoDBReactiveAuditRepository( + database, auditRepositoryName, readConcern, readPreference, writeConcern); + journalEventStore = new MongoDBReactiveJournalEventStore( + database, journalRepositoryName, readConcern, readPreference, writeConcern); + journalEventSequencerFactory = new JournalEventSequencerFactory(journalEventStore); + + lockService = new MongoDBReactiveLockService( + database, + lockRepositoryName, + readConcern, + readPreference, + writeConcern, + TimeService.getDefault() + ); + lockService.initialize(autoCreate); } @Override - public synchronized CommunityAuditPersistence getPersistence() { - if (persistence == null) { - persistence = new MongoDBReactiveAuditPersistence( + public AuditPersistenceFactory getPersistenceFactory() { + return stageId -> { + auditRepository.initialize(autoCreate); + if (isJournalEventsEnabled()) { + journalEventStore.initialize(autoCreate); + } + JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId); + MongoDBReactiveAuditPersistence stagePersistence = new MongoDBReactiveAuditPersistence( communityConfiguration, - database, - auditRepositoryName, - readConcern, - readPreference, - writeConcern, + auditRepository, + journalEventStore, + journalEventSequencer, + mongoDBTargetSystem.getTxWrapper(), autoCreate ); - persistence.initialize(runnerId); - } - return persistence; + stagePersistence.initialize(runnerId); + if (persistence == null) { + persistence = stagePersistence; + } + return stagePersistence; + }; } @Override - public synchronized CommunityLockService getLockService() { - if (lockService == null) { - lockService = new MongoDBReactiveLockService( - database, - lockRepositoryName, - readConcern, - readPreference, - writeConcern, - TimeService.getDefault() - ); - lockService.initialize(autoCreate); + public AuditReader getAuditReader() { + auditRepository.initialize(autoCreate); + return () -> auditRepository.getAuditHistory(); + } - } + @Override + public synchronized CommunityLockService getLockService() { return lockService; } @@ -165,10 +200,22 @@ 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."); + } + if (readConcern == null) { throw new FlamingockException("The 'readConcern' property is required."); } @@ -182,6 +229,14 @@ private void validate() { } } + private static boolean isJournalEventsEnabled() { + try { + return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false); + } catch (RuntimeException exception) { + return false; + } + } + @Override public Set> getNonGuardedTypes() { return new HashSet<>(Collections.singletonList(ClientSession.class)); diff --git a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java index 5e2c06587..7936f6c1f 100644 --- a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java +++ b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java @@ -21,27 +21,47 @@ import com.mongodb.reactivestreams.client.ClientSession; import com.mongodb.reactivestreams.client.MongoDatabase; 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.context.BasicRuntimeContext; import io.flamingock.internal.core.configuration.community.CommunityConfigurable; import io.flamingock.internal.core.external.store.audit.community.AbstractCommunityAuditPersistence; +import io.flamingock.internal.core.journal.JournalEventSequencer; +import io.flamingock.internal.util.FeatureFlag; import io.flamingock.internal.util.Result; import io.flamingock.internal.util.id.RunnerId; -import java.util.Collections; -import java.util.HashSet; import java.util.List; -import java.util.Set; + +import static io.flamingock.internal.common.mongodb.journal.JournalEventPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME; public class MongoDBReactiveAuditPersistence extends AbstractCommunityAuditPersistence { - private MongoDBReactiveAuditor auditor; - private final MongoDatabase database; - private final String auditCollectionName; - private final ReadConcern readConcern; - private final ReadPreference readPreference; - private final WriteConcern writeConcern; + private final MongoDBReactiveAuditRepository auditRepository; + private final MongoDBReactiveJournalEventStore journalEventStore; + private final JournalEventSequencer journalEventSequencer; + private final TransactionWrapper txWrapper; private final boolean autoCreate; + public MongoDBReactiveAuditPersistence(CommunityConfigurable localConfiguration, + MongoDBReactiveAuditRepository auditRepository, + MongoDBReactiveJournalEventStore journalEventStore, + JournalEventSequencer journalEventSequencer, + TransactionWrapper txWrapper, + boolean autoCreate) { + super(localConfiguration); + this.auditRepository = auditRepository; + this.journalEventStore = journalEventStore; + this.journalEventSequencer = journalEventSequencer; + this.txWrapper = txWrapper; + this.autoCreate = autoCreate; + } + /** + * Backward-compatible constructor for callers that only need the historical audit path. + */ public MongoDBReactiveAuditPersistence(CommunityConfigurable localConfiguration, MongoDatabase database, String auditCollectionName, @@ -49,31 +69,58 @@ public MongoDBReactiveAuditPersistence(CommunityConfigurable localConfiguration, ReadPreference readPreference, WriteConcern writeConcern, boolean autoCreate) { - super(localConfiguration); - this.database = database; - this.auditCollectionName = auditCollectionName; - this.readConcern = readConcern; - this.readPreference = readPreference; - this.writeConcern = writeConcern; - this.autoCreate = autoCreate; + this( + localConfiguration, + new MongoDBReactiveAuditRepository(database, auditCollectionName, readConcern, readPreference, writeConcern), + new MongoDBReactiveJournalEventStore(database, DEFAULT_JOURNAL_STORE_NAME, + readConcern, readPreference, writeConcern), + null, + null, + autoCreate); } @Override protected void doInitialize(RunnerId runnerId) { - //Auditor - auditor = new MongoDBReactiveAuditor(database, auditCollectionName, readConcern, readPreference, writeConcern); - auditor.initialize(autoCreate); + auditRepository.initialize(autoCreate); + 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()) { + return auditRepository.append(auditEntry); + } + + if (journalEventStore == null || journalEventSequencer == null || txWrapper == null) { + throw new IllegalStateException("MongoDB reactive journal writes require a transaction wrapper and sequencer"); + } + + RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId()); + Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> { + ClientSession clientSession = runtimeContext.getContext().getRequiredDependencyValue(ClientSession.class); + JournalEvent journalEvent = journalEventSequencer.newEvent(auditEntry); + journalEventStore.append(clientSession, journalEvent); + return auditRepository.save(clientSession, auditEntry); + }); + + // Result cannot represent FailedStep. The transaction wrapper has therefore committed successfully + // whenever control reaches this line; only then is the in-memory stream position spent. + journalEventSequencer.confirm(); + return result; } + private static boolean isJournalEventsEnabled() { + try { + return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false); + } catch (RuntimeException exception) { + return false; + } + } } diff --git a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditor.java b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditRepository.java similarity index 54% rename from community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditor.java rename to community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditRepository.java index dff532d50..fda00dc75 100644 --- a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditor.java +++ b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditRepository.java @@ -21,15 +21,14 @@ import com.mongodb.client.model.Filters; import com.mongodb.client.model.ReplaceOptions; import com.mongodb.client.result.UpdateResult; +import com.mongodb.reactivestreams.client.ClientSession; import com.mongodb.reactivestreams.client.MongoCollection; import com.mongodb.reactivestreams.client.MongoDatabase; import io.flamingock.internal.common.core.audit.AuditEntry; -import io.flamingock.internal.common.core.audit.AuditReader; -import io.flamingock.internal.common.core.audit.AuditWriter; import io.flamingock.internal.common.mongodb.CollectionInitializator; import io.flamingock.internal.common.mongodb.MongoDBAuditMapper; -import io.flamingock.internal.common.mongodb.MongoDBReactiveCollectionHelper; import io.flamingock.internal.common.mongodb.MongoDBDocumentHelper; +import io.flamingock.internal.common.mongodb.MongoDBReactiveCollectionHelper; import io.flamingock.internal.util.Result; import io.flamingock.internal.util.log.FlamingockLoggerFactory; import io.flamingock.reactive.util.PublisherSync; @@ -44,58 +43,93 @@ import static io.flamingock.internal.util.constants.AuditEntryFieldConstants.KEY_EXECUTION_ID; import static io.flamingock.internal.util.constants.AuditEntryFieldConstants.KEY_STATE; -public class MongoDBReactiveAuditor implements AuditWriter, AuditReader { +/** + * Native MongoDB Reactive Streams implementation of the audit repository. + * + *

The repository exposes two write shapes because the journal feature has two deliberately different + * persistence models. The legacy append path keeps one document per {@code (executionId, changeId, state)}; + * the journal path keeps one current-state document per {@code changeId} and joins the caller's transaction. + */ +public class MongoDBReactiveAuditRepository { - private static final Logger logger = FlamingockLoggerFactory.getLogger("MongoDBReactiveAuditor"); + private static final Logger logger = FlamingockLoggerFactory.getLogger("MongoDBReactiveAuditRepository"); private final MongoCollection collection; - private final MongoDBAuditMapper mapper = new MongoDBAuditMapper<>(() -> new MongoDBDocumentHelper(new Document())); + private final CollectionInitializator initializer; + private final MongoDBAuditMapper mapper = + new MongoDBAuditMapper<>(() -> new MongoDBDocumentHelper(new Document())); + private boolean initialized; - MongoDBReactiveAuditor(MongoDatabase database, - String collectionName, - ReadConcern readConcern, - ReadPreference readPreference, - WriteConcern writeConcern) { + public MongoDBReactiveAuditRepository(MongoDatabase database, + String collectionName, + ReadConcern readConcern, + ReadPreference readPreference, + WriteConcern writeConcern) { this.collection = database.getCollection(collectionName) .withReadConcern(readConcern) .withReadPreference(readPreference) .withWriteConcern(writeConcern); - } - - protected void initialize(boolean autoCreate) { - CollectionInitializator initializer = new CollectionInitializator<>( + this.initializer = new CollectionInitializator<>( new MongoDBReactiveCollectionHelper(collection), () -> new MongoDBDocumentHelper(new Document()), - new String[]{KEY_EXECUTION_ID, KEY_CHANGE_ID, KEY_STATE} - ); + new String[]{KEY_EXECUTION_ID, KEY_CHANGE_ID, KEY_STATE}); + } + + public synchronized void initialize(boolean autoCreate) { + if (initialized) { + return; + } if (autoCreate) { initializer.initialize(); } else { initializer.justValidateCollection(); } + initialized = true; + } + /** + * Saves the current state of a change in a caller-owned MongoDB transaction. + * + * @param clientSession session owning the transaction + * @param auditEntry current change state + * @return successful write result; driver failures are propagated + */ + Result save(ClientSession clientSession, AuditEntry auditEntry) { + Bson filter = Filters.eq(KEY_CHANGE_ID, auditEntry.getChangeId()); + Document entryDocument = mapper.toDocument(auditEntry).getDocument(); + + UpdateResult result = PublisherSync.first( + collection.replaceOne(clientSession, filter, entryDocument, new ReplaceOptions().upsert(true))); + logger.debug("Save changeState[{}] with result" + + "\n[upsertId:{}, matches: {}, modifies: {}, acknowledged: {}]", + auditEntry, result.getUpsertedId(), result.getMatchedCount(), result.getModifiedCount(), + result.wasAcknowledged()); + return Result.OK(); } - @Override - public Result writeEntry(AuditEntry auditEntry) { + /** + * Keeps the historical one-document-per-state behavior used while journal events are disabled. + * + * @param auditEntry entry to append or replace + * @return successful write result; driver failures are propagated + */ + Result append(AuditEntry auditEntry) { Bson filter = Filters.and( Filters.eq(KEY_EXECUTION_ID, auditEntry.getExecutionId()), Filters.eq(KEY_CHANGE_ID, auditEntry.getChangeId()), Filters.eq(KEY_STATE, auditEntry.getState().name()) ); - Document entryDocument = mapper.toDocument(auditEntry).getDocument(); UpdateResult result = PublisherSync.first( collection.replaceOne(filter, entryDocument, new ReplaceOptions().upsert(true))); - logger.debug("SaveOrUpdate[{}] with result" + - "\n[upsertId:{}, matches: {}, modifies: {}, acknowledged: {}]", auditEntry, result.getUpsertedId(), result.getMatchedCount(), result.getModifiedCount(), result.wasAcknowledged()); - + logger.debug("SaveOrUpdate[{}] with result" + + "\n[upsertId:{}, matches: {}, modifies: {}, acknowledged: {}]", + auditEntry, result.getUpsertedId(), result.getMatchedCount(), result.getModifiedCount(), + result.wasAcknowledged()); return Result.OK(); } - - @Override public List getAuditHistory() { return PublisherSync.collect(collection.find()) .stream() diff --git a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveJournalEventStore.java b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveJournalEventStore.java new file mode 100644 index 000000000..e0799941d --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveJournalEventStore.java @@ -0,0 +1,170 @@ +/* + * 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.mongodb.reactive.internal; + +import com.mongodb.ReadConcern; +import com.mongodb.ReadPreference; +import com.mongodb.WriteConcern; +import com.mongodb.client.model.Filters; +import com.mongodb.client.model.Sorts; +import com.mongodb.client.model.Updates; +import com.mongodb.reactivestreams.client.ClientSession; +import com.mongodb.reactivestreams.client.MongoCollection; +import com.mongodb.reactivestreams.client.MongoDatabase; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.mongodb.CollectionInitializator; +import io.flamingock.internal.common.mongodb.IndexDefinition; +import io.flamingock.internal.common.mongodb.MongoDBDocumentHelper; +import io.flamingock.internal.common.mongodb.MongoDBJournalEventMapper; +import io.flamingock.internal.common.mongodb.MongoDBReactiveCollectionHelper; +import io.flamingock.internal.core.journal.JournalEventStore; +import io.flamingock.internal.util.Result; +import io.flamingock.internal.util.log.FlamingockLoggerFactory; +import io.flamingock.reactive.util.PublisherSync; +import org.bson.Document; +import org.slf4j.Logger; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static io.flamingock.internal.common.mongodb.journal.JournalEventFieldConstants.KEY_ACKNOWLEDGED; +import static io.flamingock.internal.common.mongodb.journal.JournalEventFieldConstants.KEY_EVENT_ID; +import static io.flamingock.internal.common.mongodb.journal.JournalEventFieldConstants.KEY_STREAM_ID; +import static io.flamingock.internal.common.mongodb.journal.JournalEventFieldConstants.KEY_STREAM_SEQUENCE; + +/** + * Native MongoDB Reactive Streams implementation of the local journal event buffer. + * + *

The collection and indexes intentionally mirror the synchronous MongoDB implementation. Appending is + * kept outside {@link JournalEventStore} because it must receive the native driver {@link ClientSession} that + * also owns the current-state audit write. + */ +public class MongoDBReactiveJournalEventStore implements JournalEventStore { + + private static final Logger logger = FlamingockLoggerFactory.getLogger("MongoDBReactiveJournal"); + + static final String UNIQUE_INDEX_NAME = "unique_key_sequence"; + static final String UNACKNOWLEDGED_INDEX_NAME = "unacknowledged_by_key_sequence"; + static final String EVENT_ID_INDEX_NAME = "unique_event_id"; + + private final MongoCollection collection; + private final MongoDBJournalEventMapper mapper = new MongoDBJournalEventMapper(); + private final CollectionInitializator initializer; + private boolean initialized; + + public MongoDBReactiveJournalEventStore(MongoDatabase database, + String collectionName, + ReadConcern readConcern, + ReadPreference readPreference, + WriteConcern writeConcern) { + this.collection = database.getCollection(collectionName) + .withReadConcern(readConcern) + .withReadPreference(readPreference) + .withWriteConcern(writeConcern); + this.initializer = new CollectionInitializator<>( + new MongoDBReactiveCollectionHelper(collection), + () -> new MongoDBDocumentHelper(new Document()), + indexDefinitions()); + } + + public synchronized void initialize(boolean autoCreate) { + if (initialized) { + return; + } + if (autoCreate) { + initializer.initialize(); + } else { + initializer.justValidateCollection(); + } + initialized = true; + } + + private static List indexDefinitions() { + LinkedHashMap streamUniqueKeys = new LinkedHashMap<>(); + streamUniqueKeys.put(KEY_STREAM_ID, 1); + streamUniqueKeys.put(KEY_STREAM_SEQUENCE, 1); + IndexDefinition streamUniqueIndex = new IndexDefinition(streamUniqueKeys, true, null, UNIQUE_INDEX_NAME); + + LinkedHashMap partialKeys = new LinkedHashMap<>(); + partialKeys.put(KEY_ACKNOWLEDGED, 1); + partialKeys.put(KEY_STREAM_ID, 1); + partialKeys.put(KEY_STREAM_SEQUENCE, 1); + Map partialFilter = new LinkedHashMap<>(); + partialFilter.put(KEY_ACKNOWLEDGED, false); + IndexDefinition unacknowledgedIndex = + new IndexDefinition(partialKeys, false, partialFilter, UNACKNOWLEDGED_INDEX_NAME); + + LinkedHashMap eventIdKeys = new LinkedHashMap<>(); + eventIdKeys.put(KEY_EVENT_ID, 1); + IndexDefinition eventIdIndex = new IndexDefinition(eventIdKeys, true, null, EVENT_ID_INDEX_NAME); + + return Arrays.asList(streamUniqueIndex, unacknowledgedIndex, eventIdIndex); + } + + /** + * Appends an immutable event in the supplied transaction session. + * + *

This is deliberately an insert, never an upsert. A duplicate stream position or event id is a + * transaction failure, so the corresponding audit-state write is rolled back as well. + * + * @param clientSession session owning the transaction + * @param event event to append + * @return successful write result; driver failures are propagated + */ + Result append(ClientSession clientSession, JournalEvent event) { + if (!initialized) { + throw new IllegalStateException("MongoDB reactive journal is not initialized"); + } + PublisherSync.first(collection.insertOne(clientSession, mapper.toDocument(event))); + logger.debug("Journal event appended [eventId={} type={} stream={} sequence={}]", + event.getEventId(), event.getEventType(), event.getStreamId(), event.getStreamSequence()); + return Result.OK(); + } + + @Override + public Optional> getLastEventByStream(String streamId) { + Document document = PublisherSync.first(collection.find(Filters.eq(KEY_STREAM_ID, streamId)) + .sort(Sorts.descending(KEY_STREAM_SEQUENCE)) + .limit(1) + .first()); + return document == null ? Optional.empty() : Optional.of(mapper.fromDocument(document)); + } + + @Override + public List> getUnacknowledgedEvents(int limit) { + List> events = new ArrayList<>(); + PublisherSync.collect(collection.find(Filters.eq(KEY_ACKNOWLEDGED, false)) + .sort(Sorts.ascending(KEY_STREAM_ID, KEY_STREAM_SEQUENCE)) + .limit(limit)) + .forEach(document -> events.add(mapper.fromDocument(document))); + return events; + } + + @Override + public long acknowledgeEvents(Collection eventIds) { + if (eventIds == null || eventIds.isEmpty()) { + return 0L; + } + return PublisherSync.first(collection.updateMany( + Filters.in(KEY_EVENT_ID, eventIds), Updates.set(KEY_ACKNOWLEDGED, true))).getModifiedCount(); + } +} diff --git a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStoreJournalTest.java b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStoreJournalTest.java new file mode 100644 index 000000000..2ed577fb6 --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStoreJournalTest.java @@ -0,0 +1,387 @@ +/* + * 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.mongodb.reactive; + +import com.mongodb.reactivestreams.client.MongoClient; +import com.mongodb.reactivestreams.client.MongoClients; +import com.mongodb.client.model.IndexOptions; +import com.mongodb.reactivestreams.client.MongoDatabase; +import io.flamingock.core.kit.audit.AuditEntryTestFactory; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.audit.AuditPersistenceFactory; +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.common.mongodb.MongoDBJournalEventMapper; +import io.flamingock.internal.core.configuration.community.CommunityConfiguration; +import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence; +import io.flamingock.internal.core.context.SimpleContext; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.id.RunnerId; +import io.flamingock.reactive.util.PublisherSync; +import io.flamingock.targetsystem.mongodb.reactive.MongoDBReactiveTargetSystem; +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.junit.jupiter.api.function.Executable; +import org.bson.Document; +import org.testcontainers.containers.MongoDBContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +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 MongoDBReactiveAuditStoreJournalTest { + + private static final String DB_NAME = "test"; + private static final String AUDIT_COLLECTION = "storeAudit"; + private static final String LOCK_COLLECTION = "storeLock"; + private static final String JOURNAL_COLLECTION = "storeJournal"; + + @Container + static final MongoDBContainer mongoDBContainer = + new MongoDBContainer(DockerImageName.parse("mongo:6")).withReuse(true); + + private final MongoDBJournalEventMapper mapper = new MongoDBJournalEventMapper(); + + private MongoClient mongoClient; + private MongoDatabase database; + private SimpleContext context; + + @BeforeEach + void setUp() { + mongoClient = MongoClients.create(mongoDBContainer.getConnectionString()); + database = mongoClient.getDatabase(DB_NAME); + context = new SimpleContext(); + context.addDependency(RunnerId.generate()); + context.addDependency(new CommunityConfiguration()); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + PublisherSync.complete(database.drop()); + mongoClient.close(); + } + + @Test + @DisplayName("the flag-off store keeps historical audit rows and does not create the journal") + void flagOffKeepsHistoricalAuditRows() { + MongoDBReactiveAuditStore auditStore = initializeStore(); + CommunityAuditPersistence persistence = auditStore.getPersistenceFactory().get("stage-one"); + + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.STARTED)); + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED)); + + assertEquals(2, persistence.getAuditHistory().size()); + assertFalse(collectionExists(JOURNAL_COLLECTION)); + } + + @Test + @DisplayName("journal-enabled store creates independent streams for each stage") + void flagOnCreatesStageAwareJournalStreams() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBReactiveAuditStore auditStore = initializeStore(); + AuditPersistenceFactory factory = auditStore.getPersistenceFactory(); + + factory.get("stage-one").writeEntry(auditEntry("change-one", AuditEntry.Status.APPLIED)); + factory.get("stage-two").writeEntry(auditEntry("change-two", AuditEntry.Status.APPLIED)); + + List> events = storedEvents(); + assertEquals(2, events.size()); + assertTrue(events.stream().anyMatch(event -> "stage-one".equals(event.getStreamId()) + && event.getStreamSequence() == 1L)); + assertTrue(events.stream().anyMatch(event -> "stage-two".equals(event.getStreamId()) + && event.getStreamSequence() == 1L)); + } + + @Test + @DisplayName("recreated stage persistence resumes its stream without affecting other stages") + void recreatedStagePersistenceReseedsOnlyItsOwnStream() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + AuditPersistenceFactory firstFactory = initializeStore().getPersistenceFactory(); + firstFactory.get("stage-one").writeEntry(auditEntry("stage-one-first", AuditEntry.Status.APPLIED)); + firstFactory.get("stage-two").writeEntry(auditEntry("stage-two-first", AuditEntry.Status.APPLIED)); + + initializeStore().getPersistenceFactory() + .get("stage-one") + .writeEntry(auditEntry("stage-one-second", AuditEntry.Status.APPLIED)); + + Map>> eventsByStream = storedEvents().stream() + .collect(Collectors.groupingBy(JournalEvent::getStreamId)); + + assertEquals(Arrays.asList(1L, 2L), sequences(eventsByStream.get("stage-one"))); + assertEquals(Arrays.asList(1L), sequences(eventsByStream.get("stage-two"))); + } + + @Test + @DisplayName("an audit-only installation remains readable until journal persistence is first accessed") + void auditOnlyUpgradeCreatesJournalSchemaLazily() { + MongoDBReactiveAuditStore legacyStore = initializeStore(); + CommunityAuditPersistence legacyPersistence = legacyStore.getPersistenceFactory().get("legacy-stage"); + legacyPersistence.writeEntry(auditEntry("legacy-started", AuditEntry.Status.STARTED)); + legacyPersistence.writeEntry(auditEntry("legacy-applied", AuditEntry.Status.APPLIED)); + + assertFalse(collectionExists(JOURNAL_COLLECTION)); + assertEquals(Arrays.asList("legacy-applied", "legacy-started"), auditChangeIds(legacyPersistence.getAuditHistory())); + + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBReactiveAuditStore upgradedStore = initializeStore(); + assertEquals(Arrays.asList("legacy-applied", "legacy-started"), auditChangeIds(upgradedStore.getAuditReader().getAuditHistory())); + assertFalse(collectionExists(JOURNAL_COLLECTION)); + + upgradedStore.getPersistenceFactory().get("upgrade-stage"); + + Map indexes = indexesByName(JOURNAL_COLLECTION); + assertTrue(collectionExists(JOURNAL_COLLECTION)); + assertTrue(indexes.get("unique_key_sequence").getBoolean("unique")); + assertFalse(indexes.get("unacknowledged_by_key_sequence").getBoolean("unique", false)); + assertEquals(new Document("acknowledged", false), indexes.get("unacknowledged_by_key_sequence") + .get("partialFilterExpression", Document.class)); + assertTrue(indexes.get("unique_event_id").getBoolean("unique")); + assertEquals(0, storedEvents().size()); + } + + @Test + @DisplayName("manual schemas validate at their lazy access boundaries") + void manualSchemaValidationIsLazyAndAcceptsValidSchema() { + createManualSchema(false); + MongoDBReactiveAuditStore validStore = initializeManualStore(); + validStore.getAuditReader().getAuditHistory(); + validStore.getPersistenceFactory().get("manual-stage"); + + PublisherSync.complete(database.drop()); + assertInvalidSchema(LOCK_COLLECTION, () -> initializeManualStore()); + + PublisherSync.complete(database.drop()); + createLockSchema(); + MongoDBReactiveAuditStore missingAuditStore = initializeManualStore(); + assertInvalidSchema(AUDIT_COLLECTION, () -> missingAuditStore.getAuditReader()); + assertInvalidSchema(AUDIT_COLLECTION, () -> missingAuditStore.getPersistenceFactory().get("manual-stage")); + + PublisherSync.complete(database.drop()); + createLockSchema(); + createAuditSchema(); + MongoDBReactiveAuditStore missingJournalStore = initializeManualStore(); + missingJournalStore.getAuditReader().getAuditHistory(); + assertInvalidSchema(JOURNAL_COLLECTION, () -> missingJournalStore.getPersistenceFactory().get("manual-stage")); + } + + @Test + @DisplayName("manual journal validation rejects each required-index contract boundary") + void manualJournalValidationRejectsInvalidRequiredIndexes() { + assertInvalidJournalIndex(() -> { }); + assertInvalidJournalIndex(() -> createIndex(JOURNAL_COLLECTION, new Document("streamId", 1).append("streamSequence", 1), + new IndexOptions().name("unique_key_sequence").unique(false))); + assertInvalidJournalIndex(() -> createIndex(JOURNAL_COLLECTION, + new Document("acknowledged", 1).append("streamId", 1).append("streamSequence", 1), + new IndexOptions().name("unacknowledged_by_key_sequence").partialFilterExpression(new Document("acknowledged", true)))); + assertInvalidJournalIndex(() -> createIndex(JOURNAL_COLLECTION, new Document("wrong", 1), + new IndexOptions().name("unique_event_id").unique(true))); + } + + @Test + @DisplayName("manual journal validation accepts descending keys and extra non-unique indexes") + void manualJournalValidationAcceptsToleratedIndexDifferences() { + createManualSchema(true); + MongoDBReactiveAuditStore descendingStore = initializeManualStore(); + descendingStore.getPersistenceFactory().get("descending-stage"); + + PublisherSync.complete(database.drop()); + createManualSchema(false); + createIndex(JOURNAL_COLLECTION, new Document("diagnostic", 1), new IndexOptions().name("diagnostic_index")); + MongoDBReactiveAuditStore extraIndexStore = initializeManualStore(); + extraIndexStore.getPersistenceFactory().get("extra-index-stage"); + } + + @Test + @DisplayName("stage persistence and audit reader coexist without duplicate journal events") + void stagePersistenceAndAuditReaderCoexist() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBReactiveAuditStore auditStore = initializeStore(); + CommunityAuditPersistence persistence = auditStore.getPersistenceFactory().get("coexist-stage"); + persistence.writeEntry(auditEntry("coexist-change", AuditEntry.Status.APPLIED)); + + assertEquals(Arrays.asList("coexist-change"), auditChangeIds(auditStore.getAuditReader().getAuditHistory())); + assertEquals(1, storedEvents().size()); + assertEquals("coexist-stage", storedEvents().get(0).getStreamId()); + assertEquals(1L, storedEvents().get(0).getStreamSequence()); + } + + @Test + @DisplayName("journal repository name is required and must be different from audit and lock names") + void journalRepositoryNameMustBeUnique() { + MongoDBReactiveTargetSystem targetSystem = initializedTargetSystem(); + MongoDBReactiveAuditStore sameAsAudit = MongoDBReactiveAuditStore.from(targetSystem) + .withAuditRepositoryName(AUDIT_COLLECTION) + .withLockRepositoryName(LOCK_COLLECTION) + .withJournalRepositoryName(AUDIT_COLLECTION); + + assertThrows(FlamingockException.class, () -> sameAsAudit.initialize(context)); + + MongoDBReactiveAuditStore blank = MongoDBReactiveAuditStore.from(targetSystem) + .withAuditRepositoryName(AUDIT_COLLECTION) + .withLockRepositoryName(LOCK_COLLECTION) + .withJournalRepositoryName(" "); + MongoDBReactiveAuditStore nullName = MongoDBReactiveAuditStore.from(targetSystem) + .withAuditRepositoryName(AUDIT_COLLECTION) + .withLockRepositoryName(LOCK_COLLECTION) + .withJournalRepositoryName(null); + MongoDBReactiveAuditStore normalizedAuditCollision = MongoDBReactiveAuditStore.from(targetSystem) + .withAuditRepositoryName(AUDIT_COLLECTION) + .withLockRepositoryName(LOCK_COLLECTION) + .withJournalRepositoryName(" STOREAUDIT "); + MongoDBReactiveAuditStore normalizedLockCollision = MongoDBReactiveAuditStore.from(targetSystem) + .withAuditRepositoryName(AUDIT_COLLECTION) + .withLockRepositoryName(LOCK_COLLECTION) + .withJournalRepositoryName(" STORELOCK "); + + assertThrows(FlamingockException.class, () -> blank.initialize(context)); + assertThrows(FlamingockException.class, () -> nullName.initialize(context)); + assertThrows(FlamingockException.class, () -> normalizedAuditCollision.initialize(context)); + assertThrows(FlamingockException.class, () -> normalizedLockCollision.initialize(context)); + } + + private MongoDBReactiveAuditStore initializeManualStore() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBReactiveAuditStore auditStore = MongoDBReactiveAuditStore.from(initializedTargetSystem()) + .withAuditRepositoryName(AUDIT_COLLECTION) + .withLockRepositoryName(LOCK_COLLECTION) + .withJournalRepositoryName(JOURNAL_COLLECTION) + .withAutoCreate(false); + auditStore.initialize(context); + return auditStore; + } + + private void assertInvalidJournalIndex(Executable mutation) { + PublisherSync.complete(database.drop()); + createLockSchema(); + createAuditSchema(); + createCollection(JOURNAL_COLLECTION); + try { + mutation.execute(); + } catch (Throwable throwable) { + throw new RuntimeException(throwable); + } + assertInvalidSchema(JOURNAL_COLLECTION, () -> initializeManualStore().getPersistenceFactory().get("invalid-stage")); + } + + private void assertInvalidSchema(String collectionName, Executable access) { + RuntimeException exception = assertThrows(RuntimeException.class, access); + assertTrue(exception.getMessage().contains(collectionName)); + } + + private void createManualSchema(boolean descending) { + createLockSchema(); + createAuditSchema(); + createJournalSchema(descending); + } + + private void createLockSchema() { + createCollection(LOCK_COLLECTION); + createIndex(LOCK_COLLECTION, new Document("key", 1), new IndexOptions().unique(true)); + } + + private void createAuditSchema() { + createCollection(AUDIT_COLLECTION); + createIndex(AUDIT_COLLECTION, new Document("executionId", 1).append("changeId", 1).append("state", 1), + new IndexOptions().unique(true)); + } + + private void createJournalSchema(boolean descending) { + createCollection(JOURNAL_COLLECTION); + int direction = descending ? -1 : 1; + createIndex(JOURNAL_COLLECTION, new Document("streamId", direction).append("streamSequence", direction), + new IndexOptions().name("unique_key_sequence").unique(true)); + createIndex(JOURNAL_COLLECTION, new Document("acknowledged", direction).append("streamId", direction) + .append("streamSequence", direction), + new IndexOptions().name("unacknowledged_by_key_sequence") + .partialFilterExpression(new Document("acknowledged", false))); + createIndex(JOURNAL_COLLECTION, new Document("eventId", direction), + new IndexOptions().name("unique_event_id").unique(true)); + } + + private void createCollection(String collectionName) { + PublisherSync.complete(database.createCollection(collectionName)); + } + + private void createIndex(String collectionName, Document keys, IndexOptions options) { + PublisherSync.first(database.getCollection(collectionName).createIndex(keys, options)); + } + + private Map indexesByName(String collectionName) { + return PublisherSync.collect(database.getCollection(collectionName).listIndexes()).stream() + .collect(Collectors.toMap(index -> index.getString("name"), index -> index)); + } + + private static List auditChangeIds(List entries) { + return entries.stream().map(AuditEntry::getChangeId).sorted().collect(Collectors.toList()); + } + + private MongoDBReactiveAuditStore initializeStore() { + MongoDBReactiveTargetSystem targetSystem = initializedTargetSystem(); + MongoDBReactiveAuditStore auditStore = MongoDBReactiveAuditStore.from(targetSystem) + .withAuditRepositoryName(AUDIT_COLLECTION) + .withLockRepositoryName(LOCK_COLLECTION) + .withJournalRepositoryName(JOURNAL_COLLECTION); + auditStore.initialize(context); + return auditStore; + } + + private MongoDBReactiveTargetSystem initializedTargetSystem() { + MongoDBReactiveTargetSystem targetSystem = new MongoDBReactiveTargetSystem("mongodb", mongoClient, DB_NAME); + targetSystem.initialize(context); + return targetSystem; + } + + private List> storedEvents() { + if (!collectionExists(JOURNAL_COLLECTION)) { + return new ArrayList<>(); + } + return PublisherSync.collect(database.getCollection(JOURNAL_COLLECTION).find()) + .stream() + .map(mapper::fromDocument) + .collect(Collectors.toList()); + } + + private boolean collectionExists(String collectionName) { + return PublisherSync.collect(database.listCollectionNames()).contains(collectionName); + } + + private static List sequences(List> events) { + return events.stream() + .map(JournalEvent::getStreamSequence) + .sorted() + .collect(Collectors.toList()); + } + + private static AuditEntry auditEntry(String changeId, AuditEntry.Status status) { + return AuditEntryTestFactory.createTestAuditEntry(changeId, status, AuditTxType.NON_TX, (Class) null); + } +} diff --git a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveJournalFeatureFlagE2ETest.java b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveJournalFeatureFlagE2ETest.java new file mode 100644 index 000000000..31383c599 --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveJournalFeatureFlagE2ETest.java @@ -0,0 +1,155 @@ +/* + * 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.mongodb.reactive; + +import com.mongodb.reactivestreams.client.ClientSession; +import com.mongodb.reactivestreams.client.MongoClient; +import com.mongodb.reactivestreams.client.MongoClients; +import com.mongodb.reactivestreams.client.MongoDatabase; +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.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.common.mongodb.MongoDBJournalEventMapper; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.mongodb.reactive.kit.MongoDBReactiveTestKit; +import io.flamingock.store.mongodb.reactive.changes._001__create_client_collection_happy; +import io.flamingock.store.mongodb.reactive.changes._002__insert_federico_happy_transactional; +import io.flamingock.targetsystem.mongodb.reactive.MongoDBReactiveTargetSystem; +import io.flamingock.reactive.util.PublisherSync; +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.MongoDBContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.util.ArrayList; +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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * End-to-end coverage for the journal feature gate through a complete native reactive runner execution. + */ +@Testcontainers +class MongoDBReactiveJournalFeatureFlagE2ETest { + + private static final String DB_NAME = "test"; + private static final String JOURNAL_COLLECTION = "flamingockJournalEvents"; + private static final String DEFAULT_STAGE_NAME = "default-stage-name"; + + @Container + static final MongoDBContainer mongoDBContainer = + new MongoDBContainer(DockerImageName.parse("mongo:6")).withReuse(true); + + private final MongoDBJournalEventMapper mapper = new MongoDBJournalEventMapper(); + + private MongoClient mongoClient; + private MongoDatabase database; + private TestKit testKit; + private MongoDBTestHelper mongoDBTestHelper; + + @BeforeEach + void setUp() { + mongoClient = MongoClients.create(mongoDBContainer.getConnectionString()); + database = mongoClient.getDatabase(DB_NAME); + MongoDBReactiveTargetSystem targetSystem = new MongoDBReactiveTargetSystem("mongodb", mongoClient, DB_NAME); + testKit = MongoDBReactiveTestKit.create( + MongoDBReactiveAuditStore.from(targetSystem), mongoClient, database); + mongoDBTestHelper = new MongoDBTestHelper(database); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + testKit.cleanUp(); + mongoClient.close(); + } + + @Test + @DisplayName("journal disabled keeps the audit history and creates no journal collection") + void journalDisabledLeavesNoJournalCollection() { + runPipeline( + STARTED("create-client-collection"), + APPLIED("create-client-collection"), + STARTED("insert-federico-document"), + APPLIED("insert-federico-document")); + + assertFalse(mongoDBTestHelper.collectionExists(JOURNAL_COLLECTION)); + } + + @Test + @DisplayName("journal enabled keeps current audit state and the complete event history") + void journalEnabledSplitsCurrentStateFromHistory() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + + runPipeline( + APPLIED("create-client-collection"), + APPLIED("insert-federico-document")); + + assertTrue(mongoDBTestHelper.collectionExists(JOURNAL_COLLECTION)); + List> events = storedEvents(); + assertEquals(4, events.size()); + assertTrue(events.stream().allMatch(event -> DEFAULT_STAGE_NAME.equals(event.getStreamId()))); + assertEquals(Arrays.asList(1L, 2L, 3L, 4L), events.stream() + .map(JournalEvent::getStreamSequence) + .sorted() + .collect(Collectors.toList())); + assertEquals(Arrays.asList(AuditEntry.Status.STARTED, AuditEntry.Status.STARTED, + AuditEntry.Status.APPLIED, AuditEntry.Status.APPLIED), + events.stream().map(event -> event.getData().getState()).sorted().collect(Collectors.toList())); + } + + private void runPipeline(AuditEntryExpectation... expectedAudits) { + MongoDBReactiveTargetSystem targetSystem = new MongoDBReactiveTargetSystem("mongodb", mongoClient, DB_NAME); + AuditTestSupport.withTestKit(testKit) + .GIVEN_Changes( + new CodeChangeTestDefinition(_001__create_client_collection_happy.class, + Collections.singletonList(MongoDatabase.class)), + new CodeChangeTestDefinition(_002__insert_federico_happy_transactional.class, + Arrays.asList(MongoDatabase.class, ClientSession.class))) + .WHEN(() -> testKit.createBuilder() + .setAuditStore(MongoDBReactiveAuditStore.from(targetSystem)) + .addTargetSystem(targetSystem) + .build() + .run()) + .THEN_VerifyAuditSequenceStrict(expectedAudits) + .run(); + } + + private List> storedEvents() { + if (!mongoDBTestHelper.collectionExists(JOURNAL_COLLECTION)) { + return new ArrayList<>(); + } + return PublisherSync.collect(database.getCollection(JOURNAL_COLLECTION).find()) + .stream() + .map(mapper::fromDocument) + .collect(Collectors.toList()); + } +} diff --git a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistenceJournalTest.java b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistenceJournalTest.java new file mode 100644 index 000000000..7a86ac358 --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistenceJournalTest.java @@ -0,0 +1,242 @@ +/* + * 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.mongodb.reactive.internal; + +import com.mongodb.ReadConcern; +import com.mongodb.ReadPreference; +import com.mongodb.WriteConcern; +import com.mongodb.reactivestreams.client.ClientSession; +import com.mongodb.reactivestreams.client.MongoClient; +import com.mongodb.reactivestreams.client.MongoClients; +import com.mongodb.reactivestreams.client.MongoDatabase; +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.id.RunnerId; +import io.flamingock.reactive.util.PublisherSync; +import io.flamingock.targetsystem.mongodb.reactive.MongoDBReactiveTxWrapper; +import org.bson.Document; +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.mockito.ArgumentMatchers; +import org.testcontainers.containers.MongoDBContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +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; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; + +/** + * Verifies the feature gate and the atomic native {@code ClientSession} boundary between audit state and + * journal events. + */ +@Testcontainers +class MongoDBReactiveAuditPersistenceJournalTest { + + private static final String DB_NAME = "test"; + private static final String AUDIT_COLLECTION = "flamingockAuditLog"; + private static final String JOURNAL_COLLECTION = "flamingockJournalEvents"; + private static final String STREAM_ID = "stage-under-test"; + + @Container + static final MongoDBContainer mongoDBContainer = + new MongoDBContainer(DockerImageName.parse("mongo:6")).withReuse(true); + + private final io.flamingock.internal.common.mongodb.MongoDBJournalEventMapper mapper = + new io.flamingock.internal.common.mongodb.MongoDBJournalEventMapper(); + + private MongoClient mongoClient; + private MongoDatabase database; + private MongoDBReactiveAuditRepository auditRepository; + private MongoDBReactiveJournalEventStore journalEventStore; + private MongoDBReactiveTxWrapper txWrapper; + + @BeforeEach + void setUp() { + mongoClient = MongoClients.create(mongoDBContainer.getConnectionString()); + database = mongoClient.getDatabase(DB_NAME); + auditRepository = new MongoDBReactiveAuditRepository( + database, AUDIT_COLLECTION, + ReadConcern.MAJORITY, ReadPreference.primary(), WriteConcern.MAJORITY.withJournal(true)); + journalEventStore = new MongoDBReactiveJournalEventStore( + database, JOURNAL_COLLECTION, + ReadConcern.MAJORITY, ReadPreference.primary(), WriteConcern.MAJORITY.withJournal(true)); + txWrapper = new MongoDBReactiveTxWrapper( + new TransactionManager<>(() -> PublisherSync.first(mongoClient.startSession()))); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + PublisherSync.complete(database.drop()); + mongoClient.close(); + } + + @Test + @DisplayName("journal disabled preserves append records and creates no journal collection") + void journalDisabledPreservesHistoricalAuditPath() { + MongoDBReactiveAuditPersistence persistence = persistenceFor(auditRepository); + + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.STARTED)); + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED)); + + assertEquals(2, auditRepository.getAuditHistory().size()); + assertFalse(collectionExists(JOURNAL_COLLECTION)); + } + + @Test + @DisplayName("journal enabled writes a compatible event with the current audit record") + void journalEnabledWritesEventWithAuditEntry() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBReactiveAuditPersistence persistence = persistenceFor(auditRepository); + AuditEntry entry = auditEntry("change-1", AuditEntry.Status.APPLIED); + + persistence.writeEntry(entry); + + assertEquals(1, auditRepository.getAuditHistory().size()); + List> events = storedEvents(); + assertEquals(1, events.size()); + assertEquals(STREAM_ID, events.get(0).getStreamId()); + assertEquals(1L, events.get(0).getStreamSequence()); + assertEquals(JournalEventType.CHANGE_STATE, events.get(0).getEventType()); + assertFalse(events.get(0).isAcknowledged()); + assertEquals(entry.getChangeId(), events.get(0).getData().getChangeId()); + } + + @Test + @DisplayName("journal enabled keeps only the final audit state and retains every event") + void journalEnabledSeparatesCurrentStateFromHistory() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBReactiveAuditPersistence persistence = persistenceFor(auditRepository); + + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.STARTED)); + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED)); + + List auditRecords = auditRepository.getAuditHistory(); + assertEquals(1, auditRecords.size()); + assertEquals(AuditEntry.Status.APPLIED, auditRecords.get(0).getState()); + assertEquals(2, storedEvents().size()); + } + + @Test + @DisplayName("a journal append failure rolls back the audit record") + void journalFailureRollsBackAuditEntry() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBReactiveAuditPersistence persistence = persistenceFor(auditRepository); + occupyStreamPosition(1L); + + assertThrows(DatabaseTransactionException.class, + () -> persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED))); + + assertTrue(auditRepository.getAuditHistory().isEmpty()); + assertEquals(1, storedEvents().size()); + } + + @Test + @DisplayName("an audit write failure rolls back the journal event") + void auditFailureRollsBackJournalEvent() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBReactiveAuditRepository failingRepository = mock(MongoDBReactiveAuditRepository.class); + doThrow(new IllegalStateException("audit write failed")) + .when(failingRepository).save(ArgumentMatchers.any(ClientSession.class), ArgumentMatchers.any(AuditEntry.class)); + MongoDBReactiveAuditPersistence persistence = persistenceFor(failingRepository); + + assertThrows(DatabaseTransactionException.class, + () -> persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED))); + + assertTrue(storedEvents().isEmpty()); + } + + @Test + @DisplayName("a failed write does not consume its stream position") + void failedWriteLeavesNoGap() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + JournalEventSequencer sequencer = new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID); + MongoDBReactiveAuditRepository failingRepository = mock(MongoDBReactiveAuditRepository.class); + doThrow(new IllegalStateException("audit write failed")) + .when(failingRepository).save(ArgumentMatchers.any(ClientSession.class), ArgumentMatchers.any(AuditEntry.class)); + + assertThrows(DatabaseTransactionException.class, + () -> persistenceFor(failingRepository, sequencer) + .writeEntry(auditEntry("failed-change", AuditEntry.Status.APPLIED))); + + persistenceFor(auditRepository, sequencer) + .writeEntry(auditEntry("successful-change", AuditEntry.Status.APPLIED)); + + List> events = storedEvents(); + assertEquals(1, events.size()); + assertEquals(1L, events.get(0).getStreamSequence()); + assertEquals("successful-change", events.get(0).getData().getChangeId()); + } + + private MongoDBReactiveAuditPersistence persistenceFor(MongoDBReactiveAuditRepository repository) { + return persistenceFor(repository, new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID)); + } + + private MongoDBReactiveAuditPersistence persistenceFor(MongoDBReactiveAuditRepository repository, + JournalEventSequencer sequencer) { + MongoDBReactiveAuditPersistence persistence = new MongoDBReactiveAuditPersistence( + new CommunityConfiguration(), repository, journalEventStore, sequencer, txWrapper, true); + persistence.initialize(RunnerId.generate()); + return persistence; + } + + private void occupyStreamPosition(long streamSequence) { + JournalEvent event = new JournalEvent<>( + "pre-existing-event", JournalEventType.CHANGE_STATE, JournalEvent.DEFAULT_VERSION, + STREAM_ID, streamSequence, Instant.now(), auditEntry("pre-existing-change", AuditEntry.Status.APPLIED), false); + PublisherSync.first(database.getCollection(JOURNAL_COLLECTION).insertOne(mapper.toDocument(event))); + } + + private List> storedEvents() { + if (!collectionExists(JOURNAL_COLLECTION)) { + return new ArrayList<>(); + } + return PublisherSync.collect(database.getCollection(JOURNAL_COLLECTION).find()) + .stream() + .map(mapper::fromDocument) + .collect(Collectors.toList()); + } + + private boolean collectionExists(String collectionName) { + return PublisherSync.collect(database.listCollectionNames()).contains(collectionName); + } + + private static AuditEntry auditEntry(String changeId, AuditEntry.Status status) { + return AuditEntryTestFactory.createTestAuditEntry(changeId, status, AuditTxType.NON_TX, (Class) null); + } +} diff --git a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveJournalEventStoreE2ETest.java b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveJournalEventStoreE2ETest.java new file mode 100644 index 000000000..dcdd98c7b --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveJournalEventStoreE2ETest.java @@ -0,0 +1,369 @@ +/* + * 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.mongodb.reactive.internal; + +import com.mongodb.ReadConcern; +import com.mongodb.ReadPreference; +import com.mongodb.WriteConcern; +import com.mongodb.reactivestreams.client.ClientSession; +import com.mongodb.reactivestreams.client.MongoClient; +import com.mongodb.reactivestreams.client.MongoClients; +import com.mongodb.reactivestreams.client.MongoCollection; +import com.mongodb.reactivestreams.client.MongoDatabase; +import io.flamingock.api.RecoveryStrategy; +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.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.core.journal.JournalEventType; +import io.flamingock.internal.common.mongodb.MongoDBJournalEventMapper; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.Result; +import io.flamingock.reactive.util.PublisherSync; +import org.bson.Document; +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.MongoDBContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.temporal.ChronoUnit; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +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.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@Testcontainers +class MongoDBReactiveJournalEventStoreE2ETest { + + private static final String DB_NAME = "test"; + private static final String JOURNAL_COLLECTION = "flamingockJournalEvents"; + + @Container + static final MongoDBContainer mongoDBContainer = + new MongoDBContainer(DockerImageName.parse("mongo:6")).withReuse(true); + + private final MongoDBJournalEventMapper mapper = new MongoDBJournalEventMapper(); + + private MongoClient mongoClient; + private MongoDatabase database; + private MongoDBReactiveJournalEventStore journalEventStore; + + @BeforeEach + void setUp() { + mongoClient = MongoClients.create(mongoDBContainer.getConnectionString()); + database = mongoClient.getDatabase(DB_NAME); + journalEventStore = new MongoDBReactiveJournalEventStore( + database, JOURNAL_COLLECTION, + ReadConcern.MAJORITY, ReadPreference.primary(), WriteConcern.MAJORITY.withJournal(true)); + journalEventStore.initialize(true); + } + + @AfterEach + void tearDown() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + PublisherSync.complete(database.drop()); + mongoClient.close(); + } + + @Test + @DisplayName("initialize creates the unique, partial-unacknowledged and unique-eventId indexes") + void createsExpectedIndexes() { + Map byName = PublisherSync.collect(database.getCollection(JOURNAL_COLLECTION).listIndexes()) + .stream() + .filter(index -> index.getString("name") != null) + .collect(Collectors.toMap(index -> index.getString("name"), index -> index)); + + Document unique = byName.get(MongoDBReactiveJournalEventStore.UNIQUE_INDEX_NAME); + assertNotNull(unique); + assertTrue(unique.getBoolean("unique", false)); + assertEquals(new Document("streamId", 1).append("streamSequence", 1), unique.get("key")); + + Document unacknowledged = byName.get(MongoDBReactiveJournalEventStore.UNACKNOWLEDGED_INDEX_NAME); + assertNotNull(unacknowledged); + assertFalse(unacknowledged.getBoolean("unique", false)); + assertEquals(new Document("acknowledged", 1).append("streamId", 1) + .append("streamSequence", 1), unacknowledged.get("key")); + assertEquals(new Document("acknowledged", false), unacknowledged.get("partialFilterExpression")); + + Document eventId = byName.get(MongoDBReactiveJournalEventStore.EVENT_ID_INDEX_NAME); + assertNotNull(eventId); + assertTrue(eventId.getBoolean("unique", false)); + assertEquals(new Document("eventId", 1), eventId.get("key")); + } + + @Test + @DisplayName("initialize is idempotent") + void initializeIsIdempotent() { + Map before = listIndexesByName(); + + journalEventStore.initialize(true); + MongoDBReactiveJournalEventStore secondInstance = new MongoDBReactiveJournalEventStore( + database, JOURNAL_COLLECTION, + ReadConcern.MAJORITY, ReadPreference.primary(), WriteConcern.MAJORITY.withJournal(true)); + secondInstance.initialize(true); + + assertEquals(before.keySet(), listIndexesByName().keySet()); + } + + @Test + @DisplayName("append joins the caller transaction and abort discards the event") + void appendJoinsCallerTransaction() { + ClientSession session = PublisherSync.first(mongoClient.startSession()); + try { + session.startTransaction(); + journalEventStore.append(session, event("evt-A1", "stageA", 1L, false)); + abortTransaction(session); + } finally { + session.close(); + } + + assertFalse(journalEventStore.getLastEventByStream("stageA").isPresent()); + } + + @Test + @DisplayName("append confirms after commit and reads the committed event without conflating acknowledgement") + void appendConfirmsAndReadsCommittedEvent() { + JournalEvent event = fixedEvent("committed-event", "committed-stream", 1L, false); + + Result result = appendInCommittedTransaction(event); + + assertEquals(Result.OK(), result); + assertFalse(event.isAcknowledged()); + JournalEvent stored = journalEventStore.getLastEventByStream("committed-stream").orElseThrow(AssertionError::new); + assertEquals("committed-event", stored.getEventId()); + assertEquals(1L, stored.getStreamSequence()); + assertFalse(stored.isAcknowledged()); + } + + @Test + @DisplayName("append round-trips every journal envelope and audit-entry field") + void appendRoundTripsCompleteEventAndAuditEntry() { + JournalEvent event = fixedEvent("mapping-event", "mapping-stream", 7L, true); + + assertEquals(Result.OK(), appendInCommittedTransaction(event)); + + JournalEvent stored = journalEventStore.getLastEventByStream("mapping-stream").orElseThrow(AssertionError::new); + assertEquals(event.getEventId(), stored.getEventId()); + assertEquals(event.getEventType(), stored.getEventType()); + assertEquals(event.getEventVersion(), stored.getEventVersion()); + assertEquals(event.getStreamId(), stored.getStreamId()); + assertEquals(event.getStreamSequence(), stored.getStreamSequence()); + assertEquals(toMillis(event.getOccurredAt()), toMillis(stored.getOccurredAt())); + assertTrue(stored.isAcknowledged()); + + assertAuditEntryEquals(event.getData(), stored.getData()); + } + + @Test + @DisplayName("committed events order and resolve latest independently per stream") + void committedEventsOrderAndResolveLatestPerStream() { + appendInCommittedTransaction(fixedEvent("A1", "stream-A", 1L, false)); + appendInCommittedTransaction(fixedEvent("A2", "stream-A", 2L, false)); + appendInCommittedTransaction(fixedEvent("A3", "stream-A", 3L, false)); + appendInCommittedTransaction(fixedEvent("B1", "stream-B", 1L, false)); + + Map> sequencesByStream = journalEventStore.getUnacknowledgedEvents(10).stream() + .collect(Collectors.groupingBy(JournalEvent::getStreamId, + Collectors.mapping(JournalEvent::getStreamSequence, Collectors.toList()))); + + assertEquals(Arrays.asList(1L, 2L, 3L), sequencesByStream.get("stream-A")); + assertEquals(Collections.singletonList(1L), sequencesByStream.get("stream-B")); + assertEquals("A3", journalEventStore.getLastEventByStream("stream-A") + .orElseThrow(AssertionError::new).getEventId()); + assertEquals("B1", journalEventStore.getLastEventByStream("stream-B") + .orElseThrow(AssertionError::new).getEventId()); + assertFalse(journalEventStore.getLastEventByStream("missing-stream").isPresent()); + } + + @Test + @DisplayName("append inserts immutable events and rejects duplicate stream positions") + void appendRejectsDuplicateStreamPosition() { + writeInCommittedTransaction(event("evt-A1", "stageA", 1L, false)); + + assertThrows(RuntimeException.class, + () -> writeInCommittedTransaction(event("evt-other", "stageA", 1L, false))); + + Optional> stored = journalEventStore.getLastEventByStream("stageA"); + assertTrue(stored.isPresent()); + assertEquals("evt-A1", stored.get().getEventId()); + } + + @Test + @DisplayName("append rejects a duplicate event id even on another stream") + void appendRejectsDuplicateEventId() { + writeInCommittedTransaction(event("evt-A1", "stageA", 1L, false)); + + assertThrows(RuntimeException.class, + () -> writeInCommittedTransaction(event("evt-A1", "stageB", 1L, false))); + + assertTrue(journalEventStore.getLastEventByStream("stageA").isPresent()); + assertFalse(journalEventStore.getLastEventByStream("stageB").isPresent()); + } + + @Test + @DisplayName("reads and acknowledgement preserve native driver ordering") + void readsAndAcknowledgementPreserveOrder() { + seed(Arrays.asList( + event("evt-A1", "stageA", 1L, true), + event("evt-A2", "stageA", 2L, false), + event("evt-A3", "stageA", 3L, false), + event("evt-B1", "stageB", 1L, false))); + + assertEquals("evt-A3", journalEventStore.getLastEventByStream("stageA").get().getEventId()); + assertEquals(Arrays.asList("evt-A2", "evt-A3", "evt-B1"), ids( + journalEventStore.getUnacknowledgedEvents(10))); + assertEquals(Arrays.asList("evt-A2", "evt-A3"), ids( + journalEventStore.getUnacknowledgedEvents(2))); + + assertEquals(2L, journalEventStore.acknowledgeEvents(Arrays.asList("evt-A2", "evt-B1"))); + assertEquals(Collections.singletonList("evt-A3"), ids( + journalEventStore.getUnacknowledgedEvents(10))); + assertEquals(0L, journalEventStore.acknowledgeEvents(Collections.emptyList())); + } + + private Map listIndexesByName() { + return PublisherSync.collect(database.getCollection(JOURNAL_COLLECTION).listIndexes()) + .stream() + .filter(index -> index.getString("name") != null) + .collect(Collectors.toMap(index -> index.getString("name"), index -> index)); + } + + private void writeInCommittedTransaction(JournalEvent event) { + appendInCommittedTransaction(event); + } + + private Result appendInCommittedTransaction(JournalEvent event) { + ClientSession session = PublisherSync.first(mongoClient.startSession()); + try { + session.startTransaction(); + Result result = journalEventStore.append(session, event); + PublisherSync.complete(session.commitTransaction()); + return result; + } catch (RuntimeException exception) { + try { + abortTransaction(session); + } catch (RuntimeException ignored) { + // The original duplicate-write failure is the assertion target. + } + throw exception; + } finally { + session.close(); + } + } + + private static void abortTransaction(ClientSession session) { + PublisherSync.complete(session.abortTransaction()); + } + + private void seed(List> events) { + MongoCollection collection = database.getCollection(JOURNAL_COLLECTION); + PublisherSync.collect(collection.insertMany(events.stream() + .map(mapper::toDocument) + .collect(Collectors.toList()))); + } + + private static List ids(List> events) { + return events.stream().map(JournalEvent::getEventId).collect(Collectors.toList()); + } + + private static void assertAuditEntryEquals(AuditEntry expected, AuditEntry actual) { + assertEquals(expected.getExecutionId(), actual.getExecutionId()); + assertEquals(expected.getStageId(), actual.getStageId()); + assertEquals(expected.getChangeId(), actual.getChangeId()); + assertEquals(expected.getAuthor(), actual.getAuthor()); + assertEquals(toMillis(expected.getCreatedAt()), toMillis(actual.getCreatedAt())); + assertEquals(expected.getState(), actual.getState()); + assertEquals(expected.getType(), actual.getType()); + assertEquals(expected.getClassName(), actual.getClassName()); + assertEquals(expected.getMethodName(), actual.getMethodName()); + assertEquals(expected.getSourceFile(), actual.getSourceFile()); + assertEquals(expected.getExecutionMillis(), actual.getExecutionMillis()); + assertEquals(expected.getExecutionHostname(), actual.getExecutionHostname()); + assertEquals(expected.getMetadata(), actual.getMetadata()); + assertEquals(expected.getSystemChange(), actual.getSystemChange()); + assertEquals(expected.getErrorTrace(), actual.getErrorTrace()); + assertEquals(expected.getTxType(), actual.getTxType()); + assertEquals(expected.getTargetSystemId(), actual.getTargetSystemId()); + assertEquals(expected.getOrder(), actual.getOrder()); + assertEquals(expected.getRecoveryStrategy(), actual.getRecoveryStrategy()); + assertEquals(expected.getTransactionFlag(), actual.getTransactionFlag()); + } + + private static Instant toMillis(Instant instant) { + return instant.truncatedTo(ChronoUnit.MILLIS); + } + + private static LocalDateTime toMillis(LocalDateTime dateTime) { + return dateTime.truncatedTo(ChronoUnit.MILLIS); + } + + private static JournalEvent fixedEvent(String eventId, + String streamId, + long sequence, + boolean acknowledged) { + return new JournalEvent<>( + eventId, + JournalEventType.CHANGE_STATE, + 3, + streamId, + sequence, + Instant.parse("2025-02-03T04:05:06.789123456Z"), + new AuditEntry( + "execution-fixed", "stage-fixed", "change-fixed", "author-fixed", + LocalDateTime.parse("2025-02-03T04:05:06.789123456"), + AuditEntry.Status.FAILED, AuditEntry.ChangeType.STANDARD_TEMPLATE, + "example.Change", "apply", "Change.java", 9876L, "host-fixed", + new Document("metadata", "value").append("count", 2), true, "error-fixed", + AuditTxType.TX_SEPARATE_WITH_MARKER, "target-fixed", "order-fixed", + RecoveryStrategy.ALWAYS_RETRY, Boolean.TRUE), + acknowledged); + } + + private static JournalEvent event(String eventId, + String streamId, + long sequence, + boolean acknowledged) { + return new JournalEvent<>( + eventId, + JournalEventType.CHANGE_STATE, + JournalEvent.DEFAULT_VERSION, + streamId, + sequence, + Instant.now(), + auditEntry(eventId), + acknowledged); + } + + private static AuditEntry auditEntry(String changeId) { + return AuditEntryTestFactory.createTestAuditEntry( + changeId, AuditEntry.Status.APPLIED, AuditTxType.NON_TX, (Class) null); + } +} diff --git a/core/target-systems/flamingock-mongodb-reactive-externalsystem-api/build.gradle.kts b/core/target-systems/flamingock-mongodb-reactive-externalsystem-api/build.gradle.kts index 63012fb52..012718c96 100644 --- a/core/target-systems/flamingock-mongodb-reactive-externalsystem-api/build.gradle.kts +++ b/core/target-systems/flamingock-mongodb-reactive-externalsystem-api/build.gradle.kts @@ -1,7 +1,7 @@ val coreApiVersion: String by extra dependencies { - implementation("io.flamingock:flamingock-core-api:${coreApiVersion}") + api(project(":core:flamingock-core-commons")) compileOnly("org.mongodb:mongodb-driver-reactivestreams:4.0.0") } diff --git a/core/target-systems/flamingock-mongodb-reactive-externalsystem-api/src/main/java/io/flamingock/externalsystem/mongodb/reactive/api/MongoDBReactiveExternalSystem.java b/core/target-systems/flamingock-mongodb-reactive-externalsystem-api/src/main/java/io/flamingock/externalsystem/mongodb/reactive/api/MongoDBReactiveExternalSystem.java index c2ba24770..459fb3b70 100644 --- a/core/target-systems/flamingock-mongodb-reactive-externalsystem-api/src/main/java/io/flamingock/externalsystem/mongodb/reactive/api/MongoDBReactiveExternalSystem.java +++ b/core/target-systems/flamingock-mongodb-reactive-externalsystem-api/src/main/java/io/flamingock/externalsystem/mongodb/reactive/api/MongoDBReactiveExternalSystem.java @@ -16,9 +16,9 @@ package io.flamingock.externalsystem.mongodb.reactive.api; import com.mongodb.reactivestreams.client.MongoDatabase; -import io.flamingock.api.external.ExternalSystem; +import io.flamingock.internal.common.core.transaction.TransactionalExternalSystem; -public interface MongoDBReactiveExternalSystem extends ExternalSystem { +public interface MongoDBReactiveExternalSystem extends TransactionalExternalSystem { MongoDatabase getMongoDatabase(); } diff --git a/utils/mongodb-reactive-util/build.gradle.kts b/utils/mongodb-reactive-util/build.gradle.kts index 1e742461d..5bec1186c 100644 --- a/utils/mongodb-reactive-util/build.gradle.kts +++ b/utils/mongodb-reactive-util/build.gradle.kts @@ -1,5 +1,6 @@ dependencies { implementation(project(":utils:mongodb-util")) + implementation(project(":utils:flamingock-reactive-util")) implementation(project(":core:flamingock-core-commons")) compileOnly("org.mongodb:mongodb-driver-reactivestreams:4.0.0") diff --git a/utils/mongodb-reactive-util/src/main/java/io/flamingock/internal/common/mongodb/MongoDBReactiveCollectionHelper.java b/utils/mongodb-reactive-util/src/main/java/io/flamingock/internal/common/mongodb/MongoDBReactiveCollectionHelper.java index d8dec7a1e..972e414e1 100644 --- a/utils/mongodb-reactive-util/src/main/java/io/flamingock/internal/common/mongodb/MongoDBReactiveCollectionHelper.java +++ b/utils/mongodb-reactive-util/src/main/java/io/flamingock/internal/common/mongodb/MongoDBReactiveCollectionHelper.java @@ -17,16 +17,10 @@ import com.mongodb.client.model.IndexOptions; import com.mongodb.reactivestreams.client.MongoCollection; -import io.flamingock.internal.common.core.error.FlamingockException; +import io.flamingock.reactive.util.PublisherSync; import org.bson.Document; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; import java.util.List; -import java.util.ArrayList; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; public class MongoDBReactiveCollectionHelper implements CollectionHelper { @@ -44,7 +38,7 @@ public String getCollectionName() { @Override public Iterable listIndexes() { - List indexes = collect(collection.listIndexes()); + List indexes = PublisherSync.collect(collection.listIndexes()); return indexes.stream().map(MongoDBDocumentHelper::new).collect(Collectors.toList()); } @@ -60,67 +54,16 @@ public String createIndex(MongoDBDocumentHelper keyDocument, if (partialFilterExpression != null) { options.partialFilterExpression(partialFilterExpression.getDocument()); } - return first(collection.createIndex(keyDocument.getDocument(), options)); + return PublisherSync.first(collection.createIndex(keyDocument.getDocument(), options)); } @Override public void dropIndex(String indexName) { - complete(collection.dropIndex(indexName)); + PublisherSync.complete(collection.dropIndex(indexName)); } @Override public void deleteMany(MongoDBDocumentHelper documentWrapper) { - first(collection.deleteMany(documentWrapper.getDocument())); - } - - private static T first(Publisher publisher) { - List values = collect(publisher); - return values.isEmpty() ? null : values.get(0); - } - - private static List collect(Publisher publisher) { - List values = new ArrayList<>(); - AtomicReference error = new AtomicReference<>(); - CountDownLatch latch = new CountDownLatch(1); - publisher.subscribe(new Subscriber() { - @Override - public void onSubscribe(Subscription subscription) { - subscription.request(Long.MAX_VALUE); - } - - @Override - public void onNext(T value) { - values.add(value); - } - - @Override - public void onError(Throwable throwable) { - error.set(throwable); - latch.countDown(); - } - - @Override - public void onComplete() { - latch.countDown(); - } - }); - try { - latch.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new FlamingockException(e); - } - Throwable throwable = error.get(); - if (throwable instanceof RuntimeException) { - throw (RuntimeException) throwable; - } - if (throwable != null) { - throw new FlamingockException(throwable); - } - return values; - } - - private static void complete(Publisher publisher) { - collect(publisher); + PublisherSync.first(collection.deleteMany(documentWrapper.getDocument())); } }