From b1822b5614ad631a06adaaf16dba66c451069ade Mon Sep 17 00:00:00 2001 From: davidfrigolet Date: Thu, 6 Aug 2026 15:37:34 +0100 Subject: [PATCH] feat(couchbase): add journal event persistence --- .../store/couchbase/CouchbaseAuditStore.java | 101 +++++-- .../internal/CouchbaseAuditPersistence.java | 67 ++++- .../couchbase/internal/CouchbaseAuditor.java | 63 +++- .../internal/CouchbaseJournalEventStore.java | 214 +++++++++++++ .../CouchbaseJournalFeatureFlagE2ETest.java | 178 +++++++++++ .../CouchbaseAuditPersistenceJournalTest.java | 284 ++++++++++++++++++ .../build.gradle.kts | 2 +- .../api/CouchbaseExternalSystem.java | 4 +- .../couchbase/CouchbaseCollectionHelper.java | 17 ++ .../CouchbaseJournalEventMapper.java | 89 ++++++ .../journal/JournalEventFieldConstants.java | 37 +++ .../JournalEventPersistenceConstants.java | 28 ++ 12 files changed, 1039 insertions(+), 45 deletions(-) create mode 100644 community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseJournalEventStore.java create mode 100644 community/flamingock-couchbase-auditstore/src/test/java/io/flamingock/store/couchbase/CouchbaseJournalFeatureFlagE2ETest.java create mode 100644 community/flamingock-couchbase-auditstore/src/test/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistenceJournalTest.java create mode 100644 utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/CouchbaseJournalEventMapper.java create mode 100644 utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/journal/JournalEventFieldConstants.java create mode 100644 utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/journal/JournalEventPersistenceConstants.java diff --git a/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/CouchbaseAuditStore.java b/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/CouchbaseAuditStore.java index 8208476ad..1d6f04385 100644 --- a/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/CouchbaseAuditStore.java +++ b/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/CouchbaseAuditStore.java @@ -18,38 +18,58 @@ import com.couchbase.client.core.io.CollectionIdentifier; import com.couchbase.client.java.Bucket; import com.couchbase.client.java.Cluster; +import com.couchbase.client.java.transactions.TransactionAttemptContext; +import io.flamingock.internal.common.core.audit.AuditPersistenceFactory; +import io.flamingock.internal.common.core.audit.AuditReader; import io.flamingock.internal.common.core.context.ContextResolver; import io.flamingock.internal.common.core.error.FlamingockException; import io.flamingock.internal.core.configuration.community.CommunityConfigurable; import io.flamingock.internal.core.external.store.CommunityAuditStore; import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence; import io.flamingock.internal.core.external.store.lock.community.CommunityLockService; +import io.flamingock.internal.core.journal.JournalEventSequencer; +import io.flamingock.internal.core.journal.JournalEventSequencerFactory; import io.flamingock.internal.util.Constants; import io.flamingock.internal.util.TimeService; import io.flamingock.internal.util.constants.CommunityPersistenceConstants; +import io.flamingock.internal.common.couchbase.journal.JournalEventPersistenceConstants; import io.flamingock.internal.util.id.RunnerId; import io.flamingock.store.couchbase.internal.CouchbaseAuditPersistence; +import io.flamingock.store.couchbase.internal.CouchbaseAuditor; +import io.flamingock.store.couchbase.internal.CouchbaseJournalEventStore; import io.flamingock.store.couchbase.internal.CouchbaseLockService; import io.flamingock.externalsystem.couchbase.api.CouchbaseExternalSystem; +import java.util.Collections; +import java.util.Set; + public class CouchbaseAuditStore implements CommunityAuditStore { + private final CouchbaseExternalSystem targetSystem; private final Cluster cluster; private final String bucketName; private RunnerId runnerId; private CommunityConfigurable communityConfiguration; - private CouchbaseAuditPersistence persistence; private CouchbaseLockService lockService; private Bucket bucket; private String scopeName = CollectionIdentifier.DEFAULT_SCOPE; private String auditRepositoryName = CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME; private String lockRepositoryName = CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME; + private String journalRepositoryName = JournalEventPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME; private boolean autoCreate = true; - - - private CouchbaseAuditStore(Cluster cluster, String bucketName) { - this.cluster = cluster; - this.bucketName = bucketName; + private CouchbaseAuditor auditor; + private CouchbaseJournalEventStore journalEventStore; + private JournalEventSequencerFactory journalEventSequencerFactory; + + + private CouchbaseAuditStore(CouchbaseExternalSystem targetSystem) { + // Cannot resolve targetSystem.getTxWrapper() here: the target system's own initialize() — which is + // what sets it — runs later than this constructor (called eagerly when the caller builds this audit + // store), so it would still be null at this point. Kept as a live reference and resolved lazily in + // getPersistenceFactory(), by which point both the target system and this audit store are initialized. + this.targetSystem = targetSystem; + this.cluster = targetSystem.getCluster(); + this.bucketName = targetSystem.getBucketName(); } /** @@ -63,7 +83,7 @@ private CouchbaseAuditStore(Cluster cluster, String bucketName) { * @return a new audit store bound to the same Couchbase instance as the target system */ public static CouchbaseAuditStore from(CouchbaseExternalSystem targetSystem) { - return new CouchbaseAuditStore(targetSystem.getCluster(), targetSystem.getBucketName()); + return new CouchbaseAuditStore(targetSystem); } @Override @@ -86,6 +106,11 @@ public CouchbaseAuditStore withLockRepositoryName(String lockRepositoryName) { return this; } + public CouchbaseAuditStore withJournalRepositoryName(String journalRepositoryName) { + this.journalRepositoryName = journalRepositoryName; + return this; + } + public CouchbaseAuditStore withAutoCreate(boolean autoCreate) { this.autoCreate = autoCreate; return this; @@ -93,38 +118,58 @@ public CouchbaseAuditStore withAutoCreate(boolean autoCreate) { @Override public void initialize(ContextResolver baseContext) { + this.validate(); runnerId = baseContext.getRequiredDependencyValue(RunnerId.class); communityConfiguration = baseContext.getRequiredDependencyValue(CommunityConfigurable.class); - this.validate(); + + auditor = new CouchbaseAuditor(cluster, bucket); + journalEventStore = new CouchbaseJournalEventStore(cluster, bucket); + journalEventSequencerFactory = new JournalEventSequencerFactory(journalEventStore); + + lockService = new CouchbaseLockService(cluster, bucket, TimeService.getDefault()); + lockService.initialize(autoCreate, scopeName, lockRepositoryName); } @Override - public synchronized CommunityAuditPersistence getPersistence() { - if (persistence == null) { - persistence = new CouchbaseAuditPersistence( + public AuditPersistenceFactory getPersistenceFactory() { + return stageId -> { + JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId); + CouchbaseAuditPersistence persistence = new CouchbaseAuditPersistence( communityConfiguration, - cluster, - bucket, + auditor, + journalEventStore, + journalEventSequencer, + targetSystem.getTxWrapper(), scopeName, auditRepositoryName, + journalRepositoryName, autoCreate); persistence.initialize(runnerId); - } - return persistence; + return persistence; + }; + } + + @Override + public CommunityAuditPersistence getPersistence() { + throw new UnsupportedOperationException("getPersistence shouldn't be called at Couchbase audit store; use getPersistenceFactory(stageId)"); + } + + @Override + public AuditReader getAuditReader() { + auditor.initialize(autoCreate, scopeName, auditRepositoryName); + return () -> auditor.getAuditHistory(); } @Override public synchronized CommunityLockService getLockService() { - if (lockService == null) { - lockService = new CouchbaseLockService(cluster, bucket, TimeService.getDefault()); - lockService.initialize( - autoCreate, - scopeName, - lockRepositoryName); - } return lockService; } + @Override + public Set> getNonGuardedTypes() { + return Collections.singleton(TransactionAttemptContext.class); + } + private void validate() { if (cluster == null) { @@ -152,8 +197,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."); + } } } diff --git a/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistence.java b/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistence.java index 6ffe9a77d..2f91f9c47 100644 --- a/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistence.java +++ b/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistence.java @@ -15,11 +15,17 @@ */ package io.flamingock.store.couchbase.internal; -import com.couchbase.client.java.Bucket; -import com.couchbase.client.java.Cluster; +import com.couchbase.client.java.transactions.TransactionAttemptContext; 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.util.FeatureFlag; import io.flamingock.internal.util.Result; import io.flamingock.internal.util.id.RunnerId; @@ -27,35 +33,47 @@ public class CouchbaseAuditPersistence extends AbstractCommunityAuditPersistence { - private final Cluster cluster; - private final Bucket bucket; + private final CouchbaseAuditor auditor; + private final CouchbaseJournalEventStore journalEventStore; + private final JournalEventSequencer journalEventSequencer; + private final TransactionWrapper txWrapper; private final String scopeName; private final String auditRepositoryName; + private final String journalRepositoryName; private final boolean autoCreate; - private CouchbaseAuditor auditor; - public CouchbaseAuditPersistence(CommunityConfigurable localConfiguration, - Cluster cluster, - Bucket bucket, + CouchbaseAuditor auditor, + CouchbaseJournalEventStore journalEventStore, + JournalEventSequencer journalEventSequencer, + TransactionWrapper txWrapper, String scopeName, String auditRepositoryName, + String journalRepositoryName, boolean autoCreate) { super(localConfiguration); - this.cluster = cluster; - this.bucket = bucket; + this.auditor = auditor; + this.journalEventStore = journalEventStore; + this.journalEventSequencer = journalEventSequencer; + this.txWrapper = txWrapper; this.scopeName = scopeName; this.auditRepositoryName = auditRepositoryName; + this.journalRepositoryName = journalRepositoryName; this.autoCreate = autoCreate; } @Override protected void doInitialize(RunnerId runnerId) { - auditor = new CouchbaseAuditor(cluster, bucket); auditor.initialize(autoCreate, scopeName, auditRepositoryName); + // Creating the collection/indexes is what brings the journal collection into existence, so skipping + // this keeps it from ever appearing while the flag is off. It must stay in step with the append in + // writeEntry: skipping setup while still appending would let ctx.insert create the collection + // implicitly and without indexes, voiding the stream-position and eventId-lookup guarantees. + FeatureFlag.ifEnabled(Features.JOURNAL_EVENTS, () -> journalEventStore.initialize(autoCreate, scopeName, journalRepositoryName)); } + @Override public List getAuditHistory() { return auditor.getAuditHistory(); @@ -63,6 +81,31 @@ public List getAuditHistory() { @Override public Result writeEntry(AuditEntry auditEntry) { - return auditor.writeEntry(auditEntry); + // Read once rather than per branch: the journal append and the audit write shape are two halves of one + // model. With events, the audit record is the change's current state and the journal is the history; + // without them, the audit record set is itself the history. + if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS)) { + RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId()); + Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> { + TransactionAttemptContext ctx = runtimeContext.getContext().getRequiredDependencyValue(TransactionAttemptContext.class); + JournalEvent journalEvent = journalEventSequencer.newEvent(auditEntry); + journalEventStore.contributeToTransaction(ctx, journalEvent); + return auditor.contributeToTransaction(ctx, auditEntry); + }); + // Spends the stream position, and only a committed transaction attempt may reach this line. A + // normal return from wrapInTransaction does NOT in general mean commit — CouchbaseTxWrapper + // returns normally after a deliberate rollback too, when the operation's result is a FailedStep. + // It is sound here because this operation returns a Result, which can never be a FailedStep, so + // the only way to return normally is a committed attempt; a failing attempt is caught and + // rethrown as TransactionFailedException (see CouchbaseTxWrapper — it doesn't yet wrap that as + // DatabaseTransactionException, a known deviation from the TransactionWrapper contract, tracked + // separately from this ticket). Keep that true: an operation that could return a failed step + // would silently burn a position and gap the stream, and a contiguous sequence is what lets a + // consumer tell "in flight" from "lost". + journalEventSequencer.confirm(); + return result; + } else { + return auditor.append(auditEntry); + } } } diff --git a/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditor.java b/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditor.java index 3e9f84248..e67f8928b 100644 --- a/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditor.java +++ b/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditor.java @@ -16,6 +16,7 @@ package io.flamingock.store.couchbase.internal; import com.couchbase.client.core.error.CouchbaseException; +import com.couchbase.client.core.error.DocumentNotFoundException; import com.couchbase.client.java.Bucket; import com.couchbase.client.java.Cluster; import com.couchbase.client.java.Collection; @@ -23,9 +24,9 @@ import com.couchbase.client.java.kv.PersistTo; import com.couchbase.client.java.kv.ReplicateTo; import com.couchbase.client.java.kv.UpsertOptions; +import com.couchbase.client.java.transactions.TransactionAttemptContext; +import com.couchbase.client.java.transactions.TransactionGetResult; 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.couchbase.CouchbaseAuditMapper; import io.flamingock.internal.common.couchbase.CouchbaseCollectionHelper; import io.flamingock.internal.common.couchbase.CouchbaseCollectionInitializator; @@ -37,7 +38,11 @@ import java.util.stream.Collectors; -public class CouchbaseAuditor implements AuditWriter, AuditReader { +/** + * Internal to this module — not exposed as {@code AuditWriter}/{@code AuditReader}, since callers reach it + * only through {@link CouchbaseAuditPersistence}, which owns the {@code Features.JOURNAL_EVENTS} branching. + */ +public class CouchbaseAuditor { private static final Logger logger = FlamingockLoggerFactory.getLogger("CouchbaseAuditor"); @@ -45,23 +50,40 @@ public class CouchbaseAuditor implements AuditWriter, AuditReader { protected final Bucket bucket; protected Collection collection; protected CouchbaseCollectionInitializator collectionInitializator; + private boolean initialized = false; private final CouchbaseAuditMapper mapper = new CouchbaseAuditMapper(); - protected CouchbaseAuditor(Cluster cluster, Bucket bucket) { + public CouchbaseAuditor(Cluster cluster, Bucket bucket) { this.cluster = cluster; this.bucket = bucket; } - protected void initialize(boolean autoCreate, String scopeName, String collectionName) { + /** + * A no-op past the first call: shared across every stage's persistence, so each new stage would otherwise + * re-run the create-if-not-exists calls needlessly. + */ + public synchronized void initialize(boolean autoCreate, String scopeName, String collectionName) { + if (initialized) { + return; + } this.collectionInitializator = new CouchbaseCollectionInitializator(cluster, bucket, scopeName, collectionName); this.collectionInitializator.initialize(autoCreate); this.collection = this.bucket.scope(scopeName).collection(collectionName); + initialized = true; } - @Override - public Result writeEntry(AuditEntry auditEntry) { + /** + * Keeps one record per {@code (executionId, changeId, state)} — the append-oriented audit ledger, where a + * change accumulates a document per state transition and the collection is itself the history. + *

+ * This is the behaviour used when journal events are disabled, and it is what the Mongock importer needs + * regardless: a legacy changelog can hold several entries for the same change across executions, and + * {@link #contributeToTransaction} would collapse them onto each other, discarding the very history being + * imported. + */ + Result append(AuditEntry auditEntry) { String key = toKey(auditEntry); logger.debug("Saving audit entry with key {}", key); @@ -79,8 +101,33 @@ public Result writeEntry(AuditEntry auditEntry) { return Result.OK(); } + /** + * Keeps a single document per change, overwritten on every state transition — the change's current + * state — within the caller's transaction attempt. + *

+ * The history of how it got there lives in the journal, so this is only correct when journal events are + * being written; see {@code CouchbaseAuditPersistence.writeEntry}. + *

+ * Keyed on {@code changeId} alone, which is safe because {@code LoadedPipeline.validate()} rejects + * duplicate change ids across all stages. Nothing at the database level enforces one-document-per-change — + * the single-writer guarantee comes from the stage lock. Couchbase transactions have no {@code upsert}, so + * this reads first and replaces on a hit, inserting only on {@link DocumentNotFoundException} — the same + * idiom {@code CouchbaseTargetSystemAuditMarker.mark()} already uses. + */ + Result contributeToTransaction(TransactionAttemptContext ctx, AuditEntry auditEntry) { + String key = auditEntry.getChangeId(); + JsonObject document = mapper.toDocument(auditEntry); + try { + TransactionGetResult existing = ctx.get(collection, key); + ctx.replace(existing, document); + } catch (DocumentNotFoundException e) { + ctx.insert(collection, key, document); + } + logger.debug("Staged current-state audit entry with key {}", key); + return Result.OK(); + } + - @Override public List getAuditHistory() { return CouchbaseCollectionHelper.selectAllDocuments(cluster, collection.bucketName(), collection.scopeName(), collection.name()) .stream() diff --git a/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseJournalEventStore.java b/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseJournalEventStore.java new file mode 100644 index 000000000..4497c5315 --- /dev/null +++ b/community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseJournalEventStore.java @@ -0,0 +1,214 @@ +/* + * 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.couchbase.internal; + +import com.couchbase.client.java.Bucket; +import com.couchbase.client.java.Cluster; +import com.couchbase.client.java.Collection; +import com.couchbase.client.java.json.JsonArray; +import com.couchbase.client.java.json.JsonObject; +import com.couchbase.client.java.query.QueryOptions; +import com.couchbase.client.java.query.QueryResult; +import com.couchbase.client.java.query.QueryScanConsistency; +import com.couchbase.client.java.transactions.TransactionAttemptContext; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.couchbase.CouchbaseCollectionHelper; +import io.flamingock.internal.common.couchbase.CouchbaseCollectionInitializator; +import io.flamingock.internal.common.couchbase.CouchbaseJournalEventMapper; +import io.flamingock.internal.core.journal.JournalEventStore; +import io.flamingock.internal.util.Result; +import io.flamingock.internal.util.log.FlamingockLoggerFactory; +import org.slf4j.Logger; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_ACKNOWLEDGED; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_EVENT_ID; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_STREAM_ID; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_STREAM_SEQUENCE; + +/** + * Couchbase implementation of the local journal ({@code flamingockJournalEvents}). + *

+ * Sibling of {@link CouchbaseAuditor}/{@link CouchbaseLockService}: it owns its own collection and index setup. + *

+ * Couchbase has no unique secondary indexes, so the document key is the only place the stream-position + * invariant can be enforced: each event is keyed {@code journal::::} and appended + * with an {@code insert} (never an {@code upsert}), so a colliding position fails loudly with a + * {@code DocumentExistsException} instead of being silently overwritten. {@code eventId} uniqueness is not + * enforced here — the backend deduplicates on it independently — so its index only serves the acknowledgement + * lookup. + *

+ * Reads and acknowledgements are exposed through {@link JournalEventStore}. The append + * ({@link #contributeToTransaction(TransactionAttemptContext, JournalEvent)}) deliberately is not: it takes + * the {@link TransactionAttemptContext} the transaction wrapper injects, so the event lands in the same + * transaction attempt as the audit entry it mirrors, and a driver-specific transaction handle has no place in + * a core interface. Package-private, and only {@link CouchbaseAuditPersistence} — which owns that transaction + * boundary — calls it. + */ +public class CouchbaseJournalEventStore implements JournalEventStore { + + private static final Logger logger = FlamingockLoggerFactory.getLogger("CouchbaseJournal"); + + private static final String KEY_PREFIX = "journal"; + + static final String STREAM_SEQUENCE_INDEX_NAME = "idx_journal_stream_sequence"; + static final String UNACKNOWLEDGED_INDEX_NAME = "idx_journal_unacknowledged"; + static final String EVENT_ID_INDEX_NAME = "idx_journal_event_id"; + + private final Cluster cluster; + private final Bucket bucket; + private final CouchbaseJournalEventMapper mapper = new CouchbaseJournalEventMapper(); + + private Collection collection; + private boolean initialized = false; + + public CouchbaseJournalEventStore(Cluster cluster, Bucket bucket) { + this.cluster = cluster; + this.bucket = bucket; + } + + /** + * Creates the collection and its indexes (or validates them, when {@code autoCreate} is {@code false}). + * A no-op if already initialized, or gated out entirely by the caller under + * {@code Features.JOURNAL_EVENTS} — see {@link CouchbaseAuditPersistence#doInitialize}. + */ + synchronized void initialize(boolean autoCreate, String scopeName, String collectionName) { + if (initialized) { + return; + } + // Collection + primary index: the same concern CouchbaseAuditor/CouchbaseLockService delegate to this + // helper, so ad-hoc tooling and test cleanup that scan "all documents" keep working here too. + new CouchbaseCollectionInitializator(cluster, bucket, scopeName, collectionName).initialize(autoCreate); + if (autoCreate) { + createIndexes(scopeName, collectionName); + } else { + requireIndex(scopeName, collectionName, STREAM_SEQUENCE_INDEX_NAME); + requireIndex(scopeName, collectionName, UNACKNOWLEDGED_INDEX_NAME); + requireIndex(scopeName, collectionName, EVENT_ID_INDEX_NAME); + } + this.collection = bucket.scope(scopeName).collection(collectionName); + initialized = true; + } + + /** + * Three indexes for the event buffer: + *

    + *
  • {@code (streamId, streamSequence)} serving "last event per stream" (reverse scan);
    • + *
  • a partial {@code (acknowledged, streamId, streamSequence)} index over {@code acknowledged = false}, + * serving the ordered unacknowledged batch scan while staying sized to the backlog;
    • + *
  • a non-unique {@code eventId} index serving the {@link #acknowledgeEvents(java.util.Collection)} + * lookup — {@code eventId} uniqueness cannot be enforced by Couchbase and is not this index's job; + * the stream-position document key is the real backstop.
    • + *
+ */ + private void createIndexes(String scopeName, String collectionName) { + CouchbaseCollectionHelper.createIndexIfNotExists(cluster, bucket.name(), scopeName, collectionName, + STREAM_SEQUENCE_INDEX_NAME, KEY_STREAM_ID + ", " + KEY_STREAM_SEQUENCE, null); + CouchbaseCollectionHelper.createIndexIfNotExists(cluster, bucket.name(), scopeName, collectionName, + UNACKNOWLEDGED_INDEX_NAME, KEY_ACKNOWLEDGED + ", " + KEY_STREAM_ID + ", " + KEY_STREAM_SEQUENCE, + KEY_ACKNOWLEDGED + " = false"); + CouchbaseCollectionHelper.createIndexIfNotExists(cluster, bucket.name(), scopeName, collectionName, + EVENT_ID_INDEX_NAME, KEY_EVENT_ID, null); + } + + private void requireIndex(String scopeName, String collectionName, String indexName) { + if (!CouchbaseCollectionHelper.indexExists(cluster, bucket.name(), scopeName, collectionName, indexName)) { + throw new RuntimeException(String.format( + "Auto-creation is disabled and required journal index '%s' does not exist on `%s`.`%s`.`%s`", + indexName, bucket.name(), scopeName, collectionName)); + } + } + + /** + * Appends an event within the caller's transaction attempt, keyed {@code journal::::}. + *

+ * Uses {@code insert}, never {@code upsert}: an event that is already there is a defect, not something to + * overwrite. If the single-writer-per-stream assumption is ever violated, the second writer collides on + * this key instead of silently duplicating or clobbering, and the resulting {@code DocumentExistsException} + * aborts the transaction attempt, taking the audit entry with it. + * + * @param ctx the transaction attempt this append must join + * @param event the event to append + * @return {@link Result#OK()} — failures surface as exceptions, not as a result + */ + Result contributeToTransaction(TransactionAttemptContext ctx, JournalEvent event) { + if (!initialized) { + throw new IllegalStateException("Couchbase journal store is not initialized"); + } + String key = toKey(event.getStreamId(), event.getStreamSequence()); + JsonObject document = mapper.toDocument(event); + ctx.insert(collection, key, document); + 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) { + if (!initialized) { + return Optional.empty(); + } + String query = String.format( + "SELECT c.* FROM `%s`.`%s`.`%s` AS c WHERE c.%s = $streamId ORDER BY c.%s DESC LIMIT 1", + collection.bucketName(), collection.scopeName(), collection.name(), KEY_STREAM_ID, KEY_STREAM_SEQUENCE); + QueryResult result = cluster.query(query, QueryOptions.queryOptions() + .scanConsistency(QueryScanConsistency.REQUEST_PLUS) + .parameters(JsonObject.create().put("streamId", streamId))); + List rows = result.rowsAsObject(); + return rows.isEmpty() ? Optional.empty() : Optional.of(mapper.fromDocument(rows.get(0))); + } + + @Override + public List> getUnacknowledgedEvents(int limit) { + if (!initialized) { + return new ArrayList<>(); + } + String query = String.format( + "SELECT c.* FROM `%s`.`%s`.`%s` AS c WHERE c.%s = false ORDER BY c.%s, c.%s LIMIT $limit", + collection.bucketName(), collection.scopeName(), collection.name(), + KEY_ACKNOWLEDGED, KEY_STREAM_ID, KEY_STREAM_SEQUENCE); + QueryResult result = cluster.query(query, QueryOptions.queryOptions() + .scanConsistency(QueryScanConsistency.REQUEST_PLUS) + .parameters(JsonObject.create().put("limit", limit))); + List> events = new ArrayList<>(); + for (JsonObject row : result.rowsAsObject()) { + events.add(mapper.fromDocument(row)); + } + return events; + } + + @Override + public long acknowledgeEvents(java.util.Collection eventIds) { + if (!initialized || eventIds == null || eventIds.isEmpty()) { + return 0L; + } + String query = String.format( + "UPDATE `%s`.`%s`.`%s` SET %s = true WHERE %s IN $eventIds RETURNING META().id", + collection.bucketName(), collection.scopeName(), collection.name(), KEY_ACKNOWLEDGED, KEY_EVENT_ID); + QueryResult result = cluster.query(query, QueryOptions.queryOptions() + .scanConsistency(QueryScanConsistency.REQUEST_PLUS) + .parameters(JsonObject.create().put("eventIds", JsonArray.from(new ArrayList<>(eventIds))))); + return result.rowsAsObject().size(); + } + + private static String toKey(String streamId, long streamSequence) { + return KEY_PREFIX + "::" + streamId + "::" + streamSequence; + } +} diff --git a/community/flamingock-couchbase-auditstore/src/test/java/io/flamingock/store/couchbase/CouchbaseJournalFeatureFlagE2ETest.java b/community/flamingock-couchbase-auditstore/src/test/java/io/flamingock/store/couchbase/CouchbaseJournalFeatureFlagE2ETest.java new file mode 100644 index 000000000..953fd8fa5 --- /dev/null +++ b/community/flamingock-couchbase-auditstore/src/test/java/io/flamingock/store/couchbase/CouchbaseJournalFeatureFlagE2ETest.java @@ -0,0 +1,178 @@ +/* + * 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.couchbase; + +import com.couchbase.client.core.io.CollectionIdentifier; +import com.couchbase.client.java.Bucket; +import com.couchbase.client.java.Cluster; +import com.couchbase.client.java.ClusterOptions; +import com.couchbase.client.java.Collection; +import io.flamingock.common.test.pipeline.CodeChangeTestDefinition; +import io.flamingock.core.kit.audit.AuditEntryExpectation; +import io.flamingock.core.kit.audit.AuditTestSupport; +import io.flamingock.couchbase.kit.CouchbaseTestKit; +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.couchbase.CouchbaseCollectionHelper; +import io.flamingock.internal.common.couchbase.CouchbaseJournalEventMapper; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.constants.CommunityPersistenceConstants; +import io.flamingock.store.couchbase.changes.happyPath._002__insert_document; +import io.flamingock.targetsystem.couchbase.CouchbaseTargetSystem; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.testcontainers.couchbase.BucketDefinition; +import org.testcontainers.couchbase.CouchbaseContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import java.time.Duration; +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 {@link Features#JOURNAL_EVENTS} gate through a complete runner execution. + */ +@Testcontainers +class CouchbaseJournalFeatureFlagE2ETest { + + private static final String BUCKET_NAME = "test"; + private static final String SCOPE_NAME = CollectionIdentifier.DEFAULT_SCOPE; + private static final String JOURNAL_COLLECTION = "flamingockJournalEvents"; + private static final String DEFAULT_STAGE_NAME = "default-stage-name"; + + @Container + static final CouchbaseContainer couchbaseContainer = new CouchbaseContainer("couchbase/server:7.2.4") + .withBucket(new BucketDefinition(BUCKET_NAME)); + + private static Cluster cluster; + + private final CouchbaseJournalEventMapper mapper = new CouchbaseJournalEventMapper(); + + private CouchbaseTargetSystem targetSystem; + private CouchbaseAuditStore auditStore; + private CouchbaseTestKit testKit; + + @BeforeAll + static void beforeAll() { + couchbaseContainer.start(); + // Default KV timeout (2.5s) is too tight for a just-created collection: Couchbase's KV service can + // take a few seconds to pick up a brand-new collection's manifest entry, and the very first KV op + // against it (here, the test kit's own audit-history check) can hit that gap and time out with + // UnambiguousTimeoutException/KV_COLLECTION_OUTDATED — nothing to do with the journal logic under + // test. Widening it gives the SDK's own retry loop room to ride out that window. + cluster = Cluster.connect( + couchbaseContainer.getConnectionString(), + ClusterOptions.clusterOptions(couchbaseContainer.getUsername(), couchbaseContainer.getPassword()) + .environment(env -> env.timeoutConfig(timeouts -> timeouts.kvTimeout(Duration.ofSeconds(10))))); + cluster.bucket(BUCKET_NAME).waitUntilReady(Duration.ofSeconds(10)); + } + + @BeforeEach + void setUp() { + targetSystem = new CouchbaseTargetSystem("couchbase", cluster, BUCKET_NAME); + auditStore = CouchbaseAuditStore.from(targetSystem); + testKit = CouchbaseTestKit.create(auditStore, cluster, BUCKET_NAME, SCOPE_NAME); + } + + @AfterEach + void tearDown() { + // The flag is process-global and every test class in this module shares one JVM, so leaving it on + // would silently make later classes create and write the journal collection. + FeatureFlag.remove(Features.JOURNAL_EVENTS); + testKit.cleanUp(); + } + + @Test + @DisplayName("journal disabled: the audit log retains every state transition") + void journalDisabledRetainsHistoricalAuditEntries() { + runPipeline(STARTED("insert-document"), APPLIED("insert-document")); + + assertFalse(CouchbaseCollectionHelper.collectionExists(cluster, BUCKET_NAME, SCOPE_NAME, JOURNAL_COLLECTION), + "the journal collection must not exist when the feature is disabled"); + } + + @Test + @DisplayName("journal enabled: an audit-only installation transparently creates the journal") + void journalEnabledSplitsCurrentStateFromHistory() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + + runPipeline(APPLIED("insert-document")); + + assertTrue(CouchbaseCollectionHelper.collectionExists(cluster, BUCKET_NAME, SCOPE_NAME, + CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME), + "the existing audit collection must remain available"); + assertTrue(CouchbaseCollectionHelper.collectionExists(cluster, BUCKET_NAME, SCOPE_NAME, JOURNAL_COLLECTION)); + + List auditRecords = new CouchbaseTestHelper(cluster) + .getAuditEntriesSorted(cluster.bucket(BUCKET_NAME).scope(SCOPE_NAME) + .collection(CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME)); + assertEquals(1, auditRecords.size(), "the audit collection must retain only the current state when journal is enabled"); + assertEquals(AuditEntry.Status.APPLIED, auditRecords.get(0).getState()); + + List> events = storedEvents(); + assertEquals(2, events.size(), "one event must be stored for each audit state transition"); + assertTrue(events.stream().allMatch(event -> DEFAULT_STAGE_NAME.equals(event.getStreamId())), + "journal events must use the pipeline stage as their stream"); + assertEquals(Arrays.asList(1L, 2L), events.stream() + .map(JournalEvent::getStreamSequence) + .sorted() + .collect(Collectors.toList()), + "journal stream sequences must be contiguous from one"); + assertEquals(Arrays.asList(AuditEntry.Status.STARTED, AuditEntry.Status.APPLIED), events.stream() + .map(event -> event.getData().getState()) + .sorted() + .collect(Collectors.toList()), + "the journal must retain both audit state transitions"); + } + + private void runPipeline(AuditEntryExpectation... expectedAudits) { + Bucket bucket = cluster.bucket(BUCKET_NAME); + Collection testCollection = bucket.defaultCollection(); + AuditTestSupport.withTestKit(testKit) + .GIVEN_Changes(new CodeChangeTestDefinition( + _002__insert_document.class, + Collections.singletonList(Collection.class))) + .WHEN(() -> testKit.createBuilder() + .setAuditStore(auditStore) + .addTargetSystem(targetSystem) + .addDependency(testCollection) + .build() + .run()) + .THEN_VerifyAuditSequenceStrict(expectedAudits) + .run(); + } + + private List> storedEvents() { + List> events = new ArrayList<>(); + CouchbaseCollectionHelper.selectAllDocuments(cluster, BUCKET_NAME, SCOPE_NAME, JOURNAL_COLLECTION) + .forEach(document -> events.add(mapper.fromDocument(document))); + return events; + } +} diff --git a/community/flamingock-couchbase-auditstore/src/test/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistenceJournalTest.java b/community/flamingock-couchbase-auditstore/src/test/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistenceJournalTest.java new file mode 100644 index 000000000..182f210e6 --- /dev/null +++ b/community/flamingock-couchbase-auditstore/src/test/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistenceJournalTest.java @@ -0,0 +1,284 @@ +/* + * 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.couchbase.internal; + +import com.couchbase.client.core.io.CollectionIdentifier; +import com.couchbase.client.java.Bucket; +import com.couchbase.client.java.Cluster; +import com.couchbase.client.java.ClusterOptions; +import com.couchbase.client.java.transactions.TransactionAttemptContext; +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.common.couchbase.CouchbaseCollectionHelper; +import io.flamingock.internal.common.couchbase.CouchbaseJournalEventMapper; +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.targetsystem.couchbase.CouchbaseTxWrapper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.testcontainers.couchbase.BucketDefinition; +import org.testcontainers.couchbase.CouchbaseContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; + +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.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; + +/** + * Covers the {@code Features.JOURNAL_EVENTS} gate in {@link CouchbaseAuditPersistence}, and the transaction + * boundary the journal exists for: the audit entry and its event must commit or roll back together. + *

+ * Drives the persistence directly rather than a full runner, because forcing a failure between the two writes + * is only practical at this level. + */ +@Testcontainers +class CouchbaseAuditPersistenceJournalTest { + + private static final String BUCKET_NAME = "test"; + private static final String SCOPE_NAME = CollectionIdentifier.DEFAULT_SCOPE; + private static final String AUDIT_COLLECTION = "flamingockAuditPersistenceJournalTest"; + private static final String JOURNAL_COLLECTION = "flamingockJournalPersistenceJournalTest"; + private static final String STREAM_ID = "stage-under-test"; + + @Container + static final CouchbaseContainer couchbaseContainer = new CouchbaseContainer("couchbase/server:7.2.4") + .withBucket(new BucketDefinition(BUCKET_NAME)); + + private static Cluster cluster; + private static Bucket bucket; + + private final CouchbaseJournalEventMapper mapper = new CouchbaseJournalEventMapper(); + + private CouchbaseAuditor auditor; + private CouchbaseJournalEventStore journalEventStore; + private CouchbaseTxWrapper txWrapper; + + @BeforeAll + static void beforeAll() { + couchbaseContainer.start(); + // Default KV timeout (2.5s) is too tight for a just-created collection: Couchbase's KV service can + // take a few seconds to pick up a brand-new collection's manifest entry, and the first KV op against + // it can hit that gap and time out with UnambiguousTimeoutException/KV_COLLECTION_OUTDATED. Widening + // it gives the SDK's own retry loop room to ride out that window. + cluster = Cluster.connect( + couchbaseContainer.getConnectionString(), + ClusterOptions.clusterOptions(couchbaseContainer.getUsername(), couchbaseContainer.getPassword()) + .environment(env -> env.timeoutConfig(timeouts -> timeouts.kvTimeout(Duration.ofSeconds(10))))); + bucket = cluster.bucket(BUCKET_NAME); + bucket.waitUntilReady(Duration.ofSeconds(10)); + } + + @BeforeEach + void setUp() { + auditor = new CouchbaseAuditor(cluster, bucket); + journalEventStore = new CouchbaseJournalEventStore(cluster, bucket); + txWrapper = new CouchbaseTxWrapper(cluster, new TransactionManager<>(() -> { + throw new UnsupportedOperationException( + "Supplier is unused: the wrapper registers the TransactionAttemptContext itself"); + })); + } + + @AfterEach + void tearDown() { + // The flag is process-global and every test class in this module shares one JVM, so leaving it on + // would silently make later classes create and write the journal collection. + FeatureFlag.remove(Features.JOURNAL_EVENTS); + CouchbaseCollectionHelper.dropCollectionIfExists(cluster, BUCKET_NAME, SCOPE_NAME, AUDIT_COLLECTION); + CouchbaseCollectionHelper.dropCollectionIfExists(cluster, BUCKET_NAME, SCOPE_NAME, JOURNAL_COLLECTION); + } + + @Test + @DisplayName("journal disabled: the audit entry is written, and the journal collection is never created") + void journalDisabledWritesNoEventAndCreatesNoCollection() { + CouchbaseAuditPersistence persistence = persistenceFor(auditor); + + persistence.writeEntry(auditEntry("change-1")); + + assertEquals(1, auditor.getAuditHistory().size(), "the audit entry must still be written"); + assertFalse(CouchbaseCollectionHelper.collectionExists(cluster, BUCKET_NAME, SCOPE_NAME, JOURNAL_COLLECTION), + "the journal collection must not exist when the feature is disabled"); + } + + @Test + @DisplayName("journal enabled: the event is written alongside the audit entry, on the persistence's stream") + void journalEnabledWritesEventWithAuditEntry() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + CouchbaseAuditPersistence persistence = persistenceFor(auditor); + + persistence.writeEntry(auditEntry("change-1")); + + assertEquals(1, auditor.getAuditHistory().size()); + List> events = storedEvents(); + assertEquals(1, events.size(), "exactly one event per audit write"); + + JournalEvent event = events.get(0); + assertEquals(STREAM_ID, event.getStreamId()); + assertEquals(1L, event.getStreamSequence(), "the first event on a fresh stream is sequence 1"); + assertEquals(JournalEventType.CHANGE_STATE, event.getEventType()); + assertFalse(event.isAcknowledged(), "a freshly appended event is pending synchronization"); + assertEquals("change-1", event.getData().getChangeId(), "the audit entry travels as the event payload"); + } + + @Test + @DisplayName("journal enabled: successive states collapse to one audit document, the history staying in the journal") + void journalEnabledKeepsOneRecordPerChange() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + CouchbaseAuditPersistence persistence = persistenceFor(auditor); + + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.STARTED)); + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED)); + + List auditRecords = auditor.getAuditHistory(); + assertEquals(1, auditRecords.size(), "the audit document is the change's current state, not a ledger"); + assertEquals(AuditEntry.Status.APPLIED, auditRecords.get(0).getState(), "and it holds the latest state"); + + assertEquals(2, storedEvents().size(), "while every transition is kept as an event"); + } + + @Test + @DisplayName("journal disabled: successive states accumulate as separate audit documents") + void journalDisabledKeepsRecordPerStateTransition() { + CouchbaseAuditPersistence persistence = persistenceFor(auditor); + + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.STARTED)); + persistence.writeEntry(auditEntry("change-1", AuditEntry.Status.APPLIED)); + + assertEquals(2, auditor.getAuditHistory().size(), + "without the journal the audit collection is itself the history"); + } + + @Test + @DisplayName("journal enabled: a failing journal append rolls the audit entry back with it") + void journalFailureRollsBackAuditEntry() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + // Built while the stream is empty, so its sequencer is seeded at 1. Seeding the collision first would + // instead make forStream() seed at 2 and no conflict would happen. + CouchbaseAuditPersistence persistence = persistenceFor(auditor); + occupyStreamPosition(1L); + + assertThrows(RuntimeException.class, () -> persistence.writeEntry(auditEntry("change-1"))); + + assertTrue(auditor.getAuditHistory().isEmpty(), + "the audit entry must not survive a failed journal append"); + assertEquals(1, storedEvents().size(), "only the pre-existing event remains"); + } + + @Test + @DisplayName("journal enabled: a failing audit write rolls the journal event back with it") + void auditFailureRollsBackJournalEvent() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + // The audit write is a get-then-replace/insert on a single key, so it cannot be made to fail with a + // collision the way the journal can — the failure has to be injected. + CouchbaseAuditor failingAuditor = mock(CouchbaseAuditor.class); + doThrow(new IllegalStateException("audit write failed")) + .when(failingAuditor).contributeToTransaction(any(TransactionAttemptContext.class), any(AuditEntry.class)); + CouchbaseAuditPersistence persistence = persistenceFor(failingAuditor); + + assertThrows(RuntimeException.class, () -> persistence.writeEntry(auditEntry("change-1"))); + + assertTrue(storedEvents().isEmpty(), "the journal event must not survive a failed audit write"); + } + + @Test + @DisplayName("journal enabled: a failed write leaves no gap — its stream position is handed out again") + void failedWriteLeavesNoGapInTheStream() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + journalEventStore.initialize(true, SCOPE_NAME, JOURNAL_COLLECTION); + JournalEventSequencer sequencer = new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID); + + CouchbaseAuditor failingAuditor = mock(CouchbaseAuditor.class); + doThrow(new IllegalStateException("audit write failed")) + .when(failingAuditor).contributeToTransaction(any(TransactionAttemptContext.class), any(AuditEntry.class)); + CouchbaseAuditPersistence failing = persistenceFor(failingAuditor, sequencer); + assertThrows(RuntimeException.class, () -> failing.writeEntry(auditEntry("change-1"))); + + // Same sequencer: the aborted attempt must not have spent position 1. + persistenceFor(auditor, sequencer).writeEntry(auditEntry("change-1")); + + List> events = storedEvents(); + assertEquals(1, events.size()); + assertEquals(1L, events.get(0).getStreamSequence(), + "the stream must stay contiguous, so consumers can tell in-flight from lost"); + } + + // ----------------------------- helpers ----------------------------- + + private CouchbaseAuditPersistence persistenceFor(CouchbaseAuditor auditor) { + return persistenceFor(auditor, new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID)); + } + + private CouchbaseAuditPersistence persistenceFor(CouchbaseAuditor auditor, JournalEventSequencer sequencer) { + CouchbaseAuditPersistence persistence = new CouchbaseAuditPersistence( + new CommunityConfiguration(), auditor, journalEventStore, sequencer, txWrapper, + SCOPE_NAME, AUDIT_COLLECTION, JOURNAL_COLLECTION, true); + persistence.initialize(RunnerId.generate()); + return persistence; + } + + /** + * Takes a stream position directly, so the next append by a sequencer already seeded below it collides on + * the {@code journal::::} document key. + */ + private void occupyStreamPosition(long streamSequence) { + journalEventStore.initialize(true, SCOPE_NAME, JOURNAL_COLLECTION); + JournalEvent squatter = new JournalEvent<>( + "pre-existing-event", JournalEventType.CHANGE_STATE, JournalEvent.DEFAULT_VERSION, + STREAM_ID, streamSequence, Instant.now(), auditEntry("pre-existing-change"), false); + bucket.scope(SCOPE_NAME).collection(JOURNAL_COLLECTION) + .insert("journal::" + STREAM_ID + "::" + streamSequence, mapper.toDocument(squatter)); + } + + private List> storedEvents() { + if (!CouchbaseCollectionHelper.collectionExists(cluster, BUCKET_NAME, SCOPE_NAME, JOURNAL_COLLECTION)) { + return new ArrayList<>(); + } + List> events = new ArrayList<>(); + CouchbaseCollectionHelper.selectAllDocuments(cluster, BUCKET_NAME, SCOPE_NAME, JOURNAL_COLLECTION) + .forEach(document -> events.add(mapper.fromDocument(document))); + return events; + } + + private static AuditEntry auditEntry(String changeId) { + return auditEntry(changeId, AuditEntry.Status.APPLIED); + } + + private static AuditEntry auditEntry(String changeId, AuditEntry.Status status) { + return AuditEntryTestFactory.createTestAuditEntry( + changeId, status, AuditTxType.NON_TX, (Class) null); + } +} diff --git a/core/target-systems/flamingock-couchbase-externalsystem-api/build.gradle.kts b/core/target-systems/flamingock-couchbase-externalsystem-api/build.gradle.kts index 827c0ae61..255508d6c 100644 --- a/core/target-systems/flamingock-couchbase-externalsystem-api/build.gradle.kts +++ b/core/target-systems/flamingock-couchbase-externalsystem-api/build.gradle.kts @@ -1,6 +1,6 @@ val coreApiVersion: String by extra dependencies { - implementation("io.flamingock:flamingock-core-api:${coreApiVersion}") + api(project(":core:flamingock-core-commons")) //General compileOnly("com.couchbase.client:java-client:3.6.0") diff --git a/core/target-systems/flamingock-couchbase-externalsystem-api/src/main/java/io/flamingock/externalsystem/couchbase/api/CouchbaseExternalSystem.java b/core/target-systems/flamingock-couchbase-externalsystem-api/src/main/java/io/flamingock/externalsystem/couchbase/api/CouchbaseExternalSystem.java index 8635ac31f..59fa72063 100644 --- a/core/target-systems/flamingock-couchbase-externalsystem-api/src/main/java/io/flamingock/externalsystem/couchbase/api/CouchbaseExternalSystem.java +++ b/core/target-systems/flamingock-couchbase-externalsystem-api/src/main/java/io/flamingock/externalsystem/couchbase/api/CouchbaseExternalSystem.java @@ -17,9 +17,9 @@ import com.couchbase.client.java.Bucket; import com.couchbase.client.java.Cluster; -import io.flamingock.api.external.ExternalSystem; +import io.flamingock.internal.common.core.transaction.TransactionalExternalSystem; -public interface CouchbaseExternalSystem extends ExternalSystem { +public interface CouchbaseExternalSystem extends TransactionalExternalSystem { Cluster getCluster(); Bucket getBucket(); diff --git a/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/CouchbaseCollectionHelper.java b/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/CouchbaseCollectionHelper.java index 8744a27b0..e73af3bfe 100644 --- a/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/CouchbaseCollectionHelper.java +++ b/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/CouchbaseCollectionHelper.java @@ -43,6 +43,7 @@ public final class CouchbaseCollectionHelper { private final static String CREATE_PRIMARY_INDEX_TEMPLATE = "CREATE PRIMARY INDEX IF NOT EXISTS ON `%s`.`%s`.`%s`"; private final static String DROP_PRIMARY_INDEX_TEMPLATE = "DROP PRIMARY INDEX IF EXISTS ON `%s`.`%s`.`%s`"; private final static String DROP_INDEX_TEMPLATE = "DROP INDEX `%s` IF EXISTS ON `%s`.`%s`.`%s`"; + private final static String CREATE_INDEX_TEMPLATE = "CREATE INDEX `%s` IF NOT EXISTS ON `%s`.`%s`.`%s`(%s)"; private CouchbaseCollectionHelper() {} @@ -167,6 +168,22 @@ public static void createPrimaryIndexIfNotExists(Cluster cluster, String bucketN cluster.query(String.format(CREATE_PRIMARY_INDEX_TEMPLATE, bucketName, scopeName, collectionName)); } + /** + * Creates a named secondary index over the given field list (already comma-separated, e.g. + * {@code "streamId, streamSequence"}), optionally scoped to a {@code WHERE} predicate for a partial index. + * + * @param whereClause N1QL boolean expression (without the {@code WHERE} keyword), or {@code null}/blank + * for a non-partial index + */ + public static void createIndexIfNotExists(Cluster cluster, String bucketName, String scopeName, String collectionName, + String indexName, String fields, String whereClause) { + String query = String.format(CREATE_INDEX_TEMPLATE, indexName, bucketName, scopeName, collectionName, fields); + if (whereClause != null && !whereClause.trim().isEmpty()) { + query += " WHERE " + whereClause; + } + cluster.query(query); + } + public static void dropPrimaryIndexIfExists(Cluster cluster, String bucketName, String scopeName, String collectionName) { cluster.query(String.format(DROP_PRIMARY_INDEX_TEMPLATE, bucketName, scopeName, collectionName)); } diff --git a/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/CouchbaseJournalEventMapper.java b/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/CouchbaseJournalEventMapper.java new file mode 100644 index 000000000..edbefd304 --- /dev/null +++ b/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/CouchbaseJournalEventMapper.java @@ -0,0 +1,89 @@ +/* + * Copyright 2026 Flamingock (https://www.flamingock.io) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.flamingock.internal.common.couchbase; + +import com.couchbase.client.java.json.JsonObject; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.core.journal.JournalEventType; + +import java.time.Instant; + +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_ACKNOWLEDGED; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_DATA; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_EVENT_ID; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_EVENT_TYPE; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_EVENT_VERSION; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_OCCURRED_AT; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_STREAM_ID; +import static io.flamingock.internal.common.couchbase.journal.JournalEventFieldConstants.KEY_STREAM_SEQUENCE; + +/** + * Maps a {@link JournalEvent} carrying an {@link AuditEntry} payload to/from a Couchbase {@link JsonObject}. + *

+ * The {@code data} payload is nested as a sub-object, delegating to {@link CouchbaseAuditMapper} so the audit + * representation stays single-sourced. {@code occurredAt} is stored as epoch millis, matching how + * {@link CouchbaseUtils#addFieldToDocument} already represents dates in this module. + *

+ * Only {@link JournalEventType#CHANGE_STATE} events carry an {@link AuditEntry} payload today. Other event + * types (e.g. {@link JournalEventType#EXECUTION_STATE}) carry different payloads and are not yet implemented, + * so this mapper rejects them rather than silently mis-mapping their data as an audit entry. + */ +public class CouchbaseJournalEventMapper { + + /** The only event type whose {@code data} is an {@link AuditEntry} and is supported for now. */ + private static final JournalEventType SUPPORTED_EVENT_TYPE = JournalEventType.CHANGE_STATE; + + private final CouchbaseAuditMapper dataMapper = new CouchbaseAuditMapper(); + + public JsonObject toDocument(JournalEvent event) { + requireSupportedType(event.getEventType()); + JsonObject document = JsonObject.create(); + document.put(KEY_EVENT_ID, event.getEventId()); + document.put(KEY_EVENT_TYPE, event.getEventType().name()); + document.put(KEY_EVENT_VERSION, event.getEventVersion()); + document.put(KEY_STREAM_ID, event.getStreamId()); + document.put(KEY_STREAM_SEQUENCE, event.getStreamSequence()); + document.put(KEY_OCCURRED_AT, event.getOccurredAt().toEpochMilli()); + document.put(KEY_ACKNOWLEDGED, event.isAcknowledged()); + document.put(KEY_DATA, dataMapper.toDocument(event.getData())); + return document; + } + + public JournalEvent fromDocument(JsonObject document) { + JournalEventType eventType = JournalEventType.valueOf(document.getString(KEY_EVENT_TYPE)); + requireSupportedType(eventType); + AuditEntry data = dataMapper.fromDocument(document.getObject(KEY_DATA)); + Instant occurredAt = Instant.ofEpochMilli(document.getLong(KEY_OCCURRED_AT)); + return new JournalEvent<>( + document.getString(KEY_EVENT_ID), + eventType, + document.getInt(KEY_EVENT_VERSION), + document.getString(KEY_STREAM_ID), + document.getLong(KEY_STREAM_SEQUENCE), + occurredAt, + data, + document.getBoolean(KEY_ACKNOWLEDGED)); + } + + private static void requireSupportedType(JournalEventType eventType) { + if (eventType != SUPPORTED_EVENT_TYPE) { + throw new UnsupportedOperationException( + "CouchbaseJournalEventMapper only supports " + SUPPORTED_EVENT_TYPE + " events (AuditEntry payload); " + + "event type " + eventType + " is not yet implemented"); + } + } +} diff --git a/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/journal/JournalEventFieldConstants.java b/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/journal/JournalEventFieldConstants.java new file mode 100644 index 000000000..f4358f4b2 --- /dev/null +++ b/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/journal/JournalEventFieldConstants.java @@ -0,0 +1,37 @@ +/* + * Copyright 2026 Flamingock (https://www.flamingock.io) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.flamingock.internal.common.couchbase.journal; + +/** + * JSON field names for the local journal collection ({@code flamingockJournalEvents}). + *

+ * Declared locally because the shared {@code AuditEntryFieldConstants}/{@code CommunityPersistenceConstants} + * live in the external {@code flamingock-general-util} artifact and cannot be extended from this repository. + */ +public final class JournalEventFieldConstants { + + public static final String KEY_EVENT_ID = "eventId"; + public static final String KEY_EVENT_TYPE = "eventType"; + public static final String KEY_EVENT_VERSION = "eventVersion"; + public static final String KEY_STREAM_ID = "streamId"; + public static final String KEY_STREAM_SEQUENCE = "streamSequence"; + public static final String KEY_OCCURRED_AT = "occurredAt"; + public static final String KEY_DATA = "data"; + public static final String KEY_ACKNOWLEDGED = "acknowledged"; + + private JournalEventFieldConstants() { + } +} diff --git a/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/journal/JournalEventPersistenceConstants.java b/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/journal/JournalEventPersistenceConstants.java new file mode 100644 index 000000000..7c3aec456 --- /dev/null +++ b/utils/couchbase-util/src/main/java/io/flamingock/internal/common/couchbase/journal/JournalEventPersistenceConstants.java @@ -0,0 +1,28 @@ +/* + * Copyright 2026 Flamingock (https://www.flamingock.io) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.flamingock.internal.common.couchbase.journal; + +/** + * Default persistence name for the local event buffer, mirroring the (external, un-extendable) + * {@code CommunityPersistenceConstants} defaults used for the audit and lock collections. + */ +public final class JournalEventPersistenceConstants { + + public static final String DEFAULT_JOURNAL_STORE_NAME = "flamingockJournalEvents"; + + private JournalEventPersistenceConstants() { + } +}