From da03500b3058a313781559f776268533cca1c2a8 Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:51:20 +0300 Subject: [PATCH] feat(kanban): add KanbanSyncCoordinator and harden WebSocket event decoding --- .../data/repository/KanbanSyncCoordinator.kt | 125 ++++++++++++++++++ .../data/ws/KanbanEventsClient.kt | 34 +++++ .../repository/KanbanSyncCoordinatorTest.kt | 104 +++++++++++++++ .../data/ws/KanbanEventsClientTest.kt | 10 ++ 4 files changed, 273 insertions(+) create mode 100644 app/src/main/java/com/m57/hermescontrol/data/repository/KanbanSyncCoordinator.kt create mode 100644 app/src/test/java/com/m57/hermescontrol/data/repository/KanbanSyncCoordinatorTest.kt diff --git a/app/src/main/java/com/m57/hermescontrol/data/repository/KanbanSyncCoordinator.kt b/app/src/main/java/com/m57/hermescontrol/data/repository/KanbanSyncCoordinator.kt new file mode 100644 index 00000000..6f9c56f5 --- /dev/null +++ b/app/src/main/java/com/m57/hermescontrol/data/repository/KanbanSyncCoordinator.kt @@ -0,0 +1,125 @@ +package com.m57.hermescontrol.data.repository + +import com.m57.hermescontrol.data.ws.KanbanEventsClient +import com.m57.hermescontrol.data.ws.KanbanLiveStatus +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.launch + +sealed interface KanbanSyncStatus { + data object Live : KanbanSyncStatus + + data class Polling( + val lastSyncEpoch: Long = System.currentTimeMillis(), + ) : KanbanSyncStatus + + data object AuthFailed : KanbanSyncStatus + + data object Disconnected : KanbanSyncStatus +} + +class KanbanSyncCoordinator( + private val repository: KanbanRepository, + private val eventsClientProvider: () -> KanbanEventsClient = { KanbanEventsClient() }, + private val reloadDebounceMs: Long = 250L, + private val pollIntervalMs: Long = 60_000L, + private val nudgeDebounceMs: Long = 400L, +) { + private var eventsClient: KanbanEventsClient? = null + private var activeBoard: String? = null + private var reloadJob: Job? = null + private var pollJob: Job? = null + private var nudgeJob: Job? = null + + private val _syncStatus = MutableStateFlow(KanbanSyncStatus.Disconnected) + val syncStatus: StateFlow = _syncStatus.asStateFlow() + + fun connect( + scope: CoroutineScope, + board: String, + onReload: () -> Unit, + ) { + if (activeBoard == board && _syncStatus.value == KanbanSyncStatus.Live) return + activeBoard = board + + val client = eventsClient ?: eventsClientProvider().also { eventsClient = it } + client.connect( + scope = scope, + board = board, + onEvents = { + reloadJob?.cancel() + reloadJob = + scope.launch { + delay(reloadDebounceMs) + onReload() + } + }, + onStatus = { status -> + when (status) { + KanbanLiveStatus.CONNECTED -> { + _syncStatus.value = KanbanSyncStatus.Live + } + + KanbanLiveStatus.DISCONNECTED -> { + _syncStatus.value = KanbanSyncStatus.Polling() + } + + KanbanLiveStatus.AUTH_FAILED -> { + _syncStatus.value = KanbanSyncStatus.AuthFailed + } + } + }, + ) + } + + fun startVisiblePolling( + scope: CoroutineScope, + board: String, + onPoll: () -> Unit, + ) { + pollJob?.cancel() + pollJob = + scope.launch { + while (true) { + delay(pollIntervalMs) + if (_syncStatus.value !is KanbanSyncStatus.Live) { + onPoll() + _syncStatus.value = KanbanSyncStatus.Polling(System.currentTimeMillis()) + } + } + } + } + + fun stopVisiblePolling() { + pollJob?.cancel() + pollJob = null + } + + fun scheduleNudge( + scope: CoroutineScope, + board: String, + ) { + nudgeJob?.cancel() + nudgeJob = + scope.launch { + delay(nudgeDebounceMs) + repository.nudgeDispatcher(board) + } + } + + fun disconnect() { + activeBoard = null + reloadJob?.cancel() + reloadJob = null + pollJob?.cancel() + pollJob = null + nudgeJob?.cancel() + nudgeJob = null + eventsClient?.disconnect() + _syncStatus.value = KanbanSyncStatus.Disconnected + } +} diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/KanbanEventsClient.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/KanbanEventsClient.kt index d09fd10d..2551a208 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/KanbanEventsClient.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/KanbanEventsClient.kt @@ -8,11 +8,19 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.delay import kotlinx.coroutines.launch +import kotlinx.serialization.KSerializer import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable import kotlinx.serialization.SerializationException import kotlinx.serialization.decodeFromString +import kotlinx.serialization.descriptors.PrimitiveKind +import kotlinx.serialization.descriptors.PrimitiveSerialDescriptor +import kotlinx.serialization.descriptors.SerialDescriptor +import kotlinx.serialization.encoding.Decoder +import kotlinx.serialization.encoding.Encoder +import kotlinx.serialization.json.JsonDecoder import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive import okhttp3.Request import okhttp3.Response import okhttp3.WebSocket @@ -31,11 +39,37 @@ enum class KanbanLiveStatus { AUTH_FAILED, } +@OptIn(kotlinx.serialization.ExperimentalSerializationApi::class) +object StringOrLongSerializer : KSerializer { + override val descriptor: SerialDescriptor = PrimitiveSerialDescriptor("StringOrLong", PrimitiveKind.STRING) + + override fun serialize( + encoder: Encoder, + value: String?, + ) { + if (value == null) { + encoder.encodeNull() + } else { + encoder.encodeString(value) + } + } + + override fun deserialize(decoder: Decoder): String? { + val jsonDecoder = decoder as? JsonDecoder ?: return decoder.decodeString() + val element = jsonDecoder.decodeJsonElement() + if (element is JsonPrimitive) { + return element.content + } + return null + } +} + /** One row of the backend's ``task_events`` table (``/api/plugins/kanban/events``). */ @Serializable data class KanbanEvent( val id: Long, @SerialName("task_id") val taskId: String? = null, + @Serializable(with = StringOrLongSerializer::class) @SerialName("run_id") val runId: String? = null, val kind: String, /** Partial update (e.g. ``{"status": "done"}``) — never applied directly. */ diff --git a/app/src/test/java/com/m57/hermescontrol/data/repository/KanbanSyncCoordinatorTest.kt b/app/src/test/java/com/m57/hermescontrol/data/repository/KanbanSyncCoordinatorTest.kt new file mode 100644 index 00000000..ec8a3d3c --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/data/repository/KanbanSyncCoordinatorTest.kt @@ -0,0 +1,104 @@ +package com.m57.hermescontrol.data.repository + +import com.m57.hermescontrol.data.ws.KanbanEvent +import com.m57.hermescontrol.data.ws.KanbanEventsClient +import com.m57.hermescontrol.data.ws.KanbanEventsEnvelope +import com.m57.hermescontrol.data.ws.KanbanLiveStatus +import io.mockk.coVerify +import io.mockk.mockk +import io.mockk.slot +import io.mockk.verify +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test + +@OptIn(ExperimentalCoroutinesApi::class) +class KanbanSyncCoordinatorTest { + private val testDispatcher = StandardTestDispatcher() + private val testScope = TestScope(testDispatcher) + private val mockRepository = mockk(relaxed = true) + private val mockEventsClient = mockk(relaxed = true) + + private lateinit var coordinator: KanbanSyncCoordinator + + @Before + fun setUp() { + coordinator = + KanbanSyncCoordinator( + repository = mockRepository, + eventsClientProvider = { mockEventsClient }, + reloadDebounceMs = 250L, + pollIntervalMs = 60_000L, + nudgeDebounceMs = 400L, + ) + } + + @Test + fun testLiveStatusTransitions() = + runTest(testDispatcher) { + val statusSlot = slot<(KanbanLiveStatus) -> Unit>() + verify(exactly = 0) { mockEventsClient.connect(any(), any(), any(), any(), any()) } + + coordinator.connect(testScope, "dev", onReload = {}) + verify { mockEventsClient.connect(any(), eq("dev"), any(), any(), capture(statusSlot)) } + + statusSlot.captured(KanbanLiveStatus.CONNECTED) + assertEquals(KanbanSyncStatus.Live, coordinator.syncStatus.value) + + statusSlot.captured(KanbanLiveStatus.DISCONNECTED) + assertTrue(coordinator.syncStatus.value is KanbanSyncStatus.Polling) + + statusSlot.captured(KanbanLiveStatus.AUTH_FAILED) + assertEquals(KanbanSyncStatus.AuthFailed, coordinator.syncStatus.value) + } + + @Test + fun testDebouncedEventReload() = + runTest(testDispatcher) { + val eventsSlot = slot<(KanbanEventsEnvelope) -> Unit>() + var reloadCount = 0 + + coordinator.connect(testScope, "dev", onReload = { reloadCount++ }) + verify { mockEventsClient.connect(any(), eq("dev"), any(), capture(eventsSlot), any()) } + + eventsSlot.captured(KanbanEventsEnvelope(events = listOf(KanbanEvent(id = 1, kind = "status")))) + + testDispatcher.scheduler.advanceTimeBy(100) + assertEquals(0, reloadCount) + + testDispatcher.scheduler.advanceTimeBy(151) + assertEquals(1, reloadCount) + } + + @Test + fun testDebouncedNudge() = + runTest(testDispatcher) { + coordinator.scheduleNudge(testScope, "dev") + testDispatcher.scheduler.advanceTimeBy(200) + + // Still within debounce window + coVerify(exactly = 0) { mockRepository.nudgeDispatcher(any()) } + + testDispatcher.scheduler.advanceTimeBy(201) + testDispatcher.scheduler.advanceUntilIdle() + + coVerify(exactly = 1) { mockRepository.nudgeDispatcher("dev") } + } + + @Test + fun testDisconnectCancelsEverything() = + runTest(testDispatcher) { + coordinator.connect(testScope, "dev", onReload = {}) + coordinator.disconnect() + + verify { mockEventsClient.disconnect() } + assertEquals(KanbanSyncStatus.Disconnected, coordinator.syncStatus.value) + } +} diff --git a/app/src/test/java/com/m57/hermescontrol/data/ws/KanbanEventsClientTest.kt b/app/src/test/java/com/m57/hermescontrol/data/ws/KanbanEventsClientTest.kt index 6e21802a..7dea6fb6 100644 --- a/app/src/test/java/com/m57/hermescontrol/data/ws/KanbanEventsClientTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/data/ws/KanbanEventsClientTest.kt @@ -30,6 +30,16 @@ class KanbanEventsClientTest { assertEquals(1710000000L, event.createdAt) } + @Test + fun `parses event with numeric run_id from sqlite`() { + val frame = + """{"events":[{"id":8,"task_id":"t-2","run_id":42,"kind":"status","payload":null,"created_at":1710000001}],"cursor":8}""" + val envelope = parseKanbanEventsFrame(frame) + assertNotNull(envelope) + assertEquals(1, envelope!!.events.size) + assertEquals("42", envelope.events.first().runId) + } + @Test fun `parses empty events batch`() { val envelope = parseKanbanEventsFrame("""{"events":[],"cursor":3}""")