feat: stream scene execution progress

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

@@ -9,6 +9,8 @@ import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
 
 /**
  * Production implementation of [SceneExecutor].
@@ -24,6 +26,50 @@ class DefaultSceneExecutor(
 ) : SceneExecutor {
     private val runningScenes = ConcurrentHashMap<SceneId, Boolean>()
 
+    override fun execute(
+        scene: Scene,
+        executionId: SceneExecutionId,
+        source: SceneExecutionSource,
+    ): Flow<SceneExecutionUpdate> = flow {
+        val alreadyRunning = runningScenes.putIfAbsent(scene.id, true) != null
+        if (alreadyRunning) {
+            emit(SceneExecutionUpdate.Rejected(DomainError.OperationFailed(detail = "Scene is already running")))
+            return@flow
+        }
+        val startTime = System.currentTimeMillis()
+        try {
+            emit(SceneExecutionUpdate.Started(executionId, scene))
+            val results = mutableListOf<SceneActionExecutionResult>()
+            var aborted = false
+            for (action in scene.actions) {
+                if (aborted) {
+                    val skipped = SceneActionExecutionResult(action, SceneActionOutcome.SKIPPED, null)
+                    results += skipped
+                    emit(SceneExecutionUpdate.StepFinished(skipped))
+                    continue
+                }
+                emit(SceneExecutionUpdate.StepPending(action))
+                emit(SceneExecutionUpdate.StepRunning(action))
+                val started = System.currentTimeMillis()
+                val result = executeAction(scene.profileId, action).copy(
+                    executionDurationMillis = System.currentTimeMillis() - started,
+                )
+                results += result
+                emit(SceneExecutionUpdate.StepFinished(result))
+                if (result.outcome == SceneActionOutcome.FAILURE && scene.strategy == ExecutionStrategy.ABORT_ON_ERROR) {
+                    aborted = true
+                }
+            }
+            emit(
+                SceneExecutionUpdate.Completed(
+                    SceneExecutionResult(scene, results, System.currentTimeMillis() - startTime),
+                ),
+            )
+        } finally {
+            runningScenes.remove(scene.id)
+        }
+    }
+
     override suspend fun execute(scene: Scene): DomainResult<SceneExecutionResult> {
         val alreadyRunning = runningScenes.putIfAbsent(scene.id, true) != null
         if (alreadyRunning) {