diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepository.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepository.kt index d2d6862f..a294e52e 100644 --- a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepository.kt +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepository.kt @@ -12,6 +12,7 @@ import com.correx.core.events.stores.EventStore import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job @@ -55,11 +56,27 @@ class LiveArtifactRepository( private fun ensureSubscribed(sessionId: SessionId) { subscriptions.computeIfAbsent(sessionId) { - scope.launch { + // Rebuild from persisted history so a cold (post-restart) or first mid-session consumer + // sees prior artifact state immediately, not empty (invariant #1). The reducer's phase + // transitions are strictly ordered, so application must stay single-threaded and in + // sequence order — hence the gate below rather than a concurrent replay. + val rebuildWatermark = CompletableDeferred() + val job = scope.launch { + // Start collecting immediately so events appended during the rebuild are captured (flow + // back-pressure buffers them), but hold each until the rebuild's high-water mark is + // known, then drop anything already folded by the rebuild (sequence <= watermark). + // Live delivery is in append order, so post-watermark events fold in order too. eventStore.subscribe(sessionId).collect { event -> - processEvent(sessionId, event) + if (event.sequence > rebuildWatermark.await()) processEvent(sessionId, event) } } + var watermark = 0L + eventStore.read(sessionId).forEach { event -> + watermark = maxOf(watermark, event.sequence) + processEvent(sessionId, event) + } + rebuildWatermark.complete(watermark) + job } } diff --git a/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepositoryTest.kt b/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepositoryTest.kt new file mode 100644 index 00000000..a78ab9b0 --- /dev/null +++ b/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepositoryTest.kt @@ -0,0 +1,41 @@ +package com.correx.infrastructure.persistence.artifact + +import com.correx.core.artifacts.DefaultArtifactReducer +import com.correx.core.events.events.ArtifactCreatedEvent +import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.ArtifactValidatingEvent +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.ArtifactLifecyclePhase +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.infrastructure.persistence.InMemoryEventStore +import com.correx.testing.fixtures.EventFixtures +import kotlinx.coroutines.runBlocking +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test + +class LiveArtifactRepositoryTest { + + @Test + fun `cold access rebuilds artifact state from the persisted log, in order`(): Unit = runBlocking { + val store = InMemoryEventStore() + val session = SessionId("s1") + val artifact = ArtifactId("a1") + val stage = StageId("stage1") + + // Persist the full lifecycle BEFORE any repository/consumer exists — the pre-fix repo + // subscribed with replay=0 and never read() this history, so a cold consumer saw null. + store.append(EventFixtures.newEvent(EventId("e1"), session, ArtifactCreatedEvent(artifact, session, stage, 1))) + store.append(EventFixtures.newEvent(EventId("e2"), session, ArtifactValidatingEvent(artifact, session, stage))) + store.append(EventFixtures.newEvent(EventId("e3"), session, ArtifactValidatedEvent(artifact, session, stage))) + + // Fresh repository = cold consumer (mimics a restart / first mid-session access). + val repo = LiveArtifactRepository(store, DefaultArtifactReducer()) + + // Rebuilt, and folded in sequence order — VALIDATED only reachable via CREATED→VALIDATING→VALIDATED. + assertEquals(ArtifactLifecyclePhase.VALIDATED, repo.getById(session, artifact)?.phase) + assertNull(repo.getById(session, ArtifactId("does-not-exist"))) + } +}