chore: add more tests to saga orchestrations

core-domain/src/commonTest/kotlin/com/hailie/domain/saga/DomainInvariantsTest.kt

@@ -0,0 +1,52 @@
+package com.hailie.domain.model
+
+import com.hailie.domain.DeviceId
+import com.hailie.domain.DomainError
+import com.hailie.domain.EventOffset
+import com.hailie.domain.TimestampMs
+import com.hailie.domain.events.Disconnected
+import com.hailie.domain.events.EventsAcked
+import com.hailie.domain.events.SyncFailed
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class DomainInvariantsTest {
+    @Test
+    fun lastAckedExclusive_is_monotonic() {
+        val device = DeviceId("mono")
+        val t = TimestampMs(10_000)
+        val s0 = initialSyncAggregate(device)
+
+        // Apply ack to 50
+        val s1 = s0.applyEvent(EventsAcked(device, t, upTo = EventOffset(50)))
+        assertEquals(50, s1.lastAckedExclusive.value)
+
+        // Apply ack to 40 (lower) → must NOT decrease
+        val s2 = s1.applyEvent(EventsAcked(device, t, upTo = EventOffset(40)))
+        assertEquals(50, s2.lastAckedExclusive.value)
+
+        // Apply ack to 50 again (same) → no double-advance
+        val s3 = s2.applyEvent(EventsAcked(device, t, upTo = EventOffset(50)))
+        assertEquals(50, s3.lastAckedExclusive.value)
+
+        // Apply ack to 80 → advances
+        val s4 = s3.applyEvent(EventsAcked(device, t, upTo = EventOffset(80)))
+        assertEquals(80, s4.lastAckedExclusive.value)
+    }
+
+    @Test
+    fun permission_and_user_action_errors_are_surfacable_via_events() {
+        val device = DeviceId("perm")
+        val t = TimestampMs(11_000)
+        val s0 = initialSyncAggregate(device)
+
+        // Surface a permission required error via SyncFailed
+        val s1 = s0.applyEvent(SyncFailed(device, t, reason = DomainError.PermissionRequired("android.permission.BLUETOOTH_CONNECT")))
+        assertTrue(s1.lastError is DomainError.PermissionRequired)
+
+        // Surface a transport-level disconnect error via Disconnected
+        val s2 = s1.applyEvent(Disconnected(device, t, reason = com.hailie.domain.DisconnectReason.Timeout, gattCode = 133))
+        assertTrue(s2.lastError is DomainError.Transport)
+    }
+}

core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaHappyPathTest.kt

