fix(scene): address bot review comments on concurrency, cancellation, validation, and race conditions

domain/src/main/kotlin/com/flagship/abox/manager/domain/DefaultSceneExecutor.kt

@@ -2,8 +2,12 @@ package com.flagship.abox.manager.domain
 
 import com.flagship.abox.manager.domain.infrared.SendVirtualRemoteAction
 import com.flagship.abox.manager.domain.infrared.VirtualRemoteId
-import kotlinx.coroutines.delay
 import java.util.concurrent.ConcurrentHashMap
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
 
 /**
  * Production implementation of [SceneExecutor].
@@ -29,32 +33,47 @@ class DefaultSceneExecutor(
             )
         }
         val startTime = System.currentTimeMillis()
-        val results = mutableListOf<SceneActionExecutionResult>()
-        var aborted = false
 
         return try {
-            for (action in scene.actions) {
-                if (aborted) {
-                    results.add(
-                        SceneActionExecutionResult(
-                            action = action,
-                            outcome = SceneActionOutcome.SKIPPED,
-                            error = null,
-                        ),
-                    )
-                    continue
+            val results: List<SceneActionExecutionResult> = if (scene.concurrencyMode == ConcurrencyMode.PARALLEL) {
+                coroutineScope {
+                    scene.actions.map { action ->
+                        async {
+                            val actionStart = System.currentTimeMillis()
+                            val execResult = executeAction(scene.profileId, action)
+                            val duration = System.currentTimeMillis() - actionStart
+                            execResult.copy(executionDurationMillis = duration)
+                        }
+                    }.awaitAll()
                 }
+            } else {
+                val sequentialResults = mutableListOf<SceneActionExecutionResult>()
+                var aborted = false
+                for (action in scene.actions) {
+                    if (aborted) {
+                        sequentialResults.add(
+                            SceneActionExecutionResult(
+                                action = action,
+                                outcome = SceneActionOutcome.SKIPPED,
+                                error = null,
+                            ),
+                        )
+                        continue
+                    }
 
-                val actionStart = System.currentTimeMillis()
-                val execResult = executeAction(scene.profileId, action)
-                val duration = System.currentTimeMillis() - actionStart
-                val enrichedResult = execResult.copy(executionDurationMillis = duration)
-                results.add(enrichedResult)
+                    val actionStart = System.currentTimeMillis()
+                    val execResult = executeAction(scene.profileId, action)
+                    val duration = System.currentTimeMillis() - actionStart
+                    val enrichedResult = execResult.copy(executionDurationMillis = duration)
+                    sequentialResults.add(enrichedResult)
 
-                if (enrichedResult.outcome == SceneActionOutcome.FAILURE && scene.strategy == ExecutionStrategy.ABORT_ON_ERROR) {
-                    aborted = true
+                    if (enrichedResult.outcome == SceneActionOutcome.FAILURE && scene.strategy == ExecutionStrategy.ABORT_ON_ERROR) {
+                        aborted = true
+                    }
                 }
+                sequentialResults
             }
+
             val totalDuration = System.currentTimeMillis() - startTime
             DomainResult.Success(
                 SceneExecutionResult(
@@ -63,6 +82,8 @@ class DefaultSceneExecutor(
                     totalDurationMillis = totalDuration,
                 ),
             )
+        } catch (e: CancellationException) {
+            throw e
         } finally {
             runningScenes.remove(scene.id)
         }
@@ -72,66 +93,70 @@ class DefaultSceneExecutor(
         profileId: ProviderProfileId,
         action: SceneAction,
     ): SceneActionExecutionResult {
-        return when (action) {
-            is SceneAction.SocketPower -> {
-                val res = socketGateway.setPower(
-                    profileId = profileId,
-                    deviceName = action.deviceName,
-                    powerState = action.powerState,
-                )
-                toExecutionResult(action, res)
-            }
-
-            is SceneAction.InfraredKey -> {
-                val res = infraredGateway.sendKey(
-                    profileId = profileId,
-                    deviceName = action.deviceName,
-                    key = action.key,
-                )
-                toExecutionResult(action, res)
-            }
-
-            is SceneAction.VirtualRemoteButton -> {
-                if (sendVirtualRemoteAction != null) {
-                    val res = sendVirtualRemoteAction.sendButton(
+        return try {
+            when (action) {
+                is SceneAction.SocketPower -> {
+                    val res = socketGateway.setPower(
                         profileId = profileId,
-                        remoteId = VirtualRemoteId(action.remoteId),
-                        buttonId = action.buttonId,
+                        deviceName = action.deviceName,
+                        powerState = action.powerState,
                     )
                     toExecutionResult(action, res)
-                } else {
-                    SceneActionExecutionResult(
-                        action = action,
-                        outcome = SceneActionOutcome.SUCCESS,
-                        error = null,
-                    )
                 }
-            }
 
-            is SceneAction.Delay -> {
-                try {
+                is SceneAction.InfraredKey -> {
+                    val res = infraredGateway.sendKey(
+                        profileId = profileId,
+                        deviceName = action.deviceName,
+                        key = action.key,
+                    )
+                    toExecutionResult(action, res)
+                }
+
+                is SceneAction.VirtualRemoteButton -> {
+                    if (sendVirtualRemoteAction != null) {
+                        val res = sendVirtualRemoteAction.sendButton(
+                            profileId = profileId,
+                            remoteId = VirtualRemoteId(action.remoteId),
+                            buttonId = action.buttonId,
+                        )
+                        toExecutionResult(action, res)
+                    } else {
+                        SceneActionExecutionResult(
+                            action = action,
+                            outcome = SceneActionOutcome.FAILURE,
+                            error = DomainError.OperationFailed(
+                                detail = "Virtual remote action executor is not available",
+                            ),
+                        )
+                    }
+                }
+
+                is SceneAction.Delay -> {
                     delayDurationProvider(action.durationSeconds * 1000L)
                     SceneActionExecutionResult(
                         action = action,
                         outcome = SceneActionOutcome.SUCCESS,
                         error = null,
                     )
-                } catch (e: Exception) {
+                }
+
+                is SceneAction.Notification -> {
                     SceneActionExecutionResult(
                         action = action,
-                        outcome = SceneActionOutcome.FAILURE,
-                        error = DomainError.OperationFailed(detail = e.message),
+                        outcome = SceneActionOutcome.SUCCESS,
+                        error = null,
                     )
                 }
             }
-
-            is SceneAction.Notification -> {
-                SceneActionExecutionResult(
-                    action = action,
-                    outcome = SceneActionOutcome.SUCCESS,
-                    error = null,
-                )
-            }
+        } catch (e: CancellationException) {
+            throw e
+        } catch (e: Exception) {
+            SceneActionExecutionResult(
+                action = action,
+                outcome = SceneActionOutcome.FAILURE,
+                error = DomainError.OperationFailed(detail = e.message),
+            )
         }
     }
 

domain/src/main/kotlin/com/flagship/abox/manager/domain/InMemorySceneRepository.kt

@@ -3,6 +3,8 @@ package com.flagship.abox.manager.domain
 import java.util.concurrent.ConcurrentHashMap
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 
 /**
  * Provider-isolated in-memory [SceneRepository] implementation that supports
@@ -10,6 +12,10 @@ import kotlinx.coroutines.flow.MutableStateFlow
  */
 class InMemorySceneRepository : SceneRepository {
     private val flows = ConcurrentHashMap<ProviderProfileId, MutableStateFlow<DomainResult<List<Scene>>>>()
+    private val mutexes = ConcurrentHashMap<ProviderProfileId, Mutex>()
+
+    private fun mutexFor(profileId: ProviderProfileId): Mutex =
+        mutexes.getOrPut(profileId) { Mutex() }
 
     override fun observeScenes(profileId: ProviderProfileId): Flow<DomainResult<List<Scene>>> =
         flowFor(profileId)
@@ -25,10 +31,13 @@ class InMemorySceneRepository : SceneRepository {
     override suspend fun createScene(
         profileId: ProviderProfileId,
         scene: Scene,
-    ): DomainResult<Scene> {
+    ): DomainResult<Scene> = mutexFor(profileId).withLock {
         require(scene.profileId == profileId) {
             "Scene profileId must match the supplied profileId"
         }
+        require(scene.actions.all { it.profileId == profileId }) {
+            "All scene actions must belong to the supplied profileId"
+        }
         val scenes = currentScenes(profileId).orEmpty()
         if (scenes.any { it.id == scene.id }) {
             return DomainResult.Failure(
@@ -39,39 +48,45 @@ class InMemorySceneRepository : SceneRepository {
         }
         val updated = scenes + scene
         publish(profileId, updated)
-        return DomainResult.Success(scene)
+        DomainResult.Success(scene)
     }
 
     override suspend fun updateScene(
         profileId: ProviderProfileId,
         scene: Scene,
-    ): DomainResult<Scene> {
+    ): DomainResult<Scene> = mutexFor(profileId).withLock {
         require(scene.profileId == profileId) {
             "Scene profileId must match the supplied profileId"
         }
+        require(scene.actions.all { it.profileId == profileId }) {
+            "All scene actions must belong to the supplied profileId"
+        }
         val scenes = currentScenes(profileId).orEmpty()
         val index = scenes.indexOfFirst { it.id == scene.id }
         if (index < 0) return notFound(scene.id)
         val updated = scenes.toMutableList().apply { set(index, scene) }
         publish(profileId, updated)
-        return DomainResult.Success(scene)
+        DomainResult.Success(scene)
     }
 
     override suspend fun deleteScene(
         profileId: ProviderProfileId,
         sceneId: SceneId,
-    ): DomainResult<Unit> {
+    ): DomainResult<Unit> = mutexFor(profileId).withLock {
         val scenes = currentScenes(profileId).orEmpty()
         val updated = scenes.filterNot { it.id == sceneId }
         if (updated.size == scenes.size) return notFound(sceneId)
         publish(profileId, updated)
-        return DomainResult.Success(Unit)
+        DomainResult.Success(Unit)
     }
 
     fun seedDefaults(profileId: ProviderProfileId, scenes: List<Scene>) {
         require(scenes.all { it.profileId == profileId }) {
             "All scenes must belong to the supplied profileId"
         }
+        require(scenes.all { s -> s.actions.all { it.profileId == profileId } }) {
+            "All scene actions must belong to the supplied profileId"
+        }
         publish(profileId, scenes)
     }
 

domain/src/test/kotlin/com/flagship/abox/manager/domain/DefaultSceneExecutorTest.kt

@@ -221,5 +221,92 @@ class DefaultSceneExecutorTest {
         assertEquals(SceneActionOutcome.SUCCESS, executionResult.results[0].outcome)
         assertEquals(5000L, delayCalled)
     }
+
+    @Test
+    fun executesActionsConcurrentlyWhenParallelMode() = runTest {
+        val socketGateway = TestSocketGateway()
+        val infraredGateway = TestInfraredGateway()
+        val executor = DefaultSceneExecutor(socketGateway, infraredGateway)
+
+        val action1 = SceneAction.SocketPower(
+            id = SceneActionId("a1"),
+            profileId = profileId,
+            deviceName = DeviceName("socket-1"),
+            powerState = SocketPowerState.ON,
+        )
+        val action2 = SceneAction.InfraredKey(
+            id = SceneActionId("a2"),
+            profileId = profileId,
+            deviceName = DeviceName("ir-1"),
+            key = "power",
+        )
+        val scene = Scene(
+            id = SceneId("s1"),
+            profileId = profileId,
+            name = "Parallel Scene",
+            concurrencyMode = ConcurrencyMode.PARALLEL,
+            actions = listOf(action1, action2),
+        )
+
+        val result = executor.execute(scene)
+        assertTrue(result is DomainResult.Success)
+        val executionResult = (result as DomainResult.Success).value
+        assertEquals(2, executionResult.results.size)
+        assertEquals(SceneActionOutcome.SUCCESS, executionResult.results[0].outcome)
+        assertEquals(SceneActionOutcome.SUCCESS, executionResult.results[1].outcome)
+        assertEquals(1, socketGateway.calls.size)
+        assertEquals(1, infraredGateway.calls.size)
+    }
+
+    @Test
+    fun failsVirtualRemoteButtonActionWhenSenderNotConfigured() = runTest {
+        val executor = DefaultSceneExecutor(
+            socketGateway = TestSocketGateway(),
+            infraredGateway = TestInfraredGateway(),
+            sendVirtualRemoteAction = null,
+        )
+        val action = SceneAction.VirtualRemoteButton(
+            id = SceneActionId("vr1"),
+            profileId = profileId,
+            remoteId = "tv-remote",
+            remoteDisplayName = "TV",
+            buttonId = "power",
+            buttonDisplayName = "Power",
+        )
+        val scene = Scene(
+            id = SceneId("s-vr"),
+            profileId = profileId,
+            name = "VR Scene",
+            actions = listOf(action),
+        )
+
+        val result = executor.execute(scene)
+        assertTrue(result is DomainResult.Success)
+        val executionResult = (result as DomainResult.Success).value
+        assertEquals(1, executionResult.results.size)
+        assertEquals(SceneActionOutcome.FAILURE, executionResult.results[0].outcome)
+        assertTrue(executionResult.results[0].error is DomainError.OperationFailed)
+    }
+
+    @Test(expected = kotlinx.coroutines.CancellationException::class)
+    fun rethrowsCancellationExceptionWhenDelayCancelled() = runTest {
+        val executor = DefaultSceneExecutor(
+            socketGateway = TestSocketGateway(),
+            infraredGateway = TestInfraredGateway(),
+            delayDurationProvider = { throw kotlinx.coroutines.CancellationException("Job cancelled") },
+        )
+        val action = SceneAction.Delay(
+            id = SceneActionId("d1"),
+            profileId = profileId,
+            durationSeconds = 10,
+        )
+        val scene = Scene(
+            id = SceneId("s-cancelled"),
+            profileId = profileId,
+            name = "Cancelled Scene",
+            actions = listOf(action),
+        )
+        executor.execute(scene)
+    }
 }
 

domain/src/test/kotlin/com/flagship/abox/manager/domain/InMemorySceneRepositoryTest.kt

@@ -1,6 +1,9 @@
 package com.flagship.abox.manager.domain
 
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.test.runTest
 import org.junit.Assert.assertEquals
@@ -83,4 +86,43 @@ class InMemorySceneRepositoryTest {
         val getAfterDelete = repository.getScene(aboxProfile, sceneId)
         assertNull((getAfterDelete as DomainResult.Success).value)
     }
+
+    @Test(expected = IllegalArgumentException::class)
+    fun rejectsSceneWithMismatchedActionProfileId() = runTest {
+        val repository = InMemorySceneRepository()
+        val crossAction = SceneAction.SocketPower(
+            id = SceneActionId("a-cross"),
+            profileId = ProviderProfileId("other-profile"),
+            deviceName = DeviceName("socket-1"),
+            powerState = SocketPowerState.ON,
+        )
+        val scene = Scene(
+            id = SceneId("abox:admin:scene-invalid"),
+            profileId = aboxProfile,
+            name = "Invalid Scene",
+            actions = listOf(crossAction),
+        )
+        repository.createScene(aboxProfile, scene)
+    }
+
+    @Test
+    fun concurrentCreationsAreThreadSafe() = runTest {
+        val repository = InMemorySceneRepository()
+        val count = 20
+        coroutineScope {
+            (1..count).map { idx ->
+                async {
+                    val scene = Scene(
+                        id = SceneId("abox:admin:scene-$idx"),
+                        profileId = aboxProfile,
+                        name = "Scene $idx",
+                        actions = emptyList(),
+                    )
+                    repository.createScene(aboxProfile, scene)
+                }
+            }.awaitAll()
+        }
+        val scenes = (repository.observeScenes(aboxProfile).first() as DomainResult.Success).value
+        assertEquals(count, scenes.size)
+    }
 }