Flagship/abox-mgr
5
2
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(scene): isolate scheduled automation execution

Browse Source
This commit is contained in:
KremeCN
2026-08-28 02:36:11 +08:00
parent 076a1daf5d
commit fe285322ef
6 changed files with 86 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
android/feature/automation/src/main/kotlin/com/flagship/abox/manager/automation/runtime/AutomationBootstrapper.kt
+6 -5
View File
@@ -11,6 +11,7 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.launch
@@ -30,20 +31,20 @@ class AutomationBootstrapper(
combine(
profileStore.observeActiveProfile(),
sessionCoordinator.observeActiveSession(),
isForeground,
) { active, session, foreground -> Triple(active, session, foreground) }
.collectLatest { (active, session, foreground) ->
) { active, session -> active to session }
.collectLatest { (active, session) ->
previousProfileId?.takeIf { it != active?.id }?.let { scheduler.cancel(it) }
previousProfileId = active?.id
if (active != null && session?.profileId == active.id) {
scheduler.reconcile(active.id)
while (foreground) {
while (true) {
isForeground.first { it }
try {
engine.evaluate(active.id)
} catch (cancelled: CancellationException) {
throw cancelled
} catch (_: Exception) {}
delay(60_000L)
if (isForeground.value) delay(60_000L)
}
} else if (active != null) {
scheduler.cancel(active.id)
android/feature/automation/src/main/kotlin/com/flagship/abox/manager/automation/runtime/AutomationEngine.kt
+28 -5
View File
@@ -28,9 +28,18 @@ import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
interface AutomationEngine {
suspend fun evaluate(profileId: ProviderProfileId): DomainResult<Int>
suspend fun evaluate(
profileId: ProviderProfileId,
scheduledOccurrence: ScheduledAutomationOccurrence? = null,
): DomainResult<Int>
}
data class ScheduledAutomationOccurrence(
val sceneId: String,
val triggerId: String,
val occurrenceKey: String,
)
class DefaultAutomationEngine(
private val sceneRepository: SceneRepository,
private val runtimeRepository: AutomationRuntimeRepository,
@@ -43,11 +52,17 @@ class DefaultAutomationEngine(
) : AutomationEngine {
private val evaluationMutex = Mutex()
override suspend fun evaluate(profileId: ProviderProfileId): DomainResult<Int> = evaluationMutex.withLock {
evaluateLocked(profileId)
override suspend fun evaluate(
profileId: ProviderProfileId,
scheduledOccurrence: ScheduledAutomationOccurrence?,
): DomainResult<Int> = evaluationMutex.withLock {
evaluateLocked(profileId, scheduledOccurrence)
}
private suspend fun evaluateLocked(profileId: ProviderProfileId): DomainResult<Int> {
private suspend fun evaluateLocked(
profileId: ProviderProfileId,
scheduledOccurrence: ScheduledAutomationOccurrence?,
): DomainResult<Int> {
val scenes = when (val result = sceneRepository.observeScenes(profileId).first()) {
is DomainResult.Failure -> return result
is DomainResult.Success -> result.value.filter { it.enabled && it.type == AutomationType.AUTOMATIC }
@@ -70,7 +85,15 @@ class DefaultAutomationEngine(
is DomainResult.Success -> state.value
}
}
val evaluation = triggerEvaluator.evaluate(trigger, snapshot, now, wasMatched)
val evaluation = if (
trigger is SceneTrigger.Time &&
scheduledOccurrence?.sceneId == scene.id.value &&
scheduledOccurrence.triggerId == trigger.id.value
) {
TriggerEvaluation.Matched(scheduledOccurrence.occurrenceKey)
} else {
triggerEvaluator.evaluate(trigger, snapshot, now, wasMatched)
}
val isMatched = evaluation is TriggerEvaluation.Matched
if (trigger is SceneTrigger.Time && evaluation is TriggerEvaluation.Matched) {
newlyMatched = true
android/feature/automation/src/main/kotlin/com/flagship/abox/manager/automation/runtime/AutomationPollWorker.kt
+7 -2
View File
@@ -28,12 +28,17 @@ class AutomationPollWorker @AssistedInject constructor(
if (profileStore.activeProfile()?.id != profileId) return Result.success()
sessionCoordinator.bootstrap()
return when (val validity = sessionCoordinator.validateActiveSession()) {
return when (val validity = sessionCoordinator.validateSession(profileId)) {
is DomainResult.Failure -> Result.retry()
is DomainResult.Success -> if (validity.value != SessionValidity.VALID) {
Result.retry()
} else {
when (automationEngine.evaluate(profileId)) {
val scheduledOccurrence = inputData.getString(WorkManagerAutomationScheduleStore.KEY_OCCURRENCE_KEY)?.let { key ->
val sceneId = inputData.getString(WorkManagerAutomationScheduleStore.KEY_SCENE_ID)
val triggerId = inputData.getString(WorkManagerAutomationScheduleStore.KEY_TRIGGER_ID)
if (sceneId != null && triggerId != null) ScheduledAutomationOccurrence(sceneId, triggerId, key) else null
}
when (automationEngine.evaluate(profileId, scheduledOccurrence)) {
is DomainResult.Success -> {
automationScheduler.reconcile(profileId)
Result.success()
android/feature/automation/src/main/kotlin/com/flagship/abox/manager/automation/runtime/AutomationScheduler.kt
+1 -1
View File
@@ -36,7 +36,7 @@ class ProfileAutomationScheduler(
override suspend fun reconcile(profileId: ProviderProfileId): DomainResult<Unit> {
val active = profileStore.activeProfile()
if (active?.id != profileId) return DomainResult.Failure(DomainError.InvalidRequest(detail = "Provider profile is not active"))
return when (val validity = sessionCoordinator.validateActiveSession()) {
return when (val validity = sessionCoordinator.validateSession(profileId)) {
is DomainResult.Success -> if (validity.value == SessionValidity.VALID) {
when (val result = sceneRepository.observeScenes(profileId).first()) {
is DomainResult.Failure -> result
android/feature/automation/src/main/kotlin/com/flagship/abox/manager/automation/runtime/WorkManagerAutomationScheduleStore.kt
+26 -6
View File
@@ -39,16 +39,25 @@ class WorkManagerAutomationScheduleStore(context: Context) : AutomationScheduleS
private fun scheduleNextTimeTrigger(profileId: ProviderProfileId, scenes: List<Scene>) {
val now = LocalDateTime.now()
val candidates = scenes.flatMap { scene -> scene.triggers.filterIsInstance<SceneTrigger.Time>() }
.mapNotNull { trigger -> nextOccurrence(trigger, now) }
val next = candidates.minOrNull() ?: run {
val next = scenes.flatMap { scene ->
scene.triggers.filterIsInstance<SceneTrigger.Time>().mapNotNull { trigger ->
nextOccurrence(trigger, now)?.let { occurrence -> ScheduledWorkOccurrence(scene, trigger, occurrence) }
}
}.minByOrNull { it.scheduledAt } ?: run {
workManager.cancelUniqueWork(timeWorkName(profileId))
return
}
val delay = Duration.between(now.atZone(ZoneId.systemDefault()).toInstant(), next.atZone(ZoneId.systemDefault()).toInstant())
val delay = Duration.between(now.atZone(ZoneId.systemDefault()).toInstant(), next.scheduledAt.atZone(ZoneId.systemDefault()).toInstant())
val request = OneTimeWorkRequestBuilder<AutomationPollWorker>()
.setInitialDelay(delay.toMillis().coerceAtLeast(0), TimeUnit.MILLISECONDS)
.setInputData(androidx.work.Data.Builder().putString(KEY_PROFILE_ID, profileId.value).build())
.setInputData(
androidx.work.Data.Builder()
.putString(KEY_PROFILE_ID, profileId.value)
.putString(KEY_SCENE_ID, next.scene.id.value)
.putString(KEY_TRIGGER_ID, next.trigger.id.value)
.putString(KEY_OCCURRENCE_KEY, "${next.scheduledAt.toLocalDate()}:${next.trigger.timeOfDay}:${next.trigger.id.value}")
.build(),
)
.build()
workManager.enqueueUniqueWork(timeWorkName(profileId), androidx.work.ExistingWorkPolicy.REPLACE, request)
}
@@ -64,5 +73,16 @@ class WorkManagerAutomationScheduleStore(context: Context) : AutomationScheduleS
return null
}
companion object { const val KEY_PROFILE_ID = "profile_id" }
private data class ScheduledWorkOccurrence(
val scene: Scene,
val trigger: SceneTrigger.Time,
val scheduledAt: LocalDateTime,
)
companion object {
const val KEY_PROFILE_ID = "profile_id"
const val KEY_SCENE_ID = "scene_id"
const val KEY_TRIGGER_ID = "trigger_id"
const val KEY_OCCURRENCE_KEY = "occurrence_key"
}
}
domain/src/main/kotlin/com/flagship/abox/manager/domain/SessionCoordinator.kt
+18 -1
View File
@@ -3,6 +3,7 @@ package com.flagship.abox.manager.domain
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flatMapLatest
import kotlinx.coroutines.flow.flowOf
@@ -173,7 +174,23 @@ class SessionCoordinator(
suspend fun validateActiveSession(): DomainResult<SessionValidity> = storeOperation {
val profile = profileStore.activeProfile()
?: return@storeOperation DomainResult.Success(SessionValidity.INVALID)
profile.gateway().validateSession(profile.id)
validateSession(profile.id)
}
suspend fun validateSession(profileId: ProviderProfileId): DomainResult<SessionValidity> = storeOperation {
val profile = profileStore.activeProfile()
if (profile?.id != profileId) return@storeOperation DomainResult.Success(SessionValidity.INVALID)
val session = profile.gateway().observeSession(profileId).first()
if (session?.profileId != profileId) return@storeOperation DomainResult.Success(SessionValidity.INVALID)
when (val validity = profile.gateway().validateSession(profileId)) {
is DomainResult.Failure -> validity
is DomainResult.Success -> {
val stillActive = profileStore.activeProfile()?.id == profileId
val currentSession = profile.gateway().observeSession(profileId).first()
if (stillActive && currentSession?.profileId == profileId) validity
else DomainResult.Success(SessionValidity.INVALID)
}
}
}
suspend fun logout(): DomainResult<Unit> = storeOperation {