@@ -0,0 +1,108 @@
+package com.hailie.domain.saga
+
+import com.hailie.domain.BondStatus
+import com.hailie.domain.ConnectionStatus
+import com.hailie.domain.DeviceId
+import com.hailie.domain.EventCount
+import com.hailie.domain.EventOffset
+import com.hailie.domain.EventRange
+import com.hailie.domain.PageSize
+import com.hailie.domain.TimestampMs
+import com.hailie.domain.commands.Acknowledge
+import com.hailie.domain.commands.BondDevice
+import com.hailie.domain.commands.ConnectGatt
+import com.hailie.domain.commands.DeliverToApp
+import com.hailie.domain.commands.ReadEventCount
+import com.hailie.domain.commands.ReadEvents
+import com.hailie.domain.events.DeviceBonded
+import com.hailie.domain.events.DeviceConnected
+import com.hailie.domain.events.EventCountLoaded
+import com.hailie.domain.events.EventsAcked
+import com.hailie.domain.events.EventsDelivered
+import com.hailie.domain.events.EventsRead
+import com.hailie.domain.model.initialSyncAggregate
+import com.hailie.domain.policy.AdaptivePageSizingPolicy
+import com.hailie.domain.policy.RetryDecision
+import com.hailie.domain.policy.RetryPolicy
+import com.hailie.domain.policy.SimpleBreakerPolicy
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+private class RetryGiveUp : RetryPolicy {
+    override fun decide(
+        now: TimestampMs,
+        attemptsForOp: Int,
+        reason: com.hailie.domain.RetryReason,
+    ): RetryDecision {
+        return RetryDecision.GiveUp
+    }
+}
+
+private fun defaultSaga(): Saga =
+    DefaultSaga(
+        retryPolicy = RetryGiveUp(),
+        breakerForConnect = SimpleBreakerPolicy(failuresToOpen = 1, coolDownMs = 1),
+        breakerForRead = SimpleBreakerPolicy(),
+        breakerForDeliver = SimpleBreakerPolicy(),
+        breakerForAck = SimpleBreakerPolicy(),
+        pageSizingPolicy = AdaptivePageSizingPolicy(minPage = 20, maxPage = 100, growStep = 20, shrinkStep = 20),
+    )
+
+class SagaHappyPathTest {
+    @Test
+    fun decision_sequence_happy_path() {
+        val device = DeviceId("d1")
+        val t = TimestampMs(1_000)
+
+        // Start: not bonded → BondDevice
+        val s0 = initialSyncAggregate(deviceId = device, initialPageSize = 50)
+        val c0 = defaultSaga().decide(s0, lastEvent = null, now = t)
+        assertTrue(c0.first() is BondDevice)
+
+        // DeviceBonded → ConnectGatt
+        val s1 = s0.copy(bondStatus = BondStatus.Bonded)
+        val c1 = defaultSaga().decide(s1, lastEvent = DeviceBonded(device, t), now = t)
+        assertTrue(c1.first() is ConnectGatt)
+
+        // DeviceConnected → ReadEventCount
+        val s2 = s1.copy(connectionStatus = ConnectionStatus.Connected)
+        val c2 = defaultSaga().decide(s2, lastEvent = DeviceConnected(device, t), now = t)
+        assertTrue(c2.first() is ReadEventCount)
+
+        // EventCountLoaded(total=120) → ReadEvents(offset=0, count=50)
+        val s3 =
+            s2.copy(
+                totalOnDevice = EventCount(120),
+                pageSize = PageSize(50),
+                lastAckedExclusive = EventOffset(0),
+            )
+        val c3 = defaultSaga().decide(s3, lastEvent = EventCountLoaded(device, t, total = EventCount(120)), now = t)
+        val read1 = c3.first() as ReadEvents
+        assertEquals(0, read1.offset.value)
+        assertEquals(50, read1.count)
+
+        // EventsRead([0,50)) → DeliverToApp([0,50))
+        val range1 = EventRange(EventOffset(0), EventOffset(50))
+        val c4 = defaultSaga().decide(s3, lastEvent = EventsRead(device, t, range1), now = t)
+        val deliver1 = c4.first() as DeliverToApp
+        assertEquals(range1, deliver1.range)
+
+        // EventsDelivered([0,50)) → Acknowledge(upTo=50)
+        val c5 = defaultSaga().decide(s3, lastEvent = EventsDelivered(device, t, range1), now = t)
+        val ack1 = c5.first() as Acknowledge
+        assertEquals(50, ack1.upTo.value)
+
+        // After ack to 50 and total=120 → next ReadEvents(offset=50, count=70) due to tuning (+20 growStep)
+        val s4 = s3.copy(lastAckedExclusive = EventOffset(50))
+        val c6 = defaultSaga().decide(s4, lastEvent = EventsAcked(device, t, upTo = EventOffset(50)), now = t)
+        val read2 = c6.first() as ReadEvents
+        assertEquals(50, read2.offset.value)
+        assertEquals(70, read2.count) // tuned from 50 → 70 (Stable outcome, +20 growStep)
+
+        // If we caught up (ack upTo == total), next step is ReadEventCount (re-check growth)
+        val sDone = s3.copy(lastAckedExclusive = EventOffset(120))
+        val cDone = defaultSaga().decide(sDone, lastEvent = EventsAcked(device, t, upTo = EventOffset(120)), now = t)
+        assertTrue(cDone.first() is ReadEventCount)
+    }
+}

core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaResilienceTest.kt

