feat(kanban): add KanbanSyncCoordinator and harden WebSocket event decoding

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>(KanbanSyncStatus.Disconnected)
+    val syncStatus: StateFlow<KanbanSyncStatus> = _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
+    }
+}

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<String?> {
+    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. */

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<KanbanRepository>(relaxed = true)
+    private val mockEventsClient = mockk<KanbanEventsClient>(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)
+        }
+}

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}""")