correx/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt

import com.correx.core.events.events.StageCompletedEvent
import com.correx.core.events.events.StageFailedEvent
import com.correx.core.events.events.TransitionExecutedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.types.EventId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.events.types.TransitionId
import com.correx.core.sessions.DefaultSessionReducer
import com.correx.core.sessions.SessionProjector
import com.correx.core.sessions.SessionStatus
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
import com.correx.infrastructure.persistence.InMemoryEventStore
import com.correx.testing.fixtures.EventFixtures.newEvent
import kotlinx.coroutines.runBlocking
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test

class TransitionReplayIntegrationTest {

    private val sessionId = SessionId("session-1")

    @Test
    fun `workflow replay reconstructs COMPLETED session`(): Unit = runBlocking {

        val store = InMemoryEventStore()

        store.appendAll(
            listOf(
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-1"),
                    payload = WorkflowStartedEvent(sessionId, workflowId = "test", startStageId = StageId("stage-1"))
                ),
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-2"),
                    payload = TransitionExecutedEvent(
                        sessionId = sessionId,
                        from = StageId("draft"),
                        to = StageId("review"),
                        transitionId = TransitionId("transition-1")
                    )
                ),
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-3"),
                    payload = TransitionExecutedEvent(
                        sessionId = sessionId,
                        from = StageId("review"),
                        to = StageId("review"),
                        transitionId = TransitionId("transition-1")
                    )
                ),
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-4"),
                    payload = StageCompletedEvent(
                        sessionId = sessionId,
                        stageId = StageId("review"),
                        transitionId = TransitionId("transition-1")
                    )
                ),
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-5"),
                    payload = WorkflowCompletedEvent(sessionId, terminalStageId = StageId("stage-1"), totalStages = 1)
                )
            )
        )

        val replayer = DefaultEventReplayer(
            store = store,
            projection = SessionProjector(
                reducer = DefaultSessionReducer()
            )
        )

        val state = replayer.rebuild(sessionId)

        assertEquals(
            SessionStatus.COMPLETED,
            state.status
        )
    }

    @Test
    fun `workflow replay reconstructs FAILED session`(): Unit = runBlocking {

        val store = InMemoryEventStore()

        store.appendAll(
            listOf(
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-1"),
                    payload = WorkflowStartedEvent(sessionId, workflowId = "test", startStageId = StageId("stage-1"))
                ),
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-2"),
                    payload = TransitionExecutedEvent(
                        sessionId = sessionId,
                        from = StageId("draft"),
                        to = StageId("review"),
                        transitionId = TransitionId("transition-1")
                    )
                ),
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-3"),
                    payload = TransitionExecutedEvent(
                        sessionId = sessionId,
                        from = StageId("review"),
                        to = StageId("review"),
                        transitionId = TransitionId("transition-1")
                    )
                ),
                newEvent(
                    sessionId = sessionId,
                    eventId = EventId("event-4"),
                    payload = StageFailedEvent(
                        sessionId = sessionId,
                        stageId = StageId("review"),
                        transitionId = TransitionId("transition-1"),
                        reason = "validation failed"
                    )
                )
            )
        )

        val replayer = DefaultEventReplayer(
            store = store,
            projection = SessionProjector(
                reducer = DefaultSessionReducer()
            )
        )

        val state = replayer.rebuild(sessionId)

        assertEquals(
            SessionStatus.FAILED,
            state.status
        )
    }

    @Test
    fun `replay is deterministic for identical event stream`(): Unit = runBlocking {
        val events = listOf(
            newEvent(
                sessionId = sessionId,
                eventId = EventId("event-1"),
                payload = WorkflowStartedEvent(sessionId, workflowId = "test", startStageId = StageId("stage-1"))
            ),
            newEvent(
                sessionId = sessionId,
                eventId = EventId("event-2"),
                payload = TransitionExecutedEvent(
                    sessionId = sessionId,
                    from = StageId("draft"),
                    to = StageId("review"),
                    transitionId = TransitionId("transition-1")
                )
            ),
            newEvent(
                sessionId = sessionId,
                eventId = EventId("event-3"),
                payload = TransitionExecutedEvent(
                    sessionId = sessionId,
                    from = StageId("review"),
                    to = StageId("review"),
                    transitionId = TransitionId("transition-1")
                )
            ),
            newEvent(
                sessionId = sessionId,
                eventId = EventId("event-4"),
                payload = StageCompletedEvent(
                    sessionId = sessionId,
                    stageId = StageId("review"),
                    transitionId = TransitionId("transition-1")
                )
            )
        )

        val store1 = InMemoryEventStore()
        val store2 = InMemoryEventStore()

        store1.appendAll(events)
        store2.appendAll(events)

        val projection = SessionProjector(
            reducer = DefaultSessionReducer()
        )

        val replayer1 = DefaultEventReplayer(
            store = store1,
            projection = projection
        )

        val replayer2 = DefaultEventReplayer(
            store = store2,
            projection = projection
        )

        val state1 = replayer1.rebuild(sessionId)
        val state2 = replayer2.rebuild(sessionId)

        assertEquals(state1, state2)
    }
}