@@ -0,0 +1,214 @@
+package com.hailie.domain.saga
+
+import com.hailie.domain.AttemptCounters
+import com.hailie.domain.AttemptKey
+import com.hailie.domain.BondStatus
+import com.hailie.domain.ConnectionStatus
+import com.hailie.domain.DeviceId
+import com.hailie.domain.DisconnectReason
+import com.hailie.domain.DomainError
+import com.hailie.domain.EventCount
+import com.hailie.domain.EventOffset
+import com.hailie.domain.PageSize
+import com.hailie.domain.TimestampMs
+import com.hailie.domain.commands.ConnectGatt
+import com.hailie.domain.commands.ReadEventCount
+import com.hailie.domain.commands.ReadEvents
+import com.hailie.domain.commands.ScheduleRetry
+import com.hailie.domain.events.DeviceConnected
+import com.hailie.domain.events.Disconnected
+import com.hailie.domain.events.EventCountLoaded
+import com.hailie.domain.policy.AdaptivePageSizingPolicy
+import com.hailie.domain.policy.BreakerPolicy
+import com.hailie.domain.policy.RetryDecision
+import com.hailie.domain.policy.RetryPolicy
+import com.hailie.domain.policy.SimpleBreakerPolicy
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+private class RetryScheduleFixed(private val delayMs: Long, private val cap: Int = Int.MAX_VALUE) : RetryPolicy {
+    override fun decide(
+        now: TimestampMs,
+        attemptsForOp: Int,
+        reason: com.hailie.domain.RetryReason,
+    ): RetryDecision {
+        return if (attemptsForOp >= cap) {
+            RetryDecision.GiveUp
+        } else {
+            RetryDecision.Schedule(TimestampMs(now.value + delayMs))
+        }
+    }
+}
+
+private class BreakerAlwaysAllowed : BreakerPolicy {
+    override fun isCallAllowed(
+        now: TimestampMs,
+        state: com.hailie.domain.BreakerState,
+    ): Boolean = true
+
+    override fun onSuccess(
+        now: TimestampMs,
+        state: com.hailie.domain.BreakerState,
+    ): com.hailie.domain.BreakerState = state
+
+    override fun onFailure(
+        now: TimestampMs,
+        state: com.hailie.domain.BreakerState,
+        error: DomainError,
+    ): com.hailie.domain.BreakerState = state
+}
+
+private class BreakerAllowedAfter(private val allowAtMs: Long) : BreakerPolicy {
+    override fun isCallAllowed(
+        now: TimestampMs,
+        state: com.hailie.domain.BreakerState,
+    ): Boolean {
+        return now.value >= allowAtMs
+    }
+
+    override fun onSuccess(
+        now: TimestampMs,
+        state: com.hailie.domain.BreakerState,
+    ): com.hailie.domain.BreakerState = state
+
+    override fun onFailure(
+        now: TimestampMs,
+        state: com.hailie.domain.BreakerState,
+        error: DomainError,
+    ): com.hailie.domain.BreakerState = state
+}
+
+private fun sagaWith(
+    retry: RetryPolicy,
+    connectBreaker: BreakerPolicy,
+): Saga =
+    DefaultSaga(
+        retryPolicy = retry,
+        breakerForConnect = connectBreaker,
+        breakerForRead = SimpleBreakerPolicy(),
+        breakerForDeliver = SimpleBreakerPolicy(),
+        breakerForAck = SimpleBreakerPolicy(),
+        pageSizingPolicy = AdaptivePageSizingPolicy(minPage = 20, maxPage = 100, growStep = 20, shrinkStep = 20),
+    )
+
+/**
+ * 1) Disconnect during paging → reconnect → resume at lastAckedExclusive
+ * 2) Breaker denies at t0 (schedules retry) and allows at t1 (connects)
+ * 3) Capped retries → stop scheduling when attempts reach cap
+ */
+class SagaResilienceTest {
+    @Test
+    fun disconnect_during_paging_reconnect_and_resume_from_high_water() {
+        val device = DeviceId("dev-paging")
+        val t0 = TimestampMs(10_000)
+
+        // We were in the middle of paging previously (ack at 50, total 120, pageSize 50), but got disconnected.
+        val s0 =
+            com.hailie.domain.model.SyncAggregate(
+                deviceId = device,
+                bondStatus = BondStatus.Bonded,
+                connectionStatus = ConnectionStatus.Disconnected,
+                lastAckedExclusive = EventOffset(50),
+                totalOnDevice = EventCount(120),
+                pageSize = PageSize(50),
+            )
+
+        // On Disconnected: breaker allows → ConnectGatt
+        val commands0 =
+            sagaWith(
+                retry = RetryScheduleFixed(delayMs = 1_000),
+                connectBreaker = BreakerAlwaysAllowed(),
+            ).decide(s0, lastEvent = Disconnected(device, t0, reason = DisconnectReason.Timeout, gattCode = null), now = t0)
+        assertTrue(commands0.first() is ConnectGatt)
+
+        // After connected: ReadEventCount
+        val s1 = s0.copy(connectionStatus = ConnectionStatus.Connected)
+        val commands1 =
+            sagaWith(
+                retry = RetryScheduleFixed(delayMs = 1_000),
+                connectBreaker = BreakerAlwaysAllowed(),
+            ).decide(s1, lastEvent = DeviceConnected(device, t0), now = t0)
+        assertTrue(commands1.first() is ReadEventCount)
+
+        // Count loaded (still behind): next ReadEvents must resume at offset=50 with current pageSize (50 here)
+        val s2 = s1.copy(totalOnDevice = EventCount(120))
+        val commands2 =
+            sagaWith(
+                retry = RetryScheduleFixed(delayMs = 1_000),
+                connectBreaker = BreakerAlwaysAllowed(),
+            ).decide(s2, lastEvent = EventCountLoaded(device, t0, total = EventCount(120)), now = t0)
+
+        val read = commands2.first() as ReadEvents
+        assertEquals(50, read.offset.value)
+        assertEquals(50, read.count)
+    }
+
+    @Test
+    fun breaker_half_open_probe_denied_then_allowed() {
+        val device = DeviceId("dev-halfopen")
+        val t0 = TimestampMs(5_000)
+        val t1 = TimestampMs(6_000)
+
+        val s =
+            com.hailie.domain.model.SyncAggregate(
+                deviceId = device,
+                bondStatus = BondStatus.Bonded,
+                connectionStatus = ConnectionStatus.Disconnected,
+                lastAckedExclusive = EventOffset(0),
+                totalOnDevice = EventCount(0),
+                pageSize = PageSize(50),
+            )
+
+        // At t0 breaker denies → should schedule retry (since we injected a scheduling retry policy)
+        val commands0 =
+            sagaWith(
+                retry = RetryScheduleFixed(delayMs = 500),
+                // denies at 5_000
+                connectBreaker = BreakerAllowedAfter(allowAtMs = 5_500),
+            ).decide(s, lastEvent = Disconnected(device, t0, reason = DisconnectReason.Timeout, gattCode = null), now = t0)
+        assertTrue(commands0.first() is ScheduleRetry)
+
+        // At t1 breaker allows → should ConnectGatt
+        val commands1 =
+            sagaWith(
+                retry = RetryScheduleFixed(delayMs = 500),
+                // allows at 6_000
+                connectBreaker = BreakerAllowedAfter(allowAtMs = 5_500),
+            ).decide(s, lastEvent = Disconnected(device, t1, reason = DisconnectReason.Timeout, gattCode = null), now = t1)
+        assertTrue(commands1.first() is ConnectGatt)
+    }
+
+    @Test
+    fun capped_retries_stop_scheduling_when_attempts_reach_cap() {
+        val device = DeviceId("dev-cap")
+        val now = TimestampMs(7_000)
+
+        // Already attempted once for ConnectGatt
+        val attempts =
+            AttemptCounters(
+                byKey = mapOf(AttemptKey("ConnectGatt") to 1),
+            )
+
+        val s =
+            com.hailie.domain.model.SyncAggregate(
+                deviceId = device,
+                bondStatus = BondStatus.Bonded,
+                connectionStatus = ConnectionStatus.Disconnected,
+                lastAckedExclusive = EventOffset(0),
+                totalOnDevice = EventCount(0),
+                pageSize = PageSize(50),
+                attempts = attempts,
+            )
+
+        // Retry policy cap = 1 → with attemptsForOp == 1, policy returns GiveUp → no commands should be emitted
+        val commands =
+            sagaWith(
+                retry = RetryScheduleFixed(delayMs = 500, cap = 1),
+                // keep denying path to invoke retry logic
+                connectBreaker = BreakerAllowedAfter(allowAtMs = 100_000),
+            ).decide(s, lastEvent = Disconnected(device, now, reason = DisconnectReason.Timeout, gattCode = null), now = now)
+
+        assertTrue(commands.isEmpty())
+    }
+}