fix(artifacts): rebuild LiveArtifactRepository from event log on cold access

2026-07-12 13:05:52 +04:00
parent b93729008d
commit 8c45650e21
2 changed files with 60 additions and 2 deletions
@@ -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<Long>()
+            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 ->
+                    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
        }
    }

@@ -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")))
+    }
+}